scripts/post_credential_live_proof.py

#!/usr/bin/env python3
"""Run the post-credential live proof sequence without storing secrets."""

from __future__ import annotations

import argparse
from dataclasses import dataclass
from datetime import datetime, timezone
import json
from pathlib import Path
import re
import shlex
import subprocess
import sys
from typing import Any, Callable, Sequence


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_ENV_FILE = ROOT / ".env.final.local"
DEFAULT_JSON = ROOT / "docs" / "assets" / "post-credential-live-proof-plan.json"
DEFAULT_MD = ROOT / "docs" / "assets" / "post-credential-live-proof-plan.md"
SCHEMA = "proofframe.post_credential_live_proof.v1"
B2_EVIDENCE = ROOT / "docs" / "assets" / "b2-live-proof-evidence.json"
FINAL_EVIDENCE = ROOT / "docs" / "assets" / "final-live-proof-evidence.json"
FORBIDDEN_EVIDENCE_KEYS = re.compile(
    r"(?i)(api[_-]?key|application[_-]?key|authorization|cookie|password|secret|token)"
)
FORBIDDEN_EVIDENCE_VALUES = [
    re.compile(r"(?i)authorization:\s*bearer\s+[A-Za-z0-9._\-]{20,}"),
    re.compile(r"(?i)(api[_-]?key|application[_-]?key|secret|token|cookie)=[^&\s]{8,}"),
    re.compile(r"(?i)x-amz-(credential|security-token|signature)=[^&\s]{8,}"),
    re.compile(r"(?i)gmi-[A-Za-z0-9_\-]{16,}"),
]
REQUIRED_EVIDENCE_VALUES = ("asset_sha256", "manifest_sha256", "asset_storage_key", "manifest_key")
FINAL_ALLOWED_ASSET_PROVIDERS = {
    "genblaze/gmicloud-image",
    "genblaze/openai-image",
    "genblaze/local-image",
}
EXPECTED_EVIDENCE_FIELDS = {
    "b2": {
        "ok": True,
        "storage_backend": "b2",
        "generation_backend": "mock",
        "asset_storage_backend": "b2",
        "asset_provider": "mock",
        "manifest_storage_backend": "b2",
    },
    "final": {
        "ok": True,
        "storage_backend": "b2",
        "generation_backend": "genblaze",
        "asset_storage_backend": "b2",
        "manifest_storage_backend": "b2",
    },
}


@dataclass(frozen=True)
class CommandSpec:
    command_id: str
    label: str
    command: list[str]


def utc_now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def rel(path: Path) -> str:
    try:
        return str(path.resolve().relative_to(ROOT.resolve()))
    except ValueError:
        return str(path)


def display_arg(arg: str) -> str:
    path = Path(arg)
    if path.is_absolute():
        try:
            return str(path.resolve().relative_to(ROOT.resolve()))
        except ValueError:
            if path.name.startswith("python"):
                return "python"
            return path.name
    return arg


def display_command(command: Sequence[str]) -> str:
    return shlex.join(display_arg(arg) for arg in command)


def script_command(script: str, *args: str, python: str = sys.executable) -> list[str]:
    return [python, str(ROOT / "scripts" / script), *args]


def task_done_command(task_id: str, note: str, python: str = sys.executable) -> list[str]:
    return script_command("task.py", "done", task_id, "--note", note, python=python)


def validate_evidence_command(kind: str, evidence_path: Path, python: str = sys.executable) -> list[str]:
    return [
        python,
        str(ROOT / "scripts" / "post_credential_live_proof.py"),
        "--validate-evidence",
        kind,
        "--evidence-path",
        rel(evidence_path),
    ]


