Spaces:
Running
Running
| # Copyright 2026 SZL Holdings - SPDX-License-Identifier: Apache-2.0 | |
| """Offline tests for the installed-authority verifier (no network, no secrets).""" | |
| from __future__ import annotations | |
| import ast | |
| import importlib.util | |
| import json | |
| from pathlib import Path | |
| import socket | |
| import sys | |
| import pytest | |
| from cryptography.hazmat.primitives import serialization | |
| from cryptography.hazmat.primitives.asymmetric import ec | |
| SCRIPT = Path(__file__).with_name("verify_installed_authority.py") | |
| SPEC = importlib.util.spec_from_file_location("verify_installed_authority_under_test", SCRIPT) | |
| assert SPEC and SPEC.loader | |
| verifier = importlib.util.module_from_spec(SPEC) | |
| SPEC.loader.exec_module(verifier) | |
| ROOT = SCRIPT.resolve().parent.parent | |
| PINNED_PEM = (ROOT / verifier.PINNED_SIGNING_PUBLIC_KEY_PATH).read_bytes() | |
| ORIGIN = verifier.CANONICAL_ORIGIN | |
| NAMES = {verifier.SIGNING_SECRET, verifier.GDW_SECRET} | |
| SECRET_SENTINEL = "-----BEGIN PRIVATE KEY-----synthetic-must-not-echo" | |
| def static_fallback_pem() -> bytes: | |
| tree = ast.parse((ROOT / "szl_dsse.py").read_text(encoding="utf-8")) | |
| values = [node.value.value for node in tree.body if isinstance(node, ast.Assign) | |
| and getattr(node.targets[0], "id", "") == "COSIGN_PUBLIC_PEM"] | |
| assert len(values) == 1 | |
| return values[0].strip().encode() | |
| def public_pem(curve=ec.SECP256R1()) -> bytes: | |
| return ec.generate_private_key(curve).public_key().public_bytes( | |
| serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) | |
| def serving(body: bytes, status: int = 200, calls: list | None = None): | |
| def get(url): | |
| if calls is not None: | |
| calls.append(url) | |
| if url.endswith("/cosign.pub"): | |
| return status, {"content-type": "text/plain"}, body | |
| return 200, {}, json.dumps({"git_sha": "b" * 40}).encode() | |
| return get | |
| def no_network(monkeypatch): | |
| def refuse(*_args, **_kwargs): | |
| raise AssertionError("tests must stay offline") | |
| monkeypatch.setattr(verifier, "default_get", refuse) | |
| monkeypatch.setattr(socket, "create_connection", refuse) | |
| def test_pinned_file_matches_hardcoded_fingerprint(): | |
| assert verifier.pinned_fingerprint() == verifier.PINNED_SIGNING_KEY_DER_SHA256 | |
| assert verifier.static_fallback_fingerprint() != verifier.PINNED_SIGNING_KEY_DER_SHA256 | |
| def test_pinned_runtime_key_is_verified(): | |
| calls = [] | |
| report = verifier.verify_installed_authority(NAMES, (), origin=ORIGIN, get=serving(PINNED_PEM, calls=calls)) | |
| assert report["credential_authority_state"] == "VERIFIED" | |
| assert report["diagnostic_code"] == "INSTALLED_AUTHORITY_VERIFIED" | |
| assert report["signing"]["state"] == verifier.VERIFIED_PINNED_RUNTIME_KEY | |
| assert report["signing"]["served_fingerprint_sha256"] == verifier.PINNED_SIGNING_KEY_DER_SHA256 | |
| assert report["signing"]["fresh_signature_verified"] is False | |
| assert report["gdw"]["state"] == verifier.GDW_NAME_PRESENT | |
| assert report["live_git_sha"] == "b" * 40 | |
| assert report["origin"] == ORIGIN | |
| assert report["observed_at_start"] and report["observed_at_end"] | |
| assert calls[0] == ORIGIN + "/cosign.pub" | |
| assert report["secret_values_read"] is False and report["secret_values_written"] is False | |
| def test_trailing_slash_origin_and_whitespace_pem_are_normalized(): | |
| body = b"\n" + PINNED_PEM.replace(b"\n", b"\r\n") + b"\n" | |
| report = verifier.verify_signing_authority(ORIGIN + "/", serving(body)) | |
| assert report["state"] == verifier.VERIFIED_PINNED_RUNTIME_KEY | |
| def test_static_fallback_key_means_private_key_not_installed(): | |
| report = verifier.verify_installed_authority(NAMES, (), origin=ORIGIN, get=serving(static_fallback_pem())) | |
| assert report["signing"]["state"] == verifier.SIGNING_KEY_NOT_INSTALLED | |
| assert report["credential_authority_state"] == "BLOCKED" | |
| assert report["diagnostic_code"] == "SIGNING_KEY_NOT_INSTALLED" | |
| def test_foreign_or_malformed_key_is_a_mismatch(body): | |
| report = verifier.verify_installed_authority(NAMES, (), origin=ORIGIN, get=serving(body)) | |
| assert report["signing"]["state"] == verifier.SIGNING_KEY_MISMATCH | |
| assert report["credential_authority_state"] == "BLOCKED" | |
| assert "must-not-echo" not in json.dumps(report) | |
| def raising(exc): | |
| def get(_url): | |
| raise exc | |
| return get | |
| def test_unavailable_origin_is_unknown_never_verified(get): | |
| report = verifier.verify_installed_authority(NAMES, (), origin=ORIGIN, get=get) | |
| assert report["signing"]["state"] == verifier.ORIGIN_UNAVAILABLE | |
| assert report["credential_authority_state"] == "UNKNOWN" | |
| assert report["diagnostic_code"] == "AUTHORITY_ORIGIN_UNAVAILABLE" | |
| assert report["signing"]["served_fingerprint_sha256"] is None | |
| assert "must-not-echo" not in json.dumps(report) | |
| def test_default_get_refuses_redirects_and_bounds_body(monkeypatch): | |
| # The real getter surfaces a 3xx as a non-200 status instead of following it. | |
| import http.server | |
| import threading | |
| class Redirect(http.server.BaseHTTPRequestHandler): | |
| def do_GET(self): # noqa: N802 - stdlib hook | |
| self.send_response(302) | |
| self.send_header("Location", "http://127.0.0.1:9/cosign.pub") | |
| self.end_headers() | |
| def log_message(self, *_args): | |
| pass | |
| monkeypatch.undo() # this single test uses loopback only | |
| server = http.server.HTTPServer(("127.0.0.1", 0), Redirect) | |
| thread = threading.Thread(target=server.handle_request, daemon=True) | |
| thread.start() | |
| try: | |
| status, _headers, body = SPEC_DEFAULT_GET(f"http://127.0.0.1:{server.server_port}/cosign.pub") | |
| finally: | |
| server.server_close() | |
| assert status == 302 and body == b"" | |
| SPEC_DEFAULT_GET = verifier.default_get | |
| def test_noncanonical_origin_is_never_fetched(origin): | |
| calls = [] | |
| report = verifier.verify_installed_authority(NAMES, (), origin=origin, get=serving(PINNED_PEM, calls=calls)) | |
| assert calls == [] | |
| assert report["credential_authority_state"] == "UNKNOWN" | |
| assert report["diagnostic_code"] == "AUTHORITY_ORIGIN_NOT_CANONICAL" | |
| def test_gdw_name_missing_is_blocking(): | |
| report = verifier.verify_installed_authority({verifier.SIGNING_SECRET}, (), origin=ORIGIN, get=serving(PINNED_PEM)) | |
| assert report["signing"]["state"] == verifier.VERIFIED_PINNED_RUNTIME_KEY | |
| assert report["gdw"] == {**report["gdw"], "state": verifier.GDW_CREDENTIALS_MISSING, "blocking": True} | |
| assert report["credential_authority_state"] == "BLOCKED" | |
| assert report["diagnostic_code"] == "GDW_CREDENTIALS_MISSING" | |
| def test_signing_secret_name_missing_is_blocking_even_with_pinned_key(): | |
| report = verifier.verify_installed_authority({verifier.GDW_SECRET}, (), origin=ORIGIN, get=serving(PINNED_PEM)) | |
| assert report["credential_authority_state"] == "BLOCKED" | |
| assert report["diagnostic_code"] == "SIGNING_SECRET_MISSING" | |
| def test_github_reader_is_optional_and_never_blocking(present): | |
| names = NAMES | ({verifier.GITHUB_PUBLIC_READ_SECRET} if present else set()) | |
| report = verifier.verify_installed_authority(names, (), origin=ORIGIN, get=serving(PINNED_PEM)) | |
| assert report["github_public_reader"]["state"] == ("INSTALLED_NAME_ONLY" if present else "PUBLIC_ANONYMOUS") | |
| assert report["github_public_reader"]["blocking"] is False | |
| assert report["github_public_reader"]["value_verified"] is False | |
| assert report["credential_authority_state"] == "VERIFIED" | |
| def test_required_or_optional_name_collision_blocks(variable): | |
| report = verifier.verify_installed_authority(NAMES, {variable}, origin=ORIGIN, get=serving(PINNED_PEM)) | |
| assert report["credential_authority_state"] == "BLOCKED" | |
| assert report["diagnostic_code"] == "PUBLIC_VARIABLE_COLLISION" | |
| def test_pin_file_disagreeing_with_hardcoded_fingerprint_fails_closed(tmp_path): | |
| foreign = public_pem() | |
| pin = tmp_path / "pin.pub" | |
| pin.write_bytes(foreign) | |
| # Even when the runtime serves exactly the (tampered) pin file's key. | |
| report = verifier.verify_installed_authority(NAMES, (), origin=ORIGIN, get=serving(foreign), pinned_pem_path=pin) | |
| assert report["signing"]["state"] == verifier.SIGNING_KEY_MISMATCH | |
| assert report["signing"]["diagnostic_code"] == "PINNED_KEY_INCONSISTENT" | |
| assert report["credential_authority_state"] == "BLOCKED" | |
| def test_hardcoded_fingerprint_disagreeing_with_pin_file_fails_closed(monkeypatch): | |
| monkeypatch.setattr(verifier, "PINNED_SIGNING_KEY_DER_SHA256", "0" * 64) | |
| report = verifier.verify_installed_authority(NAMES, (), origin=ORIGIN, get=serving(PINNED_PEM)) | |
| assert report["signing"]["state"] == verifier.SIGNING_KEY_MISMATCH | |
| assert report["signing"]["diagnostic_code"] == "PINNED_KEY_INCONSISTENT" | |
| assert report["credential_authority_state"] == "BLOCKED" | |
| def test_missing_oversized_or_malformed_pin_file_fails_closed(tmp_path, content): | |
| pin = tmp_path / "pin.pub" | |
| pin.write_bytes(content) | |
| report = verifier.verify_signing_authority(ORIGIN, serving(PINNED_PEM), pin) | |
| assert report["state"] == verifier.SIGNING_KEY_MISMATCH | |
| assert report["diagnostic_code"] == "PINNED_KEY_INCONSISTENT" | |
| missing = verifier.verify_signing_authority(ORIGIN, serving(PINNED_PEM), tmp_path / "absent.pub") | |
| assert missing["diagnostic_code"] == "PINNED_KEY_INCONSISTENT" | |
| def test_pin_equal_to_static_fallback_fails_closed(tmp_path): | |
| static = tmp_path / "szl_dsse.py" | |
| static.write_text("COSIGN_PUBLIC_PEM = " + repr(PINNED_PEM.decode()) + "\n", encoding="utf-8") | |
| report = verifier.verify_signing_authority(ORIGIN, serving(PINNED_PEM), static_source_path=static) | |
| assert report["state"] == verifier.SIGNING_KEY_MISMATCH | |
| assert report["diagnostic_code"] == "PINNED_KEY_INCONSISTENT" | |
| def test_module_has_no_provider_import_and_report_is_bounded(): | |
| tree = ast.parse(SCRIPT.read_text(encoding="utf-8")) | |
| imported = set() | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.Import): | |
| imported |= {alias.name.split(".")[0] for alias in node.names} | |
| elif isinstance(node, ast.ImportFrom): | |
| imported.add((node.module or "").split(".")[0]) | |
| assert imported <= set(sys.stdlib_module_names) | {"cryptography"} | |
| assert "huggingface_hub" not in imported | |
| report = verifier.verify_installed_authority(NAMES, (), origin=ORIGIN, get=serving(PINNED_PEM)) | |
| assert len(json.dumps(report).encode()) < 4 * 1024 | |