diff --git a/docs/operations.md b/docs/operations.md index c27010e..6c71e59 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -40,6 +40,35 @@ pkill -TERM -f trex-core-daemon - `config/.env`는 **라이브 KIS 키 + portfolio_hub 토큰 + 라이브 PostgreSQL DSN**을 담는다 — 커밋 금지, 라이브 DB에 파괴적 작업 금지. - 테스트는 `TREX_BROKER=paper` + 파생 `trex_core_test` DSN. +### 배포/복구/수동 재기동 provenance gate (issue #87) + +배포, 복구, 수동 stop/start/recreate 전에 아래 gate를 실행해 **검증 가능한 release record**를 남긴다. + +```bash +# PRE (대상 인스턴스 1개만 지정) +python scripts/release_provenance_gate.py \ + --actor "" \ + --reason "" \ + --instance "" \ + --command-class "manual-stop" \ + --expected-oci-revision "" \ + --image-digest "" \ + --runtime-file "trex_core/runtime/daemon.py" \ + --runtime-file "trex_core/runtime/cli.py" \ + --capture-command "pending_slots=.venv/bin/trex-core slots --status pending --json" \ + --capture-command "reconciliation_slots=.venv/bin/trex-core slots --status all --json" \ + --capture-command "queue_consumers=rabbitmqctl list_consumers" +``` + +- gate는 `git status --porcelain` clean 여부를 검사한다. dirty checkout이면 종료 코드 `2`로 중단하고 변경 파일을 `artifacts/validation/release_provenance_isolation/`으로 복사해 격리한다. +- `checkout.head_revision`과 `--expected-oci-revision` 일치 여부를 검사해 audit checkout HEAD가 실행 이미지 OCI revision과 같은지 검증한다. +- `--image-digest`와 `--runtime-file` 검증 결과(sha256 + revision 일치 여부)가 JSON record에 저장된다. +- actor/timestamp/instance/command-class/reason이 JSON record에 함께 저장되어 stop/start/recreate 감사 추적이 가능하다. +- PRE/POST 모두 실행해 재기동 전후의 `pending_slots`, `reconciliation_slots`, `queue_consumers`, broker-side-effect 로그(`order_submit/fill/reject/cancel`)를 자동 캡처한다. +- 이 스크립트는 read-only capture 전용이다. 주문 제출·취소, DB 복구, 자동 sync/verify live 실행을 암묵적으로 수행하지 않는다. + +재기동 후에는 동일 명령(예: `--command-class "manual-start"`)으로 POST record를 추가 저장한다. + --- ## 2. 운영자 CLI diff --git a/scripts/release_provenance_gate.py b/scripts/release_provenance_gate.py new file mode 100644 index 0000000..39a86eb --- /dev/null +++ b/scripts/release_provenance_gate.py @@ -0,0 +1,88 @@ +"""Collect release/restart provenance and enforce audit-checkout clean gate.""" + +from __future__ import annotations + +import argparse +import json +from datetime import UTC, datetime +from pathlib import Path + +from trex_core.adapters.persistence.ops_checks import ( + collect_release_provenance, + isolate_dirty_checkout_files, +) + +_DEFAULT_OUTPUT_DIR = Path("artifacts/validation/release_provenance") +_DEFAULT_ISOLATE_DIR = Path("artifacts/validation/release_provenance_isolation") + + +def _record_path(output_dir: Path, *, instance: str, command_class: str) -> Path: + stamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%SZ") + return output_dir / f"{stamp}--{instance}--{command_class}.json" + + +def _parse_capture_command(raw: str) -> tuple[str, list[str]]: + if "=" not in raw: + raise ValueError("capture command must be KEY=command") + key, command = raw.split("=", 1) + key = key.strip() + parts = [x for x in command.strip().split(" ") if x] + if not key or not parts: + raise ValueError("capture command must be KEY=command") + return key, parts + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--actor", required=True) + parser.add_argument("--reason", required=True) + parser.add_argument("--instance", required=True) + parser.add_argument("--command-class", required=True) + parser.add_argument("--checkout-dir", default=".") + parser.add_argument("--expected-oci-revision") + parser.add_argument("--image-digest") + parser.add_argument("--runtime-file", action="append", default=[]) + parser.add_argument("--broker-log-path", default="/tmp/trex-core.log") + parser.add_argument("--capture-command", action="append", default=[]) + parser.add_argument("--output-dir", default=str(_DEFAULT_OUTPUT_DIR)) + parser.add_argument("--isolate-dir", default=str(_DEFAULT_ISOLATE_DIR)) + parser.add_argument("--allow-dirty", action="store_true") + args = parser.parse_args(argv) + + capture_commands = {} + for raw in args.capture_command: + key, command = _parse_capture_command(raw) + capture_commands[key] = command + + report = collect_release_provenance( + actor=args.actor, + reason=args.reason, + instance=args.instance, + command_class=args.command_class, + checkout_dir=args.checkout_dir, + expected_oci_revision=args.expected_oci_revision, + image_digest=args.image_digest, + runtime_files=args.runtime_file, + capture_commands=capture_commands, + broker_log_path=args.broker_log_path, + ) + + output_dir = Path(args.output_dir).resolve() + output_dir.mkdir(parents=True, exist_ok=True) + output_path = _record_path(output_dir, instance=args.instance, command_class=args.command_class) + + if report["checkout"]["dirty_paths"] and not args.allow_dirty: + report["dirty_checkout_isolation"] = isolate_dirty_checkout_files( + checkout_dir=args.checkout_dir, + isolate_dir=args.isolate_dir, + ) + + output_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + print(str(output_path)) + if report["checkout"]["dirty_paths"] and not args.allow_dirty: + return 2 + return 0 if report["gate"]["passed"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/adapters/persistence/test_ops_checks.py b/tests/adapters/persistence/test_ops_checks.py index 047dda8..76eafb7 100644 --- a/tests/adapters/persistence/test_ops_checks.py +++ b/tests/adapters/persistence/test_ops_checks.py @@ -7,13 +7,19 @@ find_ledger_gap_slots detects only the gaps. """ from __future__ import annotations +import subprocess import uuid -from datetime import date, datetime, timezone +from datetime import UTC, date, datetime +from pathlib import Path -from trex_core.adapters.persistence.ops_checks import find_ledger_gap_slots +from trex_core.adapters.persistence.ops_checks import ( + collect_release_provenance, + find_ledger_gap_slots, + isolate_dirty_checkout_files, +) _DAY = date(2026, 6, 10) -_TS = datetime(2026, 6, 10, 6, 0, tzinfo=timezone.utc) +_TS = datetime(2026, 6, 10, 6, 0, tzinfo=UTC) def _insert_slot(conn, *, status, portfolio_id, symbol="005930"): @@ -79,10 +85,78 @@ def test_no_gaps_when_all_recorded(pg_clean): def test_date_scope_filters_by_sell_fill_day(pg_clean): import psycopg dsn = pg_clean - other_day_ts = datetime(2026, 6, 9, 6, 0, tzinfo=timezone.utc) + other_day_ts = datetime(2026, 6, 9, 6, 0, tzinfo=UTC) with psycopg.connect(dsn, autocommit=True) as conn: sid = _insert_slot(conn, status="inactive", portfolio_id="PF-OLD") _insert_sell_fill(conn, sid, ts=other_day_ts) assert find_ledger_gap_slots(dsn, trading_date=_DAY) == [] gaps = find_ledger_gap_slots(dsn, trading_date=date(2026, 6, 9)) assert {str(g["slot_id"]) for g in gaps} == {sid} + + +def _init_git_repo(tmp_path: Path) -> tuple[Path, str]: + repo = tmp_path / "repo" + repo.mkdir() + subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True) + subprocess.run(["git", "config", "user.name", "test"], cwd=repo, check=True, capture_output=True) + subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=repo, check=True, capture_output=True) + (repo / "tracked.txt").write_text("v1\n", encoding="utf-8") + subprocess.run(["git", "add", "tracked.txt"], cwd=repo, check=True, capture_output=True) + subprocess.run(["git", "commit", "-m", "init"], cwd=repo, check=True, capture_output=True) + head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=repo, text=True).strip() + return repo, head + + +def test_collect_release_provenance_passes_on_clean_checkout(tmp_path): + repo, head = _init_git_repo(tmp_path) + report = collect_release_provenance( + actor="operator-a", + reason="deploy", + instance="inst-a", + command_class="manual-start", + checkout_dir=repo, + expected_oci_revision=head, + image_digest="sha256:" + ("a" * 64), + runtime_files=["tracked.txt"], + capture_commands={}, + broker_log_path=None, + ) + assert report["checkout"]["git_clean"] is True + assert report["artifact_identity"]["oci_revision_matches_head"] is True + assert report["runtime_file_verification"][0]["matches_expected_revision"] is True + assert report["gate"]["passed"] is True + + +def test_collect_release_provenance_blocks_on_revision_or_runtime_mismatch(tmp_path): + repo, old_head = _init_git_repo(tmp_path) + (repo / "tracked.txt").write_text("v2\n", encoding="utf-8") + subprocess.run(["git", "add", "tracked.txt"], cwd=repo, check=True, capture_output=True) + subprocess.run(["git", "commit", "-m", "v2"], cwd=repo, check=True, capture_output=True) + report = collect_release_provenance( + actor="operator-a", + reason="deploy", + instance="inst-a", + command_class="manual-start", + checkout_dir=repo, + expected_oci_revision=old_head, + runtime_files=["tracked.txt"], + capture_commands={}, + broker_log_path=None, + ) + assert report["checkout"]["git_clean"] is True + assert report["artifact_identity"]["oci_revision_matches_head"] is False + assert report["runtime_file_verification"][0]["matches_expected_revision"] is False + assert report["gate"]["passed"] is False + + +def test_isolate_dirty_checkout_files_copies_dirty_and_untracked(tmp_path): + repo, _ = _init_git_repo(tmp_path) + (repo / "tracked.txt").write_text("dirty\n", encoding="utf-8") + (repo / "new.txt").write_text("new\n", encoding="utf-8") + isolated = tmp_path / "isolated" + manifest = isolate_dirty_checkout_files(checkout_dir=repo, isolate_dir=isolated) + assert set(manifest["dirty_paths"]) == {"tracked.txt", "new.txt"} + assert set(manifest["copied_paths"]) == {"tracked.txt", "new.txt"} + assert (isolated / "tracked.txt").read_text(encoding="utf-8") == "dirty\n" + assert (isolated / "new.txt").read_text(encoding="utf-8") == "new\n" + assert (isolated / "manifest.json").is_file() diff --git a/trex_core/adapters/persistence/ops_checks.py b/trex_core/adapters/persistence/ops_checks.py index a34888d..484facb 100644 --- a/trex_core/adapters/persistence/ops_checks.py +++ b/trex_core/adapters/persistence/ops_checks.py @@ -11,8 +11,13 @@ change; the transition log is durable so this is a permanent detector. """ from __future__ import annotations -from datetime import date -from typing import Any, Optional +import hashlib +import json +import shutil +import subprocess +from datetime import UTC, date, datetime +from pathlib import Path +from typing import Any from trex_core.adapters.persistence.postgresql import _require_psycopg @@ -37,7 +42,7 @@ _COLS = ("slot_id", "symbol", "portfolio_id", "broker_order_id", "sold_at") def find_ledger_gap_slots( dsn: str, *, - trading_date: Optional[date] = None, + trading_date: date | None = None, psycopg_module: Any = None, ) -> list[dict]: """Return sell fills with no corresponding ledger_outbox row (FR-5). @@ -45,7 +50,192 @@ def find_ledger_gap_slots( Pass `trading_date` to scope to one (UTC) day of sell fills; omit for all. """ psycopg = _require_psycopg(psycopg_module) - with psycopg.connect(dsn) as conn: - with conn.cursor() as cur: - cur.execute(_LEDGER_GAP_SQL, {"td": trading_date}) - return [dict(zip(_COLS, row)) for row in cur.fetchall()] + with psycopg.connect(dsn) as conn, conn.cursor() as cur: + cur.execute(_LEDGER_GAP_SQL, {"td": trading_date}) + return [dict(zip(_COLS, row)) for row in cur.fetchall()] + + +def _run_git(checkout_dir: Path, *args: str) -> str: + return subprocess.check_output( + ["git", *args], + cwd=str(checkout_dir), + text=True, + ).strip() + + +def _porcelain_paths(status_output: str) -> list[str]: + paths: list[str] = [] + for raw in status_output.splitlines(): + line = raw.rstrip() + if len(line) < 3: + continue + if line.startswith("?? ") or (len(line) >= 4 and line[2] == " "): + body = line[3:] + elif len(line) >= 3 and line[1] == " ": + body = line[2:] + else: + body = line[3:] if len(line) > 3 else "" + if " -> " in body: + body = body.split(" -> ", 1)[1] + body = body.strip() + if body: + paths.append(body) + return paths + + +def _git_blob_bytes(checkout_dir: Path, revision: str, relpath: str) -> bytes | None: + try: + return subprocess.check_output( + ["git", "show", f"{revision}:{relpath}"], + cwd=str(checkout_dir), + ) + except subprocess.CalledProcessError: + return None + + +def _sha256_bytes(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() + + +def _capture_command(command: list[str], checkout_dir: Path) -> dict: + try: + done = subprocess.run( + command, + cwd=str(checkout_dir), + text=True, + capture_output=True, + check=False, + timeout=20, + ) + return {"command": command, "returncode": done.returncode, "stdout": done.stdout, "stderr": done.stderr} + except (OSError, subprocess.SubprocessError) as exc: # pragma: no cover + return {"command": command, "returncode": None, "stdout": "", "stderr": str(exc)} + + +def _capture_broker_side_effect_logs(log_path: Path) -> dict: + if not log_path.exists(): + return {"path": str(log_path), "exists": False, "matches": []} + matches: list[str] = [] + for line in log_path.read_text(encoding="utf-8", errors="replace").splitlines(): + lowered = line.lower() + if ( + "event=order_submit" in lowered + or "event=order_fill" in lowered + or "event=order_reject" in lowered + or "cancel" in lowered + ): + matches.append(line) + return {"path": str(log_path), "exists": True, "matches": matches[-200:]} + + +def collect_release_provenance( + *, + actor: str, + reason: str, + instance: str, + command_class: str, + checkout_dir: str | Path = ".", + expected_oci_revision: str | None = None, + image_digest: str | None = None, + runtime_files: list[str] | None = None, + capture_commands: dict[str, list[str]] | None = None, + broker_log_path: str | Path | None = "/tmp/trex-core.log", +) -> dict: + """Collect immutable release/restart provenance with fail-closed gates.""" + checkout = Path(checkout_dir).resolve() + head = _run_git(checkout, "rev-parse", "HEAD") + status = _run_git(checkout, "status", "--porcelain") + dirty_paths = _porcelain_paths(status) + git_clean = len(dirty_paths) == 0 + ref_revision = expected_oci_revision or head + runtime_reports = [] + for relpath in runtime_files or []: + abs_path = (checkout / relpath).resolve() + exists = abs_path.is_file() + current_bytes = abs_path.read_bytes() if exists else b"" + current_sha = _sha256_bytes(current_bytes) if exists else None + expected_bytes = _git_blob_bytes(checkout, ref_revision, relpath) + expected_sha = _sha256_bytes(expected_bytes) if expected_bytes is not None else None + runtime_reports.append( + { + "path": relpath, + "exists": exists, + "current_sha256": current_sha, + "expected_revision": ref_revision, + "expected_sha256": expected_sha, + "matches_expected_revision": bool(exists and expected_bytes is not None and current_bytes == expected_bytes), + } + ) + command_outputs = {} + for key, cmd in (capture_commands or {}).items(): + command_outputs[key] = _capture_command(cmd, checkout) + oci_revision_matches_head = expected_oci_revision is None or expected_oci_revision == head + runtime_ok = all(item["matches_expected_revision"] for item in runtime_reports) + gate_passed = git_clean and oci_revision_matches_head and runtime_ok + payload = { + "timestamp_utc": datetime.now(tz=UTC).isoformat(), + "actor": actor, + "reason": reason, + "instance": instance, + "command_class": command_class, + "checkout": { + "path": str(checkout), + "head_revision": head, + "git_status_porcelain": status.splitlines(), + "git_clean": git_clean, + "dirty_paths": dirty_paths, + }, + "artifact_identity": { + "expected_oci_revision": expected_oci_revision, + "oci_revision_matches_head": oci_revision_matches_head, + "image_digest": image_digest, + }, + "runtime_file_verification": runtime_reports, + "state_capture": { + "command_outputs": command_outputs, + "broker_side_effect_logs": ( + _capture_broker_side_effect_logs(Path(broker_log_path)) + if broker_log_path + else None + ), + }, + "gate": { + "passed": gate_passed, + "deployment_blocked": not gate_passed, + }, + } + return payload + + +def isolate_dirty_checkout_files( + *, + checkout_dir: str | Path, + isolate_dir: str | Path, +) -> dict: + """Copy dirty/untracked checkout files to an isolation directory.""" + checkout = Path(checkout_dir).resolve() + dst = Path(isolate_dir).resolve() + status = _run_git(checkout, "status", "--porcelain") + dirty_paths = _porcelain_paths(status) + copied: list[str] = [] + for rel in dirty_paths: + src = (checkout / rel).resolve() + if not src.exists() or src.is_dir(): + continue + target = dst / rel + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(src, target) + copied.append(rel) + manifest = { + "timestamp_utc": datetime.now(tz=UTC).isoformat(), + "checkout_dir": str(checkout), + "isolate_dir": str(dst), + "dirty_paths": dirty_paths, + "copied_paths": copied, + } + dst.mkdir(parents=True, exist_ok=True) + (dst / "manifest.json").write_text( + json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + return manifest