def build_commands(
    *,
    env_file: Path = DEFAULT_ENV_FILE,
    update_tasks: bool = False,
    python: str = sys.executable,
) -> list[CommandSpec]:
    env_arg = rel(env_file)
    commands = [
        CommandSpec(
            "credential_handoff",
            "Verify local live credential presence without printing values",
            script_command("live_env_handoff.py", "--env-file", env_arg, "--strict", python=python),
        ),
        CommandSpec(
            "b2_live_proof",
            "Capture Backblaze B2 storage proof with mock generation",
            script_command(
                "run_b2_live_proof.py",
                "--env-file",
                env_arg,
                "--evidence-out",
                "docs/assets/b2-live-proof-evidence.json",
                python=python,
            ),
        ),
        CommandSpec(
            "validate_b2_evidence",
            "Validate sanitized B2 evidence before any T020 task update",
            validate_evidence_command("b2", B2_EVIDENCE, python=python),
        ),
    ]
    if update_tasks:
        commands.append(
            CommandSpec(
                "mark_t020_done",
                "Mark T020 done after sanitized B2 evidence exists",
                task_done_command(
                    "T020",
                    "B2 live proof evidence captured in docs/assets/b2-live-proof-evidence.json.",
                    python=python,
                ),
            )
        )
    commands.append(
        CommandSpec(
            "final_live_proof",
            "Capture final B2 plus Genblaze proof",
            script_command(
                "run_final_live_proof.py",
                "--env-file",
                env_arg,
                "--evidence-out",
                "docs/assets/final-live-proof-evidence.json",
                python=python,
            ),
        )
    )
    commands.append(
        CommandSpec(
            "validate_final_evidence",
            "Validate sanitized final B2 plus Genblaze evidence before any T021 task update",
            validate_evidence_command("final", FINAL_EVIDENCE, python=python),
        )
    )
    if update_tasks:
        commands.append(
            CommandSpec(
                "mark_t021_done",
                "Mark T021 done after sanitized final evidence exists",
                task_done_command(
                    "T021",
                    "Final B2 plus Genblaze live proof evidence captured in docs/assets/final-live-proof-evidence.json.",
                    python=python,
                ),
            )
        )
    commands.extend(
        [
            CommandSpec(
                "live_env_handoff_report",
                "Regenerate live credential handoff report",
                script_command("live_env_handoff.py", "--env-file", env_arg, python=python),
            ),
            CommandSpec("devpost_form_kit", "Regenerate Devpost form kit", script_command("devpost_form_kit.py", python=python)),
            CommandSpec(
                "devpost_submission_checklist",
                "Regenerate Devpost submission checklist",
                script_command("devpost_submission_checklist.py", python=python),
            ),
            CommandSpec("judge_brief", "Regenerate judge brief", script_command("judge_brief.py", python=python)),
            CommandSpec("judge_crosswalk", "Regenerate judge crosswalk", script_command("judge_crosswalk.py", python=python)),
            CommandSpec("demo_storyboard", "Regenerate demo storyboard", script_command("demo_storyboard.py", python=python)),
            CommandSpec("demo_readiness", "Regenerate demo readiness report", script_command("demo_readiness.py", python=python)),
            CommandSpec(
                "recording_assets",
                "Regenerate recording assets report and verify public mock assets",
                script_command("recording_assets.py", "--verify-public", python=python),
            ),
            CommandSpec(
                "award_readiness",
                "Regenerate award readiness scorecard",
                script_command("award_readiness.py", "--min-score", "75", python=python),
            ),
            CommandSpec(
                "final_operator_brief",
                "Regenerate final operator brief",
                script_command("final_operator_brief.py", python=python),
            ),
            CommandSpec(
                "final_launch_plan",
                "Regenerate final launch plan",
                script_command("final_launch_plan.py", python=python),
            ),
            CommandSpec(
                "final_rehearsal",
                "Regenerate final rehearsal checklist",
                script_command("final_rehearsal.py", python=python),
            ),
            CommandSpec(
                "final_submission_control",
                "Regenerate final submission control",
                script_command("final_submission_control.py", python=python),
            ),
            CommandSpec(
                "submission_audit",
                "Regenerate pre-submit audit report",
                script_command("submission_audit.py", python=python),
            ),
            CommandSpec(
                "devpost_submission_preview",
                "Regenerate one-page Devpost submission preview",
                script_command("devpost_submission_preview.py", python=python),
            ),
            CommandSpec("secret_scan", "Run no-value secret scan", script_command("secret_scan.py", python=python)),
            CommandSpec("submission_bundle", "Regenerate submission bundle manifest", script_command("submission_bundle.py", python=python)),
        ]
    )
    return commands


