From d2627b8a18e068cca02c80e30569068621e85e49 Mon Sep 17 00:00:00 2001 From: Derek Date: Mon, 16 Nov 2026 05:26:08 +1100 Subject: [PATCH 1/3] fix(system): report the stack whose manifest matches the running images A Docker install's compose keeps its build-time stack name. A later stack still shows that older name while running newer images. The engine reads published stack manifests (ghcr.io/hyperi-io/dfe-stack-manifest) in the background after startup. It reports the newest release, then the newest prerelease, whose digests match every image in DFE_ENGINE_STACK_IMAGES. This matches dfe-infra's ordering: a release beats release candidates with identical images. - /system/version and /system/deployment report the matching stack with source "registry". The deploy-repo pins still win. The stack_version label stays the fallback. - Startup and readiness never wait on the registry. A failure or no match logs one warning and keeps the label. - DFE_ENGINE_STACK_RESOLVE=false switches off registry lookup for an air-gapped site. DFE_ENGINE_STACK_MANIFEST_REPOSITORY points the lookup at a mirror. - The settings loader now keeps false or zero top-level environment values. Dropping them caused DFE_ENGINE_STACK_RESOLVE=false to be ignored. The resolver does not verify the artifact signature. --- docs/data-plane/source-flow.md | 2 +- openapi/openapi.json | 6 +- src/dfe_engine/api/app.py | 12 + src/dfe_engine/api/v1/system.py | 40 +- src/dfe_engine/settings.py | 55 +- src/dfe_engine/stack_resolve.py | 397 ++++++++++++++ .../test_stack_resolve_registry.py | 77 +++ tests/unit/test_api/test_system_version.py | 29 ++ tests/unit/test_settings.py | 28 + tests/unit/test_stack_resolve.py | 492 ++++++++++++++++++ 10 files changed, 1122 insertions(+), 16 deletions(-) create mode 100644 src/dfe_engine/stack_resolve.py create mode 100644 tests/integration/test_stack_resolve_registry.py create mode 100644 tests/unit/test_stack_resolve.py diff --git a/docs/data-plane/source-flow.md b/docs/data-plane/source-flow.md index 1737d1c3..01f2df53 100644 --- a/docs/data-plane/source-flow.md +++ b/docs/data-plane/source-flow.md @@ -129,7 +129,7 @@ Every app rolls on every config change, so nothing here depends on which app can ## Asking what the deployment is, and what a source is doing -Two reads carry the facts a console would otherwise have to guess at. `GET /api/v1/system/deployment` answers what this deployment IS: the profile the deployer injected as `DFE_ENGINE_PROFILE`, the one transport it carries a source on - which is both the only one a source may name and the one a source that names none takes - whether the stage pools sit behind mesh listeners and where, whether compiled routing reaches its running apps (`applies_routing`), and the stack, engine, dfe-ui and per-app pinned versions (`apps`, name to version tag, from the deploy repo's pins). `GET /api/v1/system/version` serves the version half from the same function, so the footer and the deployment card cannot disagree, and `GET /api/v1/apps` judges each app's `offered` against the deployment's own profile when the caller names none. +Two reads carry the facts a console would otherwise have to guess at. `GET /api/v1/system/deployment` answers what this deployment IS: the profile the deployer injected as `DFE_ENGINE_PROFILE`, the one transport it carries a source on - which is both the only one a source may name and the one a source that names none takes - whether the stage pools sit behind mesh listeners and where, whether compiled routing reaches its running apps (`applies_routing`), and the stack, engine, dfe-ui and per-app pinned versions (`apps`, name to version tag, from the deploy repo's pins). The stack comes from the deploy repo's pins, else from the published stack whose image digests match the `DFE_ENGINE_STACK_IMAGES` this deployment runs, else from the `DFE_ENGINE_STACK_VERSION` label, and `source` names which (`deploy-repo`, `registry`, `deployment`, `engine`). `DFE_ENGINE_STACK_RESOLVE=false` skips the registry read for an air-gapped site, and `DFE_ENGINE_STACK_MANIFEST_REPOSITORY` points it at a mirror. `GET /api/v1/system/version` serves the version half from the same function, so the footer and the deployment card cannot disagree, and `GET /api/v1/apps` judges each app's `offered` against the deployment's own profile when the caller names none. `GET /api/v1/sources/{name}/signals` answers what one source is doing: `records_per_min` and `last_seen` over a bounded five-minute window, read from the loader's `loader_messages_by_table_total` counter in the otel database, plus `landed_in_default`. The counter is per TABLE, so `landed_in_default` says the number covers the whole shared landing table rather than this source alone - which is the case for a fetcher pushing to the main topic. Either reading is null where no series answers, and the route still returns 200: a source only just defined has nothing to report, and a zero would claim a measurement nobody took. diff --git a/openapi/openapi.json b/openapi/openapi.json index be73b324..05308c5d 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -39431,11 +39431,12 @@ "type": "string", "enum": [ "deploy-repo", + "registry", "deployment", "engine" ], "title": "Source", - "description": "deploy-repo when the stack version came from pins.yaml, deployment when it came from what deployed this pod, else engine." + "description": "deploy-repo when the stack version came from pins.yaml, registry when the published stack manifests matched the images this deployment runs, deployment when it came from what deployed this pod, else engine." }, "backing": { "additionalProperties": { @@ -56247,11 +56248,12 @@ "type": "string", "enum": [ "deploy-repo", + "registry", "deployment", "engine" ], "title": "Source", - "description": "deploy-repo when the stack version came from pins.yaml, deployment when it came from what deployed this pod, else engine." + "description": "deploy-repo when the stack version came from pins.yaml, registry when the published stack manifests matched the images this deployment runs, deployment when it came from what deployed this pod, else engine." }, "python_version": { "type": "string", diff --git a/src/dfe_engine/api/app.py b/src/dfe_engine/api/app.py index 34460727..036a0974 100644 --- a/src/dfe_engine/api/app.py +++ b/src/dfe_engine/api/app.py @@ -656,11 +656,23 @@ def reconcile_service_roles_now() -> ReconcileResult: name="gitops-consolidation", ) + # In the background: the registry is the internet's, so startup and readiness never wait on it. + from dfe_engine.stack_resolve import resolve_into + + app.state.resolved_stack = None + app.state.stack_resolution = asyncio.create_task( + resolve_into(app.state, settings), name="stack-resolution" + ) + health.set_started() health.set_ready() logger.info(f"DFE Engine API started (port={settings.api.port})") yield health.set_ready(False) + stack_resolution = getattr(app.state, "stack_resolution", None) + if stack_resolution is not None: + stack_resolution.cancel() + await asyncio.gather(stack_resolution, return_exceptions=True) oidc_group_sync = getattr(app.state, "oidc_group_sync", None) if oidc_group_sync is not None: # A sync writes group files and asks for a reconcile, so it stops before both close. diff --git a/src/dfe_engine/api/v1/system.py b/src/dfe_engine/api/v1/system.py index a5119be9..0f0aeaa6 100644 --- a/src/dfe_engine/api/v1/system.py +++ b/src/dfe_engine/api/v1/system.py @@ -145,10 +145,11 @@ class VersionResponse(BaseModel): "dfe-ui, which then matches the ui field above." ) ) - source: Literal["deploy-repo", "deployment", "engine"] = Field( + source: Literal["deploy-repo", "registry", "deployment", "engine"] = Field( description=( - "deploy-repo when the stack version came from pins.yaml, deployment when it " - "came from what deployed this pod, else engine." + "deploy-repo when the stack version came from pins.yaml, registry when the " + "published stack manifests matched the images this deployment runs, deployment " + "when it came from what deployed this pod, else engine." ) ) python_version: str = Field(description="Python interpreter version") @@ -285,10 +286,11 @@ class DeploymentResponse(BaseModel): "dfe-ui, which then matches the ui field above." ) ) - source: Literal["deploy-repo", "deployment", "engine"] = Field( + source: Literal["deploy-repo", "registry", "deployment", "engine"] = Field( description=( - "deploy-repo when the stack version came from pins.yaml, deployment when it " - "came from what deployed this pod, else engine." + "deploy-repo when the stack version came from pins.yaml, registry when the " + "published stack manifests matched the images this deployment runs, deployment " + "when it came from what deployed this pod, else engine." ) ) backing: dict[str, BackingFacts] = Field( @@ -357,6 +359,17 @@ class SettingsSummary(BaseModel): # -- Endpoints ------------------------------------------------ +def _stack_source( + pinned: str | None, resolved: str | None, stack: str | None +) -> Literal["deploy-repo", "registry", "deployment", "engine"]: + """Name where ``stack`` came from, in the order deployment_facts prefers them.""" + if pinned: + return "deploy-repo" + if resolved: + return "registry" + return "deployment" if stack else "engine" + + def deployment_facts(request: Request, settings: Any) -> DeploymentResponse: """Everything this deployment can state about itself, read once. @@ -364,10 +377,12 @@ def deployment_facts(request: Request, settings: Any) -> DeploymentResponse: the console footer shows and the versions its deployment card shows cannot disagree. - The pins win where there are any: they are what the operator chose. A deploy - with no pins base still knows what stood it up, because the chart passes that - in, so the footer shows a stack version rather than nothing. With neither, the - engine's own version is the whole answer. + The pins win where there are any: they are what the operator chose. Next comes + the published stack whose image digests match what this deployment runs, which + the engine resolves in the background after startup. A deploy with neither still + knows what stood it up, because the chart or compose passes that in, so the footer + shows a stack version rather than nothing. With none of them, the engine's own + version is the whole answer. Blocking: it reads the deploy repo, so a route calls it off the event loop. """ @@ -380,7 +395,8 @@ def deployment_facts(request: Request, settings: Any) -> DeploymentResponse: with gc.reading(): pins = load_pins(gc.repo_path) pinned = stack_version(pins) - stack = pinned or settings.stack_version or None + resolved = getattr(request.app.state, "resolved_stack", None) + stack = pinned or resolved or settings.stack_version or None apps = component_overrides(pins) transport = settings.transport return DeploymentResponse( @@ -402,7 +418,7 @@ def deployment_facts(request: Request, settings: Any) -> DeploymentResponse: engine=__version__, ui=apps.get(UI_COMPONENT) or settings.ui_version or None, apps=apps, - source="deploy-repo" if pinned else ("deployment" if stack else "engine"), + source=_stack_source(pinned, resolved, stack), backing=backing_facts(request), ) diff --git a/src/dfe_engine/settings.py b/src/dfe_engine/settings.py index ae5f7d8a..3b479e0b 100644 --- a/src/dfe_engine/settings.py +++ b/src/dfe_engine/settings.py @@ -2159,6 +2159,49 @@ class DFESettings(BaseModel): "repo pins no stack. DFE_ENGINE_STACK_VERSION." ), ) + stack_images: list[str] = Field( + default_factory=list, + description=( + "The DFE images this deployment runs, each as ref@sha256:digest, comma-separated " + "in DFE_ENGINE_STACK_IMAGES. When set, the engine reads the published stack " + "manifests at startup and reports the newest stack whose digests agree with " + "every image listed, in place of the stack_version label. List DFE apps only: " + "an image no stack manifest names stops any stack matching." + ), + ) + stack_manifest_repository: str = Field( + default="ghcr.io/hyperi-io/dfe-stack-manifest", + description=( + "host/path of the registry repository the stack manifests are published to, " + "read anonymously. Point it at a mirror for another organisation. " + "DFE_ENGINE_STACK_MANIFEST_REPOSITORY." + ), + ) + stack_resolve: bool = Field( + default=True, + description=( + "Resolve the running stack from stack_images against the registry. Set false " + "on an air-gapped site, which then reports the stack_version label. " + "DFE_ENGINE_STACK_RESOLVE." + ), + ) + stack_resolve_request_timeout_seconds: float = Field( + default=5.0, + gt=0, + description=( + "Longest one registry request in the stack resolution may take. " + "DFE_ENGINE_STACK_RESOLVE_REQUEST_TIMEOUT_SECONDS." + ), + ) + stack_resolve_budget_seconds: float = Field( + default=30.0, + gt=0, + description=( + "Longest the whole stack resolution may take, after which the stack_version " + "label answers. It runs in the background and never holds startup or readiness. " + "DFE_ENGINE_STACK_RESOLVE_BUDGET_SECONDS." + ), + ) ui_version: str = Field( default="", description=( @@ -2377,6 +2420,7 @@ def _strip_lower(value: str) -> str: "DFE_ENGINE_SECRETS_RECOVERY_WINDOW_DAYS": "secrets.recovery_window_days", "DFE_ENGINE_SECRETS_TAGS": "secrets.tags", "DFE_ENGINE_STACK_VERSION": "stack_version", + "DFE_ENGINE_STACK_MANIFEST_REPOSITORY": "stack_manifest_repository", "DFE_ENGINE_UI_VERSION": "ui_version", "DFE_ENGINE_ENV": "env", }, @@ -2446,6 +2490,8 @@ def _strip_lower(value: str) -> str: { "DFE_ENGINE_CLICKHOUSE_BOOTSTRAP_WAIT_SECONDS": "clickhouse.bootstrap_wait_seconds", "DFE_ENGINE_DETECTION_GUARD_PREVIEW_TIMEOUT_SECONDS": "hunts.detection_guard.preview_timeout_seconds", + "DFE_ENGINE_STACK_RESOLVE_REQUEST_TIMEOUT_SECONDS": "stack_resolve_request_timeout_seconds", + "DFE_ENGINE_STACK_RESOLVE_BUDGET_SECONDS": "stack_resolve_budget_seconds", "DFE_ENGINE_HUNTS_PHASE_SPREAD_FRACTION": "hunts.phase_spread_fraction", "DFE_ENGINE_HUNTS_FAILURE_BACKOFF_INITIAL_SECONDS": "hunts.failure_backoff_initial_seconds", "DFE_ENGINE_HUNTS_FAILURE_BACKOFF_MAX_SECONDS": "hunts.failure_backoff_max_seconds", @@ -2497,6 +2543,7 @@ def _strip_lower(value: str) -> str: "DFE_ENGINE_GITOPS_PUSH": "gitops.push", "DFE_ENGINE_API_DOCS_ENABLED": "api.docs_enabled", "DFE_ENGINE_E2E_SERVER": "e2e_server", + "DFE_ENGINE_STACK_RESOLVE": "stack_resolve", }, ), ( @@ -2541,6 +2588,7 @@ def _strip_lower(value: str) -> str: "DFE_ENGINE_HUNTS_ALERT_CHANNELS": "hunts.alert_channels", "DFE_ENGINE_HUNTS_LOOKUP_DATABASES": "hunts.lookup_databases", "DFE_ENGINE_API_CORS_ORIGINS": "api.cors_origins", + "DFE_ENGINE_STACK_IMAGES": "stack_images", }, ), ) @@ -2607,7 +2655,12 @@ def _get_env_overrides() -> dict: if key not in overrides.get(section, {}): _set_path(overrides, path, str(Path(config_dir) / subdir)) - return {section: value for section, value in overrides.items() if value} + # Only an empty section goes: a top-level false or zero is the operator's answer. + return { + section: value + for section, value in overrides.items() + if value or not isinstance(value, dict) + } def _deep_merge(base: dict, override: dict) -> dict: diff --git a/src/dfe_engine/stack_resolve.py b/src/dfe_engine/stack_resolve.py new file mode 100644 index 00000000..398f5bc4 --- /dev/null +++ b/src/dfe_engine/stack_resolve.py @@ -0,0 +1,397 @@ +# Project: dfe-engine +# File: stack_resolve.py +# Purpose: Name the published DFE stack whose images a deployment is running +# Language: Python +# +# License: BUSL-1.1 +# Copyright: (c) 2026 HYPERI PTY LIMITED +"""Resolve the stack a deployment runs from its image digests. + +A stack name a build wrote down says what the build was made against, not what is +running. Each cut stack is published as an OCI artifact (``dfe-stack-manifest:``) +whose ``dfe-manifest-.yaml`` layer pins every app image by digest. This module +walks those artifacts, newest release first and then newest prerelease, and returns +the first whose digests agree with every image the deployment runs. + +Reads are anonymous: a registry that wants a bearer token gets the anonymous one its +own challenge names. Signature verification of the artifact is not done here. +""" + +import asyncio +import hashlib +import re +import threading +import time +from collections.abc import Mapping, Sequence +from typing import Any +from urllib.parse import urlsplit + +import httpx +from ruamel.yaml import YAML, YAMLError +from scalo.http import HttpClient +from scalo.logger import logger + +_SEMVER = re.compile(r"^v?(\d+)\.(\d+)\.(\d+)(?:-([0-9A-Za-z.-]+))?$") +_CHALLENGE_PARAM = re.compile(r'(\w+)="([^"]*)"') +_MANIFEST_LAYER = re.compile(r"^dfe-manifest-.+\.yaml$") +_TITLE_ANNOTATION = "org.opencontainers.image.title" +_OCI_ACCEPT = ", ".join( + ( + "application/vnd.oci.image.manifest.v1+json", + "application/vnd.docker.distribution.manifest.v2+json", + ) +) +_TAGS_PAGE_SIZE = 1000 +_MAX_MANIFEST_BYTES = 1024 * 1024 +_LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1"}) +_BUDGET_SPENT = "time budget spent before a matching stack was found" + + +class StackResolveError(Exception): + """The registry could not be read or answered something unusable.""" + + +class BudgetSpentError(StackResolveError): + """The time allowed for the whole resolution ran out.""" + + +def semver_key(tag: str) -> tuple[int, int, int, int, tuple[tuple[int, Any], ...]] | None: + """Return the SemVer precedence key of ``tag``, or None when it is not semver-shaped. + + A release outranks every prerelease of the same ``X.Y.Z``, a numeric prerelease + identifier ranks below an alphanumeric one, and identifiers compare left to right. + This is the ordering dfe-infra's ``scripts/dfe-stack`` ``semver_key`` uses. + + Args: + tag: A stack tag such as ``2.2.2`` or ``2.2.2-rc.5``. + """ + match = _SEMVER.match(tag.strip()) + if not match: + return None + major, minor, patch = (int(match.group(i)) for i in (1, 2, 3)) + prerelease = match.group(4) + if prerelease is None: + return (major, minor, patch, 1, ()) + identifiers = tuple( + (0, int(part)) if part.isdigit() else (1, part) for part in prerelease.split(".") + ) + return (major, minor, patch, 0, identifiers) + + +def newest_first(tags: Sequence[str]) -> list[str]: + """Order stack tags releases newest first, then prereleases newest first. + + Tags that are not semver-shaped are dropped. + """ + keyed = [(key, tag) for tag in tags if (key := semver_key(tag)) is not None] + releases = [item for item in keyed if item[0][3] == 1] + prereleases = [item for item in keyed if item[0][3] == 0] + return [tag for _, tag in sorted(releases, reverse=True)] + [ + tag for _, tag in sorted(prereleases, reverse=True) + ] + + +def image_name(reference: str) -> str: + """Return the image name of ``registry/path/name[:tag][@sha256:digest]``.""" + without_digest = reference.strip().split("@", 1)[0] + last = without_digest.rsplit("/", 1)[-1] + return last.split(":", 1)[0] + + +def image_digest(reference: str) -> str | None: + """Return the ``sha256:...`` digest a reference pins, or None when it pins none.""" + _, separator, digest = reference.strip().partition("@") + return digest if separator and digest.startswith("sha256:") else None + + +def running_digests(references: Sequence[str]) -> dict[str, str]: + """Map image name to digest for the images a deployment runs. + + Raises: + StackResolveError: A reference pins no digest, or two references name one image + with different digests. + """ + running: dict[str, str] = {} + for reference in references: + digest = image_digest(reference) + if digest is None: + raise StackResolveError(f"image {reference!r} pins no sha256 digest") + name = image_name(reference) + if running.setdefault(name, digest) != digest: + raise StackResolveError(f"image {name} appears with two different digests") + return running + + +def matches(manifest_images: Mapping[str, str], running: Mapping[str, str]) -> bool: + """Return True when every running image is in the manifest at the same digest. + + The manifest may list more apps than the deployment runs. An image the manifest + does not list at all, or lists at another digest, is not that stack. + """ + if not running: + return False + for name, digest in running.items(): + listed = manifest_images.get(name) + if not isinstance(listed, str) or image_digest(listed) != digest: + return False + return True + + +def manifest_images(document: Any) -> dict[str, str]: + """Return the ``images`` mapping of a parsed stack manifest, name to reference.""" + images = document.get("images") if isinstance(document, dict) else None + if not isinstance(images, dict): + raise StackResolveError("manifest has no images mapping") + return {str(name): ref for name, ref in images.items() if isinstance(ref, str)} + + +def _split_repository(repository: str) -> tuple[str, str]: + """Split ``host/path`` into the registry base URL and the repository path. + + The base URL is https, except for a loopback host, which a registry serves over + plain http the way ``docker`` treats it. + """ + host, separator, path = repository.strip().strip("/").partition("/") + looks_like_host = "." in host or ":" in host or host == "localhost" + if not separator or not path or not looks_like_host: + raise StackResolveError(f"manifest repository {repository!r} is not host/path") + scheme = "http" if host.rsplit(":", 1)[0] in _LOOPBACK_HOSTS else "https" + return f"{scheme}://{host}", path + + +def _layer_title(layer: Mapping[str, Any]) -> str: + """Return the file name an OCI layer descriptor carries, or an empty string.""" + annotations = layer.get("annotations") + title = annotations.get(_TITLE_ANNOTATION) if isinstance(annotations, dict) else None + return title if isinstance(title, str) else "" + + +def _json_object(response: httpx.Response) -> dict[str, Any]: + """Return the JSON object a registry answered with. + + Raises: + StackResolveError: The body is not JSON, or is JSON that is not an object. + """ + try: + body = response.json() + except ValueError as exc: + raise StackResolveError("registry answered with something that is not JSON") from exc + if not isinstance(body, dict): + raise StackResolveError("registry answered with JSON that is not an object") + return body + + +class RegistryReader: + """Anonymous reader of one repository on an OCI registry. + + Every request is bounded by ``request_timeout_seconds`` and by what is left of + ``budget_seconds`` for the whole resolution, so a slow registry costs the budget + and no more. + """ + + def __init__( + self, + repository: str, + *, + request_timeout_seconds: float, + budget_seconds: float, + ) -> None: + """Prepare a reader for ``host/path``. No request is sent until it is used. + + Args: + repository: ``host/path``, e.g. ``ghcr.io/hyperi-io/dfe-stack-manifest``. + request_timeout_seconds: Longest one request may take. + budget_seconds: Longest the whole resolution may take. + """ + self._base, self._path = _split_repository(repository) + self._request_timeout = request_timeout_seconds + self._deadline = time.monotonic() + budget_seconds + self._token: str | None = None + # One attempt per request: the caller falls back, so a retry only spends budget. + self._client = HttpClient( + base_url=self._base, + timeout=request_timeout_seconds, + retries=1, + follow_redirects=True, + ) + + def __enter__(self) -> RegistryReader: + """Return the reader.""" + return self + + def __exit__(self, *_exc: object) -> None: + """Close the HTTP client.""" + self._client.close() + + def _remaining(self) -> float: + remaining = self._deadline - time.monotonic() + if remaining <= 0: + raise BudgetSpentError(_BUDGET_SPENT) + return min(self._request_timeout, remaining) + + def _authenticate(self, challenge: str) -> None: + """Fetch the anonymous bearer token the registry's own challenge names.""" + if not challenge.lower().startswith("bearer "): + raise StackResolveError("registry wants credentials that are not a bearer token") + params = dict(_CHALLENGE_PARAM.findall(challenge)) + realm = params.pop("realm", "") + scheme = self._base.partition(":")[0] + if not realm.startswith(f"{scheme}://"): + raise StackResolveError(f"registry token realm is not an {scheme} URL") + params.setdefault("scope", f"repository:{self._path}:pull") + body = _json_object(self._client.get(realm, params=params, timeout=self._remaining())) + token = body.get("token") or body.get("access_token") + if not isinstance(token, str) or not token: + raise StackResolveError("registry issued no anonymous token") + self._token = token + + def get(self, url: str, **kwargs: Any) -> httpx.Response: + """GET ``url`` anonymously, authenticating once if the registry asks. + + Raises: + StackResolveError: The request failed, was refused, or the budget is spent. + """ + base_headers = dict(kwargs.pop("headers", None) or {}) + try: + for _ in range(2): + headers = dict(base_headers) + if self._token: + headers["Authorization"] = f"Bearer {self._token}" + try: + return self._client.get( + url, timeout=self._remaining(), headers=headers, **kwargs + ) + except httpx.HTTPStatusError as exc: + challenge = exc.response.headers.get("www-authenticate", "") + if exc.response.status_code != 401 or self._token or not challenge: + raise + self._authenticate(challenge) + raise StackResolveError("registry kept refusing the anonymous token") + except httpx.HTTPError as exc: + raise StackResolveError(f"{type(exc).__name__} reading {urlsplit(url).path}") from exc + + def tags(self) -> list[str]: + """List every tag of the repository, following the registry's paging.""" + tags: list[str] = [] + url: str | None = f"/v2/{self._path}/tags/list?n={_TAGS_PAGE_SIZE}" + while url: + response = self.get(url) + tags.extend(str(tag) for tag in _json_object(response).get("tags") or []) + url = response.links.get("next", {}).get("url") + return tags + + def manifest_document(self, tag: str) -> Any: + """Fetch and parse the ``dfe-manifest-*.yaml`` layer of ``tag``. + + Raises: + StackResolveError: The artifact has no such layer, or the layer's bytes do + not hash to the digest its descriptor names. + """ + response = self.get(f"/v2/{self._path}/manifests/{tag}", headers={"Accept": _OCI_ACCEPT}) + layers = [ + item for item in _json_object(response).get("layers") or [] if isinstance(item, dict) + ] + layer = next((item for item in layers if _MANIFEST_LAYER.match(_layer_title(item))), None) + if layer is None: + raise StackResolveError(f"{tag} carries no dfe-manifest layer") + digest = str(layer.get("digest", "")) + blob = self.get(f"/v2/{self._path}/blobs/{digest}").content + if len(blob) > _MAX_MANIFEST_BYTES: + raise StackResolveError(f"{tag} manifest layer is over {_MAX_MANIFEST_BYTES} bytes") + if f"sha256:{hashlib.sha256(blob).hexdigest()}" != digest: + raise StackResolveError(f"{tag} manifest layer does not match its digest") + try: + return YAML(typ="safe").load(blob.decode("utf-8")) + except (YAMLError, UnicodeDecodeError) as exc: + raise StackResolveError(f"{tag} manifest layer is not valid YAML") from exc + + +def resolve_stack( + images: Sequence[str], + repository: str, + *, + request_timeout_seconds: float, + budget_seconds: float, + stop: threading.Event | None = None, +) -> str | None: + """Return the newest published stack whose digests agree with ``images``, or None. + + Blocking: it makes network calls, so a caller on the event loop runs it in a thread. + A tag whose artifact cannot be read is skipped. A registry that cannot be reached, + or a spent budget, ends the search. + + Args: + images: The ``ref@sha256:...`` of every DFE image the deployment runs. + repository: ``host/path`` of the stack-manifest repository. + request_timeout_seconds: Longest one registry request may take. + budget_seconds: Longest the whole resolution may take. + stop: When set, the search ends before the next tag and returns None. + + Returns: + The stack tag, or None when no published stack matches every image. + + Raises: + StackResolveError: The registry could not be read, or an image has no digest. + """ + running = running_digests(images) + with RegistryReader( + repository, + request_timeout_seconds=request_timeout_seconds, + budget_seconds=budget_seconds, + ) as reader: + for tag in newest_first(reader.tags()): + if stop is not None and stop.is_set(): + return None + try: + listed = manifest_images(reader.manifest_document(tag)) + except BudgetSpentError: + raise + except StackResolveError as exc: + logger.debug("stack manifest skipped", tag=tag, reason=str(exc)) + continue + if matches(listed, running): + return tag + return None + + +async def resolve_into(state: Any, settings: Any) -> None: + """Resolve the running stack off the event loop and store it on ``state``. + + Sets ``state.resolved_stack`` to the tag on a match. A failure, or no match, logs + once at warn and leaves the configured ``stack_version`` to answer. It never raises + for a registry problem, so neither startup nor readiness waits on it. + """ + images = list(settings.stack_images) + if not settings.stack_resolve or not images: + return + fallback = settings.stack_version or None + stop = threading.Event() + try: + tag = await asyncio.to_thread( + resolve_stack, + images, + settings.stack_manifest_repository, + request_timeout_seconds=settings.stack_resolve_request_timeout_seconds, + budget_seconds=settings.stack_resolve_budget_seconds, + stop=stop, + ) + except asyncio.CancelledError: + # The thread cannot be cancelled, so it is told to stop before its next tag. + stop.set() + raise + except StackResolveError as exc: + logger.warning( + "stack not resolved from the registry, keeping the configured label", + reason=str(exc), + fallback=fallback, + ) + return + if tag is None: + logger.warning( + "no published stack matches the running images, keeping the configured label", + repository=settings.stack_manifest_repository, + fallback=fallback, + ) + return + state.resolved_stack = tag + logger.info("stack resolved from the registry", stack=tag) diff --git a/tests/integration/test_stack_resolve_registry.py b/tests/integration/test_stack_resolve_registry.py new file mode 100644 index 00000000..39e8200a --- /dev/null +++ b/tests/integration/test_stack_resolve_registry.py @@ -0,0 +1,77 @@ +# Project: dfe-engine +# File: tests/integration/test_stack_resolve_registry.py +# Purpose: The stack resolver against the real published stack manifests +# Language: Python +# +# License: BUSL-1.1 +# Copyright: (c) 2026 HYPERI PTY LIMITED +"""The resolver reads the stack manifests dfe-infra really publishes to ghcr.io. + +The image references come from a published tag's own manifest, so the answer is checked +against the registry's current content and not a number written into the test. An +unreachable registry skips the test locally and fails it in CI, where a skip would be a +green build that resolved nothing. +""" + +import pytest + +from dfe_engine.stack_resolve import ( + RegistryReader, + StackResolveError, + manifest_images, + newest_first, + resolve_stack, +) +from tests.support.producer_contract import in_ci + +pytestmark = pytest.mark.integration + +REPOSITORY = "ghcr.io/hyperi-io/dfe-stack-manifest" +REQUEST_TIMEOUT_SECONDS = 10.0 +BUDGET_SECONDS = 120.0 + + +@pytest.fixture(scope="module") +def published() -> tuple[str, dict[str, str]]: + """The newest published release tag and its manifest's images, name to reference.""" + try: + with RegistryReader( + REPOSITORY, + request_timeout_seconds=REQUEST_TIMEOUT_SECONDS, + budget_seconds=BUDGET_SECONDS, + ) as reader: + tags = newest_first(reader.tags()) + releases = [tag for tag in tags if "-" not in tag] + assert releases, f"{REPOSITORY} publishes no release tag" + return releases[0], manifest_images(reader.manifest_document(releases[0])) + except StackResolveError as exc: + if in_ci(): + pytest.fail(f"{REPOSITORY} unreadable: {exc}") + pytest.skip(f"{REPOSITORY} unreadable: {exc}") + + +def _resolve(images: list[str]) -> str | None: + return resolve_stack( + images, + REPOSITORY, + request_timeout_seconds=REQUEST_TIMEOUT_SECONDS, + budget_seconds=BUDGET_SECONDS, + ) + + +def test_every_image_of_the_newest_release_resolves_to_that_release(published): + tag, images = published + assert len(images) > 1 + assert _resolve(list(images.values())) == tag + + +def test_one_image_of_the_newest_release_resolves_to_that_release(published): + tag, images = published + assert _resolve([images["dfe-engine"]]) == tag + + +def test_a_digest_no_stack_published_resolves_to_nothing(published): + _, images = published + name, reference = next(iter(images.items())) + unpublished = reference.split("@", 1)[0] + "@sha256:" + "0" * 64 + assert _resolve([unpublished]) is None, name diff --git a/tests/unit/test_api/test_system_version.py b/tests/unit/test_api/test_system_version.py index 13107cb6..aff931b5 100644 --- a/tests/unit/test_api/test_system_version.py +++ b/tests/unit/test_api/test_system_version.py @@ -99,6 +99,35 @@ def test_the_deployment_answers_when_the_deploy_repo_pins_nothing( assert body["stack"] == "2.2.0-rc.13" assert body["source"] == "deployment" + def test_the_registry_match_answers_in_place_of_the_baked_label( + self, client, app, admin_headers + ): + # The label a build wrote down said 2.2.2-rc.5, the images running are 2.2.2's. + app.state.settings.stack_version = "2.2.2-rc.5" + app.state.resolved_stack = "2.2.2" + + body = client.get("/api/v1/system/version", headers=admin_headers).json() + + assert body["stack"] == "2.2.2" + assert body["source"] == "registry" + + def test_the_registry_match_answers_with_no_baked_label(self, client, app, admin_headers): + app.state.resolved_stack = "2.2.2" + + body = client.get("/api/v1/system/deployment", headers=admin_headers).json() + + assert body["stack"] == "2.2.2" + assert body["source"] == "registry" + + def test_the_pins_win_over_the_registry_match(self, client, app, admin_headers, tmp_path): + _wire(app, tmp_path) + app.state.resolved_stack = "2.2.2" + + body = client.get("/api/v1/system/version", headers=admin_headers).json() + + assert body["stack"] == "2.2.0-rc.13" + assert body["source"] == "deploy-repo" + def test_the_pins_win_over_what_deployed_the_pod(self, client, app, admin_headers, tmp_path): # The pins are what the operator chose. The deployment value is only # what happened to stand this pod up. diff --git a/tests/unit/test_settings.py b/tests/unit/test_settings.py index 00f97a14..af0198e8 100644 --- a/tests/unit/test_settings.py +++ b/tests/unit/test_settings.py @@ -209,6 +209,25 @@ class TestEnvOverrides: ), ("DFE_ENGINE_PROFILE", " mesh ", "deployment.profile", "mesh"), ("DFE_ENGINE_UI_VERSION", "v1.5.1", "ui_version", "v1.5.1"), + ("DFE_ENGINE_STACK_RESOLVE", "false", "stack_resolve", False), + ( + "DFE_ENGINE_STACK_MANIFEST_REPOSITORY", + "registry.example/org/dfe-stack-manifest", + "stack_manifest_repository", + "registry.example/org/dfe-stack-manifest", + ), + ( + "DFE_ENGINE_STACK_RESOLVE_REQUEST_TIMEOUT_SECONDS", + "2.5", + "stack_resolve_request_timeout_seconds", + 2.5, + ), + ( + "DFE_ENGINE_STACK_RESOLVE_BUDGET_SECONDS", + "12", + "stack_resolve_budget_seconds", + 12.0, + ), ( "DFE_ENGINE_SERVICES_CONFIG_YAML_DIR", "/custom/svc", @@ -751,6 +770,14 @@ def _listed(settings: DFESettings, path: str, key: str | None) -> list: # Every list-typed setting load_settings() carries: its path, the env var that sets # it, the env value, the item key that identifies an entry, and what that value reads as. _LIST_SETTINGS = [ + pytest.param( + "stack_images", + "DFE_ENGINE_STACK_IMAGES", + "r.example/dfe-loader:1@sha256:aa, r.example/dfe-receiver:1@sha256:bb", + None, + ["r.example/dfe-loader:1@sha256:aa", "r.example/dfe-receiver:1@sha256:bb"], + id="stack_images", + ), pytest.param( "api.cors_origins", "DFE_ENGINE_API_CORS_ORIGINS", @@ -795,6 +822,7 @@ def _listed(settings: DFESettings, path: str, key: str | None) -> list: # The same fields as a config file sets them, plus admin_links, whose env form is a string. _FILE_LISTS = { + "stack_images": (["r.example/file:1@sha256:cc"], None, ["r.example/file:1@sha256:cc"]), "api.cors_origins": (["https://file.example"], None, ["https://file.example"]), "hunts.alert_channels": (["json://file.example"], None, ["json://file.example"]), "hunts.lookup_databases": (["intel"], None, ["intel"]), diff --git a/tests/unit/test_stack_resolve.py b/tests/unit/test_stack_resolve.py new file mode 100644 index 00000000..24d27d42 --- /dev/null +++ b/tests/unit/test_stack_resolve.py @@ -0,0 +1,492 @@ +# Project: dfe-engine +# File: tests/unit/test_stack_resolve.py +# Purpose: Resolving the running stack from published manifests and image digests +# Language: Python +# +# License: BUSL-1.1 +# Copyright: (c) 2026 HYPERI PTY LIMITED +"""The stack a deployment runs is the newest published stack whose digests match its images. + +The registry here is a real HTTP server on loopback that speaks the OCI distribution +calls the resolver makes, including the bearer-token challenge and paged tag lists. The +published artifact is the real shape: a manifest whose layer is a ``dfe-manifest-.yaml`` +file. The same resolver is held to the real ``ghcr.io`` registry in +``tests/integration/test_stack_resolve_registry.py``. +""" + +import hashlib +import json +import threading +import time +from collections.abc import Iterator +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from types import SimpleNamespace +from urllib.parse import parse_qs, urlsplit + +import pytest + +from dfe_engine.settings import DFESettings +from dfe_engine.stack_resolve import ( + StackResolveError, + image_name, + matches, + newest_first, + resolve_into, + resolve_stack, + running_digests, + semver_key, +) + +REPO_PATH = "org/dfe-stack-manifest" +LOADER = "dfe-loader" +RECEIVER = "dfe-receiver" +UI = "dfe-ui" +PAGE_SIZE = 2 + + +def digest_of(label: str) -> str: + return "sha256:" + hashlib.sha256(label.encode()).hexdigest() + + +def image_ref(app: str, label: str) -> str: + return f"registry.example/org/{app}:{label}@{digest_of(f'{app}-{label}')}" + + +def manifest_yaml(tag: str, apps: dict[str, str]) -> str: + """A stack manifest as dfe-infra renders it: apps by name, each pinned by digest.""" + lines = [f'stack_version: "{tag}"', "images:"] + lines += [f' {app}: "{image_ref(app, label)}"' for app, label in sorted(apps.items())] + return "\n".join(lines) + "\n" + + +def running(apps: dict[str, str]) -> list[str]: + return [image_ref(app, label) for app, label in apps.items()] + + +ALL_V1 = {LOADER: "v1", RECEIVER: "v1", UI: "v1"} +ALL_V2 = {LOADER: "v2", RECEIVER: "v2", UI: "v2"} + + +@dataclass +class FakeRegistry: + """A loopback OCI registry holding stack-manifest artifacts, counting what it serves.""" + + host: str = "" + stacks: dict[str, dict[str, str]] = field(default_factory=dict) + damage: dict[str, str] = field(default_factory=dict) + delay_seconds: float = 0.0 + token_requests: int = 0 + requests: int = 0 + + @property + def repository(self) -> str: + return f"{self.host}/{REPO_PATH}" + + def layer(self, tag: str) -> bytes: + if self.damage.get(tag) == "not-yaml": + return b"images: [unclosed\n - {" + return manifest_yaml(tag, self.stacks[tag]).encode() + + def manifest(self, tag: str) -> dict: + blob = self.layer(tag) + title = "readme.txt" if self.damage.get(tag) == "no-layer" else f"dfe-manifest-{tag}.yaml" + digest = "sha256:" + hashlib.sha256(blob).hexdigest() + return { + "schemaVersion": 2, + "layers": [ + { + "digest": digest, + "size": len(blob), + "annotations": {"org.opencontainers.image.title": title}, + } + ], + } + + def blob(self, digest: str) -> bytes | None: + for tag in self.stacks: + real = self.layer(tag) + if "sha256:" + hashlib.sha256(real).hexdigest() == digest: + return b"tampered: true\n" if self.damage.get(tag) == "tampered" else real + return None + + +def _handler(registry: FakeRegistry) -> type[BaseHTTPRequestHandler]: + class Handler(BaseHTTPRequestHandler): + def log_message(self, format: str, *args: object) -> None: + pass + + def _send(self, status: int, body: bytes, headers: dict[str, str] | None = None) -> None: + self.send_response(status) + for name, value in (headers or {}).items(): + self.send_header(name, value) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: + registry.requests += 1 + time.sleep(registry.delay_seconds) + parts = urlsplit(self.path) + if parts.path == "/token": + registry.token_requests += 1 + self._send(200, json.dumps({"token": "anonymous"}).encode()) + return + if self.headers.get("Authorization") != "Bearer anonymous": + challenge = ( + f'Bearer realm="http://{registry.host}/token",service="test",' + f'scope="repository:{REPO_PATH}:pull"' + ) + self._send(401, b"{}", {"WWW-Authenticate": challenge}) + return + prefix = f"/v2/{REPO_PATH}/" + if not parts.path.startswith(prefix): + self._send(404, b"{}") + return + tail = parts.path.removeprefix(prefix) + if tail == "tags/list": + self._tags(parse_qs(parts.query)) + elif ( + tail.startswith("manifests/") and tail.removeprefix("manifests/") in registry.stacks + ): + tag = tail.removeprefix("manifests/") + answer = ( + [1, 2] + if registry.damage.get(tag) == "manifest-array" + else registry.manifest(tag) + ) + body = json.dumps(answer) + self._send(200, body.encode(), {"Content-Type": "application/json"}) + elif tail.startswith("blobs/") and registry.blob(tail.removeprefix("blobs/")): + self._send(200, registry.blob(tail.removeprefix("blobs/")) or b"") + else: + self._send(404, b"{}") + + def _tags(self, query: dict[str, list[str]]) -> None: + tags = sorted(registry.stacks) + last = query.get("last", [""])[0] + remaining = [tag for tag in tags if tag > last] if last else tags + page, more = remaining[:PAGE_SIZE], len(remaining) > PAGE_SIZE + headers = {} + if more: + headers["Link"] = f'; rel="next"' + self._send(200, json.dumps({"name": REPO_PATH, "tags": page}).encode(), headers) + + return Handler + + +@pytest.fixture +def registry() -> Iterator[FakeRegistry]: + fake = FakeRegistry() + server = ThreadingHTTPServer(("127.0.0.1", 0), _handler(fake)) + fake.host = f"127.0.0.1:{server.server_address[1]}" + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield fake + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def resolve(registry: FakeRegistry, apps: dict[str, str], **limits: float) -> str | None: + return resolve_stack( + running(apps), + registry.repository, + request_timeout_seconds=limits.get("request", 5.0), + budget_seconds=limits.get("budget", 30.0), + ) + + +@pytest.mark.parametrize( + ("tags", "expected"), + [ + pytest.param( + ["2.2.1", "2.2.2-rc.5", "2.2.2", "2.2.0", "2.2.2-rc.8"], + ["2.2.2", "2.2.1", "2.2.0", "2.2.2-rc.8", "2.2.2-rc.5"], + id="releases_then_prereleases", + ), + pytest.param( + ["2.2.2-rc.9", "2.2.2-rc.10", "2.2.2-rc.2"], + ["2.2.2-rc.10", "2.2.2-rc.9", "2.2.2-rc.2"], + id="prerelease_numbers_compare_as_numbers", + ), + pytest.param( + ["2.2.2-rc.1", "2.2.2-beta.1", "2.2.2-alpha.3"], + ["2.2.2-rc.1", "2.2.2-beta.1", "2.2.2-alpha.3"], + id="alphanumeric_identifiers_compare_lexically", + ), + pytest.param( + ["2.10.0", "2.9.0", "10.0.0"], + ["10.0.0", "2.10.0", "2.9.0"], + id="versions_compare_as_numbers", + ), + pytest.param( + ["latest", "v2.2.1", "sha-abc123", "2.2", "2.2.2"], + ["2.2.2", "v2.2.1"], + id="tags_that_are_not_semver_are_dropped", + ), + pytest.param([], [], id="no_tags"), + ], +) +def test_newest_first(tags, expected): + assert newest_first(tags) == expected + + +def test_a_release_outranks_its_own_prerelease_and_not_the_next_patch(): + candidate, release, next_patch = ( + semver_key(tag) for tag in ("2.2.2-rc.9", "2.2.2", "2.2.3-rc.1") + ) + assert candidate + assert release + assert next_patch + assert candidate < release < next_patch + assert semver_key("not-a-version") is None + + +@pytest.mark.parametrize( + ("reference", "name"), + [ + ("ghcr.io/hyperi-io/dfe-loader:v1.2.3@sha256:ab", "dfe-loader"), + ("ghcr.io/hyperi-io/dfe-loader@sha256:ab", "dfe-loader"), + ("dfe-loader:v1", "dfe-loader"), + ("localhost:5000/org/dfe-loader:v1@sha256:ab", "dfe-loader"), + ], +) +def test_image_name(reference, name): + assert image_name(reference) == name + + +@pytest.mark.parametrize( + ("manifest", "run", "expected"), + [ + pytest.param(ALL_V1, ALL_V1, True, id="same_apps_same_digests"), + pytest.param(ALL_V1, {LOADER: "v1"}, True, id="deployment_runs_a_subset"), + pytest.param(ALL_V1, {LOADER: "v2"}, False, id="digest_differs"), + pytest.param(ALL_V1, {LOADER: "v1", RECEIVER: "v2"}, False, id="one_of_two_differs"), + pytest.param({LOADER: "v1"}, {LOADER: "v1", UI: "v1"}, False, id="app_the_stack_lacks"), + pytest.param(ALL_V1, {}, False, id="nothing_running"), + ], +) +def test_matches(manifest, run, expected): + listed = {app: image_ref(app, label) for app, label in manifest.items()} + assert matches(listed, running_digests(running(run))) is expected + + +@pytest.mark.parametrize( + ("references", "refusal"), + [ + pytest.param(["ghcr.io/hyperi-io/dfe-loader:v1"], "pins no sha256 digest", id="no_digest"), + pytest.param(["ghcr.io/hyperi-io/dfe-loader@sha512:ab"], "pins no sha256", id="wrong_algo"), + pytest.param( + ["r/dfe-loader:v1@sha256:aa", "r/dfe-loader:v2@sha256:bb"], + "two different digests", + id="one_image_two_digests", + ), + ], +) +def test_running_images_must_pin_one_digest_each(references, refusal): + with pytest.raises(StackResolveError, match=refusal): + running_digests(references) + + +@pytest.mark.parametrize( + ("stacks", "damage", "run", "expected"), + [ + pytest.param( + {"2.2.2-rc.6": ALL_V1, "2.2.2": ALL_V1, "2.2.2-rc.5": ALL_V1}, + {}, + ALL_V1, + "2.2.2", + id="a_release_wins_over_prereleases_with_the_same_images", + ), + pytest.param( + {"2.2.3": ALL_V2, "2.2.2": ALL_V1}, + {}, + ALL_V1, + "2.2.2", + id="an_older_release_matches_when_the_newest_differs", + ), + pytest.param( + {"2.2.3-rc.1": ALL_V1, "2.2.2": ALL_V2}, + {}, + ALL_V1, + "2.2.3-rc.1", + id="a_prerelease_answers_when_no_release_matches", + ), + pytest.param( + {"2.2.2-rc.9": ALL_V1, "2.2.2-rc.10": ALL_V1}, + {}, + ALL_V1, + "2.2.2-rc.10", + id="prereleases_order_by_number", + ), + pytest.param( + {"2.2.2": ALL_V1}, + {}, + {LOADER: "v1"}, + "2.2.2", + id="a_subset_of_the_apps_matches", + ), + pytest.param( + {"2.2.1": ALL_V1, "2.2.2": {**ALL_V2, LOADER: "v1"}}, + {}, + {LOADER: "v1"}, + "2.2.2", + id="stacks_sharing_the_running_subset_resolve_to_the_newest", + ), + pytest.param( + {"2.2.2": ALL_V2}, + {}, + ALL_V1, + None, + id="no_stack_has_the_digests", + ), + pytest.param( + {"2.2.2": {LOADER: "v1"}}, + {}, + ALL_V1, + None, + id="no_stack_lists_every_running_app", + ), + pytest.param({"latest": ALL_V1, "main": ALL_V1}, {}, ALL_V1, None, id="no_semver_tags"), + pytest.param({}, {}, ALL_V1, None, id="an_empty_repository"), + pytest.param( + {"2.2.3": ALL_V1, "2.2.2": ALL_V1}, + {"2.2.3": "no-layer"}, + ALL_V1, + "2.2.2", + id="an_artifact_with_no_manifest_layer_is_skipped", + ), + pytest.param( + {"2.2.3": ALL_V1, "2.2.2": ALL_V1}, + {"2.2.3": "tampered"}, + ALL_V1, + "2.2.2", + id="a_layer_that_does_not_hash_to_its_digest_is_skipped", + ), + pytest.param( + {"2.2.3": ALL_V1, "2.2.2": ALL_V1}, + {"2.2.3": "manifest-array"}, + ALL_V1, + "2.2.2", + id="an_artifact_manifest_that_is_not_an_object_is_skipped", + ), + pytest.param( + {"2.2.3": ALL_V1, "2.2.2": ALL_V1}, + {"2.2.3": "not-yaml"}, + ALL_V1, + "2.2.2", + id="a_layer_that_is_not_yaml_is_skipped", + ), + ], +) +def test_resolve_stack(registry, stacks, damage, run, expected): + registry.stacks = stacks + registry.damage = damage + assert resolve(registry, run) == expected + + +def test_one_anonymous_token_serves_every_request_and_every_page(registry): + registry.stacks = {f"2.2.{patch}": ALL_V2 for patch in range(5)} | {"2.1.0": ALL_V1} + assert resolve(registry, ALL_V1) == "2.1.0" + assert registry.token_requests == 1 + + +@pytest.mark.parametrize( + ("repository", "run", "refusal"), + [ + pytest.param("org/dfe-stack-manifest", ALL_V1, "is not host/path", id="no_host"), + pytest.param("registry.example", ALL_V1, "is not host/path", id="no_path"), + pytest.param("{host}/org/other", ALL_V1, "HTTPStatusError", id="repository_not_found"), + pytest.param("127.0.0.1:1/org/x", ALL_V1, "ConnectError", id="registry_unreachable"), + ], +) +def test_a_registry_that_cannot_be_read_stops_the_search(registry, repository, run, refusal): + registry.stacks = {"2.2.2": ALL_V1} + target = repository.format(host=registry.host) + with pytest.raises(StackResolveError, match=refusal): + resolve_stack(running(run), target, request_timeout_seconds=2.0, budget_seconds=10.0) + + +def test_a_slow_registry_costs_the_budget_and_no_more(registry): + registry.stacks = {"2.2.2": ALL_V1} + registry.delay_seconds = 0.5 + started = time.monotonic() + with pytest.raises(StackResolveError): + resolve(registry, ALL_V1, request=0.2, budget=0.4) + assert time.monotonic() - started < 3 + + +def test_a_stop_request_ends_the_search_before_the_next_tag(registry): + registry.stacks = {"2.2.3": ALL_V2, "2.2.2": ALL_V1} + stop = threading.Event() + stop.set() + found = resolve_stack( + running(ALL_V1), + registry.repository, + request_timeout_seconds=5.0, + budget_seconds=30.0, + stop=stop, + ) + assert found is None + + +def settings_for(registry: FakeRegistry, run: dict[str, str] | None, **extra) -> DFESettings: + return DFESettings( + env="dev", + stack_images=running(run) if run else [], + stack_manifest_repository=registry.repository, + stack_version="2.2.2-rc.5", + **extra, + ) + + +@pytest.mark.parametrize( + ("run", "extra", "stacks", "expected", "calls_registry"), + [ + pytest.param(ALL_V1, {}, {"2.2.2": ALL_V1}, "2.2.2", True, id="resolved"), + pytest.param(ALL_V1, {}, {"2.2.2": ALL_V2}, None, True, id="no_match_leaves_the_label"), + pytest.param(None, {}, {"2.2.2": ALL_V1}, None, False, id="no_images_skips_resolution"), + pytest.param( + ALL_V1, + {"stack_resolve": False}, + {"2.2.2": ALL_V1}, + None, + False, + id="switched_off_for_an_air_gapped_site", + ), + ], +) +async def test_resolve_into(registry, run, extra, stacks, expected, calls_registry): + registry.stacks = stacks + state = SimpleNamespace(resolved_stack=None) + + await resolve_into(state, settings_for(registry, run, **extra)) + + assert state.resolved_stack == expected + assert (registry.requests > 0) is calls_registry + + +@pytest.mark.parametrize( + "stack_images", + [ + pytest.param(["registry.example/org/dfe-loader:v1"], id="an_image_with_no_digest"), + pytest.param([image_ref(LOADER, "v1")], id="a_registry_that_is_down"), + ], +) +async def test_a_failure_leaves_the_label_and_never_raises(stack_images): + state = SimpleNamespace(resolved_stack=None) + settings = DFESettings( + env="dev", + stack_images=stack_images, + stack_manifest_repository="127.0.0.1:1/org/dfe-stack-manifest", + stack_version="2.2.2-rc.5", + stack_resolve_request_timeout_seconds=1.0, + stack_resolve_budget_seconds=3.0, + ) + + await resolve_into(state, settings) + + assert state.resolved_stack is None From 66f8ddc2a9707c282793bf8b583918f850e4e752 Mon Sep 17 00:00:00 2001 From: Derek Date: Mon, 16 Nov 2026 05:44:47 +1100 Subject: [PATCH 2/3] fix(system): log stack resolution failures and skip pinned stacks The stack resolution task caught only registry errors. Other exceptions ended it with no log. - Each failure except cancellation logs one warning with its reason. The stack_version label is kept. Cancellation passes through. - A non-list tag response is rejected as a registry error. - A deploy repo's stack pin takes precedence, so registry reads are skipped. The pin is read the same way as in deployment facts. - Tests cover startup with a stalled registry, task cancellation at shutdown, and skipping registry reads for pinned stacks. --- src/dfe_engine/api/app.py | 12 +- src/dfe_engine/stack_resolve.py | 33 ++- tests/support/oci_registry.py | 178 +++++++++++++ .../test_api/test_startup_stack_resolution.py | 137 ++++++++++ tests/unit/test_stack_resolve.py | 243 +++++++----------- 5 files changed, 443 insertions(+), 160 deletions(-) create mode 100644 tests/support/oci_registry.py create mode 100644 tests/unit/test_api/test_startup_stack_resolution.py diff --git a/src/dfe_engine/api/app.py b/src/dfe_engine/api/app.py index 036a0974..55d53c5a 100644 --- a/src/dfe_engine/api/app.py +++ b/src/dfe_engine/api/app.py @@ -657,11 +657,21 @@ def reconcile_service_roles_now() -> ReconcileResult: ) # In the background: the registry is the internet's, so startup and readiness never wait on it. + from dfe_engine.gitops.pins import load_pins, stack_version from dfe_engine.stack_resolve import resolve_into app.state.resolved_stack = None + read_pin = None + if gitcrud is not None: + deploy_repo = gitcrud + + def read_pin() -> str | None: + # The pins are read as deployment_facts reads them, and they win over the registry. + with deploy_repo.reading(): + return stack_version(load_pins(deploy_repo.repo_path)) + app.state.stack_resolution = asyncio.create_task( - resolve_into(app.state, settings), name="stack-resolution" + resolve_into(app.state, settings, read_pin), name="stack-resolution" ) health.set_started() diff --git a/src/dfe_engine/stack_resolve.py b/src/dfe_engine/stack_resolve.py index 398f5bc4..d1c55ea1 100644 --- a/src/dfe_engine/stack_resolve.py +++ b/src/dfe_engine/stack_resolve.py @@ -22,7 +22,7 @@ import re import threading import time -from collections.abc import Mapping, Sequence +from collections.abc import Callable, Mapping, Sequence from typing import Any from urllib.parse import urlsplit @@ -276,7 +276,10 @@ def tags(self) -> list[str]: url: str | None = f"/v2/{self._path}/tags/list?n={_TAGS_PAGE_SIZE}" while url: response = self.get(url) - tags.extend(str(tag) for tag in _json_object(response).get("tags") or []) + page = _json_object(response).get("tags") or [] + if not isinstance(page, list): + raise StackResolveError("registry tag list is not a list") + tags.extend(str(tag) for tag in page) url = response.links.get("next", {}).get("url") return tags @@ -354,12 +357,21 @@ def resolve_stack( return None -async def resolve_into(state: Any, settings: Any) -> None: +async def resolve_into( + state: Any, settings: Any, deploy_repo_stack: Callable[[], str | None] | None = None +) -> None: """Resolve the running stack off the event loop and store it on ``state``. - Sets ``state.resolved_stack`` to the tag on a match. A failure, or no match, logs - once at warn and leaves the configured ``stack_version`` to answer. It never raises - for a registry problem, so neither startup nor readiness waits on it. + Sets ``state.resolved_stack`` to the tag on a match. A failure of any kind, or no + match, logs once at warn and leaves the configured ``stack_version`` to answer. It + raises nothing but cancellation, so neither startup nor readiness waits on it and a + dead task never goes unremarked. + + Args: + state: Where ``resolved_stack`` is stored. + settings: The engine settings carrying the ``stack_*`` fields. + deploy_repo_stack: Reads the stack the deploy repo pins, blocking. A pin always + wins over the registry, so when one exists the registry is not read. """ images = list(settings.stack_images) if not settings.stack_resolve or not images: @@ -367,6 +379,9 @@ async def resolve_into(state: Any, settings: Any) -> None: fallback = settings.stack_version or None stop = threading.Event() try: + if deploy_repo_stack is not None and await asyncio.to_thread(deploy_repo_stack): + logger.debug("deploy repo pins the stack, registry not read") + return tag = await asyncio.to_thread( resolve_stack, images, @@ -379,10 +394,12 @@ async def resolve_into(state: Any, settings: Any) -> None: # The thread cannot be cancelled, so it is told to stop before its next tag. stop.set() raise - except StackResolveError as exc: + except Exception as exc: + # Whatever ends this task is logged here, since nothing awaits it before shutdown. + reason = str(exc) if isinstance(exc, StackResolveError) else f"{type(exc).__name__}: {exc}" logger.warning( "stack not resolved from the registry, keeping the configured label", - reason=str(exc), + reason=reason, fallback=fallback, ) return diff --git a/tests/support/oci_registry.py b/tests/support/oci_registry.py new file mode 100644 index 00000000..2175a916 --- /dev/null +++ b/tests/support/oci_registry.py @@ -0,0 +1,178 @@ +# Project: dfe-engine +# File: tests/support/oci_registry.py +# Purpose: A loopback OCI registry serving dfe-stack-manifest artifacts +# Language: Python +# +# License: BUSL-1.1 +# Copyright: (c) 2026 HYPERI PTY LIMITED +"""A real HTTP server speaking the registry calls the stack resolver makes. + +It serves the bearer-token challenge, paged tag lists, artifact manifests and layer blobs +for stacks a test defines, and can damage an artifact or slow every answer. +""" + +import hashlib +import json +import threading +import time +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlsplit + +REPO_PATH = "org/dfe-stack-manifest" +LOADER = "dfe-loader" +RECEIVER = "dfe-receiver" +UI = "dfe-ui" +PAGE_SIZE = 2 + + +def digest_of(label: str) -> str: + return "sha256:" + hashlib.sha256(label.encode()).hexdigest() + + +def image_ref(app: str, label: str) -> str: + return f"registry.example/org/{app}:{label}@{digest_of(f'{app}-{label}')}" + + +def manifest_yaml(tag: str, apps: dict[str, str]) -> str: + """A stack manifest as dfe-infra renders it: apps by name, each pinned by digest.""" + lines = [f'stack_version: "{tag}"', "images:"] + lines += [f' {app}: "{image_ref(app, label)}"' for app, label in sorted(apps.items())] + return "\n".join(lines) + "\n" + + +def running(apps: dict[str, str]) -> list[str]: + return [image_ref(app, label) for app, label in apps.items()] + + +ALL_V1 = {LOADER: "v1", RECEIVER: "v1", UI: "v1"} +ALL_V2 = {LOADER: "v2", RECEIVER: "v2", UI: "v2"} + + +@dataclass +class FakeRegistry: + """A loopback OCI registry holding stack-manifest artifacts, counting what it serves.""" + + host: str = "" + stacks: dict[str, dict[str, str]] = field(default_factory=dict) + damage: dict[str, str] = field(default_factory=dict) + delay_seconds: float = 0.0 + tags_answer: object = None + token_requests: int = 0 + requests: int = 0 + + @property + def repository(self) -> str: + return f"{self.host}/{REPO_PATH}" + + def layer(self, tag: str) -> bytes: + if self.damage.get(tag) == "not-yaml": + return b"images: [unclosed\n - {" + return manifest_yaml(tag, self.stacks[tag]).encode() + + def manifest(self, tag: str) -> dict: + blob = self.layer(tag) + title = "readme.txt" if self.damage.get(tag) == "no-layer" else f"dfe-manifest-{tag}.yaml" + digest = "sha256:" + hashlib.sha256(blob).hexdigest() + return { + "schemaVersion": 2, + "layers": [ + { + "digest": digest, + "size": len(blob), + "annotations": {"org.opencontainers.image.title": title}, + } + ], + } + + def blob(self, digest: str) -> bytes | None: + for tag in self.stacks: + real = self.layer(tag) + if "sha256:" + hashlib.sha256(real).hexdigest() == digest: + return b"tampered: true\n" if self.damage.get(tag) == "tampered" else real + return None + + +def _handler(registry: FakeRegistry) -> type[BaseHTTPRequestHandler]: + class Handler(BaseHTTPRequestHandler): + def log_message(self, format: str, *args: object) -> None: + pass + + def _send(self, status: int, body: bytes, headers: dict[str, str] | None = None) -> None: + self.send_response(status) + for name, value in (headers or {}).items(): + self.send_header(name, value) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: + registry.requests += 1 + time.sleep(registry.delay_seconds) + parts = urlsplit(self.path) + if parts.path == "/token": + registry.token_requests += 1 + self._send(200, json.dumps({"token": "anonymous"}).encode()) + return + if self.headers.get("Authorization") != "Bearer anonymous": + challenge = ( + f'Bearer realm="http://{registry.host}/token",service="test",' + f'scope="repository:{REPO_PATH}:pull"' + ) + self._send(401, b"{}", {"WWW-Authenticate": challenge}) + return + prefix = f"/v2/{REPO_PATH}/" + if not parts.path.startswith(prefix): + self._send(404, b"{}") + return + tail = parts.path.removeprefix(prefix) + if tail == "tags/list": + self._tags(parse_qs(parts.query)) + elif ( + tail.startswith("manifests/") and tail.removeprefix("manifests/") in registry.stacks + ): + tag = tail.removeprefix("manifests/") + answer = ( + [1, 2] + if registry.damage.get(tag) == "manifest-array" + else registry.manifest(tag) + ) + body = json.dumps(answer) + self._send(200, body.encode(), {"Content-Type": "application/json"}) + elif tail.startswith("blobs/") and registry.blob(tail.removeprefix("blobs/")): + self._send(200, registry.blob(tail.removeprefix("blobs/")) or b"") + else: + self._send(404, b"{}") + + def _tags(self, query: dict[str, list[str]]) -> None: + if registry.tags_answer is not None: + self._send(200, json.dumps(registry.tags_answer).encode()) + return + tags = sorted(registry.stacks) + last = query.get("last", [""])[0] + remaining = [tag for tag in tags if tag > last] if last else tags + page, more = remaining[:PAGE_SIZE], len(remaining) > PAGE_SIZE + headers = {} + if more: + headers["Link"] = f'; rel="next"' + self._send(200, json.dumps({"name": REPO_PATH, "tags": page}).encode(), headers) + + return Handler + + +@contextmanager +def serve_registry() -> Iterator[FakeRegistry]: + """Run a registry on a loopback port for the length of the block.""" + fake = FakeRegistry() + server = ThreadingHTTPServer(("127.0.0.1", 0), _handler(fake)) + fake.host = f"127.0.0.1:{server.server_address[1]}" + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield fake + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) diff --git a/tests/unit/test_api/test_startup_stack_resolution.py b/tests/unit/test_api/test_startup_stack_resolution.py new file mode 100644 index 00000000..ff8d1bed --- /dev/null +++ b/tests/unit/test_api/test_startup_stack_resolution.py @@ -0,0 +1,137 @@ +# Project: dfe-engine +# File: tests/unit/test_api/test_startup_stack_resolution.py +# Purpose: The stack resolution never holds the lifespan, and a deploy-repo pin skips it +# Language: Python +# +# License: BUSL-1.1 +# Copyright: (c) 2026 HYPERI PTY LIMITED +"""Startup resolves the running stack in the background, never before it serves. + +The registry is a real loopback HTTP server. A registry that stalls must not hold the +lifespan, shutdown must cancel the resolution cleanly, and a stack the deploy repo pins +is never looked up because the pin wins. +""" + +import secrets +import time +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from dfe_engine.api.app import create_app +from dfe_engine.api.deps import _registries +from dfe_engine.settings import ( + APISettings, + AuthSettings, + ClickHouseSettings, + DFESettings, + GitopsPackSettings, + GitopsSettings, + HuntsSettings, + LocalAuthSettings, + SchemasSettings, + SecretsSettings, + ServicesSettings, + SourceSettings, +) +from tests.support.oci_registry import ALL_V1, FakeRegistry, running, serve_registry + +STALL_SECONDS = 3.0 +PINS = 'base:\n dfe-infra: "2.2.0-rc.13"\n' + + +@pytest.fixture +def registry(): + with serve_registry() as fake: + yield fake + + +def _settings(tmp_path: Path, registry: FakeRegistry) -> DFESettings: + """A deployment with a deploy repo that runs ALL_V1's images.""" + for sub in ("sources", "services", "rules", "hunts", "auth", "schemas", "secrets", "orgs"): + (tmp_path / sub).mkdir(exist_ok=True) + return DFESettings( + env="dev", + config_dir=str(tmp_path), + clickhouse=ClickHouseSettings(bootstrap_tables=False), + source=SourceSettings(sources_dir=str(tmp_path / "sources")), + services=ServicesSettings(config_yaml_dir=str(tmp_path / "services")), + schemas=SchemasSettings(schemas_dir=str(tmp_path / "schemas")), + hunts=HuntsSettings(rules_dir=str(tmp_path / "rules"), hunt_dir=str(tmp_path / "hunts")), + auth=AuthSettings( + enabled=True, + auth_dir=str(tmp_path / "auth"), + local=LocalAuthSettings(enabled=True, admin_password=secrets.token_urlsafe(16)), + ), + secrets=SecretsSettings(provider="file", path=str(tmp_path / "secrets")), + api=APISettings(jwt_secret="test-secret-key-for-unit-tests-hmac32"), + gitops=GitopsSettings( + enabled=True, + local_path=str(tmp_path / "deploy"), + push=False, + mode="solo", + pack=GitopsPackSettings(interval_seconds=3600.0), + ), + stack_images=running(ALL_V1), + stack_manifest_repository=registry.repository, + stack_version="2.2.2-rc.5", + ) + + +def _await(task, what: str) -> None: + deadline = time.monotonic() + 30 + while not task.done(): + assert time.monotonic() < deadline, what + time.sleep(0.1) + + +def test_a_stalled_registry_never_holds_startup_and_shutdown_cancels_the_resolution( + tmp_path: Path, registry: FakeRegistry +) -> None: + registry.stacks = {"2.2.2": ALL_V1} + registry.delay_seconds = STALL_SECONDS + app = create_app(settings=_settings(tmp_path, registry)) + try: + began = time.monotonic() + with TestClient(app, raise_server_exceptions=False) as client: + assert time.monotonic() - began < STALL_SECONDS + assert client.get("/livez").status_code == 200 + resolution = app.state.stack_resolution + assert not resolution.done() + assert app.state.resolved_stack is None + assert resolution.cancelled() + finally: + _registries.clear() + + +def test_the_registry_answers_once_startup_is_done(tmp_path: Path, registry: FakeRegistry) -> None: + registry.stacks = {"2.2.2": ALL_V1} + app = create_app(settings=_settings(tmp_path, registry)) + try: + with TestClient(app, raise_server_exceptions=False): + _await(app.state.stack_resolution, "the stack resolution never finished") + assert app.state.resolved_stack == "2.2.2" + finally: + _registries.clear() + + +def test_a_stack_the_deploy_repo_pins_is_never_looked_up( + tmp_path: Path, registry: FakeRegistry +) -> None: + registry.stacks = {"2.2.2": ALL_V1} + settings = _settings(tmp_path, registry) + # A first start creates the deploy repo, and the pin lands in its working tree as a publish leaves it. + app = create_app(settings=settings) + try: + with TestClient(app, raise_server_exceptions=False): + _await(app.state.stack_resolution, "the first resolution never finished") + (tmp_path / "deploy" / "pins.yaml").write_text(PINS) + registry.requests = 0 + app = create_app(settings=settings) + with TestClient(app, raise_server_exceptions=False): + _await(app.state.stack_resolution, "the pinned resolution never finished") + assert app.state.resolved_stack is None + assert registry.requests == 0 + finally: + _registries.clear() diff --git a/tests/unit/test_stack_resolve.py b/tests/unit/test_stack_resolve.py index 24d27d42..f81b6865 100644 --- a/tests/unit/test_stack_resolve.py +++ b/tests/unit/test_stack_resolve.py @@ -14,15 +14,10 @@ ``tests/integration/test_stack_resolve_registry.py``. """ -import hashlib -import json import threading import time from collections.abc import Iterator -from dataclasses import dataclass, field -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from types import SimpleNamespace -from urllib.parse import parse_qs, urlsplit import pytest @@ -37,157 +32,23 @@ running_digests, semver_key, ) - -REPO_PATH = "org/dfe-stack-manifest" -LOADER = "dfe-loader" -RECEIVER = "dfe-receiver" -UI = "dfe-ui" -PAGE_SIZE = 2 - - -def digest_of(label: str) -> str: - return "sha256:" + hashlib.sha256(label.encode()).hexdigest() - - -def image_ref(app: str, label: str) -> str: - return f"registry.example/org/{app}:{label}@{digest_of(f'{app}-{label}')}" - - -def manifest_yaml(tag: str, apps: dict[str, str]) -> str: - """A stack manifest as dfe-infra renders it: apps by name, each pinned by digest.""" - lines = [f'stack_version: "{tag}"', "images:"] - lines += [f' {app}: "{image_ref(app, label)}"' for app, label in sorted(apps.items())] - return "\n".join(lines) + "\n" - - -def running(apps: dict[str, str]) -> list[str]: - return [image_ref(app, label) for app, label in apps.items()] - - -ALL_V1 = {LOADER: "v1", RECEIVER: "v1", UI: "v1"} -ALL_V2 = {LOADER: "v2", RECEIVER: "v2", UI: "v2"} - - -@dataclass -class FakeRegistry: - """A loopback OCI registry holding stack-manifest artifacts, counting what it serves.""" - - host: str = "" - stacks: dict[str, dict[str, str]] = field(default_factory=dict) - damage: dict[str, str] = field(default_factory=dict) - delay_seconds: float = 0.0 - token_requests: int = 0 - requests: int = 0 - - @property - def repository(self) -> str: - return f"{self.host}/{REPO_PATH}" - - def layer(self, tag: str) -> bytes: - if self.damage.get(tag) == "not-yaml": - return b"images: [unclosed\n - {" - return manifest_yaml(tag, self.stacks[tag]).encode() - - def manifest(self, tag: str) -> dict: - blob = self.layer(tag) - title = "readme.txt" if self.damage.get(tag) == "no-layer" else f"dfe-manifest-{tag}.yaml" - digest = "sha256:" + hashlib.sha256(blob).hexdigest() - return { - "schemaVersion": 2, - "layers": [ - { - "digest": digest, - "size": len(blob), - "annotations": {"org.opencontainers.image.title": title}, - } - ], - } - - def blob(self, digest: str) -> bytes | None: - for tag in self.stacks: - real = self.layer(tag) - if "sha256:" + hashlib.sha256(real).hexdigest() == digest: - return b"tampered: true\n" if self.damage.get(tag) == "tampered" else real - return None - - -def _handler(registry: FakeRegistry) -> type[BaseHTTPRequestHandler]: - class Handler(BaseHTTPRequestHandler): - def log_message(self, format: str, *args: object) -> None: - pass - - def _send(self, status: int, body: bytes, headers: dict[str, str] | None = None) -> None: - self.send_response(status) - for name, value in (headers or {}).items(): - self.send_header(name, value) - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - - def do_GET(self) -> None: - registry.requests += 1 - time.sleep(registry.delay_seconds) - parts = urlsplit(self.path) - if parts.path == "/token": - registry.token_requests += 1 - self._send(200, json.dumps({"token": "anonymous"}).encode()) - return - if self.headers.get("Authorization") != "Bearer anonymous": - challenge = ( - f'Bearer realm="http://{registry.host}/token",service="test",' - f'scope="repository:{REPO_PATH}:pull"' - ) - self._send(401, b"{}", {"WWW-Authenticate": challenge}) - return - prefix = f"/v2/{REPO_PATH}/" - if not parts.path.startswith(prefix): - self._send(404, b"{}") - return - tail = parts.path.removeprefix(prefix) - if tail == "tags/list": - self._tags(parse_qs(parts.query)) - elif ( - tail.startswith("manifests/") and tail.removeprefix("manifests/") in registry.stacks - ): - tag = tail.removeprefix("manifests/") - answer = ( - [1, 2] - if registry.damage.get(tag) == "manifest-array" - else registry.manifest(tag) - ) - body = json.dumps(answer) - self._send(200, body.encode(), {"Content-Type": "application/json"}) - elif tail.startswith("blobs/") and registry.blob(tail.removeprefix("blobs/")): - self._send(200, registry.blob(tail.removeprefix("blobs/")) or b"") - else: - self._send(404, b"{}") - - def _tags(self, query: dict[str, list[str]]) -> None: - tags = sorted(registry.stacks) - last = query.get("last", [""])[0] - remaining = [tag for tag in tags if tag > last] if last else tags - page, more = remaining[:PAGE_SIZE], len(remaining) > PAGE_SIZE - headers = {} - if more: - headers["Link"] = f'; rel="next"' - self._send(200, json.dumps({"name": REPO_PATH, "tags": page}).encode(), headers) - - return Handler +from tests.support.oci_registry import ( + ALL_V1, + ALL_V2, + LOADER, + RECEIVER, + UI, + FakeRegistry, + image_ref, + running, + serve_registry, +) @pytest.fixture def registry() -> Iterator[FakeRegistry]: - fake = FakeRegistry() - server = ThreadingHTTPServer(("127.0.0.1", 0), _handler(fake)) - fake.host = f"127.0.0.1:{server.server_address[1]}" - thread = threading.Thread(target=server.serve_forever, daemon=True) - thread.start() - try: + with serve_registry() as fake: yield fake - finally: - server.shutdown() - server.server_close() - thread.join(timeout=5) def resolve(registry: FakeRegistry, apps: dict[str, str], **limits: float) -> str | None: @@ -490,3 +351,83 @@ async def test_a_failure_leaves_the_label_and_never_raises(stack_images): await resolve_into(state, settings) assert state.resolved_stack is None + + +def _warning(events: list[dict]) -> dict: + warned = [e for e in events if str(e["event"]).startswith("stack not resolved")] + assert len(warned) == 1, events + return warned[0] + + +@pytest.mark.parametrize( + ("tags_answer", "reason"), + [ + pytest.param({"tags": 5}, "tag list is not a list", id="tags_is_a_number"), + pytest.param({"tags": "2.2.2"}, "tag list is not a list", id="tags_is_a_string"), + pytest.param([1, 2], "not an object", id="answer_is_an_array"), + ], +) +async def test_a_malformed_tag_list_warns_once_and_keeps_the_label( + registry, audit_events, tags_answer, reason +): + registry.stacks = {"2.2.2": ALL_V1} + registry.tags_answer = tags_answer + state = SimpleNamespace(resolved_stack=None) + + await resolve_into(state, settings_for(registry, ALL_V1)) + + warned = _warning(audit_events) + assert reason in warned["reason"] + assert warned["fallback"] == "2.2.2-rc.5" + assert state.resolved_stack is None + + +async def test_any_other_failure_warns_once_with_its_type_and_keeps_the_label(audit_events): + state = SimpleNamespace(resolved_stack=None) + # A budget that is not a number fails inside the resolver thread with a TypeError. + settings = SimpleNamespace( + stack_images=[image_ref(LOADER, "v1")], + stack_resolve=True, + stack_version="2.2.2-rc.5", + stack_manifest_repository="127.0.0.1:1/org/dfe-stack-manifest", + stack_resolve_request_timeout_seconds=1.0, + stack_resolve_budget_seconds="soon", + ) + + await resolve_into(state, settings) + + warned = _warning(audit_events) + assert warned["reason"].startswith("TypeError") + assert warned["fallback"] == "2.2.2-rc.5" + assert state.resolved_stack is None + + +@pytest.mark.parametrize( + ("pin", "expected", "calls_registry"), + [ + pytest.param(lambda: "2.2.0", None, False, id="a_pin_wins_so_the_registry_is_not_read"), + pytest.param(lambda: None, "2.2.2", True, id="no_pin_resolves"), + ], +) +async def test_the_registry_is_read_only_when_the_deploy_repo_pins_no_stack( + registry, pin, expected, calls_registry +): + registry.stacks = {"2.2.2": ALL_V1} + state = SimpleNamespace(resolved_stack=None) + + await resolve_into(state, settings_for(registry, ALL_V1), pin) + + assert state.resolved_stack == expected + assert (registry.requests > 0) is calls_registry + + +async def test_a_pin_that_cannot_be_read_warns_and_keeps_the_label(registry, audit_events): + def unreadable() -> str | None: + raise OSError("deploy repo gone") + + state = SimpleNamespace(resolved_stack=None) + + await resolve_into(state, settings_for(registry, ALL_V1), unreadable) + + assert "OSError" in _warning(audit_events)["reason"] + assert registry.requests == 0 From 283b6b649f259c580f9f8d46179fae49615e5b84 Mon Sep 17 00:00:00 2001 From: Derek Date: Mon, 16 Nov 2026 05:47:07 +1100 Subject: [PATCH 3/3] fix(system): type the test registry helper as a Generator ty flags a contextmanager annotated as Iterator. --- tests/support/oci_registry.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/support/oci_registry.py b/tests/support/oci_registry.py index 2175a916..d3c6559c 100644 --- a/tests/support/oci_registry.py +++ b/tests/support/oci_registry.py @@ -15,7 +15,7 @@ import json import threading import time -from collections.abc import Iterator +from collections.abc import Generator from contextlib import contextmanager from dataclasses import dataclass, field from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -163,7 +163,7 @@ def _tags(self, query: dict[str, list[str]]) -> None: @contextmanager -def serve_registry() -> Iterator[FakeRegistry]: +def serve_registry() -> Generator[FakeRegistry]: """Run a registry on a loopback port for the length of the block.""" fake = FakeRegistry() server = ThreadingHTTPServer(("127.0.0.1", 0), _handler(fake))