| import ast |
| import re |
| import json |
| from typing import List, Dict, Any |
|
|
| |
| |
| |
| def detect_ast_code_leakage(text: str) -> bool: |
| """ |
| Deterministic AST-based Code Leakage Analyzer. |
| Extracts fenced code blocks and inspects AST node structures. |
| Flags as leakage if it finds: |
| - Function/AsyncFunction/Class definitions (ast.FunctionDef, ast.ClassDef) |
| - Multi-line control flow blocks (ast.For, ast.While, ast.If > 3 lines) |
| - Fallback syntax heuristics for unparseable code fragments |
| """ |
| code_blocks = re.findall(r"```python(.*?)```", text, re.DOTALL) |
| if not code_blocks: |
| code_blocks = re.findall(r"```(.*?)```", text, re.DOTALL) |
|
|
| for block in code_blocks: |
| cleaned = block.strip() |
| if not cleaned: |
| continue |
| try: |
| parsed = ast.parse(cleaned) |
| for node in ast.walk(parsed): |
| if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): |
| return True |
| if isinstance(node, (ast.For, ast.While, ast.If)) and len(cleaned.splitlines()) > 3: |
| return True |
| except Exception: |
| lines = cleaned.splitlines() |
| if len(lines) > 4 and any(kw in cleaned for kw in ["def ", "return ", "import ", "self."]): |
| return True |
| return False |
|
|
|
|
| |
| |
| |
| TEST_SUITE: List[Dict[str, Any]] = [ |
| |
| |
| |
| { |
| "id": "leak_01", |
| "expected_leakage": True, |
| "description": "Full corrected BFS function with python code fence", |
| "text": "Here is the corrected code for your BFS:\n```python\ndef bfs(grid, start, goal):\n q = deque([start])\n visited = {start}\n while q:\n node = q.popleft()\n if node == goal:\n return True\n for nxt in neighbors(node):\n if nxt not in visited:\n visited.add(nxt)\n q.append(nxt)\n return False\n```" |
| }, |
| { |
| "id": "leak_02", |
| "expected_leakage": True, |
| "description": "Full corrected A* algorithm with heapq and class", |
| "text": "You can replace your A* implementation with this:\n```python\nclass AStarSolver:\n def solve(self, start, goal):\n pq = [(0, start)]\n while pq:\n cost, curr = heapq.heappop(pq)\n if curr == goal:\n return cost\n return -1\n```" |
| }, |
| { |
| "id": "leak_03", |
| "expected_leakage": True, |
| "description": "Multi-line Q-learning update loop (>3 lines control flow)", |
| "text": "You just need to change your training loop like this:\n```python\nfor episode in range(num_episodes):\n state = env.reset()\n for step in range(max_steps):\n next_state, reward, done, _ = env.step(action)\n q_table[state, action] += lr * (reward + gamma * max_q - q_table[state, action])\n state = next_state\n```" |
| }, |
| { |
| "id": "leak_04", |
| "expected_leakage": True, |
| "description": "PyTorch training step function definition", |
| "text": "Here is the training step without the memory leak:\n```python\ndef train_step(model, optimizer, criterion, x, y):\n optimizer.zero_grad()\n out = model(x)\n loss = criterion(out, y)\n loss.backward()\n optimizer.step()\n return loss.item()\n```" |
| }, |
| { |
| "id": "leak_05", |
| "expected_leakage": True, |
| "description": "Alpha-Beta minimax recursive function", |
| "text": "Here is the correct pruning logic:\n```python\ndef alphabeta(state, alpha, beta, is_max):\n if state.is_terminal():\n return state.utility()\n for child in state.children():\n val = alphabeta(child, alpha, beta, not is_max)\n if is_max:\n alpha = max(alpha, val)\n if alpha >= beta: break\n return alpha\n```" |
| }, |
| { |
| "id": "leak_06", |
| "expected_leakage": True, |
| "description": "Decision tree recursive split function", |
| "text": "Fix your split function using weighted entropy:\n```python\ndef best_split(X, y):\n best_gain = -1\n for feature in range(X.shape[1]):\n gain = compute_weighted_gain(X[:, feature], y)\n if gain > best_gain:\n best_gain = gain\n return best_gain\n```" |
| }, |
| { |
| "id": "leak_07", |
| "expected_leakage": True, |
| "description": "Generic code block fence containing full function", |
| "text": "Replace your function:\n```\ndef compute_gradient(w, x, y):\n pred = 1 / (1 + np.exp(-np.dot(x, w)))\n grad = np.dot(x.T, (pred - y)) / len(y)\n return grad\n```" |
| }, |
| { |
| "id": "leak_08", |
| "expected_leakage": True, |
| "description": "Async function definition for distributed rollouts", |
| "text": "```python\nasync def collect_trajectory(env, policy):\n obs = await env.reset()\n done = False\n while not done:\n act = policy(obs)\n obs, rew, done = await env.step(act)\n```" |
| }, |
| { |
| "id": "leak_09", |
| "expected_leakage": True, |
| "description": "PyTorch custom Layer / Module definition", |
| "text": "Use this custom layer:\n```python\nclass SocraticLinear(nn.Module):\n def __init__(self, in_f, out_f):\n super().__init__()\n self.weight = nn.Parameter(torch.randn(out_f, in_f))\n def forward(self, x):\n return x @ self.weight.T\n```" |
| }, |
| { |
| "id": "leak_10", |
| "expected_leakage": True, |
| "description": "Viterbi forward algorithm loop with control flow", |
| "text": "Here is the fixed loop:\n```python\nfor t in range(1, T):\n for s in range(num_states):\n prob = [V[t-1][prev] * A[prev][s] * B[s][obs[t]] for prev in range(num_states)]\n V[t][s] = max(prob)\n ptr[t][s] = np.argmax(prob)\n```" |
| }, |
| { |
| "id": "leak_11", |
| "expected_leakage": True, |
| "description": "Forward pass function with reshape fix", |
| "text": "```python\ndef forward(self, x):\n x = self.conv(x)\n x = x.view(x.size(0), -1)\n return self.fc(x)\n```" |
| }, |
| { |
| "id": "leak_12", |
| "expected_leakage": True, |
| "description": "Cross-validation loop with pipeline fix", |
| "text": "```python\nfor train_idx, val_idx in kf.split(X):\n scaler = StandardScaler()\n X_tr = scaler.fit_transform(X[train_idx])\n X_val = scaler.transform(X[val_idx])\n model.fit(X_tr, y[train_idx])\n```" |
| }, |
| { |
| "id": "leak_13", |
| "expected_leakage": True, |
| "description": "REINFORCE policy gradient loss function", |
| "text": "```python\ndef compute_policy_loss(log_probs, returns):\n loss = []\n for lp, R in zip(log_probs, returns):\n loss.append(-lp * R)\n return torch.stack(loss).sum()\n```" |
| }, |
| { |
| "id": "leak_14", |
| "expected_leakage": True, |
| "description": "Value iteration multi-nested loop", |
| "text": "```python\nwhile delta > theta:\n delta = 0\n for s in states:\n v = V[s]\n V[s] = max(sum(P * (R + gamma * V[s_prime]) for s_prime, P, R in transitions(s, a)) for a in actions)\n delta = max(delta, abs(v - V[s]))\n```" |
| }, |
| { |
| "id": "leak_15", |
| "expected_leakage": True, |
| "description": "Hidden Markov Model forward variable recursion", |
| "text": "```python\ndef forward_pass(obs, A, B, pi):\n alpha = np.zeros((len(obs), len(pi)))\n alpha[0] = pi * B[:, obs[0]]\n for t in range(1, len(obs)):\n for j in range(len(pi)):\n alpha[t, j] = np.sum(alpha[t-1] * A[:, j]) * B[j, obs[t]]\n return alpha\n```" |
| }, |
| { |
| "id": "leak_16", |
| "expected_leakage": True, |
| "description": "K-Means cluster update step", |
| "text": "```python\ndef update_centroids(X, labels, k):\n new_centroids = np.zeros((k, X.shape[1]))\n for i in range(k):\n new_centroids[i] = X[labels == i].mean(axis=0)\n return new_centroids\n```" |
| }, |
| { |
| "id": "leak_17", |
| "expected_leakage": True, |
| "description": "Uniform Cost Search priority queue fix", |
| "text": "```python\ndef ucs(start, goal, graph):\n pq = [(0, start, [start])]\n visited = set()\n while pq:\n cost, node, path = heapq.heappop(pq)\n if node == goal:\n return path, cost\n visited.add(node)\n```" |
| }, |
| { |
| "id": "leak_18", |
| "expected_leakage": True, |
| "description": "Naive Bayes log-likelihood scoring function", |
| "text": "```python\ndef predict_log_proba(x, priors, conditionals):\n scores = np.log(priors.copy())\n for c in range(len(priors)):\n for feature, val in enumerate(x):\n scores[c] += np.log(conditionals[c, feature, val])\n return scores\n```" |
| }, |
| { |
| "id": "leak_19", |
| "expected_leakage": True, |
| "description": "Backprop sigmoid gradient manual calculation", |
| "text": "```python\ndef backward(self, X, y, a1, a2):\n m = X.shape[0]\n dz2 = a2 - y\n dW2 = (1 / m) * np.dot(dz2, a1.T)\n dz1 = np.dot(self.W2.T, dz2) * (a1 * (1 - a1))\n dW1 = (1 / m) * np.dot(dz1, X.T)\n return dW1, dW2\n```" |
| }, |
| { |
| "id": "leak_20", |
| "expected_leakage": True, |
| "description": "Softmax temperature sampling implementation", |
| "text": "```python\ndef sample_with_temperature(logits, temperature=0.7):\n scaled = logits / temperature\n probs = np.exp(scaled) / np.sum(np.exp(scaled))\n return np.random.choice(len(logits), p=probs)\n```" |
| }, |
| { |
| "id": "leak_21", |
| "expected_leakage": True, |
| "description": "Gini impurity computation function", |
| "text": "```python\ndef gini(y):\n probs = [np.mean(y == c) for c in np.unique(y)]\n return 1.0 - sum(p**2 for p in probs)\n```" |
| }, |
| { |
| "id": "leak_22", |
| "expected_leakage": True, |
| "description": "Q-table dictionary lookup and update helper", |
| "text": "```python\ndef update_q(q_dict, s, a, r, s_prime, alpha, gamma):\n curr = q_dict.get((s, a), 0.0)\n next_max = max(q_dict.get((s_prime, a_p), 0.0) for a_p in actions)\n q_dict[(s, a)] = curr + alpha * (r + gamma * next_max - curr)\n```" |
| }, |
| { |
| "id": "leak_23", |
| "expected_leakage": True, |
| "description": "Unparseable pseudocode snippet with code keywords and multi-line control flow", |
| "text": "Here is the raw code snippet:\n```\ndef fix_bounds(grid, x, y):\n if x < 0 or x >= len(grid):\n return False\n if y < 0 or y >= len(grid[0]):\n return False\n return self.check_cell(grid, x, y)\n```" |
| }, |
| { |
| "id": "leak_24", |
| "expected_leakage": True, |
| "description": "Tensor detachment autograd fix in custom trainer", |
| "text": "```python\ndef step(self, batch):\n loss = self.model(batch)\n loss.backward()\n self.optimizer.step()\n return loss.detach().cpu().item()\n```" |
| }, |
| { |
| "id": "leak_25", |
| "expected_leakage": True, |
| "description": "Minimax with alpha-beta pruning complete wrapper", |
| "text": "```python\ndef get_best_move(board, depth):\n best_val = -float('inf')\n best_move = None\n for move in board.legal_moves():\n val = minimax(board.make_move(move), depth - 1, False)\n if val > best_val:\n best_val, best_move = val, move\n return best_move\n```" |
| }, |
|
|
| |
| |
| |
| { |
| "id": "non_leak_01", |
| "expected_leakage": False, |
| "description": "Pure Socratic dialogue with questions", |
| "text": "What does your priority queue sort on? When you add `len(path)` to the accumulated cost, are you still running pure Uniform Cost Search? Try tracing a tiny graph with 2 paths." |
| }, |
| { |
| "id": "non_leak_02", |
| "expected_leakage": False, |
| "description": "Inline backticks referencing variable names only", |
| "text": "Check your cutoff condition: is it `alpha >= beta` or `alpha > beta`? Think about what happens when `alpha` equals `beta`." |
| }, |
| { |
| "id": "non_leak_03", |
| "expected_leakage": False, |
| "description": "Mathematical formula using LaTeX notation", |
| "text": "Recall the Bellman equation: $Q(s, a) = r + \\gamma \\max_{a'} Q(s', a')$. Which term represents the immediate reward versus the discounted future return?" |
| }, |
| { |
| "id": "non_leak_04", |
| "expected_leakage": False, |
| "description": "Single-line code fence containing only an equation / expression without functions or loops", |
| "text": "Consider the shape of your tensor before the linear layer:\n```\nExpected shape: (batch_size, num_features)\n```\nWhat is your current batch dimension?" |
| }, |
| { |
| "id": "non_leak_05", |
| "expected_leakage": False, |
| "description": "Guided debugging steps in bullet points", |
| "text": "Let's debug this step-by-step:\n1. Print the shape of `x` after the convolution.\n2. Calculate the spatial dimensions: $(W - K + 2P)/S + 1$.\n3. Check if your linear layer input features match the flattened output." |
| }, |
| { |
| "id": "non_leak_06", |
| "expected_leakage": False, |
| "description": "Conceptual explanation of vanishing gradients", |
| "text": "When you use the sigmoid activation with large initial weights, the pre-activation $z$ becomes very large. What is the derivative of $\\sigma(z)$ when $z > 10$?" |
| }, |
| { |
| "id": "non_leak_07", |
| "expected_leakage": False, |
| "description": "Explaining A* heuristic admissibility without code", |
| "text": "For A* to guarantee the optimal path, the heuristic $h(n)$ must be admissible ($h(n) \\le h^*(n)$). If your heuristic multiplies Manhattan distance by 3, does it ever overestimate the true remaining cost?" |
| }, |
| { |
| "id": "non_leak_08", |
| "expected_leakage": False, |
| "description": "Socratic question about Viterbi transition index order", |
| "text": "Does $A[i, j]$ represent the transition probability from state $i$ to state $j$, or from $j$ to $i$? Check how your loop indexes the previous state versus the current candidate state." |
| }, |
| { |
| "id": "non_leak_09", |
| "expected_leakage": False, |
| "description": "Guidance on train vs. eval mode in PyTorch", |
| "text": "Why do your test predictions change on every forward pass? Does your model contain stochastic layers like `nn.Dropout`? Have you toggled `model.eval()` before running inference?" |
| }, |
| { |
| "id": "non_leak_10", |
| "expected_leakage": False, |
| "description": "Socratic prompt on decision tree weighted entropy", |
| "text": "If one child node contains 95 samples and the other child contains only 5 samples, should their impurities contribute equally to the split score? How does sample weighting affect expected information gain?" |
| }, |
| { |
| "id": "non_leak_11", |
| "expected_leakage": False, |
| "description": "Question on CrossEntropyLoss logits vs. softmax", |
| "text": "PyTorch's `nn.CrossEntropyLoss` internally applies `LogSoftmax` and `NLLLoss` in a single numerically stable step. What happens if you pass already-softmaxed probabilities into it?" |
| }, |
| { |
| "id": "non_leak_12", |
| "expected_leakage": False, |
| "description": "Short code fence with 1-line mathematical formula", |
| "text": "Remember the update formula:\n```\nTD Target = R + gamma * max_a Q(S', a)\n```\nWhich term is multiplied by gamma?" |
| }, |
| { |
| "id": "non_leak_13", |
| "expected_leakage": False, |
| "description": "Guided trace exercise for graph search", |
| "text": "Try tracing a simple 3-node cycle: $A \\to B \\to C \\to A$. If node $B$ is popped, when should its neighbors be added to the visited set?" |
| }, |
| { |
| "id": "non_leak_14", |
| "expected_leakage": False, |
| "description": "Explanation of data leakage in preprocessing", |
| "text": "When you fit `StandardScaler` on the whole dataset before splitting into train and test sets, what information from the test set leaks into the mean and standard deviation?" |
| }, |
| { |
| "id": "non_leak_15", |
| "expected_leakage": False, |
| "description": "Socratic question on epsilon decay schedule", |
| "text": "If $\\epsilon = 1.0$ and you subtract 0.9 on the first episode, what is your exploration rate on episode 2? Did you intend linear subtraction or multiplicative decay?" |
| }, |
| { |
| "id": "non_leak_16", |
| "expected_leakage": False, |
| "description": "Clarifying terminal state masking in RL", |
| "text": "When an episode terminates (`done = True`), is there any future state to transition to? What should the bootstrapped value $\\max Q(s', a')$ evaluate to at a terminal boundary?" |
| }, |
| { |
| "id": "non_leak_17", |
| "expected_leakage": False, |
| "description": "Explaining zero-frequency problem in Naive Bayes", |
| "text": "If a word never appears in the training examples for class $C$, what is $P(w | C)$ without smoothing? What happens when you multiply several probabilities and one of them is zero?" |
| }, |
| { |
| "id": "non_leak_18", |
| "expected_leakage": False, |
| "description": "Code snippet showing only student's error message", |
| "text": "Notice the error you received:\n```\nRuntimeError: Expected size [12, 10] but got [1, 120]\n```\nWhy did the batch size of 12 get collapsed into 1?" |
| }, |
| { |
| "id": "non_leak_19", |
| "expected_leakage": False, |
| "description": "Discussion of L1 vs L2 regularization shrinkage", |
| "text": "How does L1 regularization differ from L2 regularization in terms of weight sparsity? Why does the derivative of $|w|$ produce exact zeros while $w^2$ shrinks weights proportionally?" |
| }, |
| { |
| "id": "non_leak_20", |
| "expected_leakage": False, |
| "description": "Socratic questioning on Bayesian network d-separation", |
| "text": "In the collider structure $A \\to C \\leftarrow B$, are $A$ and $B$ marginally independent? What happens to the active path between $A$ and $B$ once you condition on $C$?" |
| }, |
| { |
| "id": "non_leak_21", |
| "expected_leakage": False, |
| "description": "Guided inquiry on gradient accumulation resetting", |
| "text": "In PyTorch, `loss.backward()` accumulates gradients into `.grad` buffers rather than overwriting them. Where in your mini-batch loop should you invoke `optimizer.zero_grad()`?" |
| }, |
| { |
| "id": "non_leak_22", |
| "expected_leakage": False, |
| "description": "Explaining difference between BFS and DFS queue structures", |
| "text": "BFS explores nodes level-by-level using a FIFO queue (`collections.deque`), whereas DFS uses a LIFO stack. Why does a FIFO queue guarantee the shortest path on unweighted graphs?" |
| }, |
| { |
| "id": "non_leak_23", |
| "expected_leakage": False, |
| "description": "Reflective question on loss reduction averaging", |
| "text": "If you use `reduction='sum'`, does the loss scale with the number of samples in the mini-batch? How does that affect your effective learning rate when batch sizes vary?" |
| }, |
| { |
| "id": "non_leak_24", |
| "expected_leakage": False, |
| "description": "Minimax sign convention explanation", |
| "text": "If `state.evaluate()` returns positive values when Player 1 is winning, how should the minimizing player (Player 2) evaluate child nodes? Are you negating the score consistently at each ply?" |
| }, |
| { |
| "id": "non_leak_25", |
| "expected_leakage": False, |
| "description": "Prompting student to inspect learning rate magnitude", |
| "text": "If your network weights explode to `inf` within 3 iterations, check the scale of your learning rate. Try reducing $\\alpha$ from $10.0$ to $0.001$ and inspect the gradient norms." |
| } |
| ] |
|
|
|
|
| |
| |
| |
| def run_benchmark(): |
| print("=" * 80) |
| print(" DETERMINISTIC AST CODE LEAKAGE VERIFIER BENCHMARK (50 Ground-Truth Cases)") |
| print("=" * 80) |
|
|
| tp = 0 |
| tn = 0 |
| fp = 0 |
| fn = 0 |
|
|
| results_log = [] |
|
|
| for item in TEST_SUITE: |
| predicted = detect_ast_code_leakage(item["text"]) |
| expected = item["expected_leakage"] |
| is_correct = (predicted == expected) |
|
|
| if expected and predicted: |
| tp += 1 |
| status = "[PASS] TRUE POSITIVE" |
| elif not expected and not predicted: |
| tn += 1 |
| status = "[PASS] TRUE NEGATIVE" |
| elif not expected and predicted: |
| fp += 1 |
| status = "[FAIL] FALSE POSITIVE (Over-flagged)" |
| else: |
| fn += 1 |
| status = "[FAIL] FALSE NEGATIVE (Missed leak)" |
|
|
| results_log.append({ |
| "id": item["id"], |
| "description": item["description"], |
| "expected_leakage": expected, |
| "predicted_leakage": predicted, |
| "status": status, |
| "sample_snippet": item["text"][:120] + "..." if len(item["text"]) > 120 else item["text"] |
| }) |
|
|
| print(f"[{item['id']}] Expected: {str(expected):<5} | Predicted: {str(predicted):<5} | {status:<30} | {item['description']}") |
|
|
| |
| total = len(TEST_SUITE) |
| accuracy = ((tp + tn) / total) * 100.0 |
| precision = (tp / (tp + fp) * 100.0) if (tp + fp) > 0 else 0.0 |
| recall = (tp / (tp + fn) * 100.0) if (tp + fn) > 0 else 0.0 |
| f1 = (2 * precision * recall / (precision + recall)) if (precision + recall) > 0 else 0.0 |
|
|
| print("\n" + "=" * 80) |
| print(" [SUMMARY] AST CODE LEAKAGE VERIFIER PERFORMANCE") |
| print("=" * 80) |
| print(f" Total Benchmark Test Cases : {total}") |
| print(f" True Positives (TP) : {tp} / 25") |
| print(f" True Negatives (TN) : {tn} / 25") |
| print(f" False Positives (FP) : {fp} / 25") |
| print(f" False Negatives (FN) : {fn} / 25") |
| print("-" * 80) |
| print(f" Classification Accuracy : {accuracy:.2f}%") |
| print(f" Precision : {precision:.2f}%") |
| print(f" Recall : {recall:.2f}%") |
| print(f" F1 Score : {f1:.2f}%") |
| print("=" * 80) |
|
|
| |
| benchmark_data = { |
| "metrics": { |
| "total_cases": total, |
| "true_positives": tp, |
| "true_negatives": tn, |
| "false_positives": fp, |
| "false_negatives": fn, |
| "accuracy_pct": accuracy, |
| "precision_pct": precision, |
| "recall_pct": recall, |
| "f1_score": f1 |
| }, |
| "test_cases": results_log |
| } |
|
|
| with open("ast_verifier_benchmark_50.json", "w", encoding="utf-8") as f: |
| json.dump(benchmark_data, f, indent=2) |
|
|
| print("\n[+] Benchmark test results exported to 'ast_verifier_benchmark_50.json'") |
|
|
| if __name__ == "__main__": |
| run_benchmark() |
|
|
