"""anu_v2.owner_trigger_only — OWNER_TRIGGER_ONLY_CAPABILITY core module. 회장 §명시 14장 1:1 박제 (2026-05-11 KST, OWNER_TRIGGER_ONLY_CAPABILITY). task-2554+1 회장 §명시 (2026-05-12 KST) HIGH race condition fix 반영. task-2554+2 회장 §명시 (2026-05-12 KST) §1 1:1 완성: - ``RESULT_PENDING`` import + sentinel 활용. - ``http_post`` 호출 직전 ``txn.record(PENDING)`` 으로 영구 기록 — process crash 후 다음 runner 가 DEDUPED 판정 (crash-safety fail-closed). 본 모듈 범위: - public action **1 종**: ``POST_GEMINI_REVIEW_TRIGGER_COMMENT`` - comment body 상수 (외부 입력 X): ``COMMENT_BODY = "/gemini review"`` - 단일 허용 endpoint: ``POST /repos/{owner}/{repo}/issues/{pr_number}/comments`` - 금지 11 endpoint hard-block (``PermissionError`` raise) - 전용 token 1 종: ``OWNER_GEMINI_TRIGGER_TOKEN`` (다른 token fallback 0) - merge path 와 완전 분리: ``BOT_GITHUB_TOKEN`` 본 모듈 어디서도 사용 0 - dedupe atomic (audit JSONL + fcntl flock + sidecar lock transaction) - token redaction guard (audit 에 raw value 미출력) - decision JSON v1 검증 fail-closed - result enum: POSTED / DEDUPED / FAILED / PENDING result 흐름 (task-2554+2 §1): 1. txn.check_dedupe (POSTED + PENDING) 2. (DEDUPED 인 경우 audit DEDUPED + return) 3. token presence + hash_prefix 계산 4. **txn.record(PENDING)** — http_post 직전 영구 기록 (crash-safety) 5. http_post 호출 6. 성공: txn.record(POSTED) / 실패: txn.record(FAILED) one-way isolation: anu_v2/ 외부 import 금지. """ from __future__ import annotations import json import re from dataclasses import dataclass from pathlib import Path from typing import Callable, Final from anu_v2.owner_trigger_audit import ( AuditRedactionError, DedupeViolation, OwnerTriggerAudit, RESULT_DEDUPED, RESULT_FAILED, RESULT_PENDING, RESULT_POSTED, token_hash_prefix, ) from anu_v2.owner_trigger_decision import ( ALLOWED_ACTION, ALLOWED_COMMENT_BODY, DecisionInvalidError, validate_decision, ) # ─── 코드 상수 (외부 입력 절대 X) ──────────────────────────────────────────────── COMMENT_BODY: Final[str] = "/gemini review" TOKEN_ENV_NAME: Final[str] = "OWNER_GEMINI_TRIGGER_TOKEN" _ALLOWED_PATH_RE: Final[re.Pattern[str]] = re.compile( r"^/repos/[^/]+/[^/]+/issues/\d+/comments$" ) # 금지 11 endpoint 패턴 (회장 §7 명시) — static + dynamic 양쪽 차단. FORBIDDEN_ENDPOINT_PATTERNS: Final[tuple[re.Pattern[str], ...]] = ( re.compile(r"/pulls/\d+/merge\b"), re.compile(r"/pulls/\d+/reviews\b"), re.compile(r"/git/refs\b"), re.compile(r"/contents\b"), re.compile(r"/issues/\d+(?:\?|$)"), re.compile(r"/labels\b"), re.compile(r"/branches\b"), re.compile(r"/actions/(?:runs|workflows)/.*/rerun"), re.compile(r"/check-runs/.*/rerequest|/check-suites/.*/rerequest"), re.compile(r"/pulls/\d+\?|/pulls/\d+$"), ) # 다른 token 환경변수 이름 (모두 본 모듈에서 사용 금지) FORBIDDEN_TOKEN_ENV_NAMES: Final[tuple[str, ...]] = ( "BOT_GITHUB_TOKEN", "GH_TOKEN", "GITHUB_TOKEN", "OWNER_PAT", "PAT_TOKEN", ) # ─── 결과 dataclass ─────────────────────────────────────────────────────────── @dataclass(frozen=True) class TriggerResult: """``trigger_gemini_review`` 결과 객체. token / authorization header 미포함.""" status: str # "POSTED" | "DEDUPED" | "FAILED" | "PENDING" pr: int head: str action: str comment_body: str endpoint: str token_present: bool token_hash_prefix: str # sha256 앞 8자 (raw value 아님) error_code: str | None = None # ─── 예외 분류 ─────────────────────────────────────────────────────────────── class ForbiddenEndpointError(PermissionError): """금지 11 endpoint 호출 시도 (회장 §7 hard-block).""" class TokenBoundaryViolation(PermissionError): """다른 token 사용 시도 (회장 §6 fail).""" class CommentBodyViolation(ValueError): """COMMENT_BODY 외부 변조 / 외부 입력 사용 시도 (회장 §4 fail).""" class MergePathViolation(PermissionError): """merge_queue_executor merge path 와의 분리 위반 (회장 §9 fail).""" # ─── 단일 endpoint 보안 빌더 ─────────────────────────────────────────────────── def _build_post_comment_path(owner: str, repo: str, pr_number: int) -> str: if not isinstance(owner, str) or not owner or "/" in owner: raise ValueError("owner must be a non-empty string without '/'") if not isinstance(repo, str) or not repo or "/" in repo: raise ValueError("repo must be a non-empty string without '/'") if not isinstance(pr_number, int) or isinstance(pr_number, bool) or pr_number <= 0: raise ValueError("pr_number must be a positive int") return f"/repos/{owner}/{repo}/issues/{pr_number}/comments" def assert_endpoint_allowed(method: str, path: str) -> None: """static + dynamic endpoint hard-block guard.""" if not isinstance(method, str) or method.upper() != "POST": raise ForbiddenEndpointError(f"forbidden method: {method!r}") if not isinstance(path, str): raise ForbiddenEndpointError("path must be string") for pat in FORBIDDEN_ENDPOINT_PATTERNS: if pat.search(path): raise ForbiddenEndpointError(f"forbidden endpoint path: {path!r}") if not _ALLOWED_PATH_RE.match(path): raise ForbiddenEndpointError( f"path not in allow-list (POST issues/{{n}}/comments only): {path!r}" ) def assert_token_boundary(env: dict | None = None) -> None: """token 경계 검증 (회장 §6). 명시적으로 ``env`` dict 에 다른 token env 이름이 키로 전달되면 fail. """ if env is None: return for forbidden in FORBIDDEN_TOKEN_ENV_NAMES: if forbidden in env: raise TokenBoundaryViolation( f"forbidden token env injected: {forbidden} — owner trigger path " f"must use {TOKEN_ENV_NAME} only" ) # ─── core class ──────────────────────────────────────────────────────────────── class OwnerTriggerOnly: """OWNER_TRIGGER_ONLY_CAPABILITY 단일 진입점. public action: ``trigger_gemini_review(decision_path, owner, repo, current_head_actual)``. side-effect 추상화 (test injection): - ``http_post``: ``Callable[[method:str, path:str, body:dict, headers:dict], dict]`` - ``token_provider``: ``Callable[[], str]`` — OWNER_GEMINI_TRIGGER_TOKEN value 반환. task-2554+2 §1 흐름: 1. token boundary 검증 2. decision JSON 검증 3. comment_body 검증 4. endpoint hard-block 5. ── ATOMIC TRANSACTION ──────────────────────────────── 5a. audit.transaction() 진입 (sidecar lock LOCK_EX) 5b. txn.check_dedupe (POSTED + PENDING) 5c. DEDUPED 시 audit record + return 5d. token presence + hash_prefix 계산 5e. **txn.record(PENDING)** — http_post 호출 직전 영구 기록 (crash-safety) 5f. http_post 호출 5g. 성공 시 audit POSTED / 실패 시 audit FAILED 6. lock 해제 """ def __init__( self, *, workspace_root: str | Path, http_post: Callable[[str, str, dict, dict], dict], token_provider: Callable[[], str], audit: OwnerTriggerAudit | None = None, ) -> None: if http_post is None: raise NotImplementedError("http_post callable must be injected") if token_provider is None: raise NotImplementedError("token_provider callable must be injected") self._workspace_root = Path(workspace_root).resolve() self._http_post = http_post self._token_provider = token_provider self._audit = audit if audit is not None else OwnerTriggerAudit(self._workspace_root) # ─── 내부 helpers ─────────────────────────────────────────────────────── def _read_decision(self, decision_path: str | Path) -> dict: path = Path(decision_path) if not path.is_absolute(): path = self._workspace_root / path if not path.exists(): raise DecisionInvalidError("E_FILE_MISSING", f"decision file not found: {path}") data = json.loads(path.read_text(encoding="utf-8")) return data # ─── public API ──────────────────────────────────────────────────────── def trigger_gemini_review( self, *, decision_path: str | Path, owner: str, repo: str, current_head_actual: str, env_override: dict | None = None, comment_body: str | None = None, ) -> TriggerResult: """단일 public action. /gemini review 댓글 1회 작성. race condition: 동시 2 process 가 같은 (pr, head) 호출 시 lock 직렬화로 후속 process 의 check_dedupe 가 항상 DedupeViolation 차단. crash-safety (task-2554+2 §1): http_post 호출 직전 ``txn.record(PENDING)`` 으로 영구 기록. process crash 후 다음 runner 가 ``_has_active_trigger`` 에서 PENDING 발견 → DedupeViolation → DEDUPED 판정 (fail-closed). """ # 1) token boundary assert_token_boundary(env_override) # 2) decision 검증 decision = self._read_decision(decision_path) validate_decision(decision, current_head_actual=current_head_actual) # 3) comment_body 검증 (외부 입력 0) body = COMMENT_BODY if comment_body is None else comment_body if body != ALLOWED_COMMENT_BODY: raise CommentBodyViolation( f"comment_body must be {ALLOWED_COMMENT_BODY!r}, got {body!r}" ) # 4) endpoint 빌드 + hard-block pr_num = int(decision["pr"]) head = decision["current_head"] path = _build_post_comment_path(owner, repo, pr_num) assert_endpoint_allowed("POST", path) decision_path_str = str(decision_path) task_id_val = decision.get("task_id", "") # 5) ATOMIC TRANSACTION with self._audit.transaction() as txn: # 5a) atomic dedupe check (POSTED + PENDING) try: txn.check_dedupe(pr=pr_num, head=head) except DedupeViolation: txn.record( { "task_id": task_id_val, "pr": pr_num, "head": head, "action": ALLOWED_ACTION, "result": RESULT_DEDUPED, "comment_body": COMMENT_BODY, "endpoint": path, "decision_path": decision_path_str, "token_present": False, "token_hash_prefix": "", "token_value_logged": False, "error_code": "DEDUPE", } ) return TriggerResult( status=RESULT_DEDUPED, pr=pr_num, head=head, action=ALLOWED_ACTION, comment_body=COMMENT_BODY, endpoint=path, token_present=False, token_hash_prefix="", error_code="DEDUPE", ) # 5b) token presence + hash prefix (raw value 보관/노출 X) token = self._token_provider() if not isinstance(token, str) or not token: raise TokenBoundaryViolation( f"{TOKEN_ENV_NAME} not provided — owner trigger fail-closed" ) hash_prefix = token_hash_prefix(token) # 5c) ★ task-2554+2 §1: http_post 직전 PENDING 영구 기록 (crash-safety) # process 가 http_post 호출 전후 어디서 crash 해도 PENDING 흔적이 남아 # 다음 runner 가 DEDUPED 판정 → 중복 trigger 차단. txn.record( { "task_id": task_id_val, "pr": pr_num, "head": head, "action": ALLOWED_ACTION, "result": RESULT_PENDING, "comment_body": COMMENT_BODY, "endpoint": path, "decision_path": decision_path_str, "token_present": True, "token_hash_prefix": hash_prefix, "token_value_logged": False, } ) # 5d) HTTP POST — 외부에 노출되는 본 모듈 유일 호출 지점 headers = { "Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28", } body_payload = {"body": COMMENT_BODY} try: self._http_post("POST", path, body_payload, headers) except Exception as exc: # task-2554+1 medium #4: http_post 예외 시 audit FAILED 기록. # PENDING sentinel 은 이미 남아있으므로 다음 runner 가 fail-closed. try: txn.record( { "task_id": task_id_val, "pr": pr_num, "head": head, "action": ALLOWED_ACTION, "result": RESULT_FAILED, "comment_body": COMMENT_BODY, "endpoint": path, "decision_path": decision_path_str, "token_present": True, "token_hash_prefix": hash_prefix, "token_value_logged": False, "error_code": "HTTP_POST_FAIL", } ) finally: headers = {} token = "" # type: ignore[assignment] raise exc finally: headers = {} token = "" # type: ignore[assignment] # 5e) audit POSTED append try: txn.record( { "task_id": task_id_val, "pr": pr_num, "head": head, "action": ALLOWED_ACTION, "result": RESULT_POSTED, "comment_body": COMMENT_BODY, "endpoint": path, "decision_path": decision_path_str, "token_present": True, "token_hash_prefix": hash_prefix, "token_value_logged": False, } ) except (DedupeViolation, AuditRedactionError): raise return TriggerResult( status=RESULT_POSTED, pr=pr_num, head=head, action=ALLOWED_ACTION, comment_body=COMMENT_BODY, endpoint=path, token_present=True, token_hash_prefix=hash_prefix, error_code=None, ) # ─── merge path 분리 — 본 모듈에서 merge 시도 0 ────────────────────────── def merge(self, *args, **kwargs): # pragma: no cover — 정적 정책 표현 raise MergePathViolation( "owner_trigger_only must not perform merge. " "Use merge_queue_executor with BOT_GITHUB_TOKEN." ) def approve(self, *args, **kwargs): # pragma: no cover raise ForbiddenEndpointError("approve forbidden (review submit blocked)") def close(self, *args, **kwargs): # pragma: no cover raise ForbiddenEndpointError("close forbidden") def reopen(self, *args, **kwargs): # pragma: no cover raise ForbiddenEndpointError("reopen forbidden")