"""task-2693 callback authority 4-source validator (★ 회장 verbatim PR #152 hardening #4). ★ envelope 텍스트는 증거 아님 (회장 verbatim ANCHOR-2). ★ 4-source 모두 일치 시에만 PASS. 하나라도 실패 → fail-closed → HOLD_FOR_CHAIR. Sources (4): 1. envelope owner_key == ANU key (c119085addb0f8b7) — self-key fail-closed 2. actual cron channel owner_key (cokacdir --cron-history 결과) == ANU key 3. envelope 3 SID (chair_facing / collector / delivery) 모두 동일 4. schedule_history `.log` last entry status == 'ok' AND chair_facing_session_id 필드 일치 Usage: validate( envelope=, actual_cron_owner_key=, schedule_history_record=, anu_keys=DEFAULT_ANU_KEYS, ) -> Verdict Verdict.classification ∈ { "AUTHORITATIVE_4SOURCE_OK", "ENVELOPE_OWNER_NOT_ANU", "ACTUAL_CRON_OWNER_NOT_ANU", "ENVELOPE_3SID_MISMATCH", "SCHEDULE_HISTORY_STATUS_FAIL", "SCHEDULE_HISTORY_SID_MISMATCH", "ENVELOPE_ACTUAL_CRON_OWNER_MISMATCH", } ★ 본 validator 는 PR #152 expected_files 내부에서 분류만 수행한다. 분류 결과 hold/merge 결정은 호출 측에서 회장 verbatim 도구를 통해 수행한다. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Iterable, List, Optional, Sequence VALIDATOR_SCHEMA = "utils.callback_authority_4source_validator.v1" DEFAULT_ANU_KEYS: Sequence[str] = ("c119085addb0f8b7",) # verdict classification 상수 AUTHORITATIVE_4SOURCE_OK = "AUTHORITATIVE_4SOURCE_OK" ENVELOPE_OWNER_NOT_ANU = "ENVELOPE_OWNER_NOT_ANU" ACTUAL_CRON_OWNER_NOT_ANU = "ACTUAL_CRON_OWNER_NOT_ANU" ENVELOPE_3SID_MISMATCH = "ENVELOPE_3SID_MISMATCH" SCHEDULE_HISTORY_STATUS_FAIL = "SCHEDULE_HISTORY_STATUS_FAIL" SCHEDULE_HISTORY_SID_MISMATCH = "SCHEDULE_HISTORY_SID_MISMATCH" ENVELOPE_ACTUAL_CRON_OWNER_MISMATCH = "ENVELOPE_ACTUAL_CRON_OWNER_MISMATCH" @dataclass(frozen=True) class Verdict: schema: str classification: str reasons: List[str] = field(default_factory=list) envelope_owner_key: Optional[str] = None actual_cron_owner_key: Optional[str] = None chair_facing_session_id: Optional[str] = None collector_session_id: Optional[str] = None delivery_session_id: Optional[str] = None schedule_history_status: Optional[str] = None schedule_history_session_id: Optional[str] = None @property def passed(self) -> bool: return self.classification == AUTHORITATIVE_4SOURCE_OK def _is_anu(key: Optional[Any], anu_keys: Iterable[str]) -> bool: if not isinstance(key, str): return False return key.strip() in set(anu_keys) def _get_str(d: Any, key: str) -> Optional[str]: if not isinstance(d, dict): return None v = d.get(key) return v.strip() if isinstance(v, str) and v.strip() else None def validate( *, envelope: Any, actual_cron_owner_key: Optional[str], schedule_history_record: Any = None, anu_keys: Sequence[str] = DEFAULT_ANU_KEYS, ) -> Verdict: """4-source 검증. envelope: ANU normal callback envelope (schemas/anu_normal_callback_envelope_v1.json). 필수 필드: owner_key, chair_facing_session_id, collector_session_id, delivery_session_id. actual_cron_owner_key: cokacdir --cron-history 결과로 관측된 실제 cron channel owner_key (해당 SID 채널에 실제 등록된 key — envelope 텍스트와 무관한 actual evidence). schedule_history_record: schedule_history `.log` 최근 1줄 JSON dict. 필수 필드: status (ok/cancelled/error), chair_facing_session_id (옵션). """ reasons: List[str] = [] env_owner = _get_str(envelope, "owner_key") chair_sid = _get_str(envelope, "chair_facing_session_id") collector_sid = _get_str(envelope, "collector_session_id") delivery_sid = _get_str(envelope, "delivery_session_id") actual_owner = actual_cron_owner_key.strip() if isinstance(actual_cron_owner_key, str) and actual_cron_owner_key.strip() else None sh_status = _get_str(schedule_history_record, "status") sh_sid = _get_str(schedule_history_record, "chair_facing_session_id") common_fields = dict( envelope_owner_key=env_owner, actual_cron_owner_key=actual_owner, chair_facing_session_id=chair_sid, collector_session_id=collector_sid, delivery_session_id=delivery_sid, schedule_history_status=sh_status, schedule_history_session_id=sh_sid, ) # Source 1: envelope owner_key == ANU if not _is_anu(env_owner, anu_keys): reasons.append( f"envelope.owner_key={env_owner!r} ∉ anu_keys — self-key 또는 누락 " "(★ envelope 텍스트만 보고 PASS 금지)." ) return Verdict(VALIDATOR_SCHEMA, ENVELOPE_OWNER_NOT_ANU, reasons, **common_fields) # Source 2: actual cron owner_key == ANU if not _is_anu(actual_owner, anu_keys): reasons.append( f"actual_cron_owner_key={actual_owner!r} ∉ anu_keys — " "cokacdir --cron-history 채널 owner 가 ANU 가 아님 (self-key 등록 또는 미등록)." ) return Verdict(VALIDATOR_SCHEMA, ACTUAL_CRON_OWNER_NOT_ANU, reasons, **common_fields) # envelope owner_key vs actual cron owner_key cross-check if env_owner != actual_owner: reasons.append( f"envelope.owner_key={env_owner!r} != actual_cron_owner_key={actual_owner!r} — " "envelope-actual 양방향 owner 불일치." ) return Verdict(VALIDATOR_SCHEMA, ENVELOPE_ACTUAL_CRON_OWNER_MISMATCH, reasons, **common_fields) # Source 3: envelope 3 SID 모두 동일 sids = [chair_sid, collector_sid, delivery_sid] if not all(sids) or len(set(sids)) != 1: reasons.append( f"envelope 3 SID mismatch: chair={chair_sid!r} collector={collector_sid!r} " f"delivery={delivery_sid!r}" ) return Verdict(VALIDATOR_SCHEMA, ENVELOPE_3SID_MISMATCH, reasons, **common_fields) # Source 4a: schedule_history status == ok if sh_status != "ok": reasons.append( f"schedule_history.status={sh_status!r} != 'ok' — cron 실행이 정상 완료되지 않음." ) return Verdict(VALIDATOR_SCHEMA, SCHEDULE_HISTORY_STATUS_FAIL, reasons, **common_fields) # Source 4b: schedule_history chair_facing_session_id 일치 (필드 존재 시) if sh_sid is not None and sh_sid != chair_sid: reasons.append( f"schedule_history.chair_facing_session_id={sh_sid!r} != " f"envelope.chair_facing_session_id={chair_sid!r}" ) return Verdict(VALIDATOR_SCHEMA, SCHEDULE_HISTORY_SID_MISMATCH, reasons, **common_fields) reasons.append( "4-source 모두 일치: envelope.owner_key=ANU + actual_cron.owner_key=ANU + " "envelope 3 SID 동일 + schedule_history.status=ok (+ SID 일치)." ) return Verdict(VALIDATOR_SCHEMA, AUTHORITATIVE_4SOURCE_OK, reasons, **common_fields) __all__ = [ "VALIDATOR_SCHEMA", "DEFAULT_ANU_KEYS", "AUTHORITATIVE_4SOURCE_OK", "ENVELOPE_OWNER_NOT_ANU", "ACTUAL_CRON_OWNER_NOT_ANU", "ENVELOPE_3SID_MISMATCH", "SCHEDULE_HISTORY_STATUS_FAIL", "SCHEDULE_HISTORY_SID_MISMATCH", "ENVELOPE_ACTUAL_CRON_OWNER_MISMATCH", "Verdict", "validate", ]