# -*- coding: utf-8 -*- """ TEITAN SELF-REPAIR V1 — CANDIDATE VERIFIER (FASE 2 — AR-74 HARDENED) Performs static compile validation, hash tracking, and isolated test execution (focused and regression) against sandbox candidate code. Security & Verification Guarantees (AR-74): 1. No implicit execution of tests outside candidate_dir (canonical containment required). 2. Zero inheritance of API credentials or sensitive environment variables. 3. No arbitrary or unauthorized shell/test commands permitted. 4. Real OS kernel isolation (Windows Job Object with ActiveProcessLimit=1, UI restrictions, KILL_ON_JOB_CLOSE) + unremovable CPython PEP 578 C-runtime sys.addaudithook blocking network, subprocesses, external writes, and production/secret reads. 5. For Optimus internal components: requires real focused tests AND real regression tests; never treats 0 executed tests as PASS; records commands, exit codes, and isolation evidence. """ from __future__ import annotations import ctypes from ctypes import wintypes import hashlib import importlib.util import json import os import py_compile import re import shlex import shutil import subprocess import sys 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 OS-LEVEL SANDBOXING # ============================================================================== 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 def _create_windows_sandbox_job() -> Optional[int]: """ Creates and configures a Windows Kernel Job Object enforcing: - ActiveProcessLimit = 1 (kernel block on child process spawning) - ProcessMemoryLimit = 512 MB - KILL_ON_JOB_CLOSE & DIE_ON_UNHANDLED_EXCEPTION - Full UI restrictions (clipboard, desktop, display, global atoms, external handles) """ if os.name != "nt": return None kernel32 = ctypes.windll.kernel32 kernel32.CreateJobObjectW.argtypes = [ ctypes.c_void_p, wintypes.LPCWSTR ] kernel32.CreateJobObjectW.restype = wintypes.HANDLE h_job = kernel32.CreateJobObjectW(None, None) 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 ) # On Windows venvs, Scripts\python.exe is a launcher stub (process 1) that spawns # the base Python interpreter (process 2) and waits. Setting ActiveProcessLimit = 2 # when using a venv stub (or 1 otherwise) fills all job process slots immediately once # the interpreter starts, preventing the test from spawning any child processes at the OS kernel level. is_venv_stub = ( os.path.normcase(os.path.abspath(sys.executable)) != os.path.normcase(os.path.abspath(getattr(sys, "_base_executable", sys.executable))) ) 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-74 OS & C-RUNTIME SANDBOX ENFORCEMENT (PEP 578 sys.addaudithook) Automatically loaded by CPython at startup before pytest or candidate code executes. Cannot be removed or bypassed by user/candidate Python code. """ import ctypes import json import os import sys _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.environ.get("TEITAN_SANDBOX_PROOF_FILE", "") _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 = [] for _p in list(sys.path): if not _p: _clean_sys_path.append(_CAND_DIR) 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 _clean_sys_path: _clean_sys_path.append(_p) sys.path[:] = _clean_sys_path # 2. Check Windows Kernel Job containment _in_job = False if os.name == "nt": try: _k32 = ctypes.windll.kernel32 _b = ctypes.c_int(0) if _k32.IsProcessInJob(_k32.GetCurrentProcess(), None, ctypes.byref(_b)): _in_job = bool(_b.value) except Exception: _in_job = False # 3. Record startup proof before locking down filesystem writes if _PROOF_FILE: try: with open(_PROOF_FILE, "w", encoding="utf-8") as _pf: json.dump({ "audit_hook_installed": True, "kernel_job_active": _in_job, "candidate_dir": _CAND_DIR, "pid": os.getpid(), }, _pf) except Exception: pass _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", ) def _is_inside_allowed_write(target_path): try: if isinstance(target_path, int): return True p_str = os.fsdecode(target_path) rp = os.path.normcase(os.path.realpath(p_str)) 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() # Allow reads inside candidate_dir, temp_dir, guard_dir, and python installation 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 _ar74_audit_hook(event, args): 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" ) sys.addaudithook(_ar74_audit_hook) ''' _STDLIB_PYTEST_SHIM = '''# -*- coding: utf-8 -*- """ Minimal stdlib-backed pytest runner & module for isolated environments without third-party pytest wheel. Supports @pytest.fixture (including yield & autouse), @pytest.mark.parametrize, pytest.raises, monkeypatch, tmp_path, unittest.TestCase, and test_* functions. """ import contextlib import importlib.util import inspect import os import re import shutil import sys import tempfile import traceback import unittest from pathlib import Path class _MonkeyPatch: def __init__(self): self._env_Initial = {} self._env_added = set() self._attrs = [] def setenv(self, name, value): name = str(name) if name not in self._env_Initial and name not in self._env_added: if name in os.environ: self._env_Initial[name] = os.environ[name] else: self._env_added.add(name) os.environ[name] = str(value) def delenv(self, name, raising=True): name = str(name) if name not in self._env_Initial and name not in self._env_added: if name in os.environ: self._env_Initial[name] = os.environ[name] if name in os.environ: del os.environ[name] elif raising: raise KeyError(name) def setattr(self, target, name, value): had = hasattr(target, name) orig = getattr(target, name, None) self._attrs.append((target, name, had, orig)) setattr(target, name, value) def undo(self): for k, v in self._env_Initial.items(): os.environ[k] = v for k in self._env_added: os.environ.pop(k, None) for target, name, had, orig in reversed(self._attrs): if had: setattr(target, name, orig) else: try: delattr(target, name) except Exception: pass class _ExceptionInfo: def __init__(self, value=None): self.value = value @contextlib.contextmanager def raises(expected_exception, match=None): info = _ExceptionInfo() try: yield info except expected_exception as exc: info.value = exc if match is not None: if not re.search(match, str(exc)): raise AssertionError(f"Regex pattern {match!r} did not match {str(exc)!r}") return raise AssertionError(f"DID NOT RAISE {expected_exception}") def fixture(func=None, *, autouse=False, **kwargs): def _decor(f): f._is_pytest_fixture = True f._pytest_autouse = bool(autouse) return f if func is not None and callable(func): return _decor(func) return _decor class _MarkNamespace: @staticmethod def parametrize(argnames, argvalues): def _decor(f): if not hasattr(f, "_pytest_params"): f._pytest_params = [] f._pytest_params.append((argnames, list(argvalues))) return f return _decor def __getattr__(self, name): def _noop(*args, **kwargs): if len(args) == 1 and callable(args[0]) and not kwargs: return args[0] return lambda f: f return _noop mark = _MarkNamespace() def fail(msg=""): raise AssertionError(msg) def skip(msg=""): raise unittest.SkipTest(msg) def _run_test_file(target_spec): cwd_abs = os.path.abspath(os.getcwd()) if cwd_abs not in sys.path: sys.path.insert(0, cwd_abs) parts = target_spec.split("::", 1) file_path = os.path.abspath(parts[0]) node_filter = parts[1] if len(parts) > 1 else None mod_name = "test_mod_" + Path(file_path).stem spec = importlib.util.spec_from_file_location(mod_name, file_path) if spec is None or spec.loader is None: raise RuntimeError(f"Cannot load {file_path}") mod = importlib.util.module_from_spec(spec) sys.modules[mod_name] = mod spec.loader.exec_module(mod) fixtures = {} autouse_fixtures = [] test_funcs = [] unittest_classes = [] for name, obj in inspect.getmembers(mod): if getattr(obj, "_is_pytest_fixture", False): fixtures[name] = obj if getattr(obj, "_pytest_autouse", False): autouse_fixtures.append(name) elif inspect.isfunction(obj) and name.startswith("test_"): if node_filter is None or name == node_filter: test_funcs.append((name, obj)) elif inspect.isclass(obj) and issubclass(obj, unittest.TestCase) and obj is not unittest.TestCase: if node_filter is None or name == node_filter: unittest_classes.append(obj) passed = 0 failed = 0 collected = 0 failures_out = [] for cls in unittest_classes: suite = unittest.defaultTestLoader.loadTestsFromTestCase(cls) for test_case in suite: collected += 1 res = unittest.TestResult() test_case.run(res) if res.wasSuccessful(): passed += 1 else: failed += 1 for _, err_tb in res.failures + res.errors: failures_out.append((str(test_case), err_tb)) def _resolve_fixture(fname, active_cache, teardowns): if fname in active_cache: return active_cache[fname] if fname == "monkeypatch": mp = _MonkeyPatch() teardowns.append(mp.undo) active_cache[fname] = mp return mp if fname in ("tmp_path", "tmpdir"): td = Path(tempfile.mkdtemp(prefix="pytest_tmp_")) teardowns.append(lambda: shutil.rmtree(td, ignore_errors=True)) active_cache[fname] = td return td if fname not in fixtures: raise RuntimeError(f"Unknown fixture '{fname}'") ffunc = fixtures[fname] sig = inspect.signature(ffunc) fkwargs = {} for pname in sig.parameters: fkwargs[pname] = _resolve_fixture(pname, active_cache, teardowns) if inspect.isgeneratorfunction(ffunc): gen = ffunc(**fkwargs) val = next(gen) def _td(): try: next(gen) except StopIteration: pass teardowns.append(_td) active_cache[fname] = val return val else: val = ffunc(**fkwargs) active_cache[fname] = val return val for tname, tfunc in test_funcs: param_specs = getattr(tfunc, "_pytest_params", []) call_variants = [{}] for argnames, argvalues in param_specs: names = [a.strip() for a in argnames.split(",")] if isinstance(argnames, str) else list(argnames) new_variants = [] for base_v in call_variants: for val in argvalues: v_copy = dict(base_v) if len(names) == 1: v_copy[names[0]] = val else: for n_i, n_k in enumerate(names): v_copy[n_k] = val[n_i] new_variants.append(v_copy) call_variants = new_variants for p_kwargs in call_variants: collected += 1 active_cache = {} teardowns = [] try: for af in autouse_fixtures: _resolve_fixture(af, active_cache, teardowns) sig = inspect.signature(tfunc) call_args = {} for pname in sig.parameters: if pname in p_kwargs: call_args[pname] = p_kwargs[pname] else: call_args[pname] = _resolve_fixture(pname, active_cache, teardowns) tfunc(**call_args) passed += 1 except unittest.SkipTest: pass except Exception: failed += 1 failures_out.append((f"{Path(file_path).name}::{tname}", traceback.format_exc())) finally: for td in reversed(teardowns): try: td() except Exception: pass return collected, passed, failed, failures_out def main(argv=None): args = list(argv if argv is not None else sys.argv[1:]) targets = [a for a in args if not a.startswith("-")] total_collected = 0 total_passed = 0 total_failed = 0 all_failures = [] for t in targets: try: c, p, f, fails = _run_test_file(t) total_collected += c total_passed += p total_failed += f all_failures.extend(fails) except Exception: total_failed += 1 all_failures.append((t, traceback.format_exc())) print(f"collected {total_collected} items") for label, tb_str in all_failures: print(f"FAILED {label}") print(tb_str) if total_collected == 0 and total_failed == 0: print("no tests ran in 0.00s") return 5 summary_parts = [] if total_failed > 0: summary_parts.append(f"{total_failed} failed") if total_passed > 0: summary_parts.append(f"{total_passed} passed") print("============================== " + ", ".join(summary_parts) + " in 0.01s ==============================") return 0 if (total_failed == 0 and total_passed > 0) else 1 if __name__ == "__main__": raise SystemExit(main()) ''' class CandidateVerifier: """ Validates patched candidate files in the sandbox via compilation and targeted tests under strict OS/kernel and C-runtime isolation. """ 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") @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, ) -> 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["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["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 (e.g. 'tests/test_foo.py' or 'tests/test_foo.py::test_bar') resolves canonically inside candidate_dir, is not a symlink/junction, and exists inside candidate_dir. Strictly blocks external test execution. Returns (is_valid, canonical_rel_target_with_nodeid, error_reason). """ 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 @classmethod def _parse_and_validate_focused_command( cls, 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. """ raw_cmd = str(cmd_str or "").strip() if not raw_cmd: return False, [], "Empty focused_test_command" for bad_ch in cls.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" # Normalize leading python / pytest invocation 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 = cls._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" cand_real = os.path.realpath(os.path.abspath(candidate_dir)) canonical_cmd = [ sys.executable, "-m", "pytest", f"--rootdir={cand_real}", f"--confcutdir={cand_real}", ] + 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, error, and collected counts from 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) 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 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 + error_c) if "no tests ran" in combined.lower(): passed_c = 0 collected_c = 0 return { "passed_count": passed_c, "failed_count": failed_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 pytest command inside candidate_dir under: 1. Windows Kernel Job Object (ActiveProcessLimit=1, memory cap, UI restrictions, KILL_ON_JOB_CLOSE) 2. CPython PEP 578 C-runtime sys.addaudithook via sitecustomize.py 3. Sanitized environment (zero credentials, zero external PYTHONPATH) 4. Non-zero test execution verification (0 tests executed -> FAIL). """ 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", "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) 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) # Always provide stdlib-backed pytest runner inside guard_dir so the isolated # child process (which has external PYTHONPATH scrubbed) can execute python -m pytest. pytest_shim_path = os.path.join(guard_dir, "pytest.py") with open(pytest_shim_path, "w", encoding="utf-8") as pf: pf.write(_STDLIB_PYTEST_SHIM) 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, ) h_job = _create_windows_sandbox_job() if os.name == "nt" and not h_job: return { "command": display_cmd, "status": "OS_ISOLATION_UNAVAILABLE", "error": "Failed to create Windows Kernel Job Object for candidate isolation", "passed": False, "returncode": -1, } kernel32 = ctypes.windll.kernel32 if os.name == "nt" else None job_assigned = False try: proc = subprocess.Popen( cmd_tokens, cwd=cand_real, env=clean_env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) if os.name == "nt" and h_job and kernel32: job_assigned = bool( kernel32.AssignProcessToJobObject(wintypes.HANDLE(h_job), wintypes.HANDLE(int(proc._handle))) ) 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)) # Verify that the OS/C-runtime isolation hook physically ran inside the child process 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 = {} isolation_verified = bool( isolation_proof.get("audit_hook_installed") is True and (job_assigned or isolation_proof.get("kernel_job_active") is True) ) counts = self._parse_pytest_counts(stdout_str, stderr_str) tests_passed = counts["passed_count"] tests_failed = counts["failed_count"] + counts["error_count"] if not isolation_verified: passed = False status_str = "OS_ISOLATION_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, "returncode": rc, "status": status_str, "passed": passed, "tests_passed": tests_passed, "tests_failed": tests_failed, "tests_collected": counts["collected_count"], "os_isolation_active": isolation_verified, "os_isolation_mechanism": "WINDOWS_JOB_OBJECT+PEP578_AUDIT_HOOK", "kernel_job_assigned": job_assigned, "isolation_proof": isolation_proof, "credentials_inherited": False, "stripped_env_vars_count": len(stripped_vars), "stdout_snippet": (stdout_str or "")[-600:], "stderr_snippet": (stderr_str or "")[-600:], } 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, focused tests, and regression tests under strict isolation. """ 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. 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_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, ) else: focused_pass = False focused_summary = { "status": "NO_REAL_FOCUSED_TEST_AVAILABLE", "passed": False, "error": "No focused test specified (Fail-Closed)", } # 4. 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: # Fail-closed for Optimus internal components: regressions are mandatory! 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: # External disposable workspace without internal regressions 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 under OS/C-runtime isolation. Never executes external test files outside candidate_dir and never treats 0 executed tests as PASS when targets are specified or required. """ 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} executed_count = 0 total_tests_passed = 0 all_passed = True details: Dict[str, Any] = {} cand_real = os.path.realpath(os.path.abspath(candidate_dir)) 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 = [ sys.executable, "-m", "pytest", f"--rootdir={cand_real}", f"--confcutdir={cand_real}", 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 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_passed": total_tests_passed, "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