#!/usr/bin/env python3 """Capture a read-only, secret-free GeoIntel release evidence manifest.""" from __future__ import annotations import argparse import hashlib import importlib.metadata import json import os import platform import subprocess import sys import urllib.error import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any, Sequence ROOT = Path(__file__).resolve().parents[1] BACKEND = ROOT / "backend" DEFAULT_HASHED_FILES = ( "AGENTS.md", "backend/pyproject.toml", "backend/alembic.ini", "frontend/package.json", "frontend/package-lock.json", "docker-compose.yml", "docker-compose.unraid.yml", "deploy/unraid/Dockerfile.all-in-one", "docs/RC_SCOPE_FREEZE_BELGIUM_NORTH_SEA.md", "docs/RC_ROADMAP_BELGIUM_NORTH_SEA.md", ) DEPENDENCIES = ( "alembic", "fastapi", "geoalchemy2", "geopandas", "numpy", "pydantic", "pyproj", "rasterio", "shapely", "sqlalchemy", "torch", "ultralytics", "uvicorn", ) def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def run_command( command: Sequence[str], *, cwd: Path = ROOT, timeout_seconds: int = 30, ) -> dict[str, Any]: try: result = subprocess.run( list(command), cwd=cwd, capture_output=True, text=True, check=False, timeout=timeout_seconds, ) except (OSError, subprocess.TimeoutExpired) as exc: return { "ok": False, "exit_code": None, "stdout": "", "stderr": str(exc), } return { "ok": result.returncode == 0, "exit_code": result.returncode, "stdout": result.stdout.strip(), "stderr": result.stderr.strip(), } def git_evidence() -> dict[str, Any]: commit = run_command(("git", "rev-parse", "HEAD")) branch = run_command(("git", "branch", "--show-current")) status = run_command(("git", "status", "--porcelain=v1")) remote = run_command(("git", "remote", "get-url", "origin")) dirty_paths = [ line[3:].strip() for line in status["stdout"].splitlines() if len(line) >= 4 ] return { "commit": commit["stdout"] or None, "branch": branch["stdout"] or None, "origin": remote["stdout"] or None, "dirty": bool(dirty_paths), "dirty_paths": dirty_paths, "commands_ok": all(item["ok"] for item in (commit, branch, status)), } def migration_evidence() -> dict[str, Any]: heads = run_command((sys.executable, "-m", "alembic", "heads"), cwd=BACKEND) head_lines = [ line.strip() for line in heads["stdout"].splitlines() if line.strip() ] return { "command_ok": heads["ok"], "heads": head_lines, "single_head": heads["ok"] and len(head_lines) == 1, "error": heads["stderr"] or None, } def dependency_evidence() -> dict[str, str | None]: versions: dict[str, str | None] = {} for name in DEPENDENCIES: try: versions[name] = importlib.metadata.version(name) except importlib.metadata.PackageNotFoundError: versions[name] = None return versions def file_evidence() -> dict[str, dict[str, Any]]: evidence: dict[str, dict[str, Any]] = {} for relative_path in DEFAULT_HASHED_FILES: path = ROOT / relative_path evidence[relative_path] = { "exists": path.is_file(), "size_bytes": path.stat().st_size if path.is_file() else None, "sha256": sha256_file(path) if path.is_file() else None, } return evidence def storage_evidence(storage_root: Path | None) -> dict[str, Any]: if storage_root is None: return {"requested": False} resolved = storage_root.expanduser().resolve() if not resolved.is_dir(): return { "requested": True, "root": str(resolved), "available": False, } total_size = 0 file_count = 0 errors: list[str] = [] for path in resolved.rglob("*"): if path.is_symlink() or not path.is_file(): continue try: total_size += path.stat().st_size file_count += 1 except OSError as exc: errors.append(f"{path}: {exc}") if len(errors) >= 20: break return { "requested": True, "root": str(resolved), "available": True, "file_count": file_count, "size_bytes": total_size, "scan_complete": not errors, "errors": errors, } def fetch_json(url: str, timeout_seconds: int) -> dict[str, Any]: request = urllib.request.Request( url, headers={"Accept": "application/json", "User-Agent": "GeoIntel-RC-Evidence/1.0"}, ) try: with urllib.request.urlopen(request, timeout=timeout_seconds) as response: body = response.read(2 * 1024 * 1024 + 1) if len(body) > 2 * 1024 * 1024: raise ValueError("response exceeds 2 MiB evidence limit") return { "ok": 200 <= response.status < 300, "status_code": response.status, "payload": json.loads(body.decode("utf-8")), "error": None, } except (urllib.error.URLError, ValueError, json.JSONDecodeError) as exc: return { "ok": False, "status_code": getattr(exc, "code", None), "payload": None, "error": str(exc), } def live_evidence(base_url: str | None, timeout_seconds: int) -> dict[str, Any]: if not base_url: return {"requested": False} normalized = base_url.rstrip("/") return { "requested": True, "base_url": normalized, "health": fetch_json(f"{normalized}/health", timeout_seconds), "capabilities": fetch_json( f"{normalized}/api/v1/system/capabilities", timeout_seconds, ), } def build_manifest(args: argparse.Namespace) -> dict[str, Any]: storage_root = Path(args.storage_root) if args.storage_root else None return { "schema_version": 1, "release_id": args.release_id, "captured_at": utc_now(), "read_only": True, "scope": "Belgium and the Belgian North Sea", "host": { "platform": platform.platform(), "python": platform.python_version(), "hostname": platform.node(), }, "git": git_evidence(), "migrations": migration_evidence(), "dependencies": dependency_evidence(), "files": file_evidence(), "storage": storage_evidence(storage_root), "live": live_evidence(args.live_base_url, args.timeout_seconds), "environment_presence": { "DATABASE_URL": bool(os.environ.get("DATABASE_URL")), "STORAGE_ROOT": bool(os.environ.get("STORAGE_ROOT")), "YOLO_ENABLED": bool(os.environ.get("YOLO_ENABLED")), "YOLO_MODEL_PATH": bool(os.environ.get("YOLO_MODEL_PATH")), "OLLAMA_ENABLED": bool(os.environ.get("OLLAMA_ENABLED")), }, } def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--release-id", default="rc-current") parser.add_argument("--live-base-url") parser.add_argument("--storage-root") parser.add_argument("--timeout-seconds", type=int, default=10) return parser.parse_args(argv) def main(argv: Sequence[str] | None = None) -> int: args = parse_args(argv) if args.timeout_seconds < 1 or args.timeout_seconds > 120: raise SystemExit("--timeout-seconds must be between 1 and 120") manifest = build_manifest(args) output = args.output.expanduser().resolve() output.parent.mkdir(parents=True, exist_ok=True) output.write_text( json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) print(f"Wrote read-only release evidence to {output}") if not manifest["git"]["commands_ok"]: return 2 if not manifest["migrations"]["single_head"]: return 3 return 0 if __name__ == "__main__": raise SystemExit(main())