# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR V1 — CANDIDATE VERIFIER (AR-76 HARDENED) Performs static compile validation, AST pre-execution risk classification, hash tracking, and isolated test execution with authentic pytest against sandbox candidate code. AR-76 Security & Verification Guarantees: 1. Never declares full OS isolation (os_isolation_active=False) when only Windows Job Object + Python PEP 578 audit hook are present. 2. Blocks execution of untrusted or high-risk candidates with BLOCKED_WITH_EVIDENCE when an OS-level virtualization / restricted identity with independent file and network controls is not present (never executing risky candidate code to get an artificial PASS). 3. Enforces deterministic startup order: CreateProcess/Popen (CREATE_SUSPENDED) -> AssignProcessToJobObject -> NtResumeProcess -> Python startup -> sitecustomize.py -> Verify IsProcessInJob(specific_job)==True (64-bit HANDLE) -> Register sys.addaudithook -> Trigger self-test audit event -> Write isolation_proof.json -> Lock proof file before pytest or candidate code loads. 4. Eliminates all OR conditions in isolation verification: requires strict AND across parent assignment and child in-process proof. 5. Uses authentic pytest (e.g. pytest 9.1.1 from isolated test_venv) and never substitutes a custom _STDLIB_PYTEST_SHIM. If authentic pytest is unavailable, reports PYTEST_REAL=NOT_TESTED. """ from __future__ import annotations import ast import ctypes from ctypes import wintypes import hashlib import json import os import py_compile import re import shlex import subprocess import sys import uuid from typing import Any, Callable, Dict, List, Optional, Tuple from app.remediation.self_repair.patch_executor import PatchExecutor from app.remediation.self_repair.sandbox_builder import ( _is_canonically_within, _is_reparse_point_or_symlink, ) from app.remediation.self_repair.schemas import ( PatchResult, RepairProposal, SandboxManifest, VerificationResult, ) def _compute_sha256(path: str) -> Optional[str]: try: if _is_reparse_point_or_symlink(path): return None if not os.path.isfile(path): return None with open(path, "rb") as f: return hashlib.sha256(f.read()).hexdigest().upper() except Exception: return None # ============================================================================== # WINDOWS KERNEL JOB OBJECT STRUCTURES FOR PROCESS CONTAINMENT # ============================================================================== class _IO_COUNTERS(ctypes.Structure): _fields_ = [ ("ReadOperationCount", ctypes.c_ulonglong), ("WriteOperationCount", ctypes.c_ulonglong), ("OtherOperationCount", ctypes.c_ulonglong), ("ReadTransferCount", ctypes.c_ulonglong), ("WriteTransferCount", ctypes.c_ulonglong), ("OtherTransferCount", ctypes.c_ulonglong), ] class _JOBOBJECT_BASIC_LIMIT_INFORMATION(ctypes.Structure): _fields_ = [ ("PerProcessUserTimeLimit", ctypes.c_longlong), ("PerJobUserTimeLimit", ctypes.c_longlong), ("LimitFlags", wintypes.DWORD), ("MinimumWorkingSetSize", ctypes.c_size_t), ("MaximumWorkingSetSize", ctypes.c_size_t), ("ActiveProcessLimit", wintypes.DWORD), ("Affinity", ctypes.c_size_t), ("PriorityClass", wintypes.DWORD), ("SchedulingClass", wintypes.DWORD), ] class _JOBOBJECT_EXTENDED_LIMIT_INFORMATION(ctypes.Structure): _fields_ = [ ("BasicLimitInformation", _JOBOBJECT_BASIC_LIMIT_INFORMATION), ("IoInfo", _IO_COUNTERS), ("ProcessMemoryLimit", ctypes.c_size_t), ("JobMemoryLimit", ctypes.c_size_t), ("PeakProcessMemoryUsed", ctypes.c_size_t), ("PeakJobMemoryUsed", ctypes.c_size_t), ] class _JOBOBJECT_BASIC_UI_RESTRICTIONS(ctypes.Structure): _fields_ = [ ("UIRestrictionsClass", wintypes.DWORD), ] JOB_OBJECT_LIMIT_ACTIVE_PROCESS = 0x00000008 JOB_OBJECT_LIMIT_PROCESS_MEMORY = 0x00000100 JOB_OBJECT_LIMIT_DIE_ON_UNHANDLED_EXCEPTION = 0x00000400 JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE = 0x00002000 JobObjectBasicUIRestrictions = 4 JobObjectExtendedLimitInformation = 9 UI_RESTRICTIONS_ALL = 0x000000FF CREATE_SUSPENDED = 0x00000004 def _is_venv_launcher_executable(python_exe: str) -> bool: """ Detects whether python_exe is a Windows venv launcher stub (Scripts\\python.exe) that spawns a child base python.exe process. """ exe_abs = os.path.abspath(python_exe) scripts_dir = os.path.dirname(exe_abs) venv_root = os.path.dirname(scripts_dir) if os.path.isfile(os.path.join(venv_root, "pyvenv.cfg")): return True return ( os.path.normcase(exe_abs) != os.path.normcase(os.path.abspath(getattr(sys, "_base_executable", sys.executable))) ) def _create_windows_sandbox_job(job_name: str, python_exe: str) -> Optional[int]: """ Creates and configures a uniquely named Windows Kernel Job Object enforcing: - ActiveProcessLimit = 2 (for venv launcher stub + interpreter) or 1 (direct interpreter) - ProcessMemoryLimit = 512 MB - KILL_ON_JOB_CLOSE & DIE_ON_UNHANDLED_EXCEPTION - Full UI restrictions """ if os.name != "nt": return None kernel32 = ctypes.windll.kernel32 kernel32.CreateJobObjectW.argtypes = [ctypes.c_void_p, wintypes.LPCWSTR] kernel32.CreateJobObjectW.restype = wintypes.HANDLE kernel32.SetInformationJobObject.argtypes = [ wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD, ] kernel32.SetInformationJobObject.restype = wintypes.BOOL kernel32.CloseHandle.argtypes = [wintypes.HANDLE] kernel32.CloseHandle.restype = wintypes.BOOL h_job = kernel32.CreateJobObjectW(None, job_name) if not h_job: return None ext_info = _JOBOBJECT_EXTENDED_LIMIT_INFORMATION() ext_info.BasicLimitInformation.LimitFlags = ( JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE | JOB_OBJECT_LIMIT_DIE_ON_UNHANDLED_EXCEPTION | JOB_OBJECT_LIMIT_ACTIVE_PROCESS | JOB_OBJECT_LIMIT_PROCESS_MEMORY ) is_venv_stub = _is_venv_launcher_executable(python_exe) ext_info.BasicLimitInformation.ActiveProcessLimit = 2 if is_venv_stub else 1 ext_info.ProcessMemoryLimit = 512 * 1024 * 1024 res_ext = kernel32.SetInformationJobObject( wintypes.HANDLE(h_job), JobObjectExtendedLimitInformation, ctypes.byref(ext_info), ctypes.sizeof(ext_info), ) if not res_ext: kernel32.CloseHandle(wintypes.HANDLE(h_job)) return None ui_info = _JOBOBJECT_BASIC_UI_RESTRICTIONS() ui_info.UIRestrictionsClass = UI_RESTRICTIONS_ALL kernel32.SetInformationJobObject( wintypes.HANDLE(h_job), JobObjectBasicUIRestrictions, ctypes.byref(ui_info), ctypes.sizeof(ui_info), ) return int(h_job) _SITECUSTOMIZE_TEMPLATE = '''# -*- coding: utf-8 -*- """ AR-76 PROCESS CONTAINMENT & C-RUNTIME AUDIT HOOK (PEP 578 sys.addaudithook) Loaded by CPython during site initialization BEFORE pytest or any candidate code executes. Enforces strict ordering: 1. Scrub sys.path (isolated to candidate_dir, guard_dir, and stdlib/venv site-packages). 2. Verify 64-bit Windows Kernel Job Object containment (both any-job and named-job). If not in the specific Job Object, abort immediately via os._exit(96). 3. Register sys.addaudithook(_ar76_audit_hook) BEFORE writing any proof file. 4. Trigger sys.audit("teitan.ar76.self_test", pid) and verify the hook physically fired. If self-test did not fire, abort immediately via os._exit(97). 5. Write isolation_proof.json and lock it against any subsequent writes. """ import ctypes from ctypes import wintypes import json import os import sys import time _CAND_DIR = os.path.normcase(os.path.realpath(os.environ.get("TEITAN_SANDBOX_CANDIDATE_DIR", ""))) _TEMP_DIR = os.path.normcase(os.path.realpath(os.environ.get("TEITAN_SANDBOX_TEMP_DIR", ""))) _GUARD_DIR = os.path.normcase(os.path.realpath(os.environ.get("TEITAN_SANDBOX_GUARD_DIR", ""))) _PROOF_FILE = os.path.normcase(os.path.realpath(os.environ.get("TEITAN_SANDBOX_PROOF_FILE", ""))) if os.environ.get("TEITAN_SANDBOX_PROOF_FILE") else "" _JOB_NAME = os.environ.get("TEITAN_SANDBOX_JOB_NAME", "") _STDLIB_PREFIX = os.path.normcase(os.path.realpath(sys.prefix)) _BASE_PREFIX = os.path.normcase(os.path.realpath(sys.base_prefix)) # 1. Scrub sys.path to prevent importing from production or parent workspace _clean_sys_path = [_CAND_DIR] for _p in list(sys.path): if not _p: continue _rp = os.path.normcase(os.path.realpath(_p)) if ( _rp == _CAND_DIR or _rp.startswith(_CAND_DIR + os.sep) or _rp == _GUARD_DIR or _rp.startswith(_STDLIB_PREFIX) or _rp.startswith(_BASE_PREFIX) ): if _rp not in [os.path.normcase(os.path.realpath(x)) for x in _clean_sys_path]: _clean_sys_path.append(_p) sys.path[:] = _clean_sys_path # 2. Verify Windows Kernel Job containment with proper 64-bit HANDLE signatures _in_any_job = False _in_specific_job = False if os.name == "nt": try: _k32 = ctypes.windll.kernel32 _k32.GetCurrentProcess.argtypes = [] _k32.GetCurrentProcess.restype = wintypes.HANDLE _k32.OpenJobObjectW.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.LPCWSTR] _k32.OpenJobObjectW.restype = wintypes.HANDLE _k32.IsProcessInJob.argtypes = [wintypes.HANDLE, wintypes.HANDLE, ctypes.POINTER(wintypes.BOOL)] _k32.IsProcessInJob.restype = wintypes.BOOL _k32.CloseHandle.argtypes = [wintypes.HANDLE] _k32.CloseHandle.restype = wintypes.BOOL _cur_proc = _k32.GetCurrentProcess() _b_any = wintypes.BOOL(0) if _k32.IsProcessInJob(_cur_proc, None, ctypes.byref(_b_any)): _in_any_job = bool(_b_any.value) if _JOB_NAME: _JOB_OBJECT_QUERY = 0x0004 _h_named = _k32.OpenJobObjectW(_JOB_OBJECT_QUERY, False, _JOB_NAME) if _h_named: _b_spec = wintypes.BOOL(0) if _k32.IsProcessInJob(_cur_proc, _h_named, ctypes.byref(_b_spec)): _in_specific_job = bool(_b_spec.value) _k32.CloseHandle(_h_named) except Exception: _in_any_job = False _in_specific_job = False if not (_in_any_job and _in_specific_job): os._exit(96) _ALLOWED_WRITE_ROOTS = tuple(r for r in (_CAND_DIR, _TEMP_DIR) if r) _FORBIDDEN_READ_MARKERS = ( ".dpapi", "openai_api_key", "device_token", "devices.json", "continuity.db", "digital optimus", "optimusruntime\\\\secrets", "optimusruntime/secrets", "teitan_lab\\\\ar13z", "teitan_lab/ar13z", ) _PROOF_LOCKED = False _HOOK_SELF_TEST_PASSED = False _HOOK_SELF_TEST_NS = 0 def _is_inside_allowed_write(target_path): try: if isinstance(target_path, int): return True p_str = os.fsdecode(target_path) p_slash = p_str.strip().replace(chr(92), "/").lower() if p_slash in ("nul", "//./nul", "/dev/null"): return True rp = os.path.normcase(os.path.realpath(p_str)) rp_slash = rp.replace(chr(92), "/").lower() if rp_slash in ("nul", "//./nul", "/dev/null"): return True if _PROOF_LOCKED and _PROOF_FILE and rp == _PROOF_FILE: return False for root in _ALLOWED_WRITE_ROOTS: if rp == root or rp.startswith(root + os.sep): return True return False except Exception: return False def _is_forbidden_read(target_path): try: if isinstance(target_path, int): return False p_str = os.fsdecode(target_path) rp = os.path.normcase(os.path.realpath(p_str)).lower() for safe_root in (_CAND_DIR, _TEMP_DIR, _GUARD_DIR, _STDLIB_PREFIX, _BASE_PREFIX): if safe_root and (rp == safe_root.lower() or rp.startswith(safe_root.lower() + os.sep)): return False for marker in _FORBIDDEN_READ_MARKERS: if marker in rp: return True return False except Exception: return True _WRITE_MODES = ("w", "a", "x", "+") _WRITE_FLAGS = os.O_WRONLY | os.O_RDWR | os.O_CREAT | os.O_TRUNC | os.O_APPEND def _ar76_audit_hook(event, args): global _HOOK_SELF_TEST_PASSED, _HOOK_SELF_TEST_NS if event == "teitan.ar76.self_test": _HOOK_SELF_TEST_PASSED = True _HOOK_SELF_TEST_NS = time.perf_counter_ns() return if event in ("socket.connect", "socket.bind", "socket.sendto", "socket.getaddrinfo", "urllib.Request"): raise PermissionError(f"OS_SANDBOX_NETWORK_BLOCKED: Event '{event}' is prohibited in candidate verification") if event in ("subprocess.Popen", "os.system", "os.spawn", "os.exec", "os.startfile", "os.posix_spawn"): raise PermissionError(f"OS_SANDBOX_SUBPROCESS_BLOCKED: Event '{event}' is prohibited in candidate verification") if event == "open": path_arg = args[0] if len(args) > 0 else None mode_arg = args[1] if len(args) > 1 else "r" flags_arg = args[2] if len(args) > 2 else 0 is_write = False if isinstance(mode_arg, str) and any(m in mode_arg for m in _WRITE_MODES): is_write = True if isinstance(flags_arg, int) and (flags_arg & _WRITE_FLAGS): is_write = True if is_write: if not _is_inside_allowed_write(path_arg): raise PermissionError( f"OS_SANDBOX_FS_WRITE_BLOCKED: Write to '{path_arg}' outside sandbox is prohibited" ) else: if _is_forbidden_read(path_arg): raise PermissionError( f"OS_SANDBOX_FS_READ_BLOCKED: Read of protected/secret path '{path_arg}' is prohibited" ) elif event in ("os.remove", "os.unlink", "os.rmdir", "os.mkdir", "os.putenv"): if event != "os.putenv" and len(args) > 0: if not _is_inside_allowed_write(args[0]): raise PermissionError( f"OS_SANDBOX_FS_MUTATION_BLOCKED: '{event}' on '{args[0]}' outside sandbox is prohibited" ) elif event in ("os.rename", "os.replace", "os.link", "os.symlink", "shutil.copyfile", "shutil.copymode", "shutil.copystat"): if event in ("os.link", "os.symlink"): raise PermissionError(f"OS_SANDBOX_LINK_BLOCKED: '{event}' is prohibited in candidate verification") for idx in (0, 1): if len(args) > idx and idx == 1 and not _is_inside_allowed_write(args[idx]): raise PermissionError( f"OS_SANDBOX_FS_MUTATION_BLOCKED: '{event}' target '{args[idx]}' outside sandbox is prohibited" ) # 3. Register audit hook FIRST before writing any proof _HOOK_REGISTERED_NS = time.perf_counter_ns() sys.addaudithook(_ar76_audit_hook) # 4. Trigger self-test audit event to prove the hook is actively intercepting events sys.audit("teitan.ar76.self_test", os.getpid()) if not _HOOK_SELF_TEST_PASSED or _HOOK_SELF_TEST_NS < _HOOK_REGISTERED_NS: os._exit(97) # 5. Write startup proof AFTER hook registration and self-test succeeded, then lock proof file if _PROOF_FILE: try: _PROOF_WRITTEN_NS = time.perf_counter_ns() with open(_PROOF_FILE, "w", encoding="utf-8") as _pf: json.dump({ "kernel_job_active": _in_any_job, "specific_job_verified": _in_specific_job, "job_name": _JOB_NAME, "audit_hook_installed": _HOOK_SELF_TEST_PASSED, "audit_hook_self_test_passed": _HOOK_SELF_TEST_PASSED, "audit_hook_registered_before_proof": (_HOOK_REGISTERED_NS <= _HOOK_SELF_TEST_NS <= _PROOF_WRITTEN_NS), "hook_registered_ns": _HOOK_REGISTERED_NS, "hook_self_test_ns": _HOOK_SELF_TEST_NS, "proof_written_ns": _PROOF_WRITTEN_NS, "order_verified": (_in_any_job and _in_specific_job and _HOOK_REGISTERED_NS <= _HOOK_SELF_TEST_NS <= _PROOF_WRITTEN_NS), "candidate_dir": _CAND_DIR, "pid": os.getpid(), }, _pf) _PROOF_LOCKED = True except Exception: os._exit(98) ''' class CandidateVerifier: """ Validates patched candidate files in the sandbox via compilation, static AST risk classification, and targeted tests executed with authentic pytest. """ SAFE_ENV_ALLOWLIST = { "SYSTEMROOT", "WINDIR", "COMSPEC", "PATHEXT", "SYSTEMDRIVE", "PATH", "NUMBER_OF_PROCESSORS", "PROCESSOR_ARCHITECTURE", "OS", } SENSITIVE_ENV_SUBSTRINGS = ( "KEY", "TOKEN", "SECRET", "PASSWORD", "CREDENTIAL", "DPAPI", "OPENAI", "ANTHROPIC", "GEMINI", "GOOGLE", "AUTH", "PROXY", "PYTHONPATH", "PYTHONSTARTUP", "PYTHONHOME", ) FORBIDDEN_SHELL_CHARS = (";", "&", "|", "`", "$", ">", "<", "\n", "\r") HIGH_RISK_AST_IMPORTS = { "ctypes", "cffi", "win32api", "win32con", "win32process", "win32security", "win32file", "win32pipe", "msvcrt", "_winapi", "subprocess", "socket", "ssl", "urllib", "http", "ftplib", "smtplib", "telnetlib", "xmlrpc", "multiprocessing", "webbrowser", } HIGH_RISK_AST_CALLS = { "eval", "exec", "compile", "__import__", "system", "popen", "spawnl", "spawnle", "spawnv", "spawnve", "execl", "execle", "execlp", "execv", "execve", "execvp", "startfile", "addaudithook", "_getframe", } def __init__(self, pytest_interpreter: Optional[str] = None) -> None: self._explicit_pytest_interpreter = pytest_interpreter self._cached_pytest_info: Optional[Dict[str, Any]] = None def _resolve_authentic_pytest(self) -> Dict[str, Any]: """ Detects whether authentic pytest (with pluggy in site-packages, never a custom shim) is available in an independent test environment. """ if self._cached_pytest_info is not None: return self._cached_pytest_info candidates: List[str] = [] for c in ( self._explicit_pytest_interpreter, os.environ.get("TEITAN_PYTEST_INTERPRETER"), os.path.join( os.environ.get("LOCALAPPDATA", r"C:\Users\intel\AppData\Local"), "OptimusRuntime", "Labs", "AR74", "06_ar76_candidate_hardening", "test_venv", "Scripts", "python.exe", ), sys.executable, ): if c and os.path.isfile(c): c_abs = os.path.abspath(c) if c_abs not in candidates: candidates.append(c_abs) probe_code = ( "import json, os, pytest, pluggy; " "p_file = os.path.abspath(getattr(pytest, '__file__', '')); " "is_site_pkg = ('site-packages' in p_file.lower() and '_pytest_shim' not in p_file.lower() and '_os_isolation_guard' not in p_file.lower()); " "print(json.dumps({'available': bool(is_site_pkg), 'pytest_version': getattr(pytest, '__version__', ''), " "'pluggy_version': getattr(pluggy, '__version__', ''), 'pytest_file': p_file}))" ) for py_exe in candidates: try: clean_probe_env = { k: os.environ[k] for k in self.SAFE_ENV_ALLOWLIST if k in os.environ } clean_probe_env["PYTHONNOUSERSITE"] = "1" proc = subprocess.run( [py_exe, "-c", probe_code], env=clean_probe_env, capture_output=True, text=True, timeout=10, ) if proc.returncode == 0 and proc.stdout.strip(): info = json.loads(proc.stdout.strip().splitlines()[-1]) if info.get("available") and info.get("pytest_version"): res = { "available": True, "PYTEST_REAL": "PASS", "interpreter": py_exe, "pytest_version": info["pytest_version"], "pluggy_version": info.get("pluggy_version", ""), "pytest_file": info.get("pytest_file", ""), } self._cached_pytest_info = res return res except Exception: continue res = { "available": False, "PYTEST_REAL": "NOT_TESTED", "interpreter": None, "pytest_version": None, "pluggy_version": None, "pytest_file": None, } self._cached_pytest_info = res return res @staticmethod def _probe_os_level_untrusted_isolation() -> Dict[str, Any]: """ Evaluates whether a true OS-level virtualization / container / restricted OS identity with independent OS filesystem ACLs and OS network firewall/namespace controls is active. A Windows Job Object + Python PEP 578 hook is NOT considered full OS isolation for untrusted or high-risk code. """ return { "full_os_isolation_available": False, "job_object_plus_pep578_is_full_isolation": False, "available_mechanisms": [ "WINDOWS_JOB_OBJECT_PROCESS_QUOTA_AND_KILL_ON_CLOSE", "CPYTHON_PEP578_IN_PROCESS_AUDIT_HOOK", "ENVIRONMENT_CREDENTIAL_SCRUBBING", "STATIC_AST_PRE_EXECUTION_GATE", ], "missing_os_barriers_for_untrusted_code": [ "OS_VIRTUALIZATION_OR_HYPERV_CONTAINER", "RESTRICTED_OS_IDENTITY_WITH_INDEPENDENT_FS_ACLS", "OS_KERNEL_NETWORK_NAMESPACE_OR_WFP_FIREWALL_FILTER", ], } @classmethod def _classify_candidate_risk( cls, manifest: SandboxManifest, proposal: RepairProposal, patch_result: PatchResult, ) -> Dict[str, Any]: """ Static pre-execution risk classifier. Determines whether the candidate code is LOW_RISK_AST_VERIFIED_INTERNAL or HIGH_RISK_OR_UNTRUSTED before any candidate code is executed. """ contract = proposal.sandbox_contract or {} explicit_trust = str(contract.get("candidate_trust_level") or "INTERNAL_CONTROLLED").upper() explicit_risk = str(contract.get("risk_level") or "AUTO").upper() findings: List[str] = [] if explicit_trust == "UNTRUSTED": findings.append("Explicit candidate_trust_level=UNTRUSTED") if explicit_risk == "HIGH": findings.append("Explicit risk_level=HIGH") modified_files = list((patch_result.file_diffs or {}).keys()) if not modified_files: modified_files = list(proposal.files_allowed or []) for rel_path in modified_files: norm_rel = os.path.normpath(str(rel_path)).replace("\\", "/") if not norm_rel.endswith(".py"): findings.append(f"Non-Python modified file '{norm_rel}' requires full OS virtualization") continue cand_file = os.path.join(manifest.candidate_dir, norm_rel) if not os.path.isfile(cand_file) or _is_reparse_point_or_symlink(cand_file): findings.append(f"Missing or symlink candidate file '{norm_rel}'") continue # Compare AST of source vs candidate so only newly introduced risky primitives # or untrusted files trigger high-risk blocking, OR if the patch diff itself touches risky primitives diff_text = (patch_result.file_diffs or {}).get(rel_path, "") or (patch_result.file_diffs or {}).get(norm_rel, "") added_lines = [ line[1:] for line in diff_text.splitlines() if line.startswith("+") and not line.startswith("+++") ] added_text = "\n".join(added_lines) for risky_imp in cls.HIGH_RISK_AST_IMPORTS: if re.search(rf"\b{re.escape(risky_imp)}\b", added_text): findings.append(f"Patch in '{norm_rel}' introduces high-risk module reference '{risky_imp}'") for risky_call in cls.HIGH_RISK_AST_CALLS: if re.search(rf"\b{re.escape(risky_call)}\s*\(", added_text): findings.append(f"Patch in '{norm_rel}' introduces high-risk call '{risky_call}(...)'") try: with open(cand_file, "r", encoding="utf-8") as f: cand_src = f.read() tree = ast.parse(cand_src, filename=norm_rel) # If file is not an internal Optimus module, inspect entire AST strictly is_internal_file = norm_rel.startswith(("app/", "runtime_v1/")) if not is_internal_file: for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: top_mod = alias.name.split(".")[0] if top_mod in cls.HIGH_RISK_AST_IMPORTS: findings.append(f"External candidate '{norm_rel}' imports '{alias.name}'") elif isinstance(node, ast.ImportFrom): top_mod = (node.module or "").split(".")[0] if top_mod in cls.HIGH_RISK_AST_IMPORTS: findings.append(f"External candidate '{norm_rel}' imports from '{node.module}'") elif isinstance(node, ast.Call): func_name = None if isinstance(node.func, ast.Name): func_name = node.func.id elif isinstance(node.func, ast.Attribute): func_name = node.func.attr if func_name in cls.HIGH_RISK_AST_CALLS: findings.append(f"External candidate '{norm_rel}' invokes '{func_name}'") except Exception as ast_exc: findings.append(f"AST parse failure on '{norm_rel}': {ast_exc}") is_high_risk = len(findings) > 0 return { "risk_classification": "HIGH_RISK_OR_UNTRUSTED" if is_high_risk else "LOW_RISK_AST_VERIFIED_INTERNAL", "is_high_risk_or_untrusted": is_high_risk, "findings": findings, } @staticmethod def _is_optimus_internal_repair(proposal: RepairProposal, manifest: SandboxManifest) -> bool: """Returns True if the repair targets Optimus's own internal modules.""" internal_prefixes = ("app/", "artifacts/", "runtime_v1/", "tools/", "new_mission_dispatcher.py") all_files = list(proposal.files_allowed or []) + [ sh.relative_path for sh in (manifest.source_hashes or []) ] for f in all_files: norm = os.path.normpath(str(f)).replace("\\", "/") if any(norm == p or norm.startswith(p) for p in internal_prefixes): return True return False @classmethod def _build_sanitized_env( cls, candidate_dir: str, guard_dir: str, sandbox_temp_dir: str, proof_file: str, job_name: str, ) -> Tuple[Dict[str, str], List[str]]: """ Builds a strictly sanitized environment dictionary for candidate subprocesses. Never inherits API keys, secrets, tokens, or parent PYTHONPATH. """ stripped_vars: List[str] = [] for k in os.environ.keys(): k_upper = k.upper() if any(sub in k_upper for sub in cls.SENSITIVE_ENV_SUBSTRINGS) or k_upper not in cls.SAFE_ENV_ALLOWLIST: stripped_vars.append(k) clean_env: Dict[str, str] = {} for safe_k in cls.SAFE_ENV_ALLOWLIST: val = os.environ.get(safe_k) if val is not None: clean_env[safe_k] = val cand_real = os.path.realpath(os.path.abspath(candidate_dir)) guard_real = os.path.realpath(os.path.abspath(guard_dir)) temp_real = os.path.realpath(os.path.abspath(sandbox_temp_dir)) clean_env["PYTHONPATH"] = guard_real + os.pathsep + cand_real clean_env["PYTHONNOUSERSITE"] = "1" clean_env["PYTHONDONTWRITEBYTECODE"] = "1" clean_env["PYTEST_DISABLE_PLUGIN_AUTOLOAD"] = "1" clean_env["TEITAN_E2E_DISABLE_PROVIDER"] = "1" clean_env["NO_PROXY"] = "*" clean_env["TEITAN_SANDBOX_CANDIDATE_DIR"] = cand_real clean_env["TEITAN_SANDBOX_GUARD_DIR"] = guard_real clean_env["TEITAN_SANDBOX_TEMP_DIR"] = temp_real clean_env["TEITAN_SANDBOX_PROOF_FILE"] = os.path.realpath(os.path.abspath(proof_file)) clean_env["TEITAN_SANDBOX_JOB_NAME"] = job_name clean_env["TEMP"] = temp_real clean_env["TMP"] = temp_real clean_env["TMPDIR"] = temp_real clean_env["USERPROFILE"] = temp_real clean_env["HOME"] = temp_real clean_env["LOCALAPPDATA"] = temp_real clean_env["APPDATA"] = temp_real return clean_env, sorted(stripped_vars) @staticmethod def _validate_candidate_test_target(target_spec: str, candidate_dir: str) -> Tuple[bool, Optional[str], Optional[str]]: """ Validates that a test target resolves canonically inside candidate_dir, is not a symlink/junction, and exists inside candidate_dir. Strictly blocks external test execution. """ raw = str(target_spec or "").strip() if not raw: return False, None, "Empty test target" parts = raw.split("::", 1) file_part = parts[0].strip() node_part = parts[1].strip() if len(parts) > 1 else None if node_part is not None: if not re.match(r"^[A-Za-z0-9_:\[\]\-\.]+$", node_part): return False, None, f"Invalid pytest node selector '{node_part}'" try: clean_rel, full_real = PatchExecutor._resolve_and_verify_sandbox_path( candidate_dir, file_part, role_label="candidate_test", ) except Exception as exc: return False, None, f"EXTERNAL_TEST_TARGET_BLOCKED: {exc}" if not clean_rel.endswith(".py"): return False, None, f"UNAUTHORIZED_TEST_FILE_TYPE: '{clean_rel}' is not a .py test file" if not os.path.isfile(full_real) or _is_reparse_point_or_symlink(full_real): return False, None, f"TARGET_FILE_NOT_FOUND_IN_CANDIDATE: '{clean_rel}' does not exist in candidate_dir" final_spec = f"{clean_rel}::{node_part}" if node_part else clean_rel return True, final_spec, None def _parse_and_validate_focused_command( self, cmd_str: str, candidate_dir: str, ) -> Tuple[bool, List[str], Optional[str]]: """ Validates a focused_test_command string: 1. Rejects shell metacharacters or chaining. 2. Requires pytest invocation (python -m pytest ... or pytest ...). 3. Rejects arbitrary code flags (-c, -o, -p, --rootdir, etc.). 4. Validates that every test target resolves canonically inside candidate_dir. 5. Uses the verified authentic pytest interpreter. """ raw_cmd = str(cmd_str or "").strip() if not raw_cmd: return False, [], "Empty focused_test_command" for bad_ch in self.FORBIDDEN_SHELL_CHARS: if bad_ch in raw_cmd: return False, [], f"UNAUTHORIZED_COMMAND_BLOCKED: Shell metacharacter '{bad_ch}' forbidden" try: tokens = shlex.split(raw_cmd) except Exception as exc: return False, [], f"UNAUTHORIZED_COMMAND_BLOCKED: Malformed command '{raw_cmd}': {exc}" if not tokens: return False, [], "Empty command tokens" idx = 0 exe0 = os.path.basename(tokens[0]).lower() if exe0 in ("python", "python3", "python.exe"): if len(tokens) < 3 or tokens[1] != "-m" or tokens[2] != "pytest": return ( False, [], f"UNAUTHORIZED_COMMAND_BLOCKED: Only 'python -m pytest' or 'pytest' is authorized (got '{raw_cmd}')", ) idx = 3 elif exe0 in ("pytest", "pytest.exe"): idx = 1 else: return ( False, [], f"UNAUTHORIZED_COMMAND_BLOCKED: Executable '{tokens[0]}' is not authorized in candidate verifier", ) allowed_flags = {"-q", "-v", "-vv", "-x", "--tb=short", "--tb=no", "--tb=line", "-s"} validated_targets: List[str] = [] extra_flags: List[str] = [] i = idx while i < len(tokens): tok = tokens[i] if tok == "-k" and (i + 1) < len(tokens): k_expr = tokens[i + 1] if not re.match(r"^[A-Za-z0-9_ \-\(\)]+$", k_expr): return False, [], f"UNAUTHORIZED_COMMAND_BLOCKED: Unsafe -k expression '{k_expr}'" extra_flags.extend(["-k", k_expr]) i += 2 continue elif tok.startswith("-"): if tok not in allowed_flags: return False, [], f"UNAUTHORIZED_COMMAND_BLOCKED: Disallowed pytest flag '{tok}'" extra_flags.append(tok) i += 1 continue else: ok_t, clean_spec, err_t = self._validate_candidate_test_target(tok, candidate_dir) if not ok_t or not clean_spec: return False, [], err_t or f"Invalid test target '{tok}'" validated_targets.append(clean_spec) i += 1 if not validated_targets: return False, [], "UNAUTHORIZED_COMMAND_BLOCKED: Command specifies no valid test file inside candidate_dir" pytest_info = self._resolve_authentic_pytest() if not pytest_info.get("available") or not pytest_info.get("interpreter"): return False, [], "AUTHENTIC_PYTEST_UNAVAILABLE: PYTEST_REAL=NOT_TESTED" cand_real = os.path.realpath(os.path.abspath(candidate_dir)) sandbox_basetemp = os.path.join(os.path.dirname(cand_real), "_os_isolation_tmp", "pytest_basetemp") canonical_cmd = [ pytest_info["interpreter"], "-m", "pytest", f"--rootdir={cand_real}", f"--confcutdir={cand_real}", f"--basetemp={sandbox_basetemp}", "-o", f"cache_dir={os.path.join(os.path.dirname(cand_real), '_os_isolation_tmp', 'pytest_cache')}", ] + validated_targets + extra_flags return True, canonical_cmd, None @staticmethod def _parse_pytest_counts(stdout_text: str, stderr_text: str = "") -> Dict[str, int]: """Extracts passed, failed, skipped, error, and collected counts from authentic pytest output.""" combined = f"{stdout_text or ''}\n{stderr_text or ''}" passed_m = re.search(r"(\d+)\s+passed", combined) failed_m = re.search(r"(\d+)\s+failed", combined) skipped_m = re.search(r"(\d+)\s+skipped", combined) error_m = re.search(r"(\d+)\s+errors?", combined) collected_m = re.search(r"collected\s+(\d+)\s+items?", combined) passed_c = int(passed_m.group(1)) if passed_m else 0 failed_c = int(failed_m.group(1)) if failed_m else 0 skipped_c = int(skipped_m.group(1)) if skipped_m else 0 error_c = int(error_m.group(1)) if error_m else 0 collected_c = int(collected_m.group(1)) if collected_m else (passed_c + failed_c + skipped_c + error_c) if "no tests ran" in combined.lower() or "collected 0 items" in combined.lower(): passed_c = 0 collected_c = 0 return { "passed_count": passed_c, "failed_count": failed_c, "skipped_count": skipped_c, "error_count": error_c, "collected_count": collected_c, } def _execute_isolated_pytest( self, cmd_tokens: List[str], candidate_dir: str, display_cmd: str, ) -> Dict[str, Any]: """ Executes a validated authentic pytest command inside candidate_dir with: 1. CREATE_SUSPENDED -> AssignProcessToJobObject -> NtResumeProcess 2. Named Job Object verified inside child process via 64-bit IsProcessInJob 3. CPython PEP 578 sys.addaudithook registered and self-tested BEFORE proof is written 4. Authentic pytest (never _STDLIB_PYTEST_SHIM) 5. Strict AND verification of all containment conditions. """ pytest_info = self._resolve_authentic_pytest() if not pytest_info.get("available"): return { "command": display_cmd, "status": "AUTHENTIC_PYTEST_UNAVAILABLE", "PYTEST_REAL": "NOT_TESTED", "passed": False, "returncode": -1, } cand_real = os.path.realpath(os.path.abspath(candidate_dir)) if not os.path.isdir(cand_real) or _is_reparse_point_or_symlink(cand_real): return { "command": display_cmd, "status": "SANDBOX_CANDIDATE_DIR_INVALID", "PYTEST_REAL": pytest_info["PYTEST_REAL"], "passed": False, "returncode": -1, } sandbox_parent = os.path.dirname(cand_real) guard_dir = os.path.join(sandbox_parent, "_os_isolation_guard") sandbox_temp = os.path.join(sandbox_parent, "_os_isolation_tmp") os.makedirs(guard_dir, exist_ok=True) os.makedirs(sandbox_temp, exist_ok=True) # Ensure no fake pytest.py shim exists in guard_dir legacy_shim = os.path.join(guard_dir, "pytest.py") if os.path.exists(legacy_shim): try: os.remove(legacy_shim) except OSError: pass sitecust_path = os.path.join(guard_dir, "sitecustomize.py") proof_file = os.path.join(sandbox_temp, "isolation_proof.json") if os.path.exists(proof_file): try: os.remove(proof_file) except OSError: pass with open(sitecust_path, "w", encoding="utf-8") as f: f.write(_SITECUSTOMIZE_TEMPLATE) job_name = f"Local\\TeitanAR76Job_{uuid.uuid4().hex[:12]}" clean_env, stripped_vars = self._build_sanitized_env( candidate_dir=cand_real, guard_dir=guard_dir, sandbox_temp_dir=sandbox_temp, proof_file=proof_file, job_name=job_name, ) python_exe = cmd_tokens[0] h_job = _create_windows_sandbox_job(job_name=job_name, python_exe=python_exe) if os.name == "nt" and not h_job: return { "command": display_cmd, "status": "PROCESS_CONTAINMENT_JOB_UNAVAILABLE", "error": "Failed to create named Windows Kernel Job Object", "passed": False, "returncode": -1, } kernel32 = ctypes.windll.kernel32 if os.name == "nt" else None ntdll = ctypes.windll.ntdll if os.name == "nt" else None if os.name == "nt" and kernel32 and ntdll: kernel32.AssignProcessToJobObject.argtypes = [wintypes.HANDLE, wintypes.HANDLE] kernel32.AssignProcessToJobObject.restype = wintypes.BOOL kernel32.CloseHandle.argtypes = [wintypes.HANDLE] kernel32.CloseHandle.restype = wintypes.BOOL ntdll.NtResumeProcess.argtypes = [wintypes.HANDLE] ntdll.NtResumeProcess.restype = ctypes.c_long job_assigned = False resumed_ok = False try: proc = subprocess.Popen( cmd_tokens, cwd=cand_real, env=clean_env, creationflags=CREATE_SUSPENDED if os.name == "nt" else 0, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) if os.name == "nt" and h_job and kernel32 and ntdll: job_assigned = bool( kernel32.AssignProcessToJobObject(wintypes.HANDLE(h_job), wintypes.HANDLE(int(proc._handle))) ) if not job_assigned: proc.kill() proc.communicate() return { "command": display_cmd, "status": "JOB_ASSIGNMENT_FAILED_BEFORE_RESUME", "passed": False, "returncode": -1, } nt_status = ntdll.NtResumeProcess(wintypes.HANDLE(int(proc._handle))) resumed_ok = (nt_status == 0) if not resumed_ok: proc.kill() proc.communicate() return { "command": display_cmd, "status": "PROCESS_RESUME_FAILED", "passed": False, "returncode": -1, } try: stdout_str, stderr_str = proc.communicate(timeout=30) except subprocess.TimeoutExpired: proc.kill() stdout_str, stderr_str = proc.communicate() return { "command": display_cmd, "status": "TIMEOUT_EXPIRED", "passed": False, "returncode": -1, "stdout_snippet": (stdout_str or "")[-500:], "stderr_snippet": (stderr_str or "")[-500:], } rc = proc.returncode except Exception as exc: return { "command": display_cmd, "status": "EXECUTION_EXCEPTION", "error": str(exc), "passed": False, "returncode": -1, } finally: if os.name == "nt" and h_job and kernel32: kernel32.CloseHandle(wintypes.HANDLE(h_job)) isolation_proof: Dict[str, Any] = {} if os.path.isfile(proof_file): try: with open(proof_file, "r", encoding="utf-8") as pf: isolation_proof = json.load(pf) except Exception: isolation_proof = {} # Strict AND condition — zero OR shortcuts permitted process_containment_verified = bool( job_assigned is True and resumed_ok is True and isolation_proof.get("kernel_job_active") is True and isolation_proof.get("specific_job_verified") is True and isolation_proof.get("job_name") == job_name and isolation_proof.get("audit_hook_installed") is True and isolation_proof.get("audit_hook_self_test_passed") is True and isolation_proof.get("audit_hook_registered_before_proof") is True and isolation_proof.get("order_verified") is True ) counts = self._parse_pytest_counts(stdout_str, stderr_str) tests_passed = counts["passed_count"] tests_failed = counts["failed_count"] + counts["error_count"] tests_skipped = counts["skipped_count"] if not process_containment_verified: passed = False status_str = "PROCESS_CONTAINMENT_NOT_VERIFIED_FAIL_CLOSED" elif rc != 0 or tests_failed > 0: passed = False status_str = "TESTS_FAILED" elif tests_passed <= 0: passed = False status_str = "ZERO_TESTS_EXECUTED_FAIL_CLOSED" else: passed = True status_str = "PASS" return { "command": display_cmd, "canonical_argv": cmd_tokens, "interpreter_used": pytest_info["interpreter"], "PYTEST_REAL": pytest_info["PYTEST_REAL"], "pytest_version": pytest_info["pytest_version"], "pluggy_version": pytest_info["pluggy_version"], "pytest_file": pytest_info["pytest_file"], "returncode": rc, "status": status_str, "passed": passed, "tests_collected": counts["collected_count"], "tests_passed": tests_passed, "tests_failed": tests_failed, "tests_skipped": tests_skipped, "tests_errors": counts["error_count"], # Honest reporting per AR-76 Section 2: # Windows Job Object + PEP 578 is process containment, NOT full OS isolation for untrusted code. "os_isolation_active": False, "full_os_isolation_for_untrusted_code": False, "process_containment_verified": process_containment_verified, "containment_mechanism": "CREATE_SUSPENDED+NAMED_WINDOWS_JOB_OBJECT+PEP578_AUDIT_HOOK+AST_PRECHECK", "kernel_job_assigned": job_assigned, "kernel_job_active": bool(isolation_proof.get("kernel_job_active") is True), "specific_job_verified": bool(isolation_proof.get("specific_job_verified") is True), "audit_hook_installed": bool(isolation_proof.get("audit_hook_installed") is True), "audit_hook_self_test_passed": bool(isolation_proof.get("audit_hook_self_test_passed") is True), "audit_hook_registered_before_proof": bool(isolation_proof.get("audit_hook_registered_before_proof") is True), "isolation_proof": isolation_proof, "credentials_inherited": False, "stripped_env_vars_count": len(stripped_vars), "stdout_snippet": (stdout_str or "")[-800:], "stderr_snippet": (stderr_str or "")[-800:], } def verify_candidate( self, manifest: SandboxManifest, proposal: RepairProposal, patch_result: PatchResult, focused_test_runner: Optional[Callable[[str], Tuple[bool, Dict[str, Any]]]] = None, regression_test_runner: Optional[Callable[[str], Tuple[bool, Dict[str, Any]]]] = None, incident_dir: Optional[str] = None, ) -> VerificationResult: """ Runs compilation, AST risk classification, focused tests, and regression tests. Blocks execution with BLOCKED_WITH_EVIDENCE if candidate is high-risk/untrusted and full OS-level virtualization/restricted identity is not available. """ cid = manifest.candidate_id inc_id = manifest.incident_id rep_id = manifest.repair_id is_internal = self._is_optimus_internal_repair(proposal, manifest) # 1. Hashes before vs after before_hashes: Dict[str, str] = {sh.relative_path: sh.source_sha256 for sh in manifest.source_hashes} after_hashes: Dict[str, str] = {} for rel in before_hashes.keys(): cand_p = os.path.join(manifest.candidate_dir, rel) h = _compute_sha256(cand_p) if h: after_hashes[rel] = h # 2. Compile Check compile_pass = True compile_errors: List[str] = [] if patch_result.status != "SUCCESS": compile_pass = False compile_errors.append(f"Patch execution did not succeed: {patch_result.error_message}") for rel in before_hashes.keys(): if rel.endswith(".py"): cand_p = os.path.join(manifest.candidate_dir, rel) if os.path.isfile(cand_p) and not _is_reparse_point_or_symlink(cand_p): try: py_compile.compile(cand_p, doraise=True) except Exception as ce: compile_pass = False compile_errors.append(f"{rel}: {str(ce)}") else: compile_pass = False compile_errors.append(f"{rel}: Candidate file missing or invalid") # 3. Pre-Execution Risk & OS Isolation Gate (AR-76 Section 2) os_iso_probe = self._probe_os_level_untrusted_isolation() risk_eval = self._classify_candidate_risk(manifest, proposal, patch_result) if compile_pass and risk_eval["is_high_risk_or_untrusted"] and not os_iso_probe["full_os_isolation_available"]: blocked_summary = { "status": "BLOCKED_WITH_EVIDENCE", "passed": False, "executed": 0, "blocked_reason": "OS_ISOLATION_INSUFFICIENT_FOR_UNTRUSTED_OR_HIGH_RISK_CANDIDATE", "risk_classification": risk_eval["risk_classification"], "risk_findings": risk_eval["findings"], "os_isolation_probe": os_iso_probe, "os_isolation_active": False, "full_os_isolation_for_untrusted_code": False, } v_res_blocked = VerificationResult( candidate_id=cid, incident_id=inc_id, repair_id=rep_id, compile_pass=compile_pass, compile_errors=compile_errors, candidate_before_sha256=before_hashes, candidate_after_sha256=after_hashes, focused_tests_pass=False, focused_tests_summary=blocked_summary, regression_tests_pass=False, regression_tests_summary={ "status": "BLOCKED_WITH_EVIDENCE", "passed": False, "executed": 0, "blocked_reason": "OS_ISOLATION_INSUFFICIENT_FOR_UNTRUSTED_OR_HIGH_RISK_CANDIDATE", }, ) if incident_dir: self.persist_result(v_res_blocked, incident_dir) return v_res_blocked # 4. Focused Tests focused_pass = False focused_summary: Dict[str, Any] = {} focused_cmd = ( getattr(proposal, "focused_test_command", None) or proposal.sandbox_contract.get("focused_test_command") ) if not compile_pass: focused_pass = False focused_summary = {"status": "SKIPPED_COMPILE_FAILED", "passed": False, "error": "Compilation failed"} elif focused_test_runner: try: focused_pass, focused_summary = focused_test_runner(manifest.candidate_dir) if not isinstance(focused_summary, dict): focused_summary = {"summary": str(focused_summary)} if focused_summary.get("executed") == 0: focused_pass = False focused_summary["status"] = "ZERO_TESTS_EXECUTED_FAIL_CLOSED" focused_summary["passed"] = bool(focused_pass) except Exception as fe: focused_pass = False focused_summary = {"status": "ERROR", "passed": False, "error": str(fe)} elif focused_cmd: ok_cmd, cmd_tokens, cmd_err = self._parse_and_validate_focused_command( focused_cmd, manifest.candidate_dir, ) if not ok_cmd: focused_pass = False focused_summary = { "command": focused_cmd, "status": "UNAUTHORIZED_OR_EXTERNAL_TEST_BLOCKED", "error": cmd_err, "passed": False, } else: focused_summary = self._execute_isolated_pytest( cmd_tokens=cmd_tokens, candidate_dir=manifest.candidate_dir, display_cmd=focused_cmd, ) focused_summary["risk_classification"] = risk_eval["risk_classification"] focused_pass = bool(focused_summary.get("passed")) elif proposal.focused_tests: focused_pass, focused_summary = self._run_test_targets( proposal.focused_tests, manifest.candidate_dir, is_regression=False, require_non_empty=True, ) focused_summary["risk_classification"] = risk_eval["risk_classification"] else: focused_pass = False focused_summary = { "status": "NO_REAL_FOCUSED_TEST_AVAILABLE", "passed": False, "error": "No focused test specified (Fail-Closed)", } # 5. Regression Tests regression_pass = False regression_summary: Dict[str, Any] = {} if not compile_pass or not focused_pass: regression_pass = False regression_summary = {"status": "SKIPPED_OR_FAILED_EARLIER", "passed": False} elif regression_test_runner: try: regression_pass, regression_summary = regression_test_runner(manifest.candidate_dir) if not isinstance(regression_summary, dict): regression_summary = {"summary": str(regression_summary)} if regression_summary.get("executed") == 0: regression_pass = False regression_summary["status"] = "ZERO_TESTS_EXECUTED_FAIL_CLOSED" regression_summary["passed"] = bool(regression_pass) except Exception as re_exc: regression_pass = False regression_summary = {"status": "ERROR", "passed": False, "error": str(re_exc)} elif proposal.regression_tests: regression_pass, regression_summary = self._run_test_targets( proposal.regression_tests, manifest.candidate_dir, is_regression=True, require_non_empty=True, ) elif is_internal: regression_pass = False regression_summary = { "executed": 0, "status": "MISSING_MANDATORY_REGRESSION_TESTS_FOR_INTERNAL_COMPONENT", "passed": False, "error": "Repairing internal Optimus components requires real regression tests (0 executed is FAIL).", } else: regression_pass = True regression_summary = {"executed": 0, "status": "NO_REGRESSION_TESTS_REQUIRED", "passed": True} v_res = VerificationResult( candidate_id=cid, incident_id=inc_id, repair_id=rep_id, compile_pass=compile_pass, compile_errors=compile_errors, candidate_before_sha256=before_hashes, candidate_after_sha256=after_hashes, focused_tests_pass=focused_pass, focused_tests_summary=focused_summary, regression_tests_pass=regression_pass, regression_tests_summary=regression_summary, ) if incident_dir: self.persist_result(v_res, incident_dir) return v_res def _run_test_targets( self, targets: List[str], candidate_dir: str, is_regression: bool = False, require_non_empty: bool = True, ) -> Tuple[bool, Dict[str, Any]]: """ Executes verified in-candidate test files using authentic pytest under process containment. """ if not targets: if is_regression and not require_non_empty: return True, {"executed": 0, "status": "NO_REGRESSION_TESTS_REQUIRED", "passed": True} return False, {"executed": 0, "status": "NO_TESTS_SPECIFIED", "passed": False} pytest_info = self._resolve_authentic_pytest() if not pytest_info.get("available") or not pytest_info.get("interpreter"): return False, { "executed": 0, "status": "AUTHENTIC_PYTEST_UNAVAILABLE", "PYTEST_REAL": "NOT_TESTED", "passed": False, } executed_count = 0 total_tests_passed = 0 total_tests_collected = 0 total_tests_failed = 0 total_tests_skipped = 0 all_passed = True details: Dict[str, Any] = {} cand_real = os.path.realpath(os.path.abspath(candidate_dir)) sandbox_basetemp = os.path.join(os.path.dirname(cand_real), "_os_isolation_tmp", "pytest_basetemp") sandbox_cache = os.path.join(os.path.dirname(cand_real), "_os_isolation_tmp", "pytest_cache") for t in targets: ok_t, clean_spec, err_t = self._validate_candidate_test_target(t, cand_real) if not ok_t or not clean_spec: details[str(t)] = { "status": "EXTERNAL_OR_MISSING_TEST_TARGET_BLOCKED", "error": err_t, "passed": False, } all_passed = False continue cmd_tokens = [ pytest_info["interpreter"], "-m", "pytest", f"--rootdir={cand_real}", f"--confcutdir={cand_real}", f"--basetemp={sandbox_basetemp}", "-o", f"cache_dir={sandbox_cache}", clean_spec, "-v", ] run_info = self._execute_isolated_pytest( cmd_tokens=cmd_tokens, candidate_dir=cand_real, display_cmd=f"python -m pytest {clean_spec} -v", ) details[str(t)] = run_info total_tests_collected += int(run_info.get("tests_collected", 0)) total_tests_failed += int(run_info.get("tests_failed", 0)) total_tests_skipped += int(run_info.get("tests_skipped", 0)) if not run_info.get("passed"): all_passed = False else: executed_count += 1 total_tests_passed += int(run_info.get("tests_passed", 0)) if executed_count == 0 or total_tests_passed == 0: all_passed = False status_str = "ZERO_TESTS_EXECUTED_FAIL_CLOSED" else: status_str = "PASS" if all_passed else "FAIL" return all_passed, { "executed": executed_count, "total_tests_collected": total_tests_collected, "total_tests_passed": total_tests_passed, "total_tests_failed": total_tests_failed, "total_tests_skipped": total_tests_skipped, "PYTEST_REAL": pytest_info["PYTEST_REAL"], "pytest_version": pytest_info["pytest_version"], "interpreter_used": pytest_info["interpreter"], "status": status_str, "passed": all_passed, "targets": details, } def persist_result(self, result: VerificationResult, incident_dir: str) -> str: """Persists verification_result.json.""" os.makedirs(incident_dir, exist_ok=True) path = os.path.join(incident_dir, "verification_result.json") with open(path, "w", encoding="utf-8") as f: json.dump(result.to_dict(), f, indent=2, ensure_ascii=False) return path