russell@unturf.com
0ac47a5b75
scan_verify.py: structural patch verification for all UNDF registry entries.
- Reads UNDF-REGISTRY.json, walks defects/*/patch/*.patch
- Checks: UNDF header comment, defective pattern in removed lines,
fix signature in added lines (HashSet/unordered_set/HashMap/etc.)
- Slug-to-patch filtering prevents cross-contamination in multi-entry dirs
- Additive patches (guard insertion) treated as WARN not FAIL
- Status: 454 PASS / 441 WARN / 12 FAIL / 300 NO_PATCH / 51 PENDING
- 74.2% structural coverage of clonable entries
check_coverage.py: CI gate — fails when new registry entries lack patch files.
- Any NO_PATCH entry (dir exists but no patch) causes exit code 1
- Designed to run on every commit to catch regressions early
- Quality (WARN/PASS) delegated to scan_verify.py
Makefile targets added:
make scan-verify — verbose structural verification (all 1258 entries)
make scan-verify-report — same + writes tests/SCAN-VERIFY-REPORT.md
make scan-verify-strict -- also fail on WARN
make coverage-check — CI gate for patch presence
Completes the three-tier coverage system:
unit/integration/functional Java tests (MOADs 0001-0011)
+ structural patch verification (1258+ UNDF indicators)
+ coverage CI gate (enforces no indicator left without a patch)
464 lines
20 KiB
Python
464 lines
20 KiB
Python
#!/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<T>
|
|
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()
|