Sync sp3arm4n → main: bring in PR #30 codex review fixes

sample/tests/test_analysis_cache.py

@@ -155,8 +155,8 @@ def test_has_reflects_state(self):
 
 
 class CacheVersioningTests(unittest.TestCase):
-    def test_schema_version_invalidates_pre_smali_tier_rows(self):
-        self.assertGreaterEqual(SCHEMA_VERSION, 5)
+    def test_schema_version_invalidates_pre_phase11_rows(self):
+        self.assertGreaterEqual(SCHEMA_VERSION, 6)
 
     def test_different_versions_coexist(self):
         with tempfile.TemporaryDirectory() as td:

sample/tests/test_code_audit.py

@@ -618,6 +618,18 @@ def test_same_class_same_rule_collapses(self):
         # occurrence_count = primary + extras
         self.assertEqual(out[0].occurrence_count, 3)
 
+    def test_dedup_does_not_mutate_inputs(self):
+        f1 = self._f(line=10, text="http://a.test")
+        f2 = self._f(line=20, text="http://b.test")
+        out1 = dedup_findings_by_class([f1, f2])
+        out2 = dedup_findings_by_class([f1, f2])
+
+        self.assertIsNot(out1[0], f1)
+        self.assertEqual(f1.occurrences, [])
+        self.assertEqual(f2.occurrences, [])
+        self.assertEqual(out1[0].occurrence_count, 2)
+        self.assertEqual(out2[0].occurrence_count, 2)
+
     def test_same_class_same_line_not_duplicated(self):
         """Two rules firing on the same line still create one occurrence,
         and a single rule firing twice on the same line (defensive case)

sample/tests/test_poc_generator.py

@@ -775,6 +775,45 @@ def test_same_shape_different_component_collapses(self):
         self.assertEqual(out[0].component, "com.demo.A")
         self.assertEqual(out[0].applies_to, ["com.demo.B"])
 
+    def test_dedup_does_not_mutate_inputs(self):
+        a = self._p(
+            title="외부 노출 액티비티 'com.demo.A' 호출",
+            component="com.demo.A",
+            commands=["adb shell am start -n com.demo/com.demo.A"],
+        )
+        b = self._p(
+            title="외부 노출 액티비티 'com.demo.B' 호출",
+            component="com.demo.B",
+            commands=["adb shell am start -n com.demo/com.demo.B"],
+        )
+
+        out1 = dedup_pocs_by_template([a, b])
+        out2 = dedup_pocs_by_template([a, b])
+
+        self.assertIsNot(out1[0], a)
+        self.assertEqual(a.applies_to, [])
+        self.assertEqual(b.applies_to, [])
+        self.assertEqual(out1[0].applies_to, ["com.demo.B"])
+        self.assertEqual(out2[0].applies_to, ["com.demo.B"])
+
+    def test_existing_applies_to_targets_are_merged(self):
+        a = self._p(
+            title="외부 노출 액티비티 'com.demo.A' 호출",
+            component="com.demo.A",
+            commands=["adb shell am start -n com.demo/com.demo.A"],
+        )
+        b = self._p(
+            title="외부 노출 액티비티 'com.demo.B' 호출",
+            component="com.demo.B",
+            commands=["adb shell am start -n com.demo/com.demo.B"],
+        )
+        b.applies_to = ["com.demo.C"]
+
+        out = dedup_pocs_by_template([a, b])
+
+        self.assertEqual(out[0].applies_to, ["com.demo.B", "com.demo.C"])
+        self.assertEqual(b.applies_to, ["com.demo.C"])
+
     def test_three_components_collapse_to_one(self):
         ps = [
             self._p(

sample/tests/test_smali_audit.py

@@ -321,6 +321,31 @@ def test_java_wins_on_overlap_with_smali_fold(self):
         # files_scanned is sum
         self.assertEqual(merged.files_scanned, 3)
 
+    def test_merge_does_not_mutate_inputs_or_duplicate_on_repeat(self):
+        java = CodeAuditReport(
+            package_name="com.x",
+            findings=[CodeFinding(
+                rule_id="CODE-001", title="java", severity="medium",
+                file="x.java", class_fqn="com.x.A", line_no=10,
+            )],
+        )
+        smali = CodeAuditReport(
+            package_name="com.x",
+            findings=[CodeFinding(
+                rule_id="CODE-001", title="smali", severity="medium",
+                file="x.smali", class_fqn="com.x.A",
+                evidence_tier="smali", line_no=25,
+            )],
+        )
+
+        merged1 = merge_code_reports(java, smali)
+        merged2 = merge_code_reports(java, smali)
+
+        self.assertEqual(java.findings[0].occurrences, [])
+        self.assertEqual(smali.findings[0].occurrences, [])
+        self.assertEqual(len(merged1.findings[0].occurrences), 1)
+        self.assertEqual(len(merged2.findings[0].occurrences), 1)
+
     def test_smali_occurrences_also_fold_into_java_rep(self):
         """If smali tier itself dedup-grouped multiple lines into one
         finding + occurrences, the WHOLE smali bundle (primary + occs)
