diff --git a/.github/workflows/app-ci.yml b/.github/workflows/app-ci.yml index e8f445748..f55aa8688 100644 --- a/.github/workflows/app-ci.yml +++ b/.github/workflows/app-ci.yml @@ -46,6 +46,49 @@ jobs: cache: pip cache-dependency-path: backend/requirements-hashes.txt + - name: Validate Python lock provenance + run: | + status=0 + receipt="$(python scripts/ci/python_lock_provenance.py --json)" || status=$? + printf '%s\n' "$receipt" + { + echo '### Python lock provenance' + echo '```json' + printf '%s\n' "$receipt" + echo '```' + } >> "$GITHUB_STEP_SUMMARY" + exit "$status" + + - name: Determine whether PyPI registry provenance is required + id: registry_scope + env: + BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.before }} + run: | + set -euo pipefail + required=true + if [[ -n "${BASE_SHA:-}" && "$BASE_SHA" != "0000000000000000000000000000000000000000" ]]; then + git fetch --no-tags --depth=1 origin "$BASE_SHA" + changed_files="$(git diff --name-only "$BASE_SHA" HEAD)" + if ! grep -Eq '(^|/)requirements[^/]*\.txt$|^scripts/ci/python_lock_(registry_)?provenance\.py$|^backend/tests/test_python_lock_|^docs/doctoring/python-lock-|^\.github/workflows/app-ci\.yml$' <<<"$changed_files"; then + required=false + fi + fi + echo "required=$required" >> "$GITHUB_OUTPUT" + + - name: Validate PyPI release hash provenance + if: steps.registry_scope.outputs.required == 'true' + 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/tests/test_python_lock_provenance.py b/backend/tests/test_python_lock_provenance.py new file mode 100644 index 000000000..81aed9bd0 --- /dev/null +++ b/backend/tests/test_python_lock_provenance.py @@ -0,0 +1,432 @@ +"""Contract tests for deterministic Python lock provenance receipts. + +The validator is intentionally exercised from backend CI because hash-locked +requirements are part of the release supply-chain boundary rather than an +optional developer convenience. +""" + +from __future__ import annotations + +import importlib.util +import json +import runpy +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_provenance.py" + +_spec = importlib.util.spec_from_file_location("python_lock_provenance", SCRIPT_PATH) +assert _spec is not None and _spec.loader is not None +python_lock_provenance = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = python_lock_provenance +_spec.loader.exec_module(python_lock_provenance) + + +def _write(path: Path, text: str) -> Path: + """Create one UTF-8 fixture file and return its path.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return path + + +def _sha(character: str = "a") -> str: + """Return one syntactically valid SHA-256 hex digest for fixtures.""" + return character * 64 + + +def _violation_codes(receipt: dict[str, object]) -> set[str]: + """Return stable violation codes from one lock receipt.""" + violations = receipt["violations"] + assert isinstance(violations, list) + return {str(item["code"]) for item in violations} + + +def _simple_lock(name: str = "example", version: str = "1.0") -> str: + """Return one exact hash-pinned requirement fixture.""" + return f"{name}=={version} \\\n --hash=sha256:{_sha()}\n" + + +def test_manual_download_generation_version_mismatch_is_rejected(tmp_path: Path) -> None: + """A stale generator command must not attest a newer declared package.""" + lock_path = _write( + tmp_path / "connector" / "requirements-hashes.txt", + "# Regenerate with:\n" + "# python3 -m pip download --only-binary=:all: websockets==16.1\n" + "websockets==17.0 \\\n" + f" --hash=sha256:{_sha()}\n", + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == {"generation-version-mismatch"} + + +def test_manual_download_generation_matching_version_passes(tmp_path: Path) -> None: + """A matching manual generator command and SHA-256 pin form a valid declaration.""" + lock_path = _write( + tmp_path / "connector" / "requirements-hashes.txt", + "# Regenerate with:\n" + "# python3 -m pip download --only-binary=:all: websockets==17.0\n" + "websockets==17.0 \\\n" + f" --hash=sha256:{_sha('b')}\n", + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert receipt["status"] == "passed" + assert receipt["violations"] == [] + assert receipt["requirement_count"] == 1 + assert receipt["sha256_hash_count"] == 1 + assert receipt["generation_mode"] == "pip-download" + + +def test_manual_download_generation_with_extras_passes(tmp_path: Path) -> None: + """PEP 508 extras remain bound to the same normalized project/version pin.""" + lock_path = _write( + tmp_path / "requirements-hashes.txt", + "# Regenerate with:\n" + "# python3 -m pip download 'SomePackage[PDF]==3.0'\n" + "SomePackage[PDF]==3.0 \\\n" + f" --hash=sha256:{_sha('d')}\n", + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert receipt["status"] == "passed" + assert receipt["generation_mode"] == "pip-download" + assert receipt["violations"] == [] + + +def test_manual_download_without_exact_generator_pin_is_rejected(tmp_path: Path) -> None: + """A recognized manual generator must name the package/version it attests.""" + lock_path = _write( + tmp_path / "requirements-hashes.txt", + "# Regenerate with:\n" + "# python3 -m pip download --only-binary=:all:\n" + + _simple_lock(), + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == {"generation-input-missing"} + + +def test_unpinned_and_unhashed_requirements_fail_closed(tmp_path: Path) -> None: + """Hash-checking evidence rejects non-exact pins and missing SHA-256 hashes.""" + lock_path = _write( + tmp_path / "requirements-hashes.txt", + "example>=1.0\n" + "other==2.0\n", + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == { + "missing-sha256", + "requirement-not-exactly-pinned", + } + + +def test_malformed_orphan_and_duplicate_hash_evidence_is_rejected(tmp_path: Path) -> None: + """Malformed hash structure fails with stable, independently useful codes.""" + lock_path = _write( + tmp_path / "requirements-hashes.txt", + f"--hash=sha256:{_sha('a')}\n" + "example==1.0 \\\n" + " --hash=sha256:not-a-digest\n" + "example==1.0 \\\n" + f" --hash=sha256:{_sha('b')}\n" + "--hash=sha512:not-supported\n", + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == { + "duplicate-requirement", + "malformed-sha256", + "missing-sha256", + "orphan-hash", + } + + +def test_uv_generation_source_version_mismatch_is_rejected(tmp_path: Path) -> None: + """A uv-generated lock must agree with exact direct pins in its declared input.""" + _write( + tmp_path / "requirements.txt", + "# direct dependencies\n--index-url https://example.invalid/simple\n" + "ignored>=1\nexample==2.0\n", + ) + lock_path = _write( + tmp_path / "requirements-hashes.txt", + "# This file was autogenerated by uv via the following command:\n" + "# uv pip compile --generate-hashes --output-file requirements-hashes.txt requirements.txt\n" + "example==1.0 \\\n" + f" --hash=sha256:{_sha('c')}\n", + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == {"generation-version-mismatch"} + + +def test_uv_generation_source_inline_comment_still_binds_version( + tmp_path: Path, +) -> None: + """A valid source pin with an inline comment remains part of the contract.""" + _write(tmp_path / "requirements.txt", "example==2.0 # updated source pin\n") + lock_path = _write( + tmp_path / "requirements-hashes.txt", + "# uv pip compile --generate-hashes --output-file requirements-hashes.txt requirements.txt\n" + + _simple_lock(version="1.0"), + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == {"generation-version-mismatch"} + + +def test_uv_generation_accepts_requirements_in_source(tmp_path: Path) -> None: + """The conventional requirements.in source form is valid uv provenance.""" + _write(tmp_path / "requirements.in", "example==1.0\n") + lock_path = _write( + tmp_path / "requirements-hashes.txt", + "# uv pip compile requirements.in --generate-hashes --output-file requirements-hashes.txt\n" + + _simple_lock(), + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert receipt["status"] == "passed" + assert receipt["generation_mode"] == "uv" + assert receipt["violations"] == [] + + +def test_uv_generation_missing_input_is_rejected(tmp_path: Path) -> None: + """A generated lock cannot claim provenance from a source file that is absent.""" + lock_path = _write( + tmp_path / "requirements-hashes.txt", + "# This file was autogenerated by uv via the following command:\n" + "# uv pip compile --generate-hashes --output-file requirements-hashes.txt requirements.txt\n" + + _simple_lock(), + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == {"generation-input-missing"} + + +def test_uv_generation_requires_output_and_source_declarations(tmp_path: Path) -> None: + """A uv command must bind both its output lock and input requirements path.""" + lock_path = _write( + tmp_path / "requirements-hashes.txt", + "# uv pip compile --generate-hashes\n" + _simple_lock(), + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == { + "generation-input-missing", + "generation-output-missing", + } + + +def test_uv_generation_output_path_mismatch_is_rejected(tmp_path: Path) -> None: + """A generator cannot attest a different lock path than the file under test.""" + _write(tmp_path / "requirements.txt", "example==1.0\n") + lock_path = _write( + tmp_path / "requirements-hashes.txt", + "# uv pip compile --generate-hashes --output-file other-hashes.txt requirements.txt\n" + + _simple_lock(), + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == {"generation-output-mismatch"} + + +def test_uv_generation_rejects_source_outside_repository(tmp_path: Path) -> None: + """A generator declaration cannot make CI read a source outside the repo.""" + repository_root = tmp_path / "repo" + repository_root.mkdir() + _write(tmp_path / "outside" / "requirements.in", "external-secret==9.9\n") + lock_path = _write( + repository_root / "requirements-hashes.txt", + "# uv pip compile ../outside/requirements.in --output-file requirements-hashes.txt\n" + + _simple_lock(), + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, repository_root) + serialized = json.dumps(receipt, sort_keys=True) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == {"generation-input-outside-repository"} + assert "external-secret" not in serialized + assert str(tmp_path) not in serialized + + +def test_repository_receipt_covers_every_active_hash_lock() -> None: + """The current repository must expose one passing receipt for every active lock.""" + receipt = python_lock_provenance.validate_repository(REPO_ROOT) + lock_files = receipt["lock_files"] + assert isinstance(lock_files, list) + paths = {str(item["path"]) for item in lock_files} + + assert receipt["status"] == "passed" + assert paths == { + "backend/requirements-agent.txt", + "backend/requirements-hashes.txt", + "connector/requirements-hashes.txt", + "requirements-bandit-ci-hashes.txt", + "requirements-strix-ci-hashes.txt", + } + assert all(len(str(item["sha256"])) == 64 for item in lock_files) + assert receipt["violations"] == [] + + +def test_repository_receipt_is_deterministic_and_path_relative(tmp_path: Path) -> None: + """Machine evidence is stable and never leaks an absolute runner path.""" + _write(tmp_path / "requirements-hashes.txt", _simple_lock()) + _write(tmp_path / "requirements.txt", "example==1.0\n") + _write(tmp_path / "notes.txt", "not a requirements file\n") + _write(tmp_path / ".venv" / "requirements-hidden.txt", _simple_lock()) + (tmp_path / "requirements-binary.txt").write_bytes(b"\xff\xfe\x00") + + first = python_lock_provenance.validate_repository(tmp_path) + second = python_lock_provenance.validate_repository(tmp_path) + + assert json.dumps(first, sort_keys=True) == json.dumps(second, sort_keys=True) + serialized = json.dumps(first, sort_keys=True) + assert str(tmp_path) not in serialized + assert first["schema_version"] == "naruon.python-lock-provenance.v1" + assert [item["path"] for item in first["lock_files"]] == [ + "requirements-hashes.txt" + ] + + +def test_discovery_skips_non_file_requirements_candidates(tmp_path: Path) -> None: + """A directory or broken link named like a lock cannot crash discovery.""" + (tmp_path / "requirements-directory.txt").mkdir() + (tmp_path / "requirements-broken.txt").symlink_to( + tmp_path / "missing-target.txt" + ) + + receipt = python_lock_provenance.validate_repository(tmp_path) + + assert receipt["status"] == "passed" + assert receipt["lock_files"] == [] + + +def test_outside_repository_lock_fails_without_reading_payload(tmp_path: Path) -> None: + """Direct validation rejects an out-of-root lock without serializing its data.""" + root = tmp_path / "root" + root.mkdir() + lock_path = _write( + tmp_path / "outside" / "requirements-hashes.txt", + "TOP_SECRET_PACKAGE>=9.9\n", + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, root) + serialized = json.dumps(receipt, sort_keys=True) + + assert receipt["status"] == "failed" + assert receipt["path"] == "requirements-hashes.txt" + assert _violation_codes(receipt) == {"lock-path-outside-repository"} + assert "TOP_SECRET_PACKAGE" not in serialized + assert str(tmp_path) not in serialized + + +def test_discovery_rejects_symlinked_lock_outside_repository(tmp_path: Path) -> None: + """Repository discovery never follows a requirements symlink outside root.""" + repository_root = tmp_path / "repo" + repository_root.mkdir() + outside_lock = _write( + tmp_path / "outside" / "secret.txt", + "TOP_SECRET_PACKAGE>=9.9\n", + ) + (repository_root / "requirements-hashes.txt").symlink_to(outside_lock) + + receipt = python_lock_provenance.validate_repository(repository_root) + serialized = json.dumps(receipt, sort_keys=True) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == {"lock-path-outside-repository"} + assert "TOP_SECRET_PACKAGE" not in serialized + assert str(tmp_path) not in serialized + + +def test_cli_json_and_human_modes_report_pass_and_fail( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + """Both operator surfaces preserve exit status and actionable reason codes.""" + _write(tmp_path / "requirements-hashes.txt", _simple_lock()) + assert python_lock_provenance.main(["--repository-root", str(tmp_path), "--json"]) == 0 + parsed = json.loads(capsys.readouterr().out) + assert parsed["status"] == "passed" + + _write(tmp_path / "requirements-hashes.txt", "example>=1.0\n") + assert python_lock_provenance.main(["--repository-root", str(tmp_path)]) == 1 + human_output = capsys.readouterr().out + assert "Python lock provenance: failed" in human_output + assert "requirement-not-exactly-pinned" in human_output + + +def test_script_main_guard_propagates_failed_exit( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Direct script execution exits nonzero when repository validation fails.""" + _write(tmp_path / "requirements-hashes.txt", "example>=1.0\n") + monkeypatch.setattr( + sys, + "argv", + [str(SCRIPT_PATH), "--repository-root", str(tmp_path), "--json"], + ) + + with pytest.raises(SystemExit) as exc_info: + runpy.run_path(str(SCRIPT_PATH), run_name="__main__") + + assert exc_info.value.code == 1 + + +def test_application_ci_publishes_lock_provenance_receipt() -> None: + """The backend install job must publish validation evidence before installation.""" + workflow_path = REPO_ROOT / ".github" / "workflows" / "app-ci.yml" + workflow = yaml.safe_load(workflow_path.read_text(encoding="utf-8")) + jobs = workflow["jobs"] + backend_jobs = [ + job + for job in jobs.values() + if any( + step.get("name") == "Install backend dependencies" + for step in job.get("steps", []) + if isinstance(step, dict) + ) + ] + assert len(backend_jobs) == 1 + + steps = backend_jobs[0]["steps"] + step_names = [step.get("name") for step in steps] + validation_index = step_names.index("Validate Python lock provenance") + install_index = step_names.index("Install backend dependencies") + assert validation_index < install_index + + validation_step = steps[validation_index] + validation_run = validation_step["run"] + assert "python scripts/ci/python_lock_provenance.py --json" in validation_run + assert "GITHUB_STEP_SUMMARY" in validation_run + assert "|| status=$?" in validation_run + assert 'exit "$status"' in validation_run diff --git a/backend/tests/test_python_lock_provenance_includes.py b/backend/tests/test_python_lock_provenance_includes.py new file mode 100644 index 000000000..86d5cca1a --- /dev/null +++ b/backend/tests/test_python_lock_provenance_includes.py @@ -0,0 +1,190 @@ +"""Focused contracts for requirements-file include provenance validation.""" + +from __future__ import annotations + +import hashlib +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT_PATH = REPO_ROOT / "scripts" / "ci" / "python_lock_provenance.py" + +_spec = importlib.util.spec_from_file_location( + "python_lock_provenance_includes", SCRIPT_PATH +) +assert _spec is not None and _spec.loader is not None +python_lock_provenance = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = python_lock_provenance +_spec.loader.exec_module(python_lock_provenance) + + +def _write(path: Path, text: str) -> Path: + """Create one UTF-8 requirements fixture and return its path.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return path + + +def _sha(character: str = "a") -> str: + """Return one syntactically valid SHA-256 fixture digest.""" + return character * 64 + + +def _lock(name: str = "root-package", version: str = "1.0") -> str: + """Return one exact requirement with SHA-256 evidence.""" + return f"{name}=={version} \\\n --hash=sha256:{_sha()}\n" + + +def _violation_codes(receipt: dict[str, object]) -> set[str]: + """Return stable violation codes from one receipt.""" + violations = receipt["violations"] + assert isinstance(violations, list) + return {str(item["code"]) for item in violations} + + +@pytest.mark.parametrize("directive", ["-r", "--requirement"]) +def test_requirement_include_forms_are_validated_recursively( + tmp_path: Path, + directive: str, +) -> None: + """Both pip include forms expose invalid included requirements.""" + included_path = _write( + tmp_path / "backend" / "generated-lock.txt", + "included-package>=2.0\n", + ) + root_path = _write( + tmp_path / "requirements-hashes.txt", + f"{directive} backend/generated-lock.txt\n" + _lock(), + ) + + receipt = python_lock_provenance.validate_lock_file(root_path, tmp_path) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == { + "missing-sha256", + "requirement-not-exactly-pinned", + } + assert receipt["requirement_count"] == 1 + assert receipt["sha256_hash_count"] == 1 + included_files = receipt["included_files"] + assert isinstance(included_files, list) + assert [item["path"] for item in included_files] == [ + included_path.relative_to(tmp_path).as_posix() + ] + assert included_files[0]["sha256"] == hashlib.sha256( + included_path.read_bytes() + ).hexdigest() + + +def test_valid_nested_requirement_include_contributes_receipt_counts( + tmp_path: Path, +) -> None: + """A contained valid include is represented and counted deterministically.""" + _write(tmp_path / "nested" / "included-lock.txt", _lock("child-package", "2.0")) + root_path = _write( + tmp_path / "requirements-hashes.txt", + "--requirement=nested/included-lock.txt\n" + _lock(), + ) + + first = python_lock_provenance.validate_lock_file(root_path, tmp_path) + second = python_lock_provenance.validate_lock_file(root_path, tmp_path) + + assert first["status"] == "passed" + assert first["requirement_count"] == 2 + assert first["sha256_hash_count"] == 2 + assert json.dumps(first, sort_keys=True) == json.dumps(second, sort_keys=True) + + +def test_requirement_include_outside_repository_fails_without_reading_payload( + tmp_path: Path, +) -> None: + """An escaping include cannot expose external file content or absolute paths.""" + repository_root = tmp_path / "repo" + repository_root.mkdir() + _write( + tmp_path / "outside" / "generated-lock.txt", + "EXTERNAL_SECRET_PACKAGE>=9.9\n", + ) + root_path = _write( + repository_root / "requirements-hashes.txt", + "-r ../outside/generated-lock.txt\n" + _lock(), + ) + + receipt = python_lock_provenance.validate_lock_file(root_path, repository_root) + serialized = json.dumps(receipt, sort_keys=True) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == { + "requirement-include-outside-repository" + } + assert "EXTERNAL_SECRET_PACKAGE" not in serialized + assert str(tmp_path) not in serialized + + +def test_missing_requirement_include_fails_closed(tmp_path: Path) -> None: + """A missing or non-file include has one stable operator-facing reason code.""" + (tmp_path / "missing-lock.txt").mkdir() + root_path = _write( + tmp_path / "requirements-hashes.txt", + "-r missing-lock.txt\n" + _lock(), + ) + + receipt = python_lock_provenance.validate_lock_file(root_path, tmp_path) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == {"requirement-include-missing"} + + +def test_malformed_requirement_include_fails_closed(tmp_path: Path) -> None: + """An include option without exactly one path cannot be silently skipped.""" + root_path = _write( + tmp_path / "requirements-hashes.txt", + "--requirement\n" + _lock(), + ) + + receipt = python_lock_provenance.validate_lock_file(root_path, tmp_path) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == {"requirement-include-invalid"} + + +def test_requirement_include_cycle_fails_closed(tmp_path: Path) -> None: + """A recursive include cycle terminates with a stable reason code.""" + root_path = _write( + tmp_path / "requirements-hashes.txt", + "-r nested/child-lock.txt\n" + _lock(), + ) + _write( + tmp_path / "nested" / "child-lock.txt", + "--requirement ../requirements-hashes.txt\n" + _lock("child-package", "2.0"), + ) + + receipt = python_lock_provenance.validate_lock_file(root_path, tmp_path) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == {"requirement-include-cycle"} + + +def test_requirement_include_depth_is_bounded(tmp_path: Path) -> None: + """A hostile acyclic include chain cannot exhaust Python recursion.""" + depth = python_lock_provenance._MAX_REQUIREMENT_INCLUDE_DEPTH + 2 + for index in range(depth): + include = f"-r lock-{index + 1}.txt\n" if index + 1 < depth else "" + _write( + tmp_path / f"lock-{index}.txt", + include + _lock(f"package-{index}", f"{index + 1}.0"), + ) + + receipt = python_lock_provenance.validate_lock_file( + tmp_path / "lock-0.txt", + tmp_path, + ) + + assert receipt["status"] == "failed" + assert _violation_codes(receipt) == { + "requirement-include-depth-exceeded" + } diff --git a/backend/tests/test_python_lock_provenance_review_edges.py b/backend/tests/test_python_lock_provenance_review_edges.py new file mode 100644 index 000000000..e4ba67745 --- /dev/null +++ b/backend/tests/test_python_lock_provenance_review_edges.py @@ -0,0 +1,58 @@ +"""Review regressions for Python lock provenance edge contracts.""" + +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + + +REPOSITORY_ROOT = Path(__file__).resolve().parents[2] +SCRIPT_PATH = REPOSITORY_ROOT / "scripts" / "ci" / "python_lock_provenance.py" + +_spec = importlib.util.spec_from_file_location("python_lock_provenance_edges", SCRIPT_PATH) +assert _spec is not None and _spec.loader is not None +python_lock_provenance = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = python_lock_provenance +_spec.loader.exec_module(python_lock_provenance) + + +def _hash(character: str) -> str: + """Return one syntactically valid SHA-256 fixture digest.""" + return character * 64 + + +def _codes(receipt: dict[str, object]) -> set[str]: + """Return top-level stable violation codes from a lock receipt.""" + violations = receipt["violations"] + assert isinstance(violations, list) + return {str(item["code"]) for item in violations} + + +def test_uv_generation_checks_every_declared_source_file(tmp_path: Path) -> None: + """A stale pin in an earlier uv input must not escape version agreement.""" + (tmp_path / "first.in").write_text("alpha==1.0\n", encoding="utf-8") + (tmp_path / "second.in").write_text("beta==2.0\n", encoding="utf-8") + lock_path = tmp_path / "requirements-hashes.txt" + lock_path.write_text( + "# uv pip compile first.in second.in --output-file requirements-hashes.txt\n" + f"alpha==9.0 \\\n --hash=sha256:{_hash('a')}\n" + f"beta==2.0 \\\n --hash=sha256:{_hash('b')}\n", + encoding="utf-8", + ) + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert "generation-version-mismatch" in _codes(receipt) + + +def test_non_utf8_requirement_include_returns_stable_failure(tmp_path: Path) -> None: + """An undecodable included file must fail closed without a Python traceback.""" + lock_path = tmp_path / "requirements-hashes.txt" + lock_path.write_text("-r binary.in\n", encoding="utf-8") + (tmp_path / "binary.in").write_bytes(b"\xff\xfe\x00") + + receipt = python_lock_provenance.validate_lock_file(lock_path, tmp_path) + + assert receipt["status"] == "failed" + assert "lock-read-failed" in _codes(receipt) diff --git a/backend/tests/test_python_lock_registry_ci_scope.py b/backend/tests/test_python_lock_registry_ci_scope.py new file mode 100644 index 000000000..ce056eb28 --- /dev/null +++ b/backend/tests/test_python_lock_registry_ci_scope.py @@ -0,0 +1,54 @@ +"""Regression contracts for PyPI registry-provenance CI scoping.""" + +from __future__ import annotations + +from pathlib import Path + + +REPOSITORY_ROOT = Path(__file__).resolve().parents[2] +APPLICATION_CI = REPOSITORY_ROOT / ".github" / "workflows" / "app-ci.yml" + + +def _application_ci_text() -> str: + """Read the repository-owned Application CI workflow as UTF-8 text.""" + return APPLICATION_CI.read_text(encoding="utf-8") + + +def test_registry_provenance_is_scoped_after_offline_validation() -> None: + """Keep live-PyPI evidence off unrelated PRs without weakening lock validation.""" + workflow = _application_ci_text() + + offline_index = workflow.index("- name: Validate Python lock provenance") + scope_index = workflow.index( + "- name: Determine whether PyPI registry provenance is required" + ) + registry_index = workflow.index("- name: Validate PyPI release hash provenance") + install_index = workflow.index("- name: Install backend dependencies") + + assert offline_index < scope_index < registry_index < install_index + assert "id: registry_scope" in workflow[scope_index:registry_index] + assert 'git diff --name-only "$BASE_SHA" HEAD' in workflow[scope_index:registry_index] + assert "requirements[^/]*\\.txt" in workflow[scope_index:registry_index] + assert 'echo "required=$required" >> "$GITHUB_OUTPUT"' in workflow[ + scope_index:registry_index + ] + + registry_block = workflow[registry_index:install_index] + assert "if: steps.registry_scope.outputs.required == 'true'" in registry_block + + +def test_registry_scope_fails_safe_when_base_cannot_be_compared() -> None: + """Unknown comparison state must require the network provenance gate.""" + workflow = _application_ci_text() + scope_index = workflow.index( + "- name: Determine whether PyPI registry provenance is required" + ) + registry_index = workflow.index("- name: Validate PyPI release hash provenance") + scope_block = workflow[scope_index:registry_index] + + assert "required=true" in scope_block + base_sha_expression = ( + 'BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.before }}' + ) + assert base_sha_expression in scope_block + assert '"0000000000000000000000000000000000000000"' in scope_block 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/docs/doctoring/python-lock-provenance-receipt.md b/docs/doctoring/python-lock-provenance-receipt.md new file mode 100644 index 000000000..d5128918a --- /dev/null +++ b/docs/doctoring/python-lock-provenance-receipt.md @@ -0,0 +1,103 @@ +# Python lock provenance receipt + +## Status boundary + +**Protected `develop` shipped truth (before PR #1369):** naruon installs its active Python lock files with pip hash-checking mode, but protected `develop` does not first attest that each repository-controlled lock declaration still agrees with its declared generator/source contract. + +**Active PR #1369:** adds an offline, deterministic declaration receipt before backend dependency installation. The receipt covers repository-controlled exact pins, SHA-256 hash syntax/presence, recognized generator command binding, declared `uv pip compile` output/source paths (including conventional `requirements.in` inputs and multiple declared source files), PEP 508-style extras on manual `pip download` pins, agreement between exact direct source pins and the generated lock, and repository-root containment before any candidate lock/source payload is read. Valid pip `-r` and `--requirement` directives are resolved recursively relative to the including file, represented as nested receipts, and bounded against malformed, missing, unreadable, escaping, cyclic, or excessively deep include graphs. + +The same active supply-chain lane now also contains the companion PyPI release-hash validator documented in `python-lock-registry-provenance.md`. That network-derived evidence is separate from this offline receipt and is diff-scoped in Application CI so unrelated product changes do not depend on live PyPI availability. Platform-specific artifact selection/hash matching and a clean `pip install --require-hashes` rehearsal remain issue #1229 follow-on work. + +## Customer and operator decision + +A passing offline receipt means that the checked-in Python lock declarations are internally consistent with the repository evidence this validator can verify without network access. It does **not** prove that a package index currently serves the expected distributions, that a distribution is available for the target platform, that a remote artifact's bytes match the checked-in hash, or that a clean installation succeeds. + +A failing receipt is actionable and fail-closed. The operator should read the stable reason code and affected relative path, regenerate or repair the affected lock from its declared source/generator, review the resulting dependency delta, and rerun Application CI. Do not bypass the receipt or remove hash-checking mode to make a dependency update green. A path-containment failure means the declaration or symlink must first be moved back under the repository root; the validator intentionally does not read the escaping payload. Include-graph failures require correcting the directive, restoring or re-encoding the referenced file, removing the cycle, or flattening an over-deep chain before dependency installation proceeds. + +## Evidence flow + +```mermaid +flowchart LR + A[Checked-in requirements sources] --> B[Declared lock generator] + B --> C[Hash-pinned lock files] + C --> C1[Bounded -r / --requirement include graph] + C1 --> D[Repository-root containment] + A --> D + C1 -->|invalid / missing / unreadable / cycle / depth| H[Stable include reason code] + D -->|contained| E[Offline provenance validator] + D -->|escapes root| F[Stable containment reason code] + E -->|pass| G[Deterministic JSON receipt] + E -->|fail| H[Stable reason code + relative path] + G --> I[pip install --require-hashes] + F --> J[Repair path / symlink] + H --> K[Regenerate / repair / review] + J --> D + K --> E + I --> L[Registry + artifact + clean-install evidence] +``` + +For safely contained lock paths, the validator emits repository-relative paths, SHA-256 digests of the checked-in lock text, aggregate requirement/hash counts, generation mode, nested `included_files` receipts, and stable validation findings. Include paths are resolved relative to the including file, checked against the repository root before `is_file()` or payload reads, and traversed to a maximum depth of 32. Missing, non-regular, or non-UTF-8 lock/include payloads fail with bounded reason data rather than an unhandled traceback. For an escaping lock path, it emits a failed receipt with `sha256: null`, zero counts, and a containment reason code without reading the target payload. It performs no network request and reads no credentials or package-index tokens. + +## Validation contract + +The active slice discovers `requirements*.txt` files containing SHA-256 lock entries and validates the following repository-controlled properties: + +- each requirement declaration is an exact `==` pin; +- each pinned requirement carries at least one syntactically valid SHA-256 entry; +- detached hashes, malformed SHA-256 entries, and duplicate project declarations fail with stable reason codes; +- valid `-r path`, `-rpath`, `--requirement path`, and `--requirement=path` directives are recursively validated relative to the including file rather than silently skipped; +- every include target must be a readable UTF-8 regular in-repository file, and malformed, missing, unreadable, escaping, cyclic, or deeper-than-32 include graphs fail closed before unsafe payload data can leak; +- nested included-file digests and counts are retained in deterministic `included_files` receipts while their findings are flattened into the parent lock decision; +- a recognized manual `pip download` regeneration command names at least one exact package/version, accepts standard extras such as `SomePackage[PDF]==3.0`, and agrees with the lock; +- a recognized `uv pip compile` command names the lock output and one or more `.txt` or `.in` source requirements files, and exact direct pins from every declared source agree with the generated lock; +- resolved lock/source candidates must remain under the resolved repository root before file existence checks or payload reads, including symlink targets and `..` traversal; +- an unreadable or non-UTF-8 declared `uv` source fails with a stable source reason rather than being silently ignored; +- the machine receipt is deterministic and does not serialize an absolute runner path, escaping file payload, or provider exception text; +- Application CI publishes the receipt before network dependency installation even when validation fails, then exits with the validator status. + +The implementation intentionally ignores arbitrary explanatory prose as provenance metadata. Only recognized generator command forms create generator-binding obligations. Recorded `uv pip compile` paths are interpreted as repository-root-relative because the checked-in generator comments use that convention; `-r` / `--requirement` includes follow pip's including-file-relative convention. This asymmetry is explicit rather than inferred from whichever path happens to appear last. + +## Reason-code handling + +| Code | Meaning | Operator action | +| --- | --- | --- | +| `requirement-not-exactly-pinned` | A lock entry is not an exact `==` requirement. | Regenerate the lock from the intended source requirements and review the resolved version. | +| `missing-sha256` | A requirement has no valid SHA-256 evidence. | Regenerate hashes for the intended artifacts; do not install without hash checking. | +| `malformed-sha256` | A SHA-256 entry is syntactically invalid. | Recompute the digest through the declared lock-generation path. | +| `orphan-hash` | A hash is not attached to a requirement declaration. | Regenerate or repair the lock structure. | +| `duplicate-requirement` | The same normalized project is declared more than once. | Consolidate the declaration through the source requirements and regenerate. | +| `requirement-include-invalid` | A `-r` or `--requirement` directive does not name exactly one file path. | Correct the directive to one supported file reference. | +| `requirement-include-missing` | The contained include target is absent or not a regular file. | Restore the referenced requirements file or remove the stale directive. | +| `requirement-include-outside-repository` | An include resolves outside the repository root, including through traversal or a symlink. | Move the target under repository control and rewrite the directive; do not expose the external payload to CI. | +| `requirement-include-cycle` | The include graph returns to a file already on the active traversal path. | Remove or flatten the cyclic include relationship. | +| `requirement-include-depth-exceeded` | The include graph exceeds the bounded depth of 32. | Flatten or consolidate the requirements graph before validation. | +| `generation-output-missing` | A recognized `uv` generator omits its output lock path. | Restore the exact `--output-file` declaration and regenerate. | +| `generation-output-mismatch` | The declared generator output is a different lock file. | Correct the generator command or validate the intended lock. | +| `generation-input-missing` | A recognized generator does not identify a usable source/package pin. | Restore the source requirement path or exact manual package pin, then regenerate. | +| `generation-input-outside-repository` | A declared `uv` source resolves outside the repository root. | Move or rewrite the source declaration so the resolved file stays inside the repository; do not expose the external payload to CI. | +| `generation-input-unreadable` | A declared `uv` source cannot be read as repository UTF-8 text. | Restore or re-encode the source requirements file before regenerating. | +| `generation-version-mismatch` | At least one generator/source exact pin disagrees with the lock. | Regenerate from all current source declarations and review the dependency delta. | +| `lock-path-outside-repository` | A discovered or directly validated lock resolves outside the repository root, including through a symlink. | Replace the escaping path/symlink with an in-repository lock before validation. | +| `lock-read-failed` | A contained lock/include path is missing, non-regular, unreadable, or not valid UTF-8. | Restore a readable repository-controlled UTF-8 requirements file; do not rely on traceback-only failure. | + +## TDD and acceptance evidence + +The first PR head intentionally introduced tests before the validator existed so collection failed closed rather than silently passing. Follow-up regressions cover a stale manual generator version, missing manual generator pin, manual extras, unpinned/unhashed declarations, malformed/orphan/duplicate hash structure, `.txt` and `.in` uv sources, missing or mismatched `uv` source/output bindings, traversal and symlink escapes, deterministic path-relative receipts without escaping payload disclosure, CLI exit behavior, the direct-script guard, the current repository lock inventory, job-scoped workflow ordering, and failure-receipt publication before CI exits. A later RED commit proves that both `-r` and `--requirement` previously bypassed included-file validation; the GREEN contract covers both forms, valid nested receipt counts/digests, deterministic output, missing and malformed targets, outside-root non-disclosure, cycle detection, and bounded-depth termination. + +The current review-edge RED additionally pins two latent failure modes from live review: a stale exact pin in an earlier source of a multi-input `uv pip compile` command must be checked rather than only the final `.txt`/`.in` path, and a non-UTF-8 included requirements file must return `lock-read-failed` instead of raising a traceback. The production repair iterates every declared source path and converts lock/source read failures to stable non-secret reason codes. + +For the current exact PR head, merge evidence remains the live protected-branch gate set, not this document and not predecessor-head success. Required CI/security/review evidence must be terminal and exact-head current before merge is considered. + +## Standards and primary technical grounding + +pip's current secure-install guidance defines `--require-hashes` as hash-checking mode and describes hash checking as protection against remote package tampering. pip's requirements-file format documents `-r` / `--requirement` as a supported include directive, so an attestation that skips those lines is incomplete. The Python Packaging User Guide distinguishes concrete requirements files used for repeatable complete-environment installations from abstract package dependency declarations. NIST SSDF v1.1 remains the final SP 800-218 publication and provides the broader secure-development and provenance-oriented practice context for protecting software and its components. This slice uses those sources to define a deterministic local evidence boundary; it does not claim that local declaration validation substitutes for remote artifact verification or the remaining issue #1229 controls. + +### References (APA 7th) + +National Institute of Standards and Technology. (2022). *Secure Software Development Framework (SSDF) version 1.1: Recommendations for mitigating the risk of software vulnerabilities* (NIST Special Publication 800-218). https://doi.org/10.6028/NIST.SP.800-218 + +Python Packaging Authority. (2026). *install_requires vs requirements files*. Python Packaging User Guide. https://packaging.python.org/en/latest/discussions/install-requires-vs-requirements/ + +Python Packaging Authority. (2026). *Secure installs*. pip documentation. https://pip.pypa.io/en/stable/topics/secure-installs/ + +Python Packaging Authority. (2026). *Requirements file format*. pip documentation. https://pip.pypa.io/en/stable/reference/requirements-file-format/ diff --git a/docs/doctoring/python-lock-registry-provenance.md b/docs/doctoring/python-lock-registry-provenance.md new file mode 100644 index 000000000..8563a15cc --- /dev/null +++ b/docs/doctoring/python-lock-registry-provenance.md @@ -0,0 +1,90 @@ +# 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. For changes that can alter Python lock or provenance evidence, Naruon therefore compares each exact project/version lock entry with trusted PyPI release metadata before dependency installation and requires at least one SHA-256 intersection with an eligible artifact. + +The deterministic offline lock-declaration validator remains unconditional in Application CI. The network-derived PyPI gate is intentionally scoped to supply-chain-relevant changes so an unrelated product PR does not become non-deterministically blocked by a public-index outage. If the workflow cannot establish a comparison base, it fails safe by requiring the network gate rather than silently skipping it. + +A passing registry receipt means the operator may continue to later dependency-install and platform-compatibility gates for the relevant change. A failing required 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 always runs deterministic offline lock provenance first. It then classifies the pull-request or push diff against the event base. Changes to `requirements*.txt`, either lock-provenance validator, their focused backend tests/doctoring, or the Application CI workflow itself require the network-derived PyPI evidence before dependency installation. Other changes skip only this public-network check; they do not skip exact-pin/hash validation, dependency installation with `--require-hashes`, tests, or the rest of the protected CI gate. A missing, zero, or otherwise unusable comparison base defaults to `required=true`. + +## Failure semantics + +When selected, the registry 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. Skipping the network gate because a diff is outside the supply-chain scope is a workflow routing decision, not a passing registry receipt and not evidence that PyPI was queried. + +## 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 risks are authority scope and availability. Proving a hash is published by PyPI is not the same as proving publisher identity, artifact intent, target compatibility, or absence of compromise. Conversely, a temporary PyPI availability failure is not evidence that unrelated Naruon product code is invalid. The CI scope therefore preserves fail-closed registry evidence whenever lock/provenance authority can change while keeping unrelated product CI independent of the public index. + +## 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, CI ordering before installation, and the diff-scoped/fail-safe network-gate contract. 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_provenance.py b/scripts/ci/python_lock_provenance.py new file mode 100644 index 000000000..d292fe5cc --- /dev/null +++ b/scripts/ci/python_lock_provenance.py @@ -0,0 +1,538 @@ +#!/usr/bin/env python3 +"""Validate deterministic offline provenance for hash-pinned Python locks. + +This utility deliberately validates only repository-controlled declarations: +exact pins, SHA-256 entries, generator-command binding, and source-lock version +agreement. It does not contact package indexes and therefore does not claim +artifact availability or registry provenance. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +from pathlib import Path +from typing import Iterable, cast + +SCHEMA_VERSION = "naruon.python-lock-provenance.v1" +_EXACT_PIN = re.compile( + r"^(?P[A-Za-z0-9][A-Za-z0-9._-]*(?:\[[A-Za-z0-9._,-]+\])?)" + r"==(?P[^\s\\;]+)(?:\s*;\s*[^\\]+)?\s*\\?$" +) +_SHA256 = re.compile(r"^--hash=sha256:(?P[0-9a-fA-F]{64})\s*\\?$") +_SHA256_PREFIX = "--hash=sha256:" +_MANUAL_PIN = re.compile( + r"(?[A-Za-z0-9][A-Za-z0-9._-]*(?:\[[A-Za-z0-9._,-]+\])?)" + r"==(?P[A-Za-z0-9][A-Za-z0-9.!+_-]*)" +) +_TEXT_PATH = re.compile( + r"(?[A-Za-z0-9_./-]+\.(?:txt|in))(?=\s|$)" +) +_REQUIREMENT_INCLUDE = re.compile( + r"^(?:-r\s*|--requirement(?:=|\s+))(?P\S+)\s*$" +) +_MAX_REQUIREMENT_INCLUDE_DEPTH = 32 + + +def _normalized_name(name: str) -> str: + """Return the canonical comparison form for one Python project name.""" + return re.sub(r"[-_.]+", "-", name.split("[", 1)[0].lower()) + + +def _relative_path(path: Path, repository_root: Path) -> str: + """Return a stable POSIX path without leaking an absolute runner location.""" + try: + relative = path.resolve().relative_to(repository_root.resolve()) + except ValueError: + return path.name + return relative.as_posix() + + +def _resolve_repository_path(path: Path, repository_root: Path) -> Path | None: + """Resolve ``path`` only when its target remains within ``repository_root``.""" + 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]: + """Create one stable machine-readable validation finding.""" + return {"code": code, "path": path, "detail": detail} + + +def _header_command(header_lines: list[str], marker: str) -> str | None: + """Return the first comment command containing ``marker``, if present.""" + for line in header_lines: + cleaned = line.lstrip("#").strip() + if marker in cleaned: + return cleaned + return None + + +def _parse_source_pins(text: str) -> dict[str, str]: + """Return exact direct pins declared by a source requirements file.""" + pins: dict[str, str] = {} + for raw_line in text.splitlines(): + stripped = re.split(r"\s+#", raw_line, maxsplit=1)[0].strip() + if not stripped or stripped.startswith(("#", "-")): + continue + match = _EXACT_PIN.fullmatch(stripped) + if match is not None: + pins[_normalized_name(match.group("name"))] = match.group("version") + return pins + + +def _parse_lock( + text: str, path: str +) -> tuple[list[str], dict[str, str], int, list[str], list[dict[str, str]]]: + """Parse pins, hashes, and requirement includes from one lock file.""" + header_lines: list[str] = [] + pins: dict[str, str] = {} + hash_count = 0 + include_paths: list[str] = [] + violations: list[dict[str, str]] = [] + current_label: str | None = None + current_hashes = 0 + seen_requirement = False + + def finalize() -> None: + nonlocal current_label, current_hashes + if current_label is not None and current_hashes == 0: + violations.append( + _violation( + "missing-sha256", + path, + f"{current_label} has no SHA-256 hash entry", + ) + ) + current_label = None + current_hashes = 0 + + for raw_line in text.splitlines(): + stripped = raw_line.strip() + if not stripped: + continue + if stripped.startswith("#"): + if not seen_requirement: + header_lines.append(stripped) + continue + if stripped.startswith("--hash="): + if current_label is None: + violations.append( + _violation( + "orphan-hash", + path, + "hash entry is not attached to a requirement", + ) + ) + continue + sha_match = _SHA256.fullmatch(stripped) + if sha_match is None: + if stripped.startswith(_SHA256_PREFIX): + violations.append( + _violation( + "malformed-sha256", + path, + f"{current_label} has a malformed SHA-256 digest", + ) + ) + continue + current_hashes += 1 + hash_count += 1 + continue + if stripped.startswith("-r") or stripped.startswith("--requirement"): + finalize() + seen_requirement = True + include_match = _REQUIREMENT_INCLUDE.fullmatch(stripped) + if include_match is None: + violations.append( + _violation( + "requirement-include-invalid", + path, + "requirements include must name exactly one file path", + ) + ) + else: + include_paths.append(include_match.group("path")) + continue + if stripped.startswith("-"): + continue + + finalize() + seen_requirement = True + match = _EXACT_PIN.fullmatch(stripped) + if match is None: + current_label = stripped.rstrip("\\").strip() + violations.append( + _violation( + "requirement-not-exactly-pinned", + path, + f"{current_label} is not an exact == pin", + ) + ) + continue + + current_label = f"{match.group('name')}=={match.group('version')}" + normalized_name = _normalized_name(match.group("name")) + if normalized_name in pins: + violations.append( + _violation( + "duplicate-requirement", + path, + f"{current_label} duplicates project {normalized_name}", + ) + ) + pins[normalized_name] = match.group("version") + + finalize() + return header_lines, pins, hash_count, include_paths, violations + + +def _validate_generation( + *, + repository_root: Path, + header_lines: list[str], + pins: dict[str, str], + relative_path: str, +) -> tuple[str, list[dict[str, str]]]: + """Validate recognized lock-generation declarations without network access.""" + violations: list[dict[str, str]] = [] + uv_command = _header_command(header_lines, "uv pip compile") + if uv_command is not None: + output_match = re.search(r"--output-file(?:=|\s+)(?P\S+)", uv_command) + if output_match is None: + violations.append( + _violation( + "generation-output-missing", + relative_path, + "uv generation command does not name --output-file", + ) + ) + elif Path(output_match.group("path")).as_posix() != relative_path: + violations.append( + _violation( + "generation-output-mismatch", + relative_path, + "uv generation output does not match the validated lock path", + ) + ) + + text_paths = [match.group("path") for match in _TEXT_PATH.finditer(uv_command)] + output_path = output_match.group("path") if output_match is not None else None + source_paths = [candidate for candidate in text_paths if candidate != output_path] + if not source_paths: + violations.append( + _violation( + "generation-input-missing", + relative_path, + "uv generation command does not name a source requirements file", + ) + ) + return "uv", violations + + for source_reference in source_paths: + source_path = repository_root / source_reference + resolved_source = _resolve_repository_path(source_path, repository_root) + if resolved_source is None: + violations.append( + _violation( + "generation-input-outside-repository", + relative_path, + "declared source resolves outside repository root", + ) + ) + continue + if not resolved_source.is_file(): + violations.append( + _violation( + "generation-input-missing", + relative_path, + "declared source requirements file is missing", + ) + ) + continue + try: + source_text = resolved_source.read_text(encoding="utf-8") + except (UnicodeDecodeError, OSError): + violations.append( + _violation( + "generation-input-unreadable", + relative_path, + "declared source is not readable as repository UTF-8 text", + ) + ) + continue + + source_pins = _parse_source_pins(source_text) + for name, version in sorted(source_pins.items()): + locked_version = pins.get(name) + if locked_version != version: + locked_description = locked_version or "missing" + violations.append( + _violation( + "generation-version-mismatch", + relative_path, + ( + f"source pin {name}=={version} is locked as " + f"{locked_description}" + ), + ) + ) + return "uv", violations + + pip_command = _header_command(header_lines, "pip download") + if pip_command is not None: + command_pins = { + _normalized_name(match.group("name")): match.group("version") + for match in _MANUAL_PIN.finditer(pip_command) + } + if not command_pins: + violations.append( + _violation( + "generation-input-missing", + relative_path, + "pip download generation command does not name an exact package pin", + ) + ) + return "pip-download", violations + for name, version in sorted(command_pins.items()): + locked_version = pins.get(name) + if locked_version != version: + locked_description = locked_version or "missing" + violations.append( + _violation( + "generation-version-mismatch", + relative_path, + ( + f"generator pin {name}=={version} is locked as " + f"{locked_description}" + ), + ) + ) + return "pip-download", violations + + return "manual", violations + + +def _failed_lock_receipt( + *, + relative_path: str, + code: str, + detail: str, +) -> dict[str, object]: + """Return a deterministic unread-lock receipt for one path failure.""" + return { + "path": relative_path, + "sha256": None, + "status": "failed", + "generation_mode": "unread", + "requirement_count": 0, + "sha256_hash_count": 0, + "included_files": [], + "violations": [_violation(code, relative_path, detail)], + } + + +def _validate_lock_tree( + lock_path: Path, + repository_root: Path, + *, + ancestors: tuple[Path, ...], +) -> dict[str, object]: + """Validate one lock and every safely contained requirements include.""" + relative_path = _relative_path(lock_path, repository_root) + resolved_lock = _resolve_repository_path(lock_path, repository_root) + if resolved_lock is None: + return _failed_lock_receipt( + relative_path=relative_path, + code="lock-path-outside-repository", + detail="lock path resolves outside repository root", + ) + if not resolved_lock.is_file(): + return _failed_lock_receipt( + relative_path=relative_path, + code="lock-read-failed", + detail="lock path is missing or not a regular file", + ) + try: + text = resolved_lock.read_text(encoding="utf-8") + except (UnicodeDecodeError, OSError): + return _failed_lock_receipt( + relative_path=relative_path, + code="lock-read-failed", + detail="lock is not readable as repository UTF-8 text", + ) + + ( + header_lines, + pins, + hash_count, + include_paths, + violations, + ) = _parse_lock(text, relative_path) + included_files: list[dict[str, object]] = [] + requirement_count = len(pins) + + for include_reference in include_paths: + if len(ancestors) >= _MAX_REQUIREMENT_INCLUDE_DEPTH: + violations.append( + _violation( + "requirement-include-depth-exceeded", + relative_path, + "requirements include depth exceeds the bounded validation limit", + ) + ) + continue + + include_candidate = resolved_lock.parent / include_reference + resolved_include = _resolve_repository_path(include_candidate, repository_root) + if resolved_include is None: + violations.append( + _violation( + "requirement-include-outside-repository", + relative_path, + "included requirements path resolves outside repository root", + ) + ) + continue + if resolved_include in (*ancestors, resolved_lock): + violations.append( + _violation( + "requirement-include-cycle", + relative_path, + "requirements include graph contains a cycle", + ) + ) + continue + if not resolved_include.is_file(): + violations.append( + _violation( + "requirement-include-missing", + relative_path, + "included requirements file is missing or not a regular file", + ) + ) + continue + + child_receipt = _validate_lock_tree( + resolved_include, + repository_root, + ancestors=(*ancestors, resolved_lock), + ) + included_files.append(child_receipt) + requirement_count += cast(int, child_receipt["requirement_count"]) + hash_count += cast(int, child_receipt["sha256_hash_count"]) + violations.extend( + cast(list[dict[str, str]], child_receipt["violations"]) + ) + + generation_mode, generation_violations = _validate_generation( + repository_root=repository_root, + header_lines=header_lines, + pins=pins, + relative_path=relative_path, + ) + violations.extend(generation_violations) + violations.sort(key=lambda item: (item["code"], item["path"], item["detail"])) + return { + "path": relative_path, + "sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(), + "status": "failed" if violations else "passed", + "generation_mode": generation_mode, + "requirement_count": requirement_count, + "sha256_hash_count": hash_count, + "included_files": included_files, + "violations": violations, + } + + +def validate_lock_file(lock_path: Path, repository_root: Path) -> dict[str, object]: + """Validate one in-repository lock and its bounded include graph.""" + return _validate_lock_tree(lock_path, repository_root, ancestors=()) + + +def discover_hash_locks(repository_root: Path) -> list[Path]: + """Discover hash locks without reading files whose targets escape the repository.""" + 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_path = _resolve_repository_path(path, repository_root) + if resolved_path is None: + candidates.append(path) + continue + if not resolved_path.is_file(): + continue + try: + text = resolved_path.read_text(encoding="utf-8") + except UnicodeDecodeError: + continue + if _SHA256_PREFIX 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(repository_root: Path) -> dict[str, object]: + """Validate every active Python hash lock and aggregate one repository receipt.""" + lock_receipts = [ + validate_lock_file(path, repository_root) + for path in discover_hash_locks(repository_root) + ] + violations = [ + violation + for receipt in lock_receipts + for violation in receipt["violations"] + if isinstance(violation, dict) + ] + 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 validation.""" + parser = argparse.ArgumentParser( + description="Validate offline provenance declarations for Python hash locks." + ) + 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 the deterministic JSON receipt to stdout.", + ) + return parser + + +def main(argv: Iterable[str] | None = None) -> int: + """Run repository 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(args.repository_root) + if args.json: + print(json.dumps(receipt, sort_keys=True, separators=(",", ":"))) + else: + print(f"Python lock 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()) \ No newline at end of file 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())