# -*- coding: utf-8 -*- """anu_v3.callback_owner_validator — callback owner/key/role runtime validator 결선 (task-2553+49 AUTHORITATIVE §4.1 / §5.A / §5.C). 회장 §3/§4: "callback owner/key/role runtime validation" 을 실제 runtime path 에 결선한다. narrow +49 는 ``dispatch.callback_owner_enforcer`` 를 standalone 으로만 두었다. 본 모듈은 그 검증 로직을 **anu_v3 runtime 계층**에 결선하는 어댑터이자, callback registration helper 가 등록 직전 **반드시 경유**해야 하는 fail-closed validator 다 (§5.C: "normal/fallback callback 등록 직전 owner/key /role 검증 / mismatch 면 등록 안 함"). reuse, not re-implement: 실제 검증 규칙은 비-frozen ``dispatch.callback_owner_enforcer`` (narrow +49 산출, byte-0 carve-out 결정으로 non-frozen) 를 단일 진실원으로 호출한다. 규칙 중복 = drift 위험이므로 금지. Layer A / NO-CRON: 순수 검증. ZERO cron / dispatch / subprocess / cokacdir. """ from __future__ import annotations from dataclasses import dataclass, field from typing import List, Optional, Sequence from dispatch.callback_owner_enforcer import ( DEFAULT_ANU_KEYS, FAIL, HOLD, PASS, enforce_callback_owner, is_anu_key, ) VALIDATOR_SCHEMA = "anu_v3.callback_owner_validation.v1" CALLBACK_OWNER_MISMATCH = "CALLBACK_OWNER_MISMATCH" CALLBACK_COLLECTOR_NOT_ANU = "CALLBACK_COLLECTOR_NOT_ANU" SELF_COLLECTOR_FORBIDDEN = "SELF_COLLECTOR_FORBIDDEN" CALLBACK_4TUPLE_INVALID = "CALLBACK_4TUPLE_INVALID" DISPATCH_PATH_BYPASSED_CONTRACT = "DISPATCH_PATH_BYPASSED_CONTRACT" CALLBACK_REGISTRATION_BLOCKED = "CALLBACK_REGISTRATION_BLOCKED" @dataclass class CallbackOwnerValidationResult: schema: str verdict: str # PASS | FAIL | HOLD_FOR_CHAIR classifications: List[str] registration_allowed: bool owner_is_independent_anu: bool task_id: str executor_key: str collector_key: str collector_role: str enforcement: dict reasons: List[str] = field(default_factory=list) @property def ok(self) -> bool: return self.verdict == PASS @property def primary_classification(self) -> Optional[str]: return self.classifications[0] if self.classifications else None def to_json(self) -> dict: return { "schema": self.schema, "verdict": self.verdict, "classifications": list(self.classifications), "primary_classification": self.primary_classification, "registration_allowed": self.registration_allowed, "owner_is_independent_anu": self.owner_is_independent_anu, "task_id": self.task_id, "executor_key": self.executor_key, "collector_key": self.collector_key, "collector_role": self.collector_role, "enforcement": self.enforcement, "reasons": list(self.reasons), } def validate_callback_owner_runtime( *, task_id: str, executor_key: str, collector_key: str, collector_owner_key: Optional[str] = None, collector_role: str, normal_collector_cron_id: Optional[str], fallback_callback_cron_id: Optional[str], dispatch_cron_id: str, chat_id: str = "", prompt_claims_anu_collector: bool = False, entry_path: str = "cokacdir_cron_direct", anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), no_fallback: bool = False, anu_keys_resolvable: bool = True, ) -> CallbackOwnerValidationResult: """Runtime fail-closed owner/key/role validator (§5.A/§5.C). Delegates the rule set to ``enforce_callback_owner`` (single source of truth) and maps it to a *registration decision*: PASS -> registration allowed; FAIL -> registration blocked (mismatch -> 등록 안 함, §5.C); HOLD -> blocked + escalate (§11). The prompt-text claim is NEVER authoritative — owner/key/role identity is (§5.A/regression 4). """ enf = enforce_callback_owner( task_id=task_id, executor_key=executor_key, collector_key=collector_key, collector_owner_key=collector_owner_key, collector_role=collector_role, normal_collector_cron_id=normal_collector_cron_id, fallback_callback_cron_id=fallback_callback_cron_id, dispatch_cron_id=dispatch_cron_id, chat_id=chat_id, prompt_claims_anu_collector=prompt_claims_anu_collector, entry_path=entry_path, anu_keys=anu_keys, no_fallback=no_fallback, anu_keys_resolvable=anu_keys_resolvable, ) registration_allowed = enf.verdict == PASS reasons = list(enf.reasons) if not registration_allowed: reasons.insert( 0, "callback registration BLOCKED at the runtime validator — " "owner/key/role did not fail-closed validate; the registration " "helper MUST NOT register this callback (§5.C / regression 23).", ) return CallbackOwnerValidationResult( schema=VALIDATOR_SCHEMA, verdict=enf.verdict, classifications=list(enf.classifications), registration_allowed=registration_allowed, owner_is_independent_anu=enf.owner_is_independent_anu, task_id=task_id, executor_key=executor_key, collector_key=collector_key, collector_role=collector_role, enforcement=enf.to_json(), reasons=reasons, ) def assert_registration_permitted( result: CallbackOwnerValidationResult, ) -> None: """Hard gate used by the registration helper just before cron register. Raises ``CallbackRegistrationBlocked`` unless the validator PASSed. This is the structural "mismatch -> 등록 안 함" enforcement point (§5.C): a caller that ignores the return value still cannot register a self-owned callback because this raises (fail-closed). """ if result.verdict != PASS: raise CallbackRegistrationBlocked( f"callback registration blocked: verdict={result.verdict} " f"classifications={result.classifications} " f"({result.primary_classification or CALLBACK_REGISTRATION_BLOCKED})" ) class CallbackRegistrationBlocked(RuntimeError): """Raised by ``assert_registration_permitted`` on a non-PASS verdict.""" __all__ = [ "VALIDATOR_SCHEMA", "PASS", "FAIL", "HOLD", "CALLBACK_OWNER_MISMATCH", "CALLBACK_COLLECTOR_NOT_ANU", "SELF_COLLECTOR_FORBIDDEN", "CALLBACK_4TUPLE_INVALID", "DISPATCH_PATH_BYPASSED_CONTRACT", "CALLBACK_REGISTRATION_BLOCKED", "is_anu_key", "CallbackOwnerValidationResult", "validate_callback_owner_runtime", "assert_registration_permitted", "CallbackRegistrationBlocked", ]