# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR V1 — PHASE 4C: PROVIDER SAFETY FENCE (AR-74 HARDENED) Provides a multi-layer verifiable provider & network safety fence backed by: 1. Environment fence flag (TEITAN_E2E_DISABLE_PROVIDER=1) & temporary credential env scrubbing 2. Raw socket interception (socket.socket.connect, connect_ex, create_connection) 3. HTTP client interception (http.client.HTTPConnection.connect, HTTPSConnection.connect, urllib.request.urlopen) 4. CPython PEP 578 C-runtime audit hook (sys.addaudithook) catching low-level C socket.connect, socket.sendto, urllib.Request, and .dpapi credential file reads even if Python functions were cached prior to fence entry. Strict Invariants: PROVIDER_ALLOWED = False SELF_REPAIR_PROVIDER = None AUTO_PROMOTION = False PRODUCTION_WRITE_AUTHORITY = False MAX_REPAIR_ATTEMPTS = 1 PROVIDER_CALL_ATTEMPTS = 0 NETWORK_PROVIDER_CALLS = 0 """ from __future__ import annotations import http.client import json import logging import os import socket import sys import threading import time import urllib.request from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Set logger = logging.getLogger("teitan.self_repair.safety_fence") TEITAN_E2E_DISABLE_PROVIDER_ENV = "TEITAN_E2E_DISABLE_PROVIDER" # Global Invariants PROVIDER_ALLOWED: bool = False SELF_REPAIR_PROVIDER: Optional[str] = None AUTO_PROMOTION: bool = False PRODUCTION_WRITE_AUTHORITY: bool = False MAX_REPAIR_ATTEMPTS: int = 1 BRAIN_STARTED: bool = False DEVELOPER_STARTED: bool = False NETWORK_USED: bool = False PROVIDER_CALLED: bool = False EVIDENCE_BASE_DIR = r"C:\OptimusProjects\CreatorE2E_Evidence" SENSITIVE_CREDENTIAL_ENV_VARS = ( "OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GEMINI_API_KEY", "GOOGLE_API_KEY", "OPTIMUS_API_KEY", "OPENAI_ORG_ID", ) def is_provider_fence_active() -> bool: """Returns True if the provider safety fence environment flag is active.""" return os.environ.get(TEITAN_E2E_DISABLE_PROVIDER_ENV) == "1" def get_controlled_failure_spec() -> Dict[str, Any]: """ Returns the canonical, deterministic failure specification. """ return { "instruction": "modifica el archivo nonexistent_module.py y cambia version a 2.0", "expected_status": "VERIFICATION_FAILED", "expected_task_status": "VERIFICATION_FAILED", "expected_canonical_status": "VERIFICATION_FAILED", "requires_provider": False, "requires_browser": False, "requires_network": False, "destructive_write": False, "production_mutation": False, "failure_reason_pattern": "nonexistent_module.py", } try: from app.remediation.self_repair.telemetry_contract import ( AuthenticTelemetryEvaluator, BaselineIntegrityReport, SCOPE_HTTP_CLIENT, SCOPE_PEP578_AUDIT_HOOK, SCOPE_RAW_SOCKET, SCOPE_URLLIB_HTTP, ) except (ImportError, ModuleNotFoundError): from .telemetry_contract import ( AuthenticTelemetryEvaluator, BaselineIntegrityReport, SCOPE_HTTP_CLIENT, SCOPE_PEP578_AUDIT_HOOK, SCOPE_RAW_SOCKET, SCOPE_URLLIB_HTTP, ) # Process-wide PEP 578 audit hook registry for active ProviderSafetyFence instances _ACTIVE_FENCES_LOCK = threading.Lock() _ACTIVE_FENCES: List["ProviderSafetyFence"] = [] _GLOBAL_AUDIT_HOOK_INSTALLED: bool = False _IN_GUARD_THREAD_LOCAL = threading.local() def _global_fence_audit_hook(event: str, args: Any) -> None: """ C-runtime PEP 578 audit hook that intercepts low-level socket connections, datagram sends, urllib requests, and .dpapi credential reads whenever any ProviderSafetyFence is active, even if the caller cached socket.socket.connect. """ if getattr(_IN_GUARD_THREAD_LOCAL, "handling", False): return if not _ACTIVE_FENCES: return if event in ("socket.connect", "socket.sendto"): addr = args[1] if len(args) > 1 else "unknown_address" target_str = str(addr) with _ACTIVE_FENCES_LOCK: fences = list(_ACTIVE_FENCES) if fences: _IN_GUARD_THREAD_LOCAL.handling = True try: for f in fences: f.record_attempt(target_str, call_type="PEP578_RAW_SOCKET") finally: _IN_GUARD_THREAD_LOCAL.handling = False raise RuntimeError( f"BLOCKED_BY_PROVIDER_SAFETY_FENCE: Kernel/C-runtime audit hook blocked '{event}' " f"to '{target_str}' while TEITAN_E2E_DISABLE_PROVIDER=1." ) elif event == "open": path_arg = args[0] if len(args) > 0 else None if isinstance(path_arg, (str, bytes, os.PathLike)): p_lower = os.fsdecode(path_arg).lower() if p_lower.endswith(".dpapi") or "openai_api_key" in p_lower or "device_token.dpapi" in p_lower: with _ACTIVE_FENCES_LOCK: fences = list(_ACTIVE_FENCES) if fences: _IN_GUARD_THREAD_LOCAL.handling = True try: for f in fences: f.record_attempt(str(path_arg), call_type="CREDENTIAL_FILE_ACCESS") finally: _IN_GUARD_THREAD_LOCAL.handling = False raise RuntimeError( f"BLOCKED_BY_PROVIDER_SAFETY_FENCE: Access to protected credential file '{path_arg}' " f"is strictly forbidden while ProviderSafetyFence is active." ) def _ensure_global_audit_hook() -> bool: global _GLOBAL_AUDIT_HOOK_INSTALLED with _ACTIVE_FENCES_LOCK: if not _GLOBAL_AUDIT_HOOK_INSTALLED: sys.addaudithook(_global_fence_audit_hook) _GLOBAL_AUDIT_HOOK_INSTALLED = True return _GLOBAL_AUDIT_HOOK_INSTALLED class ProviderSafetyFence: """ Verifiable Provider & Network Safety Fence context manager and telemetry collector. Guarantees: 1. Sets TEITAN_E2E_DISABLE_PROVIDER=1 and scrubs sensitive credential env vars while active. 2. Intercepts urllib.request.urlopen, http.client.HTTP(S)Connection.connect, socket.socket.connect, socket.socket.connect_ex, and socket.create_connection. 3. Installs a C-runtime PEP 578 sys.addaudithook blocking raw C socket.connect, socket.sendto, and .dpapi credential reads independent of urllib monkeypatching. 4. Records blocked attempts (PROVIDER_CALL_ATTEMPTS) while keeping NETWORK_PROVIDER_CALLS == 0. 5. Reports truthful telemetry backed only by physically active enforcement mechanisms. """ def __init__(self, blocked_domains: Optional[List[str]] = None): self.provider_call_attempts: int = 0 self.network_provider_calls: int = 0 self.blocked_attempts: List[Dict[str, Any]] = [] self._lock = threading.Lock() self._prev_env_val: Optional[str] = None self._saved_credentials_env: Dict[str, str] = {} self._orig_urlopen: Optional[Callable] = None self._orig_socket_connect: Optional[Callable] = None self._orig_socket_connect_ex: Optional[Callable] = None self._orig_create_connection: Optional[Callable] = None self._orig_http_connect: Optional[Callable] = None self._orig_https_connect: Optional[Callable] = None self._active: bool = False self._was_active_completed: bool = False self._urllib_hooked: bool = False self._socket_hooked: bool = False self._http_client_hooked: bool = False self._audit_hook_active: bool = False self._credentials_scrubbed: bool = False self._observer_fault: Optional[str] = None self._start_time_iso: Optional[str] = None self._end_time_iso: Optional[str] = None self.blocked_domains = [d.lower() for d in (blocked_domains or [ "api.openai.com", "openai.com", "api.anthropic.com", "generativelanguage.googleapis.com" ])] @property def is_active(self) -> bool: return self._active def _get_verified_scopes(self) -> List[str]: if not (self._active or self._was_active_completed): return [] scopes: List[str] = [] if self._urllib_hooked: scopes.append(SCOPE_URLLIB_HTTP) if self._socket_hooked: scopes.append(SCOPE_RAW_SOCKET) if self._http_client_hooked: scopes.append(SCOPE_HTTP_CLIENT) if self._audit_hook_active: scopes.append(SCOPE_PEP578_AUDIT_HOOK) return scopes def record_attempt(self, target: str, call_type: str = "HTTP") -> None: """Records an attempted call intercepted by the fence.""" with self._lock: self.provider_call_attempts += 1 self.blocked_attempts.append({ "timestamp": datetime.now(timezone.utc).isoformat(), "target": str(target), "call_type": call_type, "action": "BLOCKED_BY_SAFETY_FENCE", }) logger.warning( f"[ProviderSafetyFence] BLOCKED {call_type} attempt to '{target}'. Total attempts={self.provider_call_attempts}" ) def _guarded_urlopen(self, url, *args, **kwargs): if hasattr(url, "full_url"): target_str = str(url.full_url) elif hasattr(url, "get_full_url"): target_str = str(url.get_full_url()) else: target_str = str(url) _IN_GUARD_THREAD_LOCAL.handling = True try: self.record_attempt(target_str, call_type="URLLIB_HTTP") finally: _IN_GUARD_THREAD_LOCAL.handling = False raise RuntimeError( f"BLOCKED_BY_PROVIDER_SAFETY_FENCE: External network call to '{target_str}' " f"strictly forbidden while TEITAN_E2E_DISABLE_PROVIDER=1." ) def _guarded_socket_connect(self, sock_self, address): target_str = str(address) _IN_GUARD_THREAD_LOCAL.handling = True try: self.record_attempt(target_str, call_type="RAW_SOCKET") finally: _IN_GUARD_THREAD_LOCAL.handling = False raise RuntimeError( f"BLOCKED_BY_PROVIDER_SAFETY_FENCE: Raw socket.connect to '{target_str}' " f"strictly forbidden while TEITAN_E2E_DISABLE_PROVIDER=1." ) def _guarded_socket_connect_ex(self, sock_self, address): target_str = str(address) _IN_GUARD_THREAD_LOCAL.handling = True try: self.record_attempt(target_str, call_type="RAW_SOCKET_EX") finally: _IN_GUARD_THREAD_LOCAL.handling = False raise RuntimeError( f"BLOCKED_BY_PROVIDER_SAFETY_FENCE: Raw socket.connect_ex to '{target_str}' " f"strictly forbidden while TEITAN_E2E_DISABLE_PROVIDER=1." ) def _guarded_create_connection(self, address, *args, **kwargs): target_str = str(address) _IN_GUARD_THREAD_LOCAL.handling = True try: self.record_attempt(target_str, call_type="SOCKET_CREATE_CONNECTION") finally: _IN_GUARD_THREAD_LOCAL.handling = False raise RuntimeError( f"BLOCKED_BY_PROVIDER_SAFETY_FENCE: socket.create_connection to '{target_str}' " f"strictly forbidden while TEITAN_E2E_DISABLE_PROVIDER=1." ) def _guarded_http_connect(self, conn_self): target_str = f"{getattr(conn_self, 'host', 'unknown')}:{getattr(conn_self, 'port', 80)}" _IN_GUARD_THREAD_LOCAL.handling = True try: self.record_attempt(target_str, call_type="HTTP_CLIENT") finally: _IN_GUARD_THREAD_LOCAL.handling = False raise RuntimeError( f"BLOCKED_BY_PROVIDER_SAFETY_FENCE: HTTPConnection.connect to '{target_str}' " f"strictly forbidden while TEITAN_E2E_DISABLE_PROVIDER=1." ) def __enter__(self) -> "ProviderSafetyFence": self._start_time_iso = datetime.now(timezone.utc).isoformat() self._prev_env_val = os.environ.get(TEITAN_E2E_DISABLE_PROVIDER_ENV) os.environ[TEITAN_E2E_DISABLE_PROVIDER_ENV] = "1" # Scrub sensitive API credentials from environment while fence is active self._saved_credentials_env = {} for cred_key in SENSITIVE_CREDENTIAL_ENV_VARS: if cred_key in os.environ: self._saved_credentials_env[cred_key] = os.environ.pop(cred_key) self._credentials_scrubbed = True # 1. Intercept urllib.request.urlopen self._orig_urlopen = urllib.request.urlopen urllib.request.urlopen = self._guarded_urlopen self._urllib_hooked = True # 2. Intercept raw socket connection methods fence_ref = self self._orig_socket_connect = socket.socket.connect self._orig_socket_connect_ex = socket.socket.connect_ex self._orig_create_connection = socket.create_connection def _sock_connect_wrapper(sock_self, address): return fence_ref._guarded_socket_connect(sock_self, address) def _sock_connect_ex_wrapper(sock_self, address): return fence_ref._guarded_socket_connect_ex(sock_self, address) socket.socket.connect = _sock_connect_wrapper # type: ignore[assignment] socket.socket.connect_ex = _sock_connect_ex_wrapper # type: ignore[assignment] socket.create_connection = self._guarded_create_connection # type: ignore[assignment] self._socket_hooked = True # 3. Intercept http.client HTTPConnection & HTTPSConnection self._orig_http_connect = http.client.HTTPConnection.connect self._orig_https_connect = http.client.HTTPSConnection.connect def _http_conn_wrapper(conn_self): return fence_ref._guarded_http_connect(conn_self) http.client.HTTPConnection.connect = _http_conn_wrapper # type: ignore[assignment] http.client.HTTPSConnection.connect = _http_conn_wrapper # type: ignore[assignment] self._http_client_hooked = True # 4. Activate C-runtime PEP 578 audit hook registration self._audit_hook_active = _ensure_global_audit_hook() with _ACTIVE_FENCES_LOCK: if self not in _ACTIVE_FENCES: _ACTIVE_FENCES.append(self) self._active = True self._was_active_completed = False self._observer_fault = None return self def __exit__(self, exc_type, exc_val, exc_tb) -> None: self._end_time_iso = datetime.now(timezone.utc).isoformat() if exc_type is not None and not issubclass(exc_type, (RuntimeError, AssertionError)): self._observer_fault = f"UNEXPECTED_EXCEPTION_{exc_type.__name__}" with _ACTIVE_FENCES_LOCK: if self in _ACTIVE_FENCES: _ACTIVE_FENCES.remove(self) # Restore environment flag & scrubbed credentials if self._prev_env_val is None: os.environ.pop(TEITAN_E2E_DISABLE_PROVIDER_ENV, None) else: os.environ[TEITAN_E2E_DISABLE_PROVIDER_ENV] = self._prev_env_val for cred_key, cred_val in self._saved_credentials_env.items(): os.environ[cred_key] = cred_val self._saved_credentials_env.clear() # Restore urlopen if self._orig_urlopen is not None: urllib.request.urlopen = self._orig_urlopen self._orig_urlopen = None # Restore raw socket methods if self._orig_socket_connect is not None: socket.socket.connect = self._orig_socket_connect # type: ignore[assignment] self._orig_socket_connect = None if self._orig_socket_connect_ex is not None: socket.socket.connect_ex = self._orig_socket_connect_ex # type: ignore[assignment] self._orig_socket_connect_ex = None if self._orig_create_connection is not None: socket.create_connection = self._orig_create_connection # type: ignore[assignment] self._orig_create_connection = None # Restore http.client methods if self._orig_http_connect is not None: http.client.HTTPConnection.connect = self._orig_http_connect # type: ignore[assignment] self._orig_http_connect = None if self._orig_https_connect is not None: http.client.HTTPSConnection.connect = self._orig_https_connect # type: ignore[assignment] self._orig_https_connect = None self._active = False self._was_active_completed = (self._observer_fault is None) def get_telemetry(self) -> Dict[str, Any]: """Returns structured physical proof of safety fence execution.""" is_active_now = self._active is_completed = self._was_active_completed verified_scopes = self._get_verified_scopes() has_real_multi_layer = ( SCOPE_URLLIB_HTTP in verified_scopes and SCOPE_RAW_SOCKET in verified_scopes and SCOPE_PEP578_AUDIT_HOOK in verified_scopes ) if self.provider_call_attempts > 0: provider_telemetry_status = "TRUE" in_process_network_status = "TRUE" elif (is_active_now or is_completed) and has_real_multi_layer and not self._observer_fault: provider_telemetry_status = "FALSE_VERIFIED" in_process_network_status = "FALSE_VERIFIED" else: provider_telemetry_status = "UNKNOWN" in_process_network_status = "UNKNOWN" return { "PROVIDER_ALLOWED": False, "PROVIDER_CALL_ATTEMPTS": self.provider_call_attempts, "NETWORK_PROVIDER_CALLS": self.network_provider_calls, "FENCE_ACTIVE": self._active, "ENV_FLAG_ACTIVE": is_provider_fence_active(), "CREDENTIALS_SCRUBBED_DURING_FENCE": self._credentials_scrubbed, "RAW_SOCKET_HOOK_ACTIVE": self._socket_hooked, "HTTP_CLIENT_HOOK_ACTIVE": self._http_client_hooked, "PEP578_AUDIT_HOOK_ACTIVE": self._audit_hook_active, "SELF_REPAIR_PROVIDER": None, "AUTO_PROMOTION": False, "PRODUCTION_WRITE_AUTHORITY": False, "MAX_REPAIR_ATTEMPTS": 1, "BLOCKED_ATTEMPTS_COUNT": len(self.blocked_attempts), "BLOCKED_ATTEMPTS_DETAILS": list(self.blocked_attempts), # Authentic telemetry contract integration "OBSERVER_STATUS": "ACTIVE" if is_active_now else ("ACTIVE_COMPLETED" if is_completed else "INACTIVE"), "MONITORED_SCOPES": verified_scopes, "PROVIDER_TELEMETRY_STATUS": provider_telemetry_status, "IN_PROCESS_NETWORK_STATUS": in_process_network_status, "COMPREHENSIVE_NETWORK_STATUS": "TRUE" if self.provider_call_attempts > 0 else "UNKNOWN", "TRUE_NETWORK_OBSERVATION_PROVEN": has_real_multi_layer and (is_active_now or is_completed), "PROMOTION_ELIGIBLE": False, # Fail-closed "OBSERVER_FAULT": self._observer_fault, } def evaluate_authentic_telemetry( self, mission_id: str = "unknown", instance_id: str = "default", workspace_root: str = "", required_scopes: Optional[Set[str]] = None, ) -> Dict[str, Any]: """Runs the authentic telemetry evaluator against the current execution record.""" rec = AuthenticTelemetryEvaluator.evaluate_telemetry( mission_id=mission_id, instance_id=instance_id, workspace_root=workspace_root, observer_active=self._active or self._was_active_completed, monitored_scopes=self._get_verified_scopes(), detected_attempts=self.blocked_attempts, observer_fault=self._observer_fault, interval_start=self._start_time_iso, interval_end=self._end_time_iso, required_scopes=required_scopes, ) return rec.to_dict() def build_evidence_bundle( self, base_dir: str = EVIDENCE_BASE_DIR, timestamp_str: Optional[str] = None, brain_log: str = "", mission_record: Optional[Dict[str, Any]] = None, shadow_report: Optional[Any] = None, failure_fingerprint: str = "", hashes_before: Optional[Dict[str, str]] = None, hashes_after: Optional[Dict[str, str]] = None, ) -> Dict[str, str]: """ Creates the required evidence structure under: \\SELF_REPAIR_LIVE_SHADOW_E2E_\\ """ ts = timestamp_str or datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") run_dir = Path(base_dir) / f"SELF_REPAIR_LIVE_SHADOW_E2E_{ts}" run_dir.mkdir(parents=True, exist_ok=True) created_files: Dict[str, str] = {} # 1. brain.log log_path = run_dir / "brain.log" log_path.write_text(brain_log or "[BRAIN_HOST] Live shadow execution trace.\n", encoding="utf-8") created_files["brain_log"] = str(log_path) # 2. mission_terminal_record.json rec_path = run_dir / "mission_terminal_record.json" rec_path.write_text(json.dumps(mission_record or {}, indent=2, ensure_ascii=False), encoding="utf-8") created_files["mission_terminal_record"] = str(rec_path) # 3. shadow_repair_report.json report_data = shadow_report.to_dict() if hasattr(shadow_report, "to_dict") else (shadow_report or {}) rep_path = run_dir / "shadow_repair_report.json" rep_path.write_text(json.dumps(report_data, indent=2, ensure_ascii=False), encoding="utf-8") created_files["shadow_repair_report"] = str(rep_path) # 4. failure_fingerprint.txt fp_path = run_dir / "failure_fingerprint.txt" fp_path.write_text(str(failure_fingerprint or "unknown_fingerprint"), encoding="utf-8") created_files["failure_fingerprint"] = str(fp_path) # 5. guard_evidence.json guard_data = { "max_repair_attempts": MAX_REPAIR_ATTEMPTS, "duplicate_guard": "PASS" if MAX_REPAIR_ATTEMPTS >= 1 else "BLOCKED", "recursion_guard": "PASS" if MAX_REPAIR_ATTEMPTS == 1 else "BLOCKED", "circuit_breaker": "PASS" if (self.network_provider_calls == 0 and not self._observer_fault) else "TRIPPED", "self_repair_provider": None, "auto_promotion": False, "production_write_authority": False, } guard_path = run_dir / "guard_evidence.json" guard_path.write_text(json.dumps(guard_data, indent=2), encoding="utf-8") created_files["guard_evidence"] = str(guard_path) # 6. hashes_before_after.json — Truthful baseline integrity verification baseline_rep = BaselineIntegrityReport.evaluate(hashes_before, hashes_after) hash_data = { "hashes_before": hashes_before or {}, "hashes_after": hashes_after or {}, "all_baselines_identical": baseline_rep.all_baselines_identical, "baseline_integrity_status": baseline_rep.status, "missing_baselines": baseline_rep.missing_baselines, "mismatched_baselines": baseline_rep.mismatched_baselines, "evaluated_at": baseline_rep.evaluated_at, } hash_path = run_dir / "hashes_before_after.json" hash_path.write_text(json.dumps(hash_data, indent=2), encoding="utf-8") created_files["hashes_before_after"] = str(hash_path) # 7. provider_network_proof.json telemetry = self.get_telemetry() proof_path = run_dir / "provider_network_proof.json" proof_path.write_text(json.dumps(telemetry, indent=2), encoding="utf-8") created_files["provider_network_proof"] = str(proof_path) logger.info(f"[ProviderSafetyFence] Evidence bundle created at: {run_dir}") return created_files