diff --git a/plugins/codex-security/scripts/report_projection.py b/plugins/codex-security/scripts/report_projection.py index 98b4f00d1..e5e8302a1 100644 --- a/plugins/codex-security/scripts/report_projection.py +++ b/plugins/codex-security/scripts/report_projection.py @@ -523,6 +523,53 @@ 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} + 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 {source_id}: {text}"]) + for field, label in ( + ("remediationTests", "Tests"), + ("preventiveControls", "Preventive controls"), + ): + values = list( + dict.fromkeys( + value for _, original 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 +663,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 +781,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 +803,14 @@ 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 any( + finding.get("provenance", {}).get(field) for field in ("sourceFindings", "previousFindings") + ): + 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 a83bb8359..447482bd3 100644 --- a/plugins/codex-security/tests/test_deep_scan_successful_publication.py +++ b/plugins/codex-security/tests/test_deep_scan_successful_publication.py @@ -174,6 +174,89 @@ def assert_published_aggregate(scan): assert (scan.scan_dir / "report.md").is_file() +@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="deep" if mode == "deep-nested" else mode) + finding = scan.findings[0] + first = copy.deepcopy(finding) + 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) + 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.") + 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"] + 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" + 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})) + + completed = complete(workbench_api, workbench_db, scan) + + assert completed["progress"]["status"] == "complete" + 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"): + 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"): + 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 + 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 + 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"]) 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..94b769439 100644 --- a/plugins/codex-security/tests/test_report_projection.py +++ b/plugins/codex-security/tests/test_report_projection.py @@ -75,6 +75,63 @@ def test_projection_normalizes_multiline_and_block_structural_text() -> None: assert "Text: ## Injected remediation - unsafe instruction" in markdown +@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] + 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.", + "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) + + 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: manifest, findings, coverage = canonical_documents() finding = findings["findings"][0]