From 637d285ecb5f575331f3572cac121e5ed76758d9 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Fri, 10 Jul 2026 03:39:44 +0200 Subject: [PATCH] Add Leo production preflight collector --- ...-leo-authorized-apply-runbook-current.json | 13 +- ...ng-leo-authorized-apply-runbook-current.md | 11 +- ...ld_working_leo_authorized_apply_runbook.py | 29 ++ ...ollect_working_leo_production_preflight.py | 270 ++++++++++++++++++ ...ld_working_leo_authorized_apply_runbook.py | 8 + ...ollect_working_leo_production_preflight.py | 102 +++++++ 6 files changed, 431 insertions(+), 2 deletions(-) create mode 100644 scripts/collect_working_leo_production_preflight.py create mode 100644 tests/test_collect_working_leo_production_preflight.py diff --git a/docs/reports/leo-working-state-20260709/working-leo-authorized-apply-runbook-current.json b/docs/reports/leo-working-state-20260709/working-leo-authorized-apply-runbook-current.json index 0b8b473..f76ee60 100644 --- a/docs/reports/leo-working-state-20260709/working-leo-authorized-apply-runbook-current.json +++ b/docs/reports/leo-working-state-20260709/working-leo-authorized-apply-runbook-current.json @@ -5,6 +5,7 @@ "all_phase_files_non_empty": true, "apply_order_matches_manifest": true, "post_apply_regression_commands_present": true, + "pre_apply_validation_commands_present": true, "production_apply_receipt_schema_present": true, "readiness_passes": true, "readiness_says_not_authorized": true, @@ -26,7 +27,7 @@ "rio_strategy_nodes": 2, "sources": 28 }, - "generated_at_utc": "2026-07-10T01:25:52.995962+00:00", + "generated_at_utc": "2026-07-10T01:39:08.583122+00:00", "mode": "operator_runbook_generation_only", "operator_environment": { "operator_commands_run_sql_if_executed": true, @@ -333,6 +334,16 @@ "Run the post-apply regression command bundle and retain every artifact.", "Record production apply receipt, row counts, service MainPID/NRestarts, and cleanup readback." ], + "pre_apply_validation_commands": [ + { + "applies_production_packet": false, + "command_template": ".venv/bin/python scripts/collect_working_leo_production_preflight.py --execute --out docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.json --markdown-out docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.md", + "description": "Collect a production baseline for every preflight SQL file through a read-only transaction wrapper. During an authorized apply window, still run each phase preflight again immediately before its phase.", + "expected_artifact": "docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.md", + "id": "readonly_production_preflight_collector", + "mutates_production_db": false + } + ], "production_apply_authorization_present": false, "production_apply_executed": false, "production_apply_receipt_required_fields": [ diff --git a/docs/reports/leo-working-state-20260709/working-leo-authorized-apply-runbook-current.md b/docs/reports/leo-working-state-20260709/working-leo-authorized-apply-runbook-current.md index cf3703a..141f6a5 100644 --- a/docs/reports/leo-working-state-20260709/working-leo-authorized-apply-runbook-current.md +++ b/docs/reports/leo-working-state-20260709/working-leo-authorized-apply-runbook-current.md @@ -1,6 +1,6 @@ # Working Leo Authorized Apply Runbook -Generated UTC: `2026-07-10T01:25:52.995962+00:00` +Generated UTC: `2026-07-10T01:39:08.583122+00:00` Mode: `operator_runbook_generation_only` Status: `ready_not_executed` Ready for authorized operator: `True` @@ -22,6 +22,7 @@ Do not run apply commands unless the operator explicitly authorizes production D - `all_phase_files_non_empty`: `True` - `all_apply_and_rollback_sql_have_transactions`: `True` - `post_apply_regression_commands_present`: `True` +- `pre_apply_validation_commands_present`: `True` - `production_apply_receipt_schema_present`: `True` ## Preflight Order @@ -91,6 +92,14 @@ Do not run apply commands unless the operator explicitly authorizes production D - `rio_strategy_nodes`: `2` - `sources`: `28` +## Pre-Apply Validation Commands + +- `readonly_production_preflight_collector`: Collect a production baseline for every preflight SQL file through a read-only transaction wrapper. During an authorized apply window, still run each phase preflight again immediately before its phase. + - command: `.venv/bin/python scripts/collect_working_leo_production_preflight.py --execute --out docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.json --markdown-out docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.md` + - artifact: `docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.md` + - mutates production DB: `False` + - applies production packet: `False` + ## Post-Apply Required Actions - Run every postflight SQL file and retain output. diff --git a/scripts/build_working_leo_authorized_apply_runbook.py b/scripts/build_working_leo_authorized_apply_runbook.py index 9cc6e26..5b844d5 100644 --- a/scripts/build_working_leo_authorized_apply_runbook.py +++ b/scripts/build_working_leo_authorized_apply_runbook.py @@ -133,6 +133,24 @@ PRODUCTION_APPLY_RECEIPT_REQUIRED_FIELDS: list[str] = [ "known_residual_risks", ] +PRE_APPLY_VALIDATION_COMMANDS: list[dict[str, Any]] = [ + { + "id": "readonly_production_preflight_collector", + "description": ( + "Collect a production baseline for every preflight SQL file through a read-only transaction wrapper. " + "During an authorized apply window, still run each phase preflight again immediately before its phase." + ), + "command_template": ( + ".venv/bin/python scripts/collect_working_leo_production_preflight.py --execute " + "--out docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.json " + "--markdown-out docs/reports/leo-working-state-20260709/working-leo-production-preflight-current.md" + ), + "expected_artifact": str(REPO_REPORT_DIR / "working-leo-production-preflight-current.md"), + "mutates_production_db": False, + "applies_production_packet": False, + }, +] + def _sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() @@ -194,6 +212,9 @@ def build_runbook(report_dir: Path = REPO_REPORT_DIR) -> dict[str, Any]: "post_apply_regression_commands_present": all( item.get("command_template") and item.get("expected_artifact") for item in POST_APPLY_REGRESSION_COMMANDS ), + "pre_apply_validation_commands_present": all( + item.get("command_template") and item.get("expected_artifact") for item in PRE_APPLY_VALIDATION_COMMANDS + ), "production_apply_receipt_schema_present": all(PRODUCTION_APPLY_RECEIPT_REQUIRED_FIELDS), } ready = all(checks.values()) @@ -228,6 +249,7 @@ def build_runbook(report_dir: Path = REPO_REPORT_DIR) -> dict[str, Any]: "Run the post-apply regression command bundle and retain every artifact.", "Record production apply receipt, row counts, service MainPID/NRestarts, and cleanup readback.", ], + "pre_apply_validation_commands": PRE_APPLY_VALIDATION_COMMANDS, "post_apply_regression_commands": POST_APPLY_REGRESSION_COMMANDS, "production_apply_receipt_required_fields": PRODUCTION_APPLY_RECEIPT_REQUIRED_FIELDS, "claim_ceiling": ( @@ -268,6 +290,13 @@ def write_markdown(path: Path, runbook: dict[str, Any]) -> None: lines.append(f" - `{entry['command_template']}`") lines.extend(["", "## Expected Counts After Apply", ""]) lines.extend(f"- `{key}`: `{value}`" for key, value in sorted(runbook["expected_after_apply_counts"].items())) + lines.extend(["", "## Pre-Apply Validation Commands", ""]) + for item in runbook["pre_apply_validation_commands"]: + lines.append(f"- `{item['id']}`: {item['description']}") + lines.append(f" - command: `{item['command_template']}`") + lines.append(f" - artifact: `{item['expected_artifact']}`") + lines.append(f" - mutates production DB: `{item['mutates_production_db']}`") + lines.append(f" - applies production packet: `{item['applies_production_packet']}`") lines.extend(["", "## Post-Apply Required Actions", ""]) lines.extend(f"- {item}" for item in runbook["post_apply_required_actions"]) lines.extend(["", "## Post-Apply Regression Commands", ""]) diff --git a/scripts/collect_working_leo_production_preflight.py b/scripts/collect_working_leo_production_preflight.py new file mode 100644 index 0000000..9ca59fc --- /dev/null +++ b/scripts/collect_working_leo_production_preflight.py @@ -0,0 +1,270 @@ +#!/usr/bin/env python3 +"""Collect live read-only production preflight output for the Working Leo packet. + +This tool runs only the preflight SQL files from the authorized apply runbook. +It does not run apply, postflight, or rollback SQL. Before any SQL is sent to a +database, each file is statically checked for mutating SQL and then wrapped in a +Postgres read-only transaction. The intended use is to attach fresh row-level +preflight evidence to an operator receipt without crossing the production apply +authorization boundary. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import subprocess +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import build_working_leo_authorized_apply_runbook as runbook + +REPO_REPORT_DIR = Path("docs/reports/leo-working-state-20260709") +DEFAULT_OUT = REPO_REPORT_DIR / "working-leo-production-preflight-current.json" +DEFAULT_MARKDOWN_OUT = REPO_REPORT_DIR / "working-leo-production-preflight-current.md" +DEFAULT_SSH_TARGET = "root@77.42.65.182" +DEFAULT_SSH_KEY = Path.home() / ".ssh/livingip_hetzner_20260604_ed25519" + +FORBIDDEN_SQL_TOKENS = { + "alter", + "analyze", + "begin", + "call", + "cluster", + "commit", + "copy", + "create", + "delete", + "drop", + "execute", + "grant", + "insert", + "merge", + "refresh", + "reindex", + "revoke", + "rollback", + "truncate", + "update", + "vacuum", +} + + +def _sha256_text(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _strip_sql_for_token_scan(sql: str) -> str: + without_block_comments = re.sub(r"/\*.*?\*/", " ", sql, flags=re.S) + without_line_comments = re.sub(r"--.*?$", " ", without_block_comments, flags=re.M) + without_strings = re.sub(r"'(?:''|[^'])*'", "''", without_line_comments) + without_meta = "\n".join( + line for line in without_strings.splitlines() if not line.lstrip().startswith("\\") + ) + return without_meta + + +def forbidden_tokens(sql: str) -> list[str]: + scan = _strip_sql_for_token_scan(sql) + tokens = {match.group(0).lower() for match in re.finditer(r"\b[a-z_]+\b", scan, flags=re.I)} + return sorted(tokens & FORBIDDEN_SQL_TOKENS) + + +def readonly_sql(sql: str) -> str: + body_lines = [ + line for line in sql.splitlines() if line.strip().lower() != r"\set on_error_stop on" + ] + body = "\n".join(body_lines).strip() + return f"\\set ON_ERROR_STOP on\nbegin transaction read only;\n{body}\ncommit;\n" + + +def preflight_entries(report_dir: Path = REPO_REPORT_DIR) -> list[dict[str, Any]]: + generated = runbook.build_runbook(report_dir) + return generated["phases"]["preflight_order"] + + +def build_ssh_command(args: argparse.Namespace) -> list[str]: + command = [ + "ssh", + "-i", + str(args.ssh_key), + "-o", + "BatchMode=yes", + "-o", + "StrictHostKeyChecking=accept-new", + "-o", + f"ConnectTimeout={args.connect_timeout}", + args.ssh_target, + ( + f"docker exec -i {args.container} " + f"psql -U {args.db_user} -d {args.db} -v ON_ERROR_STOP=1 -P pager=off" + ), + ] + return command + + +def command_for_report(args: argparse.Namespace) -> str: + return ( + "ssh -i [SSH_KEY] -o BatchMode=yes -o StrictHostKeyChecking=accept-new " + f"-o ConnectTimeout={args.connect_timeout} {args.ssh_target} " + f"'docker exec -i {args.container} psql -U {args.db_user} -d {args.db} " + "-v ON_ERROR_STOP=1 -P pager=off'" + ) + + +def collect_preflight(args: argparse.Namespace) -> dict[str, Any]: + entries = preflight_entries(args.report_dir) + results: list[dict[str, Any]] = [] + mode = "live_vps_readonly_preflight" if args.execute else "validated_not_executed" + command = build_ssh_command(args) + + for index, entry in enumerate(entries, start=1): + path = args.report_dir / entry["file"] + sql = path.read_text(encoding="utf-8") + forbidden = forbidden_tokens(sql) + wrapped = readonly_sql(sql) + result: dict[str, Any] = { + "index": index, + "packet_id": entry["packet_id"], + "preflight_file": entry["file"], + "path": str(path), + "sql_sha256": _sha256_text(sql), + "readonly_sql_sha256": _sha256_text(wrapped), + "forbidden_tokens": forbidden, + "static_readonly_ok": not forbidden, + "executed": False, + "ok": False, + } + if forbidden: + result["error"] = "preflight SQL contains forbidden mutating or transaction token(s)" + results.append(result) + continue + + if args.execute: + proc = subprocess.run( + command, + input=wrapped, + text=True, + capture_output=True, + timeout=args.timeout, + check=False, + ) + result.update( + { + "executed": True, + "returncode": proc.returncode, + "stdout": proc.stdout, + "stderr": proc.stderr, + "ok": proc.returncode == 0, + } + ) + else: + result.update({"ok": True, "stdout": "", "stderr": ""}) + results.append(result) + + status = "pass" if all(item["ok"] for item in results) else "fail" + return { + "generated_at_utc": datetime.now(timezone.utc).isoformat(), + "mode": mode, + "status": status, + "execute_requested": args.execute, + "mutates_production_db": False, + "applies_production_packet": False, + "production_apply_executed": False, + "readonly_transaction_wrapper": True, + "ssh_target": args.ssh_target if args.execute else None, + "command_redacted": command_for_report(args) if args.execute else None, + "preflight_count": len(results), + "passes": sum(1 for item in results if item["ok"]), + "failures": [item for item in results if not item["ok"]], + "results": results, + "claim_ceiling": ( + "This is read-only production preflight evidence. It can prove current row-level readiness or drift before " + "an apply window. It does not authorize production apply and does not prove canonical rows were changed." + ), + } + + +def write_report(path: Path, report: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + +def write_markdown(path: Path, report: dict[str, Any]) -> None: + lines = [ + "# Working Leo Production Preflight", + "", + f"Generated UTC: `{report['generated_at_utc']}`", + f"Mode: `{report['mode']}`", + f"Status: `{report['status']}`", + f"Executed: `{report['execute_requested']}`", + f"Mutates production DB: `{report['mutates_production_db']}`", + f"Applies production packet: `{report['applies_production_packet']}`", + f"Production apply executed: `{report['production_apply_executed']}`", + f"Read-only transaction wrapper: `{report['readonly_transaction_wrapper']}`", + "", + "## Results", + "", + ] + for item in report["results"]: + lines.append( + f"- `{item['packet_id']}` / `{item['preflight_file']}`: " + f"`ok={item['ok']}`, `executed={item['executed']}`, " + f"`forbidden_tokens={item['forbidden_tokens']}`" + ) + if item.get("returncode") is not None: + lines.append(f" - returncode: `{item['returncode']}`") + stdout = (item.get("stdout") or "").strip() + stderr = (item.get("stderr") or "").strip() + if stdout: + lines.extend([" - stdout:", "", "```text", stdout, "```", ""]) + if stderr: + lines.extend([" - stderr:", "", "```text", stderr, "```", ""]) + lines.extend(["", "## Claim Ceiling", "", report["claim_ceiling"], ""]) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("\n".join(lines), encoding="utf-8") + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--report-dir", type=Path, default=REPO_REPORT_DIR) + parser.add_argument("--out", type=Path, default=DEFAULT_OUT) + parser.add_argument("--markdown-out", type=Path, default=DEFAULT_MARKDOWN_OUT) + parser.add_argument("--execute", action="store_true", help="run the preflight SQL against the VPS") + parser.add_argument("--ssh-target", default=DEFAULT_SSH_TARGET) + parser.add_argument("--ssh-key", type=Path, default=DEFAULT_SSH_KEY) + parser.add_argument("--connect-timeout", type=int, default=15) + parser.add_argument("--timeout", type=int, default=120) + parser.add_argument("--container", default="teleo-pg") + parser.add_argument("--db-user", default="postgres") + parser.add_argument("--db", default="teleo") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + report = collect_preflight(args) + write_report(args.out, report) + write_markdown(args.markdown_out, report) + print( + json.dumps( + { + "out": str(args.out), + "markdown_out": str(args.markdown_out), + "status": report["status"], + "execute_requested": report["execute_requested"], + "passes": report["passes"], + "preflight_count": report["preflight_count"], + }, + indent=2, + sort_keys=True, + ) + ) + return 0 if report["status"] == "pass" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_build_working_leo_authorized_apply_runbook.py b/tests/test_build_working_leo_authorized_apply_runbook.py index 40bffe9..1294de2 100644 --- a/tests/test_build_working_leo_authorized_apply_runbook.py +++ b/tests/test_build_working_leo_authorized_apply_runbook.py @@ -32,7 +32,13 @@ def test_build_runbook_is_ready_not_executed(): assert report["operator_environment"]["operator_commands_run_sql_if_executed"] is True assert "does not execute SQL" in report["claim_ceiling"] assert report["checks"]["post_apply_regression_commands_present"] is True + assert report["checks"]["pre_apply_validation_commands_present"] is True assert report["checks"]["production_apply_receipt_schema_present"] is True + assert {item["id"] for item in report["pre_apply_validation_commands"]} == { + "readonly_production_preflight_collector", + } + assert all(item["mutates_production_db"] is False for item in report["pre_apply_validation_commands"]) + assert all(item["applies_production_packet"] is False for item in report["pre_apply_validation_commands"]) assert {item["id"] for item in report["post_apply_regression_commands"]} == { "direct_claim_handler_suite", "direct_claim_strict_score", @@ -65,6 +71,8 @@ def test_write_markdown_includes_authorization_boundary(tmp_path: Path): assert "Production apply authorization present: `False`" in text assert "Do not run apply commands unless" in text assert 'psql "$TELEO_DATABASE_URL"' in text + assert "Pre-Apply Validation Commands" in text + assert "collect_working_leo_production_preflight.py --execute" in text assert "Post-Apply Regression Commands" in text assert "scripts/run_leo_direct_claim_handler_suite.py" in text assert "direct_claim_score_path" in text diff --git a/tests/test_collect_working_leo_production_preflight.py b/tests/test_collect_working_leo_production_preflight.py new file mode 100644 index 0000000..9ba7028 --- /dev/null +++ b/tests/test_collect_working_leo_production_preflight.py @@ -0,0 +1,102 @@ +"""Tests for scripts/collect_working_leo_production_preflight.py.""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT / "scripts")) + +import collect_working_leo_production_preflight as preflight # noqa: E402 + + +def _args(tmp_path: Path, *, execute: bool = False) -> argparse.Namespace: + return argparse.Namespace( + report_dir=REPO_ROOT / preflight.REPO_REPORT_DIR, + out=tmp_path / "preflight.json", + markdown_out=tmp_path / "preflight.md", + execute=execute, + ssh_target="root@example.invalid", + ssh_key=tmp_path / "ssh-key", + connect_timeout=3, + timeout=10, + container="teleo-pg", + db_user="postgres", + db="teleo", + ) + + +def test_current_preflight_files_validate_without_execution(tmp_path: Path): + report = preflight.collect_preflight(_args(tmp_path)) + + assert report["status"] == "pass" + assert report["execute_requested"] is False + assert report["mutates_production_db"] is False + assert report["applies_production_packet"] is False + assert report["production_apply_executed"] is False + assert report["readonly_transaction_wrapper"] is True + assert report["preflight_count"] == 5 + assert report["passes"] == 5 + assert all(item["static_readonly_ok"] for item in report["results"]) + assert all(item["executed"] is False for item in report["results"]) + + +def test_forbidden_tokens_reject_mutating_sql(): + sql = """ + -- update in a comment is ignored + select 'delete from public.claims' as harmless_string; + update public.claims set status = 'applied'; + """ + + assert preflight.forbidden_tokens(sql) == ["update"] + + +def test_execute_mode_wraps_sql_readonly_and_uses_ssh(monkeypatch, tmp_path: Path): + calls = [] + + class Completed: + returncode = 0 + stdout = "ok\n" + stderr = "" + + def fake_run(command, *, input, text, capture_output, timeout, check): + calls.append( + { + "command": command, + "input": input, + "text": text, + "capture_output": capture_output, + "timeout": timeout, + "check": check, + } + ) + return Completed() + + monkeypatch.setattr(preflight.subprocess, "run", fake_run) + + report = preflight.collect_preflight(_args(tmp_path, execute=True)) + + assert report["status"] == "pass" + assert report["execute_requested"] is True + assert report["passes"] == 5 + assert len(calls) == 5 + for call in calls: + assert call["command"][0] == "ssh" + assert "docker exec -i teleo-pg psql" in call["command"][-1] + assert call["input"].startswith("\\set ON_ERROR_STOP on\nbegin transaction read only;\n") + assert call["input"].rstrip().endswith("commit;") + + +def test_markdown_preserves_claim_ceiling(tmp_path: Path): + report = preflight.collect_preflight(_args(tmp_path)) + out = tmp_path / "preflight.md" + + preflight.write_markdown(out, report) + + text = out.read_text() + assert "Working Leo Production Preflight" in text + assert "Mutates production DB: `False`" in text + assert "Production apply executed: `False`" in text + assert "does not authorize production apply" in text