# -*- coding: utf-8 -*- """utils.snapshot_crossref_validator — Step 0b/0c/0d 통합 helper. task-2639 — real_merge_hooks chair_authorization snapshot 교차검증 정책 수정 (real merge 실행 0). Spec: memory/specs/system_real_merge_hooks_snapshot_crossref_spec_260523.md sha256: 12b8af006913833596562c55ab9a0acca935830be90c5f17f2af4b7e1e632621 회장 verbatim 10항목 (spec §2): 1. forbidden prefix 검사 전에 chair_authorization expected_files_snapshot 로드 2. changed_files 전체가 expected_files_snapshot 의 부분집합인지 확인 3. PR number / head_sha 가 chair_authorization 과 정확 일치 확인 4. snapshot 에 없는 forbidden prefix 파일 → NO_OP_FORBIDDEN_PATH 유지 5. snapshot 에 있는 파일이라도 production / secret / admin override 시 차단 6. .tasks/locks/* sanctioned push-gate artifact 는 분리 기록 7. tests/fixtures/INDEX.md 같은 doc-only → snapshot 일치 시 통과 가능 8. allow_reason 을 merge_decision.json 에 명시 9. broad prefix allowlist 금지 (snapshot exact match 만) 10. 기존 forbidden path 보안 의미 유지 (snapshot 정합 없으면 동일 동작) ANCHOR-2 (task md): "forbidden prefix 보안 가드 유지 · snapshot exact match 만 통과 · broad allowlist 금지" """ from __future__ import annotations from typing import Any, Dict, List, Optional # Public schema constant — embedded into merge_decision.json snapshot_crossref block. SNAPSHOT_CROSSREF_SCHEMA = "utils.snapshot_crossref_validator.v1" # allow_reason verbatim token (spec §2 #8 / fixture expected.json). ALLOW_REASON_SNAPSHOT_CROSSREF = "chair_authorization_snapshot_crossref" # Sanctioned push-gate artifact prefix (spec §2 #6 — .tasks/locks/* separation). SANCTIONED_LOCK_PREFIX = ".tasks/locks/" # Production-area prefixes — snapshot 에 포함된 경우 CHAIR_REQUIRED 격상 신호 # (ANCHOR-4 / spec §3 Step 0e). 정적 정의 — snapshot 외부 forbidden 가드는 # real_merge_hooks.FORBIDDEN_PATHS / FORBIDDEN_DIR_PREFIXES 가 우선. _PRODUCTION_DIR_PREFIXES = ( "utils/", "scripts/", "dispatch/", ) # Blocking-secret 의심 토큰 — snapshot key 자체에 포함 시 CHAIR_REQUIRED 격상 # (spec §4 CHAIR_REQUIRED_BLOCKING_SECRET_IN_SNAPSHOT). # ★ broad allowlist 금지 doctrine 보존을 위해 보수적 키워드만 사용. _SECRET_TOKEN_PATTERNS = ( "ghp_", "github_pat_", "-----BEGIN", # PEM header substring (filename 에 들어가는 일은 드물지만 방어) ".pem", "secret", "private_key", ) def _normalize_pr(value: Any) -> Optional[int]: """int-convertible 한 PR 번호로 정규화. 실패 시 None.""" if value is None: return None try: return int(value) except (TypeError, ValueError): return None def _normalize_pr_list(values: Any) -> List[int]: """int convertible 한 PR 목록 정규화. 비정규 항목은 폐기.""" if not isinstance(values, list): return [] out: List[int] = [] for v in values: n = _normalize_pr(v) if n is not None: out.append(n) return out def _extract_snapshot_keys(chair_authorization: Any) -> List[str]: """expected_files_snapshot 필드를 정렬된 string list 로 정규화. - dict → key 목록 - list → str 항목만 - None / 누락 → [] """ if not isinstance(chair_authorization, dict): return [] snap = chair_authorization.get("expected_files_snapshot") if snap is None: return [] if isinstance(snap, dict): keys = [k for k in snap.keys() if isinstance(k, str)] elif isinstance(snap, list): keys = [k for k in snap if isinstance(k, str)] else: return [] return sorted(keys) def _is_forbidden( path: str, forbidden_paths: List[str], forbidden_dir_prefixes: List[str], ) -> bool: """real_merge_hooks 와 동일 doctrine 으로 forbidden 판정. 의존 사이클을 피하기 위해 caller 가 FORBIDDEN_PATHS / FORBIDDEN_DIR_PREFIXES 를 주입한다 (validator 는 real_merge_hooks 를 import 하지 않음). """ if path in forbidden_paths: return True for prefix in forbidden_dir_prefixes: if path.startswith(prefix): return True return False def _is_production_path(path: str) -> bool: """production-area 디렉토리 prefix 매칭 (ANCHOR-4).""" for prefix in _PRODUCTION_DIR_PREFIXES: if path.startswith(prefix): return True return False def _has_secret_token(path: str) -> bool: """경로 문자열에 blocking-secret 의심 토큰이 포함되어 있는지.""" lower = path.lower() for token in _SECRET_TOKEN_PATTERNS: if token.lower() in lower: return True return False def validate_snapshot_crossref( pr_identity: Any, chair_authorization: Any, changed_files: Optional[List[str]], *, forbidden_paths: Optional[List[str]] = None, forbidden_dir_prefixes: Optional[List[str]] = None, ) -> Dict[str, Any]: """chair_authorization snapshot 교차검증 결과를 반환. Args: pr_identity: ``{"pr": int|str, "head_sha": str, ...}`` 형태 dict. chair_authorization: ``{"pr_numbers": [...], "head_shas": [...], "expected_files_snapshot": [...]|{...}, ...}`` 형태 dict (None 허용). changed_files: 변경된 경로 목록. ``None`` → fail-closed sentinel (``__INPUT_NONE_FAIL_CLOSED__``) 1건이 분류에 기록되고 ``unauthorized_forbidden_hits`` 에도 포함. forbidden_paths / forbidden_dir_prefixes: real_merge_hooks 에서 주입. 누락 시 빈 리스트 → forbidden 가드 비활성화 (테스트 전용). Returns: dict with keys:: schema : SNAPSHOT_CROSSREF_SCHEMA pr_match : bool sha_match : bool snapshot_present : bool — expected_files_snapshot 필드 유무 snapshot_keys : sorted list[str] classification : { task_outputs : list[str], sanctioned_artifacts : list[str], unauthorized_forbidden_hits: list[str], authorized_forbidden_hits : list[str], # snapshot exact match } allow_reason : ALLOW_REASON_SNAPSHOT_CROSSREF | None production_in_snapshot : list[str] — snapshot 내 production prefix blocking_secret_in_snapshot : list[str] — snapshot 내 secret token """ forbidden_paths = list(forbidden_paths or []) forbidden_dir_prefixes = list(forbidden_dir_prefixes or []) # ── pr/sha match (Step 0b) ─────────────────────────────────────────────── pr_match = False sha_match = False if isinstance(chair_authorization, dict) and isinstance(pr_identity, dict): pr_int = _normalize_pr(pr_identity.get("pr")) head_sha = pr_identity.get("head_sha") authorized_prs = _normalize_pr_list(chair_authorization.get("pr_numbers")) authorized_shas = chair_authorization.get("head_shas") if pr_int is not None and pr_int in authorized_prs: pr_match = True if ( isinstance(head_sha, str) and isinstance(authorized_shas, list) and head_sha in authorized_shas ): sha_match = True # ── snapshot keys (Step 0c 로드) ───────────────────────────────────────── snapshot_present = ( isinstance(chair_authorization, dict) and "expected_files_snapshot" in chair_authorization ) snapshot_keys = _extract_snapshot_keys(chair_authorization) snapshot_set = set(snapshot_keys) # ── classification (Step 0c forbidden / Step 0d sanctioned split) ──────── task_outputs: List[str] = [] sanctioned_artifacts: List[str] = [] unauthorized_forbidden_hits: List[str] = [] authorized_forbidden_hits: List[str] = [] if changed_files is None: # fail-closed sentinel — real_merge_hooks 의 detect_forbidden_paths 와 동일 # 동작을 유지 (None 입력은 검증 불가 신호로 forbidden hit 처리). unauthorized_forbidden_hits.append("__INPUT_NONE_FAIL_CLOSED__") else: for path in changed_files: if not isinstance(path, str): # 비-string 항목은 task_outputs 로 분류하지 않고 무시 (실제 코드 # 경로에서 발생할 수 없는 입력이지만 fail-closed 유지). continue if path.startswith(SANCTIONED_LOCK_PREFIX): sanctioned_artifacts.append(path) continue forbidden_hit = _is_forbidden(path, forbidden_paths, forbidden_dir_prefixes) if forbidden_hit: if path in snapshot_set: authorized_forbidden_hits.append(path) task_outputs.append(path) else: unauthorized_forbidden_hits.append(path) else: task_outputs.append(path) allow_reason: Optional[str] = ( ALLOW_REASON_SNAPSHOT_CROSSREF if authorized_forbidden_hits else None ) # ── snapshot 내 production / secret 검사 (Step 0e 보조 — ANCHOR-4) ─────── production_in_snapshot = sorted( {k for k in snapshot_keys if _is_production_path(k)} ) blocking_secret_in_snapshot = sorted( {k for k in snapshot_keys if _has_secret_token(k)} ) return { "schema": SNAPSHOT_CROSSREF_SCHEMA, "pr_match": pr_match, "sha_match": sha_match, "snapshot_present": snapshot_present, "snapshot_keys": snapshot_keys, "classification": { "task_outputs": task_outputs, "sanctioned_artifacts": sanctioned_artifacts, "unauthorized_forbidden_hits": unauthorized_forbidden_hits, "authorized_forbidden_hits": authorized_forbidden_hits, }, "allow_reason": allow_reason, "production_in_snapshot": production_in_snapshot, "blocking_secret_in_snapshot": blocking_secret_in_snapshot, } __all__ = [ "SNAPSHOT_CROSSREF_SCHEMA", "ALLOW_REASON_SNAPSHOT_CROSSREF", "SANCTIONED_LOCK_PREFIX", "validate_snapshot_crossref", ]