diff --git a/docs/operations.md b/docs/operations.md index c27010e..2682771 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -40,6 +40,39 @@ 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개만 지정; POST도 같은 operation-id로 별도 실행) +python scripts/release_provenance_gate.py \ + --actor "" \ + --reason "" \ + --instance "" \ + --operation-id "" \ + --phase "pre" \ + --command-class "manual-stop" \ + --expected-oci-revision "<40-character-running-image-oci-revision>" \ + --image-digest "sha256:<64-hex-running-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 --instance --json" \ + --capture-command "reconciliation_slots=.venv/bin/trex-core slots --status all --instance --json" \ + --capture-command "queue_consumers=rabbitmqctl list_consumers" +``` + +- gate는 audit checkout이 detached HEAD인지 확인하고 NUL 구분 `git status --porcelain=v1`로 clean 여부를 검사한다. branch checkout 또는 dirty checkout이면 운영 상태 조회 명령도 실행하지 않고 종료 코드 `2`로 중단하며, dirty 변경 파일은 audit checkout 밖의 `-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/operation-id/phase/command-class/reason이 JSON record에 함께 저장되어 PRE/POST 및 stop/start/recreate 감사 추적이 가능하다. actor/reason에 credential 형태의 key/value가 포함되면 기록 전에 거부한다. +- PRE/POST 모두 실행해 재기동 전후의 `pending_slots`, `reconciliation_slots`, `queue_consumers`, broker-side-effect 로그(`order_submit/fill/reject/cancel`)를 자동 캡처한다. 필수 조회 또는 로그 캡처가 누락·실패하면 gate는 실패한다. +- capture command는 위의 인스턴스 지정 `trex-core slots` 두 형식과 `rabbitmqctl list_consumers`만 허용한다. 주문 제출·취소, DB 복구, 자동 sync/verify live 실행은 거부한다. +- broker-side-effect 로그 입력은 기본 `/tmp/trex-core.log`를 사용하며, 로그 파일 위치가 다르면 `--broker-log-path `로 명시한다. record에는 원문 대신 파일 sha256·스캔 행 수와 event 종류·행 번호·행 sha256만 남긴다. 파일을 정상 스캔했지만 일치 이벤트가 없으면 `captured_empty`, 파일 누락·읽기 실패는 각각 `unavailable`·`error`로 기록하며 후자는 gate를 차단한다. +- record와 격리 디렉터리는 audit checkout 밖에 둔다. 기본 record 위치는 `-release-provenance/`이며, 충돌 없는 파일명을 exclusive-create하고 read-only mode로 전환해 기존 record를 덮어쓰지 않는다. + +재기동 후에는 같은 `--operation-id`, `--phase "post"`, 해당 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 100755 index 0000000..6dcf805 --- /dev/null +++ b/scripts/release_provenance_gate.py @@ -0,0 +1,175 @@ +#!/usr/bin/env python3 +"""Collect release/restart provenance and enforce audit-checkout gates.""" + +from __future__ import annotations + +import argparse +import json +import re +import shlex +import sys +import uuid +from datetime import UTC, datetime +from pathlib import Path + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from trex_core.adapters.persistence.ops_checks import ( + collect_release_provenance, + isolate_dirty_checkout_files, +) + +_COMMAND_CLASSES = ("deploy", "recovery", "manual-stop", "manual-start", "recreate") +_REQUIRED_CAPTURE_KEYS = frozenset( + {"pending_slots", "reconciliation_slots", "queue_consumers"} +) +_SAFE_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}") + + +def _record_path( + output_dir: Path, + *, + operation_id: str, + phase: str, + instance: str, + command_class: str, +) -> Path: + for value in (operation_id, phase, instance, command_class): + if not _SAFE_ID.fullmatch(value): + raise ValueError("record filename components must be safe identifiers") + stamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%S.%fZ") + nonce = uuid.uuid4().hex[:12] + return output_dir / ( + f"{stamp}--{operation_id}--{phase}--{instance}--{command_class}--{nonce}.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 = shlex.split(command.strip()) + if not key or not parts: + raise ValueError("capture command must be KEY=command") + return key, parts + + +def _validate_capture_command(key: str, command: list[str], *, instance: str) -> None: + executable = Path(command[0]).name + args = command[1:] + expected = { + "pending_slots": ["slots", "--status", "pending", "--instance", instance, "--json"], + "reconciliation_slots": ["slots", "--status", "all", "--instance", instance, "--json"], + "queue_consumers": ["list_consumers"], + } + if key not in expected: + raise ValueError(f"unsupported capture key: {key}") + if key == "queue_consumers": + valid = executable == "rabbitmqctl" and args == expected[key] + else: + valid = executable == "trex-core" and args == expected[key] + if not valid: + raise ValueError(f"capture command is not the approved read-only form for {key}") + + +def _outside_checkout(path: Path, checkout: Path, *, label: str) -> Path: + resolved = path.resolve() + try: + resolved.relative_to(checkout.resolve()) + except ValueError: + return resolved + raise ValueError(f"{label} must be outside the audit checkout") + + +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("--operation-id", required=True) + parser.add_argument("--phase", required=True, choices=("pre", "post")) + parser.add_argument("--command-class", required=True, choices=_COMMAND_CLASSES) + parser.add_argument("--checkout-dir", default=".") + parser.add_argument("--expected-oci-revision", required=True) + parser.add_argument("--image-digest", required=True) + parser.add_argument("--runtime-file", action="append", required=True) + parser.add_argument("--broker-log-path", default="/tmp/trex-core.log") + parser.add_argument("--capture-command", action="append", required=True) + parser.add_argument("--output-dir") + parser.add_argument("--isolate-dir") + args = parser.parse_args(argv) + + try: + capture_commands: dict[str, list[str]] = {} + for raw in args.capture_command: + key, command = _parse_capture_command(raw) + if key in capture_commands: + raise ValueError(f"duplicate capture key: {key}") + _validate_capture_command(key, command, instance=args.instance) + capture_commands[key] = command + if set(capture_commands) != _REQUIRED_CAPTURE_KEYS: + missing = sorted(_REQUIRED_CAPTURE_KEYS - set(capture_commands)) + extra = sorted(set(capture_commands) - _REQUIRED_CAPTURE_KEYS) + raise ValueError(f"capture command keys mismatch: missing={missing}, extra={extra}") + + checkout = Path(args.checkout_dir).resolve() + output_default = checkout.parent / f"{checkout.name}-release-provenance" + isolate_default = checkout.parent / f"{checkout.name}-release-provenance-isolation" + output_dir = _outside_checkout( + Path(args.output_dir) if args.output_dir else output_default, + checkout, + label="output directory", + ) + isolate_dir = _outside_checkout( + Path(args.isolate_dir) if args.isolate_dir else isolate_default, + checkout, + label="isolation directory", + ) + + report = collect_release_provenance( + actor=args.actor, + reason=args.reason, + instance=args.instance, + command_class=args.command_class, + operation_id=args.operation_id, + phase=args.phase, + checkout_dir=checkout, + 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, + ) + except (RuntimeError, ValueError) as exc: + parser.error(str(exc)) + + output_dir.mkdir(parents=True, exist_ok=True) + output_path = _record_path( + output_dir, + operation_id=args.operation_id, + phase=args.phase, + instance=args.instance, + command_class=args.command_class, + ) + + dirty_checkout = bool(report["checkout"]["dirty_paths"]) + if dirty_checkout: + report["dirty_checkout_isolation"] = isolate_dirty_checkout_files( + checkout_dir=checkout, + isolate_dir=isolate_dir, + ) + + # Exclusive create prevents a provenance record from being overwritten. + with output_path.open("x", encoding="utf-8") as fp: + fp.write(json.dumps(report, ensure_ascii=False, indent=2) + "\n") + output_path.chmod(0o444) + print(str(output_path)) + if dirty_checkout: + 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..9290f88 100644 --- a/tests/adapters/persistence/test_ops_checks.py +++ b/tests/adapters/persistence/test_ops_checks.py @@ -7,13 +7,21 @@ 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 +import pytest + +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 +87,221 @@ 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() + subprocess.run(["git", "checkout", "--detach", head], cwd=repo, check=True, capture_output=True) + return repo, head + + +def _provenance_kwargs(repo: Path, head: str) -> dict: + broker_log = repo.parent / "broker.log" + broker_log.write_text("event=order_submit token=must-not-leak\n", encoding="utf-8") + bindir = repo.parent / "bin" + bindir.mkdir(exist_ok=True) + trex = bindir / "trex-core" + rabbit = bindir / "rabbitmqctl" + trex.write_text("#!/bin/sh\nprintf '{}\\n'\n", encoding="utf-8") + rabbit.write_text("#!/bin/sh\nprintf 'consumer\\n'\n", encoding="utf-8") + trex.chmod(0o755) + rabbit.chmod(0o755) + instance = "inst-a" + return { + "actor": "operator-a", + "reason": "deploy", + "instance": instance, + "command_class": "manual-start", + "operation_id": "op-1", + "phase": "pre", + "checkout_dir": repo, + "expected_oci_revision": head, + "image_digest": "sha256:" + ("a" * 64), + "runtime_files": ["tracked.txt"], + "capture_commands": { + "pending_slots": [ + str(trex), "slots", "--status", "pending", "--instance", instance, "--json" + ], + "reconciliation_slots": [ + str(trex), "slots", "--status", "all", "--instance", instance, "--json" + ], + "queue_consumers": [str(rabbit), "list_consumers"], + }, + "broker_log_path": broker_log, + } + + +def test_collect_release_provenance_passes_on_clean_checkout(tmp_path): + repo, head = _init_git_repo(tmp_path) + report = collect_release_provenance(**_provenance_kwargs(repo, head)) + assert report["checkout"]["git_clean"] is True + assert report["checkout"]["detached_head"] 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 + assert report["operation_id"] == "op-1" + assert report["phase"] == "pre" + rendered = str(report) + assert "must-not-leak" not in rendered + assert report["state_capture"]["broker_side_effect_logs"]["matches"][0]["event"] == "order_submit" + + +def test_collect_release_provenance_blocks_attached_checkout(tmp_path): + repo, head = _init_git_repo(tmp_path) + subprocess.run(["git", "switch", "-c", "audit-attached"], cwd=repo, check=True, capture_output=True) + + report = collect_release_provenance(**_provenance_kwargs(repo, head)) + + assert report["checkout"]["git_clean"] is True + assert report["checkout"]["detached_head"] is False + assert report["state_capture"]["command_outputs"] == {} + assert report["gate"]["passed"] is False + + +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) + kwargs = _provenance_kwargs(repo, old_head) + report = collect_release_provenance(**kwargs) + 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"} + snapshot = Path(manifest["isolate_dir"]) + assert snapshot.parent == isolated + assert (snapshot / "tracked.txt").read_text(encoding="utf-8") == "dirty\n" + assert (snapshot / "new.txt").read_text(encoding="utf-8") == "new\n" + assert (snapshot / "manifest.json").is_file() + + +def test_collect_release_provenance_rejects_runtime_path_escape(tmp_path): + repo, head = _init_git_repo(tmp_path) + kwargs = _provenance_kwargs(repo, head) + kwargs["runtime_files"] = ["../outside.txt"] + report = collect_release_provenance(**kwargs) + runtime = report["runtime_file_verification"][0] + assert runtime["path_within_checkout"] is False + assert runtime["matches_expected_revision"] is False + assert report["gate"]["passed"] is False + + +def test_collect_release_provenance_reports_git_failure_context(tmp_path): + not_repo = tmp_path / "not-repo" + not_repo.mkdir() + trex = tmp_path / "trex-core" + rabbit = tmp_path / "rabbitmqctl" + with pytest.raises(RuntimeError, match="git command failed in checkout="): + collect_release_provenance( + actor="operator-a", + reason="deploy", + instance="inst-a", + command_class="manual-start", + operation_id="op-1", + phase="pre", + checkout_dir=not_repo, + expected_oci_revision="0" * 40, + image_digest="sha256:" + ("a" * 64), + runtime_files=["tracked.txt"], + capture_commands={ + "pending_slots": [ + str(trex), "slots", "--status", "pending", "--instance", "inst-a", "--json" + ], + "reconciliation_slots": [ + str(trex), "slots", "--status", "all", "--instance", "inst-a", "--json" + ], + "queue_consumers": [str(rabbit), "list_consumers"], + }, + ) + + +def test_dirty_checkout_blocks_before_capture_commands_run(tmp_path): + repo, head = _init_git_repo(tmp_path) + marker = tmp_path / "capture-ran" + (repo / "tracked.txt").write_text("dirty\n", encoding="utf-8") + kwargs = _provenance_kwargs(repo, head) + for command in kwargs["capture_commands"].values(): + executable = Path(command[0]) + executable.write_text(f"#!/bin/sh\ntouch {marker}\n", encoding="utf-8") + executable.chmod(0o755) + + report = collect_release_provenance(**kwargs) + + assert report["gate"]["passed"] is False + assert report["gate"]["deployment_blocked"] is True + assert report["state_capture"]["command_outputs"] == {} + assert not marker.exists() + + +def test_capture_failure_or_missing_broker_log_blocks_gate(tmp_path): + repo, head = _init_git_repo(tmp_path) + kwargs = _provenance_kwargs(repo, head) + trex = Path(kwargs["capture_commands"]["pending_slots"][0]) + trex.write_text("#!/bin/sh\nexit 3\n", encoding="utf-8") + trex.chmod(0o755) + kwargs["broker_log_path"] = tmp_path / "missing.log" + + report = collect_release_provenance(**kwargs) + + assert report["gate"]["passed"] is False + assert report["gate"]["state_captures_succeeded"] is False + assert report["gate"]["broker_log_captured"] is False + + +def test_empty_broker_log_is_explicitly_captured_as_no_matching_events(tmp_path): + repo, head = _init_git_repo(tmp_path) + kwargs = _provenance_kwargs(repo, head) + Path(kwargs["broker_log_path"]).write_bytes(b"") + + report = collect_release_provenance(**kwargs) + + broker = report["state_capture"]["broker_side_effect_logs"] + assert broker["capture_status"] == "captured_empty" + assert broker["file_sha256"] == "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + assert broker["matches"] == [] + assert report["gate"]["passed"] is True + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("expected_oci_revision", None), + ("image_digest", "not-a-digest"), + ("runtime_files", []), + ("capture_commands", {}), + ("instance", "../escape"), + ("reason", "password=must-not-be-recorded"), + ], +) +def test_required_provenance_inputs_fail_closed(tmp_path, field, value): + repo, head = _init_git_repo(tmp_path) + kwargs = _provenance_kwargs(repo, head) + kwargs[field] = value + with pytest.raises((TypeError, ValueError)): + collect_release_provenance(**kwargs) diff --git a/tests/scripts/test_release_provenance_gate.py b/tests/scripts/test_release_provenance_gate.py new file mode 100644 index 0000000..72a7942 --- /dev/null +++ b/tests/scripts/test_release_provenance_gate.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from scripts import release_provenance_gate + + +def test_documented_python_invocation_loads_project_without_pythonpath(): + script = Path(release_provenance_gate.__file__).resolve() + env = os.environ.copy() + env.pop("PYTHONPATH", None) + + completed = subprocess.run( + [sys.executable, str(script), "--help"], + cwd=script.parent, + env=env, + capture_output=True, + text=True, + check=False, + ) + + assert completed.returncode == 0, completed.stderr + assert "--expected-oci-revision" in completed.stdout + assert script.stat().st_mode & 0o111 != 0 + + +def _init_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.email", "test@example.com"], cwd=repo, check=True) + subprocess.run(["git", "config", "user.name", "Test"], cwd=repo, check=True) + (repo / "runtime.py").write_text("print('ok')\n", encoding="utf-8") + subprocess.run(["git", "add", "runtime.py"], cwd=repo, check=True) + subprocess.run(["git", "commit", "-m", "init"], cwd=repo, check=True, capture_output=True) + head = subprocess.run( + ["git", "rev-parse", "HEAD"], cwd=repo, check=True, capture_output=True, text=True + ).stdout.strip() + subprocess.run(["git", "checkout", "--detach", head], cwd=repo, check=True, capture_output=True) + return repo, head + + +def _fake_read_only_commands(tmp_path: Path) -> tuple[Path, Path]: + bindir = tmp_path / "bin" + bindir.mkdir(parents=True) + trex = bindir / "trex-core" + rabbit = bindir / "rabbitmqctl" + trex.write_text("#!/bin/sh\nprintf '{}\\n'\n", encoding="utf-8") + rabbit.write_text("#!/bin/sh\nprintf 'consumer\\n'\n", encoding="utf-8") + trex.chmod(0o755) + rabbit.chmod(0o755) + return trex, rabbit + + +def _argv(tmp_path: Path, repo: Path, head: str) -> list[str]: + trex, rabbit = _fake_read_only_commands(tmp_path) + broker_log = tmp_path / "broker.log" + broker_log.write_text("event=order_submit account=must-not-leak\n", encoding="utf-8") + return [ + "--actor", "operator-a", + "--reason", "deploy release", + "--instance", "inst-a", + "--operation-id", "op-123", + "--phase", "pre", + "--command-class", "deploy", + "--checkout-dir", str(repo), + "--expected-oci-revision", head, + "--image-digest", "sha256:" + ("a" * 64), + "--runtime-file", "runtime.py", + "--broker-log-path", str(broker_log), + "--capture-command", f"pending_slots={trex} slots --status pending --instance inst-a --json", + "--capture-command", f"reconciliation_slots={trex} slots --status all --instance inst-a --json", + "--capture-command", f"queue_consumers={rabbit} list_consumers", + ] + + +def test_cli_writes_unique_record_outside_checkout_and_keeps_checkout_clean(tmp_path, capsys): + repo, head = _init_repo(tmp_path) + argv = _argv(tmp_path, repo, head) + + assert release_provenance_gate.main(argv) == 0 + first = Path(capsys.readouterr().out.strip()) + assert release_provenance_gate.main(argv) == 0 + second = Path(capsys.readouterr().out.strip()) + + assert first != second + assert first.parent == tmp_path / "repo-release-provenance" + assert first.stat().st_mode & 0o222 == 0 + assert second.stat().st_mode & 0o222 == 0 + payload = json.loads(first.read_text(encoding="utf-8")) + assert payload["gate"]["passed"] is True + assert "must-not-leak" not in first.read_text(encoding="utf-8") + status = subprocess.run( + ["git", "status", "--porcelain"], cwd=repo, check=True, capture_output=True, text=True + ).stdout + assert status == "" + + +def test_cli_dirty_checkout_is_fail_closed_and_isolated(tmp_path, capsys): + repo, head = _init_repo(tmp_path) + argv = _argv(tmp_path, repo, head) + (repo / "runtime.py").write_text("dirty\n", encoding="utf-8") + + assert release_provenance_gate.main(argv) == 2 + output = Path(capsys.readouterr().out.strip()) + payload = json.loads(output.read_text(encoding="utf-8")) + assert output.stat().st_mode & 0o222 == 0 + assert payload["gate"]["passed"] is False + assert payload["gate"]["deployment_blocked"] is True + assert payload["state_capture"]["command_outputs"] == {} + snapshot = Path(payload["dirty_checkout_isolation"]["isolate_dir"]) + assert snapshot.parent == tmp_path / "repo-release-provenance-isolation" + assert (snapshot / "runtime.py").read_text() == "dirty\n" + + +def test_cli_rejects_unsafe_command_and_checkout_local_output(tmp_path): + repo, head = _init_repo(tmp_path) + argv = _argv(tmp_path, repo, head) + unsafe = argv.index(next(x for x in argv if x.startswith("pending_slots="))) + argv[unsafe] = "pending_slots=trex-core cancel slot-1 --yes" + with pytest.raises(SystemExit): + release_provenance_gate.main(argv) + + argv = _argv(tmp_path / "other", repo, head) + argv.extend(["--output-dir", str(repo / "artifacts")]) + with pytest.raises(SystemExit): + release_provenance_gate.main(argv) diff --git a/trex_core/adapters/persistence/ops_checks.py b/trex_core/adapters/persistence/ops_checks.py index a34888d..d9edfc7 100644 --- a/trex_core/adapters/persistence/ops_checks.py +++ b/trex_core/adapters/persistence/ops_checks.py @@ -11,8 +11,16 @@ 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 re +import shutil +import subprocess +import uuid +from collections import deque +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 +45,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 +53,398 @@ 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()] + + +_SENSITIVE_KEY = re.compile( + r"(?:account|acct|token|credential|password|passwd|secret|api[_-]?key|dsn)", + re.IGNORECASE, +) +_SENSITIVE_VALUE = re.compile( + r"(?i)\b(account|acct|token|credential|password|passwd|secret|api[_-]?key|dsn)" + r"\s*[:=]\s*([^\s,;]+)" +) +_FULL_GIT_REVISION = re.compile(r"[0-9a-f]{40}") +_IMAGE_DIGEST = re.compile(r"sha256:[0-9a-f]{64}") +_SAFE_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}") +_REQUIRED_CAPTURE_KEYS = frozenset( + {"pending_slots", "reconciliation_slots", "queue_consumers"} +) + + +def _run_git(checkout_dir: Path, *args: str) -> str: + command = ["git", *args] + try: + return subprocess.check_output( + command, + cwd=str(checkout_dir), + text=True, + ).strip() + except subprocess.CalledProcessError as exc: + joined = " ".join(command) + raise RuntimeError( + f"git command failed in checkout={checkout_dir}: {joined}" + ) from exc + + +def _git_status(checkout_dir: Path) -> tuple[list[dict[str, str]], list[str]]: + try: + raw = subprocess.check_output( + ["git", "status", "--porcelain=v1", "-z", "--untracked-files=all"], + cwd=str(checkout_dir), + ) + except subprocess.CalledProcessError as exc: + raise RuntimeError(f"git status failed in checkout={checkout_dir}") from exc + fields = raw.decode("utf-8", errors="surrogateescape").split("\0") + entries: list[dict[str, str]] = [] + paths: list[str] = [] + index = 0 + while index < len(fields): + field = fields[index] + index += 1 + if not field: + continue + if len(field) < 4: + raise RuntimeError("unexpected git status porcelain record") + code, path = field[:2], field[3:] + entry = {"code": code, "path": path} + if "R" in code or "C" in code: + if index >= len(fields) or not fields[index]: + raise RuntimeError("unexpected git rename/copy porcelain record") + entry["original_path"] = fields[index] + index += 1 + entries.append(entry) + paths.append(path) + return entries, paths + + +def _git_head_is_detached(checkout_dir: Path) -> bool: + proc = subprocess.run( + ["git", "symbolic-ref", "-q", "HEAD"], + cwd=checkout_dir, + capture_output=True, + text=True, + check=False, + ) + if proc.returncode == 0: + return False + if proc.returncode == 1: + return True + detail = _sanitize_text(proc.stderr.strip()) or f"exit={proc.returncode}" + raise RuntimeError(f"git symbolic-ref failed in checkout={checkout_dir}: {detail}") + + +def _sanitize_text(value: str) -> str: + return _SENSITIVE_VALUE.sub(lambda match: f"{match.group(1)}=[REDACTED]", value) + + +def _sanitize_json(value: object) -> object: + if isinstance(value, dict): + return { + str(key): "[REDACTED]" if _SENSITIVE_KEY.search(str(key)) else _sanitize_json(item) + for key, item in value.items() + } + if isinstance(value, list): + return [_sanitize_json(item) for item in value] + if isinstance(value, str): + return _sanitize_text(value) + return value + + +def _validate_capture_commands( + capture_commands: dict[str, list[str]], *, instance: str +) -> None: + if set(capture_commands) != _REQUIRED_CAPTURE_KEYS: + raise ValueError("all required state capture commands must be supplied exactly once") + expected = { + "pending_slots": ["slots", "--status", "pending", "--instance", instance, "--json"], + "reconciliation_slots": ["slots", "--status", "all", "--instance", instance, "--json"], + "queue_consumers": ["list_consumers"], + } + for key, command in capture_commands.items(): + if not command: + raise ValueError(f"capture command is empty for {key}") + executable = Path(command[0]).name + valid = ( + executable == "rabbitmqctl" and command[1:] == expected[key] + if key == "queue_consumers" + else executable == "trex-core" and command[1:] == expected[key] + ) + if not valid: + raise ValueError(f"capture command is not the approved read-only form for {key}") + + +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 _resolve_under(root: Path, relpath: str) -> Path | None: + resolved_root = root.resolve() + candidate = (resolved_root / relpath).resolve() + try: + candidate.relative_to(resolved_root) + return candidate + except ValueError: + return None + + +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, + ) + try: + stdout: object = _sanitize_json(json.loads(done.stdout)) + stdout_format = "json" + except json.JSONDecodeError: + stdout = _sanitize_text(done.stdout) + stdout_format = "text" + return { + "command": command, + "returncode": done.returncode, + "stdout": stdout, + "stdout_format": stdout_format, + "stderr": _sanitize_text(done.stderr), + } + except (OSError, subprocess.SubprocessError) as exc: # pragma: no cover + return { + "command": command, + "returncode": None, + "stdout": "", + "stdout_format": "text", + "stderr": type(exc).__name__, + } + + +def _capture_broker_side_effect_logs(log_path: Path) -> dict: + if not log_path.exists(): + return { + "path": str(log_path), + "exists": False, + "capture_status": "unavailable", + "matches": [], + } + matches: deque[dict[str, object]] = deque(maxlen=200) + file_hash = hashlib.sha256() + scanned_lines = 0 + try: + with log_path.open("rb") as fp: + for line_number, raw_line in enumerate(fp, start=1): + scanned_lines = line_number + file_hash.update(raw_line) + lowered = raw_line.decode("utf-8", errors="replace").lower() + event = next( + ( + name + for marker, name in ( + ("event=order_submit", "order_submit"), + ("event=order_fill", "order_fill"), + ("event=order_reject", "order_reject"), + ("event=order_cancel", "order_cancel"), + ) + if marker in lowered + ), + None, + ) + if event is not None: + matches.append( + { + "event": event, + "line_number": line_number, + "line_sha256": _sha256_bytes(raw_line.rstrip(b"\n")), + } + ) + except OSError as exc: + return { + "path": str(log_path), + "exists": True, + "capture_status": "error", + "error_type": type(exc).__name__, + "matches": [], + } + return { + "path": str(log_path), + "exists": True, + "capture_status": "captured" if matches else "captured_empty", + "file_sha256": file_hash.hexdigest(), + "scanned_lines": scanned_lines, + "matches": list(matches), + } + + +def collect_release_provenance( + *, + actor: str, + reason: str, + instance: str, + command_class: str, + operation_id: str, + phase: str, + expected_oci_revision: str, + image_digest: str, + runtime_files: list[str], + capture_commands: dict[str, list[str]], + checkout_dir: str | Path = ".", + broker_log_path: str | Path | None = "/tmp/trex-core.log", +) -> dict: + """Collect release/restart provenance with mandatory fail-closed evidence.""" + if not actor.strip() or not reason.strip(): + raise ValueError("actor and reason must be non-empty") + if _sanitize_text(actor) != actor or _sanitize_text(reason) != reason: + raise ValueError("actor and reason must not contain sensitive key/value material") + if not _SAFE_ID.fullmatch(instance) or not _SAFE_ID.fullmatch(operation_id): + raise ValueError("instance and operation_id must be safe identifiers") + if phase not in {"pre", "post"}: + raise ValueError("phase must be pre or post") + if not _FULL_GIT_REVISION.fullmatch(expected_oci_revision): + raise ValueError("expected_oci_revision must be a full lowercase git SHA") + if not _IMAGE_DIGEST.fullmatch(image_digest): + raise ValueError("image_digest must be sha256:<64 lowercase hex>") + if not runtime_files: + raise ValueError("at least one runtime file is required") + _validate_capture_commands(capture_commands, instance=instance) + + checkout = Path(checkout_dir).resolve() + head = _run_git(checkout, "rev-parse", "HEAD") + detached_head = _git_head_is_detached(checkout) + status_entries, dirty_paths = _git_status(checkout) + git_clean = len(dirty_paths) == 0 + runtime_reports = [] + for relpath in runtime_files: + abs_path = _resolve_under(checkout, relpath) + path_valid = abs_path is not None + exists = bool(path_valid and 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, expected_oci_revision, relpath) + expected_sha = _sha256_bytes(expected_bytes) if expected_bytes is not None else None + runtime_reports.append( + { + "path": relpath, + "path_within_checkout": path_valid, + "exists": exists, + "current_sha256": current_sha, + "expected_revision": expected_oci_revision, + "expected_sha256": expected_sha, + "matches_expected_revision": bool( + path_valid and exists and expected_bytes is not None and current_bytes == expected_bytes + ), + } + ) + + # A dirty audit checkout blocks before any operator command is executed. + command_outputs: dict[str, dict] = {} + broker_logs = None + if detached_head and git_clean: + for key, cmd in capture_commands.items(): + command_outputs[key] = _capture_command(cmd, checkout) + broker_logs = ( + _capture_broker_side_effect_logs(Path(broker_log_path)) + if broker_log_path + else None + ) + oci_revision_matches_head = expected_oci_revision == head + runtime_ok = all(item["matches_expected_revision"] for item in runtime_reports) + commands_ok = set(command_outputs) == _REQUIRED_CAPTURE_KEYS and all( + item.get("returncode") == 0 for item in command_outputs.values() + ) + broker_logs_ok = bool( + broker_logs + and broker_logs.get("capture_status") in {"captured", "captured_empty"} + ) + gate_passed = ( + detached_head + and git_clean + and oci_revision_matches_head + and runtime_ok + and commands_ok + and broker_logs_ok + ) + return { + "timestamp_utc": datetime.now(tz=UTC).isoformat(), + "actor": actor, + "reason": reason, + "instance": instance, + "command_class": command_class, + "operation_id": operation_id, + "phase": phase, + "checkout": { + "path": str(checkout), + "head_revision": head, + "detached_head": detached_head, + "git_status_porcelain": status_entries, + "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": broker_logs, + }, + "gate": { + "passed": gate_passed, + "deployment_blocked": not gate_passed, + "runtime_files_verified": runtime_ok, + "state_captures_succeeded": commands_ok, + "broker_log_captured": broker_logs_ok, + }, + } + + +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_root = Path(isolate_dir).resolve() + stamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%S.%fZ") + dst = dst_root / f"{stamp}--{uuid.uuid4().hex[:12]}" + _, dirty_paths = _git_status(checkout) + copied: list[str] = [] + for rel in dirty_paths: + src = _resolve_under(checkout, rel) + if src is None or not src.exists() or src.is_dir(): + continue + target = (dst / rel).resolve() + try: + target.relative_to(dst) + except ValueError: + continue + 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) + with (dst / "manifest.json").open("x", encoding="utf-8") as fp: + fp.write(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n") + return manifest