diff --git a/.github/workflows/app-ci.yml b/.github/workflows/app-ci.yml index d5b99d42c..35e57cd46 100644 --- a/.github/workflows/app-ci.yml +++ b/.github/workflows/app-ci.yml @@ -59,6 +59,19 @@ jobs: } >> "$GITHUB_STEP_SUMMARY" exit "$status" + - name: Validate PyPI release hash provenance + run: | + status=0 + receipt="$(python scripts/ci/python_lock_registry_provenance.py --json)" || status=$? + printf '%s\n' "$receipt" + { + echo '### PyPI release hash provenance' + echo '```json' + printf '%s\n' "$receipt" + echo '```' + } >> "$GITHUB_STEP_SUMMARY" + exit "$status" + - name: Install backend dependencies run: | python -m pip install --disable-pip-version-check --require-hashes -r backend/requirements-hashes.txt diff --git a/backend/core/local_http.py b/backend/core/local_http.py index a0e7e6691..97aa25575 100644 --- a/backend/core/local_http.py +++ b/backend/core/local_http.py @@ -70,7 +70,11 @@ def validate_loopback_http_origin(value: str) -> LocalHTTPOrigin: safe_hostname = address.compressed try: - port = parsed.port or (443 if parsed.scheme == "https" else 80) + port = ( + parsed.port + if parsed.port is not None + else (443 if parsed.scheme == "https" else 80) + ) except ValueError as exc: raise LocalHTTPValidationError("local HTTP origin port is invalid") from exc if not 1 <= port <= 65535: diff --git a/backend/tests/test_local_http.py b/backend/tests/test_local_http.py index 11dc87a8d..6a99d9c04 100644 --- a/backend/tests/test_local_http.py +++ b/backend/tests/test_local_http.py @@ -15,6 +15,18 @@ def test_loopback_origin_is_canonicalized() -> None: hostname="::1", port=18080, ) + assert validate_loopback_http_origin("http://localhost") == LocalHTTPOrigin( + origin="http://localhost", + scheme="http", + hostname="localhost", + port=80, + ) + assert validate_loopback_http_origin("https://127.0.0.1:443/") == LocalHTTPOrigin( + origin="https://127.0.0.1", + scheme="https", + hostname="127.0.0.1", + port=443, + ) @pytest.mark.parametrize( @@ -32,6 +44,64 @@ def test_loopback_origin_normalizes_malformed_parser_errors(value: str) -> None: validate_loopback_http_origin(value) +@pytest.mark.parametrize( + "value", + [ + "http://localhost:80\x00/", + "http://\nlocalhost/", + ], +) +def test_loopback_origin_rejects_control_characters(value: str) -> None: + with pytest.raises(LocalHTTPValidationError, match="control characters"): + validate_loopback_http_origin(value) + + +@pytest.mark.parametrize( + "value", + [ + "ftp://localhost/", + "http://user:pass@localhost/", + "http://localhost/path", + "http://localhost/?query=1", + "http://localhost/#frag", + "http:///", # No hostname + ], +) +def test_loopback_origin_rejects_invalid_components(value: str) -> None: + with pytest.raises( + LocalHTTPValidationError, match=r"must be a loopback HTTP\(S\) origin" + ): + validate_loopback_http_origin(value) + + +@pytest.mark.parametrize( + "value", + [ + "http://example.com/", + "http://192.168.1.1/", + "http://[2001:db8::1]/", + "http://invalid.localhost/", + ], +) +def test_loopback_origin_rejects_non_allowlisted_hosts(value: str) -> None: + with pytest.raises(LocalHTTPValidationError, match="host is not allowlisted"): + validate_loopback_http_origin(value) + + +@pytest.mark.parametrize( + "value", + [ + "http://localhost:-1/", + "http://localhost:65536/", + "http://localhost:abc/", + "http://localhost:0/", + ], +) +def test_loopback_origin_rejects_invalid_ports(value: str) -> None: + with pytest.raises(LocalHTTPValidationError, match="port is invalid"): + validate_loopback_http_origin(value) + + def test_local_request_target_preserves_safe_path_and_query() -> None: assert ( validate_local_request_target("/api/emails?limit=10") == "/api/emails?limit=10" diff --git a/backend/tests/test_python_lock_registry_non_vacuous.py b/backend/tests/test_python_lock_registry_non_vacuous.py new file mode 100644 index 000000000..319be3ea8 --- /dev/null +++ b/backend/tests/test_python_lock_registry_non_vacuous.py @@ -0,0 +1,114 @@ +"""Non-vacuous evidence contracts for PyPI lock provenance.""" + +from __future__ import annotations + +import importlib.util +import json +import runpy +import sys +import urllib.request +from pathlib import Path +from typing import Any + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT_PATH = REPO_ROOT / "scripts" / "ci" / "python_lock_registry_provenance.py" +_spec = importlib.util.spec_from_file_location("python_lock_registry_non_vacuous", SCRIPT_PATH) +assert _spec is not None and _spec.loader is not None +registry_provenance = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = registry_provenance +_spec.loader.exec_module(registry_provenance) + + +def _sha() -> str: + """Return one deterministic SHA-256 fixture digest.""" + return "a" * 64 + + +def _codes(receipt: dict[str, object]) -> set[str]: + """Return stable top-level violation codes from a repository receipt.""" + return {str(item["code"]) for item in receipt["violations"]} + + +def test_repository_without_hash_locks_fails_non_vacuously(tmp_path: Path) -> None: + """A green registry receipt must represent at least one discovered hash lock.""" + (tmp_path / "requirements.txt").write_text("example==1.0\n", encoding="utf-8") + + receipt = registry_provenance.validate_repository_registry( + tmp_path, + fetch_release=lambda project, version: {}, + ) + + assert receipt["status"] == "failed" + assert receipt["lock_files"] == [] + assert _codes(receipt) == {"registry-no-hash-locks"} + + +class _Response: + """Minimal exact-origin PyPI response used by the script-entrypoint test.""" + + def __init__(self, url: str) -> None: + self.url = url + self.headers = {"Content-Type": "application/json"} + self.payload = json.dumps( + { + "info": {"name": "example", "version": "1.0"}, + "urls": [ + { + "packagetype": "sdist", + "yanked": False, + "digests": {"sha256": _sha()}, + } + ], + } + ).encode("utf-8") + + def __enter__(self) -> "_Response": + return self + + def __exit__(self, *args: Any) -> None: + return None + + def geturl(self) -> str: + """Return the unchanged trusted request URL.""" + return self.url + + def read(self, size: int) -> bytes: + """Return a bounded response body.""" + return self.payload[:size] + + +def test_script_main_guard_runs_registry_validation( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """Executing the script as __main__ publishes a passing JSON receipt and exits zero.""" + lock = tmp_path / "requirements-hashes.txt" + lock.write_text( + f"example==1.0 \\\n --hash=sha256:{_sha()}\n", + encoding="utf-8", + ) + + class _Opener: + def open(self, request: urllib.request.Request, timeout: float) -> _Response: + return _Response(request.full_url) + + monkeypatch.setattr(urllib.request, "build_opener", lambda *handlers: _Opener()) + monkeypatch.setattr( + sys, + "argv", + [ + str(SCRIPT_PATH), + "--repository-root", + str(tmp_path), + "--json", + ], + ) + + with pytest.raises(SystemExit) as exit_info: + runpy.run_path(str(SCRIPT_PATH), run_name="__main__") + + assert exit_info.value.code == 0 + assert json.loads(capsys.readouterr().out)["status"] == "passed" diff --git a/backend/tests/test_python_lock_registry_provenance.py b/backend/tests/test_python_lock_registry_provenance.py new file mode 100644 index 000000000..b8d1e286d --- /dev/null +++ b/backend/tests/test_python_lock_registry_provenance.py @@ -0,0 +1,251 @@ +"""Contract tests for PyPI release-hash provenance of Python lock files. + +The network-backed validator is a second, stacked supply-chain boundary after the +offline declaration validator. Tests inject release metadata so normal unit tests +remain deterministic and never depend on public network availability. +""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path + +import pytest +import yaml + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT_PATH = REPO_ROOT / "scripts" / "ci" / "python_lock_registry_provenance.py" + +_spec = importlib.util.spec_from_file_location( + "python_lock_registry_provenance", SCRIPT_PATH +) +assert _spec is not None and _spec.loader is not None +registry_provenance = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = registry_provenance +_spec.loader.exec_module(registry_provenance) + + +def _sha(character: str) -> str: + """Return one syntactically valid SHA-256 digest for fixtures.""" + return character * 64 + + +def _write_lock(path: Path, *, digest: str, version: str = "1.0") -> Path: + """Write one exact hash-pinned requirement and return its path.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + f"example=={version} \\\n --hash=sha256:{digest}\n", + encoding="utf-8", + ) + return path + + +def _release_metadata( + *, + digest: str, + version: str = "1.0", + yanked: bool = False, + package_type: str = "bdist_wheel", +) -> dict[str, object]: + """Return a minimal PyPI release JSON payload with one artifact.""" + return { + "info": {"name": "example", "version": version}, + "urls": [ + { + "filename": f"example-{version}-py3-none-any.whl", + "packagetype": package_type, + "yanked": yanked, + "digests": {"sha256": digest}, + "url": "https://files.pythonhosted.org/private-looking-path.whl", + } + ], + } + + +def _codes(receipt: dict[str, object]) -> set[str]: + """Return stable violation codes from a registry provenance receipt.""" + violations = receipt["violations"] + assert isinstance(violations, list) + return {str(item["code"]) for item in violations} + + +def test_matching_non_yanked_registry_artifact_hash_passes(tmp_path: Path) -> None: + """A lock hash is accepted only when PyPI publishes it for the exact release.""" + digest = _sha("a") + lock_path = _write_lock(tmp_path / "requirements-hashes.txt", digest=digest) + + receipt = registry_provenance.validate_lock_against_registry( + lock_path, + tmp_path, + fetch_release=lambda project, version: _release_metadata( + digest=digest, version=version + ), + ) + + assert receipt["status"] == "passed" + assert receipt["path"] == "requirements-hashes.txt" + assert receipt["requirements"] == [ + { + "project": "example", + "version": "1.0", + "status": "passed", + "matched_artifact_count": 1, + } + ] + assert receipt["violations"] == [] + assert "pythonhosted" not in json.dumps(receipt, sort_keys=True) + + +def test_stale_lock_hash_fails_with_stable_code(tmp_path: Path) -> None: + """A syntactically valid but non-registry SHA-256 cannot attest a release.""" + lock_path = _write_lock( + tmp_path / "requirements-hashes.txt", digest=_sha("a") + ) + + receipt = registry_provenance.validate_lock_against_registry( + lock_path, + tmp_path, + fetch_release=lambda project, version: _release_metadata( + digest=_sha("b"), version=version + ), + ) + + assert receipt["status"] == "failed" + assert _codes(receipt) == {"registry-hash-mismatch"} + + +def test_yanked_or_unknown_artifacts_do_not_satisfy_provenance( + tmp_path: Path, +) -> None: + """Only non-yanked wheel/sdist artifacts are eligible provenance evidence.""" + digest = _sha("c") + lock_path = _write_lock(tmp_path / "requirements-hashes.txt", digest=digest) + metadata = { + "info": {"name": "example", "version": "1.0"}, + "urls": [ + _release_metadata(digest=digest, yanked=True)["urls"][0], + _release_metadata(digest=digest, package_type="unknown")["urls"][0], + ], + } + + receipt = registry_provenance.validate_lock_against_registry( + lock_path, + tmp_path, + fetch_release=lambda project, version: metadata, + ) + + assert receipt["status"] == "failed" + assert _codes(receipt) == {"registry-release-has-no-allowed-artifacts"} + + +def test_release_identity_mismatch_fails_closed(tmp_path: Path) -> None: + """Metadata for another project or version cannot satisfy the requested pin.""" + digest = _sha("d") + lock_path = _write_lock(tmp_path / "requirements-hashes.txt", digest=digest) + metadata = _release_metadata(digest=digest) + metadata["info"] = {"name": "other-project", "version": "9.9"} + + receipt = registry_provenance.validate_lock_against_registry( + lock_path, + tmp_path, + fetch_release=lambda project, version: metadata, + ) + + assert receipt["status"] == "failed" + assert _codes(receipt) == { + "registry-project-mismatch", + "registry-version-mismatch", + } + + +def test_registry_fetch_failure_does_not_serialize_provider_details( + tmp_path: Path, +) -> None: + """Transient provider errors fail closed without copying exception text to CI.""" + lock_path = _write_lock( + tmp_path / "requirements-hashes.txt", digest=_sha("e") + ) + + def failing_fetch(project: str, version: str) -> dict[str, object]: + raise RuntimeError("SECRET_TOKEN=https://private.invalid/token") + + receipt = registry_provenance.validate_lock_against_registry( + lock_path, + tmp_path, + fetch_release=failing_fetch, + ) + serialized = json.dumps(receipt, sort_keys=True) + + assert receipt["status"] == "failed" + assert _codes(receipt) == {"registry-metadata-fetch-failed"} + assert "SECRET_TOKEN" not in serialized + assert "private.invalid" not in serialized + + +def test_repository_registry_receipt_deduplicates_release_fetches( + tmp_path: Path, +) -> None: + """The same project/version across multiple locks is resolved only once.""" + digest = _sha("f") + _write_lock(tmp_path / "backend" / "requirements-hashes.txt", digest=digest) + _write_lock(tmp_path / "connector" / "requirements-hashes.txt", digest=digest) + calls: list[tuple[str, str]] = [] + + def fetch_release(project: str, version: str) -> dict[str, object]: + calls.append((project, version)) + return _release_metadata(digest=digest, version=version) + + receipt = registry_provenance.validate_repository_registry( + tmp_path, + fetch_release=fetch_release, + ) + + assert receipt["status"] == "passed" + assert calls == [("example", "1.0")] + assert [item["path"] for item in receipt["lock_files"]] == [ + "backend/requirements-hashes.txt", + "connector/requirements-hashes.txt", + ] + assert receipt["schema_version"] == "naruon.python-lock-registry-provenance.v1" + + +def test_pypi_release_fetch_contract_rejects_untrusted_origin() -> None: + """The built-in network client only accepts credential-free HTTPS PyPI.""" + for origin in ( + "http://pypi.org", + "https://user:secret@pypi.org", + "https://example.invalid", + "https://pypi.org/path", + "https://pypi.org?token=secret", + ): + with pytest.raises(ValueError, match="trusted PyPI origin"): + registry_provenance.build_pypi_release_url( + "example", "1.0", pypi_origin=origin + ) + + assert registry_provenance.build_pypi_release_url("Example_Pkg", "1.0") == ( + "https://pypi.org/pypi/example-pkg/1.0/json" + ) + + +def test_application_ci_runs_registry_provenance_before_dependency_install() -> None: + """Application CI must publish registry evidence before installing backend code.""" + workflow = yaml.safe_load( + (REPO_ROOT / ".github" / "workflows" / "app-ci.yml").read_text( + encoding="utf-8" + ) + ) + backend_job = workflow["jobs"]["backend"] + steps = backend_job["steps"] + names = [step.get("name") for step in steps] + registry_index = names.index("Validate PyPI release hash provenance") + install_index = names.index("Install backend dependencies") + assert registry_index < install_index + + registry_step = steps[registry_index] + command = registry_step["run"] + assert "python scripts/ci/python_lock_registry_provenance.py --json" in command + assert "GITHUB_STEP_SUMMARY" in command + assert 'exit "$status"' in command diff --git a/backend/tests/test_python_lock_registry_provenance_edges.py b/backend/tests/test_python_lock_registry_provenance_edges.py new file mode 100644 index 000000000..140aaa282 --- /dev/null +++ b/backend/tests/test_python_lock_registry_provenance_edges.py @@ -0,0 +1,250 @@ +"""Edge and transport tests for the PyPI lock-provenance validator.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path +from typing import Any + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT_PATH = REPO_ROOT / "scripts" / "ci" / "python_lock_registry_provenance.py" +_spec = importlib.util.spec_from_file_location("python_lock_registry_edges", SCRIPT_PATH) +assert _spec is not None and _spec.loader is not None +registry_provenance = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = registry_provenance +_spec.loader.exec_module(registry_provenance) + + +def _sha(character: str = "a") -> str: + """Return a fixture SHA-256 digest.""" + return character * 64 + + +def _write(path: Path, text: str) -> Path: + """Write one UTF-8 fixture path.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return path + + +def _codes(receipt: dict[str, object]) -> set[str]: + """Return stable violation codes from a receipt.""" + return {str(item["code"]) for item in receipt["violations"]} + + +def _metadata(digest: str) -> dict[str, object]: + """Return one eligible exact-release metadata fixture.""" + return { + "info": {"name": "example", "version": "1.0"}, + "urls": [ + { + "packagetype": "sdist", + "yanked": False, + "digests": {"sha256": digest}, + } + ], + } + + +def test_lock_parser_fails_closed_on_structure_errors(tmp_path: Path) -> None: + """Orphan hashes, non-exact pins, and missing hashes are separately visible.""" + path = _write( + tmp_path / "requirements-hashes.txt", + f"--hash=sha256:{_sha()}\nexample>=1\nother==2.0\n", + ) + receipt = registry_provenance.validate_lock_against_registry( + path, + tmp_path, + fetch_release=lambda project, version: _metadata(_sha("b")), + ) + assert receipt["status"] == "failed" + assert { + "lock-orphan-sha256", + "lock-requirement-not-exact", + "lock-requirement-has-no-sha256", + }.issubset(_codes(receipt)) + + +def test_outside_symlink_is_rejected_without_reading_payload(tmp_path: Path) -> None: + """A discovered lock symlink cannot exfiltrate an external file.""" + root = tmp_path / "repo" + root.mkdir() + outside = _write(tmp_path / "outside.txt", "TOP_SECRET>=1\n") + (root / "requirements-hashes.txt").symlink_to(outside) + receipt = registry_provenance.validate_repository_registry( + root, + fetch_release=lambda project, version: _metadata(_sha()), + ) + serialized = json.dumps(receipt, sort_keys=True) + assert receipt["status"] == "failed" + assert _codes(receipt["lock_files"][0]) == {"lock-path-outside-repository"} + assert "TOP_SECRET" not in serialized + assert str(tmp_path) not in serialized + + +def test_unreadable_utf8_lock_is_ignored_by_discovery(tmp_path: Path) -> None: + """Binary requirements candidates are not interpreted as provenance locks.""" + (tmp_path / "requirements-hashes.txt").write_bytes(b"\xff\xfe") + assert registry_provenance.discover_hash_locks(tmp_path) == [] + + +def test_direct_invalid_utf8_lock_returns_stable_read_failure(tmp_path: Path) -> None: + """Direct validation reports a generic read failure without raw bytes.""" + path = tmp_path / "requirements-hashes.txt" + path.write_bytes(b"\xff\xfe") + receipt = registry_provenance.validate_lock_against_registry(path, tmp_path) + assert _codes(receipt) == {"lock-read-failed"} + assert receipt["requirements"] == [] + + +def test_artifact_filter_ignores_malformed_registry_entries() -> None: + """Only non-yanked wheel/sdist objects with valid SHA-256 values count.""" + assert registry_provenance._eligible_registry_hashes({"urls": "bad"}) == set() + metadata = { + "urls": [ + "bad", + {"packagetype": "sdist", "yanked": True, "digests": {"sha256": _sha()}}, + {"packagetype": "other", "yanked": False, "digests": {"sha256": _sha()}}, + {"packagetype": "sdist", "yanked": False, "digests": "bad"}, + {"packagetype": "sdist", "yanked": False, "digests": {"sha256": "bad"}}, + {"packagetype": "bdist_wheel", "yanked": False, "digests": {"sha256": _sha("c").upper()}}, + ] + } + assert registry_provenance._eligible_registry_hashes(metadata) == {_sha("c")} + + +class _Headers(dict[str, str]): + """Minimal urllib-compatible response header mapping.""" + + +class _Response: + """Minimal context-managed urllib response for transport tests.""" + + def __init__(self, payload: bytes, content_type: str = "application/json") -> None: + self.payload = payload + self.headers = _Headers({"Content-Type": content_type}) + + def __enter__(self) -> "_Response": + return self + + def __exit__(self, *args: Any) -> None: + return None + + def read(self, size: int) -> bytes: + return self.payload[:size] + + +def test_fetch_pypi_release_enforces_bounds_and_json_shape(monkeypatch: pytest.MonkeyPatch) -> None: + """The real transport validates configuration, media type, size, and JSON shape.""" + payload = json.dumps(_metadata(_sha())).encode() + monkeypatch.setattr( + registry_provenance, + "_open_pypi_request", + lambda request, *, timeout_seconds: _Response(payload), + ) + assert registry_provenance.fetch_pypi_release("example", "1.0")["info"] == { + "name": "example", + "version": "1.0", + } + + for kwargs in ({"timeout_seconds": 0}, {"max_metadata_bytes": 0}): + with pytest.raises(ValueError): + registry_provenance.fetch_pypi_release("example", "1.0", **kwargs) + + monkeypatch.setattr( + registry_provenance, + "_open_pypi_request", + lambda request, *, timeout_seconds: _Response(payload, "text/plain"), + ) + with pytest.raises(ValueError, match="must be JSON"): + registry_provenance.fetch_pypi_release("example", "1.0") + + monkeypatch.setattr( + registry_provenance, + "_open_pypi_request", + lambda request, *, timeout_seconds: _Response(b"{}x"), + ) + with pytest.raises(ValueError, match="byte limit"): + registry_provenance.fetch_pypi_release( + "example", "1.0", max_metadata_bytes=2 + ) + + monkeypatch.setattr( + registry_provenance, + "_open_pypi_request", + lambda request, *, timeout_seconds: _Response(b"[]"), + ) + with pytest.raises(ValueError, match="JSON object"): + registry_provenance.fetch_pypi_release("example", "1.0") + + +def test_origin_validation_rejects_invalid_port_and_fragment() -> None: + """Malformed authority and fragment-bearing origins fail before network use.""" + for origin in ("https://pypi.org:bad", "https://pypi.org/#fragment"): + with pytest.raises(ValueError, match="trusted PyPI origin"): + registry_provenance.build_pypi_release_url( + "example", "1.0", pypi_origin=origin + ) + + +def test_cached_registry_failure_is_not_retried_per_lock(tmp_path: Path) -> None: + """One failed exact release resolution is shared across repeated lock entries.""" + for directory in ("a", "b"): + _write( + tmp_path / directory / "requirements-hashes.txt", + f"example==1.0 \\\n --hash=sha256:{_sha()}\n", + ) + calls = 0 + + def fail_once(project: str, version: str) -> dict[str, object]: + nonlocal calls + calls += 1 + raise RuntimeError("provider unavailable") + + receipt = registry_provenance.validate_repository_registry( + tmp_path, + fetch_release=fail_once, + ) + assert calls == 1 + assert receipt["status"] == "failed" + assert all( + _codes(lock) == {"registry-metadata-fetch-failed"} + for lock in receipt["lock_files"] + ) + + +def test_main_json_and_human_output(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + """CLI output preserves deterministic pass/fail exit semantics.""" + monkeypatch.setattr( + registry_provenance, + "validate_repository_registry", + lambda root: { + "schema_version": registry_provenance.SCHEMA_VERSION, + "status": "passed", + "lock_files": [], + "violations": [], + }, + ) + assert registry_provenance.main(["--json"]) == 0 + assert json.loads(capsys.readouterr().out)["status"] == "passed" + + monkeypatch.setattr( + registry_provenance, + "validate_repository_registry", + lambda root: { + "schema_version": registry_provenance.SCHEMA_VERSION, + "status": "failed", + "lock_files": [], + "violations": [ + {"code": "registry-hash-mismatch", "path": "lock.txt", "detail": "mismatch"} + ], + }, + ) + assert registry_provenance.main([]) == 1 + output = capsys.readouterr().out + assert "Python lock PyPI provenance: failed" in output + assert "registry-hash-mismatch: lock.txt: mismatch" in output diff --git a/backend/tests/test_python_lock_registry_redirect_policy.py b/backend/tests/test_python_lock_registry_redirect_policy.py new file mode 100644 index 000000000..7109d1892 --- /dev/null +++ b/backend/tests/test_python_lock_registry_redirect_policy.py @@ -0,0 +1,94 @@ +"""Redirect-origin contract for the PyPI lock-provenance transport.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path +from typing import Any + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT_PATH = REPO_ROOT / "scripts" / "ci" / "python_lock_registry_provenance.py" +_spec = importlib.util.spec_from_file_location("python_lock_registry_redirect", SCRIPT_PATH) +assert _spec is not None and _spec.loader is not None +registry_provenance = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = registry_provenance +_spec.loader.exec_module(registry_provenance) + + +class _RedirectedResponse: + """Minimal urllib response exposing the final URL after redirect handling.""" + + def __init__(self, final_url: str) -> None: + self._final_url = final_url + self.headers = {"Content-Type": "application/json"} + self._payload = json.dumps( + { + "info": {"name": "example", "version": "1.0"}, + "urls": [], + } + ).encode("utf-8") + + def __enter__(self) -> "_RedirectedResponse": + return self + + def __exit__(self, *args: Any) -> None: + return None + + def geturl(self) -> str: + """Return the final response URL observed by urllib.""" + return self._final_url + + def read(self, size: int) -> bytes: + """Return a bounded JSON payload.""" + return self._payload[:size] + + +def test_fetch_rejects_redirect_to_non_pypi_origin( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An HTTPS redirect must not move trusted metadata reads off pypi.org.""" + monkeypatch.setattr( + registry_provenance, + "_open_pypi_request", + lambda request, *, timeout_seconds: _RedirectedResponse( + "https://metadata.attacker.invalid/pypi/example/1.0/json" + ), + ) + + with pytest.raises(ValueError, match="trusted PyPI origin"): + registry_provenance.fetch_pypi_release("example", "1.0") + + +def test_fetch_accepts_final_exact_pypi_release_url( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A response that remains on the exact requested PyPI URL is accepted.""" + expected_url = registry_provenance.build_pypi_release_url("example", "1.0") + monkeypatch.setattr( + registry_provenance, + "_open_pypi_request", + lambda request, *, timeout_seconds: _RedirectedResponse(expected_url), + ) + + metadata = registry_provenance.fetch_pypi_release("example", "1.0") + + assert metadata["info"] == {"name": "example", "version": "1.0"} + + +def test_redirect_handler_returns_no_follow_request() -> None: + """The transport handler refuses to construct a request for a redirect target.""" + request = registry_provenance._NoRedirectHandler().redirect_request( + registry_provenance.urllib.request.Request( + "https://pypi.org/pypi/example/1.0/json" + ), + 302, + "Found", + {"Location": "https://metadata.attacker.invalid/"}, + "https://pypi.org/pypi/example/1.0/json", + ) + + assert request is None diff --git a/backend/tests/test_url_validation.py b/backend/tests/test_url_validation.py index 2857f61a2..05c75ed43 100644 --- a/backend/tests/test_url_validation.py +++ b/backend/tests/test_url_validation.py @@ -5,6 +5,7 @@ from core.url_validation import ( parse_allowed_hosts, validate_https_url_host, + validate_same_or_subdomain_host, validate_https_url_host_details, _normalize_host, _reject_unsafe_ip_literal, @@ -12,6 +13,7 @@ _resolve_global_addresses, ) + def test_parse_allowed_hosts(): assert parse_allowed_hosts("example.com, TEST.COM. , [2001:db8::1]") == frozenset( {"example.com", "test.com", "2001:db8::1"} @@ -22,11 +24,13 @@ def test_parse_allowed_hosts(): {"example.com", "example.net"} ) + def test_normalize_host(): assert _normalize_host(" Example.COM. ") == "example.com" assert _normalize_host("[2001:db8::1]") == "2001:db8::1" assert _normalize_host("test") == "test" + def test_reject_unsafe_ip_literal(): # Safe global IP _reject_unsafe_ip_literal("setting", "8.8.8.8") @@ -38,69 +42,109 @@ def test_reject_unsafe_ip_literal(): with pytest.raises(ValueError, match="setting IP host must be globally routable"): _reject_unsafe_ip_literal("setting", "::1") - with pytest.raises(ValueError, match="setting host must not be a local or internal domain"): + with pytest.raises( + ValueError, match="setting host must not be a local or internal domain" + ): _reject_unsafe_ip_literal("setting", "localhost") - with pytest.raises(ValueError, match="setting host must not be a local or internal domain"): + with pytest.raises( + ValueError, match="setting host must not be a local or internal domain" + ): _reject_unsafe_ip_literal("setting", "test.localhost") - with pytest.raises(ValueError, match="setting host must not be a local or internal domain"): + with pytest.raises( + ValueError, match="setting host must not be a local or internal domain" + ): _reject_unsafe_ip_literal("setting", "internal") - with pytest.raises(ValueError, match="setting host must not be a local or internal domain"): + with pytest.raises( + ValueError, match="setting host must not be a local or internal domain" + ): _reject_unsafe_ip_literal("setting", "test.internal") - with pytest.raises(ValueError, match="setting host must not be a local or internal domain"): + with pytest.raises( + ValueError, match="setting host must not be a local or internal domain" + ): _reject_unsafe_ip_literal("setting", "test.local") # Standard domain name _reject_unsafe_ip_literal("setting", "example.com") + def test_validate_global_address(): assert _validate_global_address("setting", "8.8.8.8") == "8.8.8.8" - assert _validate_global_address("setting", "2001:4860:4860::8888") == "2001:4860:4860::8888" + assert ( + _validate_global_address("setting", "2001:4860:4860::8888") + == "2001:4860:4860::8888" + ) - with pytest.raises(ValueError, match="setting resolved IP host must be globally routable"): + with pytest.raises( + ValueError, match="setting resolved IP host must be globally routable" + ): _validate_global_address("setting", "127.0.0.1") - with pytest.raises(ValueError, match="setting resolved IP host must be globally routable"): + with pytest.raises( + ValueError, match="setting resolved IP host must be globally routable" + ): _validate_global_address("setting", "invalid-ip") + @patch("socket.getaddrinfo") def test_resolve_global_addresses(mock_getaddrinfo): mock_getaddrinfo.return_value = [ (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 443)), (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.4.4", 443)), - (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 443)), # duplicate - (socket.AF_INET6, socket.SOCK_STREAM, 6, "", ("2001:4860:4860::8888", 443, 0, 0)), + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 443)), # duplicate + ( + socket.AF_INET6, + socket.SOCK_STREAM, + 6, + "", + ("2001:4860:4860::8888", 443, 0, 0), + ), ] addresses = _resolve_global_addresses("setting", "example.com", 443) assert addresses == ("8.8.8.8", "8.8.4.4", "2001:4860:4860::8888") - mock_getaddrinfo.assert_called_once_with("example.com", 443, type=socket.SOCK_STREAM) + mock_getaddrinfo.assert_called_once_with( + "example.com", 443, type=socket.SOCK_STREAM + ) + @patch("socket.getaddrinfo") def test_resolve_global_addresses_gaierror(mock_getaddrinfo): mock_getaddrinfo.side_effect = socket.gaierror("Name or service not known") - with pytest.raises(ValueError, match="setting host must resolve to a global address"): + with pytest.raises( + ValueError, match="setting host must resolve to a global address" + ): _resolve_global_addresses("setting", "example.com", 443) + @patch("socket.getaddrinfo") def test_resolve_global_addresses_no_global(mock_getaddrinfo): mock_getaddrinfo.return_value = [ (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 443)), ] - with pytest.raises(ValueError, match="setting resolved IP host must be globally routable"): + with pytest.raises( + ValueError, match="setting resolved IP host must be globally routable" + ): _resolve_global_addresses("setting", "example.com", 443) + @patch("socket.getaddrinfo") def test_resolve_global_addresses_empty(mock_getaddrinfo): mock_getaddrinfo.return_value = [] - with pytest.raises(ValueError, match="setting host must resolve to a global address"): + with pytest.raises( + ValueError, match="setting host must resolve to a global address" + ): _resolve_global_addresses("setting", "example.com", 443) + @patch("core.url_validation._resolve_global_addresses") def test_validate_https_url_host_details(mock_resolve): mock_resolve.return_value = ("8.8.8.8",) # Success res = validate_https_url_host_details( - "setting", "https://example.com/path", frozenset({"example.com"}), "ALLOWED_HOSTS" + "setting", + "https://example.com/path", + frozenset({"example.com"}), + "ALLOWED_HOSTS", ) assert res.normalized_url == "https://example.com/path" assert res.hostname == "example.com" @@ -109,7 +153,10 @@ def test_validate_https_url_host_details(mock_resolve): # Success with port res2 = validate_https_url_host_details( - "setting", "https://example.com:8443/path", frozenset({"example.com"}), "ALLOWED_HOSTS" + "setting", + "https://example.com:8443/path", + frozenset({"example.com"}), + "ALLOWED_HOSTS", ) assert res2.normalized_url == "https://example.com:8443/path" assert res2.hostname == "example.com" @@ -119,19 +166,28 @@ def test_validate_https_url_host_details(mock_resolve): # Not https with pytest.raises(ValueError, match="setting must use https"): validate_https_url_host_details( - "setting", "http://example.com/path", frozenset({"example.com"}), "ALLOWED_HOSTS" + "setting", + "http://example.com/path", + frozenset({"example.com"}), + "ALLOWED_HOSTS", ) # Userinfo with pytest.raises(ValueError, match="setting must not include userinfo"): validate_https_url_host_details( - "setting", "https://user:pass@example.com/path", frozenset({"example.com"}), "ALLOWED_HOSTS" + "setting", + "https://user:pass@example.com/path", + frozenset({"example.com"}), + "ALLOWED_HOSTS", ) # Fragment with pytest.raises(ValueError, match="setting must not include a fragment"): validate_https_url_host_details( - "setting", "https://example.com/path#frag", frozenset({"example.com"}), "ALLOWED_HOSTS" + "setting", + "https://example.com/path#frag", + frozenset({"example.com"}), + "ALLOWED_HOSTS", ) # No host @@ -141,12 +197,47 @@ def test_validate_https_url_host_details(mock_resolve): ) # Host not in allowed - with pytest.raises(ValueError, match="setting host must be listed in ALLOWED_HOSTS"): + with pytest.raises( + ValueError, match="setting host must be listed in ALLOWED_HOSTS" + ): validate_https_url_host_details( - "setting", "https://bad.com/path", frozenset({"example.com"}), "ALLOWED_HOSTS" + "setting", + "https://bad.com/path", + frozenset({"example.com"}), + "ALLOWED_HOSTS", ) + @patch("core.url_validation.validate_https_url_host_details") def test_validate_https_url_host(mock_details): - validate_https_url_host("setting", "https://example.com", frozenset({"example.com"}), "ALLOWED_HOSTS") - mock_details.assert_called_once_with("setting", "https://example.com", frozenset({"example.com"}), "ALLOWED_HOSTS") + validate_https_url_host( + "setting", "https://example.com", frozenset({"example.com"}), "ALLOWED_HOSTS" + ) + mock_details.assert_called_once_with( + "setting", "https://example.com", frozenset({"example.com"}), "ALLOWED_HOSTS" + ) + + +def test_validate_same_or_subdomain_host_rejects_suffix_confusion(): + for valid_host in ( + "issuer.example.com", + "jwks.issuer.example.com", + "a.b.c.issuer.example.com", + ): + validate_same_or_subdomain_host( + "OIDC_JWKS_URL", valid_host, "OIDC_ISSUER_URL", "issuer.example.com" + ) + + for invalid_host in ( + "other.com", + "fakeissuer.example.com", + "issuer.example.com.attacker.com", + "notexample.com", + ): + with pytest.raises( + ValueError, + match="OIDC_JWKS_URL host must match or be a subdomain of OIDC_ISSUER_URL host", + ): + validate_same_or_subdomain_host( + "OIDC_JWKS_URL", invalid_host, "OIDC_ISSUER_URL", "issuer.example.com" + ) diff --git a/docs/doctoring/local-http-origin-port-validation.md b/docs/doctoring/local-http-origin-port-validation.md new file mode 100644 index 000000000..2e0b36a09 --- /dev/null +++ b/docs/doctoring/local-http-origin-port-validation.md @@ -0,0 +1,36 @@ +# Local HTTP origin port-validation boundary + +## Decision + +`validate_loopback_http_origin()` distinguishes an absent URI port from an explicitly supplied port value. Scheme defaults are applied only when the port subcomponent is absent. An explicit port `0`, a value outside the application's `1..65535` transport-port contract, or a malformed/non-numeric port is rejected rather than rewritten to the scheme default. + +This is an origin-integrity rule, not only input cleanup. A caller that supplied `:0` expressed a materially different authority from one that omitted the port. Replacing the explicit value with `80` or `443` changes caller intent and can convert malformed or attacker-controlled configuration into a valid local destination. + +The same validator continues to require the existing loopback-host allowlist and to reject credentials, path/query/fragment material outside the local-origin contract, control characters, and unsafe request-target traversal. + +## Standards basis + +RFC 3986 defines the URI authority as host plus an optional decimal port subcomponent and allows a scheme to define a default port. The default therefore belongs to the *absent-port* case; an explicitly parsed port must not be collapsed with absence merely because the application's language treats numeric zero as false. + +RFC 6335 defines the Service Name and Transport Protocol Port Number Registry and the port-number space used by transport protocols. Naruon's local-origin helper intentionally narrows its application contract to `1..65535`; port zero is not a usable destination for this product path. The validator preserves this product-level restriction without making a broader claim that RFC 3986 itself forbids the textual URI `:0`. + +## Verification contract + +Regression tests must keep these cases distinct: + +- `http://localhost` and `https://localhost` use their scheme defaults; +- explicit supported ports are preserved; +- explicit `:0` is rejected instead of defaulted; +- negative, out-of-range, and non-numeric ports fail closed; +- IPv4/IPv6 loopback canonicalization remains stable; +- userinfo, path/query/fragment material outside the origin contract, controls, and non-allowlisted hosts remain rejected. + +## Rollback + +If a future runtime genuinely needs port zero as a sentinel, introduce a separate typed configuration field or explicit sentinel contract. Do not reintroduce truthiness-based defaulting in URI parsing, because that again conflates an explicit authority with absence. + +## References (APA 7th) + +Berners-Lee, T., Fielding, R., & Masinter, L. (2005). *Uniform Resource Identifier (URI): Generic syntax* (RFC 3986). RFC Editor. https://doi.org/10.17487/RFC3986 + +Cotton, M., Eggert, L., Touch, J., Westerlund, M., & Cheshire, S. (2011). *Internet Assigned Numbers Authority (IANA) procedures for the management of the service name and transport protocol port number registry* (BCP 165, RFC 6335). RFC Editor. https://doi.org/10.17487/RFC6335 diff --git a/docs/doctoring/python-lock-registry-provenance.md b/docs/doctoring/python-lock-registry-provenance.md new file mode 100644 index 000000000..199deda61 --- /dev/null +++ b/docs/doctoring/python-lock-registry-provenance.md @@ -0,0 +1,88 @@ +# PyPI release-hash provenance for Python locks + +## Status and ownership + +**Status:** Implemented on active PR only. This document does not describe protected `develop` until the corresponding code is merged. + +Naruon owns this repository-local supply-chain gate because it validates the Python lock files Naruon executes in CI and release preparation. PyPI remains the external release-metadata authority for this bounded public-index check. The gate does not copy dependency-policy authority from another CWL repository. + +## Buyer and operator decision + +A syntactically valid `--hash=sha256:` value is not sufficient evidence that a lock actually names a file published for the declared package release. Before dependency installation, Naruon therefore compares each exact project/version lock entry with trusted PyPI release metadata and requires at least one SHA-256 intersection with an eligible artifact. + +A passing receipt means the operator may continue to later dependency-install and platform-compatibility gates. A failing receipt means the operator should regenerate or investigate the lock; it must not be treated as a transient application-test failure or bypassed. + +## Implemented boundary + +For every discovered active `requirements*.txt` hash lock, the validator: + +1. reads only repository-contained UTF-8 files; +2. requires exact `==` pins and attached SHA-256 values; +3. normalizes project names before metadata resolution; +4. queries the exact PyPI release route `GET /pypi///json` over credential-free HTTPS; +5. binds returned `info.name` and `info.version` to the requested release; +6. considers only non-yanked `bdist_wheel` and `sdist` file objects with a syntactically valid SHA-256 digest; +7. requires at least one intersection between those published digests and the hashes recorded in the lock; +8. emits path-relative, deterministic reason codes and match counts without artifact URLs, provider exception strings, credentials, or absolute runner paths; +9. caches release metadata per `(project, version)` during one repository scan so repeated pins do not multiply external requests. + +Application CI runs this network-derived evidence after the deterministic offline lock-declaration gate and before dependency installation. + +## Failure semantics + +The gate is fail-closed. Important stable reasons include: + +- `lock-path-outside-repository`: a lock resolves outside the repository root; +- `lock-read-failed`: the lock cannot be read as repository UTF-8 text; +- `lock-requirement-not-exact`: a requirement is not an exact `==` pin; +- `lock-requirement-has-no-sha256`: an exact pin has no attached SHA-256; +- `registry-metadata-fetch-failed`: exact PyPI release metadata could not be resolved; +- `registry-project-mismatch` / `registry-version-mismatch`: returned metadata does not identify the requested release; +- `registry-release-has-no-allowed-artifacts`: the release has no eligible non-yanked wheel or source distribution SHA-256; +- `registry-hash-mismatch`: eligible release artifacts exist but none of their SHA-256 values appears in the lock. + +Network/provider exception text is deliberately not copied into the machine receipt. The workflow log may contain transport diagnostics from the trusted runtime, but the persisted summary is bounded to non-secret decision evidence. + +## Why PyPI release JSON is used in this slice + +The Python Packaging User Guide defines the Simple Repository API as the standards-track index interface and specifies JSON file records with hash dictionaries; PyPI recommends JSON for new index integrations. PyPI also documents a release-specific JSON route whose `urls` entries include file type, yanked state, and SHA-256 digests for one exact release. This bounded slice uses that release-specific PyPI route because it directly binds the requested exact version to its current file list without downloading or executing distributions. + +This is intentionally a **PyPI-specific adapter**, not a claim of generic PEP 691/private-index support. A future provider-neutral index adapter should consume the Simple Repository JSON API with explicit repository authority, TLS/origin policy, version selection, and index-isolation tests rather than silently redirecting this gate to an arbitrary host. + +## Relationship to pip hash checking + +pip's secure-install guidance describes `--require-hashes` as an all-or-nothing mode: requirements and dependencies need hashes and should be pinned, with multiple hashes often necessary when multiple wheels or source distributions are acceptable. It also distinguishes locally recorded hashes from remotely supplied index hashes. Naruon's registry receipt complements rather than replaces that control: it verifies that at least one local lock hash corresponds to an eligible file PyPI currently publishes for the exact release; later CI still performs `pip install --require-hashes`. + +## Explicit non-claims and follow-on work + +A passing receipt does **not** yet prove: + +- that the matched wheel is compatible with Python 3.14, the runner ABI, operating system, or architecture; +- that a source distribution is acceptable for the deployment policy; +- complete transitive dependency closure; +- clean installation on every supported Python/platform target; +- parity with a private or mirrored package index; +- that an artifact is covered by a trusted publisher attestation or PEP 740 provenance statement; +- reproducible wheel build output from an sdist. + +Issue #1229 remains open until those applicable boundaries, especially target-aware artifact matching and clean `pip install --require-hashes` rehearsal, have executable evidence. + +## Security and privacy analysis + +The built-in network path accepts only credential-free `https://pypi.org` as its origin. Project and version values become percent-encoded path segments; the receipt never copies returned file URLs. Metadata response size and content type are bounded before JSON parsing. No provider credential is needed or permitted for this public-index slice. + +The main residual risk is authority scope: proving a hash is published by PyPI is not the same as proving publisher identity, artifact intent, target compatibility, or absence of compromise. Those remain separate gates rather than being collapsed into one green status. + +## Verification + +The active PR uses RED-first tests covering matching and stale hashes, yanked and unsupported artifact types, release-identity mismatch, provider failure redaction, repeated-release fetch deduplication, trusted-origin validation, deterministic path-relative receipts, and CI ordering before installation. Exact current-head GitHub checks and independent review remain authoritative; predecessor-head results do not transfer. + +## References + +Python Packaging Authority. (n.d.). *Simple repository API*. Python Packaging User Guide. Retrieved August 16, 2026, from https://packaging.python.org/en/latest/specifications/simple-repository-api/ + +Python Packaging Authority. (n.d.). *Secure installs*. pip documentation. Retrieved August 16, 2026, from https://pip.pypa.io/en/stable/topics/secure-installs/ + +Python Package Index. (n.d.). *Index API*. PyPI Docs. Retrieved August 16, 2026, from https://docs.pypi.org/api/index-api/ + +Python Package Index. (n.d.). *JSON API*. PyPI Docs. Retrieved August 16, 2026, from https://docs.pypi.org/api/json/ diff --git a/scripts/ci/python_lock_registry_provenance.py b/scripts/ci/python_lock_registry_provenance.py new file mode 100644 index 000000000..ea2a391c4 --- /dev/null +++ b/scripts/ci/python_lock_registry_provenance.py @@ -0,0 +1,503 @@ +#!/usr/bin/env python3 +"""Validate hash-pinned Python locks against exact PyPI release metadata. + +This validator is intentionally narrower than dependency installation. It proves +that each exact project/version pin has at least one eligible, non-yanked wheel +or source distribution published by PyPI whose SHA-256 digest is recorded in +the lock. It does not claim platform compatibility, dependency closure, install +success, private-index parity, or artifact-attestation identity. +""" + +from __future__ import annotations + +import argparse +import json +import re +import urllib.parse +import urllib.error +import urllib.request +from pathlib import Path +from typing import Callable, Iterable, Mapping + +SCHEMA_VERSION = "naruon.python-lock-registry-provenance.v1" +DEFAULT_PYPI_ORIGIN = "https://pypi.org" +MAX_METADATA_BYTES = 4 * 1024 * 1024 +ALLOWED_PACKAGE_TYPES = frozenset({"bdist_wheel", "sdist"}) +_SHA256_RE = re.compile(r"^[0-9a-fA-F]{64}$") +_EXACT_PIN_RE = re.compile( + r"^(?P[A-Za-z0-9][A-Za-z0-9._-]*(?:\[[A-Za-z0-9._,-]+\])?)" + r"==(?P[^\s\\;]+)(?:\s*;\s*[^\\]+)?\s*\\?$" +) +_HASH_LINE_RE = re.compile(r"^--hash=sha256:(?P[0-9a-fA-F]{64})\s*\\?$") + +ReleaseFetcher = Callable[[str, str], Mapping[str, object]] + + +class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): + """Prevent urllib from contacting an unvalidated redirect target.""" + + def redirect_request(self, *args: object, **kwargs: object) -> None: + """Reject every redirect so the caller can fail before a second request.""" + return None + + +def _open_pypi_request( + request: urllib.request.Request, + *, + timeout_seconds: float, +) -> object: + """Open one PyPI request without following redirects.""" + opener = urllib.request.build_opener(_NoRedirectHandler()) + try: + return opener.open(request, timeout=timeout_seconds) + except urllib.error.HTTPError as exc: + if 300 <= exc.code < 400: + raise ValueError("PyPI metadata redirects are not allowed") from exc + raise + + +def _normalized_name(name: str) -> str: + """Return the canonical comparison and PyPI lookup form for a project name.""" + return re.sub(r"[-_.]+", "-", name.split("[", 1)[0].lower()) + + +def _relative_path(path: Path, repository_root: Path) -> str: + """Return a stable repository-relative path without leaking runner paths.""" + try: + return path.resolve().relative_to(repository_root.resolve()).as_posix() + except ValueError: + return path.name + + +def _resolve_repository_path(path: Path, repository_root: Path) -> Path | None: + """Resolve ``path`` only when its final target remains inside the repository.""" + root = repository_root.resolve() + candidate = path.resolve() + try: + candidate.relative_to(root) + except ValueError: + return None + return candidate + + +def _violation(code: str, path: str, detail: str) -> dict[str, str]: + """Build one deterministic machine-readable validation finding.""" + return {"code": code, "path": path, "detail": detail} + + +def _parse_lock_requirements( + text: str, + relative_path: str, +) -> tuple[list[dict[str, object]], list[dict[str, str]]]: + """Parse exact requirements and their attached SHA-256 values from a lock.""" + requirements: list[dict[str, object]] = [] + violations: list[dict[str, str]] = [] + current: dict[str, object] | None = None + + def finalize() -> None: + nonlocal current + if current is None: + return + hashes = current["hashes"] + assert isinstance(hashes, set) + if not hashes: + violations.append( + _violation( + "lock-requirement-has-no-sha256", + relative_path, + f"{current['project']}=={current['version']} has no SHA-256", + ) + ) + current["hashes"] = sorted(hashes) + requirements.append(current) + current = None + + for raw_line in text.splitlines(): + stripped = raw_line.strip() + if not stripped or stripped.startswith("#"): + continue + hash_match = _HASH_LINE_RE.fullmatch(stripped) + if hash_match is not None: + if current is None: + violations.append( + _violation( + "lock-orphan-sha256", + relative_path, + "SHA-256 entry is not attached to an exact requirement", + ) + ) + else: + hashes = current["hashes"] + assert isinstance(hashes, set) + hashes.add(hash_match.group("digest").lower()) + continue + if stripped.startswith("-"): + continue + + finalize() + match = _EXACT_PIN_RE.fullmatch(stripped) + if match is None: + violations.append( + _violation( + "lock-requirement-not-exact", + relative_path, + "lock contains a requirement that is not an exact == pin", + ) + ) + continue + current = { + "project": _normalized_name(match.group("name")), + "version": match.group("version"), + "hashes": set(), + } + + finalize() + return requirements, violations + + +def build_pypi_release_url( + project: str, + version: str, + *, + pypi_origin: str = DEFAULT_PYPI_ORIGIN, +) -> str: + """Build an exact PyPI release JSON URL from a credential-free HTTPS origin.""" + try: + parsed = urllib.parse.urlsplit(pypi_origin) + port = parsed.port + except ValueError as exc: + raise ValueError("pypi_origin must be the trusted PyPI origin") from exc + if ( + parsed.scheme != "https" + or (parsed.hostname or "").lower() != "pypi.org" + or parsed.username is not None + or parsed.password is not None + or port is not None + or parsed.path not in {"", "/"} + or parsed.query + or parsed.fragment + ): + raise ValueError("pypi_origin must be the trusted PyPI origin") + + normalized_project = _normalized_name(project) + project_segment = urllib.parse.quote(normalized_project, safe="-._") + version_segment = urllib.parse.quote(version, safe="-._") + return f"{DEFAULT_PYPI_ORIGIN}/pypi/{project_segment}/{version_segment}/json" + + +def fetch_pypi_release( + project: str, + version: str, + *, + timeout_seconds: float = 15.0, + max_metadata_bytes: int = MAX_METADATA_BYTES, +) -> Mapping[str, object]: + """Fetch one exact PyPI release document with a bounded credential-free GET.""" + if timeout_seconds <= 0: + raise ValueError("timeout_seconds must be positive") + if max_metadata_bytes <= 0: + raise ValueError("max_metadata_bytes must be positive") + + release_url = build_pypi_release_url(project, version) + request = urllib.request.Request( + release_url, + headers={ + "Accept": "application/json", + "User-Agent": "naruon-lock-provenance/1", + }, + method="GET", + ) + with _open_pypi_request(request, timeout_seconds=timeout_seconds) as response: + final_url_getter = getattr(response, "geturl", None) + final_url = final_url_getter() if callable(final_url_getter) else release_url + if final_url != release_url: + raise ValueError("PyPI metadata response left the trusted PyPI origin") + content_type = response.headers.get("Content-Type", "") + if not content_type.lower().startswith("application/json"): + raise ValueError("PyPI release metadata must be JSON") + payload = response.read(max_metadata_bytes + 1) + if len(payload) > max_metadata_bytes: + raise ValueError("PyPI release metadata exceeds the configured byte limit") + decoded = json.loads(payload.decode("utf-8")) + if not isinstance(decoded, dict): + raise ValueError("PyPI release metadata must be a JSON object") + return decoded + + +def _eligible_registry_hashes(metadata: Mapping[str, object]) -> set[str]: + """Return non-yanked wheel/sdist SHA-256 values from a PyPI release payload.""" + urls = metadata.get("urls") + if not isinstance(urls, list): + return set() + hashes: set[str] = set() + for artifact in urls: + if not isinstance(artifact, dict): + continue + if artifact.get("yanked") is True: + continue + if artifact.get("packagetype") not in ALLOWED_PACKAGE_TYPES: + continue + digests = artifact.get("digests") + if not isinstance(digests, dict): + continue + digest = digests.get("sha256") + if isinstance(digest, str) and _SHA256_RE.fullmatch(digest): + hashes.add(digest.lower()) + return hashes + + +def _validate_requirement_metadata( + *, + project: str, + version: str, + locked_hashes: set[str], + metadata: Mapping[str, object], + relative_path: str, +) -> tuple[dict[str, object], list[dict[str, str]]]: + """Compare one exact lock pin with one exact PyPI release metadata document.""" + violations: list[dict[str, str]] = [] + info = metadata.get("info") + info_mapping = info if isinstance(info, dict) else {} + metadata_name = info_mapping.get("name") + metadata_version = info_mapping.get("version") + if not isinstance(metadata_name, str) or _normalized_name(metadata_name) != project: + violations.append( + _violation( + "registry-project-mismatch", + relative_path, + f"trusted metadata identity does not match {project}", + ) + ) + if not isinstance(metadata_version, str) or metadata_version != version: + violations.append( + _violation( + "registry-version-mismatch", + relative_path, + f"trusted metadata version does not match {project}=={version}", + ) + ) + + matched_count = 0 + if not violations: + registry_hashes = _eligible_registry_hashes(metadata) + if not registry_hashes: + violations.append( + _violation( + "registry-release-has-no-allowed-artifacts", + relative_path, + f"{project}=={version} has no eligible non-yanked wheel or sdist SHA-256", + ) + ) + else: + matched_count = len(locked_hashes & registry_hashes) + if matched_count == 0: + violations.append( + _violation( + "registry-hash-mismatch", + relative_path, + f"{project}=={version} lock hashes do not match eligible PyPI artifacts", + ) + ) + + requirement_receipt = { + "project": project, + "version": version, + "status": "failed" if violations else "passed", + "matched_artifact_count": matched_count, + } + return requirement_receipt, violations + + +def validate_lock_against_registry( + lock_path: Path, + repository_root: Path, + *, + fetch_release: ReleaseFetcher = fetch_pypi_release, +) -> dict[str, object]: + """Validate one in-repository hash lock against exact PyPI release metadata.""" + relative_path = _relative_path(lock_path, repository_root) + resolved_lock = _resolve_repository_path(lock_path, repository_root) + if resolved_lock is None: + violations = [ + _violation( + "lock-path-outside-repository", + relative_path, + "lock path resolves outside repository root", + ) + ] + return { + "path": relative_path, + "status": "failed", + "requirements": [], + "violations": violations, + } + + try: + text = resolved_lock.read_text(encoding="utf-8") + except (OSError, UnicodeError): + violations = [ + _violation( + "lock-read-failed", + relative_path, + "lock could not be read as repository UTF-8 text", + ) + ] + return { + "path": relative_path, + "status": "failed", + "requirements": [], + "violations": violations, + } + + parsed_requirements, violations = _parse_lock_requirements(text, relative_path) + requirement_receipts: list[dict[str, object]] = [] + for requirement in parsed_requirements: + project = str(requirement["project"]) + version = str(requirement["version"]) + raw_hashes = requirement["hashes"] + assert isinstance(raw_hashes, list) + locked_hashes = {str(value).lower() for value in raw_hashes} + try: + metadata = fetch_release(project, version) + except Exception: + requirement_receipts.append( + { + "project": project, + "version": version, + "status": "failed", + "matched_artifact_count": 0, + } + ) + violations.append( + _violation( + "registry-metadata-fetch-failed", + relative_path, + f"trusted PyPI metadata could not be resolved for {project}=={version}", + ) + ) + continue + requirement_receipt, metadata_violations = _validate_requirement_metadata( + project=project, + version=version, + locked_hashes=locked_hashes, + metadata=metadata, + relative_path=relative_path, + ) + requirement_receipts.append(requirement_receipt) + violations.extend(metadata_violations) + + requirement_receipts.sort(key=lambda item: (str(item["project"]), str(item["version"]))) + violations.sort(key=lambda item: (item["code"], item["path"], item["detail"])) + return { + "path": relative_path, + "status": "failed" if violations else "passed", + "requirements": requirement_receipts, + "violations": violations, + } + + +def discover_hash_locks(repository_root: Path) -> list[Path]: + """Discover active requirements hash locks without reading escaping symlinks.""" + candidates: list[Path] = [] + for path in repository_root.rglob("requirements*.txt"): + if any(part in {".git", ".venv", "node_modules"} for part in path.parts): + continue + resolved = _resolve_repository_path(path, repository_root) + if resolved is None: + candidates.append(path) + continue + try: + text = resolved.read_text(encoding="utf-8") + except (OSError, UnicodeError): + continue + if "--hash=sha256:" in text or "hash" in path.stem.lower(): + candidates.append(path) + return sorted(candidates, key=lambda path: _relative_path(path, repository_root)) + + +def validate_repository_registry( + repository_root: Path, + *, + fetch_release: ReleaseFetcher = fetch_pypi_release, +) -> dict[str, object]: + """Validate all active hash locks while resolving each release metadata once.""" + cache: dict[tuple[str, str], tuple[bool, Mapping[str, object] | None]] = {} + + def cached_fetch(project: str, version: str) -> Mapping[str, object]: + key = (project, version) + cached = cache.get(key) + if cached is None: + try: + metadata = fetch_release(project, version) + except Exception: + cache[key] = (False, None) + raise RuntimeError("registry metadata unavailable") from None + cache[key] = (True, metadata) + return metadata + success, metadata = cached + if not success or metadata is None: + raise RuntimeError("registry metadata unavailable") + return metadata + + discovered_locks = discover_hash_locks(repository_root) + lock_receipts = [ + validate_lock_against_registry(path, repository_root, fetch_release=cached_fetch) + for path in discovered_locks + ] + violations = [ + violation + for receipt in lock_receipts + for violation in receipt["violations"] + if isinstance(violation, dict) + ] + if not discovered_locks: + violations.append( + _violation( + "registry-no-hash-locks", + ".", + "no active Python requirements hash lock was discovered", + ) + ) + violations.sort(key=lambda item: (item["code"], item["path"], item["detail"])) + return { + "schema_version": SCHEMA_VERSION, + "status": "failed" if violations else "passed", + "lock_files": lock_receipts, + "violations": violations, + } + + +def _build_parser() -> argparse.ArgumentParser: + """Build the command-line parser for repository-level registry validation.""" + parser = argparse.ArgumentParser( + description="Verify Python lock SHA-256 values against exact PyPI releases." + ) + parser.add_argument( + "--repository-root", + type=Path, + default=Path.cwd(), + help="Repository root to validate (default: current working directory).", + ) + parser.add_argument( + "--json", + action="store_true", + help="Emit one deterministic credential-free JSON receipt.", + ) + return parser + + +def main(argv: Iterable[str] | None = None) -> int: + """Run registry validation and return zero only for a passing receipt.""" + args = _build_parser().parse_args(list(argv) if argv is not None else None) + receipt = validate_repository_registry(args.repository_root) + if args.json: + print(json.dumps(receipt, sort_keys=True, separators=(",", ":"))) + else: + print(f"Python lock PyPI provenance: {receipt['status']}") + for violation in receipt["violations"]: + print(f"{violation['code']}: {violation['path']}: {violation['detail']}") + return 0 if receipt["status"] == "passed" else 1 + + +if __name__ == "__main__": + raise SystemExit(main())