@@ -355,6 +380,9 @@ def test_smali_occurrences_also_fold_into_java_rep(self):
         # on the java representative
         self.assertEqual(len(rep.occurrences), 3)
         self.assertEqual({o.evidence_tier for o in rep.occurrences}, {"smali"})
+        self.assertEqual([o.file for o in rep.occurrences], [
+            "x.smali", "x.smali", "x.smali",
+        ])
 
     def test_severity_difference_keeps_smali_separate(self):
         """High and medium severity findings on the same class stay as

src/venomhook/analysis_cache.py

@@ -59,7 +59,10 @@
 # Phase 10-4 added smali-tier findings. Old v4 rows can have a populated
 # Java code_audit_report yet still miss every smali finding, so replaying
 # them would silently under-report the new tier.
-SCHEMA_VERSION = 5
+# Phase 11 added occurrence folding and PoC applies_to/template dedup. Old
+# v5 rows would replay pre-normalized findings and PoCs without the new
+# cardinality guarantees, so they must be recomputed.
+SCHEMA_VERSION = 6
 
 
 @dataclass(frozen=True)

src/venomhook/code_audit.py

@@ -41,7 +41,7 @@
 from __future__ import annotations
 
 import re
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path
 from typing import Callable, Iterable, Optional
 
@@ -616,9 +616,68 @@ def _check_mode_world(
 ]
 
 
-def dedup_findings_by_class(
-    findings: Iterable[CodeFinding],
-) -> list[CodeFinding]:
+def _copy_occurrence_for_rep(
+    occ: CodeOccurrence,
+    *,
+    source_file: str,
+    rep_file: str,
+) -> CodeOccurrence:
+    """Copy an occurrence while normalizing its file relative to a rep."""
+    effective_file = occ.file or source_file
+    return CodeOccurrence(
+        line_no=occ.line_no,
+        line_text=occ.line_text,
+        file=effective_file if effective_file != rep_file else "",
+        evidence_tier=occ.evidence_tier,
+    )
+
+
+def _copy_finding_for_dedup(finding: CodeFinding) -> CodeFinding:
+    return replace(
+        finding,
+        references=list(finding.references),
+        occurrences=[
+            _copy_occurrence_for_rep(
+                occ,
+                source_file=finding.file,
+                rep_file=finding.file,
+            )
+            for occ in finding.occurrences
+        ],
+    )
+
+
+def _finding_as_occurrence(finding: CodeFinding, rep_file: str) -> CodeOccurrence:
+    return CodeOccurrence(
+        line_no=finding.line_no,
+        line_text=finding.line_text,
+        file=finding.file if finding.file != rep_file else "",
+        evidence_tier=finding.evidence_tier,
+    )
+
+
+def _same_occurrence_file(occ: CodeOccurrence, rep_file: str) -> str:
+    return occ.file or rep_file
+
+
+def _has_occurrence(rep: CodeFinding, occ: CodeOccurrence) -> bool:
+    occ_file = _same_occurrence_file(occ, rep.file)
+    if rep.line_no == occ.line_no and rep.file == occ_file:
+        return True
+    return any(
+        existing.line_no == occ.line_no
+        and _same_occurrence_file(existing, rep.file) == occ_file
+        and existing.evidence_tier == occ.evidence_tier
+        for existing in rep.occurrences
+    )
+
+
+def _append_occurrence_if_new(rep: CodeFinding, occ: CodeOccurrence) -> None:
+    if not _has_occurrence(rep, occ):
+        rep.occurrences.append(occ)
+
+
+def dedup_findings_by_class(findings: Iterable[CodeFinding]) -> list[CodeFinding]:
     """Phase 11-1: collapse findings sharing (rule_id, class_fqn, severity)
     into one representative + occurrences list.
 
