|
| 1 | +"""Pure D3 evidence helpers, including the shared exact-bundle validator.""" |
| 2 | +from __future__ import annotations |
| 3 | + |
| 4 | +import hashlib |
| 5 | +import os |
| 6 | +import re |
| 7 | +import stat |
| 8 | +from collections.abc import Iterable, Mapping |
| 9 | +from pathlib import Path |
| 10 | + |
| 11 | +FIXED_FILES = ("manifest.json", "report.html", "report.json") |
| 12 | +EVIDENCE_VERSION = "qar.d3.build_evidence.v4" |
| 13 | +DIST_RE = re.compile(r"^([A-Za-z0-9_.-]+)==([^\s=]+)$") |
| 14 | +DEPENDENCY_INVENTORY = [ |
| 15 | + ".github/workflows/qar_d3_daily_preview_artifact.yml", "pyproject.toml", "uv.lock", |
| 16 | + "scripts/d3_build_daily_preview.py", "scripts/d3_verify_daily_preview.py", "scripts/d3_evidence.py", |
| 17 | + "src/quant_advisor_research/advisory_report.py", "src/quant_advisor_research/artifact_integrity.py", |
| 18 | + "src/quant_advisor_research/artifacts.py", "src/quant_advisor_research/contracts.py", |
| 19 | + "src/quant_advisor_research/csv_utils.py", "src/quant_advisor_research/period_contract.py", |
| 20 | + "src/quant_advisor_research/preview_bundle.py", "src/quant_advisor_research/preview_workspace.py", |
| 21 | + "src/quant_advisor_research/time_contract.py", "tests/test_d3_exact_bundle.py", |
| 22 | + "examples/political_events.example.csv", "examples/political_watchlist.example.csv", |
| 23 | +] |
| 24 | + |
| 25 | + |
| 26 | +class EvidenceContractError(ValueError): |
| 27 | + def __init__(self, code: str) -> None: |
| 28 | + self.code = code |
| 29 | + super().__init__(code) |
| 30 | + |
| 31 | + |
| 32 | +def validate_exact_bundle(workspace: str | Path) -> dict[str, Path]: |
| 33 | + """Validate all directory members before any caller reads or hashes them.""" |
| 34 | + root = Path(workspace) |
| 35 | + try: |
| 36 | + root_info = root.lstat() |
| 37 | + if not stat.S_ISDIR(root_info.st_mode) or stat.S_ISLNK(root_info.st_mode): |
| 38 | + raise EvidenceContractError("bundle_directory_invalid") |
| 39 | + entries = list(os.scandir(root)) |
| 40 | + if {entry.name for entry in entries} != set(FIXED_FILES) or len(entries) != len(FIXED_FILES): |
| 41 | + raise EvidenceContractError("bundle_member_set_invalid") |
| 42 | + result: dict[str, Path] = {} |
| 43 | + for entry in entries: |
| 44 | + info = entry.stat(follow_symlinks=False) |
| 45 | + if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode) or info.st_nlink != 1: |
| 46 | + raise EvidenceContractError("bundle_member_invalid") |
| 47 | + result[entry.name] = root / entry.name |
| 48 | + return {name: result[name] for name in FIXED_FILES} |
| 49 | + except EvidenceContractError: |
| 50 | + raise |
| 51 | + except (OSError, TypeError, ValueError): |
| 52 | + raise EvidenceContractError("bundle_member_invalid") from None |
| 53 | + |
| 54 | + |
| 55 | +def canonical_distribution_snapshot(values: Iterable[str]) -> tuple[list[str], str]: |
| 56 | + try: |
| 57 | + raw = list(values) |
| 58 | + normalized: dict[str, str] = {} |
| 59 | + for value in raw: |
| 60 | + if type(value) is not str or not (match := DIST_RE.fullmatch(value)): |
| 61 | + raise EvidenceContractError("distribution_snapshot_invalid") |
| 62 | + name = re.sub(r"[-_.]+", "-", match.group(1)).lower() |
| 63 | + item = f"{name}=={match.group(2)}" |
| 64 | + if name in normalized and normalized[name] != item: |
| 65 | + raise EvidenceContractError("distribution_snapshot_conflict") |
| 66 | + normalized[name] = item |
| 67 | + snapshot = sorted(normalized.values()) |
| 68 | + digest = hashlib.sha256("\n".join(snapshot).encode()).hexdigest() |
| 69 | + return snapshot, digest |
| 70 | + except EvidenceContractError: |
| 71 | + raise |
| 72 | + except (TypeError, ValueError, UnicodeError): |
| 73 | + raise EvidenceContractError("distribution_snapshot_invalid") from None |
| 74 | + |
| 75 | + |
| 76 | +def locked_environment_evidence(*, lock_sha256: str, uv_version: str, python_version: str, distributions: Iterable[str]) -> dict[str, object]: |
| 77 | + if type(lock_sha256) is not str or not re.fullmatch(r"[0-9a-f]{64}", lock_sha256): |
| 78 | + raise EvidenceContractError("lock_digest_invalid") |
| 79 | + if type(uv_version) is not str or not uv_version.startswith("uv "): |
| 80 | + raise EvidenceContractError("uv_version_invalid") |
| 81 | + if type(python_version) is not str or not re.fullmatch(r"3\.11(?:\.\d+)?", python_version): |
| 82 | + raise EvidenceContractError("python_version_invalid") |
| 83 | + snapshot, digest = canonical_distribution_snapshot(distributions) |
| 84 | + return {"lock_sha256": lock_sha256, "uv_version": uv_version, "python_version": python_version, "installed_distributions": snapshot, "installed_distributions_sha256": digest} |
| 85 | + |
| 86 | + |
| 87 | +def repository_file_hashes(repo_root: str | Path, paths: Iterable[str]) -> dict[str, str]: |
| 88 | + root = Path(repo_root) |
| 89 | + try: |
| 90 | + if not stat.S_ISDIR(root.lstat().st_mode): |
| 91 | + raise EvidenceContractError("dependency_root_invalid") |
| 92 | + result: dict[str, str] = {} |
| 93 | + for raw in paths: |
| 94 | + relative = Path(raw) |
| 95 | + if type(raw) is not str or not raw or relative.is_absolute() or ".." in relative.parts or raw in result: |
| 96 | + raise EvidenceContractError("dependency_path_invalid") |
| 97 | + target = root / relative |
| 98 | + info = target.lstat() |
| 99 | + if not stat.S_ISREG(info.st_mode) or stat.S_ISLNK(info.st_mode) or info.st_nlink != 1: |
| 100 | + raise EvidenceContractError("dependency_file_invalid") |
| 101 | + result[raw] = hashlib.sha256(target.read_bytes()).hexdigest() |
| 102 | + return dict(sorted(result.items())) |
| 103 | + except EvidenceContractError: |
| 104 | + raise |
| 105 | + except (OSError, TypeError, ValueError, UnicodeError): |
| 106 | + raise EvidenceContractError("dependency_file_invalid") from None |
| 107 | + |
| 108 | + |
| 109 | +def require_exact_locked_environment(value: object, expected: Mapping[str, object]) -> None: |
| 110 | + if not isinstance(value, Mapping) or dict(value) != dict(expected): |
| 111 | + raise EvidenceContractError("locked_environment_mismatch") |
0 commit comments