From b92bd33ddc18732a8cacf35531440783e22b48a7 Mon Sep 17 00:00:00 2001
From: Infinitasium <infinitasium@localhost>
Date: Fri, 1 May 2026 23:31:53 +0530
Subject: [PATCH] Add generator replay detection (Pattern 16)

Detects generator-based replay attacks where:
- A generator function is defined with yield
- The generator is primed with next()
- generator.send is exposed as custom_kernel

This patches red submission KG Red ID 94 (WazupSteve).

KernelGuard-Red-Submission: 94
---
 kernelguard.py | 25 +++++++++++++++++++++++++
 1 file changed, 25 insertions(+)

diff --git a/kernelguard.py b/kernelguard.py
index f086847..bca5602 100644
--- a/kernelguard.py
+++ b/kernelguard.py
@@ -180,6 +180,9 @@
 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)
 
+# Pattern 16: Generator-based replay attack
+RE_GEN_REPLAY = re.compile(r'def\s+\w+\s*\([^)]*\)\s*:[^}]*yield[^}]*next\s*\([^)]*\)\s*[^=]*=\s*\w+\.send')
+
 
 # ---------------------------------------------------------------------------
 # Utility: strip C++/CUDA inline source from Python code
@@ -1397,6 +1400,26 @@ def detect_token_paste_cuda_api(code: str | SubmissionFacts) -> list[dict]:
     }]
 
 
+def detect_generator_replay(code: str | SubmissionFacts) -> list[dict]:
+    """Pattern 16: Generator-based replay attack.
+
+    Detects generators whose .send() method is exposed as custom_kernel.
+    The attack: create generator, prime with next(), assign g.send to custom_kernel.
+    """
+    facts = ensure_submission_facts(code)
+    raw = facts.raw_code
+    matches = []
+
+    if RE_GEN_REPLAY.search(raw):
+        matches.append({
+            "pattern": "GENERATOR_REPLAY",
+            "severity": "high",
+            "evidence": "Generator-based replay attack: generator.send() exposed as callable",
+        })
+
+    return matches
+
+
 # Sequence-aware batch graph exploit: learns repeated call sequences
 # and builds a batch CUDA graph that launches all calls at once.
 RE_CALL_SEQUENCE = re.compile(r'\b_?call_sequence\b|\b_?known_seq\b|\bbatch_graph\b|\bbatch_handle\b')
@@ -3773,6 +3796,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool:
     detect_unsync_multistream,
     detect_cuda_event_disable_timing,
     detect_token_paste_cuda_api,
+    detect_generator_replay,
     detect_sequence_batch_graph,
     detect_runtime_package_install,
     # AST-based detectors (Layer 2)
@@ -3811,6 +3835,7 @@ def support_only_patterns(matched_patterns: list[dict]) -> bool:
     ("unsync_multistream", detect_unsync_multistream),
     ("cuda_event_disable_timing", detect_cuda_event_disable_timing),
     ("token_paste_cuda_api", detect_token_paste_cuda_api),
+    ("generator_replay", detect_generator_replay),
     ("sequence_batch_graph", detect_sequence_batch_graph),
     ("runtime_package_install", detect_runtime_package_install),
     ("trusted_module_import", detect_trusted_module_import),