diff --git a/solstone/think/providers/artifact_proof.py b/solstone/think/providers/artifact_proof.py index 4af26c8ea..75ce6d194 100644 --- a/solstone/think/providers/artifact_proof.py +++ b/solstone/think/providers/artifact_proof.py @@ -14,7 +14,7 @@ import stat import tempfile from dataclasses import dataclass from pathlib import Path -from typing import Any, Callable, Literal +from typing import Any, Literal from solstone.think.journal_io.atomic import atomic_replace from solstone.think.journal_io.locking import hold_lock @@ -284,98 +284,6 @@ def prove_manifest( return ProofResult("ready", "ready", cache_hit=False) -def prove_cuda_sidecar( - *, - provider: str, - image_ref: str, - arch: str, - wanted_files: list[str] | tuple[str, ...], - target_dir: Path, - pin_identity: dict[str, Any], - verifier: Callable[[str, str, list[str] | tuple[str, ...], Path], bool], - journal_path: str | Path | None = None, -) -> ProofResult: - """Cache successful CUDA OCI sidecar verification.""" - _validate_provider(provider) - target_dir = Path(target_dir) - sidecar = target_dir / ".oci-install.json" - try: - sidecar_stat = _required_file_stat( - sidecar, - io_reason_code="cuda_sidecar_io_error", - ) - except FileNotFoundError: - return ProofResult("missing-or-mismatched", "cuda_sidecar_missing") - except ValueError: - return ProofResult("missing-or-mismatched", "cuda_sidecar_malformed") - except ProofUnavailableError as exc: - return ProofResult("proof-unavailable", exc.reason_code) - try: - record = json.loads(sidecar.read_text(encoding="utf-8")) - except OSError as exc: - _log_unavailable("CUDA sidecar read failed", exc) - return ProofResult("proof-unavailable", "cuda_sidecar_io_error") - except (json.JSONDecodeError, ValueError): - return ProofResult("missing-or-mismatched", "cuda_sidecar_malformed") - files = record.get("files") - if record.get("image_ref") != image_ref or record.get("arch") != arch: - return ProofResult("missing-or-mismatched", "cuda_sidecar_pin_mismatch") - if not isinstance(files, dict): - return ProofResult("missing-or-mismatched", "cuda_sidecar_malformed") - - file_fingerprints: list[dict[str, Any]] = [] - for name in wanted_files: - expected = files.get(name) - if not isinstance(expected, str): - return ProofResult("missing-or-mismatched", "expected_hash_unavailable") - path = target_dir / name - try: - stat_result = _required_file_stat(path) - except FileNotFoundError: - return ProofResult("missing-or-mismatched", "inventory_member_missing") - except ProofUnavailableError as exc: - return ProofResult("proof-unavailable", exc.reason_code) - stat_info = _stat_identity_from_result(path, stat_result) - stat_info.update( - { - "path": str(path), - "relative_path": name, - "expected_sha256": expected, - "executable": os.access(path, os.X_OK), - } - ) - file_fingerprints.append(stat_info) - - cache_path = proof_cache_path(provider, journal_path=journal_path) - cache = _read_proof_cache(cache_path) - sidecar_identity = canonical_fingerprint( - _stat_identity_from_result(sidecar, sidecar_stat) - ) - proof_key = _proof_key( - provider=provider, - pin_identity={ - **pin_identity, - "image_ref": image_ref, - "arch": arch, - "sidecar_identity": sidecar_identity, - }, - manifest_hash="cuda-oci-sidecar", - files=file_fingerprints, - ) - if proof_key in cache.get("affirmative", {}): - return ProofResult("ready", "ready", cache_hit=True) - - try: - verified = verifier(image_ref, arch, wanted_files, target_dir) - except OSError as exc: - _log_unavailable("CUDA sidecar verifier could not run", exc) - return ProofResult("proof-unavailable", "cuda_sidecar_verify_unavailable") - if not verified: - return ProofResult("missing-or-mismatched", "cuda_sidecar_verify_failed") - _write_affirmative_cache(cache_path, proof_key=proof_key) - return ProofResult("ready", "ready", cache_hit=False) - - def _inventory_file_fingerprints( root: Path, manifest: dict[str, Any], @@ -648,7 +556,6 @@ __all__ = [ "mlx_snapshot_manifest_path", "mlx_variant_manifest_path", "proof_cache_path", - "prove_cuda_sidecar", "prove_manifest", "publish_staged_tree", "read_manifest", diff --git a/solstone/think/providers/oci_image.py b/solstone/think/providers/oci_image.py index f1bafcb46..a0125d90b 100644 --- a/solstone/think/providers/oci_image.py +++ b/solstone/think/providers/oci_image.py @@ -1,35 +1,22 @@ # SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc -"""Acquire selected files from pinned OCI images. - -This module performs no network access at import time. -""" +"""OCI registry helpers used by CUDA runtime repacking.""" from __future__ import annotations import hashlib -import json -import logging import re import shutil -import subprocess import tarfile -import tempfile -from collections.abc import Callable -from dataclasses import dataclass from pathlib import Path -from typing import TYPE_CHECKING, Any, Sequence +from typing import TYPE_CHECKING, Any if TYPE_CHECKING: import httpx -LOG = logging.getLogger(__name__) - -SIDECAR_NAME = ".oci-install.json" _GHCR_HOST = "ghcr.io" _DIGEST_RE = re.compile(r"^[0-9a-f]{64}$") -_COSIGN_TIMEOUT_SECONDS = 60.0 _MANIFEST_ACCEPT = ", ".join( ( "application/vnd.oci.image.index.v1+json", @@ -41,217 +28,13 @@ _MANIFEST_ACCEPT = ", ".join( class OciImageError(RuntimeError): - """OCI image acquisition failure with a recovery reason code.""" + """OCI registry or layer processing failure with a recovery reason code.""" def __init__(self, reason_code: str, message: str) -> None: super().__init__(message) self.reason_code = reason_code -@dataclass(frozen=True) -class OciInstallRecord: - image_ref: str - arch: str - files: dict[str, str] - - def to_json(self) -> str: - return ( - json.dumps( - { - "arch": self.arch, - "files": self.files, - "image_ref": self.image_ref, - }, - indent=2, - sort_keys=True, - ) - + "\n" - ) - - @classmethod - def from_json(cls, text: str) -> OciInstallRecord: - data = json.loads(text) - files = data["files"] - if not isinstance(files, dict): - raise ValueError("OCI install record files must be an object") - return cls( - image_ref=str(data["image_ref"]), - arch=str(data["arch"]), - files={str(name): str(digest) for name, digest in files.items()}, - ) - - -@dataclass(frozen=True) -class OciInstallResult: - target_dir: Path - files: dict[str, str] - already_present: bool - - -@dataclass(frozen=True, kw_only=True) -class OciSignaturePolicy: - certificate_identity: str | None = None - certificate_identity_regexp: str | None = None - oidc_issuer: str - - def __post_init__(self) -> None: - has_identity = bool(self.certificate_identity) - has_regexp = bool(self.certificate_identity_regexp) - if has_identity == has_regexp: - raise ValueError( - "exactly one of certificate_identity or " - "certificate_identity_regexp is required" - ) - if not self.oidc_issuer: - raise ValueError("oidc_issuer is required") - - -def pull_and_install( - image_ref: str, - arch: str, - wanted_files: Sequence[str], - target_dir: Path, - *, - client: httpx.Client | None = None, - policy: OciSignaturePolicy | None = None, - verifier: Callable[[str, OciSignaturePolicy], None] | None = None, -) -> OciInstallResult: - repo, digest = _parse_image_ref(image_ref) - wanted = _validate_wanted_files(wanted_files) - target_dir = Path(target_dir) - - offline = _offline_result(image_ref, arch, wanted, target_dir) - if offline is not None: - return offline - - if policy is not None: - (verifier or verify_image_signature)(image_ref, policy) - - import httpx - - created_client = client is None - if client is None: - client = httpx.Client(follow_redirects=True, timeout=600.0) - - target_dir.parent.mkdir(parents=True, exist_ok=True) - work_root = Path(tempfile.mkdtemp(dir=target_dir.parent)) - try: - token = _fetch_token(client, repo) - manifest = _fetch_image_manifest(client, repo, digest, arch, token) - rootfs = work_root / "rootfs" - blobs_dir = work_root / "blobs" - rootfs.mkdir() - blobs_dir.mkdir() - for index, layer_digest in enumerate(_layer_digests(manifest), start=1): - blob_path = blobs_dir / f"layer-{index}.tar" - _download_blob(client, repo, layer_digest, blob_path, token) - _extract_layer(blob_path, rootfs) - selected = _find_wanted_files(rootfs, wanted) - files = _publish_install(image_ref, arch, selected, target_dir) - return OciInstallResult( - target_dir=target_dir, - files=files, - already_present=False, - ) - finally: - shutil.rmtree(work_root, ignore_errors=True) - if created_client: - client.close() - - -def verify_image_signature(image_ref: str, policy: OciSignaturePolicy) -> None: - command = ["cosign", "verify", image_ref] - if policy.certificate_identity is not None: - command.extend(["--certificate-identity", policy.certificate_identity]) - elif policy.certificate_identity_regexp is not None: - command.extend( - ["--certificate-identity-regexp", policy.certificate_identity_regexp] - ) - command.extend(["--certificate-oidc-issuer", policy.oidc_issuer]) - - try: - completed = subprocess.run( - command, - capture_output=True, - text=True, - timeout=_COSIGN_TIMEOUT_SECONDS, - check=False, - ) - except subprocess.TimeoutExpired as exc: - raise OciImageError( - "signature_verify_failed", - f"cosign verify timed out after {_COSIGN_TIMEOUT_SECONDS:g}s", - ) from exc - except OSError as exc: - raise OciImageError( - "cosign_missing", f"cosign verify could not start: {exc}" - ) from exc - - if completed.returncode == 0: - return - - detail = ( - (completed.stderr or "").strip() - or (completed.stdout or "").strip() - or f"exited with status {completed.returncode}" - ) - raise OciImageError( - "signature_verify_failed", f"cosign verify failed for {image_ref}: {detail}" - ) - - -def _parse_image_ref(image_ref: str) -> tuple[str, str]: - if "@" not in image_ref: - raise OciImageError( - "invalid_image_ref", f"OCI image ref must be pinned by digest: {image_ref}" - ) - repo_part, digest_part = image_ref.split("@", 1) - if repo_part.startswith(f"{_GHCR_HOST}/"): - repo = repo_part[len(_GHCR_HOST) + 1 :] - else: - first = repo_part.split("/", 1)[0] - if "." in first or ":" in first or first == "localhost": - raise OciImageError( - "invalid_image_ref", f"OCI image ref must use ghcr.io: {image_ref}" - ) - repo = repo_part - if not repo: - raise OciImageError( - "invalid_image_ref", f"OCI image ref must include a repository: {image_ref}" - ) - if not digest_part.startswith("sha256:"): - raise OciImageError( - "invalid_image_ref", f"OCI image ref must use a sha256 digest: {image_ref}" - ) - digest = digest_part.removeprefix("sha256:") - if not _DIGEST_RE.fullmatch(digest): - raise OciImageError( - "invalid_image_ref", - f"OCI image ref digest must be 64 lowercase hex chars: {image_ref}", - ) - return repo, digest - - -def _validate_wanted_files(wanted_files: Sequence[str]) -> list[str]: - wanted: list[str] = [] - seen: set[str] = set() - for name in wanted_files: - if ( - not name - or name == "." - or name == SIDECAR_NAME - or "/" in name - or ".." in name - ): - raise OciImageError( - "invalid_wanted_file", f"wanted file must be a basename: {name!r}" - ) - if name not in seen: - wanted.append(name) - seen.add(name) - return wanted - - def _sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: @@ -269,59 +52,6 @@ def _verify_sha256(path: Path, expected: str) -> None: ) -def _offline_result( - image_ref: str, - arch: str, - wanted: Sequence[str], - target_dir: Path, -) -> OciInstallResult | None: - record = _verify_sidecar(image_ref, arch, wanted, target_dir) - if record is None: - return None - return OciInstallResult( - target_dir=target_dir, - files=dict(record.files), - already_present=True, - ) - - -def _verify_sidecar( - image_ref: str, - arch: str, - wanted: Sequence[str], - target_dir: Path, -) -> OciInstallRecord | None: - sidecar = target_dir / SIDECAR_NAME - try: - record = OciInstallRecord.from_json(sidecar.read_text(encoding="utf-8")) - except Exception: - return None - if record.image_ref != image_ref or record.arch != arch: - return None - for name in wanted: - expected = record.files.get(name) - if expected is None: - return None - path = target_dir / name - if not path.is_file(): - return None - try: - _verify_sha256(path, expected) - except OciImageError: - return None - return record - - -def verify_sidecar_install( - image_ref: str, - arch: str, - wanted_files: Sequence[str], - target_dir: Path, -) -> bool: - wanted = _validate_wanted_files(wanted_files) - return _verify_sidecar(image_ref, arch, wanted, Path(target_dir)) is not None - - def _fetch_token(client: httpx.Client, repo: str) -> str: url = ( f"https://{_GHCR_HOST}/token?service={_GHCR_HOST}&scope=repository:{repo}:pull" @@ -532,27 +262,6 @@ def _remove_path(path: Path) -> None: path.unlink(missing_ok=True) -def _find_wanted_files(rootfs: Path, wanted: Sequence[str]) -> dict[str, Path]: - found: dict[str, Path] = {} - rootfs_resolved = rootfs.resolve() - for name in wanted: - found[name] = _find_wanted_file(rootfs, rootfs_resolved, name) - return found - - -def _find_wanted_file(rootfs: Path, rootfs_resolved: Path, name: str) -> Path: - direct = rootfs / name - if direct.is_file(): - return _resolved_under_root(direct, rootfs_resolved) - matches = [path for path in rootfs.rglob(name) if path.is_file()] - if not matches: - raise OciImageError( - "wanted_file_missing", f"Extracted image did not contain {name}" - ) - matches.sort(key=lambda path: (len(path.relative_to(rootfs).parts), str(path))) - return _resolved_under_root(matches[0], rootfs_resolved) - - def _resolved_under_root(path: Path, rootfs_resolved: Path) -> Path: resolved = path.resolve() if resolved != rootfs_resolved and rootfs_resolved not in resolved.parents: @@ -562,74 +271,4 @@ def _resolved_under_root(path: Path, rootfs_resolved: Path) -> Path: return resolved -def _publish_install( - image_ref: str, - arch: str, - selected: dict[str, Path], - target_dir: Path, -) -> dict[str, str]: - target_dir.parent.mkdir(parents=True, exist_ok=True) - staging = Path(tempfile.mkdtemp(dir=target_dir.parent)) - aside: Path | None = None - published = False - try: - files: dict[str, str] = {} - for name, source in selected.items(): - dest = staging / name - shutil.copy2(source, dest) - files[name] = _sha256_file(dest) - record = OciInstallRecord(image_ref=image_ref, arch=arch, files=files) - (staging / SIDECAR_NAME).write_text(record.to_json(), encoding="utf-8") - - if target_dir.exists(): - aside = Path(tempfile.mkdtemp(dir=target_dir.parent)) - target_dir.replace(aside) - try: - staging.replace(target_dir) - published = True - except Exception: - _restore_aside(aside, target_dir) - raise - - if aside is not None: - try: - shutil.rmtree(aside) - except OSError: - LOG.warning( - "failed to remove previous OCI install target", exc_info=True - ) - aside = None - return files - except Exception as exc: - if not published: - _restore_aside(aside, target_dir) - raise OciImageError( - "install_failed", f"failed to install OCI image files: {exc}" - ) from exc - finally: - if not published and staging.exists(): - shutil.rmtree(staging, ignore_errors=True) - if not published and aside is not None and aside.exists(): - shutil.rmtree(aside, ignore_errors=True) - - -def _restore_aside(aside: Path | None, target_dir: Path) -> None: - if aside is None or not aside.exists() or target_dir.exists(): - return - try: - aside.replace(target_dir) - except Exception: - LOG.exception("failed to restore previous OCI install target") - raise - - -__all__ = [ - "OciImageError", - "OciInstallRecord", - "OciInstallResult", - "OciSignaturePolicy", - "SIDECAR_NAME", - "pull_and_install", - "verify_image_signature", - "verify_sidecar_install", -] +__all__ = ["OciImageError"] diff --git a/tests/test_artifact_proof.py b/tests/test_artifact_proof.py index 107b5ab10..9e3357f92 100644 --- a/tests/test_artifact_proof.py +++ b/tests/test_artifact_proof.py @@ -4,7 +4,6 @@ from __future__ import annotations import hashlib -import json import os from pathlib import Path @@ -18,7 +17,6 @@ from solstone.think.providers.artifact_proof import ( mlx_snapshot_manifest_path, mlx_variant_manifest_path, proof_cache_path, - prove_cuda_sidecar, prove_manifest, publish_staged_tree, write_manifest, @@ -367,132 +365,6 @@ def test_mlx_manifests_are_solstone_side(tmp_path, monkeypatch) -> None: assert not str(variant).startswith(str(hf_root)) -def test_cuda_sidecar_success_is_cached_without_second_verifier( - tmp_path, monkeypatch -) -> None: - monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) - target = tmp_path / "cuda" - target.mkdir() - (target / "llama-server").write_bytes(b"server") - (target / ".oci-install.json").write_text( - json.dumps( - { - "image_ref": "image@sha256:abc", - "arch": "amd64", - "files": {"llama-server": _sha(b"server")}, - } - ) - + "\n", - encoding="utf-8", - ) - calls = 0 - hash_calls: list[Path] = [] - real_hash = artifact_proof._sha256_file - - def verifier(_image_ref, _arch, wanted, verify_target) -> bool: - nonlocal calls - calls += 1 - for name in wanted: - path = verify_target / name - hash_calls.append(path) - real_hash(path) - return True - - first = prove_cuda_sidecar( - provider="local", - image_ref="image@sha256:abc", - arch="amd64", - wanted_files=("llama-server",), - target_dir=target, - pin_identity={"pin": "cuda"}, - verifier=verifier, - ) - second = prove_cuda_sidecar( - provider="local", - image_ref="image@sha256:abc", - arch="amd64", - wanted_files=("llama-server",), - target_dir=target, - pin_identity={"pin": "cuda"}, - verifier=verifier, - ) - - assert first.ready - assert second.ready - assert second.cache_hit is True - assert calls == 1 - assert hash_calls == [target / "llama-server"] - - -def test_cuda_sidecar_absent_is_repair_needed(tmp_path, monkeypatch) -> None: - monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) - target = tmp_path / "cuda" - target.mkdir() - - def verifier(_image_ref, _arch, _wanted, _target) -> bool: - raise AssertionError("verifier should not run when the sidecar is absent") - - result = prove_cuda_sidecar( - provider="local", - image_ref="image@sha256:abc", - arch="amd64", - wanted_files=("llama-server",), - target_dir=target, - pin_identity={"pin": "cuda"}, - verifier=verifier, - ) - - assert result.status == "missing-or-mismatched" - assert result.reason_code == "cuda_sidecar_missing" - - -def test_cuda_sidecar_unreadable_is_proof_unavailable_and_does_not_verify( - tmp_path, - monkeypatch, -) -> None: - _skip_if_root_chmod_is_ignored() - monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) - target = tmp_path / "cuda" - target.mkdir() - (target / "llama-server").write_bytes(b"server") - sidecar = target / ".oci-install.json" - sidecar.write_text( - json.dumps( - { - "image_ref": "image@sha256:abc", - "arch": "amd64", - "files": {"llama-server": _sha(b"server")}, - } - ) - + "\n", - encoding="utf-8", - ) - sidecar.chmod(0o000) - calls = 0 - - def verifier(_image_ref, _arch, _wanted, _target) -> bool: - nonlocal calls - calls += 1 - return True - - try: - result = prove_cuda_sidecar( - provider="local", - image_ref="image@sha256:abc", - arch="amd64", - wanted_files=("llama-server",), - target_dir=target, - pin_identity={"pin": "cuda"}, - verifier=verifier, - ) - assert result.status == "proof-unavailable" - assert result.reason_code == "cuda_sidecar_io_error" - assert calls == 0 - assert (target / "llama-server").read_bytes() == b"server" - finally: - sidecar.chmod(0o600) - - def test_publish_staged_tree_restores_prior_tree_on_replace_failure( tmp_path, monkeypatch ) -> None: diff --git a/tests/test_oci_image.py b/tests/test_oci_image.py index 391d02af5..b248cc134 100644 --- a/tests/test_oci_image.py +++ b/tests/test_oci_image.py @@ -5,45 +5,19 @@ from __future__ import annotations import hashlib import io -import subprocess import tarfile from pathlib import Path +from typing import Any -import httpx import pytest from solstone.think.providers import oci_image -REPO = "acme/tool" -IMAGE_DIGEST = "a" * 64 -MANIFEST_DIGEST = "b" * 64 -ARM_MANIFEST_DIGEST = "c" * 64 -IMAGE_REF = f"ghcr.io/{REPO}@sha256:{IMAGE_DIGEST}" -ALT_IMAGE_REF = f"ghcr.io/{REPO}@sha256:{'d' * 64}" -TOP_REF = f"sha256:{IMAGE_DIGEST}" -MANIFEST_REF = f"sha256:{MANIFEST_DIGEST}" -ARM_MANIFEST_REF = f"sha256:{ARM_MANIFEST_DIGEST}" - - -def _policy() -> oci_image.OciSignaturePolicy: - return oci_image.OciSignaturePolicy( - certificate_identity_regexp=r"^https://github\.com/acme/tool/.+$", - oidc_issuer="https://token.actions.githubusercontent.com", - ) - def _sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() -def _sha256_file(path: Path) -> str: - digest = hashlib.sha256() - with path.open("rb") as handle: - for chunk in iter(lambda: handle.read(1024 * 1024), b""): - digest.update(chunk) - return digest.hexdigest() - - def _digest_ref(data: bytes) -> str: return f"sha256:{_sha256_bytes(data)}" @@ -58,710 +32,167 @@ def _layer_bytes(entries: dict[str, bytes]) -> bytes: return buffer.getvalue() -def _single_member_layer(name: str, data: bytes = b"bad") -> bytes: - buffer = io.BytesIO() - with tarfile.open(fileobj=buffer, mode="w:gz") as archive: - member = tarfile.TarInfo(name) - member.size = len(data) - archive.addfile(member, io.BytesIO(data)) - return buffer.getvalue() - - -def _index(manifest_ref: str = MANIFEST_REF) -> dict: - return { - "schemaVersion": 2, - "manifests": [ - { - "mediaType": "application/vnd.oci.image.manifest.v1+json", - "digest": manifest_ref, - "platform": {"os": "linux", "architecture": "amd64"}, - }, - { - "mediaType": "application/vnd.oci.image.manifest.v1+json", - "digest": ARM_MANIFEST_REF, - "platform": {"os": "linux", "architecture": "arm64"}, - }, - ], - } - - -def _manifest_for_layers(layer_bytes: list[bytes]) -> tuple[dict, dict[str, bytes]]: - blobs = {_digest_ref(data): data for data in layer_bytes} - manifest = { - "schemaVersion": 2, - "layers": [ - { - "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip", - "digest": digest, - } - for digest in blobs - ], - } - return manifest, blobs - - -class _Registry: - def __init__( - self, - manifests: dict[str, dict], - blobs: dict[str, bytes], - *, - token_status: int = 200, - manifest_statuses: dict[str, int] | None = None, - blob_statuses: dict[str, int] | None = None, - garbage_manifests: set[str] | None = None, - ) -> None: - self.manifests = manifests - self.blobs = blobs - self.token_status = token_status - self.manifest_statuses = manifest_statuses or {} - self.blob_statuses = blob_statuses or {} - self.garbage_manifests = garbage_manifests or set() - self.requests: list[httpx.Request] = [] - - def client(self) -> httpx.Client: - return httpx.Client(transport=httpx.MockTransport(self.handle)) - - def handle(self, request: httpx.Request) -> httpx.Response: - self.requests.append(request) - path = request.url.path - if path == "/token": - return httpx.Response( - self.token_status, - json={"token": "token-1"}, - request=request, - ) - - manifest_prefix = f"/v2/{REPO}/manifests/" - if path.startswith(manifest_prefix): - digest = path.removeprefix(manifest_prefix) - status = self.manifest_statuses.get(digest) - if status is not None: - return httpx.Response(status, request=request) - if digest in self.garbage_manifests: - return httpx.Response(200, content=b"not json", request=request) - payload = self.manifests.get(digest) - if payload is None: - return httpx.Response(404, request=request) - return httpx.Response(200, json=payload, request=request) - - blob_prefix = f"/v2/{REPO}/blobs/" - if path.startswith(blob_prefix): - digest = path.removeprefix(blob_prefix) - status = self.blob_statuses.get(digest) - if status is not None: - return httpx.Response(status, request=request) - payload = self.blobs.get(digest) - if payload is None: - return httpx.Response(404, request=request) - return httpx.Response(200, content=payload, request=request) - - return httpx.Response(404, request=request) - - -def _registry_for_layers(layer_bytes: list[bytes]) -> _Registry: - manifest, blobs = _manifest_for_layers(layer_bytes) - return _Registry( - { - TOP_REF: _index(), - MANIFEST_REF: manifest, - ARM_MANIFEST_REF: {"schemaVersion": 2, "layers": []}, - }, - blobs, - ) - - -def _pull_with_registry( - registry: _Registry, - target: Path, - wanted: list[str], - *, - image_ref: str = IMAGE_REF, - policy: oci_image.OciSignaturePolicy | None = None, - verifier=None, -) -> oci_image.OciInstallResult: - with registry.client() as client: - return oci_image.pull_and_install( - image_ref, - "amd64", - wanted, - target, - client=client, - policy=policy, - verifier=verifier, - ) +def _write_layer(path: Path, entries: dict[str, bytes]) -> None: + path.write_bytes(_layer_bytes(entries)) -def _assert_reason( - registry: _Registry, - target: Path, - reason_code: str, - *, - wanted: list[str] | None = None, -) -> None: - with registry.client() as client: - with pytest.raises(oci_image.OciImageError) as exc_info: - oci_image.pull_and_install( - IMAGE_REF, - "amd64", - wanted or ["tool"], - target, - client=client, - ) - assert exc_info.value.reason_code == reason_code - - -def test_ac1_record_round_trip_and_invalid_inputs(tmp_path: Path) -> None: - record = oci_image.OciInstallRecord( - image_ref=IMAGE_REF, - arch="amd64", - files={"tool": "1" * 64}, - ) +class _Response: + def __init__(self, data: bytes, *, status_error: Exception | None = None) -> None: + self._data = data + self._status_error = status_error - assert oci_image.OciInstallRecord.from_json(record.to_json()) == record + def __enter__(self) -> _Response: + return self - for image_ref in ( - f"ghcr.io/{REPO}", - f"ghcr.io/{REPO}@sha256:{'1' * 63}", - f"docker.io/{REPO}@sha256:{IMAGE_DIGEST}", - ): - with pytest.raises(oci_image.OciImageError) as exc_info: - oci_image.pull_and_install(image_ref, "amd64", ["tool"], tmp_path / "out") - assert exc_info.value.reason_code == "invalid_image_ref" + def __exit__(self, *_args: object) -> None: + return None - for wanted in (["dir/tool"], [".."], ["tool..old"], [oci_image.SIDECAR_NAME], [""]): - with pytest.raises(oci_image.OciImageError) as exc_info: - oci_image.pull_and_install(IMAGE_REF, "amd64", wanted, tmp_path / "out") - assert exc_info.value.reason_code == "invalid_wanted_file" + def raise_for_status(self) -> None: + if self._status_error is not None: + raise self._status_error + def iter_bytes(self) -> list[bytes]: + return [self._data] -def test_signature_policy_requires_exactly_one_identity() -> None: - with pytest.raises(ValueError, match="exactly one"): - oci_image.OciSignaturePolicy( - oidc_issuer="https://token.actions.githubusercontent.com" - ) - with pytest.raises(ValueError, match="exactly one"): - oci_image.OciSignaturePolicy( - certificate_identity="identity", - certificate_identity_regexp="identity-regexp", - oidc_issuer="https://token.actions.githubusercontent.com", - ) +class _Client: + def __init__(self, data: bytes) -> None: + self.data = data + self.calls: list[tuple[str, str, dict[str, str]]] = [] -def test_verify_image_signature_invokes_cosign_with_keyless_flags( - monkeypatch: pytest.MonkeyPatch, -) -> None: - calls: list[dict[str, object]] = [] - policy = _policy() + def stream(self, method: str, url: str, *, headers: dict[str, str]) -> _Response: + self.calls.append((method, url, headers)) + return _Response(self.data) - def fake_run(command, **kwargs): - calls.append({"command": command, **kwargs}) - return subprocess.CompletedProcess(command, 0, stdout="ok", stderr="") - monkeypatch.setattr(oci_image.subprocess, "run", fake_run) +def test_retained_repack_import_surface_and_owner_apis_absent() -> None: + for name in ( + "_MANIFEST_ACCEPT", + "_download_blob", + "_extract_layer", + "_fetch_token", + "_layer_digests", + "_registry_headers", + "_resolved_under_root", + "_select_arch_manifest", + "_sha256_file", + "_valid_digest_ref", + "_verify_sha256", + "OciImageError", + ): + assert hasattr(oci_image, name) + + for retired in ( + "OciInstallRecord", + "OciInstallResult", + "OciSignaturePolicy", + "SIDECAR_NAME", + "pull_and_install", + "verify_image_signature", + "verify_sidecar_install", + ): + assert not hasattr(oci_image, retired) - oci_image.verify_image_signature(IMAGE_REF, policy) - assert calls == [ +def test_select_arch_manifest_and_layer_digests_validate_digest_refs() -> None: + manifest_ref = "sha256:" + "a" * 64 + arm_ref = "sha256:" + "b" * 64 + manifests: list[dict[str, Any]] = [ + { + "digest": manifest_ref, + "platform": {"os": "linux", "architecture": "amd64"}, + }, { - "command": [ - "cosign", - "verify", - IMAGE_REF, - "--certificate-identity-regexp", - policy.certificate_identity_regexp, - "--certificate-oidc-issuer", - policy.oidc_issuer, - ], - "capture_output": True, - "text": True, - "timeout": oci_image._COSIGN_TIMEOUT_SECONDS, - "check": False, - } + "digest": arm_ref, + "platform": {"os": "linux", "architecture": "arm64"}, + }, ] - -def test_verify_image_signature_maps_missing_cosign( - monkeypatch: pytest.MonkeyPatch, -) -> None: - def fail_run(_command, **_kwargs): - raise FileNotFoundError("cosign missing") - - monkeypatch.setattr(oci_image.subprocess, "run", fail_run) + assert oci_image._select_arch_manifest(manifests, "arm64") == arm_ref + assert oci_image._layer_digests({"layers": [{"digest": manifest_ref}]}) == [ + manifest_ref + ] with pytest.raises(oci_image.OciImageError) as exc_info: - oci_image.verify_image_signature(IMAGE_REF, _policy()) - - assert exc_info.value.reason_code == "cosign_missing" - - -def test_verify_image_signature_maps_nonzero_exit( - monkeypatch: pytest.MonkeyPatch, -) -> None: - def fake_run(command, **_kwargs): - return subprocess.CompletedProcess(command, 1, stdout="", stderr="bad sig") - - monkeypatch.setattr(oci_image.subprocess, "run", fake_run) + oci_image._select_arch_manifest( + [{"digest": "sha256:nothex", "platform": manifests[0]["platform"]}], + "amd64", + ) + assert exc_info.value.reason_code == "manifest_fetch_failed" with pytest.raises(oci_image.OciImageError) as exc_info: - oci_image.verify_image_signature(IMAGE_REF, _policy()) - - assert exc_info.value.reason_code == "signature_verify_failed" - assert "bad sig" in str(exc_info.value) + oci_image._layer_digests({"layers": [{"digest": "sha256:nothex"}]}) + assert exc_info.value.reason_code == "manifest_fetch_failed" -def test_ac2_happy_path_installs_files_and_sidecar(tmp_path: Path) -> None: - layer = _layer_bytes( - { - "usr/local/bin/llama-server": b"server", - "usr/lib/libfoo.so": b"library", - } - ) - registry = _registry_for_layers([layer]) - target = tmp_path / "target" +def test_download_blob_writes_verified_blob(tmp_path: Path) -> None: + data = b"blob" + client = _Client(data) + dest = tmp_path / "blob.tar" - result = _pull_with_registry(registry, target, ["llama-server", "libfoo.so"]) + oci_image._download_blob(client, "acme/tool", _digest_ref(data), dest, "token") - assert result.already_present is False - assert (target / "llama-server").read_bytes() == b"server" - assert (target / "libfoo.so").read_bytes() == b"library" - record = oci_image.OciInstallRecord.from_json( - (target / oci_image.SIDECAR_NAME).read_text(encoding="utf-8") - ) - assert record.image_ref == IMAGE_REF - assert record.arch == "amd64" - assert record.files == result.files - assert record.files == { - "llama-server": _sha256_file(target / "llama-server"), - "libfoo.so": _sha256_file(target / "libfoo.so"), - } - - protected = [ - request for request in registry.requests if request.url.path != "/token" - ] - assert protected - assert all( - request.headers["authorization"] == "Bearer token-1" for request in protected - ) - assert all("accept" in request.headers for request in protected) - manifest_requests = [ - request for request in protected if "/manifests/" in request.url.path + assert dest.read_bytes() == data + assert client.calls == [ + ( + "GET", + f"https://ghcr.io/v2/acme/tool/blobs/{_digest_ref(data)}", + oci_image._registry_headers("token"), + ) ] - assert len(manifest_requests) == 2 - assert all( - "application/vnd.oci.image.index.v1+json" in request.headers["accept"] - for request in manifest_requests - ) - - -def test_signature_verifier_runs_before_first_blob_fetch(tmp_path: Path) -> None: - layer = _layer_bytes({"bin/tool": b"tool"}) - registry = _registry_for_layers([layer]) - target = tmp_path / "target" - events: list[str] = [] - original_handle = registry.handle - - def handle(request: httpx.Request) -> httpx.Response: - if "/blobs/" in request.url.path: - events.append("blob") - return original_handle(request) - - def verifier(_image_ref: str, _policy: oci_image.OciSignaturePolicy) -> None: - events.append("verify") - - registry.handle = handle # type: ignore[method-assign] - - _pull_with_registry( - registry, - target, - ["tool"], - policy=_policy(), - verifier=verifier, - ) - - assert events.index("verify") < events.index("blob") -def test_signature_verifier_failure_leaves_target_unchanged(tmp_path: Path) -> None: - layer = _layer_bytes({"bin/tool": b"tool"}) - registry = _registry_for_layers([layer]) - target = tmp_path / "target" - target.mkdir() - (target / "old").write_text("old\n", encoding="utf-8") - - def verifier(_image_ref: str, _policy: oci_image.OciSignaturePolicy) -> None: - raise oci_image.OciImageError("cosign_missing", "cosign missing") +def test_download_blob_rejects_sha256_mismatch(tmp_path: Path) -> None: + client = _Client(b"actual") + dest = tmp_path / "blob.tar" with pytest.raises(oci_image.OciImageError) as exc_info: - _pull_with_registry( - registry, - target, - ["tool"], - policy=_policy(), - verifier=verifier, - ) - - assert exc_info.value.reason_code == "cosign_missing" - assert (target / "old").read_text(encoding="utf-8") == "old\n" - assert not (target / "tool").exists() - assert list(tmp_path.rglob("*.tmp")) == [] - assert {path.name for path in tmp_path.iterdir()} == {"target"} - - -def test_ac3_fetch_failures_raise_reason_codes(tmp_path: Path) -> None: - layer = _layer_bytes({"bin/tool": b"tool"}) - manifest, blobs = _manifest_for_layers([layer]) - - cases: list[tuple[str, _Registry]] = [ - ( - "token_fetch_failed", - _Registry( - {TOP_REF: _index(), MANIFEST_REF: manifest}, blobs, token_status=401 - ), - ), - ( - "manifest_fetch_failed", - _Registry( - {TOP_REF: _index(), MANIFEST_REF: manifest}, - blobs, - manifest_statuses={TOP_REF: 500}, - ), - ), - ( - "manifest_fetch_failed", - _Registry( - {TOP_REF: _index(), MANIFEST_REF: manifest}, - blobs, - garbage_manifests={TOP_REF}, - ), - ), - ( - "arch_unavailable", - _Registry( - { - TOP_REF: { - "schemaVersion": 2, - "manifests": [ - { - "digest": ARM_MANIFEST_REF, - "platform": {"os": "linux", "architecture": "arm64"}, - } - ], - }, - ARM_MANIFEST_REF: manifest, - }, - blobs, - ), - ), - ( - "blob_fetch_failed", - _Registry( - {TOP_REF: _index(), MANIFEST_REF: manifest}, - blobs, - blob_statuses={next(iter(blobs)): 500}, - ), - ), - ] - - for index, (reason_code, registry) in enumerate(cases): - target = tmp_path / f"target-{index}" - _assert_reason(registry, target, reason_code) - assert not target.exists() - - -def test_ac4_blob_sha_mismatch_leaves_target_unchanged(tmp_path: Path) -> None: - layer = _layer_bytes({"bin/tool": b"tool"}) - wrong_digest = f"sha256:{'f' * 64}" - manifest = { - "schemaVersion": 2, - "layers": [ - { - "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip", - "digest": wrong_digest, - } - ], - } - registry = _Registry( - {TOP_REF: _index(), MANIFEST_REF: manifest}, {wrong_digest: layer} - ) - target = tmp_path / "target" - target.mkdir() - (target / "old").write_text("old\n", encoding="utf-8") - - _assert_reason(registry, target, "sha256_mismatch") + oci_image._download_blob(client, "acme/tool", "sha256:" + "0" * 64, dest, "t") - assert (target / "old").read_text(encoding="utf-8") == "old\n" - assert not (target / "tool").exists() - assert list(tmp_path.rglob("*.tmp")) == [] + assert exc_info.value.reason_code == "sha256_mismatch" + assert not dest.exists() + assert not (tmp_path / "blob.tar.tmp").exists() def test_ac5_path_traversal_is_rejected(tmp_path: Path) -> None: - registry = _registry_for_layers([_single_member_layer("../escape")]) - target = tmp_path / "target" + layer = tmp_path / "layer.tar.gz" + _write_layer(layer, {"../escape": b"bad"}) + rootfs = tmp_path / "rootfs" - _assert_reason(registry, target, "archive_path_traversal") + with pytest.raises(oci_image.OciImageError) as exc_info: + oci_image._extract_layer(layer, rootfs) + assert exc_info.value.reason_code == "archive_path_traversal" assert not (tmp_path / "escape").exists() - assert not target.exists() -def test_ac5_corrupt_layer_raises_extract_failed(tmp_path: Path) -> None: - corrupt = b"this is not a tarball" - registry = _registry_for_layers([corrupt]) - target = tmp_path / "target" +def test_corrupt_layer_raises_extract_failed(tmp_path: Path) -> None: + layer = tmp_path / "layer.tar.gz" + layer.write_bytes(b"this is not a tarball") - _assert_reason(registry, target, "extract_failed") + with pytest.raises(oci_image.OciImageError) as exc_info: + oci_image._extract_layer(layer, tmp_path / "rootfs") - assert not target.exists() + assert exc_info.value.reason_code == "extract_failed" -def test_ac5_whiteout_and_opaque_directory_remove_earlier_files( - tmp_path: Path, -) -> None: - layer_one = _layer_bytes({"bin/tool": b"tool", "app/old": b"old"}) - layer_two = _layer_bytes( +def test_whiteout_and_opaque_directory_remove_earlier_files(tmp_path: Path) -> None: + layer_one = tmp_path / "one.tar.gz" + _write_layer(layer_one, {"bin/tool": b"tool", "app/old": b"old"}) + layer_two = tmp_path / "two.tar.gz" + _write_layer( + layer_two, { "bin/.wh.tool": b"", "app/.wh..wh..opq": b"", "app/new": b"new", - } - ) - - _assert_reason( - _registry_for_layers([layer_one, layer_two]), - tmp_path / "whiteout", - "wanted_file_missing", - wanted=["tool"], - ) - _assert_reason( - _registry_for_layers([layer_one, layer_two]), - tmp_path / "opaque", - "wanted_file_missing", - wanted=["old"], - ) - - -def test_ac6_shallowest_match_wins(tmp_path: Path) -> None: - layer_one = _layer_bytes({"very/deep/tool": b"deep"}) - layer_two = _layer_bytes({"bin/tool": b"shallow"}) - registry = _registry_for_layers([layer_one, layer_two]) - target = tmp_path / "target" - - _pull_with_registry(registry, target, ["tool"]) - - assert (target / "tool").read_bytes() == b"shallow" - - -def test_ac6_missing_wanted_file_raises_reason_code(tmp_path: Path) -> None: - registry = _registry_for_layers([_layer_bytes({"bin/other": b"other"})]) - - _assert_reason(registry, tmp_path / "target", "wanted_file_missing") - - -def test_ac6_publish_failure_restores_preexisting_target( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, -) -> None: - layer = _layer_bytes({"bin/one": b"one", "bin/two": b"two"}) - registry = _registry_for_layers([layer]) - target = tmp_path / "target" - target.mkdir() - (target / "old").write_text("old\n", encoding="utf-8") - (target / oci_image.SIDECAR_NAME).write_text("old sidecar\n", encoding="utf-8") - original_copy2 = oci_image.shutil.copy2 - calls = 0 - - def flaky_copy2(src: Path, dest: Path) -> Path: - nonlocal calls - calls += 1 - if calls == 2: - raise OSError("copy broke") - return Path(original_copy2(src, dest)) - - monkeypatch.setattr(oci_image.shutil, "copy2", flaky_copy2) - - _assert_reason(registry, target, "install_failed", wanted=["one", "two"]) - - assert (target / "old").read_text(encoding="utf-8") == "old\n" - assert (target / oci_image.SIDECAR_NAME).read_text( - encoding="utf-8" - ) == "old sidecar\n" - assert not (target / "one").exists() - assert {path.name for path in tmp_path.iterdir()} == {"target"} - - -def test_ac6_publish_replace_failure_restores_moved_aside_target( - tmp_path: Path, - monkeypatch: pytest.MonkeyPatch, -) -> None: - layer = _layer_bytes({"bin/tool": b"tool"}) - registry = _registry_for_layers([layer]) - target = tmp_path / "target" - target.mkdir() - old_file = target / "old" - old_file.write_text("old\n", encoding="utf-8") - old_record = oci_image.OciInstallRecord( - image_ref=ALT_IMAGE_REF, - arch="amd64", - files={"old": _sha256_file(old_file)}, - ) - (target / oci_image.SIDECAR_NAME).write_text( - old_record.to_json(), - encoding="utf-8", - ) - before = {path.name: path.read_bytes() for path in target.iterdir()} - original_replace = oci_image.Path.replace - aside_path: Path | None = None - events: list[str] = [] - - def flaky_replace(self: Path, target_path: Path) -> Path: - nonlocal aside_path - target_path = Path(target_path) - if self == target: - aside_path = target_path - events.append("move-aside") - return original_replace(self, target_path) - if target_path == target and aside_path is not None and self == aside_path: - events.append("restore") - return original_replace(self, target_path) - if target_path == target and self.parent == tmp_path: - events.append("staging-fail") - raise OSError("replace broke") - return original_replace(self, target_path) - - monkeypatch.setattr(oci_image.Path, "replace", flaky_replace) - - _assert_reason(registry, target, "install_failed", wanted=["tool"]) - - assert events == ["move-aside", "staging-fail", "restore"] - assert {path.name: path.read_bytes() for path in target.iterdir()} == before - assert not (target / "tool").exists() - assert {path.name for path in tmp_path.iterdir()} == {"target"} - - -def test_ac7_offline_short_circuit_uses_zero_requests(tmp_path: Path) -> None: - layer = _layer_bytes({"bin/tool": b"tool"}) - target = tmp_path / "target" - verifier_calls = 0 - - def verifier(_image_ref: str, _policy: oci_image.OciSignaturePolicy) -> None: - nonlocal verifier_calls - verifier_calls += 1 - - policy = _policy() - _pull_with_registry( - _registry_for_layers([layer]), - target, - ["tool"], - policy=policy, - verifier=verifier, - ) - assert verifier_calls == 1 - request_count = 0 - - def counting_handler(request: httpx.Request) -> httpx.Response: - nonlocal request_count - request_count += 1 - return httpx.Response(500, request=request) - - with httpx.Client(transport=httpx.MockTransport(counting_handler)) as client: - result = oci_image.pull_and_install( - IMAGE_REF, - "amd64", - ["tool"], - target, - client=client, - policy=policy, - verifier=verifier, - ) - - assert request_count == 0 - assert verifier_calls == 1 - assert result.already_present is True - assert result.files["tool"] == _sha256_file(target / "tool") - - with httpx.Client(transport=httpx.MockTransport(counting_handler)) as client: - with pytest.raises(oci_image.OciImageError) as exc_info: - oci_image.pull_and_install( - ALT_IMAGE_REF, - "amd64", - ["tool"], - target, - client=client, - ) - - assert exc_info.value.reason_code == "token_fetch_failed" - assert request_count == 1 - - -def test_verify_sidecar_install_reports_cached_integrity(tmp_path: Path) -> None: - layer = _layer_bytes( - { - "bin/tool": b"tool", - "usr/lib/libfoo.so": b"library", - } - ) - target = tmp_path / "target" - _pull_with_registry( - _registry_for_layers([layer]), - target, - ["tool", "libfoo.so"], - ) - - assert oci_image.verify_sidecar_install( - IMAGE_REF, - "amd64", - ["tool", "libfoo.so"], - target, - ) - assert not oci_image.verify_sidecar_install( - ALT_IMAGE_REF, - "amd64", - ["tool", "libfoo.so"], - target, - ) - assert not oci_image.verify_sidecar_install( - IMAGE_REF, - "arm64", - ["tool", "libfoo.so"], - target, - ) - - (target / "libfoo.so").unlink() - assert not oci_image.verify_sidecar_install( - IMAGE_REF, - "amd64", - ["tool", "libfoo.so"], - target, - ) - - (target / "libfoo.so").write_bytes(b"wrong") - assert not oci_image.verify_sidecar_install( - IMAGE_REF, - "amd64", - ["tool", "libfoo.so"], - target, + }, ) + rootfs = tmp_path / "rootfs" + oci_image._extract_layer(layer_one, rootfs) + oci_image._extract_layer(layer_two, rootfs) -def test_ac8_offline_short_circuit_uses_request_raising_transport( - tmp_path: Path, -) -> None: - layer = _layer_bytes({"bin/tool": b"tool"}) - target = tmp_path / "target" - _pull_with_registry(_registry_for_layers([layer]), target, ["tool"]) - - def raising_handler(request: httpx.Request) -> httpx.Response: - raise AssertionError(f"unexpected HTTP request: {request.url}") - - with httpx.Client(transport=httpx.MockTransport(raising_handler)) as client: - result = oci_image.pull_and_install( - IMAGE_REF, - "amd64", - ["tool"], - target, - client=client, - ) - - assert result.already_present is True - assert result.files["tool"] == _sha256_file(target / "tool") + assert not (rootfs / "bin" / "tool").exists() + assert not (rootfs / "app" / "old").exists() + assert (rootfs / "app" / "new").read_bytes() == b"new"