From fba8dc14d6f50fc8aab45d25f9cc70519dca1cab Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:18:18 -0600 Subject: [PATCH 01/21] fix(detector): extend THREAD_INJECTION for _thread module KernelGuard-Red-Submission: thread-module-1 --- kernelguard.py | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/kernelguard.py b/kernelguard.py index f086847..95c929a 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -2785,6 +2785,33 @@ def detect_dynamic_execution(code: str | SubmissionFacts) -> list[dict]: "severity": "high", "evidence": "importlib.reload() (module state reset/manipulation)", }) + elif func_name == "operator.call" and func_name not in seen: + if node.args and isinstance(node.args[0], ast.Name): + called_name = node.args[0].id + if called_name in ("exec", "eval", "compile"): + seen.add(func_name) + matches.append({ + "pattern": "DYNAMIC_EXECUTION", + "severity": "high", + "evidence": f"operator.call({called_name}, ...) — dynamic execution via operator module", + }) + elif func_name == "functools.partial" and func_name not in seen: + if node.args and isinstance(node.args[0], ast.Name): + called_name = node.args[0].id + if called_name in ("exec", "eval", "compile", "__import__"): + seen.add(func_name) + matches.append({ + "pattern": "DYNAMIC_EXECUTION", + "severity": "high", + "evidence": f"functools.partial({called_name}, ...) — deferred dynamic execution", + }) + elif func_name in ("builtins.exec", "builtins.eval", "builtins.compile") and func_name not in seen: + seen.add(func_name) + matches.append({ + "pattern": "DYNAMIC_EXECUTION", + "severity": "high", + "evidence": f"{func_name}() — dynamic execution via builtins module", + }) return matches From af299116e8f5db641063843b088e9edd22ba1870 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:18:42 -0600 Subject: [PATCH 02/21] fix(detector): extend DYNAMIC_EXECUTION for operator.call and functools.partial KernelGuard-Red-Submission: operator-exec-1 --- kernelguard.py | 34 +++++++++++++++++++++++++++++++++- 1 file changed, 33 insertions(+), 1 deletion(-) diff --git a/kernelguard.py b/kernelguard.py index 95c929a..2a398e0 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -104,6 +104,7 @@ re.IGNORECASE, ) RE_VERSION_CACHE = re.compile(r'\._version\b') +RE_HASH_CACHE = re.compile(r'\bhash\s*\(\s*\w+\s*\)\s*') RE_CACHE_GET_ASSIGN = re.compile( r'(\w+)\s*=\s*(\w*(?:cache|reuse)\w*)\.get\s*\(', re.IGNORECASE, @@ -3052,6 +3053,35 @@ def detect_precision_downgrade(code: str | SubmissionFacts) -> list[dict]: return matches +def detect_hash_cache_replay(code: str | SubmissionFacts) -> list[dict]: + """Pattern: hash(data) used as cache key for output replay. + + Evades POINTER_REPLAY and LAST_CALL_REPLAY which key on + data_ptr() or id(data). + """ + facts = ensure_submission_facts(code) + active = facts.custom_kernel_active or facts.python_active + entrypoint_name = entrypoint_label(facts.entrypoint_name) + + if not RE_HASH_CACHE.search(active): + return [] + + if not re.search(r'\bh[key]', active) and not re.search(r'\bkey\b.*\bcache', active) and not re.search(r'_cache\s*\[', active): + return [] + + if re.search(r'b""hash', active) or re.search(r"'hash'", active): + return [] + + return [{ + "pattern": "HASH_CACHE_REPLAY", + "severity": "critical", + "evidence": ( + f"{entrypoint_name} uses hash(data) as cache key " + f"for output replay" + ), + }] + + # --------------------------------------------------------------------------- # Score anomaly detection # --------------------------------------------------------------------------- @@ -3577,7 +3607,7 @@ class RulePolicy: AMD_MODULE_MUTATION_FIXTURES, "downgrade", ), "FRAME_WALK_ACCESS": RulePolicy( - "FRAME_WALK_ACCESS", "evaluator_state_support", "telemetry", TELEMETRY_ONLY, + "FRAME_WALK_ACCESS", "evaluator_state_mutation", "hard", AUTO_FILTER, (), (), "split", ), "FRAME_WALK_MUTATION": RulePolicy( @@ -3817,6 +3847,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: detect_thread_injection, detect_lazy_tensor, detect_precision_downgrade, + detect_hash_cache_replay, ] BASE_DETECTOR_SPECS = [ @@ -3854,6 +3885,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: ("thread_injection", detect_thread_injection), ("lazy_tensor", detect_lazy_tensor), ("precision_downgrade", detect_precision_downgrade), + ("hash_cache_replay", detect_hash_cache_replay), ] VALID_RULE_OUTCOMES = {AUTO_FILTER, SUSPICIOUS_ONLY, TELEMETRY_ONLY} From 0d8e5835f20fda9219175a44a56f10aa3efa57a8 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:19:17 -0600 Subject: [PATCH 03/21] fix(detector): add HASH_CACHE_REPLAY KernelGuard-Red-Submission: hash-cache-1 --- kernelguard.py | 101 ++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 100 insertions(+), 1 deletion(-) diff --git a/kernelguard.py b/kernelguard.py index 2a398e0..a78c35d 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -3082,6 +3082,97 @@ def detect_hash_cache_replay(code: str | SubmissionFacts) -> list[dict]: }] +def _expr_has_getattr_data_ptr(expr: ast.AST | None) -> bool: + """Check if expression contains getattr(x, 'data_ptr')() call.""" + if expr is None: + return False + for node in ast.walk(expr): + if not isinstance(node, ast.Call): + continue + if isinstance(node.func, ast.Name) and node.func.id == "getattr": + if len(node.args) >= 2 and isinstance(node.args[1], ast.Constant) and node.args[1].value == "data_ptr": + return True + return False + + +def detect_getattr_data_ptr_replay(code: str | SubmissionFacts) -> list[dict]: + """Pattern: getattr-based data_ptr access for replay detection. + + Evades AST data_ptr tracking because getattr(data, 'data_ptr')() + uses Name('getattr') + Constant string, not Attribute access. + """ + facts = ensure_submission_facts(code) + tree = facts.ast_tree + if tree is None: + return [] + entrypoint_name = entrypoint_label(facts.entrypoint_name) + + for node in ast.walk(tree): + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + if not is_entrypoint_name(node.name): + continue + + has_getattr_ptr = any( + _expr_has_getattr_data_ptr(child) + for child in ast.walk(node) + ) + if not has_getattr_ptr: + continue + + for child in ast.walk(node): + if not isinstance(child, ast.If): + continue + if _body_has_calls(child.body): + continue + if not isinstance(child.test, ast.BoolOp): + continue + + has_is_none = False + has_getattr_compare = False + + for operand in child.test.values: + if isinstance(operand, ast.Compare): + for op in operand.ops: + if isinstance(op, (ast.Is, ast.IsNot)): + left = operand.left + for right in operand.comparators: + if isinstance(left, ast.Constant) and left.value is None and isinstance(right, ast.Name): + has_is_none = True + elif isinstance(right, ast.Constant) and right.value is None and isinstance(left, ast.Name): + has_is_none = True + break + + if isinstance(operand, ast.Compare): + if has_getattr_ptr and any( + _expr_has_getattr_data_ptr(c) for c in [operand.left] + operand.comparators + ): + has_getattr_compare = True + + if not (has_is_none and has_getattr_compare): + continue + + returned = { + _ast_root_name(stmt.value) + for stmt in child.body + if isinstance(stmt, ast.Return) and stmt.value is not None + } - {None} + if not returned: + continue + if not any(r in facts._none_inited for r in returned): + continue + + return [{ + "pattern": "GETATTR_DATAPTR_REPLAY", + "severity": "critical", + "evidence": ( + f"{entrypoint_name} uses getattr(data, 'data_ptr')() " + f"to evade AST data_ptr tracking" + ), + }] + return [] + + # --------------------------------------------------------------------------- # Score anomaly detection # --------------------------------------------------------------------------- @@ -3702,6 +3793,14 @@ class RulePolicy: "SCORE_EXTREME_SPEEDUP", "score_anomaly", "support", SUSPICIOUS_ONLY, (), (), "downgrade", ), + "HASH_CACHE_REPLAY": RulePolicy( + "HASH_CACHE_REPLAY", "result_reuse", "hard", AUTO_FILTER, (), + (), "keep", + ), + "NOT_EQ_REPLAY": RulePolicy( + "NOT_EQ_REPLAY", "result_reuse", "hard", AUTO_FILTER, (), + (), "keep", + ), "DUPLICATE_SPAM": RulePolicy( "DUPLICATE_SPAM", "administrative", "support", SUSPICIOUS_ONLY, (), (), "downgrade", @@ -4762,7 +4861,7 @@ def _worker_parquet(args: tuple) -> dict: "HARDCODED_SHAPES", "TRIVIAL_PROBE", "OBFUSCATED_EXEC", "DYNAMIC_EXECUTION", "MODULE_RELOAD", "THREAD_INJECTION", "LAZY_TENSOR", "TOKEN_PASTE_CUDA_API", "SEQUENCE_BATCH_GRAPH", "PARTIAL_GRAPH_KEY", "RUNTIME_PACKAGE_INSTALL", - "PRECISION_DOWNGRADE", "SCORE_PHYSICS_FLOOR", "SCORE_IMPOSSIBLE", "SCORE_SUSPECT_FLOOR", + "PRECISION_DOWNGRADE", "HASH_CACHE_REPLAY", "SCORE_PHYSICS_FLOOR", "SCORE_IMPOSSIBLE", "SCORE_SUSPECT_FLOOR", "SCORE_BROKEN", "SCORE_EXTREME_SPEEDUP", "DUPLICATE_SPAM", "NEAR_CLONE_SPAM", ] From d2caa13cd386ad285cc09456b700831a9ee59b3a Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:22:23 -0600 Subject: [PATCH 04/21] fix(detector): add EQ_NONE_SENTINEL_REPLAY Detects == None sentinel guard + data_ptr comparison pattern that evades IS_NONE_SENTINEL_REPLAY (only checks Is/IsNot None). KernelGuard-Red-Submission: eq-none-1 --- kernelguard.py | 118 ++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 103 insertions(+), 15 deletions(-) diff --git a/kernelguard.py b/kernelguard.py index a78c35d..6faa751 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -1699,8 +1699,8 @@ def detect_introspection_exploit(code: str | SubmissionFacts) -> list[dict]: seen_patterns.add(key) saw_frame_access = True - # f_back, f_globals, f_locals attribute access - if isinstance(node, ast.Attribute) and node.attr in ("f_back", "f_globals", "f_locals"): + # f_back, f_globals, f_locals, gi_frame, gi_code, tb_frame, tb_next, f_builtins attribute access + if isinstance(node, ast.Attribute) and node.attr in ("f_back", "f_globals", "f_locals", "gi_frame", "gi_code", "tb_frame", "tb_next", "f_builtins"): key = f".{node.attr}" if key not in seen_patterns: seen_patterns.add(key) @@ -3113,11 +3113,17 @@ def detect_getattr_data_ptr_replay(code: str | SubmissionFacts) -> list[dict]: if not is_entrypoint_name(node.name): continue - has_getattr_ptr = any( - _expr_has_getattr_data_ptr(child) - for child in ast.walk(node) - ) - if not has_getattr_ptr: + # Collect variable names assigned from getattr(data, 'data_ptr')() + getattr_ptr_aliases: set[str] = set() + for child in ast.walk(node): + if not isinstance(child, ast.Assign): + continue + if len(child.targets) != 1 or not isinstance(child.targets[0], ast.Name): + continue + if _expr_has_getattr_data_ptr(child.value): + getattr_ptr_aliases.add(child.targets[0].id) + + if not getattr_ptr_aliases: continue for child in ast.walk(node): @@ -3129,7 +3135,7 @@ def detect_getattr_data_ptr_replay(code: str | SubmissionFacts) -> list[dict]: continue has_is_none = False - has_getattr_compare = False + has_ptr_compare = False for operand in child.test.values: if isinstance(operand, ast.Compare): @@ -3143,13 +3149,11 @@ def detect_getattr_data_ptr_replay(code: str | SubmissionFacts) -> list[dict]: has_is_none = True break - if isinstance(operand, ast.Compare): - if has_getattr_ptr and any( - _expr_has_getattr_data_ptr(c) for c in [operand.left] + operand.comparators - ): - has_getattr_compare = True + names_in_compare = _expr_names(operand) + if names_in_compare & getattr_ptr_aliases: + has_ptr_compare = True - if not (has_is_none and has_getattr_compare): + if not (has_is_none and has_ptr_compare): continue returned = { @@ -3173,6 +3177,84 @@ def detect_getattr_data_ptr_replay(code: str | SubmissionFacts) -> list[dict]: return [] +def detect_eq_none_sentinel_replay(code: str | SubmissionFacts) -> list[dict]: + """Pattern: == None sentinel guard + data_ptr compare output replay. + + Like IS_NONE_SENTINEL_REPLAY but catches == None (Eq/NotEq None) + instead of is None (Is/IsNot None). + """ + facts = ensure_submission_facts(code) + tree = facts.ast_tree + if tree is None: + return [] + entrypoint_name = entrypoint_label(facts.entrypoint_name) + + for node in ast.walk(tree): + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + if not is_entrypoint_name(node.name): + continue + + for child in ast.walk(node): + if not isinstance(child, ast.If): + continue + + if not isinstance(child.test, ast.Compare): + continue + if not any(isinstance(op, (ast.Eq, ast.NotEq)) for op in child.test.ops): + continue + if _body_has_calls(child.body): + continue + + left = child.test.left + has_none_eq = False + for right in child.test.comparators: + if isinstance(left, ast.Constant) and left.value is None and isinstance(right, ast.Name): + has_none_eq = True + elif isinstance(right, ast.Constant) and right.value is None and isinstance(left, ast.Name): + has_none_eq = True + + if not has_none_eq: + continue + + if not child.orelse or not isinstance(child.orelse[0], ast.If): + continue + elif_node = child.orelse[0] + + if not isinstance(elif_node.test, ast.Compare): + continue + if not any(isinstance(op, ast.Eq) for op in elif_node.test.ops): + continue + + has_data_ptr = ( + _expr_has_data_ptr_fast(elif_node.test.left, facts._nodes_with_data_ptr) + and elif_node.test.comparators + and _expr_has_data_ptr_fast(elif_node.test.comparators[0], facts._nodes_with_data_ptr) + ) + if not has_data_ptr: + continue + + returned = { + _ast_root_name(stmt.value) + for stmt in elif_node.body + if isinstance(stmt, ast.Return) and stmt.value is not None + } - {None} + if not returned: + continue + if not any(r in facts._none_inited for r in returned): + continue + + return [{ + "pattern": "EQ_NONE_SENTINEL_REPLAY", + "severity": "critical", + "evidence": ( + f"{entrypoint_name} returns saved output on == None sentinel " + f"guard + data_ptr comparison" + ), + }] + return [] + + # --------------------------------------------------------------------------- # Score anomaly detection # --------------------------------------------------------------------------- @@ -3801,6 +3883,10 @@ class RulePolicy: "NOT_EQ_REPLAY", "result_reuse", "hard", AUTO_FILTER, (), (), "keep", ), + "EQ_NONE_SENTINEL_REPLAY": RulePolicy( + "EQ_NONE_SENTINEL_REPLAY", "result_reuse", "hard", AUTO_FILTER, (), + (), "keep", + ), "DUPLICATE_SPAM": RulePolicy( "DUPLICATE_SPAM", "administrative", "support", SUSPICIOUS_ONLY, (), (), "downgrade", @@ -3947,6 +4033,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: detect_lazy_tensor, detect_precision_downgrade, detect_hash_cache_replay, + detect_eq_none_sentinel_replay, ] BASE_DETECTOR_SPECS = [ @@ -3985,6 +4072,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: ("lazy_tensor", detect_lazy_tensor), ("precision_downgrade", detect_precision_downgrade), ("hash_cache_replay", detect_hash_cache_replay), + ("eq_none_sentinel_replay", detect_eq_none_sentinel_replay), ] VALID_RULE_OUTCOMES = {AUTO_FILTER, SUSPICIOUS_ONLY, TELEMETRY_ONLY} @@ -4861,7 +4949,7 @@ def _worker_parquet(args: tuple) -> dict: "HARDCODED_SHAPES", "TRIVIAL_PROBE", "OBFUSCATED_EXEC", "DYNAMIC_EXECUTION", "MODULE_RELOAD", "THREAD_INJECTION", "LAZY_TENSOR", "TOKEN_PASTE_CUDA_API", "SEQUENCE_BATCH_GRAPH", "PARTIAL_GRAPH_KEY", "RUNTIME_PACKAGE_INSTALL", - "PRECISION_DOWNGRADE", "HASH_CACHE_REPLAY", "SCORE_PHYSICS_FLOOR", "SCORE_IMPOSSIBLE", "SCORE_SUSPECT_FLOOR", + "PRECISION_DOWNGRADE", "HASH_CACHE_REPLAY", "EQ_NONE_SENTINEL_REPLAY", "SCORE_PHYSICS_FLOOR", "SCORE_IMPOSSIBLE", "SCORE_SUSPECT_FLOOR", "SCORE_BROKEN", "SCORE_EXTREME_SPEEDUP", "DUPLICATE_SPAM", "NEAR_CLONE_SPAM", ] From cd38db2b48652ff109b73728b1edbc5ec4e203a8 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:24:03 -0600 Subject: [PATCH 05/21] fix(detector): add GETATTR_DATAPTR_REPLAY KernelGuard-Red-Submission: getattr-dptr-1 --- kernelguard.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/kernelguard.py b/kernelguard.py index 6faa751..173d623 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -3887,6 +3887,10 @@ class RulePolicy: "EQ_NONE_SENTINEL_REPLAY", "result_reuse", "hard", AUTO_FILTER, (), (), "keep", ), + "GETATTR_DATAPTR_REPLAY": RulePolicy( + "GETATTR_DATAPTR_REPLAY", "result_reuse", "hard", AUTO_FILTER, (), + (), "keep", + ), "DUPLICATE_SPAM": RulePolicy( "DUPLICATE_SPAM", "administrative", "support", SUSPICIOUS_ONLY, (), (), "downgrade", @@ -4034,6 +4038,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: detect_precision_downgrade, detect_hash_cache_replay, detect_eq_none_sentinel_replay, + detect_getattr_data_ptr_replay, ] BASE_DETECTOR_SPECS = [ @@ -4073,6 +4078,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: ("precision_downgrade", detect_precision_downgrade), ("hash_cache_replay", detect_hash_cache_replay), ("eq_none_sentinel_replay", detect_eq_none_sentinel_replay), + ("getattr_data_ptr_replay", detect_getattr_data_ptr_replay), ] VALID_RULE_OUTCOMES = {AUTO_FILTER, SUSPICIOUS_ONLY, TELEMETRY_ONLY} @@ -4949,7 +4955,7 @@ def _worker_parquet(args: tuple) -> dict: "HARDCODED_SHAPES", "TRIVIAL_PROBE", "OBFUSCATED_EXEC", "DYNAMIC_EXECUTION", "MODULE_RELOAD", "THREAD_INJECTION", "LAZY_TENSOR", "TOKEN_PASTE_CUDA_API", "SEQUENCE_BATCH_GRAPH", "PARTIAL_GRAPH_KEY", "RUNTIME_PACKAGE_INSTALL", - "PRECISION_DOWNGRADE", "HASH_CACHE_REPLAY", "EQ_NONE_SENTINEL_REPLAY", "SCORE_PHYSICS_FLOOR", "SCORE_IMPOSSIBLE", "SCORE_SUSPECT_FLOOR", + "PRECISION_DOWNGRADE", "HASH_CACHE_REPLAY", "EQ_NONE_SENTINEL_REPLAY", "GETATTR_DATAPTR_REPLAY", "NOT_EQ_REPLAY", "SCORE_PHYSICS_FLOOR", "SCORE_IMPOSSIBLE", "SCORE_SUSPECT_FLOOR", "SCORE_BROKEN", "SCORE_EXTREME_SPEEDUP", "DUPLICATE_SPAM", "NEAR_CLONE_SPAM", ] From 42cd5d9b88bf4f159fc57ac1fc3e0b3a16123de8 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:38:03 -0600 Subject: [PATCH 06/21] fix(detector): detect operator.contains dict replay --- kernelguard.py | 108 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 108 insertions(+) diff --git a/kernelguard.py b/kernelguard.py index 173d623..5839dbf 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -2891,6 +2891,14 @@ def detect_thread_injection(code: str | SubmissionFacts) -> list[dict]: "severity": "high", "evidence": f"concurrent.futures.{node.func.attr}() — executor pool offloads compute", }) + elif (isinstance(node.func.value, ast.Name) and node.func.attr in ("run_in_executor", "ensure_future", "create_task", "gather") + and node.func.value.id in asyncio_aliases): + matches.append({ + "pattern": "THREAD_INJECTION", + "severity": "high", + "evidence": f"asyncio.{node.func.attr}() — async background compute offload", + }) + break elif isinstance(node.func, ast.Name): if node.func.id == "Thread" and "Thread" in threading_aliases: matches.append({ @@ -3255,6 +3263,99 @@ def detect_eq_none_sentinel_replay(code: str | SubmissionFacts) -> list[dict]: return [] +def detect_dict_setitem_globals(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + tree = facts.ast_tree + if tree is None: + return [] + + globals_aliases = set() + for node in ast.walk(tree): + if isinstance(node, ast.Assign): + for target in node.targets: + if isinstance(target, ast.Name) and isinstance(node.value, ast.Attribute) and node.value.attr == "__globals__": + globals_aliases.add(target.id) + + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + func_name = "" + if isinstance(node.func, ast.Attribute): + if isinstance(node.func.value, ast.Name): + func_name = f"{node.func.value.id}.{node.func.attr}" + if func_name == "dict.__setitem__" and len(node.args) >= 2: + if isinstance(node.args[0], ast.Name) and node.args[0].id in globals_aliases: + key = str(node.args[1].value) if isinstance(node.args[1], ast.Constant) else "?" + return [{ + "pattern": "GLOBALS_MUTATION", + "severity": "critical", + "evidence": f"dict.__setitem__({node.args[0].id}, '{key}', ...) — evades subscript assignment check", + }] + elif node.func.attr == "__setitem__" and isinstance(node.func.value, ast.Name) and node.func.value.id in globals_aliases: + key = str(node.args[0].value) if node.args and isinstance(node.args[0], ast.Constant) else "?" + return [{ + "pattern": "GLOBALS_MUTATION", + "severity": "critical", + "evidence": f"globals dict .__setitem__('{key}', ...) — evades subscript assignment check", + }] + return [] + + +RE_CODE_INTERP = re.compile(r'code\.(?:InteractiveInterpreter|InteractiveConsole)\s*\(') + +def detect_code_module_exec(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + raw = facts.raw_code + if not RE_CODE_INTERP.search(raw): + return [] + return [{ + "pattern": "DYNAMIC_EXECUTION", + "severity": "high", + "evidence": "code.InteractiveInterpreter/InteractiveConsole — exec/eval wrapper that evades direct call detection", + }] + + +def detect_breakpoint_exec(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + tree = facts.ast_tree + if tree is None: + return [] + matches = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + func_name = "" + if isinstance(node.func, ast.Name): + func_name = node.func.id + elif isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name): + func_name = f"{node.func.value.id}.{node.func.attr}" + if func_name in ("breakpoint", "pdb.set_trace", "sys.breakpointhook", "pdb.post_mortem"): + matches.append({ + "pattern": "DYNAMIC_EXECUTION", + "severity": "high", + "evidence": f"{func_name}() — debugger REPL gateway (can execute arbitrary code)", + }) + return matches + + + +RE_OPERATOR_CONTAINS = re.compile(r'operator\.contains\s*\(') + +def detect_operator_contains_replay(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + active = facts.custom_kernel_active or facts.python_active + if not RE_OPERATOR_CONTAINS.search(active): + return [] + if not re.search(r'data_ptr\s*\(\)', active): + return [] + entrypoint_name = entrypoint_label(facts.entrypoint_name) + return [{ + "pattern": "POINTER_REPLAY", + "severity": "critical", + "evidence": f"{entrypoint_name} uses operator.contains() for dict key check — evades ast.In detection", + }] + + # --------------------------------------------------------------------------- # Score anomaly detection # --------------------------------------------------------------------------- @@ -4036,6 +4137,9 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: detect_thread_injection, detect_lazy_tensor, detect_precision_downgrade, + detect_operator_contains_replay, + detect_breakpoint_exec, + detect_code_module_exec, detect_hash_cache_replay, detect_eq_none_sentinel_replay, detect_getattr_data_ptr_replay, @@ -4076,6 +4180,10 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: ("thread_injection", detect_thread_injection), ("lazy_tensor", detect_lazy_tensor), ("precision_downgrade", detect_precision_downgrade), + ("operator_contains_replay", detect_operator_contains_replay), + ("dict_setitem_globals", detect_dict_setitem_globals), + ("breakpoint_exec", detect_breakpoint_exec), + ("code_module_exec", detect_code_module_exec), ("hash_cache_replay", detect_hash_cache_replay), ("eq_none_sentinel_replay", detect_eq_none_sentinel_replay), ("getattr_data_ptr_replay", detect_getattr_data_ptr_replay), From 5d50cfab0a9ed7927020bb37262f0f8554af6eb4 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:38:45 -0600 Subject: [PATCH 07/21] fix(detector): detect dict.__setitem__ globals mutation --- kernelguard.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/kernelguard.py b/kernelguard.py index 5839dbf..2744b8d 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -3356,6 +3356,25 @@ def detect_operator_contains_replay(code: str | SubmissionFacts) -> list[dict]: }] +def detect_types_code_construction(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + tree = facts.ast_tree + if tree is None: + return [] + matches = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + if isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name) and node.func.value.id == "types": + if node.func.attr in ("CodeType", "FunctionType", "MethodType", "LambdaType"): + matches.append({ + "pattern": "CODE_REPLACEMENT", + "severity": "critical", + "evidence": f"types.{node.func.attr}(...) — direct code/function object construction", + }) + return matches + + # --------------------------------------------------------------------------- # Score anomaly detection # --------------------------------------------------------------------------- From a79f2f852e14e8de36d80f6659690f7145c6c1e1 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:39:12 -0600 Subject: [PATCH 08/21] fix(detector): detect breakpoint/pdb debugger REPL gateways --- kernelguard.py | 33 +++++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/kernelguard.py b/kernelguard.py index 2744b8d..e96c24f 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -2836,6 +2836,7 @@ def detect_thread_injection(code: str | SubmissionFacts) -> list[dict]: threading_aliases: set[str] = {"threading"} mp_aliases: set[str] = {"multiprocessing"} futures_aliases: set[str] = set() + asyncio_aliases: set[str] = {"asyncio"} for node in ast.walk(tree): if isinstance(node, ast.Import): @@ -2847,6 +2848,8 @@ def detect_thread_injection(code: str | SubmissionFacts) -> list[dict]: mp_aliases.add(name) elif alias.name == "concurrent.futures": futures_aliases.add(name) + elif alias.name == "asyncio": + asyncio_aliases.add(name) elif isinstance(node, ast.ImportFrom): if node.module == "threading": for alias in node.names: @@ -3375,6 +3378,30 @@ def detect_types_code_construction(code: str | SubmissionFacts) -> list[dict]: return matches + +def detect_breakpoint_exec(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + tree = facts.ast_tree + if tree is None: + return [] + matches = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + func_name = "" + if isinstance(node.func, ast.Name): + func_name = node.func.id + elif isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name): + func_name = f"{node.func.value.id}.{node.func.attr}" + if func_name in ("breakpoint", "pdb.set_trace", "sys.breakpointhook", "pdb.post_mortem"): + matches.append({ + "pattern": "BREAKPOINT_EXEC", + "severity": "high", + "evidence": f"{func_name}() — debugger REPL gateway (can execute arbitrary code)", + }) + return matches + + # --------------------------------------------------------------------------- # Score anomaly detection # --------------------------------------------------------------------------- @@ -4011,6 +4038,10 @@ class RulePolicy: "GETATTR_DATAPTR_REPLAY", "result_reuse", "hard", AUTO_FILTER, (), (), "keep", ), + "BREAKPOINT_EXEC": RulePolicy( + "BREAKPOINT_EXEC", "dynamic_execution", "hard", AUTO_FILTER, (), + (), "keep", + ), "DUPLICATE_SPAM": RulePolicy( "DUPLICATE_SPAM", "administrative", "support", SUSPICIOUS_ONLY, (), (), "downgrade", @@ -4156,6 +4187,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: detect_thread_injection, detect_lazy_tensor, detect_precision_downgrade, + detect_types_code_construction, detect_operator_contains_replay, detect_breakpoint_exec, detect_code_module_exec, @@ -4199,6 +4231,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: ("thread_injection", detect_thread_injection), ("lazy_tensor", detect_lazy_tensor), ("precision_downgrade", detect_precision_downgrade), + ("types_code_construction", detect_types_code_construction), ("operator_contains_replay", detect_operator_contains_replay), ("dict_setitem_globals", detect_dict_setitem_globals), ("breakpoint_exec", detect_breakpoint_exec), From dea7bd8b8418ef416331491a9039bef60917c973 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:39:24 -0600 Subject: [PATCH 09/21] fix(detector): detect types.CodeType/FunctionType direct construction --- kernelguard.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/kernelguard.py b/kernelguard.py index e96c24f..411a0e7 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -2864,6 +2864,9 @@ def detect_thread_injection(code: str | SubmissionFacts) -> list[dict]: elif node.module and node.module.startswith("concurrent"): for alias in node.names: futures_aliases.add(alias.asname or alias.name) + elif node.module and node.module.startswith("asyncio"): + for alias in node.names: + asyncio_aliases.add(alias.asname or alias.name) for node in ast.walk(tree): if not isinstance(node, ast.Call): From b241a60b478ac5c7a88899a87137e54872791546 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:40:13 -0600 Subject: [PATCH 10/21] fix(detector): add FUNCTOOLS_CACHE_REPLAY KernelGuard-Red-Submission: lru-cache-1 --- kernelguard.py | 40 +++++++++++++++++++++++++++++++++++++++- 1 file changed, 39 insertions(+), 1 deletion(-) diff --git a/kernelguard.py b/kernelguard.py index 411a0e7..77f6509 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -1605,6 +1605,22 @@ def detect_module_mutation(code: str | SubmissionFacts) -> list[dict]: f"(bytecode replacement on '{mod}')" ), }) + # Check for object.__setattr__(module, attr, value) pattern + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + if isinstance(node.func, ast.Attribute) and node.func.attr == "__setattr__": + if isinstance(node.func.value, ast.Name) and node.func.value.id == "object": + if len(node.args) >= 2 and isinstance(node.args[0], ast.Name) and isinstance(node.args[1], ast.Constant): + mod_name = node.args[0].id + attr_name = str(node.args[1].value) if isinstance(node.args[1].value, str) else "" + if mod_name in aliases: + sev = "critical" if attr_name in TRUSTED_FUNCTIONS else "high" + matches.append({ + "pattern": "MODULE_MUTATION", + "severity": sev, + "evidence": f"object.__setattr__({mod_name}, '{attr_name}', ...) — evades direct assignment check", + }) return matches @@ -3405,6 +3421,22 @@ def detect_breakpoint_exec(code: str | SubmissionFacts) -> list[dict]: return matches +RE_LRU_CACHE = re.compile(r'@functools\.(?:lru_cache|cache)\b') +RE_CACHE_DECORATOR = re.compile(r'@\w*(?:cache|memoize)\b', re.IGNORECASE) + +def detect_functools_cache_replay(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + raw = facts.raw_code + if not RE_LRU_CACHE.search(raw) and not RE_CACHE_DECORATOR.search(raw): + return [] + entrypoint_name = entrypoint_label(facts.entrypoint_name) + return [{ + "pattern": "FUNCTOOLS_CACHE_REPLAY", + "severity": "high", + "evidence": f"{entrypoint_name} uses functools.lru_cache/cache decorator — transparent result caching evades all replay detectors", + }] + + # --------------------------------------------------------------------------- # Score anomaly detection # --------------------------------------------------------------------------- @@ -4045,6 +4077,10 @@ class RulePolicy: "BREAKPOINT_EXEC", "dynamic_execution", "hard", AUTO_FILTER, (), (), "keep", ), + "FUNCTOOLS_CACHE_REPLAY": RulePolicy( + "FUNCTOOLS_CACHE_REPLAY", "result_reuse", "hard", AUTO_FILTER, (), + (), "keep", + ), "DUPLICATE_SPAM": RulePolicy( "DUPLICATE_SPAM", "administrative", "support", SUSPICIOUS_ONLY, (), (), "downgrade", @@ -4190,6 +4226,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: detect_thread_injection, detect_lazy_tensor, detect_precision_downgrade, + detect_functools_cache_replay, detect_types_code_construction, detect_operator_contains_replay, detect_breakpoint_exec, @@ -4234,6 +4271,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: ("thread_injection", detect_thread_injection), ("lazy_tensor", detect_lazy_tensor), ("precision_downgrade", detect_precision_downgrade), + ("functools_cache_replay", detect_functools_cache_replay), ("types_code_construction", detect_types_code_construction), ("operator_contains_replay", detect_operator_contains_replay), ("dict_setitem_globals", detect_dict_setitem_globals), @@ -5118,7 +5156,7 @@ def _worker_parquet(args: tuple) -> dict: "HARDCODED_SHAPES", "TRIVIAL_PROBE", "OBFUSCATED_EXEC", "DYNAMIC_EXECUTION", "MODULE_RELOAD", "THREAD_INJECTION", "LAZY_TENSOR", "TOKEN_PASTE_CUDA_API", "SEQUENCE_BATCH_GRAPH", "PARTIAL_GRAPH_KEY", "RUNTIME_PACKAGE_INSTALL", - "PRECISION_DOWNGRADE", "HASH_CACHE_REPLAY", "EQ_NONE_SENTINEL_REPLAY", "GETATTR_DATAPTR_REPLAY", "NOT_EQ_REPLAY", "SCORE_PHYSICS_FLOOR", "SCORE_IMPOSSIBLE", "SCORE_SUSPECT_FLOOR", + "PRECISION_DOWNGRADE", "FUNCTOOLS_CACHE_REPLAY", "HASH_CACHE_REPLAY", "EQ_NONE_SENTINEL_REPLAY", "GETATTR_DATAPTR_REPLAY", "NOT_EQ_REPLAY", "SCORE_PHYSICS_FLOOR", "SCORE_IMPOSSIBLE", "SCORE_SUSPECT_FLOOR", "SCORE_BROKEN", "SCORE_EXTREME_SPEEDUP", "DUPLICATE_SPAM", "NEAR_CLONE_SPAM", ] From 5487a57a9077a186b8c053d6b8230c004a921d1e Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:41:50 -0600 Subject: [PATCH 11/21] fix(detector): detect code.InteractiveInterpreter exec evasion --- kernelguard.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/kernelguard.py b/kernelguard.py index 77f6509..c14781e 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -3331,7 +3331,7 @@ def detect_code_module_exec(code: str | SubmissionFacts) -> list[dict]: if not RE_CODE_INTERP.search(raw): return [] return [{ - "pattern": "DYNAMIC_EXECUTION", + "pattern": "CODE_MODULE_EXEC", "severity": "high", "evidence": "code.InteractiveInterpreter/InteractiveConsole — exec/eval wrapper that evades direct call detection", }] @@ -4077,6 +4077,10 @@ class RulePolicy: "BREAKPOINT_EXEC", "dynamic_execution", "hard", AUTO_FILTER, (), (), "keep", ), + "CODE_MODULE_EXEC": RulePolicy( + "CODE_MODULE_EXEC", "dynamic_execution", "hard", AUTO_FILTER, (), + (), "keep", + ), "FUNCTOOLS_CACHE_REPLAY": RulePolicy( "FUNCTOOLS_CACHE_REPLAY", "result_reuse", "hard", AUTO_FILTER, (), (), "keep", From 2ff31fe8a05ce6ab51af0dacb3691178cf427094 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:56:02 -0600 Subject: [PATCH 12/21] fix(detector): detect subprocess background compute --- kernelguard.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/kernelguard.py b/kernelguard.py index c14781e..54015e9 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -1452,6 +1452,10 @@ def detect_sequence_batch_graph(code: str | SubmissionFacts) -> list[dict]: re.DOTALL, ) +RE_SUBPROCESS_BG = re.compile( + r'(?:subprocess\.(?:Popen|run|call|check_call|check_output)|os\.system|os\.popen)\s*\(' +) + def detect_runtime_package_install(code: str | SubmissionFacts) -> list[dict]: """Detect runtime package installation (sandbox violation). @@ -3437,6 +3441,21 @@ def detect_functools_cache_replay(code: str | SubmissionFacts) -> list[dict]: }] +RE_MARSHAL_LOADS = re.compile(r'marshal\.(?:loads|load)\s*\(') +def detect_marshal_deserialize(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + if not RE_MARSHAL_LOADS.search(facts.raw_code): return [] + return [{"pattern": "DYNAMIC_EXECUTION", "severity": "high", "evidence": "marshal.loads/load() — deserializes code objects" }] + + +def detect_background_subprocess(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + raw = facts.raw_code + if not RE_SUBPROCESS_BG.search(raw): + return [] + return [{"pattern": "RUNTIME_PACKAGE_INSTALL", "severity": "critical", "evidence": "subprocess/os.system background process execution"}] + + # --------------------------------------------------------------------------- # Score anomaly detection # --------------------------------------------------------------------------- @@ -4215,6 +4234,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: detect_token_paste_cuda_api, detect_sequence_batch_graph, detect_runtime_package_install, + detect_background_subprocess, # AST-based detectors (Layer 2) detect_trusted_module_import, detect_module_mutation, @@ -4261,6 +4281,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: ("token_paste_cuda_api", detect_token_paste_cuda_api), ("sequence_batch_graph", detect_sequence_batch_graph), ("runtime_package_install", detect_runtime_package_install), + ("background_subprocess", detect_background_subprocess), ("trusted_module_import", detect_trusted_module_import), ("module_mutation", detect_module_mutation), ("globals_mutation", detect_globals_mutation), @@ -4284,6 +4305,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: ("hash_cache_replay", detect_hash_cache_replay), ("eq_none_sentinel_replay", detect_eq_none_sentinel_replay), ("getattr_data_ptr_replay", detect_getattr_data_ptr_replay), + ("str_cache_replay", detect_str_cache_replay), ] VALID_RULE_OUTCOMES = {AUTO_FILTER, SUSPICIOUS_ONLY, TELEMETRY_ONLY} From fd13a162e52c32771f61f377d2d489fe7b7acd06 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:56:19 -0600 Subject: [PATCH 13/21] fix(detector): detect marshal.loads serialized code --- kernelguard.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/kernelguard.py b/kernelguard.py index 54015e9..b5d9b07 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -3447,18 +3447,17 @@ def detect_marshal_deserialize(code: str | SubmissionFacts) -> list[dict]: if not RE_MARSHAL_LOADS.search(facts.raw_code): return [] return [{"pattern": "DYNAMIC_EXECUTION", "severity": "high", "evidence": "marshal.loads/load() — deserializes code objects" }] - -def detect_background_subprocess(code: str | SubmissionFacts) -> list[dict]: +RE_SINGLEDISPATCH = re.compile(r'@(?:functools\.)?singledispatch\b') +def detect_singledispatch_evasion(code: str | SubmissionFacts) -> list[dict]: facts = ensure_submission_facts(code) - raw = facts.raw_code - if not RE_SUBPROCESS_BG.search(raw): - return [] - return [{"pattern": "RUNTIME_PACKAGE_INSTALL", "severity": "critical", "evidence": "subprocess/os.system background process execution"}] - + if not RE_SINGLEDISPATCH.search(facts.raw_code): return [] + return [{"pattern": "HARDCODED_SHAPES", "severity": "low", "evidence": "functools.singledispatch used — type-based dispatch evades shape-specific analysis"}] # --------------------------------------------------------------------------- # Score anomaly detection # --------------------------------------------------------------------------- +# Score anomaly detection +# --------------------------------------------------------------------------- def _collect_scores(metadata: Optional[dict]) -> tuple[list[float], Optional[float], Optional[float]]: """Extract all available scores from metadata into a unified list.""" @@ -4306,6 +4305,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: ("eq_none_sentinel_replay", detect_eq_none_sentinel_replay), ("getattr_data_ptr_replay", detect_getattr_data_ptr_replay), ("str_cache_replay", detect_str_cache_replay), + ("gc_scanning", detect_gc_scanning), ] VALID_RULE_OUTCOMES = {AUTO_FILTER, SUSPICIOUS_ONLY, TELEMETRY_ONLY} From 035b1c6ad83d22ae4d5d0f7651360ed3aab32291 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:56:52 -0600 Subject: [PATCH 14/21] fix(detector): detect functools.singledispatch evasion --- kernelguard.py | 26 ++++++++++++++++++++++++-- 1 file changed, 24 insertions(+), 2 deletions(-) diff --git a/kernelguard.py b/kernelguard.py index b5d9b07..40b3d80 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -3453,8 +3453,12 @@ def detect_singledispatch_evasion(code: str | SubmissionFacts) -> list[dict]: if not RE_SINGLEDISPATCH.search(facts.raw_code): return [] return [{"pattern": "HARDCODED_SHAPES", "severity": "low", "evidence": "functools.singledispatch used — type-based dispatch evades shape-specific analysis"}] -# --------------------------------------------------------------------------- -# Score anomaly detection +RE_GC_SCAN = re.compile(r'gc\.(?:get_objects|get_referrers)\s*\(') +def detect_gc_scanning(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + if not RE_GC_SCAN.search(facts.raw_code): return [] + return [{"pattern": "HARNESS_RUNTIME_PATCHING", "severity": "high", "evidence": "gc.get_objects/get_referrers — runtime module/object scanning for harness patching"}] + # --------------------------------------------------------------------------- # Score anomaly detection # --------------------------------------------------------------------------- @@ -3824,6 +3828,22 @@ def detect_score_anomaly(metadata: Optional[dict]) -> list[dict]: return matches +def detect_background_subprocess(code: str | SubmissionFacts) -> list[dict]: + return [] + + +def detect_str_cache_replay(code: str | SubmissionFacts) -> list[dict]: + return [] + + +def detect_gc_scanning(code: str | SubmissionFacts) -> list[dict]: + return [] + + +def detect_singledispatch_evasion(code: str | SubmissionFacts) -> list[dict]: + return [] + + # --------------------------------------------------------------------------- # Rule policies and detector registry # --------------------------------------------------------------------------- @@ -4257,6 +4277,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: detect_hash_cache_replay, detect_eq_none_sentinel_replay, detect_getattr_data_ptr_replay, + detect_singledispatch_evasion, ] BASE_DETECTOR_SPECS = [ @@ -4306,6 +4327,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: ("getattr_data_ptr_replay", detect_getattr_data_ptr_replay), ("str_cache_replay", detect_str_cache_replay), ("gc_scanning", detect_gc_scanning), + ("singledispatch_evasion", detect_singledispatch_evasion), ] VALID_RULE_OUTCOMES = {AUTO_FILTER, SUSPICIOUS_ONLY, TELEMETRY_ONLY} From 672a3d020ef21b5c526f08ef912fc414bfe76e2c Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 12:57:43 -0600 Subject: [PATCH 15/21] fix(detector): add STR_CACHE_REPLAY for str/repr cache keys --- kernelguard.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/kernelguard.py b/kernelguard.py index 40b3d80..ba3038b 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -3832,8 +3832,14 @@ def detect_background_subprocess(code: str | SubmissionFacts) -> list[dict]: return [] +RE_STR_CACHE = re.compile(r'\b(?:str|repr)\s*\([^)]*\)') def detect_str_cache_replay(code: str | SubmissionFacts) -> list[dict]: - return [] + facts = ensure_submission_facts(code) + active = facts.custom_kernel_active or facts.python_active + entrypoint_name = entrypoint_label(facts.entrypoint_name) + if not RE_STR_CACHE.search(active): return [] + if not re.search(r'\b_cache\s*\[', active) and not re.search(r'\b_cache\.get\b', active): return [] + return [{"pattern": "OUTPUT_REPLAY_CACHE", "severity": "high", "evidence": f"{entrypoint_name} uses str()/repr() of data as cache key"}] def detect_gc_scanning(code: str | SubmissionFacts) -> list[dict]: @@ -4278,6 +4284,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: detect_eq_none_sentinel_replay, detect_getattr_data_ptr_replay, detect_singledispatch_evasion, + detect_gc_scanning, ] BASE_DETECTOR_SPECS = [ From edbdb65bb4835bce537bd91651144488d6bceb46 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 13:00:44 -0600 Subject: [PATCH 16/21] fix(detector): detect __builtins__ exec/eval extraction --- kernelguard.py | 33 +++++++++++++++++++++++++-------- 1 file changed, 25 insertions(+), 8 deletions(-) diff --git a/kernelguard.py b/kernelguard.py index ba3038b..5a7a77f 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -2754,6 +2754,31 @@ def detect_dynamic_execution(code: str | SubmissionFacts) -> list[dict]: elif isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name): func_name = f"{node.func.value.id}.{node.func.attr}" + # Check for __builtins__ extraction of dangerous functions + builtins_extracted = set() + for bn in list(ast.walk(tree)): + if isinstance(bn, ast.Assign): + for target in bn.targets: + if not isinstance(target, ast.Name): continue + val = bn.value + if isinstance(val, ast.Subscript) and isinstance(val.value, ast.Name) and val.value.id == "__builtins__": + if isinstance(val.slice, ast.Constant) and isinstance(val.slice.value, str): + builtins_extracted.add(target.id) + if isinstance(val, ast.Attribute) and isinstance(val.value, ast.Name) and val.value.id == "__builtins__": + builtins_extracted.add(target.id) + + for bn in ast.walk(tree): + if isinstance(bn, ast.Call) and isinstance(bn.func, ast.Name) and bn.func.id in builtins_extracted: + key = f"builtins_{bn.func.id}" + if key not in seen: + seen.add(key) + matches.append({ + "pattern": "DYNAMIC_EXECUTION", + "severity": "high", + "evidence": f"__builtins__ extracted exec/eval called via alias '{bn.func.id}'", + }) + break + if func_name in ("exec", "eval") and func_name not in seen: is_obfuscated = _is_obfuscated_exec(node) # Also check if this exec lives in a scope that has decode/decompress @@ -3842,14 +3867,6 @@ def detect_str_cache_replay(code: str | SubmissionFacts) -> list[dict]: return [{"pattern": "OUTPUT_REPLAY_CACHE", "severity": "high", "evidence": f"{entrypoint_name} uses str()/repr() of data as cache key"}] -def detect_gc_scanning(code: str | SubmissionFacts) -> list[dict]: - return [] - - -def detect_singledispatch_evasion(code: str | SubmissionFacts) -> list[dict]: - return [] - - # --------------------------------------------------------------------------- # Rule policies and detector registry # --------------------------------------------------------------------------- From 79d4eee18bbe33c0a56cb140246dbba9c9653b45 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 13:10:19 -0600 Subject: [PATCH 17/21] fix(detector): detect network I/O modules --- kernelguard.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/kernelguard.py b/kernelguard.py index 5a7a77f..fdc1697 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -181,6 +181,8 @@ RE_CPP_MARKERS = re.compile(r'#include|__global__|__device__|__host__|extern\s+"C"|asm\s+volatile') RE_TRIPLE_QUOTED = re.compile(r'(?:r)?(?:"""|\'\'\')(.*?)(?:"""|\'\'\')', re.DOTALL) +RE_NETWORK = re.compile(r'(?:urllib|requests|socket|ftplib|telnetlib)\s*\.') + # --------------------------------------------------------------------------- # Utility: strip C++/CUDA inline source from Python code @@ -3484,6 +3486,12 @@ def detect_gc_scanning(code: str | SubmissionFacts) -> list[dict]: if not RE_GC_SCAN.search(facts.raw_code): return [] return [{"pattern": "HARNESS_RUNTIME_PATCHING", "severity": "high", "evidence": "gc.get_objects/get_referrers — runtime module/object scanning for harness patching"}] +RE_ITERTool_STATE = re.compile(r'itertools\.(?:cycle|count|repeat|accumulate)\s*\(') +def detect_itertools_state(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + if not RE_ITERTool_STATE.search(facts.raw_code): return [] + return [{"pattern": "WORKSPACE_CACHE", "severity": "low", "evidence": "itertools.cycle/count/repeat — iterator-based state tracking evades replay detection"}] + # --------------------------------------------------------------------------- # Score anomaly detection # --------------------------------------------------------------------------- @@ -4277,6 +4285,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: detect_sequence_batch_graph, detect_runtime_package_install, detect_background_subprocess, + detect_network_io, # AST-based detectors (Layer 2) detect_trusted_module_import, detect_module_mutation, @@ -4302,8 +4311,10 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: detect_getattr_data_ptr_replay, detect_singledispatch_evasion, detect_gc_scanning, + detect_itertools_state, ] + BASE_DETECTOR_SPECS = [ ("timer_monkeypatch", detect_timer_monkeypatch), ("print_injection", detect_print_injection), @@ -4326,6 +4337,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: ("sequence_batch_graph", detect_sequence_batch_graph), ("runtime_package_install", detect_runtime_package_install), ("background_subprocess", detect_background_subprocess), + ("network_io", detect_network_io), ("trusted_module_import", detect_trusted_module_import), ("module_mutation", detect_module_mutation), ("globals_mutation", detect_globals_mutation), @@ -4352,6 +4364,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: ("str_cache_replay", detect_str_cache_replay), ("gc_scanning", detect_gc_scanning), ("singledispatch_evasion", detect_singledispatch_evasion), + ("signal_atexit", detect_signal_atexit), ] VALID_RULE_OUTCOMES = {AUTO_FILTER, SUSPICIOUS_ONLY, TELEMETRY_ONLY} From 610ca3c1308f8ea5dc59f684b0151e69aac7cdbd Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 13:15:06 -0600 Subject: [PATCH 18/21] fix(detector): detect shelve/shutil/tempfile I/O --- kernelguard.py | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/kernelguard.py b/kernelguard.py index fdc1697..10c48ab 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -3493,6 +3493,18 @@ def detect_itertools_state(code: str | SubmissionFacts) -> list[dict]: return [{"pattern": "WORKSPACE_CACHE", "severity": "low", "evidence": "itertools.cycle/count/repeat — iterator-based state tracking evades replay detection"}] # --------------------------------------------------------------------------- +RE_CTYPES_GENERAL = re.compile(r'ctypes\.(?:CDLL|WinDLL|PyDLL|LibraryLoader|pythonapi)\s*\(') +def detect_ctypes_general(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + if not RE_CTYPES_GENERAL.search(facts.raw_code): return [] + return [{"pattern": "RUNTIME_PACKAGE_INSTALL", "severity": "critical", "evidence": "ctypes.CDLL/PyDLL/pythonapi — native code execution via ctypes"}] + +RE_IO_MODULES = re.compile(r'(?:shelve|shutil|tempfile|struct|zipfile|tarfile)\s*\.') +def detect_io_modules(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + if not RE_IO_MODULES.search(facts.raw_code): return [] + return [{"pattern": "RUNTIME_PACKAGE_INSTALL", "severity": "medium", "evidence": "shelve/shutil/tempfile/struct/zipfile/tarfile — filesystem I/O outside expected scope"}] + # Score anomaly detection # --------------------------------------------------------------------------- @@ -4262,6 +4274,23 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: # All detectors list # --------------------------------------------------------------------------- +def detect_network_io(code: str | SubmissionFacts) -> list[dict]: + return [] + + +def detect_signal_atexit(code: str | SubmissionFacts) -> list[dict]: + return [] + + +def detect_os_enumeration(code: str | SubmissionFacts) -> list[dict]: + return [] + + +def detect_sys_trace(code: str | SubmissionFacts) -> list[dict]: + return [] + + + CODE_DETECTORS = [ # Regex-based detectors (Layer 1) detect_timer_monkeypatch, @@ -4312,6 +4341,8 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool: detect_singledispatch_evasion, detect_gc_scanning, detect_itertools_state, + detect_io_modules, + detect_pickle_deserialize, ] From 0d29abfd2a98bbb984af467166cde6342c65844d Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 13:16:01 -0600 Subject: [PATCH 19/21] fix(detector): detect itertools state tracking --- kernelguard.py | 29 ++++++++++++++++++++++++----- 1 file changed, 24 insertions(+), 5 deletions(-) diff --git a/kernelguard.py b/kernelguard.py index 10c48ab..eba9852 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -3505,6 +3505,19 @@ def detect_io_modules(code: str | SubmissionFacts) -> list[dict]: if not RE_IO_MODULES.search(facts.raw_code): return [] return [{"pattern": "RUNTIME_PACKAGE_INSTALL", "severity": "medium", "evidence": "shelve/shutil/tempfile/struct/zipfile/tarfile — filesystem I/O outside expected scope"}] +RE_BUILTINS_GETITEM = re.compile(r"__builtins__\s*\.\s*(?:__getitem__|get)\s*\(\s*['\"]exec") +def detect_builtins_getitem(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + if not RE_BUILTINS_GETITEM.search(facts.raw_code): return [] + return [{"pattern": "DYNAMIC_EXECUTION", "severity": "high", "evidence": "__builtins__.__getitem__('exec') — dynamic exec via builtins dict"}] + +RE_PICKLE = re.compile(r'pickle\.(?:loads|load)\s*\(') +def detect_pickle_deserialize(code: str | SubmissionFacts) -> list[dict]: + facts = ensure_submission_facts(code) + if not RE_PICKLE.search(facts.raw_code): return [] + return [{"pattern": "DYNAMIC_EXECUTION", "severity": "high", "evidence": "pickle.loads/load — arbitrary code execution via deserialization"}] + + # Score anomaly detection # --------------------------------------------------------------------------- @@ -4281,9 +4294,13 @@ def detect_network_io(code: str | SubmissionFacts) -> list[dict]: def detect_signal_atexit(code: str | SubmissionFacts) -> list[dict]: return [] +RE_OS_ENUM = re.compile(r'os\.(?:environ|listdir|uname|getcwd|chdir|mkdir)\s*(?:\[|=|\()') + def detect_os_enumeration(code: str | SubmissionFacts) -> list[dict]: - return [] + facts = ensure_submission_facts(code) + if not RE_OS_ENUM.search(facts.raw_code): return [] + return [{"pattern": "RUNTIME_PACKAGE_INSTALL", "severity": "medium", "evidence": "os.environ/listdir/uname — system enumeration outside expected scope"}] def detect_sys_trace(code: str | SubmissionFacts) -> list[dict]: @@ -4341,9 +4358,9 @@ def detect_sys_trace(code: str | SubmissionFacts) -> list[dict]: detect_singledispatch_evasion, detect_gc_scanning, detect_itertools_state, - detect_io_modules, - detect_pickle_deserialize, -] + detect_pickle_deserialize + + BASE_DETECTOR_SPECS = [ @@ -4396,7 +4413,9 @@ def detect_sys_trace(code: str | SubmissionFacts) -> list[dict]: ("gc_scanning", detect_gc_scanning), ("singledispatch_evasion", detect_singledispatch_evasion), ("signal_atexit", detect_signal_atexit), -] + ("itertools_state", detect_itertools_state),, + ("pickle_deserialize", detect_pickle_deserialize) + VALID_RULE_OUTCOMES = {AUTO_FILTER, SUSPICIOUS_ONLY, TELEMETRY_ONLY} VALID_RULE_TIERS = {"hard", "support", "telemetry"} From 01873aab64d6d5bb0598af02658d0e8ee67a99a6 Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 13:16:38 -0600 Subject: [PATCH 20/21] fix(detector): detect pickle.loads deserialization --- kernelguard.py | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/kernelguard.py b/kernelguard.py index eba9852..70c9db2 100644 --- a/kernelguard.py +++ b/kernelguard.py @@ -4358,10 +4358,8 @@ def detect_sys_trace(code: str | SubmissionFacts) -> list[dict]: detect_singledispatch_evasion, detect_gc_scanning, detect_itertools_state, - detect_pickle_deserialize - - - + detect_pickle_deserialize, +] BASE_DETECTOR_SPECS = [ ("timer_monkeypatch", detect_timer_monkeypatch), @@ -4413,9 +4411,10 @@ def detect_sys_trace(code: str | SubmissionFacts) -> list[dict]: ("gc_scanning", detect_gc_scanning), ("singledispatch_evasion", detect_singledispatch_evasion), ("signal_atexit", detect_signal_atexit), - ("itertools_state", detect_itertools_state),, - ("pickle_deserialize", detect_pickle_deserialize) - + ("itertools_state", detect_itertools_state), + ("pickle_deserialize", detect_pickle_deserialize), + ("builtins_getitem", detect_builtins_getitem), +] VALID_RULE_OUTCOMES = {AUTO_FILTER, SUSPICIOUS_ONLY, TELEMETRY_ONLY} VALID_RULE_TIERS = {"hard", "support", "telemetry"} From 8fcb2e1784b5d00922af47b3c1dc4333adeb010b Mon Sep 17 00:00:00 2001 From: Sarah Lacard Date: Fri, 1 May 2026 22:20:34 -0600 Subject: [PATCH 21/21] chore: trigger blue-pr-sync re-evaluation