def load_evidence(path: Path) -> tuple[dict[str, Any] | None, str | None]:
    try:
        evidence = json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return None, "Evidence file is missing."
    except json.JSONDecodeError as exc:
        return None, f"Evidence file is invalid JSON: {exc}"
    if not isinstance(evidence, dict):
        return None, "Evidence JSON must be an object."
    return evidence, None


def evidence_safety_findings(value: Any, path: str = "$") -> list[str]:
    findings: list[str] = []
    if isinstance(value, dict):
        for key, item in value.items():
            key_path = f"{path}.{key}"
            if FORBIDDEN_EVIDENCE_KEYS.search(str(key)):
                findings.append(f"{key_path}: forbidden evidence key")
            findings.extend(evidence_safety_findings(item, key_path))
    elif isinstance(value, list):
        for index, item in enumerate(value):
            findings.extend(evidence_safety_findings(item, f"{path}[{index}]"))
    elif isinstance(value, str):
        for pattern in FORBIDDEN_EVIDENCE_VALUES:
            if pattern.search(value):
                findings.append(f"{path}: forbidden evidence value")
                break
    return findings


def validate_evidence(kind: str, path: Path) -> dict[str, Any]:
    evidence, load_error = load_evidence(path)
    findings: list[dict[str, str]] = []
    if load_error:
        findings.append({"field": "file", "detail": load_error})
        evidence = {}

    expected = EXPECTED_EVIDENCE_FIELDS[kind]
    for key, expected_value in expected.items():
        actual = evidence.get(key)
        if actual != expected_value:
            findings.append({"field": key, "detail": f"Expected {expected_value!r}, got {actual!r}."})
    if kind == "final":
        asset_provider = evidence.get("asset_provider")
        if asset_provider not in FINAL_ALLOWED_ASSET_PROVIDERS:
            findings.append(
                {
                    "field": "asset_provider",
                    "detail": (
                        "Expected one of "
                        f"{sorted(FINAL_ALLOWED_ASSET_PROVIDERS)!r}, got {asset_provider!r}."
                    ),
                }
            )
    for key in REQUIRED_EVIDENCE_VALUES:
        if not evidence.get(key):
            findings.append({"field": key, "detail": "Required evidence value is missing."})
    for finding in evidence_safety_findings(evidence):
        findings.append({"field": "secret_safety", "detail": finding})

    return {
        "ok": not findings,
        "kind": kind,
        "path": rel(path),
        "checks": {
            "expected_fields": expected,
            "allowed_final_asset_providers": sorted(FINAL_ALLOWED_ASSET_PROVIDERS),
            "required_values": list(REQUIRED_EVIDENCE_VALUES),
            "secret_safety": "forbidden keys, bearer tokens, signed URL parameters, and GMI-style keys",
        },
        "findings": findings,
    }


def command_record(spec: CommandSpec, status: str, returncode: int | None = None) -> dict[str, Any]:
    return {
        "id": spec.command_id,
        "label": spec.label,
        "status": status,
        "returncode": returncode,
        "command": display_command(spec.command),
        "argv": [display_arg(arg) for arg in spec.command],
    }


def run_sequence(
    commands: Sequence[CommandSpec],
    *,
    execute: bool,
    runner: Callable[..., subprocess.CompletedProcess[Any]] = subprocess.run,
) -> dict[str, Any]:
    records: list[dict[str, Any]] = []
    if not execute:
        return {
            "ok": True,
            "mode": "plan_only",
            "commands": [command_record(spec, "planned") for spec in commands],
            "failed_command": None,
        }

    failed_command: str | None = None
    for index, spec in enumerate(commands):
        completed = runner(spec.command, cwd=ROOT, check=False)
        records.append(command_record(spec, "passed" if completed.returncode == 0 else "failed", completed.returncode))
        if completed.returncode != 0:
            failed_command = spec.command_id
            for skipped in commands[index + 1 :]:
                records.append(command_record(skipped, "skipped"))
            break
    return {
        "ok": failed_command is None,
        "mode": "executed" if failed_command is None else "failed",
        "commands": records,
        "failed_command": failed_command,
    }


