"""Fail-closed administrative checks for the DualCoach rehearsal. ``provider-auth check`` uses the production auxiliary-client resolver. Most providers are verified with a read-only ``models.list`` call. Codex OAuth has no supported models endpoint, so its billable Responses probe is unavailable unless an operator explicitly supplies ``--allow-billable-active-probe``. """ from __future__ import annotations import argparse import asyncio import hashlib import json import os import re import stat import tempfile import uuid from dataclasses import dataclass from datetime import UTC, date, datetime from enum import Enum, IntEnum from pathlib import Path from typing import Any, Callable, Mapping, Sequence, cast COMMAND = "dualcoach_admin provider-auth check" COMMAND_VERSION = "v2" PROVIDER_ADAPTER = "agent.auxiliary_client.resolve_provider_client" PROVIDER_ADAPTER_VERSION = "v1" PROBE_TIMEOUT_SECONDS = 10.0 _RECEIPT_SCHEMA = "dualcoach-provider-auth-receipt-v2" _INDEX_SCHEMA = "dualcoach-provider-auth-index-v2" _PROFILE_SNAPSHOT_SCHEMA = "dualcoach-provider-auth-profile-snapshot-v1" _CODEX_PROBE_KIND = "codex_nonpersistent_generation" _CODEX_PROMPT = "ok" _SECRET_KEY_PATTERN = re.compile(r"(?:api[_-]?key|token|secret|password|credential|auth)", re.IGNORECASE) _PROVIDER_PATTERN = re.compile(r"^[a-z0-9][a-z0-9_.:-]{0,127}$") _ENV_REFERENCE_PATTERN = re.compile(r"^[A-Z_][A-Z0-9_]{0,127}$") _PROFILE_SNAPSHOT_FILES = ( "config.yaml", "auth.json", "customers/registry.json", "gateway_state.json", "data/owner-actions/draft-deliveries.json", "data/scheduled-deliveries.jsonl", "data/onboarding/telegram-publication-outbox-v1/ledger.json", ) _CODEX_REQUIRED_REQUEST_FIELDS = frozenset( {"input", "instructions", "model", "store", "stream", "timeout"} ) _CODEX_FORBIDDEN_REQUEST_FIELDS = frozenset( { "tools", "metadata", "previous_response_id", "conversation", "conversation_id", "thread", "thread_id", } ) # These OAuth route slugs were removed from the local Codex fallback catalog # after repeated ChatGPT-account HTTP 400 responses. Rejecting them locally # prevents a known-invalid active probe from consuming its single wire attempt. _CODEX_KNOWN_RETIRED_OAUTH_MODELS = frozenset( { "gpt-5.2-codex", "gpt-5.1-codex-max", "gpt-5.1-codex-mini", } ) _CODEX_MODEL_FAILURE_CODES = frozenset( {"model_not_found", "model_not_supported", "unsupported_model"} ) _SAFE_CODEX_FAILURE_CATEGORIES = frozenset( { "bad_request", "unsupported_parameter", "context_length", "rate_limited", "authentication", "provider_unavailable", "provider_error", } ) _SAFE_CODEX_FAILURE_CODES = frozenset( { "provider_bad_request", "provider_unsupported_parameter", "provider_context_length", "provider_authentication", "provider_rate_limited", "provider_unavailable", "provider_failure", } ) # These production routes do not expose a stable, read-only models-list surface # through the auxiliary client. Treating client construction as auth success # would be an unsafe false positive. _MODEL_LIST_UNSUPPORTED_PROVIDERS = frozenset( { "bedrock", "copilot-acp", "gemini", "google-gemini-cli", "qwen-oauth", } ) class ProviderAuthResult(str, Enum): READY = "ready" CONFIG_MISSING = "config_missing" CONFIG_MALFORMED = "config_malformed" CREDENTIAL_MISSING = "credential_missing" PROVIDER_UNRESOLVED = "provider_unresolved" PROVIDER_UNSUPPORTED = "provider_unsupported" CODEX_MODEL_UNSUPPORTED = "codex_model_unsupported" SECRET_REFERENCE_MALFORMED = "secret_reference_malformed" SECRET_SOURCE_UNAVAILABLE = "secret_source_unavailable" AUTH_REJECTED = "auth_rejected" PROBE_TIMEOUT = "probe_timeout" PROBE_UNKNOWN = "probe_unknown" RECEIPT_WRITE_FAILED = "receipt_write_failed" PROBE_RATE_LIMITED = "probe_rate_limited" ACTIVE_PROBE_MALFORMED = "active_probe_malformed" PROBE_CANCELLED = "probe_cancelled" ACTIVE_PROBE_UNKNOWN = "active_probe_unknown" PROFILE_SNAPSHOT_UNAVAILABLE = "profile_snapshot_unavailable" PROFILE_MUTATED = "profile_mutated" CLIENT_CLOSE_FAILED = "client_close_failed" class ProviderAuthExit(IntEnum): READY = 0 CONFIG_MISSING = 20 CONFIG_MALFORMED = 21 CREDENTIAL_MISSING = 22 PROVIDER_UNRESOLVED = 23 PROVIDER_UNSUPPORTED = 24 CODEX_MODEL_UNSUPPORTED = 38 SECRET_REFERENCE_MALFORMED = 25 SECRET_SOURCE_UNAVAILABLE = 26 AUTH_REJECTED = 27 PROBE_TIMEOUT = 28 PROBE_UNKNOWN = 29 RECEIPT_WRITE_FAILED = 30 PROBE_RATE_LIMITED = 31 ACTIVE_PROBE_MALFORMED = 32 PROBE_CANCELLED = 33 ACTIVE_PROBE_UNKNOWN = 34 PROFILE_SNAPSHOT_UNAVAILABLE = 35 PROFILE_MUTATED = 36 CLIENT_CLOSE_FAILED = 37 _RESULT_EXIT = { ProviderAuthResult.READY: ProviderAuthExit.READY, ProviderAuthResult.CONFIG_MISSING: ProviderAuthExit.CONFIG_MISSING, ProviderAuthResult.CONFIG_MALFORMED: ProviderAuthExit.CONFIG_MALFORMED, ProviderAuthResult.CREDENTIAL_MISSING: ProviderAuthExit.CREDENTIAL_MISSING, ProviderAuthResult.PROVIDER_UNRESOLVED: ProviderAuthExit.PROVIDER_UNRESOLVED, ProviderAuthResult.PROVIDER_UNSUPPORTED: ProviderAuthExit.PROVIDER_UNSUPPORTED, ProviderAuthResult.CODEX_MODEL_UNSUPPORTED: ProviderAuthExit.CODEX_MODEL_UNSUPPORTED, ProviderAuthResult.SECRET_REFERENCE_MALFORMED: ProviderAuthExit.SECRET_REFERENCE_MALFORMED, ProviderAuthResult.SECRET_SOURCE_UNAVAILABLE: ProviderAuthExit.SECRET_SOURCE_UNAVAILABLE, ProviderAuthResult.AUTH_REJECTED: ProviderAuthExit.AUTH_REJECTED, ProviderAuthResult.PROBE_TIMEOUT: ProviderAuthExit.PROBE_TIMEOUT, ProviderAuthResult.PROBE_UNKNOWN: ProviderAuthExit.PROBE_UNKNOWN, ProviderAuthResult.RECEIPT_WRITE_FAILED: ProviderAuthExit.RECEIPT_WRITE_FAILED, ProviderAuthResult.PROBE_RATE_LIMITED: ProviderAuthExit.PROBE_RATE_LIMITED, ProviderAuthResult.ACTIVE_PROBE_MALFORMED: ProviderAuthExit.ACTIVE_PROBE_MALFORMED, ProviderAuthResult.PROBE_CANCELLED: ProviderAuthExit.PROBE_CANCELLED, ProviderAuthResult.ACTIVE_PROBE_UNKNOWN: ProviderAuthExit.ACTIVE_PROBE_UNKNOWN, ProviderAuthResult.PROFILE_SNAPSHOT_UNAVAILABLE: ProviderAuthExit.PROFILE_SNAPSHOT_UNAVAILABLE, ProviderAuthResult.PROFILE_MUTATED: ProviderAuthExit.PROFILE_MUTATED, ProviderAuthResult.CLIENT_CLOSE_FAILED: ProviderAuthExit.CLIENT_CLOSE_FAILED, } @dataclass(frozen=True) class ProviderAuthConfig: provider: str model: str custom_secret_reference: str | None @dataclass(frozen=True) class ProviderAuthReceipt: result: ProviderAuthResult exit: ProviderAuthExit payload: dict[str, object] receipt_path: Path | None index_path: Path | None class _ConfigError(Exception): def __init__(self, result: ProviderAuthResult) -> None: self.result = result def _load_production_config() -> object: from hermes_cli.config import load_config return load_config() def _resolve_provider_client(provider: str, model: str) -> tuple[object | None, str | None]: from agent.auxiliary_client import resolve_provider_client return resolve_provider_client(provider, model) def _load_codex_auth_status() -> object: from hermes_cli.auth import get_codex_auth_status return get_codex_auth_status() def _is_codex_auxiliary_client(client: object) -> bool: from agent.auxiliary_client import CodexAuxiliaryClient return isinstance(client, CodexAuxiliaryClient) def _default_receipt_directory() -> Path: from hermes_constants import get_hermes_home return get_hermes_home() / "data" / "dualcoach-provider-auth" def _profile_snapshot_sha256() -> str | None: """Hash the profile surfaces the probe is forbidden to mutate. The receipt and index directory are intentionally outside this snapshot: those are the only allowed profile writes made by this command. """ try: from hermes_constants import get_hermes_home profile = get_hermes_home() records: list[dict[str, object]] = [] for relative in _PROFILE_SNAPSHOT_FILES: path = profile / relative if not path.exists(): records.append({"path": relative, "absent": True}) continue info = path.lstat() if path.is_symlink() or not path.is_file() or stat.S_IMODE(info.st_mode) != 0o600: return None records.append( { "path": relative, "bytes": info.st_size, "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), } ) encoded = json.dumps( {"schema": _PROFILE_SNAPSHOT_SCHEMA, "files": records}, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() except (ImportError, OSError): return None def _now_utc() -> str: return datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z") def _candidate_digest() -> str: source_digest = hashlib.sha256(Path(__file__).read_bytes()).hexdigest() encoded = json.dumps( { "command": COMMAND, "command_version": COMMAND_VERSION, "source_sha256": source_digest, }, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _redacted_config(value: object, key: str = "") -> object: if _SECRET_KEY_PATTERN.search(key): return "" if isinstance(value, Mapping): return { str(item_key): _redacted_config(item_value, str(item_key)) for item_key, item_value in sorted(value.items(), key=lambda item: str(item[0])) } if isinstance(value, list): return [_redacted_config(item) for item in value] if value is None or isinstance(value, (bool, int, float, str)): return value return "" def _config_digest(config: object) -> str: encoded = json.dumps( _redacted_config(config), ensure_ascii=True, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _normalized_provider_name(value: object) -> str: return value.strip().lower() if isinstance(value, str) else "" def _string_object_mapping(value: object) -> dict[str, object] | None: if not isinstance(value, Mapping): return None result: dict[str, object] = {} for key, item in value.items(): if not isinstance(key, str): return None result[key] = item return result def _matching_custom_provider(config: Mapping[str, object], provider: str) -> dict[str, object] | None: normalized = _normalized_provider_name(provider) bare_name = normalized.removeprefix("custom:") providers = config.get("providers") if providers is not None: providers_mapping = _string_object_mapping(providers) if providers_mapping is None: raise _ConfigError(ProviderAuthResult.CONFIG_MALFORMED) for entry_name, entry in providers_mapping.items(): entry_mapping = _string_object_mapping(entry) if entry_mapping is None: raise _ConfigError(ProviderAuthResult.CONFIG_MALFORMED) entry_names = { _normalized_provider_name(entry_name), _normalized_provider_name(entry_mapping.get("name")), } if normalized in entry_names or bare_name in entry_names: return entry_mapping legacy = config.get("custom_providers") if legacy is not None and not isinstance(legacy, list): raise _ConfigError(ProviderAuthResult.CONFIG_MALFORMED) if isinstance(legacy, list): for entry in legacy: entry_mapping = _string_object_mapping(entry) if entry_mapping is None: raise _ConfigError(ProviderAuthResult.CONFIG_MALFORMED) entry_names = { _normalized_provider_name(entry_mapping.get("name")), _normalized_provider_name(entry_mapping.get("provider_key")), } if normalized in entry_names or bare_name in entry_names: return entry_mapping return None def _strict_config(config: object) -> ProviderAuthConfig: typed_config = _string_object_mapping(config) if typed_config is None: raise _ConfigError(ProviderAuthResult.CONFIG_MALFORMED) coaching = typed_config.get("physique_coach") if coaching is None: raise _ConfigError(ProviderAuthResult.CONFIG_MISSING) coaching_mapping = _string_object_mapping(coaching) if coaching_mapping is None: raise _ConfigError(ProviderAuthResult.CONFIG_MALFORMED) provider = coaching_mapping.get("draft_provider", "openai-codex") model = coaching_mapping.get("draft_model", "gpt-5.6-terra") if not isinstance(provider, str) or not isinstance(model, str): raise _ConfigError(ProviderAuthResult.CONFIG_MALFORMED) provider = provider.strip().lower() model = model.strip() if not provider or not model or not _PROVIDER_PATTERN.fullmatch(provider): raise _ConfigError(ProviderAuthResult.CONFIG_MALFORMED) custom_provider = _matching_custom_provider(typed_config, provider) secret_reference: str | None = None if custom_provider is not None: raw_secret_reference = custom_provider.get("key_env", custom_provider.get("api_key_env")) if raw_secret_reference is not None: if not isinstance(raw_secret_reference, str): raise _ConfigError(ProviderAuthResult.SECRET_REFERENCE_MALFORMED) secret_reference = raw_secret_reference.strip() if not _ENV_REFERENCE_PATTERN.fullmatch(secret_reference): raise _ConfigError(ProviderAuthResult.SECRET_REFERENCE_MALFORMED) if not os.environ.get(secret_reference): raise _ConfigError(ProviderAuthResult.SECRET_SOURCE_UNAVAILABLE) return ProviderAuthConfig(provider=provider, model=model, custom_secret_reference=secret_reference) def _resolve_canonical_provider(provider: str, custom_provider: Mapping[str, object] | None) -> str | None: if custom_provider is not None: return "custom" try: from hermes_cli.auth import AuthError, resolve_provider return resolve_provider(provider) except (AuthError, ValueError, TypeError): return None def _status_code(error: BaseException) -> int | None: direct = getattr(error, "status_code", None) if type(direct) is int: return direct response = getattr(error, "response", None) nested = getattr(response, "status_code", None) if type(nested) is int: return nested audit = _string_object_mapping(getattr(error, "provider_failure_audit", None)) audited = audit.get("http_status") if audit is not None else None return audited if type(audited) is int else None def _is_timeout(error: BaseException) -> bool: return isinstance(error, TimeoutError) or "timeout" in type(error).__name__.lower() def _is_cancelled(error: BaseException) -> bool: return isinstance(error, (KeyboardInterrupt, InterruptedError, asyncio.CancelledError)) or "cancel" in type(error).__name__.lower() def _models_list(client: object) -> Callable[..., object] | None: candidate = getattr(getattr(client, "models", None), "list", None) return candidate if callable(candidate) else None def _probe_models(client: object) -> ProviderAuthResult: models_list = _models_list(client) if models_list is None: return ProviderAuthResult.PROBE_UNKNOWN try: response = models_list(timeout=PROBE_TIMEOUT_SECONDS) except BaseException as error: # SDKs use provider-specific exception classes. if _is_timeout(error): return ProviderAuthResult.PROBE_TIMEOUT if _status_code(error) in {401, 403}: return ProviderAuthResult.AUTH_REJECTED if _status_code(error) == 429: return ProviderAuthResult.PROBE_RATE_LIMITED return ProviderAuthResult.PROBE_UNKNOWN return ProviderAuthResult.READY if isinstance(getattr(response, "data", None), (list, tuple)) else ProviderAuthResult.PROBE_UNKNOWN def _local_codex_auth_result(status: object) -> ProviderAuthResult | None: value = _string_object_mapping(status) if value is None: return ProviderAuthResult.ACTIVE_PROBE_UNKNOWN if value.get("rate_limited") is True: return ProviderAuthResult.PROBE_RATE_LIMITED if "logged_in" not in value or type(value["logged_in"]) is not bool: return ProviderAuthResult.ACTIVE_PROBE_UNKNOWN if value["logged_in"] is False: return ProviderAuthResult.CREDENTIAL_MISSING return None def _is_known_retired_codex_oauth_model(model: str) -> bool: return model.strip().lower().rsplit("/", 1)[-1] in _CODEX_KNOWN_RETIRED_OAUTH_MODELS def _usage_totals(value: object) -> dict[str, int] | None: mapping = _string_object_mapping(value) def field(*names: str) -> object: for name in names: item = mapping.get(name) if mapping is not None else getattr(value, name, None) if item is not None: return item return None input_tokens = field("input_tokens", "prompt_tokens") output_tokens = field("output_tokens", "completion_tokens") total_tokens = field("total_tokens") if ( type(input_tokens) is not int or type(output_tokens) is not int or type(total_tokens) is not int or input_tokens < 0 or output_tokens < 0 or total_tokens < input_tokens + output_tokens ): return None return {"input_tokens": input_tokens, "output_tokens": output_tokens, "total_tokens": total_tokens} def _codex_request_contract_from_audit(audit_value: object) -> dict[str, object] | None: audit = _string_object_mapping(audit_value) if audit is None: return None fields = audit.get("field_names") if not isinstance(fields, list) or not all(isinstance(field, str) for field in fields): return None field_set = frozenset(fields) if field_set != _CODEX_REQUIRED_REQUEST_FIELDS or field_set & _CODEX_FORBIDDEN_REQUEST_FIELDS: return None if audit.get("store_is_false") is not True or audit.get("stream_is_true") is not True: return None return { "store": False, "stream": True, "tools": False, "metadata": False, "previous_response_id": False, "conversation_ids": False, "thread_ids": False, "sdk_max_retries": 0, "request_attempts": 1, } def _codex_request_contract(response: object) -> dict[str, object] | None: return _codex_request_contract_from_audit(getattr(response, "provider_request_audit", None)) def _codex_provider_failure_evidence(error: BaseException) -> dict[str, object] | None: audit = _string_object_mapping(getattr(error, "provider_failure_audit", None)) if audit is None: return None status = audit.get("http_status") category = audit.get("message_category") failure_code = audit.get("failure_code") retryable = audit.get("retryable") provider_error_code = audit.get("provider_error_code") if ( type(status) is not int or not 100 <= status <= 599 or category not in _SAFE_CODEX_FAILURE_CATEGORIES or failure_code not in _SAFE_CODEX_FAILURE_CODES or type(retryable) is not bool or provider_error_code is not None and (not isinstance(provider_error_code, str) or provider_error_code not in _CODEX_MODEL_FAILURE_CODES) ): return None return { "http_status": status, "provider_error_code": provider_error_code, "message_category": category, "failure_code": failure_code, "retryable": retryable, } def _configure_codex_one_shot(client: object) -> bool: """Rebind the resolved Codex adapter to an SDK client with no retries. The OpenAI SDK otherwise retries a transport request twice by default. The active probe must make exactly one wire attempt, so failure to attest the public ``max_retries`` option is itself fail-closed. """ real_client = getattr(client, "_real_client", None) adapter = getattr(getattr(getattr(client, "chat", None), "completions", None), "_client", None) with_options = getattr(real_client, "with_options", None) if adapter is not real_client or not callable(with_options): return False try: one_shot_client = with_options(max_retries=0) except BaseException: return False if getattr(one_shot_client, "max_retries", None) != 0: return False try: setattr(client, "_real_client", one_shot_client) setattr(getattr(getattr(client, "chat"), "completions"), "_client", one_shot_client) except BaseException: return False return True def _active_error_result(error: BaseException) -> ProviderAuthResult: if _is_cancelled(error): return ProviderAuthResult.PROBE_CANCELLED if _is_timeout(error): return ProviderAuthResult.PROBE_TIMEOUT failure = _codex_provider_failure_evidence(error) if failure is not None and ( failure["http_status"] == 400 and failure["provider_error_code"] in _CODEX_MODEL_FAILURE_CODES ): return ProviderAuthResult.CODEX_MODEL_UNSUPPORTED status = _status_code(error) if status in {401, 403}: return ProviderAuthResult.AUTH_REJECTED if status == 429: return ProviderAuthResult.PROBE_RATE_LIMITED if status == 400: return ProviderAuthResult.ACTIVE_PROBE_MALFORMED return ProviderAuthResult.ACTIVE_PROBE_UNKNOWN def _probe_codex_generation(client: object, resolved_model: object) -> tuple[ProviderAuthResult, dict[str, object]]: evidence: dict[str, object] = { "prompt_sha256": hashlib.sha256(_CODEX_PROMPT.encode("utf-8")).hexdigest(), "prompt_length": len(_CODEX_PROMPT), "resolved_model_sha256": ( hashlib.sha256(resolved_model.encode("utf-8")).hexdigest() if isinstance(resolved_model, str) and resolved_model else None ), "response_model_sha256": None, "response_status": None, "usage": None, "request_contract": None, "provider_failure": None, "model_preflight": None, "sdk_max_retries": 0, "request_attempts": 1, } create = getattr(getattr(getattr(client, "chat", None), "completions", None), "create", None) if not callable(create) or evidence["resolved_model_sha256"] is None: return ProviderAuthResult.ACTIVE_PROBE_MALFORMED, evidence try: response = create( messages=[{"role": "user", "content": _CODEX_PROMPT}], stream=True, timeout=PROBE_TIMEOUT_SECONDS, ) except BaseException as error: audit = _string_object_mapping(getattr(error, "provider_failure_audit", None)) if audit is not None: evidence["request_contract"] = _codex_request_contract_from_audit(audit.get("request_audit")) evidence["provider_failure"] = _codex_provider_failure_evidence(error) return _active_error_result(error), evidence response_model = getattr(response, "model", None) response_status = getattr(response, "provider_status", None) if isinstance(response_model, str) and response_model: evidence["response_model_sha256"] = hashlib.sha256(response_model.encode("utf-8")).hexdigest() if isinstance(response_status, str): evidence["response_status"] = response_status usage = _usage_totals(getattr(response, "usage", None)) contract = _codex_request_contract(response) evidence["usage"] = usage evidence["request_contract"] = contract if ( evidence["response_model_sha256"] is None or response_status != "completed" or getattr(response, "provider_terminal_received", None) is not True or getattr(response, "provider_terminal_event_type", None) != "response.completed" or usage is None or contract is None ): return ProviderAuthResult.ACTIVE_PROBE_MALFORMED, evidence return ProviderAuthResult.READY, evidence def _close_client(client: object | None) -> bool: if client is None: return True close = getattr(client, "close", None) if not callable(close): return True try: close() except BaseException: return False return True def _receipt_payload( *, result: ProviderAuthResult, timestamp_utc: str, config_sha256: str, provider: str | None, probe_kind: str, billable: bool, active_probe_authorized: bool, active_evidence: Mapping[str, object], pre_profile_snapshot_sha256: str | None, post_profile_snapshot_sha256: str | None, ) -> dict[str, object]: payload: dict[str, object] = { "schema": _RECEIPT_SCHEMA, "command": COMMAND, "command_version": COMMAND_VERSION, "timestamp_utc": timestamp_utc, "candidate_digest": _candidate_digest(), "config_sha256": config_sha256, "provider_adapter": PROVIDER_ADAPTER, "provider_adapter_version": PROVIDER_ADAPTER_VERSION, "probe_kind": probe_kind, "probe": ( "models.list" if probe_kind == "models_list_read_only" else "chat.completions.create" if probe_kind == _CODEX_PROBE_KIND else None ), "active_probe_authorized": active_probe_authorized, "store": False if probe_kind == _CODEX_PROBE_KIND else None, "billable": billable, "prompt_sha256": active_evidence.get("prompt_sha256"), "prompt_length": active_evidence.get("prompt_length"), "resolved_model_sha256": active_evidence.get("resolved_model_sha256"), "response_model_sha256": active_evidence.get("response_model_sha256"), "response_status": active_evidence.get("response_status"), "usage": active_evidence.get("usage"), "request_contract": active_evidence.get("request_contract"), "provider_failure": active_evidence.get("provider_failure"), "model_preflight": active_evidence.get("model_preflight"), "sdk_max_retries": active_evidence.get("sdk_max_retries"), "request_attempts": active_evidence.get("request_attempts"), "pre_profile_snapshot_sha256": pre_profile_snapshot_sha256, "post_profile_snapshot_sha256": post_profile_snapshot_sha256, "profile_snapshot_unchanged": ( pre_profile_snapshot_sha256 is not None and pre_profile_snapshot_sha256 == post_profile_snapshot_sha256 ), "effects": { "delivery_actions": 0, "registry_mutations": 0, "service_actions": 0, "telegram_actions": 0, }, "result": result.value, "exit_code": int(_RESULT_EXIT[result]), "success": result is ProviderAuthResult.READY, } if provider is not None: payload["configured_provider"] = provider payload["receipt_sha256"] = hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") ).hexdigest() return payload def _atomic_private_json(path: Path, payload: object) -> None: path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) path.parent.chmod(0o700) descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) temporary_path = Path(temporary_name) try: os.fchmod(descriptor, 0o600) with os.fdopen(descriptor, "w", encoding="utf-8") as handle: descriptor = -1 json.dump(payload, handle, sort_keys=True, separators=(",", ":")) handle.write("\n") handle.flush() os.fsync(handle.fileno()) os.replace(temporary_path, path) path.chmod(0o600) finally: if descriptor >= 0: os.close(descriptor) if temporary_path.exists(): temporary_path.unlink() def _write_receipt(receipt_directory: Path, payload: dict[str, object]) -> tuple[Path, Path]: receipts_directory = receipt_directory / "receipts" timestamp = payload["timestamp_utc"] if not isinstance(timestamp, str): raise OSError("receipt timestamp is invalid") receipt_name = f"{timestamp.replace(':', '').replace('-', '')}-{uuid.uuid4().hex}.json" receipt_path = receipts_directory / receipt_name index_path = receipt_directory / "index.json" _atomic_private_json(receipt_path, payload) _atomic_private_json( index_path, { "schema": _INDEX_SCHEMA, "latest_receipt": receipt_name, "latest_receipt_sha256": payload["receipt_sha256"], "candidate_digest": payload["candidate_digest"], }, ) return receipt_path, index_path def provider_auth_check( *, receipt_directory: Path | None = None, timestamp_utc: str | None = None, allow_billable_active_probe: bool = False, ) -> ProviderAuthReceipt: """Resolve and authenticate the configured provider without leaking secrets. Codex has no supported ``models.list`` route. Its one-shot Responses probe is deliberately opt-in because it can bill the authenticated account. """ timestamp = timestamp_utc or _now_utc() raw_config: object = {} provider: str | None = None client: object | None = None result = ProviderAuthResult.PROBE_UNKNOWN probe_kind = "none" billable = False active_probe_authorized = False active_evidence: dict[str, object] = {} pre_snapshot: str | None = None post_snapshot: str | None = None active_profile_check = False client_closed = True try: raw_config = _load_production_config() config_mapping = _string_object_mapping(raw_config) if config_mapping is None: raise _ConfigError(ProviderAuthResult.CONFIG_MALFORMED) checked_config = _strict_config(config_mapping) provider = checked_config.provider custom_provider = _matching_custom_provider(config_mapping, provider) canonical_provider = _resolve_canonical_provider(provider, custom_provider) if canonical_provider is None: result = ProviderAuthResult.PROVIDER_UNRESOLVED elif canonical_provider in _MODEL_LIST_UNSUPPORTED_PROVIDERS: result = ProviderAuthResult.PROVIDER_UNSUPPORTED elif canonical_provider == "openai-codex": if not allow_billable_active_probe: client, _resolved_model = _resolve_provider_client(provider, checked_config.model) result = ProviderAuthResult.CREDENTIAL_MISSING if client is None else ProviderAuthResult.PROBE_UNKNOWN else: probe_kind = _CODEX_PROBE_KIND active_probe_authorized = True active_profile_check = True pre_snapshot = _profile_snapshot_sha256() if pre_snapshot is None: result = ProviderAuthResult.PROFILE_SNAPSHOT_UNAVAILABLE else: local_result = _local_codex_auth_result(_load_codex_auth_status()) if local_result is not None: result = local_result elif _is_known_retired_codex_oauth_model(checked_config.model): active_evidence = { "model_preflight": "known_retired_codex_oauth_model", "request_attempts": 0, "sdk_max_retries": None, } result = ProviderAuthResult.CODEX_MODEL_UNSUPPORTED else: client, resolved_model = _resolve_provider_client(provider, checked_config.model) if client is None: result = ProviderAuthResult.CREDENTIAL_MISSING elif not _is_codex_auxiliary_client(client): result = ProviderAuthResult.ACTIVE_PROBE_UNKNOWN elif not _configure_codex_one_shot(client): result = ProviderAuthResult.ACTIVE_PROBE_UNKNOWN else: billable = True result, active_evidence = _probe_codex_generation(client, resolved_model) else: client, _resolved_model = _resolve_provider_client(provider, checked_config.model) if client is None: result = ProviderAuthResult.CREDENTIAL_MISSING elif _is_codex_auxiliary_client(client): result = ProviderAuthResult.PROBE_UNKNOWN else: probe_kind = "models_list_read_only" result = _probe_models(client) except _ConfigError as error: result = error.result except BaseException: result = ProviderAuthResult.ACTIVE_PROBE_UNKNOWN if active_probe_authorized else ProviderAuthResult.PROBE_UNKNOWN finally: client_closed = _close_client(client) if active_profile_check: post_snapshot = _profile_snapshot_sha256() if pre_snapshot is None or post_snapshot is None: result = ProviderAuthResult.PROFILE_SNAPSHOT_UNAVAILABLE elif pre_snapshot != post_snapshot: result = ProviderAuthResult.PROFILE_MUTATED if not client_closed: result = ProviderAuthResult.CLIENT_CLOSE_FAILED payload = _receipt_payload( result=result, timestamp_utc=timestamp, config_sha256=_config_digest(raw_config), provider=provider, probe_kind=probe_kind, billable=billable, active_probe_authorized=active_probe_authorized, active_evidence=active_evidence, pre_profile_snapshot_sha256=pre_snapshot, post_profile_snapshot_sha256=post_snapshot, ) try: receipt_path, index_path = _write_receipt(receipt_directory or _default_receipt_directory(), payload) except OSError: result = ProviderAuthResult.RECEIPT_WRITE_FAILED payload = _receipt_payload( result=result, timestamp_utc=timestamp, config_sha256=_config_digest(raw_config), provider=provider, probe_kind=probe_kind, billable=billable, active_probe_authorized=active_probe_authorized, active_evidence=active_evidence, pre_profile_snapshot_sha256=pre_snapshot, post_profile_snapshot_sha256=post_snapshot, ) receipt_path = None index_path = None return ProviderAuthReceipt(result=result, exit=_RESULT_EXIT[result], payload=payload, receipt_path=receipt_path, index_path=index_path) async def staff_membership_preflight( *, profile_root: Path, customer_id: str, bootstrap_session_id: str, expected_generation: int, deployment_receipt_path: Path, output_path: Path, bot: object | None = None, bot_api_base_url: str | None = None, phase: str = "pre_activation", now: datetime | None = None, ) -> dict[str, object]: """Perform the canonical fresh Bot API query and write private evidence.""" from gateway.config import Platform, load_gateway_preflight_inputs from .telegram_customer_bootstrap import Role, RoomBootstrapStore, room_bootstrap_state_dir from .telegram_staff_membership_gate import ( MembershipJournal, build_staff_chat_inventory, create_pre_activation_evidence, ) preflight_inputs = load_gateway_preflight_inputs(profile_root) root = preflight_inputs.profile registry_path = root / "customers/registry.json" config_path = root / "config.yaml" registry = json.loads(registry_path.read_text(encoding="utf-8")) if not isinstance(registry, dict): raise ValueError("staff membership configuration is invalid") telegram_config = preflight_inputs.config.platforms.get(Platform.TELEGRAM) if telegram_config is None: raise ValueError("staff membership configuration is invalid") inventory = build_staff_chat_inventory( registry, {"platforms": {"telegram": telegram_config.to_dict()}}, ) session = RoomBootstrapStore(room_bootstrap_state_dir(root)).get( bootstrap_session_id ) claim = session.role_claim(Role.CUSTOMER) if ( session.customer_key != customer_id or session.generation != expected_generation or claim is None or not str(claim.user_id).isdigit() ): raise ValueError("staff membership bootstrap binding is stale") journal = MembershipJournal( root / "data/onboarding/telegram-staff-membership-v1/events.jsonl" ) armed_rows = [row for row in journal.verify() if row.get("event") == "subscription_armed"] if not armed_rows: raise ValueError("staff membership subscription is not armed") armed = armed_rows[-1] if armed.get("staff_chat_inventory_sha256") != inventory.sha256: raise ValueError("staff membership subscription inventory is stale") if bot is None: from telegram import Bot from .telegram_production_preflight import validate_loopback_api_overrides if bot_api_base_url is None: bot = Bot(preflight_inputs.telegram.secret) else: validate_loopback_api_overrides({"base_url": bot_api_base_url}) bot = Bot( preflight_inputs.telegram.secret, base_url=bot_api_base_url, ) return await create_pre_activation_evidence( cast(Any, bot), inventory, output_path=output_path, deployment_receipt_path=deployment_receipt_path, registry_path=registry_path, config_path=config_path, customer_id=customer_id, customer_user_id=int(claim.user_id), bootstrap_session_id=session.session_id, bootstrap_generation=session.generation, subscription_epoch_id=str(armed["subscription_epoch_id"]), subscription_armed_at_utc=str(armed["observed_at_utc"]), phase=phase, now=now, ) def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="dualcoach_admin") commands = parser.add_subparsers(dest="command", required=True) provider_auth = commands.add_parser("provider-auth") provider_auth_commands = provider_auth.add_subparsers(dest="provider_auth_command", required=True) check = provider_auth_commands.add_parser("check") check.add_argument("--json", action="store_true", required=True) check.add_argument("--receipt-dir", type=Path) check.add_argument("--allow-billable-active-probe", action="store_true") customer = commands.add_parser("customer") customer_commands = customer.add_subparsers(dest="customer_command", required=True) activate = customer_commands.add_parser("activate") activate.add_argument("--profile-root", type=Path, required=True) activate.add_argument("--data-root", type=Path, required=True) activate.add_argument("--customer-id", required=True) activate.add_argument("--checklist-evidence", type=Path, required=True) activate.add_argument("--staff-membership-evidence", type=Path, required=True) activate.add_argument("--deployment-receipt", type=Path, required=True) activate.add_argument("--bootstrap-session", required=True) activate.add_argument("--expected-generation", type=int, required=True) activate.add_argument("--package-root", type=Path) activate.add_argument("--telegram-api-base-url") activate.add_argument("--kst-date", type=date.fromisoformat) activate.add_argument( "--task26-authority-pin", type=Path, required=True, help="absolute private external task26-authority-pin-v1 credential", ) activate.add_argument( "--task26-candidate-digest", required=True, help="exact currently qualified 64-hex Task26 candidate digest", ) for name in ( "staff-membership-preflight", "staff-membership-finalize", "staff-membership-verify", ): membership = customer_commands.add_parser(name) membership.add_argument("--profile-root", type=Path, required=True) membership.add_argument("--customer-id", required=True) membership.add_argument("--bootstrap-session", required=True) membership.add_argument("--expected-generation", type=int, required=True) membership.add_argument("--deployment-receipt", type=Path, required=True) membership.add_argument("--staff-membership-evidence", type=Path, required=True) membership.add_argument("--telegram-api-base-url") membership.add_argument("--json", action="store_true", required=True) return parser def main(argv: Sequence[str] | None = None) -> int: args = _parser().parse_args(argv) command = cast(str, args.command) if command == "customer" and cast(str, args.customer_command) == "activate": from .dualcoach_activation_cutover import activate_customer_cutover from .task26_runtime_authority import FileCandidateAuthoritySource profile_root = cast(Path, args.profile_root) package_root = cast(Path | None, args.package_root) candidate_digest = cast(str, args.task26_candidate_digest) authority_source = FileCandidateAuthoritySource( cast(Path, args.task26_authority_pin), forbidden_roots=( profile_root, cast(Path, args.data_root), package_root if package_root is not None else profile_root / "workspace", cast(Path, args.checklist_evidence), cast(Path, args.staff_membership_evidence), cast(Path, args.deployment_receipt), ), ) with authority_source.authorize(candidate_digest, "activation") as predecessor: pass _ = asyncio.run( staff_membership_preflight( profile_root=cast(Path, args.profile_root), customer_id=cast(str, args.customer_id), bootstrap_session_id=cast(str, args.bootstrap_session), expected_generation=cast(int, args.expected_generation), deployment_receipt_path=cast(Path, args.deployment_receipt), output_path=cast(Path, args.staff_membership_evidence), bot_api_base_url=cast(str | None, args.telegram_api_base_url), ) ) result = activate_customer_cutover( profile_root, cast(Path, args.data_root), cast(str, args.customer_id), cast(Path, args.checklist_evidence), cast(Path, args.staff_membership_evidence), bootstrap_session_id=cast(str, args.bootstrap_session), expected_generation=cast(int, args.expected_generation), deployment_receipt_path=cast(Path, args.deployment_receipt), package_root=package_root, kst_date=cast(date | None, args.kst_date), task26_authority_source=authority_source, task26_candidate_digest=candidate_digest, task26_authority_predecessor=predecessor, task26_runtime_required=True, ) print(json.dumps(result.to_dict(), sort_keys=True, separators=(",", ":"))) return 0 if command == "customer" and cast(str, args.customer_command) in { "staff-membership-preflight", "staff-membership-finalize", }: membership_phase = ( "post_lifecycle" if cast(str, args.customer_command) == "staff-membership-finalize" else "pre_activation" ) payload = asyncio.run( staff_membership_preflight( profile_root=cast(Path, args.profile_root), customer_id=cast(str, args.customer_id), bootstrap_session_id=cast(str, args.bootstrap_session), expected_generation=cast(int, args.expected_generation), deployment_receipt_path=cast(Path, args.deployment_receipt), output_path=cast(Path, args.staff_membership_evidence), bot_api_base_url=cast(str | None, args.telegram_api_base_url), phase=membership_phase, ) ) print(json.dumps({ "evidence_sha256": payload["evidence_sha256"], "inventory_sha256": payload["staff_chat_inventory_sha256"], "phase": payload["phase"], "subscription_epoch_id": payload["subscription_epoch_id"], }, sort_keys=True, separators=(",", ":"))) return 0 if command == "customer" and cast(str, args.customer_command) == "staff-membership-verify": from .telegram_staff_membership_gate import ( ABSENT_STATUSES, MembershipJournal, sha256_json, ) path = cast(Path, args.staff_membership_evidence) payload = json.loads(path.read_text(encoding="utf-8")) body = {key: value for key, value in payload.items() if key != "evidence_sha256"} inventory = payload.get("staff_chat_inventory") membership_results = payload.get("membership_results") private_results = payload.get("private_dm_results") if ( payload.get("evidence_sha256") != sha256_json(body) or not isinstance(inventory, list) or payload.get("staff_chat_inventory_sha256") != sha256_json(inventory) or not isinstance(membership_results, list) or any( not isinstance(row, dict) or row.get("status") not in ABSENT_STATUSES for row in membership_results ) or not isinstance(private_results, list) or any( not isinstance(row, dict) or row.get("identity_separated") is not True for row in private_results ) ): raise ValueError("staff membership evidence digest is invalid") journal = MembershipJournal( cast(Path, args.profile_root) / "data/onboarding/telegram-staff-membership-v1/events.jsonl" ) rows = journal.verify() if not any( row.get("event") == "subscription_armed" and row.get("subscription_epoch_id") == payload.get("subscription_epoch_id") for row in rows ): raise ValueError("staff membership subscription continuity is invalid") if payload.get("phase") == "post_lifecycle": activation_path = cast(Path, args.profile_root) / "data/customer-activation-journal.json" activation_digest = hashlib.sha256(activation_path.read_bytes()).hexdigest() if not any( row.get("event") == "activation_binding" and row.get("customer_id") == payload.get("customer_id") and row.get("activation_journal_sha256") == activation_digest for row in rows ): raise ValueError("staff membership activation continuity is invalid") print(json.dumps({"evidence_sha256": payload["evidence_sha256"], "status": "verified"}, sort_keys=True, separators=(",", ":"))) return 0 if command == "provider-auth" and cast(str, args.provider_auth_command) == "check": receipt = provider_auth_check( receipt_directory=cast(Path | None, args.receipt_dir), allow_billable_active_probe=cast( bool, args.allow_billable_active_probe ), ) print(json.dumps(receipt.payload, sort_keys=True, separators=(",", ":"))) return int(receipt.exit) return int(ProviderAuthExit.PROBE_UNKNOWN) if __name__ == "__main__": raise SystemExit(main())