From 60ea0d849a337ec31f259e4cf20cfd6208655b91 Mon Sep 17 00:00:00 2001 From: Kyl-Bempah <24903083+Kyl-Bempah@users.noreply.github.com> Date: Fri, 16 Oct 2026 19:27:41 -0400 Subject: [PATCH 1/3] Adding http & feed providers Signed-off-by: Kyl-Bempah <24903083+Kyl-Bempah@users.noreply.github.com> --- README.md | 2 +- src/traust_core/v1/providers/feeds.py | 25 ++++++ src/traust_core/v1/providers/http.py | 87 ++++++++++++++++++++ tests/v1/providers/test_feeds.py | 33 ++++++++ tests/v1/providers/test_http.py | 109 ++++++++++++++++++++++++++ 5 files changed, 255 insertions(+), 1 deletion(-) create mode 100644 src/traust_core/v1/providers/http.py create mode 100644 tests/v1/providers/test_feeds.py create mode 100644 tests/v1/providers/test_http.py diff --git a/README.md b/README.md index 53e12b5..234a0dc 100644 --- a/README.md +++ b/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 diff --git a/src/traust_core/v1/providers/feeds.py b/src/traust_core/v1/providers/feeds.py index e6c79ac..9a26e84 100644 --- a/src/traust_core/v1/providers/feeds.py +++ b/src/traust_core/v1/providers/feeds.py @@ -1,8 +1,12 @@ from __future__ import annotations +import gzip +import json 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 +42,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. + """ + if not raw: + raise ValidationError(f"{filename}: empty payload") + 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) as e: + raise ValidationError(f"{filename}: corrupt payload: {e}") from e diff --git a/src/traust_core/v1/providers/http.py b/src/traust_core/v1/providers/http.py new file mode 100644 index 0000000..9f9d914 --- /dev/null +++ b/src/traust_core/v1/providers/http.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +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, with bounded retries. + + 429 waits out Retry-After (capped by max_wait_seconds); 5xx and timeouts use + capped exponential backoff; any other 4xx fails at once (retrying a 403/404 + just burns the rate-limit window). + """ + + 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: + 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) 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()) diff --git a/tests/v1/providers/test_feeds.py b/tests/v1/providers/test_feeds.py new file mode 100644 index 0000000..852a00b --- /dev/null +++ b/tests/v1/providers/test_feeds.py @@ -0,0 +1,33 @@ +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"), + ], +) +def test_corrupt_payload_raises(raw: bytes, filename: str) -> None: + with pytest.raises(ValidationError): + validate_payload(raw, filename) diff --git a/tests/v1/providers/test_http.py b/tests/v1/providers/test_http.py new file mode 100644 index 0000000..eeee3ee --- /dev/null +++ b/tests/v1/providers/test_http.py @@ -0,0 +1,109 @@ +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 From cc5e37bc5a0dcbd17040cc393bfde11c62844eaf Mon Sep 17 00:00:00 2001 From: Shaun Smith <51304449+smith-xyz@users.noreply.github.com> Date: Wed, 21 Oct 2026 15:53:19 -0400 Subject: [PATCH 2/3] docs(providers): compress the HttpFetcher docstring to the two-line convention --- src/traust_core/v1/providers/http.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/src/traust_core/v1/providers/http.py b/src/traust_core/v1/providers/http.py index 9f9d914..3fff8ea 100644 --- a/src/traust_core/v1/providers/http.py +++ b/src/traust_core/v1/providers/http.py @@ -23,12 +23,8 @@ class RetryPolicy: class HttpFetcher: - """Outbound GET for providers, with bounded retries. - - 429 waits out Retry-After (capped by max_wait_seconds); 5xx and timeouts use - capped exponential backoff; any other 4xx fails at once (retrying a 403/404 - just burns the rate-limit window). - """ + """Outbound GET for providers: Retry-After on 429, capped backoff on + 5xx/timeouts, fail at once on any other 4xx.""" def __init__( self, @@ -52,6 +48,7 @@ def get(self, url: str) -> bytes: 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}" From 8f2bf1bbaba57f09f5bb07722e7dc760771ede8a Mon Sep 17 00:00:00 2001 From: Shaun Smith <51304449+smith-xyz@users.noreply.github.com> Date: Wed, 21 Oct 2026 16:02:58 -0400 Subject: [PATCH 3/3] fix(providers): close the three review-reproduced failure paths - validate_payload checks emptiness after decompression: an empty gzip body is an empty payload, never a cache replacement - gzip.decompress failures normalize to ValidationError (zlib.error joined the caught set; invalid DEFLATE no longer escapes raw) - truncated HTTP bodies (http.client.IncompleteRead) join the retryable transport errors: retried with backoff, ServiceError when exhausted Each reproduced before fixing; regression tests for all five paths. --- src/traust_core/v1/providers/feeds.py | 7 ++++--- src/traust_core/v1/providers/http.py | 8 +++++++- tests/v1/providers/test_feeds.py | 2 ++ tests/v1/providers/test_http.py | 18 ++++++++++++++++++ 4 files changed, 31 insertions(+), 4 deletions(-) diff --git a/src/traust_core/v1/providers/feeds.py b/src/traust_core/v1/providers/feeds.py index 9a26e84..a1c4c8b 100644 --- a/src/traust_core/v1/providers/feeds.py +++ b/src/traust_core/v1/providers/feeds.py @@ -2,6 +2,7 @@ import gzip import json +import zlib from datetime import datetime, timedelta from pathlib import PurePosixPath from typing import Protocol, TypeVar @@ -50,8 +51,6 @@ def validate_payload(raw: bytes, filename: str) -> None: 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. """ - if not raw: - raise ValidationError(f"{filename}: empty payload") suffixes = PurePosixPath(filename).suffixes try: if suffixes[-1:] == [".gz"]: @@ -61,5 +60,7 @@ def validate_payload(raw: bytes, filename: str) -> None: json.loads(raw) elif suffixes[-1:] == [".csv"]: raw.decode("utf-8") - except (OSError, EOFError, ValueError) as e: + 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") diff --git a/src/traust_core/v1/providers/http.py b/src/traust_core/v1/providers/http.py index 3fff8ea..94b8eeb 100644 --- a/src/traust_core/v1/providers/http.py +++ b/src/traust_core/v1/providers/http.py @@ -1,5 +1,6 @@ from __future__ import annotations +import http.client import random import time import urllib.error @@ -56,7 +57,12 @@ def get(self, url: str) -> bytes: delay = ( self._backoff(attempt) if after is None else min(after, policy.max_wait_seconds) ) - except (urllib.error.URLError, TimeoutError, OSError) as e: + except ( + urllib.error.URLError, + TimeoutError, + OSError, + http.client.IncompleteRead, + ) as e: last = str(e) delay = self._backoff(attempt) if attempt + 1 < policy.max_attempts: diff --git a/tests/v1/providers/test_feeds.py b/tests/v1/providers/test_feeds.py index 852a00b..d2bee34 100644 --- a/tests/v1/providers/test_feeds.py +++ b/tests/v1/providers/test_feeds.py @@ -26,6 +26,8 @@ def test_well_formed_payload_passes(raw: bytes, filename: str) -> None: (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: diff --git a/tests/v1/providers/test_http.py b/tests/v1/providers/test_http.py index eeee3ee..42cf8a7 100644 --- a/tests/v1/providers/test_http.py +++ b/tests/v1/providers/test_http.py @@ -1,3 +1,4 @@ +import http.client as http_client import urllib.error from collections.abc import Callable from datetime import UTC, datetime @@ -107,3 +108,20 @@ def test_gives_up_after_max_attempts(monkeypatch: pytest.MonkeyPatch) -> None: 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