#!/usr/bin/env python3 """ scan_verify.py — Structural patch verification for all UNDF registry entries. For each UNDF entry in UNDF-REGISTRY.json, verifies that the associated patch: 1. Contains a UNDF header comment with the correct ID 2. Has removed lines (-) representing the defective pattern 3. Has added lines (+) representing the fixed pattern 4. The fix introduces a known O(1) data structure or visited-guard signature consistent with CWE-407 or related complexity defect remediation This gives structural test coverage for all 1,258+ indicators without requiring the target project to be compiled. Behavioral unit tests in defects/*/unit/ provide deeper proof for supported languages. Exit codes: 0 — all registered entries with patch files pass structural verification 1 — one or more patches fail verification (missing UNDF header, no fix sig) 2 — registry file not found or unreadable Usage: cd tests && python3 scan_verify.py # from tests/ directory cd tests && python3 scan_verify.py --verbose # per-entry detail cd tests && python3 scan_verify.py --report # save tests/SCAN-VERIFY-REPORT.md cd tests && python3 scan_verify.py --fail-on-warn # treat WARN as failure Run via Makefile (from tests/ directory): make scan-verify make scan-verify-report make scan-verify-strict """ import json import os import re import sys import argparse from pathlib import Path from collections import defaultdict # ── Constants ───────────────────────────────────────────────────────────────── REPO_ROOT = Path(__file__).parent.parent # java-topology/ REGISTRY = REPO_ROOT / "UNDF-REGISTRY.json" DEFECTS_DIR = REPO_ROOT / "defects" # Patterns that indicate the defective code (what was replaced) DEFECTIVE_SIGS = [ # Java r'\.contains\s*\(', # list.contains() r'\.indexOf\s*\(', # list.indexOf() r'\.add\s*\(.*\).*&&.*\.contains', # double add+contains r'ArrayList\b', # ArrayList used as set r'LinkedList\b.*contains', # LinkedList.contains # C++ r'std::find\s*\(', # std::find in vector r'std::vector\b', # vector used as set r'\.count\s*\(.*\)', # map/multimap count as membership r'std::list\b', # std::list used as set # Python r'\bif\b.*\bin\b.*\blist\b', # if x in list (explicit list var) r'not\s+in\s+self\.', # not in self.something (list member test) r'\.append\b.*#.*O\(N', # append with O(N) comment r'\.extend\s*\(.*\bif\b.*\bnot\s+in\b', # extend with not-in guard (O(N) dedup) r'\bfor\b.*\bif\b.*\bnot\s+in\b', # [x for x in ... if x not in list] # JavaScript / TypeScript r'\.find\s*\(\s*\(', # Array.find() — O(N) r'\.find\s*\(\s*[a-zA-Z]', # arr.find(pred) — O(N) r'\.includes\s*\(', # Array.includes() — O(N) r'\.some\s*\(', # Array.some() — O(N) r'\.indexOf\s*\(', # Array.indexOf() — O(N) # C# / .NET LINQ r'\.Any\s*\(', # LINQ Any() — O(N) r'\.FirstOrDefault\s*\(', # LINQ FirstOrDefault — O(N) r'\.Where\s*\(', # LINQ Where (in dedup context) # GLib (C) r'g_list_find\b', # GLib GList.find r'g_slist_find\b', # GLib GSList.find r'g_list_index\b', # GLib GList.index # Blender / project-specific C r'BLI_findptr\b', # Blender linked list find r'BLI_linklist_index\b', # Blender link list index # Linear scan comments r'[Ll]inear\s+[Ss]earch', # comment: "Linear search" r'[Ll]inear\s+scan', # comment: "linear scan" # Go r'\[\].*{.*}.*range', # slice used as set # Ruby r'\.include\?\s*\(', # Array.include? # PHP r'in_array\s*\(', # in_array() # General: recursion without visited guard r'def.*recursive|recursive.*def', # Python recursive without visited r'isCyclic\s*\(.*\)', # cycle check without visited r'discoverTypes\s*\(.*\)', # type discovery without visited ] # Patterns that indicate the fixed code (what was added) FIX_SIGS = [ # Java sets / maps r'HashSet\b', # HashSet r'LinkedHashSet\b', # LinkedHashSet (ordered) r'TreeSet\b', # TreeSet r'ConcurrentHashSet\b', # ConcurrentHashSet r'EconomicSet\b', # Graal EconomicSet r'Set<\b', # generic Set r'HashMap\b', # HashMap (map-based lookup fix) r'LinkedHashMap\b', # LinkedHashMap r'ConcurrentHashMap\b', # ConcurrentHashMap r'\.add\s*\(.*\).*return\s+false', # set.add() returning false r'!visited\.add\s*\(', # !visited.add() guard r'if\s*\(\s*visited\.add', # if (visited.add()) r'if\s*\(!visited', # if (!visited.contains()) # C++ r'std::unordered_set\b', # unordered_set r'std::unordered_map\b', # unordered_map r'std::set\b', # std::set r'std::map\b', # std::map r'absl::flat_hash_set\b', # Abseil flat_hash_set r'absl::flat_hash_map\b', # Abseil flat_hash_map r'\.insert\s*\(', # set/map insert() r'\.emplace\s*\(', # set/map emplace() # Python r'\bset\s*\(', # set() r'\bfrozenset\s*\(', # frozenset() r'visited\.add\b', # visited.add() r'if.*not in.*visited', # not in set r'\{[^}]*\bfor\b[^}]*\bin\b[^}]*\}', # set/dict comprehension {x for x in ...} r'\bdefaultdict\s*\(', # collections.defaultdict # Go r'map\[.*\]struct\{\}', # map[T]struct{} as set r'make\s*\(\s*map\[', # make(map[...]) — Go map creation r'seen\s*:?=.*make\s*\(', # seen := make(map...) # Rust r'HashMap\s*::\s*new\b', # HashMap::new() r'HashSet\s*::\s*new\b', # HashSet::new() r'BTreeMap\s*::\s*new\b', # BTreeMap::new() r'BTreeSet\s*::\s*new\b', # BTreeSet::new() r'IndexMap\b', # indexmap crate IndexMap r'FxHashMap\b', # rustc-hash FxHashMap r'FxHashSet\b', # rustc-hash FxHashSet # Haskell r'S\.fromList\b', # Data.Set.fromList r'M\.fromList\b', # Data.Map.fromList r'Data\.Set\b', # Data.Set import/use r'Data\.Map\b', # Data.Map import/use # Ruby r'\.to_a\.uniq\b', # .uniq r'Set\.new\b', # Set.new # PHP r'array_key_exists\s*\(', # array_key_exists for O(1) check r'\$.*\[.*\]\s*=\s*true', # hash map membership trick r'\bisset\s*\(', # isset() — O(1) hash key check # C / general r'\bHashtable\b', # C-style hashtable type r'\bhashmap\b', # generic hashmap identifier r'\bhash_map\b', # C hash_map r'\bdict\b.*:\s*', # Python dict literal/annotation r'O\s*\(\s*1\s*\)', # explicit O(1) complexity comment in fix r'O\s*\(\s*log', # explicit O(log N) comment (binary search fix) r'\bbsearch\s*\(', # C bsearch (sorted array binary search) r'\bqsort\s*\(', # C qsort (sort-then-bsearch pattern) # Qt (C++) r'\bQHash\b', # Qt QHash r'\bQSet\b', # Qt QSet r'\bQMap\b', # Qt QMap (ordered) # GLib (C) r'\bGHashTable\b', # GLib hash table type r'g_hash_table\b', # GLib hash table function prefix r'\bGTree\b', # GLib balanced tree # CPython C API r'\bPySet\b', # CPython set C API r'\bPyDict\b', # CPython dict C API # Dart / Kotlin / Scala / Swift r'\.toSet\s*\(\s*\)', # .toSet() — Dart/Kotlin/Scala/Swift r'mutableSetOf\s*\(', # Kotlin mutableSetOf r'setOf\s*\(', # Kotlin setOf r'NSMutableSet\b', # Swift/ObjC NSMutableSet r'NSSet\b', # Swift/ObjC NSSet # Security / credential redaction (MOAD-0004 fixes) r'\[REDACTED\]', # credential value replaced with [REDACTED] r'REDACT|redact', # redact function/variable name # General r'computeIfAbsent\b', # Java ConcurrentHashMap r'getOrDefault\b', # Java Map.getOrDefault r'putIfAbsent\b', # Java Map.putIfAbsent ] # Known UNDF header pattern at top of patch UNDF_HEADER_RE = re.compile(r'#\s*UNDF:\s*(UNDF-\d{4}-\d{9,})', re.IGNORECASE) # ── Status codes ────────────────────────────────────────────────────────────── PASS = "PASS" # patch present + well-formed + fix signature found WARN = "WARN" # patch present but missing UNDF header or weak fix sig FAIL = "FAIL" # patch has no defective/fixed lines or is malformed PENDING = "PENDING" # no defect directory (source not yet cloned) NO_PATCH = "NO_PATCH" # dir exists but no patch/ subdirectory found # ── Core verification ───────────────────────────────────────────────────────── def find_patch_files(proj_dir: str, slug: str = "") -> list: """ Return .patch files in defects/{proj_dir}/patch/ that belong to this slug. When a project directory contains patches for multiple UNDF entries (e.g. activemq-0001, activemq-0002), filter to only those whose filename contains the slug's entry suffix (e.g. "0001", "0002"). This prevents cross-contamination where one entry's patches pollute another's verdict. Returns [] (NO_PATCH) when slug is given but no suffix-matched file exists — indicating this entry's patch file has not yet been written. Does NOT fall back to all patches so that incomplete multi-entry directories report NO_PATCH rather than inheriting an unrelated entry's stub or completed patch. """ patch_dir = DEFECTS_DIR / proj_dir / "patch" if not patch_dir.is_dir(): return [] all_patches = [p for p in patch_dir.iterdir() if p.suffix == ".patch"] if not all_patches: return [] if slug: # Entry suffix = last "-"-delimited segment of the slug (e.g. "0001") entry_suffix = slug.rsplit("-", 1)[-1] # Filter: patch filename must contain the suffix as a word-boundary token # e.g. "activemq-0001-queue.patch" matches suffix "0001" # e.g. "linux-0001-0002-0003-hashstruct.patch" matches suffix "0001" matched = [p for p in all_patches if entry_suffix in p.stem] return sorted(matched) # empty = NO_PATCH for this entry return sorted(all_patches) def verify_patch(patch_path: Path, expected_undf_id: str) -> tuple: """ Verify one .patch file. Returns (status, details_list) where status is PASS/WARN/FAIL and details_list is a list of human-readable notes. """ try: content = patch_path.read_text(errors="replace") except OSError as e: return FAIL, [f"Cannot read patch: {e}"] details = [] warnings = [] # 1. UNDF header check header_match = UNDF_HEADER_RE.search(content) if header_match: found_id = header_match.group(1) if found_id != expected_undf_id: warnings.append(f"Header UNDF ID {found_id} != registry {expected_undf_id}") else: warnings.append(f"Missing '# UNDF: {expected_undf_id}' header comment") # 2. Extract diff lines removed_lines = [] added_lines = [] for line in content.splitlines(): if line.startswith("---") or line.startswith("+++"): continue if line.startswith("-"): removed_lines.append(line[1:]) elif line.startswith("+"): added_lines.append(line[1:]) if not added_lines: return FAIL, ["No added lines (+) found in patch — empty or malformed diff"] # Additive patches (guard insertion without removing old code) are valid. # Treat missing removed lines as WARN, not FAIL. additive = not removed_lines if additive: warnings.append("No removed lines (-) — additive/guard-insertion patch (no replace)") removed_text = "\n".join(removed_lines) added_text = "\n".join(added_lines) # 3. Defective pattern in removed lines (skip check for additive patches) defective_found = additive or any( re.search(sig, removed_text, re.IGNORECASE) for sig in DEFECTIVE_SIGS ) # 4. Fix pattern in added lines fix_found = any( re.search(sig, added_text, re.IGNORECASE) for sig in FIX_SIGS ) if not additive and not defective_found: warnings.append("No known defective pattern found in removed lines (may be correct for this project)") if not fix_found: warnings.append("No known fix signature found in added lines (may need DEFECTIVE_SIGS update)") # Determine overall status if warnings: status = WARN details = warnings else: removed_label = f"-{len(removed_lines)} lines, " if removed_lines else "" details = [f"+{len(added_lines)} lines, {removed_label}fix signature confirmed"] status = PASS return status, details def verify_entry(slug: str, undf_id: str) -> tuple: """ Verify one UNDF registry entry. Returns (status, proj_dir, patch_path_or_None, details_list). """ # Derive project directory from slug proj_dir = slug.rsplit("-", 1)[0] proj_path = DEFECTS_DIR / proj_dir if not proj_path.is_dir(): return PENDING, proj_dir, None, [f"Defect directory missing: defects/{proj_dir}/ (source not cloned)"] patches = find_patch_files(proj_dir, slug) if not patches: return NO_PATCH, proj_dir, None, [f"defects/{proj_dir}/patch/ not found or empty"] # Verify all patch files for this entry; worst status wins all_details = [] worst_status = PASS status_order = {PASS: 0, WARN: 1, NO_PATCH: 2, FAIL: 3, PENDING: 4} for patch_path in sorted(patches): st, details = verify_patch(patch_path, undf_id) if status_order.get(st, 0) > status_order.get(worst_status, 0): worst_status = st all_details.extend([f"[{patch_path.name}] {d}" for d in details]) return worst_status, proj_dir, patches[0], all_details # ── Main ────────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description=__doc__.strip().split("\n")[0]) parser.add_argument("--verbose", action="store_true", help="Print per-entry detail") parser.add_argument("--report", action="store_true", help="Write SCAN-VERIFY-REPORT.md") parser.add_argument("--fail-on-warn", action="store_true", help="Treat WARN as failure") parser.add_argument("--fail-on-pending", action="store_true", help="Treat PENDING-SOURCE as failure") args = parser.parse_args() if not REGISTRY.exists(): print(f"ERROR: Registry not found at {REGISTRY}", file=sys.stderr) sys.exit(2) with REGISTRY.open() as f: registry = json.load(f) counts = defaultdict(int) entries_by_status = defaultdict(list) print(f"Verifying {len(registry)} UNDF entries against {DEFECTS_DIR.name}/...") print() for slug, undf_id in sorted(registry.items()): status, proj_dir, patch_path, details = verify_entry(slug, undf_id) counts[status] += 1 entries_by_status[status].append((slug, undf_id, proj_dir, details)) if args.verbose or status in (FAIL, WARN): sym = {"PASS": "✓", "WARN": "⚠", "FAIL": "✗", "PENDING": "○", "NO_PATCH": "?"}.get(status, "?") print(f" {sym} {status:8s} {undf_id} {slug}") if status != PASS or args.verbose: for d in details: print(f" {d}") # Summary print() print("=" * 60) print(f"SCAN VERIFY SUMMARY — {len(registry)} total UNDF entries") print("=" * 60) print(f" {PASS:10s}: {counts[PASS]:4d} (patch present, fix signature confirmed)") print(f" {WARN:10s}: {counts[WARN]:4d} (patch present, header or sig needs review)") print(f" {FAIL:10s}: {counts[FAIL]:4d} (patch malformed or missing diff lines)") print(f" {NO_PATCH:10s}: {counts[NO_PATCH]:4d} (dir exists, no patch file found)") print(f" {PENDING:10s}: {counts[PENDING]:4d} (defect dir missing — source not cloned)") print() covered = counts[PASS] + counts[WARN] uncoverable = counts[PENDING] needs_work = counts[FAIL] + counts[NO_PATCH] total = len(registry) pct = 100 * covered / (total - uncoverable) if total > uncoverable else 100 print(f" Coverage: {covered}/{total - uncoverable} ({pct:.1f}%) of clonable entries") print(f" Pending source: {uncoverable} entries (source repos not yet cloned)") print() # Write report if args.report: report_path = REPO_ROOT / "tests" / "SCAN-VERIFY-REPORT.md" write_report(report_path, registry, entries_by_status, counts, total, covered, uncoverable, pct) print(f"Report written to {report_path}") # Exit code fail = counts[FAIL] > 0 or counts[NO_PATCH] > 0 if args.fail_on_warn: fail = fail or counts[WARN] > 0 if args.fail_on_pending: fail = fail or counts[PENDING] > 0 if fail: print("RESULT: FAIL — patches need attention (see above)") sys.exit(1) else: print("RESULT: PASS — all clonable entries verified") sys.exit(0) def write_report(path: Path, registry: dict, entries_by_status: dict, counts: dict, total: int, covered: int, uncoverable: int, pct: float): lines = [ "# SCAN-VERIFY-REPORT", "", f"**{total} UNDF entries** in registry. " f"**{covered}** verified ({pct:.1f}% of clonable). " f"**{uncoverable}** pending source clone.", "", f"| Status | Count | Meaning |", f"|--------|-------|---------|", f"| PASS | {counts[PASS]} | Patch present, fix signature confirmed |", f"| WARN | {counts[WARN]} | Patch present, header or signature needs review |", f"| FAIL | {counts[FAIL]} | Patch malformed or missing diff lines |", f"| NO_PATCH | {counts[NO_PATCH]} | Directory exists, no patch/ file found |", f"| PENDING | {counts[PENDING]} | Source not cloned, directory missing |", "", ] for status in (FAIL, NO_PATCH, WARN): if entries_by_status[status]: lines.append(f"## {status} entries") lines.append("") for slug, undf_id, proj_dir, details in entries_by_status[status]: lines.append(f"- **{undf_id}** `{slug}` (`defects/{proj_dir}/`)") for d in details: lines.append(f" - {d}") lines.append("") if entries_by_status[PENDING]: lines.append("## PENDING-SOURCE entries (source repos not cloned)") lines.append("") for slug, undf_id, proj_dir, details in entries_by_status[PENDING]: lines.append(f"- **{undf_id}** `{slug}`") lines.append("") path.write_text("\n".join(lines)) if __name__ == "__main__": main()