def build_report(args: argparse.Namespace, sequence: dict[str, Any]) -> dict[str, Any]:
    next_actions = [
        "If mode is plan_only, rerun with --execute after credentials are entered locally.",
        "If live proof succeeds, record and upload the final public demo video.",
        "After the public video is verified, run the final secret scan, strict audit, Devpost submit, and receipt capture.",
    ]
    if sequence.get("failed_command") == "credential_handoff":
        next_actions.insert(
            0,
            "Enter missing local credentials with python scripts/final_env_wizard.py --output .env.final.local --missing-only --force.",
        )
    return {
        "schema": SCHEMA,
        "created_at": utc_now(),
        "ok": sequence["ok"],
        "mode": sequence["mode"],
        "env_file": rel(args.env_file),
        "update_tasks": bool(args.update_tasks),
        "execute": bool(args.execute),
        "commands": sequence["commands"],
        "failed_command": sequence["failed_command"],
        "next_actions": next_actions,
        "secret_policy": (
            "This report stores command strings, statuses, and artifact paths only. It never stores "
            "Backblaze keys, Genblaze provider keys, Devpost cookies, provider responses, or signed URLs."
        ),
    }


def render_markdown(report: dict[str, Any]) -> str:
    lines = [
        "# ProofFrame Post-Credential Live Proof Plan",
        "",
        f"Mode: `{report['mode']}`",
        f"OK: `{str(report['ok']).lower()}`",
        f"Created: `{report['created_at']}`",
        f"Env file: `{report['env_file']}`",
        f"Update tasks: `{str(report['update_tasks']).lower()}`",
        "",
        report["secret_policy"],
        "",
        "## Commands",
        "",
        "| Status | ID | Command |",
        "| --- | --- | --- |",
    ]
    for command in report["commands"]:
        lines.append(f"| {command['status'].upper()} | `{command['id']}` | `{command['command']}` |")
    lines.extend(["", "## Next Actions", ""])
    lines.extend(f"- {action}" for action in report["next_actions"])
    return "\n".join(lines) + "\n"


def write_outputs(report: dict[str, Any], json_path: Path, markdown_path: Path) -> None:
    json_path.parent.mkdir(parents=True, exist_ok=True)
    markdown_path.parent.mkdir(parents=True, exist_ok=True)
    json_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
    markdown_path.write_text(render_markdown(report), encoding="utf-8")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Plan or run the post-credential ProofFrame live proof sequence."
    )
    parser.add_argument("--env-file", type=Path, default=DEFAULT_ENV_FILE)
    parser.add_argument("--json-out", type=Path, default=DEFAULT_JSON)
    parser.add_argument("--markdown-out", type=Path, default=DEFAULT_MD)
    parser.add_argument("--execute", action="store_true", help="Run the live proof sequence.")
    parser.add_argument(
        "--update-tasks",
        action="store_true",
        help="Mark T020/T021 done after their proof commands pass.",
    )
    parser.add_argument(
        "--validate-evidence",
        choices=sorted(EXPECTED_EVIDENCE_FIELDS),
        help="Internal fail-closed evidence validator used before task updates.",
    )
    parser.add_argument("--evidence-path", type=Path, help="Evidence JSON path for --validate-evidence.")
    return parser


def main() -> None:
    args = build_parser().parse_args()
    if args.validate_evidence:
        evidence_path = args.evidence_path
        if evidence_path is None:
            evidence_path = B2_EVIDENCE if args.validate_evidence == "b2" else FINAL_EVIDENCE
        report = validate_evidence(args.validate_evidence, evidence_path)
        print(json.dumps(report, indent=2))
        raise SystemExit(0 if report["ok"] else 2)

    commands = build_commands(env_file=args.env_file, update_tasks=args.update_tasks)
    sequence = run_sequence(commands, execute=args.execute)
    report = build_report(args, sequence)
    write_outputs(report, args.json_out, args.markdown_out)
    print(
        json.dumps(
            {
                "ok": report["ok"],
                "mode": report["mode"],
                "json": str(args.json_out),
                "markdown": str(args.markdown_out),
                "failed_command": report["failed_command"],
                "commands": len(report["commands"]),
            },
            indent=2,
        )
    )
    if args.execute and not report["ok"]:
        raise SystemExit(2)


if __name__ == "__main__":
    main()