From d013b71eb6a1ee7657fa597c1aa504c93e56107e Mon Sep 17 00:00:00 2001 From: Michael D'Angelo Date: Sat, 12 Sep 2026 03:20:20 +0000 Subject: [PATCH 1/5] Preserve distinct source remediations in published reports --- .../scripts/report_projection.py | 56 ++++++++++++++----- .../test_deep_scan_successful_publication.py | 32 +++++++++++ .../tests/test_report_projection.py | 20 +++++++ 3 files changed, 93 insertions(+), 15 deletions(-) diff --git a/plugins/codex-security/scripts/report_projection.py b/plugins/codex-security/scripts/report_projection.py index 98b4f00d1..b98b052e3 100644 --- a/plugins/codex-security/scripts/report_projection.py +++ b/plugins/codex-security/scripts/report_projection.py @@ -523,6 +523,41 @@ def _surface_notes(surface: dict[str, Any]) -> str: return _cell(f"{notes} Evidence: {evidence}") +def _remediation_section(finding: dict[str, Any]) -> list[str]: + remediation = _text(finding.get("remediation"), "No canonical remediation was recorded.") + lines = ["", "#### Remediation", "", remediation] + seen = {remediation} + sources = finding.get("provenance", {}).get("sourceFindings", []) + originals = ( + [ + source + for source in sources + if isinstance(source, dict) and isinstance(source.get("finding"), dict) + ] + if isinstance(sources, list) + else [] + ) + for source in originals: + text = _text(source["finding"].get("remediation"), "") + if text and text not in seen: + seen.add(text) + lines.extend(["", f"Source {_text(source.get('id'), 'finding')}: {text}"]) + for field, label in ( + ("remediationTests", "Tests"), + ("preventiveControls", "Preventive controls"), + ): + values = list( + dict.fromkeys( + value + for original in [finding, *(source["finding"] for source in originals)] + for value in _strings(original.get(field)) + ) + ) + if values: + lines.extend(["", f"{label}:", *_bullets(values, "None recorded.")]) + return lines + + def _finding_section(number: int, finding: dict[str, Any]) -> list[str]: validation = finding.get("validation") if isinstance(finding.get("validation"), dict) else {} _, raw_root_cause = merged_root_cause(finding) @@ -616,8 +651,6 @@ def _finding_section(number: int, finding: dict[str, Any]) -> list[str]: severity.get("changeConditions"), "Additional runtime or deployment evidence could raise or lower this severity.", ) - remediation_tests = _strings(finding.get("remediationTests")) - preventive_controls = _strings(finding.get("preventiveControls")) attack_steps = _strings(attack_path.get("steps")) cwes = ", ".join(finding["taxonomy"]["cwe"]) or "none" title = _text(finding["title"], "Untitled finding") @@ -736,18 +769,7 @@ def _finding_section(number: int, finding: dict[str, Any]) -> list[str]: lines.extend( ["", f"{label} assessment:", *(f"- **{name}:** {value}" for name, value in details)] ) - lines.extend( - [ - "", - "#### Remediation", - "", - _text(finding["remediation"], "No canonical remediation was recorded."), - ] - ) - if remediation_tests: - lines.extend(["", "Tests:", *_bullets(remediation_tests, "No tests recorded.")]) - if preventive_controls: - lines.extend(["", "Preventive controls:", *_bullets(preventive_controls, "None recorded.")]) + lines.extend(_remediation_section(finding)) return lines @@ -769,8 +791,12 @@ def _linked_finding_section(number: int, finding: dict[str, Any], report_path: s f"| CWE | {_cell(cwes)} |", f"| Affected lines | {_cell(_locations(finding))} |", ] - for heading in ("Summary", "Validation", "Dataflow", "Reachability", "Severity", "Remediation"): + for heading in ("Summary", "Validation", "Dataflow", "Reachability", "Severity"): lines.extend(["", f"#### {heading}", "", f"See the {link}."]) + if finding.get("provenance", {}).get("sourceFindings"): + lines.extend(_remediation_section(finding)) + else: + lines.extend(["", "#### Remediation", "", f"See the {link}."]) return lines diff --git a/plugins/codex-security/tests/test_deep_scan_successful_publication.py b/plugins/codex-security/tests/test_deep_scan_successful_publication.py index 8a144ece4..0c7a3ce75 100644 --- a/plugins/codex-security/tests/test_deep_scan_successful_publication.py +++ b/plugins/codex-security/tests/test_deep_scan_successful_publication.py @@ -173,6 +173,38 @@ def assert_published_aggregate(scan): assert (scan.scan_dir / "report.md").is_file() +def test_deep_publication_renders_each_source_remediation( + workbench_api, workbench_db, publication_scan +): + scan = publication_scan() + finding = scan.findings[0] + first = copy.deepcopy(finding) + first.pop("provenance") + first["remediation"] = "Check the destination before writing the archive entry." + first["remediationTests"] = ["Reject an archive entry outside the destination."] + second = copy.deepcopy(first) + second["remediation"] = "Reject symbolic links before opening the destination." + second["remediationTests"] = ["Reject a symbolic link inside the destination."] + second["preventiveControls"] = ["Use a directory-relative file handle."] + finding["remediation"] = first["remediation"] + finding["remediationTests"] = first["remediationTests"] + finding["provenance"]["sourceFindings"] = [ + {"id": "review-1:0", "finding": first}, + {"id": "review-2:0", "finding": second}, + ] + (scan.scan_dir / "findings.json").write_text(json.dumps({"findings": scan.findings})) + + complete(workbench_api, workbench_db, scan) + + assert_published_aggregate(scan) + report = (scan.scan_dir / "report.md").read_text() + for source in (first, second): + assert report.count(source["remediation"]) == 1 + for test in source["remediationTests"]: + assert report.count(test) == 1 + assert "Use a directory-relative file handle." in report + + @pytest.mark.parametrize("scope", [".", "subdir"], ids=["repository", "scoped"]) def test_deep_publication_keeps_configured_scope_without_worker_observations( workbench_api, workbench_db, publication_scan, scope diff --git a/plugins/codex-security/tests/test_report_projection.py b/plugins/codex-security/tests/test_report_projection.py index e88dcb63a..47495d7d5 100644 --- a/plugins/codex-security/tests/test_report_projection.py +++ b/plugins/codex-security/tests/test_report_projection.py @@ -75,6 +75,26 @@ def test_projection_normalizes_multiline_and_block_structural_text() -> None: assert "Text: ## Injected remediation - unsafe instruction" in markdown +def test_linked_writeup_retains_distinct_source_fixes() -> None: + manifest, findings, coverage = canonical_documents() + finding = findings["findings"][0] + finding["writeup"] = {"reportPath": "findings/parser/parser.md"} + finding["remediation"] = "Validate the record length." + finding["provenance"] = { + "sourceFindings": [ + {"id": "review-1:0", "finding": {"remediation": "Validate the record length."}}, + {"id": "review-2:0", "finding": {"remediation": "Reject duplicate record keys."}}, + {"id": "review-3:0", "finding": {"remediation": "Reject duplicate record keys."}}, + ] + } + + markdown = PROJECTION.build_report_markdown(manifest, findings, coverage) + + assert "findings/parser/parser.md" in markdown + assert markdown.count("Validate the record length.") == 1 + assert markdown.count("Reject duplicate record keys.") == 1 + + def test_projection_renders_inline_code_and_section_code_evidence() -> None: manifest, findings, coverage = canonical_documents() finding = findings["findings"][0] From 4b23ee3cbba79abc01d5171be2d01504c91fa674 Mon Sep 17 00:00:00 2001 From: Michael D'Angelo Date: Tue, 15 Sep 2026 01:57:18 -0700 Subject: [PATCH 2/5] test(reports): retain source tests across report formats --- .../tests/test_report_projection.py | 51 ++++++++++++++++--- 1 file changed, 44 insertions(+), 7 deletions(-) diff --git a/plugins/codex-security/tests/test_report_projection.py b/plugins/codex-security/tests/test_report_projection.py index 47495d7d5..94b769439 100644 --- a/plugins/codex-security/tests/test_report_projection.py +++ b/plugins/codex-security/tests/test_report_projection.py @@ -75,24 +75,61 @@ def test_projection_normalizes_multiline_and_block_structural_text() -> None: assert "Text: ## Injected remediation - unsafe instruction" in markdown -def test_linked_writeup_retains_distinct_source_fixes() -> None: +@pytest.mark.parametrize("linked_writeup", [False, True], ids=["inline", "linked"]) +def test_projection_retains_distinct_source_fixes(linked_writeup: bool) -> None: manifest, findings, coverage = canonical_documents() finding = findings["findings"][0] - finding["writeup"] = {"reportPath": "findings/parser/parser.md"} + if linked_writeup: + finding["writeup"] = {"reportPath": "findings/parser/parser.md"} finding["remediation"] = "Validate the record length." + finding["remediationTests"] = ["Reject a record longer than the allowed size."] + finding["preventiveControls"] = ["Centralize record validation."] finding["provenance"] = { "sourceFindings": [ {"id": "review-1:0", "finding": {"remediation": "Validate the record length."}}, - {"id": "review-2:0", "finding": {"remediation": "Reject duplicate record keys."}}, - {"id": "review-3:0", "finding": {"remediation": "Reject duplicate record keys."}}, + { + "id": "review-2:0", + "finding": { + "remediation": "Reject duplicate record keys.", + "remediationTests": [ + "Reject a record longer than the allowed size.", + "Cover duplicate keys in parser tests.", + ], + "preventiveControls": [ + "Centralize record validation.", + "Track keys while parsing a record.", + ], + }, + }, + { + "id": "review-3:0", + "finding": { + "remediation": "Reject duplicate record keys.", + "remediationTests": [ + "Cover duplicate keys in parser tests.", + "Reject case-variant duplicate keys.", + ], + "preventiveControls": ["Track keys while parsing a record."], + }, + }, ] } markdown = PROJECTION.build_report_markdown(manifest, findings, coverage) - assert "findings/parser/parser.md" in markdown - assert markdown.count("Validate the record length.") == 1 - assert markdown.count("Reject duplicate record keys.") == 1 + if linked_writeup: + assert "findings/parser/parser.md" in markdown + assert "Source review-2:0: Reject duplicate record keys." in markdown + for text in ( + "Validate the record length.", + "Reject duplicate record keys.", + "Reject a record longer than the allowed size.", + "Cover duplicate keys in parser tests.", + "Reject case-variant duplicate keys.", + "Centralize record validation.", + "Track keys while parsing a record.", + ): + assert markdown.count(text) == 1 def test_projection_renders_inline_code_and_section_code_evidence() -> None: From 89ea5a87b75d2791daf112828c73adb8c2cc31c0 Mon Sep 17 00:00:00 2001 From: Michael D'Angelo Date: Wed, 16 Sep 2026 00:55:54 +0000 Subject: [PATCH 3/5] test(reports): cover merged source fixes across publication modes --- .../test_deep_scan_successful_publication.py | 31 ++++++++++++++----- 1 file changed, 24 insertions(+), 7 deletions(-) diff --git a/plugins/codex-security/tests/test_deep_scan_successful_publication.py b/plugins/codex-security/tests/test_deep_scan_successful_publication.py index 7ff95cf72..b2ddcef09 100644 --- a/plugins/codex-security/tests/test_deep_scan_successful_publication.py +++ b/plugins/codex-security/tests/test_deep_scan_successful_publication.py @@ -174,10 +174,12 @@ def assert_published_aggregate(scan): assert (scan.scan_dir / "report.md").is_file() -def test_deep_publication_renders_each_source_remediation( - workbench_api, workbench_db, publication_scan +@pytest.mark.parametrize("mode", ["standard", "deep"]) +@pytest.mark.parametrize("linked_writeup", [False, True], ids=["inline", "linked"]) +def test_publication_renders_each_source_remediation( + workbench_api, workbench_db, publication_scan, mode, linked_writeup ): - scan = publication_scan() + scan = publication_scan(mode=mode) finding = scan.findings[0] first = copy.deepcopy(finding) first.pop("provenance") @@ -187,19 +189,34 @@ def test_deep_publication_renders_each_source_remediation( second["remediation"] = "Reject symbolic links before opening the destination." second["remediationTests"] = ["Reject a symbolic link inside the destination."] second["preventiveControls"] = ["Use a directory-relative file handle."] + third = copy.deepcopy(second) + third["remediationTests"].append("Reject a dangling symbolic link.") + fourth = copy.deepcopy(first) + fourth["remediation"] = "Create the output file exclusively." + fourth["remediationTests"] = ["Preserve an existing destination file."] + sources = [first, second, third, fourth] finding["remediation"] = first["remediation"] finding["remediationTests"] = first["remediationTests"] finding["provenance"]["sourceFindings"] = [ - {"id": "review-1:0", "finding": first}, - {"id": "review-2:0", "finding": second}, + {"id": f"review-{index}:0", "finding": source} for index, source in enumerate(sources, 1) ] + if linked_writeup: + finding["writeup"] = {"reportPath": "findings/archive/archive.md"} + writeup = scan.scan_dir / "findings" / "archive" / "archive.md" + writeup.parent.mkdir(parents=True) + writeup.write_text("# Archive extraction\n\nRepresentative source writeup.\n") + writeup_bytes = writeup.read_bytes() (scan.scan_dir / "findings.json").write_text(json.dumps({"findings": scan.findings})) - complete(workbench_api, workbench_db, scan) + completed = complete(workbench_api, workbench_db, scan) + assert completed["progress"]["status"] == "complete" assert_published_aggregate(scan) report = (scan.scan_dir / "report.md").read_text() - for source in (first, second): + if linked_writeup: + assert "findings/archive/archive.md" in report + assert writeup.read_bytes() == writeup_bytes + for source in sources: assert report.count(source["remediation"]) == 1 for test in source["remediationTests"]: assert report.count(test) == 1 From 321e64e09a03790172b52f1fd4b435f35c9f1c9d Mon Sep 17 00:00:00 2001 From: Michael D'Angelo Date: Wed, 16 Sep 2026 03:29:43 +0000 Subject: [PATCH 4/5] fix(reports): traverse retained remediation provenance --- .../scripts/report_projection.py | 48 +++++++++++------- .../test_deep_scan_successful_publication.py | 49 ++++++++++++++++--- 2 files changed, 72 insertions(+), 25 deletions(-) diff --git a/plugins/codex-security/scripts/report_projection.py b/plugins/codex-security/scripts/report_projection.py index b98b052e3..e5e8302a1 100644 --- a/plugins/codex-security/scripts/report_projection.py +++ b/plugins/codex-security/scripts/report_projection.py @@ -527,30 +527,42 @@ def _remediation_section(finding: dict[str, Any]) -> list[str]: remediation = _text(finding.get("remediation"), "No canonical remediation was recorded.") lines = ["", "#### Remediation", "", remediation] seen = {remediation} - sources = finding.get("provenance", {}).get("sourceFindings", []) - originals = ( - [ - source - for source in sources - if isinstance(source, dict) and isinstance(source.get("finding"), dict) - ] - if isinstance(sources, list) - else [] - ) - for source in originals: - text = _text(source["finding"].get("remediation"), "") + originals: list[tuple[str, dict[str, Any]]] = [] + pending = [("finding", finding)] + seen_findings: set[int] = set() + while pending: + source_id, original = pending.pop() + if id(original) in seen_findings: + continue + seen_findings.add(id(original)) + originals.append((source_id, original)) + provenance = original.get("provenance") + if not isinstance(provenance, dict): + continue + previous = provenance.get("previousFindings") + if isinstance(previous, list): + pending.extend( + (source_id, item) for item in reversed(previous) if isinstance(item, dict) + ) + sources = provenance.get("sourceFindings") + if isinstance(sources, list): + pending.extend( + (_text(source.get("id"), "finding"), source["finding"]) + for source in reversed(sources) + if isinstance(source, dict) and isinstance(source.get("finding"), dict) + ) + for source_id, original in originals[1:]: + text = _text(original.get("remediation"), "") if text and text not in seen: seen.add(text) - lines.extend(["", f"Source {_text(source.get('id'), 'finding')}: {text}"]) + lines.extend(["", f"Source {source_id}: {text}"]) for field, label in ( ("remediationTests", "Tests"), ("preventiveControls", "Preventive controls"), ): values = list( dict.fromkeys( - value - for original in [finding, *(source["finding"] for source in originals)] - for value in _strings(original.get(field)) + value for _, original in originals for value in _strings(original.get(field)) ) ) if values: @@ -793,7 +805,9 @@ def _linked_finding_section(number: int, finding: dict[str, Any], report_path: s ] for heading in ("Summary", "Validation", "Dataflow", "Reachability", "Severity"): lines.extend(["", f"#### {heading}", "", f"See the {link}."]) - if finding.get("provenance", {}).get("sourceFindings"): + if any( + finding.get("provenance", {}).get(field) for field in ("sourceFindings", "previousFindings") + ): lines.extend(_remediation_section(finding)) else: lines.extend(["", "#### Remediation", "", f"See the {link}."]) diff --git a/plugins/codex-security/tests/test_deep_scan_successful_publication.py b/plugins/codex-security/tests/test_deep_scan_successful_publication.py index b2ddcef09..b3f325ea6 100644 --- a/plugins/codex-security/tests/test_deep_scan_successful_publication.py +++ b/plugins/codex-security/tests/test_deep_scan_successful_publication.py @@ -174,15 +174,15 @@ def assert_published_aggregate(scan): assert (scan.scan_dir / "report.md").is_file() -@pytest.mark.parametrize("mode", ["standard", "deep"]) +@pytest.mark.parametrize("mode", ["standard", "deep", "deep-nested"]) @pytest.mark.parametrize("linked_writeup", [False, True], ids=["inline", "linked"]) def test_publication_renders_each_source_remediation( workbench_api, workbench_db, publication_scan, mode, linked_writeup ): - scan = publication_scan(mode=mode) + scan = publication_scan(mode="deep" if mode == "deep-nested" else mode) finding = scan.findings[0] first = copy.deepcopy(finding) - first.pop("provenance") + first["provenance"] = {"source": "local_plugin"} first["remediation"] = "Check the destination before writing the archive entry." first["remediationTests"] = ["Reject an archive entry outside the destination."] second = copy.deepcopy(first) @@ -191,15 +191,30 @@ def test_publication_renders_each_source_remediation( second["preventiveControls"] = ["Use a directory-relative file handle."] third = copy.deepcopy(second) third["remediationTests"].append("Reject a dangling symbolic link.") + third["preventiveControls"].append("Resolve links relative to the destination directory.") fourth = copy.deepcopy(first) fourth["remediation"] = "Create the output file exclusively." fourth["remediationTests"] = ["Preserve an existing destination file."] sources = [first, second, third, fourth] finding["remediation"] = first["remediation"] finding["remediationTests"] = first["remediationTests"] - finding["provenance"]["sourceFindings"] = [ - {"id": f"review-{index}:0", "finding": source} for index, source in enumerate(sources, 1) - ] + if mode == "standard": + # Standard completion itself consolidates these duplicate logical findings. + scan.findings[:] = sources + finding = first + elif mode == "deep-nested": + nested = copy.deepcopy(second) + nested["provenance"]["previousFindings"] = [third] + nested["provenance"]["sourceFindings"] = [{"id": "review-4:0", "finding": fourth}] + finding["provenance"]["sourceFindings"] = [ + {"id": "review-1:0", "finding": first}, + {"id": "review-2:0", "finding": nested}, + ] + else: + finding["provenance"]["sourceFindings"] = [ + {"id": f"review-{index}:0", "finding": source} + for index, source in enumerate(sources, 1) + ] if linked_writeup: finding["writeup"] = {"reportPath": "findings/archive/archive.md"} writeup = scan.scan_dir / "findings" / "archive" / "archive.md" @@ -211,7 +226,22 @@ def test_publication_renders_each_source_remediation( completed = complete(workbench_api, workbench_db, scan) assert completed["progress"]["status"] == "complete" - assert_published_aggregate(scan) + if mode == "standard": + published = json.loads((scan.scan_dir / "findings.json").read_text())["findings"] + assert len(published) == 1 + canonical = published[0] + assert "sourceFindings" not in canonical["provenance"] + retained = [canonical, *canonical["provenance"].pop("previousFindings")] + for source in retained: + for field in ("findingId", "occurrenceId", "fingerprints"): + assert source.pop(field) + assert retained == sources + coverage = json.loads((scan.scan_dir / "coverage.json").read_text()) + for field in ("documentType", "schemaVersion", "scanId"): + coverage.pop(field) + assert coverage == scan.coverage + else: + assert_published_aggregate(scan) report = (scan.scan_dir / "report.md").read_text() if linked_writeup: assert "findings/archive/archive.md" in report @@ -220,7 +250,10 @@ def test_publication_renders_each_source_remediation( assert report.count(source["remediation"]) == 1 for test in source["remediationTests"]: assert report.count(test) == 1 - assert "Use a directory-relative file handle." in report + for control in source.get("preventiveControls", []): + assert report.count(control) == 1 + positions = [report.index(text) for text in dict.fromkeys(s["remediation"] for s in sources)] + assert positions == sorted(positions) @pytest.mark.parametrize("scope", [".", "subdir"], ids=["repository", "scoped"]) From 4d729e8a9cc99d6962400c487b7c1bcb479ce087 Mon Sep 17 00:00:00 2001 From: Michael D'Angelo Date: Tue, 15 Sep 2026 22:13:10 -0700 Subject: [PATCH 5/5] test(reports): separate fixture mutation from assertions --- .../tests/test_deep_scan_successful_publication.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/plugins/codex-security/tests/test_deep_scan_successful_publication.py b/plugins/codex-security/tests/test_deep_scan_successful_publication.py index b3f325ea6..447482bd3 100644 --- a/plugins/codex-security/tests/test_deep_scan_successful_publication.py +++ b/plugins/codex-security/tests/test_deep_scan_successful_publication.py @@ -234,7 +234,8 @@ def test_publication_renders_each_source_remediation( retained = [canonical, *canonical["provenance"].pop("previousFindings")] for source in retained: for field in ("findingId", "occurrenceId", "fingerprints"): - assert source.pop(field) + value = source.pop(field) + assert value assert retained == sources coverage = json.loads((scan.scan_dir / "coverage.json").read_text()) for field in ("documentType", "schemaVersion", "scanId"):