|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Package an explicit, hash-verified prepublication evidence snapshot. |
| 3 | +
|
| 4 | +This does not record acceptance or publish anything. All required prepublication |
| 5 | +cases except EVIDENCE-01 must already pass. EVIDENCE-01 stays pending while this |
| 6 | +bundle is produced, avoiding a bundle containing its own acceptance hash. |
| 7 | +""" |
| 8 | +from __future__ import annotations |
| 9 | + |
| 10 | +import argparse |
| 11 | +import gzip |
| 12 | +import hashlib |
| 13 | +import json |
| 14 | +import os |
| 15 | +from pathlib import Path, PurePosixPath |
| 16 | +import re |
| 17 | +import stat |
| 18 | +import sys |
| 19 | +import tarfile |
| 20 | +import tempfile |
| 21 | + |
| 22 | +import verify_acceptance_4_0_0 as acceptance |
| 23 | + |
| 24 | + |
| 25 | +VERSION = "4.0.0" |
| 26 | +PREFIX = f"shadowfetch-{VERSION}-evidence" |
| 27 | +BUNDLE = f"evidence-bundle-{VERSION}.tar.gz" |
| 28 | +CONTENTS = f"evidence-bundle-{VERSION}.contents" |
| 29 | +GENERATED = tuple(f"{stem}-{VERSION}{suffix}" for stem, suffix in ( |
| 30 | + ("dossier", ".md"), ("packages", ".manifest"), ("sbom", ".cdx.json"), |
| 31 | + ("sbom-sources", ".txt"), ("release-facts", ".json"), |
| 32 | +)) |
| 33 | +CHECKSUMS = f"release-evidence-{VERSION}.sha256" |
| 34 | +QA_SOURCES = ( |
| 35 | + "tools/build_release_evidence_4_0_0.py", |
| 36 | + "tools/package_release_evidence_4_0_0.py", |
| 37 | + "tools/verify_acceptance_4_0_0.py", |
| 38 | + "tools/qa_4_0_0/README.md", |
| 39 | + "tools/qa_4_0_0/vm_harness.sh", |
| 40 | + "tools/qa_4_0_0/qga_exec.py", |
| 41 | + "tools/qa_4_0_0/stress_45m.sh", |
| 42 | + "tools/qa_4_0_0/mission_stress.py", |
| 43 | + "tools/qa_4_0_0/container_stress.py", |
| 44 | + "tools/qa_4_0_0/latency_probe.py", |
| 45 | + "tools/qa_4_0_0/engine_acceptance.py", |
| 46 | + "tools/qa_4_0_0/native_mission_acceptance.py", |
| 47 | + "tools/qa_4_0_0/upgrade_recovery_acceptance.py", |
| 48 | + "tools/qa_4_0_0/installed_audit.sh", |
| 49 | + "tools/qa_4_0_0/native_ui_probe.py", |
| 50 | + "tools/qa_4_0_0/native_atspi.py", |
| 51 | +) |
| 52 | +MAX_FILE = 256 * 1024**2 |
| 53 | +MAX_TOTAL = 1024**3 |
| 54 | +MAX_FILES = 512 |
| 55 | +HEX = re.compile(r"[0-9a-fA-F]{64}\Z") |
| 56 | +RASTER = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp", ".ppm"} |
| 57 | + |
| 58 | + |
| 59 | +def relative_name(value: str) -> str: |
| 60 | + if not isinstance(value, str) or not value or len(value) > 512: |
| 61 | + raise ValueError("Expected a bounded relative file path") |
| 62 | + path = PurePosixPath(value) |
| 63 | + if (path.is_absolute() or "\\" in value or |
| 64 | + any(ord(c) < 32 or ord(c) == 127 for c in value) or |
| 65 | + any(part in ("", ".", "..") for part in value.split("/"))): |
| 66 | + raise ValueError(f"Unsafe or escaping relative path: {value!r}") |
| 67 | + return path.as_posix() |
| 68 | + |
| 69 | + |
| 70 | +def checked_path(root: Path, value: str, *, directory: bool = False) -> Path: |
| 71 | + value = relative_name(value) |
| 72 | + current = root |
| 73 | + for part in value.split("/"): |
| 74 | + current = current / part |
| 75 | + if current.is_symlink(): |
| 76 | + raise ValueError(f"Symbolic links are not evidence inputs: {value}") |
| 77 | + current.resolve().relative_to(root) |
| 78 | + mode = current.stat().st_mode |
| 79 | + if not (stat.S_ISDIR(mode) if directory else stat.S_ISREG(mode)): |
| 80 | + raise ValueError(f"Expected a regular {'directory' if directory else 'file'}: {value}") |
| 81 | + return current |
| 82 | + |
| 83 | + |
| 84 | +def expected_hash(value: str) -> str: |
| 85 | + if not isinstance(value, str) or not HEX.fullmatch(value): |
| 86 | + raise ValueError("Expected a SHA256 digest with 64 hexadecimal characters") |
| 87 | + return value.lower() |
| 88 | + |
| 89 | + |
| 90 | +def json_bytes(value) -> bytes: |
| 91 | + return (json.dumps(value, indent=2, sort_keys=True) + "\n").encode() |
| 92 | + |
| 93 | + |
| 94 | +class Snapshot: |
| 95 | + def __init__(self, root: Path, staging: Path): |
| 96 | + self.root, self.staging = root, staging |
| 97 | + self.entries: dict[str, dict] = {} |
| 98 | + self.total = 0 |
| 99 | + |
| 100 | + def _reserve(self, name: str, size: int): |
| 101 | + relative_name(name) |
| 102 | + if name in self.entries: |
| 103 | + raise ValueError(f"Duplicate archive destination: {name}") |
| 104 | + if size > MAX_FILE or self.total + size > MAX_TOTAL or len(self.entries) >= MAX_FILES: |
| 105 | + raise ValueError("Evidence bundle exceeds bounded file/count/total limits") |
| 106 | + |
| 107 | + def add_bytes(self, name: str, data: bytes): |
| 108 | + self._reserve(name, len(data)) |
| 109 | + path = self.staging / name |
| 110 | + path.parent.mkdir(parents=True, exist_ok=True) |
| 111 | + path.write_bytes(data) |
| 112 | + self.entries[name] = {"sha256": hashlib.sha256(data).hexdigest(), "size": len(data)} |
| 113 | + self.total += len(data) |
| 114 | + |
| 115 | + def add_file(self, name: str, source: str, digest: str | None = None): |
| 116 | + path = checked_path(self.root, source) |
| 117 | + if path.name in (BUNDLE, CONTENTS): |
| 118 | + raise ValueError("Circular evidence-bundle reference is forbidden") |
| 119 | + size = path.stat().st_size |
| 120 | + self._reserve(name, size) |
| 121 | + destination = self.staging / name |
| 122 | + destination.parent.mkdir(parents=True, exist_ok=True) |
| 123 | + actual = hashlib.sha256() |
| 124 | + fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) |
| 125 | + with os.fdopen(fd, "rb") as stream, destination.open("wb") as out: |
| 126 | + before = os.fstat(stream.fileno()) |
| 127 | + if not stat.S_ISREG(before.st_mode) or before.st_size != size: |
| 128 | + raise ValueError(f"Input changed while opening: {source}") |
| 129 | + for chunk in iter(lambda: stream.read(1024**2), b""): |
| 130 | + actual.update(chunk) |
| 131 | + out.write(chunk) |
| 132 | + if out.tell() > size: |
| 133 | + raise ValueError(f"Input grew while copying: {source}") |
| 134 | + after = os.fstat(stream.fileno()) |
| 135 | + if (before.st_size, before.st_mtime_ns, before.st_ino) != (after.st_size, after.st_mtime_ns, after.st_ino): |
| 136 | + raise ValueError(f"Input changed while copying: {source}") |
| 137 | + checked_path(self.root, source) |
| 138 | + if destination.stat().st_size != size: |
| 139 | + raise ValueError(f"Input truncated while copying: {source}") |
| 140 | + actual_hash = actual.hexdigest() |
| 141 | + if digest is not None and actual_hash != expected_hash(digest): |
| 142 | + raise ValueError(f"SHA256 mismatch: {source}") |
| 143 | + self.entries[name] = {"sha256": actual_hash, "size": size} |
| 144 | + self.total += size |
| 145 | + |
| 146 | + |
| 147 | +def approve_inputs(root: Path, path: str) -> dict: |
| 148 | + approval_file = checked_path(root, path) |
| 149 | + if approval_file.stat().st_size > 1024**2: |
| 150 | + raise ValueError("Approved input manifest exceeds 1 MiB") |
| 151 | + data = json.loads(approval_file.read_text()) |
| 152 | + if not isinstance(data, dict) or set(data) != {"schema_version", "screenshots", "documents"} or data["schema_version"] != 1: |
| 153 | + raise ValueError("Approval input requires schema_version=1, screenshots and documents") |
| 154 | + for category in ("screenshots", "documents"): |
| 155 | + if not isinstance(data[category], list) or len(data[category]) > MAX_FILES: |
| 156 | + raise ValueError(f"Invalid approved {category} array") |
| 157 | + seen = set() |
| 158 | + for item in data[category]: |
| 159 | + if not isinstance(item, dict) or set(item) != {"path", "sha256", "approved"} or item["approved"] is not True: |
| 160 | + raise ValueError(f"Every {category} input needs explicit approved:true, path and sha256") |
| 161 | + name = relative_name(item["path"]) |
| 162 | + expected_hash(item["sha256"]) |
| 163 | + if name in seen: |
| 164 | + raise ValueError(f"Duplicate approved input: {name}") |
| 165 | + seen.add(name) |
| 166 | + suffix = PurePosixPath(name).suffix.lower() |
| 167 | + if category == "screenshots" and suffix != ".png": |
| 168 | + raise ValueError("Approved screenshots must be actual PNG captures") |
| 169 | + if category == "documents" and suffix not in {".md", ".txt", ".pdf", ".docx", ".html", ".json"}: |
| 170 | + raise ValueError(f"Unsupported release document type: {name}") |
| 171 | + return data |
| 172 | + |
| 173 | + |
| 174 | +def validate_prepublication(data: dict): |
| 175 | + errors = acceptance.validate_manifest(data) |
| 176 | + for case in data.get("cases", []): |
| 177 | + if not isinstance(case, dict): |
| 178 | + continue |
| 179 | + if case.get("phase") == "postpublish" and (case.get("status") != "pending" or case.get("evidence")): |
| 180 | + errors.append("Prepublication bundle cannot claim publication has passed") |
| 181 | + if case.get("id") == "EVIDENCE-01": |
| 182 | + if case.get("status") != "pending": |
| 183 | + errors.append("EVIDENCE-01 must remain pending until bundle verification") |
| 184 | + elif case.get("required") and case.get("phase") == "prepublish": |
| 185 | + if case.get("status") != "pass" or not case.get("evidence"): |
| 186 | + errors.append(f"{case.get('id')}: required prepublication evidence has not passed") |
| 187 | + if not any(c.get("id") == "EVIDENCE-01" for c in data.get("cases", []) if isinstance(c, dict)): |
| 188 | + errors.append("Missing EVIDENCE-01 packaging case") |
| 189 | + artifact = data.get("artifact", {}) |
| 190 | + if artifact.get("evidence_bundle_sha256"): |
| 191 | + errors.append("Circular bundle digest: package before recording artifact.evidence_bundle_sha256") |
| 192 | + if errors: |
| 193 | + raise ValueError("; ".join(errors)) |
| 194 | + |
| 195 | + |
| 196 | +def package(root: Path, output_dir: str, approved_path: str) -> dict: |
| 197 | + if root.is_symlink(): |
| 198 | + raise ValueError("Repository root cannot be a symbolic link") |
| 199 | + root = root.resolve(strict=True) |
| 200 | + output = checked_path(root, output_dir, directory=True) |
| 201 | + for name in (BUNDLE, CONTENTS): |
| 202 | + if (output / name).exists() or (output / name).is_symlink(): |
| 203 | + raise ValueError(f"Preserve the prior snapshot before rebuilding: {output / name}") |
| 204 | + manifest_path = checked_path(root, "qa/4.0.0/acceptance.json") |
| 205 | + if manifest_path.stat().st_size > 8 * 1024**2: |
| 206 | + raise ValueError("Acceptance manifest exceeds 8 MiB") |
| 207 | + manifest_bytes = manifest_path.read_bytes() |
| 208 | + data = json.loads(manifest_bytes) |
| 209 | + validate_prepublication(data) |
| 210 | + approvals = approve_inputs(root, approved_path) |
| 211 | + evidence_root = relative_name(data["evidence_root"]) |
| 212 | + checked_path(root, evidence_root, directory=True) |
| 213 | + artifact = data["artifact"] |
| 214 | + iso = checked_path(root, artifact["iso_path"]) |
| 215 | + if iso.stat().st_size != artifact.get("iso_size_bytes") or acceptance.sha256_file(iso) != expected_hash(artifact.get("iso_sha256")): |
| 216 | + raise ValueError("ISO identity differs from prepublication acceptance") |
| 217 | + checked_path(root, artifact["signature_path"]) |
| 218 | + if not re.fullmatch(r"[0-9A-Fa-f]{40}", artifact.get("signing_fingerprint", "")): |
| 219 | + raise ValueError("A full signing fingerprint is required") |
| 220 | + approved_shots = {item["path"]: expected_hash(item["sha256"]) for item in approvals["screenshots"]} |
| 221 | + with tempfile.TemporaryDirectory(prefix=".evidence-snapshot-", dir=output) as temp: |
| 222 | + staging = Path(temp) / "files" |
| 223 | + staging.mkdir() |
| 224 | + snapshot = Snapshot(root, staging) |
| 225 | + # Original statuses are retained; no output digest is inserted here. |
| 226 | + snapshot.add_bytes("acceptance.prepublication.json", json_bytes(data)) |
| 227 | + snapshot.add_bytes("approved-inputs.json", json_bytes(approvals)) |
| 228 | + references: dict[str, str] = {} |
| 229 | + for case in data["cases"]: |
| 230 | + for item in case["evidence"]: |
| 231 | + if not isinstance(item, dict) or item.get("kind") not in acceptance.VALID_KINDS: |
| 232 | + raise ValueError(f"Invalid evidence entry in {case['id']}") |
| 233 | + name, digest = relative_name(item["path"]), expected_hash(item["sha256"]) |
| 234 | + if name in references and references[name] != digest: |
| 235 | + raise ValueError(f"Conflicting referenced hashes: {name}") |
| 236 | + if item["kind"] == "screenshot" or PurePosixPath(name).suffix.lower() in RASTER: |
| 237 | + if approved_shots.get(name) != digest: |
| 238 | + raise ValueError(f"Screenshot has not been explicitly approved at this hash: {name}") |
| 239 | + references[name] = digest |
| 240 | + for name, digest in approved_shots.items(): |
| 241 | + if name in references and references[name] != digest: |
| 242 | + raise ValueError(f"Approved screenshot differs from acceptance: {name}") |
| 243 | + references[name] = digest |
| 244 | + for name, digest in sorted(references.items()): |
| 245 | + source = evidence_root + "/" + name |
| 246 | + if name in approved_shots: |
| 247 | + width, height = acceptance.png_size(checked_path(root, source)) |
| 248 | + if width < 1280 or height < 720: |
| 249 | + raise ValueError(f"Approved screenshot below 1280x720: {name}") |
| 250 | + snapshot.add_file("evidence/" + name, source, digest) |
| 251 | + # Generated release document hashes are a closed set, not a glob. |
| 252 | + release_dir = "work/release-4.0.0" |
| 253 | + checksum_file = checked_path(root, release_dir + "/" + CHECKSUMS) |
| 254 | + checksums = {} |
| 255 | + for line in checksum_file.read_text().splitlines(): |
| 256 | + match = re.fullmatch(r"([a-fA-F0-9]{64}) ([^/\\]+)", line) |
| 257 | + if not match or match[2] in checksums: |
| 258 | + raise ValueError("Malformed or duplicate generated release checksum") |
| 259 | + checksums[match[2]] = match[1].lower() |
| 260 | + if set(checksums) != set(GENERATED): |
| 261 | + raise ValueError("Generated release checksum list differs from the five expected documents") |
| 262 | + for name in (*GENERATED, CHECKSUMS): |
| 263 | + snapshot.add_file("release/" + name, release_dir + "/" + name, checksums.get(name)) |
| 264 | + facts_path = staging / "release" / f"release-facts-{VERSION}.json" |
| 265 | + if facts_path.stat().st_size > 8 * 1024**2: |
| 266 | + raise ValueError("Release facts exceed 8 MiB") |
| 267 | + facts = json.loads(facts_path.read_text()) |
| 268 | + if facts.get("publicationStatus") != "prepublication" or facts.get("iso") != artifact: |
| 269 | + raise ValueError("Generated release facts are stale or not a prepublication snapshot") |
| 270 | + for item in approvals["documents"]: |
| 271 | + snapshot.add_file("documents/" + item["path"], item["path"], item["sha256"]) |
| 272 | + for name in QA_SOURCES: |
| 273 | + snapshot.add_file("source/" + name, name) |
| 274 | + snapshot.add_file("release/" + PurePosixPath(artifact["signature_path"]).name, artifact["signature_path"]) |
| 275 | + snapshot.add_bytes("bundle-metadata.json", json_bytes({ |
| 276 | + "schema_version": 1, "release": VERSION, "publication_status": "prepublication", |
| 277 | + "iso_sha256": artifact["iso_sha256"], "iso_size_bytes": artifact["iso_size_bytes"], |
| 278 | + "snapshot_rule": "EVIDENCE-01 remains pending; record bundle hash only in the external acceptance manifest after inspection.", |
| 279 | + "scope": "Explicit acceptance evidence, approved captures/documents, generated release docs, signature and named QA sources. ISO bytes and unrelated work files excluded.", |
| 280 | + "source_acceptance_sha256": hashlib.sha256(manifest_bytes).hexdigest(), |
| 281 | + "member_hashes": dict(sorted(snapshot.entries.items())), |
| 282 | + })) |
| 283 | + sums = "".join(f"{entry['sha256']} {PREFIX}/{name}\n" for name, entry in sorted(snapshot.entries.items())) |
| 284 | + snapshot.add_bytes("SHA256SUMS", sums.encode()) |
| 285 | + archive_path = Path(temp) / BUNDLE |
| 286 | + with archive_path.open("wb") as raw, gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as compressed: |
| 287 | + with tarfile.open(fileobj=compressed, mode="w", format=tarfile.PAX_FORMAT) as tar: |
| 288 | + for name, entry in sorted(snapshot.entries.items()): |
| 289 | + info = tarfile.TarInfo(PREFIX + "/" + name) |
| 290 | + info.size, info.mode, info.mtime = entry["size"], 0o644, 0 |
| 291 | + info.uid = info.gid = 0 |
| 292 | + info.uname = info.gname = "" |
| 293 | + with (staging / name).open("rb") as source: |
| 294 | + tar.addfile(info, source) |
| 295 | + contents = "".join(f"{entry['sha256']} {PREFIX}/{name}\n" for name, entry in sorted(snapshot.entries.items())) |
| 296 | + # Validate actual archive bytes before exposing the final output names. |
| 297 | + with tarfile.open(archive_path, "r:gz") as tar: |
| 298 | + if tar.getnames() != [PREFIX + "/" + name for name in sorted(snapshot.entries)]: |
| 299 | + raise ValueError("Archive member list differs from snapshot") |
| 300 | + for member in tar: |
| 301 | + if not member.isfile(): |
| 302 | + raise ValueError("Archive contains a non-file member") |
| 303 | + digest = hashlib.file_digest(tar.extractfile(member), "sha256").hexdigest() |
| 304 | + if digest != snapshot.entries[member.name[len(PREFIX) + 1:]]["sha256"]: |
| 305 | + raise ValueError("Archive member hash/type mismatch") |
| 306 | + if manifest_path.read_bytes() != manifest_bytes: |
| 307 | + raise ValueError("Acceptance manifest changed during packaging") |
| 308 | + (Path(temp) / CONTENTS).write_text(contents) |
| 309 | + result = {"bundle": str(output / BUNDLE), "sha256": acceptance.sha256_file(archive_path), |
| 310 | + "contents": str(output / CONTENTS), "members": len(snapshot.entries), |
| 311 | + "uncompressed_bytes": snapshot.total, "acceptance_modified": False} |
| 312 | + os.replace(Path(temp) / CONTENTS, output / CONTENTS) |
| 313 | + os.replace(archive_path, output / BUNDLE) |
| 314 | + return result |
| 315 | + |
| 316 | + |
| 317 | +def main() -> int: |
| 318 | + parser = argparse.ArgumentParser(description=__doc__) |
| 319 | + parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[1]) |
| 320 | + parser.add_argument("--output-dir", default="work/release-4.0.0", help="Existing repository-relative output directory") |
| 321 | + parser.add_argument("--approved-inputs", required=True, help="Repository-relative explicit screenshot/document approval JSON") |
| 322 | + args = parser.parse_args() |
| 323 | + try: |
| 324 | + print(json.dumps(package(args.root, args.output_dir, args.approved_inputs), indent=2)) |
| 325 | + return 0 |
| 326 | + except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError) as exc: |
| 327 | + print(f"EVIDENCE_BUNDLE_FAILED: {exc}", file=sys.stderr) |
| 328 | + return 1 |
| 329 | + |
| 330 | + |
| 331 | +if __name__ == "__main__": |
| 332 | + raise SystemExit(main()) |
0 commit comments