# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR V1 — TELEMETRY CONTRACT & INTEGRITY AUDITOR (AR-84 TRUST BOUNDARY) Defines authentic telemetry verification for the self-repair cycle. Strict Invariants: 1. No fabricated telemetry: provider_called=False or network_used=False cannot be inferred by default without physical, verified observation across both HTTP and raw socket layers. 2. Trinary status: - TRUE: Outbound provider or network traffic was detected. - FALSE_VERIFIED: Observer was ACTIVE, covered required scopes (including RAW_SOCKET), and verified 0 events. - UNKNOWN: Observer inactive, scope unmonitored, or observer failed. 3. Promotion Gate: - Only FALSE_VERIFIED backed by a cryptographically authentic supervisor signature allows auto-promotion. - UNKNOWN and TRUE strictly block auto-promotion. 4. Cryptographic Provenance & Anti-Forgery (AR-84): - Telemetry signatures use HMAC-SHA256 with a supervisor secret kept outside candidate reach. - Signed payload strictly binds: mission_id, instance_id, process_id, workspace_root, interval_start, interval_end, monitored_scopes, detected_attempts_count, observer_active, provider_status, network_status, nonce. - Replay protection: Used nonces/signatures cannot be replayed. - Scope integrity: Scope subsets missing RAW_SOCKET or URLLIB_HTTP invalidate authenticity. """ from __future__ import annotations import copy import hashlib import hmac import json import logging import os import secrets import threading from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Set logger = logging.getLogger("teitan.self_repair.telemetry_contract") def _utc_iso() -> str: return datetime.now(timezone.utc).isoformat() # Scope coverage channels SCOPE_URLLIB_HTTP = "URLLIB_HTTP" SCOPE_RAW_SOCKET = "RAW_SOCKET" SCOPE_HTTP_CLIENT = "HTTP_CLIENT" SCOPE_PEP578_AUDIT_HOOK = "PEP578_AUDIT_HOOK" SCOPE_HTTPX_REQUESTS = "HTTPX_REQUESTS" SCOPE_SUBPROCESS_DESCENDANTS = "SUBPROCESS_DESCENDANTS" # AR-74 / AR-84: Mandatory scopes require both URLLIB_HTTP and RAW_SOCKET MANDATORY_PROVIDER_SCOPES: Set[str] = { SCOPE_URLLIB_HTTP, SCOPE_RAW_SOCKET, } COMPREHENSIVE_NETWORK_SCOPES: Set[str] = { SCOPE_URLLIB_HTTP, SCOPE_RAW_SOCKET, SCOPE_HTTP_CLIENT, SCOPE_PEP578_AUDIT_HOOK, SCOPE_HTTPX_REQUESTS, SCOPE_SUBPROCESS_DESCENDANTS, } # --- AR-84 SUPERVISOR KEY AUTHORITY (Isolated from candidate reach) --- _SUPERVISOR_SECRET_ENV_VAR = "TEITAN_SUPERVISOR_TELEMETRY_SECRET" _INTERNAL_SUPERVISOR_KEY: bytes = os.environ.get(_SUPERVISOR_SECRET_ENV_VAR, "").encode("utf-8") if not _INTERNAL_SUPERVISOR_KEY: _INTERNAL_SUPERVISOR_KEY = secrets.token_bytes(32) _USED_TELEMETRY_NONCES: Set[str] = set() _NONCES_LOCK = threading.Lock() def get_supervisor_telemetry_key() -> bytes: """Returns the trusted supervisor telemetry key.""" return _INTERNAL_SUPERVISOR_KEY def set_supervisor_telemetry_key(key: bytes) -> None: """Configures trusted supervisor telemetry key (called only by supervisor setup).""" global _INTERNAL_SUPERVISOR_KEY _INTERNAL_SUPERVISOR_KEY = key def reset_consumed_nonces() -> None: """Resets the replay protection nonce ledger (for clean lab test sessions).""" with _NONCES_LOCK: _USED_TELEMETRY_NONCES.clear() @dataclass class BaselineIntegrityReport: """Verifiable baseline hash comparison report.""" hashes_before: Dict[str, str] hashes_after: Dict[str, str] status: str # "VERIFIED_IDENTICAL", "INTEGRITY_VIOLATION", "NO_EVIDENCE" all_baselines_identical: bool missing_baselines: List[str] = field(default_factory=list) mismatched_baselines: List[str] = field(default_factory=list) evaluated_at: str = field(default_factory=_utc_iso) @classmethod def evaluate( cls, hashes_before: Optional[Dict[str, str]], hashes_after: Optional[Dict[str, str]], required_baselines: Optional[List[str]] = None, ) -> BaselineIntegrityReport: b_before = dict(hashes_before or {}) b_after = dict(hashes_after or {}) if not b_before or not b_after: return cls( hashes_before=b_before, hashes_after=b_after, status="NO_EVIDENCE", all_baselines_identical=False, missing_baselines=list(required_baselines or []), ) missing = [] if required_baselines: for req in required_baselines: if req not in b_before or req not in b_after: missing.append(req) if missing: return cls( hashes_before=b_before, hashes_after=b_after, status="NO_EVIDENCE", all_baselines_identical=False, missing_baselines=missing, ) mismatches = [] all_keys = set(b_before.keys()).union(set(b_after.keys())) for k in sorted(all_keys): if b_before.get(k) != b_after.get(k): mismatches.append(k) if mismatches: return cls( hashes_before=b_before, hashes_after=b_after, status="INTEGRITY_VIOLATION", all_baselines_identical=False, mismatched_baselines=mismatches, ) return cls( hashes_before=b_before, hashes_after=b_after, status="VERIFIED_IDENTICAL", all_baselines_identical=True, ) def to_dict(self) -> Dict[str, Any]: return asdict(self) @dataclass class TelemetryAuditRecord: """ Verifiable telemetry record cryptographically authenticated by the supervisor. Strictly binds mission, process, workspace root, interval, scopes, and observer status. """ mission_id: str instance_id: str workspace_root: str interval_start: str interval_end: str observer_active: bool observer_fault: Optional[str] monitored_scopes: List[str] unmonitored_scopes: List[str] detected_attempts: List[Dict[str, Any]] detected_attempts_count: int provider_status: str # "TRUE", "FALSE_VERIFIED", "UNKNOWN" network_status: str # "TRUE", "FALSE_VERIFIED", "UNKNOWN" promotion_eligible: bool audit_verdict: str # "PASS", "FAIL", "BLOCKED" process_id: int = field(default_factory=os.getpid) nonce: str = field(default_factory=lambda: secrets.token_hex(16)) evidence_signature: str = "" def canonical_payload(self) -> str: """Constructs canonical payload representation for HMAC verification.""" norm_ws = str(self.workspace_root).replace("\\", "/").rstrip("/").lower() sorted_scopes = ",".join(sorted(self.monitored_scopes or [])) return ( f"v2:{self.mission_id}:{self.instance_id}:{self.process_id}:{norm_ws}:" f"{self.interval_start}:{self.interval_end}:{sorted_scopes}:" f"{self.detected_attempts_count}:{self.observer_active}:" f"{self.provider_status}:{self.network_status}:{self.nonce}" ) def compute_hmac_signature(self, key: Optional[bytes] = None) -> str: """Computes HMAC-SHA256 signature using the supervisor authority key.""" signing_key = key or _INTERNAL_SUPERVISOR_KEY payload_bytes = self.canonical_payload().encode("utf-8") return hmac.new(signing_key, payload_bytes, hashlib.sha256).hexdigest().upper() def verify_authenticity( self, expected_mission_id: Optional[str] = None, expected_workspace_root: Optional[str] = None, key: Optional[bytes] = None, enforce_replay_guard: bool = True, ) -> Tuple[bool, Optional[str]]: """ Validates cryptographic provenance, scope coverage, workspace binding, and replay guards. Returns (is_valid, failure_reason). """ if not self.evidence_signature: return False, "SIGNATURE_MISSING" # 1. Scope coverage validation req_scopes = MANDATORY_PROVIDER_SCOPES mon_scopes = set(self.monitored_scopes or []) if not req_scopes.issubset(mon_scopes): return False, "INCOMPLETE_SCOPE_COVERAGE" # 2. Observer active validation if not self.observer_active: return False, "OBSERVER_INACTIVE" if self.observer_fault: return False, f"OBSERVER_FAULT: {self.observer_fault}" # 3. Mission binding validation if expected_mission_id and self.mission_id != expected_mission_id: return False, f"MISSION_ID_MISMATCH: expected {expected_mission_id}, got {self.mission_id}" # 4. Workspace root binding validation if expected_workspace_root: exp_norm = str(expected_workspace_root).replace("\\", "/").rstrip("/").lower() rec_norm = str(self.workspace_root).replace("\\", "/").rstrip("/").lower() if exp_norm != rec_norm: return False, f"WORKSPACE_ROOT_MISMATCH: expected {exp_norm}, got {rec_norm}" # 5. Cryptographic HMAC validation verification_key = key or _INTERNAL_SUPERVISOR_KEY expected_sig = self.compute_hmac_signature(verification_key) if not hmac.compare_digest(self.evidence_signature, expected_sig): return False, "HMAC_SIGNATURE_MISMATCH" # 6. Replay protection if enforce_replay_guard: with _NONCES_LOCK: if self.nonce in _USED_TELEMETRY_NONCES or self.evidence_signature in _USED_TELEMETRY_NONCES: return False, "REPLAYED_TELEMETRY_RECORD" _USED_TELEMETRY_NONCES.add(self.nonce) _USED_TELEMETRY_NONCES.add(self.evidence_signature) return True, None def to_dict(self) -> Dict[str, Any]: d = asdict(self) if not self.evidence_signature: self.evidence_signature = self.compute_hmac_signature() d["evidence_signature"] = self.evidence_signature return d class AuthenticTelemetryEvaluator: """ Evaluates telemetry gathered during a repair execution without fabricating success. Enforces that 'FALSE_VERIFIED' requires verified active observation over required scopes and signs records using supervisor HMAC authority. """ @staticmethod def evaluate_telemetry( mission_id: str, instance_id: str, workspace_root: str, observer_active: bool, monitored_scopes: Optional[List[str]] = None, detected_attempts: Optional[List[Dict[str, Any]]] = None, observer_fault: Optional[str] = None, interval_start: Optional[str] = None, interval_end: Optional[str] = None, required_scopes: Optional[Set[str]] = None, process_id: Optional[int] = None, nonce: Optional[str] = None, supervisor_key: Optional[bytes] = None, ) -> TelemetryAuditRecord: start_ts = interval_start or _utc_iso() end_ts = interval_end or _utc_iso() scopes = list(monitored_scopes or []) attempts = list(detected_attempts or []) req_scopes = required_scopes if required_scopes is not None else MANDATORY_PROVIDER_SCOPES unmonitored = [s for s in sorted(req_scopes) if s not in scopes] # 1. Check if calls detected if len(attempts) > 0: provider_status = "TRUE" network_status = "TRUE" promotion_eligible = False audit_verdict = "FAIL" # 2. If observer was inactive or experienced a fault or missed required scope elif not observer_active or observer_fault or unmonitored: provider_status = "UNKNOWN" network_status = "UNKNOWN" promotion_eligible = False audit_verdict = "BLOCKED" # 3. Verified zero calls under active, full observation else: provider_status = "FALSE_VERIFIED" network_status = "FALSE_VERIFIED" promotion_eligible = True audit_verdict = "PASS" record = TelemetryAuditRecord( mission_id=mission_id, instance_id=instance_id, workspace_root=workspace_root, interval_start=start_ts, interval_end=end_ts, observer_active=observer_active, observer_fault=observer_fault, monitored_scopes=scopes, unmonitored_scopes=unmonitored, detected_attempts=attempts, detected_attempts_count=len(attempts), provider_status=provider_status, network_status=network_status, promotion_eligible=promotion_eligible, audit_verdict=audit_verdict, process_id=process_id or os.getpid(), nonce=nonce or secrets.token_hex(16), ) record.evidence_signature = record.compute_hmac_signature(supervisor_key) return record