From 3186e2e17f524c88edb867dc4c90535291bd7142 Mon Sep 17 00:00:00 2001 From: sp3arm4n Date: Wed, 13 May 2026 02:57:55 +0900 Subject: [PATCH] Fix phase 11 normalization side effects --- sample/tests/test_analysis_cache.py | 4 +- sample/tests/test_code_audit.py | 12 ++++ sample/tests/test_poc_generator.py | 39 ++++++++++++ sample/tests/test_smali_audit.py | 28 +++++++++ src/venomhook/analysis_cache.py | 5 +- src/venomhook/code_audit.py | 97 +++++++++++++++++++++++------ src/venomhook/poc_generator.py | 42 ++++++++----- src/venomhook/smali_audit.py | 39 ++++++------ 8 files changed, 212 insertions(+), 54 deletions(-) diff --git a/sample/tests/test_analysis_cache.py b/sample/tests/test_analysis_cache.py index 6b6367b..77bd4c7 100644 --- a/sample/tests/test_analysis_cache.py +++ b/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: diff --git a/sample/tests/test_code_audit.py b/sample/tests/test_code_audit.py index db225c1..dfb2b6a 100644 --- a/sample/tests/test_code_audit.py +++ b/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) diff --git a/sample/tests/test_poc_generator.py b/sample/tests/test_poc_generator.py index 944ba62..639d2e9 100644 --- a/sample/tests/test_poc_generator.py +++ b/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( diff --git a/sample/tests/test_smali_audit.py b/sample/tests/test_smali_audit.py index 0076de2..8bde2b0 100644 --- a/sample/tests/test_smali_audit.py +++ b/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 diff --git a/src/venomhook/analysis_cache.py b/src/venomhook/analysis_cache.py index 3b1d0fa..ef597d2 100644 --- a/src/venomhook/analysis_cache.py +++ b/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) diff --git a/src/venomhook/code_audit.py b/src/venomhook/code_audit.py index 40131a9..25790d3 100644 --- a/src/venomhook/code_audit.py +++ b/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 diff --git a/src/venomhook/poc_generator.py b/src/venomhook/poc_generator.py index 4c8c45f..e6b25ce 100644 --- a/src/venomhook/poc_generator.py +++ b/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 diff --git a/src/venomhook/smali_audit.py b/src/venomhook/smali_audit.py index cc25ede..f3c87b6 100644 --- a/src/venomhook/smali_audit.py +++ b/src/venomhook/smali_audit.py @@ -35,7 +35,8 @@ severities, but evidence_tier="smali" labels each finding so HTML / JSON consumers can show "이 발견은 smali 기반" tooltips. When both tiers fire on the same (rule_id, class) pair, the pipeline keeps the -Java-tier finding (richer line text); the smali duplicate is dropped. +Java-tier finding (richer line text) and folds the smali evidence into +that finding's occurrences. Pure-Python; no external dependencies. Skips third-party prefixes that the Java tier already filters (Kotlin stdlib, AndroidX, Google @@ -52,6 +53,10 @@ from venomhook.code_audit import ( DEFAULT_THIRD_PARTY_PREFIXES, + _append_occurrence_if_new, + _copy_finding_for_dedup, + _copy_occurrence_for_rep, + _finding_as_occurrence, _strip_line_comment, # quote-aware to match code_audit conventions dedup_findings_by_class, ) @@ -424,8 +429,6 @@ def merge_code_reports( are None). When both are present, the smali tier's files_scanned is added to the result for completeness, and partial flag is OR'd. """ - from venomhook.models import CodeOccurrence - if java_report is None and smali_report is None: return None if java_report is None: @@ -436,35 +439,35 @@ def merge_code_reports( # Index java findings by (rule, class_or_file, severity) so a smali # match on the same key can fold instead of duplicate. java_index: dict[tuple[str, str, str], CodeFinding] = {} - for jf in java_report.findings: + merged: list[CodeFinding] = [ + _copy_finding_for_dedup(jf) for jf in java_report.findings + ] + for jf in merged: key = (jf.rule_id, jf.class_fqn or jf.file, jf.severity) # Last write wins on collision (shouldn't happen post P11-1 dedup # but defensive). Keep the first instead so order matches input. java_index.setdefault(key, jf) - merged: list[CodeFinding] = list(java_report.findings) for sf in smali_report.findings: key = (sf.rule_id, sf.class_fqn or sf.file, sf.severity) rep = java_index.get(key) if rep is None: # No overlap — keep the smali finding as a separate entry. - merged.append(sf) + merged.append(_copy_finding_for_dedup(sf)) continue # Same (rule, class, severity) in both tiers — fold smali into # the java representative as additional evidence. - rep.occurrences.append(CodeOccurrence( - line_no=sf.line_no, - line_text=sf.line_text, - file=sf.file if sf.file != rep.file else "", - evidence_tier="smali", - )) + primary_occ = _finding_as_occurrence(sf, rep.file) + primary_occ.evidence_tier = "smali" + _append_occurrence_if_new(rep, primary_occ) for occ in sf.occurrences: - rep.occurrences.append(CodeOccurrence( - line_no=occ.line_no, - line_text=occ.line_text, - file=occ.file if occ.file != rep.file else "", - evidence_tier="smali", - )) + copied_occ = _copy_occurrence_for_rep( + occ, + source_file=sf.file, + rep_file=rep.file, + ) + copied_occ.evidence_tier = "smali" + _append_occurrence_if_new(rep, copied_occ) return CodeAuditReport( package_name=java_report.package_name or smali_report.package_name,