# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR V1 — TELEMETRY CONTRACT & INTEGRITY AUDITOR (AR-74 HARDENED) 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 allows automated promotion. - UNKNOWN and TRUE strictly block auto-promotion. 4. Truthful Baseline Evidence: - Differentiates VERIFIED_IDENTICAL, INTEGRITY_VIOLATION, and NO_EVIDENCE. - Missing baselines evaluate to UNKNOWN/False, NEVER to True. """ from __future__ import annotations import hashlib import json import logging 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: Mandatory scopes require both URLLIB_HTTP and RAW_SOCKET (independent of a simple urllib patch) 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, } @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 linked to a specific repair cycle.""" 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" evidence_signature: str = "" def to_dict(self) -> Dict[str, Any]: d = asdict(self) if not self.evidence_signature: d["evidence_signature"] = self.compute_signature() return d def compute_signature(self) -> str: payload = ( f"{self.mission_id}:{self.instance_id}:{self.provider_status}:" f"{self.network_status}:{self.detected_attempts_count}:{self.observer_active}" ) return hashlib.sha256(payload.encode("utf-8")).hexdigest().upper() class AuthenticTelemetryEvaluator: """ Evaluates telemetry gathered during a repair execution without fabricating success. Enforces that 'FALSE_VERIFIED' requires verified active observation over required scopes. """ @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, ) -> 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, ) record.evidence_signature = record.compute_signature() return record