Spaces:
Running
Running
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 19)
da0ce7b verified | #!/usr/bin/env python3 | |
| """Deterministic, credential-free release read-back comparison. | |
| This module never opens a socket and never accepts credentials. It compares | |
| an already captured HTTP response set or an already downloaded FTPS/WebFTP | |
| tree with the exact per-file SHA-256 manifest carried by a built package. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import hashlib | |
| import json | |
| import re | |
| import sys | |
| from collections.abc import Mapping | |
| from dataclasses import dataclass | |
| from pathlib import Path, PurePosixPath | |
| from typing import Any | |
| FILE_HASH_MANIFEST_NAME = "HOMEPL_FILE_HASHES.json" | |
| FILE_HASH_MANIFEST_SCHEMA = "aureon.homepl-file-hashes.v1" | |
| _COMMIT = re.compile(r"^[0-9a-f]{40}$") | |
| _SHA256 = re.compile(r"^[0-9a-f]{64}$") | |
| class ReadbackInputError(ValueError): | |
| """Raised when a manifest or captured observation is unsafe or malformed.""" | |
| class FileHashRecord: | |
| """One immutable expected public file.""" | |
| path: str | |
| bytes: int | |
| sha256: str | |
| class FileHashManifest: | |
| """Validated release manifest with no recursive self-hash.""" | |
| source_commit: str | |
| records: tuple[FileHashRecord, ...] | |
| class HttpObservation: | |
| """Captured public response bytes; no URL, headers, cookies, or credentials.""" | |
| status_code: int | |
| body: bytes | |
| class ReadbackFinding: | |
| """One deterministic mismatch safe to retain in a release receipt.""" | |
| path: str | |
| check: str | |
| expected: str | |
| observed: str | |
| def to_dict(self) -> dict[str, str]: | |
| return { | |
| "path": self.path, | |
| "check": self.check, | |
| "expected": self.expected, | |
| "observed": self.observed, | |
| } | |
| class ReadbackReport: | |
| """Credential-free exact comparison result.""" | |
| method: str | |
| source_commit: str | |
| expected_file_count: int | |
| observed_file_count: int | |
| findings: tuple[ReadbackFinding, ...] | |
| def passed(self) -> bool: | |
| return not self.findings and self.observed_file_count == self.expected_file_count | |
| def to_dict(self) -> dict[str, object]: | |
| return { | |
| "schema": "aureon.homepl-readback-comparison.v1", | |
| "method": self.method, | |
| "source_commit": self.source_commit, | |
| "expected_file_count": self.expected_file_count, | |
| "observed_file_count": self.observed_file_count, | |
| "passed": self.passed, | |
| "findings": [finding.to_dict() for finding in self.findings], | |
| "credentials_recorded": False, | |
| "network_access_performed": False, | |
| } | |
| def _safe_relative_path(value: object) -> str: | |
| if not isinstance(value, str) or not value or "\\" in value or "\t" in value or "\n" in value: | |
| raise ReadbackInputError("manifest path must be a non-empty forward-slash relative path") | |
| path = PurePosixPath(value) | |
| if path.is_absolute() or ".." in path.parts or "." in path.parts: | |
| raise ReadbackInputError(f"unsafe manifest path: {value}") | |
| return path.as_posix() | |
| def _mapping(value: object, label: str) -> Mapping[str, Any]: | |
| if not isinstance(value, Mapping) or not all(isinstance(key, str) for key in value): | |
| raise ReadbackInputError(f"{label} must be an object with string keys") | |
| return value | |
| def parse_file_hash_manifest(payload: bytes | str) -> FileHashManifest: | |
| """Parse and strictly validate a deterministic package hash manifest.""" | |
| try: | |
| text = payload.decode("utf-8") if isinstance(payload, bytes) else payload | |
| except UnicodeDecodeError as exc: | |
| raise ReadbackInputError("file hash manifest is not UTF-8") from exc | |
| try: | |
| raw = json.loads(text) | |
| except json.JSONDecodeError as exc: | |
| raise ReadbackInputError("file hash manifest is not valid JSON") from exc | |
| data = _mapping(raw, "file hash manifest") | |
| if data.get("schema") != FILE_HASH_MANIFEST_SCHEMA: | |
| raise ReadbackInputError("unsupported file hash manifest schema") | |
| if data.get("algorithm") != "sha256" or data.get("manifest_self_included") is not False: | |
| raise ReadbackInputError("manifest must use SHA-256 and explicitly exclude itself") | |
| source_commit = data.get("source_commit") | |
| if not isinstance(source_commit, str) or not _COMMIT.fullmatch(source_commit): | |
| raise ReadbackInputError("manifest source_commit must be lowercase full-length hexadecimal") | |
| raw_records = data.get("records") | |
| if not isinstance(raw_records, list): | |
| raise ReadbackInputError("manifest records must be a list") | |
| if data.get("record_count") != len(raw_records): | |
| raise ReadbackInputError("manifest record_count does not match records") | |
| records: list[FileHashRecord] = [] | |
| for index, item in enumerate(raw_records): | |
| row = _mapping(item, f"record {index}") | |
| path = _safe_relative_path(row.get("path")) | |
| size = row.get("bytes") | |
| digest = row.get("sha256") | |
| if not isinstance(size, int) or isinstance(size, bool) or size < 0: | |
| raise ReadbackInputError(f"record {index} has an invalid byte count") | |
| if not isinstance(digest, str) or not _SHA256.fullmatch(digest): | |
| raise ReadbackInputError(f"record {index} has an invalid SHA-256") | |
| if path == FILE_HASH_MANIFEST_NAME: | |
| raise ReadbackInputError("file hash manifest must not hash itself") | |
| records.append(FileHashRecord(path=path, bytes=size, sha256=digest)) | |
| paths = [record.path for record in records] | |
| if paths != sorted(paths) or len(paths) != len(set(paths)): | |
| raise ReadbackInputError("manifest paths must be unique and sorted") | |
| return FileHashManifest(source_commit=source_commit, records=tuple(records)) | |
| def load_file_hash_manifest(package_dir: Path) -> tuple[FileHashManifest, bytes]: | |
| """Load the bounded package manifest and its exact bytes.""" | |
| package = package_dir.resolve(strict=True) | |
| if not package.is_dir(): | |
| raise ReadbackInputError("package_dir must be a directory") | |
| path = package / FILE_HASH_MANIFEST_NAME | |
| if not path.is_file(): | |
| raise ReadbackInputError(f"package is missing {FILE_HASH_MANIFEST_NAME}") | |
| payload = path.read_bytes() | |
| return parse_file_hash_manifest(payload), payload | |
| def _digest(path: Path) -> tuple[int, str]: | |
| digest = hashlib.sha256() | |
| size = 0 | |
| with path.open("rb") as handle: | |
| for chunk in iter(lambda: handle.read(131_072), b""): | |
| size += len(chunk) | |
| digest.update(chunk) | |
| return size, digest.hexdigest() | |
| def _directory_observation(root: Path, relative: str) -> tuple[int, str] | None: | |
| candidate = (root / Path(*PurePosixPath(relative).parts)).resolve() | |
| try: | |
| candidate.relative_to(root) | |
| except ValueError as exc: | |
| raise ReadbackInputError(f"read-back path escapes root: {relative}") from exc | |
| if not candidate.is_file(): | |
| return None | |
| return _digest(candidate) | |
| def _regular_file_paths(root: Path) -> tuple[set[str], tuple[ReadbackFinding, ...]]: | |
| """Enumerate an exact mirror without following links or accepting unsafe entries.""" | |
| paths: set[str] = set() | |
| findings: list[ReadbackFinding] = [] | |
| for candidate in sorted(root.rglob("*"), key=lambda item: item.relative_to(root).as_posix()): | |
| relative = candidate.relative_to(root).as_posix() | |
| if candidate.is_symlink(): | |
| findings.append(ReadbackFinding(relative, "unsafe-entry", "ordinary file/directory", "symlink")) | |
| continue | |
| if candidate.is_file(): | |
| paths.add(_safe_relative_path(relative)) | |
| return paths, tuple(findings) | |
| def compare_readback_directory(package_dir: Path, readback_dir: Path) -> ReadbackReport: | |
| """Compare an already downloaded remote tree with the exact package.""" | |
| manifest, manifest_bytes = load_file_hash_manifest(package_dir) | |
| observed_root = readback_dir.resolve(strict=True) | |
| if not observed_root.is_dir(): | |
| raise ReadbackInputError("readback_dir must be a directory") | |
| expected_manifest = FileHashRecord( | |
| path=FILE_HASH_MANIFEST_NAME, | |
| bytes=len(manifest_bytes), | |
| sha256=hashlib.sha256(manifest_bytes).hexdigest(), | |
| ) | |
| expected = (expected_manifest, *manifest.records) | |
| observed_paths, unsafe_findings = _regular_file_paths(observed_root) | |
| findings: list[ReadbackFinding] = list(unsafe_findings) | |
| expected_paths = {record.path for record in expected} | |
| for record in expected: | |
| observed = _directory_observation(observed_root, record.path) | |
| if observed is None: | |
| findings.append(ReadbackFinding(record.path, "presence", "file", "missing")) | |
| continue | |
| size, digest = observed | |
| if size != record.bytes: | |
| findings.append(ReadbackFinding(record.path, "bytes", str(record.bytes), str(size))) | |
| if digest != record.sha256: | |
| findings.append(ReadbackFinding(record.path, "sha256", record.sha256, digest)) | |
| for unexpected in sorted(observed_paths - expected_paths): | |
| findings.append(ReadbackFinding(unexpected, "unexpected-file", "absent", "present")) | |
| return ReadbackReport( | |
| method="downloaded-directory", | |
| source_commit=manifest.source_commit, | |
| expected_file_count=len(expected), | |
| observed_file_count=len(observed_paths), | |
| findings=tuple(findings), | |
| ) | |
| def compare_http_observations( | |
| package_dir: Path, | |
| observations: Mapping[str, HttpObservation], | |
| ) -> ReadbackReport: | |
| """Compare every expected package path from previously captured HTTP bodies. | |
| This proves the requested package paths, but unlike an exact downloaded | |
| served-root mirror it cannot discover unknown stale public paths. | |
| """ | |
| manifest, manifest_bytes = load_file_hash_manifest(package_dir) | |
| normalised: dict[str, HttpObservation] = {} | |
| for path, observation in observations.items(): | |
| safe = _safe_relative_path(path) | |
| if safe in normalised: | |
| raise ReadbackInputError(f"duplicate HTTP observation: {safe}") | |
| if not isinstance(observation, HttpObservation): | |
| raise ReadbackInputError(f"HTTP observation for {safe} has the wrong type") | |
| normalised[safe] = observation | |
| expected_manifest = FileHashRecord( | |
| path=FILE_HASH_MANIFEST_NAME, | |
| bytes=len(manifest_bytes), | |
| sha256=hashlib.sha256(manifest_bytes).hexdigest(), | |
| ) | |
| expected = (expected_manifest, *manifest.records) | |
| findings: list[ReadbackFinding] = [] | |
| observed_count = 0 | |
| for record in expected: | |
| current = normalised.get(record.path) | |
| if current is None: | |
| findings.append(ReadbackFinding(record.path, "presence", "HTTP 200 body", "missing")) | |
| continue | |
| observed_count += 1 | |
| if current.status_code != 200: | |
| findings.append(ReadbackFinding(record.path, "http-status", "200", str(current.status_code))) | |
| continue | |
| size = len(current.body) | |
| digest = hashlib.sha256(current.body).hexdigest() | |
| if size != record.bytes: | |
| findings.append(ReadbackFinding(record.path, "bytes", str(record.bytes), str(size))) | |
| if digest != record.sha256: | |
| findings.append(ReadbackFinding(record.path, "sha256", record.sha256, digest)) | |
| return ReadbackReport( | |
| method="captured-http-expected-paths", | |
| source_commit=manifest.source_commit, | |
| expected_file_count=len(expected), | |
| observed_file_count=observed_count, | |
| findings=tuple(findings), | |
| ) | |
| def main(argv: list[str] | None = None) -> int: | |
| parser = argparse.ArgumentParser( | |
| description="Compare an already downloaded Home.pl tree with an exact built package (no network)." | |
| ) | |
| parser.add_argument("--package", required=True, help="built package directory") | |
| parser.add_argument("--readback-dir", required=True, help="already downloaded served-root tree") | |
| args = parser.parse_args(argv) | |
| try: | |
| report = compare_readback_directory(Path(args.package), Path(args.readback_dir)) | |
| except (OSError, ReadbackInputError) as exc: | |
| print(f"read-back comparison refused: {exc}", file=sys.stderr) | |
| return 2 | |
| print(json.dumps(report.to_dict(), indent=2, sort_keys=True)) | |
| return 0 if report.passed else 1 | |
| if __name__ == "__main__": # pragma: no cover - CLI wrapper | |
| raise SystemExit(main()) | |