| |
| """Standalone verifier for an extracted OSHE Community Preview release.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import base64 |
| import hashlib |
| import json |
| import subprocess |
| import tempfile |
| from pathlib import Path, PurePosixPath |
| from typing import Any |
|
|
|
|
| class VerificationError(ValueError): |
| pass |
|
|
|
|
| TRUSTED_PUBLIC_KEY_SHA256 = "99fdaae10fd2ff341199f05c9ca04416bda5e809a59911d218c9557c88d6d425" |
|
|
|
|
| def sha256_file(path: Path) -> str: |
| digest = hashlib.sha256() |
| with path.open("rb") as handle: |
| for block in iter(lambda: handle.read(1024 * 1024), b""): |
| digest.update(block) |
| return digest.hexdigest() |
|
|
|
|
| def read_json(path: Path, label: str) -> dict[str, Any]: |
| if path.is_symlink() or not path.is_file(): |
| raise VerificationError(f"{label} is missing or unsafe") |
| try: |
| value = json.loads(path.read_text(encoding="utf-8")) |
| except (OSError, UnicodeError, json.JSONDecodeError) as exc: |
| raise VerificationError(f"{label} is invalid JSON") from exc |
| if not isinstance(value, dict): |
| raise VerificationError(f"{label} must be an object") |
| return value |
|
|
|
|
| def bound_path(root: Path, value: Any, label: str) -> Path: |
| if not isinstance(value, str) or not value or "\\" in value: |
| raise VerificationError(f"{label} is not a safe relative path") |
| relative = PurePosixPath(value) |
| if relative.is_absolute() or any(part in {"", ".", ".."} for part in relative.parts): |
| raise VerificationError(f"{label} is not a safe relative path") |
| path = root / relative |
| try: |
| path.resolve(strict=False).relative_to(root.resolve(strict=True)) |
| except (OSError, ValueError) as exc: |
| raise VerificationError(f"{label} escapes the release") from exc |
| if path.is_symlink() or not path.is_file(): |
| raise VerificationError(f"{label} is missing or unsafe") |
| return path |
|
|
|
|
| def check_hash(root: Path, record: Any, label: str) -> Path: |
| if not isinstance(record, dict): |
| raise VerificationError(f"{label} binding is malformed") |
| path = bound_path(root, record.get("path"), label) |
| expected = record.get("sha256") |
| if not isinstance(expected, str) or len(expected) != 64 or any(char not in "0123456789abcdef" for char in expected): |
| raise VerificationError(f"{label} SHA-256 is malformed") |
| if sha256_file(path) != expected: |
| raise VerificationError(f"{label} SHA-256 mismatch") |
| return path |
|
|
|
|
| def canonical_payload(manifest: dict[str, Any]) -> bytes: |
| payload = dict(manifest) |
| payload.pop("signing", None) |
| return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n" |
|
|
|
|
| def verify_signature(root: Path, manifest: dict[str, Any], signature: dict[str, Any], public_key: Path) -> None: |
| if signature.get("signature_version") != "oshe-adapter-signature-v1" or signature.get("algorithm") != "ed25519": |
| raise VerificationError("unsupported release signature") |
| if signature.get("key_id") != "oshe-community-preview-2026q3-ed25519": |
| raise VerificationError("untrusted release key ID") |
| payload = canonical_payload(manifest) |
| if hashlib.sha256(payload).hexdigest() != signature.get("payload_sha256"): |
| raise VerificationError("release signature payload hash mismatch") |
| signing = manifest.get("signing") |
| if not isinstance(signing, dict) or signing.get("signature") != signature.get("signature"): |
| raise VerificationError("manifest and detached signature disagree") |
| try: |
| signature_bytes = base64.b64decode(signature["signature"], validate=True) |
| except (KeyError, TypeError, ValueError) as exc: |
| raise VerificationError("release signature is not valid base64") from exc |
| with tempfile.TemporaryDirectory(prefix="oshe-release-verify-") as temporary: |
| payload_path = Path(temporary) / "payload.json" |
| signature_path = Path(temporary) / "signature.bin" |
| payload_path.write_bytes(payload) |
| signature_path.write_bytes(signature_bytes) |
| result = subprocess.run( |
| [ |
| "openssl", "pkeyutl", "-verify", "-pubin", "-inkey", str(public_key), |
| "-rawin", "-in", str(payload_path), "-sigfile", str(signature_path), |
| ], |
| capture_output=True, |
| text=True, |
| ) |
| if result.returncode != 0: |
| raise VerificationError("release Ed25519 signature verification failed") |
|
|
|
|
| def verify_bundle_index(root: Path, manifest: dict[str, Any]) -> None: |
| installer = manifest.get("installer") |
| if not isinstance(installer, dict): |
| raise VerificationError("installer binding is missing") |
| bundle_manifest = bound_path(root, installer.get("bundle_manifest_path"), "bundle manifest") |
| bundle_index_path = bound_path(root, installer.get("bundle_index_path"), "bundle index") |
| if sha256_file(bundle_manifest) != installer.get("bundle_manifest_sha256"): |
| raise VerificationError("bundle manifest hash mismatch") |
| if sha256_file(bundle_index_path) != installer.get("bundle_index_sha256"): |
| raise VerificationError("bundle index hash mismatch") |
| index = read_json(bundle_index_path, "bundle index") |
| files = index.get("files") |
| if not isinstance(files, list) or not files: |
| raise VerificationError("bundle index has no files") |
| records: list[dict[str, Any]] = [] |
| declared_paths: set[str] = set() |
| for item in files: |
| path = check_hash(root / "bundle", item, f"bundle file {item.get('path') if isinstance(item, dict) else '?'}") |
| relative = path.relative_to(root / "bundle").as_posix() |
| if relative in declared_paths: |
| raise VerificationError(f"bundle index contains a duplicate path: {relative}") |
| declared_paths.add(relative) |
| records.append({"path": relative, "size_bytes": path.stat().st_size, "sha256": item["sha256"]}) |
| if item.get("size_bytes") != path.stat().st_size: |
| raise VerificationError(f"bundle file size mismatch: {item.get('path')}") |
| actual_paths: set[str] = set() |
| bundle_root = root / "bundle" |
| for path in bundle_root.rglob("*"): |
| if path.is_symlink(): |
| raise VerificationError(f"bundle contains a symlink: {path.relative_to(bundle_root).as_posix()}") |
| if path.is_file(): |
| relative = path.relative_to(bundle_root).as_posix() |
| if relative != bundle_index_path.relative_to(bundle_root).as_posix(): |
| actual_paths.add(relative) |
| if actual_paths != declared_paths: |
| missing = sorted(declared_paths - actual_paths) |
| unlisted = sorted(actual_paths - declared_paths) |
| raise VerificationError(f"bundle file inventory mismatch: missing={missing}, unlisted={unlisted}") |
| canonical = json.dumps(sorted(records, key=lambda item: item["path"]), ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode("utf-8") |
| if hashlib.sha256(canonical).hexdigest() != index.get("tree_sha256"): |
| raise VerificationError("bundle tree hash mismatch") |
|
|
|
|
| def verify_release(root: Path) -> dict[str, Any]: |
| root = root.expanduser().resolve(strict=True) |
| manifest = read_json(root / "release-manifest.json", "release manifest") |
| signature = read_json(root / "release-manifest.sig.json", "release detached signature") |
| public_files = manifest.get("public_files") |
| if not isinstance(public_files, list) or not public_files: |
| raise VerificationError("public release files are missing") |
| observed = {record.get("path"): check_hash(root, record, f"public file {record.get('path')}") for record in public_files if isinstance(record, dict)} |
| public_key = observed.get("oshe-community-preview-public.pem") |
| if public_key is None: |
| raise VerificationError("embedded public release key is missing") |
| if sha256_file(public_key) != TRUSTED_PUBLIC_KEY_SHA256: |
| raise VerificationError("embedded public release key does not match the trusted fingerprint") |
| verify_signature(root, manifest, signature, public_key) |
| rights = manifest.get("rights") |
| if not isinstance(rights, dict): |
| raise VerificationError("rights binding is missing") |
| check_hash(root, {"path": rights.get("path"), "sha256": rights.get("sha256")}, "rights contract") |
| evidence = rights.get("evidence") |
| if not isinstance(evidence, list) or not evidence: |
| raise VerificationError("rights evidence is missing") |
| for record in evidence: |
| check_hash(root, record, f"rights evidence {record.get('source_path') if isinstance(record, dict) else '?'}") |
| verify_bundle_index(root, manifest) |
| runtime = manifest.get("runtime_release") |
| if not isinstance(runtime, dict): |
| raise VerificationError("runtime binding is missing") |
| for name in ("archive", "sidecar"): |
| check_hash(root, {"path": runtime.get(f"{name}_path"), "sha256": runtime.get(f"{name}_sha256")}, f"runtime {name}") |
| return { |
| "status": "verified", |
| "release_manifest_sha256": sha256_file(root / "release-manifest.json"), |
| "bundle_manifest_sha256": manifest["installer"]["bundle_manifest_sha256"], |
| "runtime_archive_sha256": runtime["archive_sha256"], |
| } |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("release", nargs="?", type=Path, default=Path(__file__).resolve().parent) |
| args = parser.parse_args() |
| try: |
| result = verify_release(args.release) |
| except (OSError, VerificationError, subprocess.SubprocessError) as exc: |
| print(json.dumps({"status": "BLOCKED", "errors": [str(exc)]}, indent=2, sort_keys=True)) |
| return 1 |
| print(json.dumps(result, indent=2, sort_keys=True)) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
