feat(providers): HttpFetcher with retry + feed payload validation (XWING-2223)

README.md

@@ -17,7 +17,7 @@ All modules live under `traust_core.v1` (see [Versioning](#versioning)).
 | **Object store** | `repositories.objects.store` | `ObjectStore.put/get/list` raw bytes by key, digest-verified (local dir; S3 later) | `write_text`, `json.dump` |
 | **Rendering** | `rendering` | `Report` → `MarkdownRenderer` → `publish()`; `markdown_table()` | 18 copies of `render_md` |
 | **Jobs** | `models.operations`, `services.operations` | `Materializer.materialize(AssetRequest) -> JobResult` (succeeded · degraded · refused · failed): the seam traust-engine calls; `Router` for continuous-ops `route()` | each CLI inventing its own exit codes |
-| **Providers** | `providers` | the outside world: `base` (`Provider`, `Readiness`, `Provenance`, registry), `tools`, `sources`, `feeds`, `process` (`ProcessRunner`, the only subprocess path), `clock` | ad hoc `subprocess`/`git`/fetch code, 22 private `run` wrappers |
+| **Providers** | `providers` | the outside world: `base` (`Provider`, `Readiness`, `Provenance`, registry), `tools`, `sources`, `feeds` (+ `validate_payload`), `http` (`HttpFetcher` + `RetryPolicy`, the outbound HTTP path), `process` (`ProcessRunner`, the only subprocess path), `clock` | ad hoc `subprocess`/`git`/fetch code, 22 private `run` wrappers |
 | **Config + Context** | `context` | one `traust.yaml`; `Context` builds everything | `os.environ`, scattered config files |
 
 ## Layers

src/traust_core/v1/providers/feeds.py

@@ -1,8 +1,13 @@
 from __future__ import annotations
 
+import gzip
+import json
+import zlib
 from datetime import datetime, timedelta
+from pathlib import PurePosixPath
 from typing import Protocol, TypeVar
 
+from traust_core.v1.errors import ValidationError
 from traust_core.v1.models.base import Dto
 from traust_core.v1.models.dependencies import Advisory, Package, Version
 from traust_core.v1.providers.base import Readiness, Result
@@ -38,3 +43,24 @@ class AdvisoryLookup(Protocol):
     def check(self) -> Readiness: ...
 
     def query(self, package: Package, version: Version) -> Result[list[Advisory]]: ...
+
+
+def validate_payload(raw: bytes, filename: str) -> None:
+    """Raise ValidationError when ``raw`` doesn't decode as ``filename``'s type.
+
+    For feeds that publish no checksum, "corrupt" means "not well-formed gzip /
+    JSON / UTF-8 CSV", so a garbage 200 never replaces a good cached copy.
+    """
+    suffixes = PurePosixPath(filename).suffixes
+    try:
+        if suffixes[-1:] == [".gz"]:
+            raw = gzip.decompress(raw)
+            suffixes = suffixes[:-1]
+        if suffixes[-1:] == [".json"]:
+            json.loads(raw)
+        elif suffixes[-1:] == [".csv"]:
+            raw.decode("utf-8")
+    except (OSError, EOFError, ValueError, zlib.error) as e:
+        raise ValidationError(f"{filename}: corrupt payload: {e}") from e
+    if not raw:  # after decompression: an empty gzip body is an empty payload
+        raise ValidationError(f"{filename}: empty payload")

src/traust_core/v1/providers/http.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import http.client
+import random
+import time
+import urllib.error
+import urllib.request
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import UTC
+from email.utils import parsedate_to_datetime
+
+from traust_core.v1.errors import ServiceError
+from traust_core.v1.providers.clock import Clock, SystemClock
+
+
+@dataclass(frozen=True, slots=True)
+class RetryPolicy:
+    max_attempts: int = 4
+    timeout_seconds: float = 120.0
+    backoff_base_seconds: float = 1.0
+    max_wait_seconds: float = 60.0
+    jitter_seconds: float = 0.0
+
+
+class HttpFetcher:
+    """Outbound GET for providers: Retry-After on 429, capped backoff on
+    5xx/timeouts, fail at once on any other 4xx."""
+
+    def __init__(
+        self,
+        user_agent: str,
+        policy: RetryPolicy | None = None,
+        *,
+        clock: Clock | None = None,
+        sleep: Callable[[float], None] = time.sleep,
+    ) -> None:
+        self._user_agent = user_agent
+        self._policy = policy or RetryPolicy()
+        self._clock = clock or SystemClock()
+        self._sleep = sleep
+
+    def get(self, url: str) -> bytes:
+        policy = self._policy
+        last = ""
+        for attempt in range(policy.max_attempts):
+            req = urllib.request.Request(url, headers={"User-Agent": self._user_agent})
+            try:
+                with urllib.request.urlopen(req, timeout=policy.timeout_seconds) as r:
+                    return r.read()
+            except urllib.error.HTTPError as e:
+                # retrying a 403/404 just burns the rate-limit window
+                if e.code != 429 and not 500 <= e.code < 600:
+                    raise ServiceError(f"{url}: HTTP {e.code}", status=e.code) from e
+                last = f"HTTP {e.code}"
+                after = self._retry_after(e.headers.get("Retry-After") if e.headers else None)
+                delay = (
+                    self._backoff(attempt) if after is None else min(after, policy.max_wait_seconds)
+                )
+            except (
+                urllib.error.URLError,
+                TimeoutError,
+                OSError,
+                http.client.IncompleteRead,
+            ) as e:
+                last = str(e)
+                delay = self._backoff(attempt)
+            if attempt + 1 < policy.max_attempts:
+                self._sleep(delay)
+        raise ServiceError(f"{url}: gave up after {policy.max_attempts} attempts ({last})")
+
+    def _backoff(self, attempt: int) -> float:
+        p = self._policy
+        wait = min(p.backoff_base_seconds * 2**attempt, p.max_wait_seconds)
+        return wait + (random.uniform(0, p.jitter_seconds) if p.jitter_seconds else 0.0)
+
+    def _retry_after(self, value: str | None) -> float | None:
+        """Retry-After as delta-seconds or an HTTP-date; None when absent or unparseable."""
+        if not value:
+            return None
+        value = value.strip()
+        if value.isdigit():
+            return float(value)
+        try:
+            when = parsedate_to_datetime(value)
+        except (TypeError, ValueError):
+            return None
+        if when.tzinfo is None:
+            when = when.replace(tzinfo=UTC)
+        return max(0.0, (when - self._clock.now()).total_seconds())

tests/v1/providers/test_feeds.py

@@ -0,0 +1,35 @@
+import gzip
+
+import pytest
+
+from traust_core.v1.errors import ValidationError
+from traust_core.v1.providers.feeds import validate_payload
+
+
+@pytest.mark.parametrize(
+    ("raw", "filename"),
+    [
+        (gzip.compress(b"cve,epss\n"), "epss_scores-current.csv.gz"),
+        (b'{"a": 1}', "known_exploited_vulnerabilities.json"),
+        (gzip.compress(b"[]"), "index.json.gz"),
+        (b"anything", "archive_latest.txt"),
+    ],
+)
+def test_well_formed_payload_passes(raw: bytes, filename: str) -> None:
+    validate_payload(raw, filename)
+
+
+@pytest.mark.parametrize(
+    ("raw", "filename"),
+    [
+        (b"", "kev.json"),
+        (b"not a gzip payload", "epss.csv.gz"),
+        (b"{broken", "kev.json"),
+        (b"\xff\xfe", "rh-cve.csv"),
+        (gzip.compress(b""), "epss_scores-current.csv.gz"),
+        (gzip.compress(b"cve,epss\n")[:10] + b"\xff\xff", "epss_scores-current.csv.gz"),
+    ],
+)
+def test_corrupt_payload_raises(raw: bytes, filename: str) -> None:
+    with pytest.raises(ValidationError):
+        validate_payload(raw, filename)

tests/v1/providers/test_http.py

@@ -0,0 +1,127 @@
+import http.client as http_client
+import urllib.error
+from collections.abc import Callable
+from datetime import UTC, datetime
+from email.message import Message
+from typing import Any
+
+import pytest
+
+from traust_core.v1.errors import ServiceError
+from traust_core.v1.providers import http
+from traust_core.v1.providers.clock import FixedClock
+from traust_core.v1.providers.http import HttpFetcher, RetryPolicy
+
+
+class _Resp:
+    def __init__(self, body: bytes) -> None:
+        self._body = body
+
+    def __enter__(self) -> "_Resp":
+        return self
+
+    def __exit__(self, *a: object) -> None:
+        return None
+
+    def read(self) -> bytes:
+        return self._body
+
+
+def _urlopen(actions: list[Any], calls: list[int]) -> Callable[..., _Resp]:
+    """Walks ``actions``: bytes are returned as the body, exceptions are raised."""
+    it = iter(actions)
+
+    def _open(req: object, timeout: float | None = None) -> _Resp:
+        calls.append(1)
+        action = next(it)
+        if isinstance(action, Exception):
+            raise action
+        return _Resp(action)
+
+    return _open
+
+
+def _http_error(code: int, retry_after: str | None = None) -> urllib.error.HTTPError:
+    headers = Message()
+    if retry_after is not None:
+        headers["Retry-After"] = retry_after
+    return urllib.error.HTTPError("http://feed.test", code, "err", headers, None)
+
+
+def _fetcher(slept: list[float], **policy: float) -> HttpFetcher:
+    base: dict[str, float] = {"max_attempts": 3, "backoff_base_seconds": 0, "max_wait_seconds": 10}
+    return HttpFetcher(
+        "test-agent",
+        RetryPolicy(**{**base, **policy}),  # type: ignore[arg-type]
+        clock=FixedClock(datetime(2026, 1, 1, tzinfo=UTC)),
+        sleep=slept.append,
+    )
+
+
+def _patch(monkeypatch: pytest.MonkeyPatch, actions: list[Any]) -> list[int]:
+    calls: list[int] = []
+    monkeypatch.setattr(http.urllib.request, "urlopen", _urlopen(actions, calls))
+    return calls
+
+
+def test_retries_5xx_then_succeeds(monkeypatch: pytest.MonkeyPatch) -> None:
+    _patch(monkeypatch, [_http_error(503), b"ok"])
+    assert _fetcher([]).get("http://feed.test") == b"ok"
+
+
+def test_retries_timeout_then_succeeds(monkeypatch: pytest.MonkeyPatch) -> None:
+    _patch(monkeypatch, [TimeoutError("slow"), b"ok"])
+    assert _fetcher([]).get("http://feed.test") == b"ok"
+
+
+@pytest.mark.parametrize(
+    ("retry_after", "expected"),
+    [("2", 2.0), ("Thu, 01 Jan 2026 00:00:05 GMT", 5.0), ("999", 10.0)],
+)
+def test_429_honours_retry_after_capped(
+    monkeypatch: pytest.MonkeyPatch, retry_after: str, expected: float
+) -> None:
+    _patch(monkeypatch, [_http_error(429, retry_after), b"ok"])
+    slept: list[float] = []
+    assert _fetcher(slept).get("http://feed.test") == b"ok"
+    assert slept == [expected]
+
+
+def test_backoff_doubles_and_is_capped(monkeypatch: pytest.MonkeyPatch) -> None:
+    _patch(monkeypatch, [_http_error(500)] * 4)
+    slept: list[float] = []
+    with pytest.raises(ServiceError):
+        _fetcher(slept, max_attempts=4, backoff_base_seconds=3, max_wait_seconds=10).get("http://x")
+    assert slept == [3.0, 6.0, 10.0]
+
+
+def test_other_4xx_fails_fast(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = _patch(monkeypatch, [_http_error(404)])
+    with pytest.raises(ServiceError) as exc:
+        _fetcher([]).get("http://feed.test")
+    assert exc.value.status == 404
+    assert len(calls) == 1
+
+
+def test_gives_up_after_max_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = _patch(monkeypatch, [TimeoutError("slow")] * 3)
+    with pytest.raises(ServiceError, match="gave up after 3 attempts"):
+        _fetcher([]).get("http://feed.test")
+    assert len(calls) == 3
+
+
+def test_truncated_body_retries_then_succeeds(monkeypatch: pytest.MonkeyPatch) -> None:
+    slept: list[float] = []
+    calls = _patch(monkeypatch, [http_client.IncompleteRead(b"partial", 100), b"the full body"])
+    fetcher = _fetcher(slept)
+    assert fetcher.get("http://feed.test/x") == b"the full body"
+    assert len(calls) == 2
+
+
+def test_truncated_body_gives_up_as_service_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    slept: list[float] = []
+    calls = _patch(monkeypatch, [http_client.IncompleteRead(b"partial", 100)] * 3)
+    fetcher = _fetcher(slept)
+    with pytest.raises(ServiceError):
+        fetcher.get("http://feed.test/x")
+    assert len(calls) == 3