@@ -639,10 +698,11 @@ def dedup_findings_by_class(
     enclosing class can't be inferred (default-package code, header
     comments, etc.).
 
-    Stable: representatives appear in the order their first occurrence
-    came from the rules. Each occurrence retains its (line_no,
-    line_text, file, evidence_tier) so HTML can render a "주요 +
-    N개 추가" expandable section.
+    Stable and non-mutating: representatives appear in the order their
+    first occurrence came from the rules, and the returned findings are
+    shallow copies with copied list fields. Each occurrence retains its
+    (line_no, line_text, file, evidence_tier) so HTML can render a
+    "주요 + N개 추가" expandable section.
     """
     out: list[CodeFinding] = []
     index: dict[tuple[str, str, str], int] = {}
@@ -652,18 +712,19 @@ def dedup_findings_by_class(
         slot = index.get(key)
         if slot is None:
             index[key] = len(out)
-            out.append(f)
+            out.append(_copy_finding_for_dedup(f))
             continue
         existing = out[slot]
-        # Don't duplicate the representative line itself.
-        if f.line_no == existing.line_no and f.file == existing.file:
-            continue
-        existing.occurrences.append(CodeOccurrence(
-            line_no=f.line_no,
-            line_text=f.line_text,
-            file=f.file if f.file != existing.file else "",
-            evidence_tier=f.evidence_tier,
-        ))
+        _append_occurrence_if_new(existing, _finding_as_occurrence(f, existing.file))
+        for occ in f.occurrences:
+            _append_occurrence_if_new(
+                existing,
+                _copy_occurrence_for_rep(
+                    occ,
+                    source_file=f.file,
+                    rep_file=existing.file,
+                ),
+            )
     return out
 
 

src/venomhook/poc_generator.py

@@ -36,6 +36,7 @@
 
 import json
 import shlex
+from dataclasses import replace
 from typing import Callable, Iterable
 
 from venomhook.models import (
@@ -1059,6 +1060,20 @@ def _template_signature(p: PoCArtifact) -> tuple:
     return (p.rule_id, p.kind, p.severity, normalized_title, normalized_cmds)
 
 
+def _copy_poc_artifact(artifact: PoCArtifact) -> PoCArtifact:
+    return replace(
+        artifact,
+        commands=list(artifact.commands),
+        references=list(artifact.references),
+        applies_to=list(artifact.applies_to),
+    )
+
+
+def _append_apply_target(rep: PoCArtifact, target: str | None) -> None:
+    if target and target != rep.component and target not in rep.applies_to:
+        rep.applies_to.append(target)
+
+
 def dedup_pocs_by_template(
     artifacts: Iterable[PoCArtifact],
 ) -> list[PoCArtifact]:
@@ -1070,17 +1085,14 @@ def dedup_pocs_by_template(
     blow-up). Operators got identical .sh scripts with only the
     component string changing — exhausting to read.
 
-    The first artifact for a signature stays as-is. Subsequent matches
-    contribute their ``component`` (or class_fqn-shaped first token of
-    the title) into the representative's ``applies_to`` list. The
-    representative's commands and metadata are untouched so any one
-    .sh stays runnable.
-
-    Stable: output order matches first-seen order of each signature.
-    Pure function; the artifacts passed in are not mutated when they
-    survive as representatives, but the merged ``applies_to`` list is
-    extended on the representative object — callers needing immutable
-    inputs should pass copies.
+    The first artifact for a signature stays as the representative.
+    Subsequent matches contribute their ``component`` (or class_fqn-shaped
+    first token of the title) into the representative's ``applies_to``
+    list. The representative's commands and metadata are copied from the
+    first artifact so any one .sh stays runnable.
+
+    Stable and non-mutating: output order matches first-seen order of
+    each signature, and the artifacts passed in are never modified.
     """
     out: list[PoCArtifact] = []
     seen: dict[tuple, int] = {}
@@ -1089,14 +1101,14 @@ def dedup_pocs_by_template(
         idx = seen.get(sig)
         if idx is None:
             seen[sig] = len(out)
-            out.append(p)
+            out.append(_copy_poc_artifact(p))
             continue
         rep = out[idx]
         # Record this artifact's component (or fall back to its title
         # fragment) on the representative.
-        target = p.component or p.title
-        if target and target != rep.component and target not in rep.applies_to:
-            rep.applies_to.append(target)
+        _append_apply_target(rep, p.component or p.title)
+        for target in p.applies_to:
+            _append_apply_target(rep, target)
     return out