Skip to content
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 40 additions & 0 deletions kernelguard.py
Original file line number Diff line number Diff line change
Expand Up @@ -1697,6 +1697,24 @@ def detect_introspection_exploit(code: str | SubmissionFacts) -> list[dict]:
if key not in seen_patterns:
seen_patterns.add(key)
saw_frame_access = True
elif (isinstance(node.func.value, ast.Name)
and node.func.value.id == "sys"
and node.func.attr == "_getframe"):
key = "sys._getframe"
if key not in seen_patterns:
seen_patterns.add(key)
saw_frame_access = True
elif (isinstance(node.func.value, ast.Name)
and node.func.value.id == "sys"
and node.func.attr == "_current_frames"):
key = "sys._current_frames"
if key not in seen_patterns:
seen_patterns.add(key)
matches.append({
"pattern": "SYS_GETFRAME",
"severity": "high",
"evidence": "sys._current_frames() — multi-threaded frame walk access",
})

# f_back, f_globals, f_locals attribute access
if isinstance(node, ast.Attribute) and node.attr in ("f_back", "f_globals", "f_locals"):
Expand Down Expand Up @@ -3025,6 +3043,22 @@ def detect_precision_downgrade(code: str | SubmissionFacts) -> list[dict]:
return matches


RE_HASH_CACHE = re.compile(r'\bhash\s*\(\s*\w+\s*\)\s*')
def detect_hash_cache_replay(code: str | SubmissionFacts) -> list[dict]:
facts = ensure_submission_facts(code)
active = facts.custom_kernel_active or facts.python_active
if not RE_HASH_CACHE.search(active): return []
if not re.search(r'_cache\s*\[', active) and not re.search(r'_cache\.get', active): return []
return [{"pattern": "OUTPUT_REPLAY_CACHE", "severity": "high", "evidence": "hash(data) used as cache key for output replay"}]


RE_LOGGING_FAKE = re.compile(r'logging\.(?:basicConfig|getLogger|setLevel|StreamHandler|FileHandler)\s*\(')
def detect_logging_output_injection(code: str | SubmissionFacts) -> list[dict]:
facts = ensure_submission_facts(code)
if not RE_LOGGING_FAKE.search(facts.raw_code): return []
return [{"pattern": "STDIO_REDIRECT", "severity": "medium", "evidence": "logging module — output injection/manipulation"}]


# ---------------------------------------------------------------------------
# Score anomaly detection
# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -3775,6 +3809,8 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool:
detect_token_paste_cuda_api,
detect_sequence_batch_graph,
detect_runtime_package_install,
detect_hash_cache_replay,
detect_logging_output_injection,
# AST-based detectors (Layer 2)
detect_trusted_module_import,
detect_module_mutation,
Expand All @@ -3790,6 +3826,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool:
detect_thread_injection,
detect_lazy_tensor,
detect_precision_downgrade,
detect_reduce_state,
]

BASE_DETECTOR_SPECS = [
Expand All @@ -3813,6 +3850,8 @@ 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),
("hash_cache_replay", detect_hash_cache_replay),
("logging_output_injection", detect_logging_output_injection),
("trusted_module_import", detect_trusted_module_import),
("module_mutation", detect_module_mutation),
("globals_mutation", detect_globals_mutation),
Expand All @@ -3827,6 +3866,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),
("exc_info_frame", detect_exc_info_frame),
]

VALID_RULE_OUTCOMES = {AUTO_FILTER, SUSPICIOUS_ONLY, TELEMETRY_ONLY}
Expand Down