diff --git a/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-000 b/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-000 index 4b7c17cca..4f8c6872d 100644 Binary files a/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-000 and b/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-000 differ diff --git a/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-001 b/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-001 index d280d00e6..14aedb294 100644 Binary files a/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-001 and b/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-001 differ diff --git a/sdk/typescript/_bundled_plugin/schemas/findings.schema.json b/sdk/typescript/_bundled_plugin/schemas/findings.schema.json index b282914da..2344b6715 100644 --- a/sdk/typescript/_bundled_plugin/schemas/findings.schema.json +++ b/sdk/typescript/_bundled_plugin/schemas/findings.schema.json @@ -263,6 +263,29 @@ } } }, + "code_evidence": { + "type": [ + "array", + "null" + ], + "items": { + "type": "object", + "required": [ + "id", + "code" + ], + "properties": { + "id": { + "type": "string", + "minLength": 1 + }, + "code": { + "type": "string", + "minLength": 1 + } + } + } + }, "rootCause": { "type": [ "object", @@ -293,6 +316,41 @@ } } }, + "root_cause": { + "type": [ + "object", + "string", + "null" + ], + "properties": { + "summary": { + "type": "string", + "minLength": 1 + }, + "evidenceRefs": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "evidence_refs": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "code": { + "type": "string", + "minLength": 1 + }, + "language": { + "type": "string", + "minLength": 1 + } + } + }, "remediation": { "type": "string", "minLength": 1 @@ -301,13 +359,280 @@ "type": [ "object", "null" - ] + ], + "properties": { + "assertions": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "counterEvidence": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "evidence": { + "type": [ + "string", + "array" + ], + "minLength": 1, + "items": { + "type": "string", + "minLength": 1 + } + }, + "evidenceRefs": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "evidence_refs": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "limitations": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "method": { + "type": "string", + "minLength": 1 + }, + "status": { + "type": [ + "string", + "null" + ], + "minLength": 1 + }, + "summary": { + "type": "string", + "minLength": 1 + }, + "disposition": { + "type": [ + "string", + "null" + ], + "minLength": 1 + }, + "result": { + "type": [ + "string", + "null" + ], + "minLength": 1 + } + } }, "attackPath": { "type": [ "object", "null" - ] + ], + "properties": { + "assumptions": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "blindspots": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "controls": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "dataFlow": { + "type": [ + "string", + "object" + ], + "minLength": 1, + "properties": { + "summary": { "type": "string", "minLength": 1 }, + "source": { "type": "string", "minLength": 1 }, + "sink": { "type": "string", "minLength": 1 }, + "outcome": { "type": "string", "minLength": 1 }, + "transformations": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + }, + "evidenceRefs": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + }, + "evidence_refs": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + } + } + }, + "data_flow": { + "type": [ + "string", + "object" + ], + "minLength": 1, + "properties": { + "summary": { "type": "string", "minLength": 1 }, + "source": { "type": "string", "minLength": 1 }, + "sink": { "type": "string", "minLength": 1 }, + "outcome": { "type": "string", "minLength": 1 }, + "transformations": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + }, + "evidenceRefs": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + }, + "evidence_refs": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + } + } + }, + "dataflow": { + "type": [ + "string", + "object" + ], + "minLength": 1, + "properties": { + "summary": { "type": "string", "minLength": 1 }, + "source": { "type": "string", "minLength": 1 }, + "sink": { "type": "string", "minLength": 1 }, + "outcome": { "type": "string", "minLength": 1 }, + "transformations": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + }, + "evidenceRefs": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + }, + "evidence_refs": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + } + } + }, + "evidenceRefs": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "evidence_refs": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "impact": { + "type": [ + "string", + "object", + "null" + ], + "minLength": 1, + "properties": { + "level": { "type": "string", "minLength": 1 }, + "rationale": { "type": "string", "minLength": 1 }, + "why": { "type": "string", "minLength": 1 } + } + }, + "likelihood": { + "type": [ + "string", + "object", + "null" + ], + "minLength": 1, + "properties": { + "level": { "type": "string", "minLength": 1 }, + "rationale": { "type": "string", "minLength": 1 }, + "why": { "type": "string", "minLength": 1 } + } + }, + "limitations": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "preconditions": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "reachability": { + "type": [ + "string", + "object" + ], + "minLength": 1, + "properties": { + "summary": { "type": "string", "minLength": 1 }, + "attacker": { "type": "string", "minLength": 1 }, + "entrypoint": { "type": "string", "minLength": 1 }, + "source": { "type": "string", "minLength": 1 }, + "sink": { "type": "string", "minLength": 1 }, + "outcome": { "type": "string", "minLength": 1 }, + "preconditions": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + }, + "evidenceRefs": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + }, + "evidence_refs": { + "type": "array", + "items": { "type": "string", "minLength": 1 } + } + } + }, + "steps": { + "type": "array", + "items": { + "type": "string", + "minLength": 1 + } + }, + "summary": { + "type": "string", + "minLength": 1 + } + } }, "remediationTests": { "type": "array", diff --git a/sdk/typescript/_bundled_plugin/schemas/tools/scan-draft.schema.json b/sdk/typescript/_bundled_plugin/schemas/tools/scan-draft.schema.json index 999e46f49..d32892223 100644 --- a/sdk/typescript/_bundled_plugin/schemas/tools/scan-draft.schema.json +++ b/sdk/typescript/_bundled_plugin/schemas/tools/scan-draft.schema.json @@ -181,6 +181,250 @@ ], "additionalProperties": true }, + "legacyCodeEvidence": { + "type": "object", + "properties": { + "id": { + "$ref": "#/$defs/text" + }, + "code": { + "$ref": "#/$defs/text" + } + }, + "required": [ + "id", + "code" + ], + "additionalProperties": true + }, + "findingAssessment": { + "anyOf": [ + { + "$ref": "#/$defs/text" + }, + { + "type": "object", + "properties": { + "level": { + "$ref": "#/$defs/text" + }, + "rationale": { + "$ref": "#/$defs/text" + }, + "why": { + "$ref": "#/$defs/text" + } + }, + "additionalProperties": true + } + ] + }, + "textOrSummary": { + "anyOf": [ + { + "$ref": "#/$defs/text" + }, + { + "type": "object", + "properties": { + "summary": { + "$ref": "#/$defs/text" + }, + "source": { + "$ref": "#/$defs/text" + }, + "sink": { + "$ref": "#/$defs/text" + }, + "outcome": { + "$ref": "#/$defs/text" + }, + "transformations": { + "$ref": "#/$defs/textList" + }, + "evidenceRefs": { + "$ref": "#/$defs/textList" + }, + "evidence_refs": { + "$ref": "#/$defs/textList" + } + }, + "additionalProperties": true + } + ] + }, + "findingReachability": { + "anyOf": [ + { + "$ref": "#/$defs/text" + }, + { + "type": "object", + "properties": { + "summary": { + "$ref": "#/$defs/text" + }, + "attacker": { + "$ref": "#/$defs/text" + }, + "entrypoint": { + "$ref": "#/$defs/text" + }, + "source": { + "$ref": "#/$defs/text" + }, + "sink": { + "$ref": "#/$defs/text" + }, + "outcome": { + "$ref": "#/$defs/text" + }, + "preconditions": { + "$ref": "#/$defs/textList" + }, + "evidenceRefs": { + "$ref": "#/$defs/textList" + }, + "evidence_refs": { + "$ref": "#/$defs/textList" + } + }, + "additionalProperties": true + } + ] + }, + "findingValidation": { + "type": "object", + "properties": { + "assertions": { + "$ref": "#/$defs/textList" + }, + "counterEvidence": { + "$ref": "#/$defs/textList" + }, + "evidence": { + "anyOf": [ + { + "$ref": "#/$defs/text" + }, + { + "$ref": "#/$defs/textList" + } + ] + }, + "evidenceRefs": { + "$ref": "#/$defs/textList" + }, + "evidence_refs": { + "$ref": "#/$defs/textList" + }, + "limitations": { + "$ref": "#/$defs/textList" + }, + "method": { + "$ref": "#/$defs/text" + }, + "status": { + "anyOf": [ + { + "$ref": "#/$defs/text" + }, + { + "type": "null" + } + ] + }, + "summary": { + "$ref": "#/$defs/text" + }, + "disposition": { + "anyOf": [ + { + "$ref": "#/$defs/text" + }, + { + "type": "null" + } + ] + }, + "result": { + "anyOf": [ + { + "$ref": "#/$defs/text" + }, + { + "type": "null" + } + ] + } + }, + "additionalProperties": true + }, + "findingAttackPath": { + "type": "object", + "properties": { + "assumptions": { + "$ref": "#/$defs/textList" + }, + "blindspots": { + "$ref": "#/$defs/textList" + }, + "controls": { + "$ref": "#/$defs/textList" + }, + "dataFlow": { + "$ref": "#/$defs/textOrSummary" + }, + "data_flow": { + "$ref": "#/$defs/textOrSummary" + }, + "dataflow": { + "$ref": "#/$defs/textOrSummary" + }, + "evidenceRefs": { + "$ref": "#/$defs/textList" + }, + "evidence_refs": { + "$ref": "#/$defs/textList" + }, + "impact": { + "anyOf": [ + { + "$ref": "#/$defs/findingAssessment" + }, + { + "type": "null" + } + ] + }, + "likelihood": { + "anyOf": [ + { + "$ref": "#/$defs/findingAssessment" + }, + { + "type": "null" + } + ] + }, + "limitations": { + "$ref": "#/$defs/textList" + }, + "preconditions": { + "$ref": "#/$defs/textList" + }, + "reachability": { + "$ref": "#/$defs/findingReachability" + }, + "steps": { + "$ref": "#/$defs/textList" + }, + "summary": { + "$ref": "#/$defs/text" + } + }, + "additionalProperties": true + }, "finding": { "type": "object", "properties": { @@ -236,6 +480,12 @@ "$ref": "#/$defs/codeEvidence" } }, + "code_evidence": { + "type": "array", + "items": { + "$ref": "#/$defs/legacyCodeEvidence" + } + }, "rootCause": { "anyOf": [ { @@ -247,6 +497,12 @@ "summary": { "$ref": "#/$defs/text" }, + "code": { + "$ref": "#/$defs/text" + }, + "language": { + "$ref": "#/$defs/text" + }, "evidenceRefs": { "$ref": "#/$defs/textList" } @@ -258,14 +514,41 @@ } ] }, + "root_cause": { + "anyOf": [ + { + "$ref": "#/$defs/text" + }, + { + "type": "object", + "properties": { + "summary": { + "$ref": "#/$defs/text" + }, + "code": { + "$ref": "#/$defs/text" + }, + "language": { + "$ref": "#/$defs/text" + }, + "evidenceRefs": { + "$ref": "#/$defs/textList" + }, + "evidence_refs": { + "$ref": "#/$defs/textList" + } + }, + "additionalProperties": true + } + ] + }, "remediation": { "$ref": "#/$defs/text" }, "validation": { "anyOf": [ { - "type": "object", - "additionalProperties": true + "$ref": "#/$defs/findingValidation" }, { "type": "null" @@ -275,8 +558,7 @@ "attackPath": { "anyOf": [ { - "type": "object", - "additionalProperties": true + "$ref": "#/$defs/findingAttackPath" }, { "type": "null" diff --git a/sdk/typescript/_bundled_plugin/scripts/finalize_scan_contract.py b/sdk/typescript/_bundled_plugin/scripts/finalize_scan_contract.py index 5a7806c7e..e84b0d7b9 100644 --- a/sdk/typescript/_bundled_plugin/scripts/finalize_scan_contract.py +++ b/sdk/typescript/_bundled_plugin/scripts/finalize_scan_contract.py @@ -694,7 +694,7 @@ def _finding_strength(finding: dict[str, Any]) -> tuple[int, int, int]: return ( ("informational", "low", "medium", "high", "critical").index(finding["severity"]["level"]), ("low", "medium", "high").index(finding["confidence"]["level"]), - len(finding.get("codeEvidence") or []), + _finding_evidence_strength(finding), ) @@ -735,6 +735,12 @@ def _recover_unsealed_findings( try: if not isinstance(finding, dict): raise ContractError(f"{context}: expected an object") + compatible_findings = _legacy_sealed_findings_for_validation( + {"findings": [finding]} + )["findings"] + compatible_finding = compatible_findings[0] + normalized_legacy_details = compatible_finding != finding + finding = compatible_finding identity = _require_dict(finding, "identity", context) fields: list[tuple[dict[str, Any], str, str, str]] = [ (finding, "ruleId", context, "rule identifier"), @@ -742,7 +748,9 @@ def _recover_unsealed_findings( ] if "instance" in identity: fields.append((identity, "instance", f"{context}.identity", "instance")) - normalized_fields = [] + normalized_fields = ( + ["legacy finding details"] if normalized_legacy_details else [] + ) for parent, field, field_context, label in fields: value = _require_str(parent, field, field_context) if SLUG_RE.fullmatch(value): @@ -1222,12 +1230,14 @@ def _validate_finding(finding: dict[str, Any], context: str) -> None: _validate_location(location, f"{context}.locations[{index}]") evidence_ids: set[str] = set() - code_evidence = finding.get("codeEvidence") - if code_evidence is not None: + for evidence_key in ("codeEvidence", "code_evidence"): + if evidence_key not in finding: + continue + code_evidence = finding[evidence_key] if not isinstance(code_evidence, list): - raise ContractError(f"{context}.codeEvidence: expected an array") + raise ContractError(f"{context}.{evidence_key}: expected an array") for index, evidence in enumerate(code_evidence): - evidence_context = f"{context}.codeEvidence[{index}]" + evidence_context = f"{context}.{evidence_key}[{index}]" if not isinstance(evidence, dict): raise ContractError(f"{evidence_context}: expected an object") evidence_id = _require_str(evidence, "id", evidence_context) @@ -1236,19 +1246,33 @@ def _validate_finding(finding: dict[str, Any], context: str) -> None: evidence_ids.add(evidence_id) _require_str(evidence, "code", evidence_context) - for section_name in ("rootCause", "validation", "attackPath"): - section = finding.get(section_name) - if not isinstance(section, dict) or "evidenceRefs" not in section: + referenced_sections = [ + (section_name, finding.get(section_name)) + for section_name in ("rootCause", "root_cause", "validation", "attackPath") + ] + attack_path = finding.get("attackPath") + if isinstance(attack_path, dict): + referenced_sections.extend( + (f"attackPath.{section_name}", attack_path.get(section_name)) + for section_name in ("dataFlow", "dataflow", "data_flow", "reachability") + ) + for section_name, section in referenced_sections: + if not isinstance(section, dict): continue - refs = section["evidenceRefs"] - if not isinstance(refs, list) or any(not isinstance(ref, str) or not ref for ref in refs): - raise ContractError(f"{context}.{section_name}.evidenceRefs: expected strings") - unknown_refs = sorted(set(refs) - evidence_ids) - if unknown_refs: - raise ContractError( - f"{context}.{section_name}.evidenceRefs: unknown code-evidence ids: " - + ", ".join(unknown_refs) - ) + for refs_key in ("evidenceRefs", "evidence_refs"): + if refs_key not in section: + continue + refs = section[refs_key] + if not isinstance(refs, list) or any( + not isinstance(ref, str) or not ref for ref in refs + ): + raise ContractError(f"{context}.{section_name}.{refs_key}: expected strings") + unknown_refs = sorted(set(refs) - evidence_ids) + if unknown_refs: + raise ContractError( + f"{context}.{section_name}.{refs_key}: unknown code-evidence ids: " + + ", ".join(unknown_refs) + ) provenance = _require_dict(finding, "provenance", context) _require_str(provenance, "source", f"{context}.provenance") @@ -1466,6 +1490,190 @@ def validate_against_schema(payload: dict[str, Any], schema_path: Path) -> None: _validate_schema_node(payload, schema, schema_path.stem) +def _filter_unknown_legacy_evidence_refs( + section: dict[str, Any], evidence_ids: set[str] +) -> None: + for refs_field in ("evidenceRefs", "evidence_refs"): + refs = section.get(refs_field) + if isinstance(refs, list): + section[refs_field] = [ + ref + for ref in refs + if isinstance(ref, str) and ref.strip() and ref in evidence_ids + ] + + +def _normalize_legacy_string_list_fields( + section: dict[str, Any], fields: tuple[str, ...] +) -> None: + for field in fields: + if field not in section: + continue + value = section[field] + if isinstance(value, str): + normalized = [value] if value.strip() else [] + elif isinstance(value, list): + normalized = [ + item for item in value if isinstance(item, str) and item.strip() + ] + else: + normalized = [] + if normalized: + section[field] = normalized + else: + section.pop(field) + + +def _remove_unsupported_legacy_scalar_fields( + section: dict[str, Any], fields: tuple[str, ...] +) -> None: + for field in fields: + if field in section and ( + not isinstance(section[field], str) or section[field] == "" + ): + section.pop(field) + + +def _legacy_sealed_findings_for_validation(findings: dict[str, Any]) -> dict[str, Any]: + compatible = copy.deepcopy(findings) + finding_items = compatible.get("findings") + if not isinstance(finding_items, list): + return compatible + for finding in finding_items: + if not isinstance(finding, dict): + continue + canonical_evidence = finding.get("codeEvidence") + canonical_evidence = canonical_evidence if isinstance(canonical_evidence, list) else [] + canonical_evidence_ids = { + evidence["id"] + for evidence in canonical_evidence + if isinstance(evidence, dict) + and isinstance(evidence.get("id"), str) + and evidence["id"] + } + legacy_evidence = finding.get("code_evidence") + if isinstance(legacy_evidence, list): + compatible_legacy_evidence = [] + seen_evidence_ids = set(canonical_evidence_ids) + for evidence in legacy_evidence: + if not isinstance(evidence, dict): + continue + evidence_id = evidence.get("id") + evidence_code = evidence.get("code") + if ( + not isinstance(evidence_id, str) + or not evidence_id.strip() + or not isinstance(evidence_code, str) + or not evidence_code.strip() + ): + continue + if evidence_id in seen_evidence_ids: + continue + seen_evidence_ids.add(evidence_id) + compatible_legacy_evidence.append(evidence) + finding["code_evidence"] = compatible_legacy_evidence + elif "code_evidence" in finding: + finding.pop("code_evidence") + compatible_legacy_evidence = finding.get("code_evidence") + compatible_legacy_evidence = ( + compatible_legacy_evidence if isinstance(compatible_legacy_evidence, list) else [] + ) + evidence_ids = canonical_evidence_ids | { + evidence["id"] + for evidence in compatible_legacy_evidence + if isinstance(evidence, dict) + and isinstance(evidence.get("id"), str) + and evidence["id"] + } + for section_name, list_fields in ( + ("root_cause", ("evidenceRefs", "evidence_refs")), + ( + "validation", + ( + "assertions", + "counterEvidence", + "evidence", + "evidenceRefs", + "evidence_refs", + "limitations", + ), + ), + ( + "attackPath", + ( + "assumptions", + "blindspots", + "controls", + "evidenceRefs", + "evidence_refs", + "limitations", + "preconditions", + "steps", + ), + ), + ): + section = finding.get(section_name) + if not isinstance(section, dict): + continue + _normalize_legacy_string_list_fields(section, list_fields) + _filter_unknown_legacy_evidence_refs(section, evidence_ids) + legacy_root_cause = finding.get("root_cause") + if isinstance(legacy_root_cause, dict): + _remove_unsupported_legacy_scalar_fields( + legacy_root_cause, ("summary", "code", "language") + ) + elif ( + "root_cause" in finding + and legacy_root_cause is not None + and (not isinstance(legacy_root_cause, str) or legacy_root_cause == "") + ): + finding.pop("root_cause") + validation = finding.get("validation") + if isinstance(validation, dict): + _remove_unsupported_legacy_scalar_fields( + validation, ("method", "status", "summary", "disposition", "result") + ) + attack_path = finding.get("attackPath") + if not isinstance(attack_path, dict): + continue + _remove_unsupported_legacy_scalar_fields(attack_path, ("summary",)) + for field in ("dataFlow", "data_flow", "dataflow", "reachability"): + if field not in attack_path: + continue + detail = attack_path.get(field) + if detail is None: + attack_path.pop(field) + continue + if not isinstance(detail, (str, dict)): + attack_path.pop(field) + continue + if isinstance(detail, str): + if detail == "": + attack_path.pop(field) + continue + detail_scalar_fields = ("summary", "source", "sink", "outcome") + if field == "reachability": + detail_scalar_fields += ("attacker", "entrypoint") + _remove_unsupported_legacy_scalar_fields(detail, detail_scalar_fields) + _normalize_legacy_string_list_fields( + detail, ("evidenceRefs", "evidence_refs", "transformations") + ) + _filter_unknown_legacy_evidence_refs(detail, evidence_ids) + if field == "reachability": + _normalize_legacy_string_list_fields(detail, ("preconditions",)) + for field in ("impact", "likelihood"): + detail = attack_path.get(field) + if isinstance(detail, dict): + _remove_unsupported_legacy_scalar_fields( + detail, ("level", "rationale", "why") + ) + elif detail is not None and ( + not isinstance(detail, str) or detail == "" + ): + attack_path.pop(field) + return compatible + + def _validate_canonical_schemas_before_projection( manifest: dict[str, Any], findings: dict[str, Any], @@ -1608,21 +1816,94 @@ def _sarif_primary_location(finding: dict[str, Any]) -> dict[str, Any]: ) +def _merged_code_evidence(finding: dict[str, Any]) -> list[dict[str, Any]]: + catalog: dict[str, dict[str, Any]] = {} + for evidence_key in ("codeEvidence", "code_evidence"): + code_evidence = finding.get(evidence_key) + if not isinstance(code_evidence, list): + continue + for evidence in code_evidence: + if not isinstance(evidence, dict): + continue + evidence_id = evidence.get("id") + if isinstance(evidence_id, str) and evidence_id: + catalog.setdefault(evidence_id, evidence) + return list(catalog.values()) + + +def _finding_evidence_strength(finding: dict[str, Any]) -> int: + evidence = _merged_code_evidence(finding) + seen_ids = { + item["id"] for item in evidence if isinstance(item.get("id"), str) and item["id"].strip() + } + seen_codes = { + item["code"] + for item in evidence + if isinstance(item.get("code"), str) and item["code"].strip() + } + strength = len(evidence) + for section_name in ("rootCause", "root_cause"): + section = finding.get(section_name) + if not isinstance(section, dict): + continue + for evidence_name in ("codeEvidence", "code_evidence"): + embedded = section.get(evidence_name) + if not isinstance(embedded, list): + continue + for item in embedded: + if not isinstance(item, dict): + continue + code = item.get("code") + if not isinstance(code, str) or not code.strip() or code in seen_codes: + continue + evidence_id = item.get("id") + if isinstance(evidence_id, str) and evidence_id.strip(): + if evidence_id in seen_ids: + continue + seen_ids.add(evidence_id) + seen_codes.add(code) + strength += 1 + code = section.get("code") + if isinstance(code, str) and code.strip() and code not in seen_codes: + seen_codes.add(code) + strength += 1 + return strength + + def _sarif_locations(finding: dict[str, Any]) -> list[dict[str, Any]]: primary = _sarif_primary_location(finding) locations = [ primary, *(location for location in finding["locations"] if location is not primary), ] - locations.extend( - { - "path": evidence["path"], - "startLine": evidence["startLine"], - "endLine": evidence.get("endLine", evidence["startLine"]), - "role": f"evidence:{evidence['id']}", - } - for evidence in finding.get("codeEvidence", []) - ) + for evidence in _merged_code_evidence(finding): + path = evidence.get("path") + start_line = evidence.get("startLine") + if ( + not isinstance(path, str) + or not isinstance(start_line, int) + or isinstance(start_line, bool) + or start_line < 1 + ): + continue + try: + path = _require_safe_relative_path(path, "SARIF evidence location") + except ContractError: + continue + locations.append( + { + "path": path, + "startLine": start_line, + "endLine": ( + evidence["endLine"] + if isinstance(evidence.get("endLine"), int) + and not isinstance(evidence["endLine"], bool) + and evidence["endLine"] >= start_line + else start_line + ), + "role": f"evidence:{evidence['id']}", + } + ) unique: dict[tuple[str, int, int], dict[str, Any]] = {} for location in locations: key = ( @@ -1894,11 +2175,12 @@ def _read_sealed_scan( }, ) _validate_manifest(manifest) - _validate_findings(manifest, findings) + findings_for_validation = _legacy_sealed_findings_for_validation(findings) + _validate_findings(manifest, findings_for_validation) _validate_coverage(manifest, coverage, scan_dir) _validate_sealed_coverage_receipts(scan, coverage) validate_against_schema(manifest, schema_dir / "scan-manifest.schema.json") - validate_against_schema(findings, schema_dir / "findings.schema.json") + validate_against_schema(findings_for_validation, schema_dir / "findings.schema.json") validate_against_schema(coverage, schema_dir / "coverage.schema.json") _validate_derived_finding_identities(manifest, findings) return manifest, findings, coverage, findings_bytes @@ -2196,8 +2478,11 @@ def _prepare_scan_finalization( scan["sealedAt"] = _require_str(scan, "completedAt", "manifest.scan") _validate_target(_require_dict(scan, "target", "manifest.scan")) _validate_completion_binding(manifest, findings, coverage, completion_binding) + findings_for_validation = ( + _legacy_sealed_findings_for_validation(findings) if was_sealed else findings + ) if was_sealed: - _validate_findings(manifest, findings) + _validate_findings(manifest, findings_for_validation) _validate_derived_finding_identities(manifest, findings) elif completion_warnings is not None: discarded_findings = _recover_unsealed_findings( @@ -2209,16 +2494,20 @@ def _prepare_scan_finalization( _recover_unsealed_hardening(manifest, scan_dir, completion_warnings) else: _populate_unsealed_finding_identities(manifest, findings) - _validate_findings(manifest, findings) + _validate_findings(manifest, findings_for_validation) _validate_coverage(manifest, coverage, scan_dir) - _validate_canonical_schemas_before_projection(manifest, findings, coverage, schema_dir) + _validate_canonical_schemas_before_projection( + manifest, findings_for_validation, coverage, schema_dir + ) _require_derived_writeup_files(scan_dir, findings) _require_hardening_portfolio_file(scan_dir, scan) if was_sealed: _validate_sealed_coverage_receipts(scan, coverage) _validate_manifest(manifest) validate_against_schema(manifest, schema_dir / "scan-manifest.schema.json") - validate_against_schema(findings, schema_dir / "findings.schema.json") + validate_against_schema( + findings_for_validation, schema_dir / "findings.schema.json" + ) validate_against_schema(coverage, schema_dir / "coverage.schema.json") report_markdown_bytes = _generate_report_projection(manifest, findings, coverage) _validate_report_output_paths(scan_dir) diff --git a/sdk/typescript/_bundled_plugin/scripts/finding_preview.py b/sdk/typescript/_bundled_plugin/scripts/finding_preview.py index 82d82e087..15ee5771b 100644 --- a/sdk/typescript/_bundled_plugin/scripts/finding_preview.py +++ b/sdk/typescript/_bundled_plugin/scripts/finding_preview.py @@ -4,11 +4,13 @@ import argparse import json +import re import sys from pathlib import Path from typing import Any sys.path.insert(0, str(Path(__file__).resolve().parent)) +from report_projection import merged_root_cause from workbench_constants import ( FINDING_ATTACK_PATH_PREVIEW_BYTES, FINDING_CODE_EVIDENCE_LIMIT, @@ -61,6 +63,7 @@ def bounded_finding_details(value: Any) -> dict[str, Any]: "evidence_refs", "assertions", "evidence", + "counterEvidence", "limitations", ), ( @@ -70,6 +73,7 @@ def bounded_finding_details(value: Any) -> dict[str, Any]: (("evidenceRefs", "evidence_refs"), 400), (("assertions",), 400), (("evidence",), 400), + (("counterEvidence",), 400), (("limitations",), 400), ), ), @@ -109,14 +113,27 @@ def bounded_finding_details(value: Any) -> dict[str, Any]: ), ), ): - key = next((alias for alias in aliases if alias in value), None) + if aliases == ("rootCause", "root_cause"): + key, section = merged_root_cause(value) + else: + key = next((alias for alias in aliases if alias in value), None) + section = value[key] if key is not None else None if key is not None: - section = value[key] if key == "attackPath" and isinstance(section, dict): section = dict(section) for assessment in ("impact", "likelihood"): if isinstance(section.get(assessment), str): - section[assessment] = {"level": section[assessment]} + assessment_value = section[assessment] + assessment_key = ( + "level" + if re.fullmatch( + r"critical|high|medium|low|informational|ignore|unknown", + assessment_value, + flags=re.IGNORECASE, + ) + else "rationale" + ) + section[assessment] = {assessment_key: assessment_value} prepared[key] = bounded_finding_section( section, maximum_bytes, @@ -128,12 +145,9 @@ def bounded_finding_details(value: Any) -> dict[str, Any]: if isinstance(writeup, dict) and isinstance(writeup.get("reportPath"), str): prepared["writeup"] = {"reportPath": bounded_json_text(writeup["reportPath"], 512)[0]} - evidence_key = next( - (key for key in ("codeEvidence", "code_evidence") if key in value), - None, - ) + evidence_key, evidence = merged_code_evidence(value) if evidence_key is not None: - prepared[evidence_key] = bounded_code_evidence(value[evidence_key]) + prepared[evidence_key] = bounded_code_evidence(evidence) for key in ( "confidence", @@ -185,9 +199,7 @@ def bounded_finding_details(value: Any) -> dict[str, Any]: ) core = {key: prepared[key] for key in core_keys if key in prepared} extras = { - key: item - for key, item in prepared.items() - if key not in core and key not in guidance + key: item for key, item in prepared.items() if key not in core and key not in guidance } complete_guidance = {key: items[:1] for key, items in guidance.items()} minimum_guidance = { @@ -233,25 +245,75 @@ def bounded_finding_section( for key in (*priority_keys, *value): if key in value and key not in ordered: ordered[key] = value[key] - evidence_key = next( - (key for key in ("codeEvidence", "code_evidence") if key in ordered), - None, - ) + evidence_key, evidence = merged_code_evidence(ordered) if evidence_key is not None: - ordered[evidence_key] = bounded_code_evidence(ordered[evidence_key]) + ordered[evidence_key] = bounded_code_evidence(evidence) ordered.pop("code_evidence" if evidence_key == "codeEvidence" else "codeEvidence", None) return bounded_json_value(ordered, [maximum_bytes]) +def merged_code_evidence(value: dict[str, Any]) -> tuple[str | None, Any]: + evidence_keys = [key for key in ("codeEvidence", "code_evidence") if key in value] + if not evidence_keys: + return None, None + catalogs = [value[key] for key in evidence_keys if isinstance(value[key], list)] + if catalogs: + merged = [] + seen_ids: set[str] = set() + for catalog in catalogs: + for item in catalog: + if not _is_valid_code_evidence(item): + continue + evidence_id = item["id"] + if evidence_id in seen_ids: + continue + seen_ids.add(evidence_id) + merged.append(item) + return evidence_keys[0], merged + return evidence_keys[0], value[evidence_keys[0]] + + +def _is_valid_code_evidence(item: Any) -> bool: + return ( + isinstance(item, dict) + and isinstance(item.get("id"), str) + and bool(item["id"].strip()) + and isinstance(item.get("code"), str) + and bool(item["code"].strip()) + ) + + def bounded_code_evidence(value: Any) -> Any: if not isinstance(value, list): return value bounded = [] - for item in value[:FINDING_CODE_EVIDENCE_LIMIT]: - if not isinstance(item, dict): - bounded.append(item) + for item in value: + if not _is_valid_code_evidence(item): continue + if len(bounded) >= FINDING_CODE_EVIDENCE_LIMIT: + break evidence = dict(item) + for field in ("explanation", "label", "language", "path"): + if field in evidence and not isinstance(evidence[field], str): + evidence.pop(field) + if ( + "role" in evidence + and evidence["role"] is not None + and not isinstance(evidence["role"], str) + ): + evidence.pop("role") + start_line = evidence.get("startLine") + if "startLine" in evidence and ( + not isinstance(start_line, int) or isinstance(start_line, bool) or start_line < 1 + ): + evidence.pop("startLine") + end_line = evidence.get("endLine") + if ( + "endLine" in evidence + and end_line is not None + and (not isinstance(end_line, int) or isinstance(end_line, bool) or end_line < 1) + ): + evidence.pop("endLine") code = evidence.get("code") if isinstance(code, str): evidence["code"] = bounded_json_text( diff --git a/sdk/typescript/_bundled_plugin/scripts/report_projection.py b/sdk/typescript/_bundled_plugin/scripts/report_projection.py index 83f4af518..6149d5f4f 100644 --- a/sdk/typescript/_bundled_plugin/scripts/report_projection.py +++ b/sdk/typescript/_bundled_plugin/scripts/report_projection.py @@ -47,6 +47,8 @@ def _escape_markdown_text(value: str) -> str: def _strings(value: Any) -> list[str]: + if isinstance(value, str): + value = [value] if not isinstance(value, list): return [] normalized: list[str] = [] @@ -240,36 +242,46 @@ def _bullets(items: list[str], fallback: str) -> list[str]: def _code_evidence_catalog(finding: dict[str, Any]) -> dict[str, dict[str, Any]]: - raw = finding.get("codeEvidence", finding.get("code_evidence", [])) - if not isinstance(raw, list): - return {} - return { - item["id"]: item - for item in raw - if isinstance(item, dict) - and isinstance(item.get("id"), str) - and isinstance(item.get("code"), str) - and item["code"].strip() - } + catalog: dict[str, dict[str, Any]] = {} + for key in ("codeEvidence", "code_evidence"): + raw = finding.get(key, []) + if not isinstance(raw, list): + continue + for item in raw: + if ( + isinstance(item, dict) + and isinstance(item.get("id"), str) + and isinstance(item.get("code"), str) + and item["code"].strip() + ): + catalog.setdefault(item["id"], item) + return catalog def _section_code_evidence( - finding: dict[str, Any], section: dict[str, Any] + finding: dict[str, Any], *sections: dict[str, Any] ) -> list[dict[str, Any]]: catalog = _code_evidence_catalog(finding) - refs = section.get("evidenceRefs", section.get("evidence_refs", [])) - resolved = ( - [catalog[ref] for ref in refs if isinstance(ref, str) and ref in catalog] - if isinstance(refs, list) - else [] - ) - embedded = section.get("codeEvidence", section.get("code_evidence", [])) - if isinstance(embedded, list): - resolved.extend( - item - for item in embedded - if isinstance(item, dict) and isinstance(item.get("code"), str) and item["code"].strip() - ) + resolved: list[dict[str, Any]] = [] + for section in sections: + for key in ("evidenceRefs", "evidence_refs"): + refs = section.get(key, []) + if isinstance(refs, str): + refs = [refs] + if isinstance(refs, list): + resolved.extend( + catalog[ref] for ref in refs if isinstance(ref, str) and ref in catalog + ) + for key in ("codeEvidence", "code_evidence"): + embedded = section.get(key, []) + if isinstance(embedded, list): + resolved.extend( + item + for item in embedded + if isinstance(item, dict) + and isinstance(item.get("code"), str) + and item["code"].strip() + ) unique: list[dict[str, Any]] = [] seen: set[tuple[str, str]] = set() for item in resolved: @@ -281,6 +293,102 @@ def _section_code_evidence( return unique +def merged_root_cause(value: dict[str, Any]) -> tuple[str | None, Any]: + keys = [key for key in ("rootCause", "root_cause") if key in value] + if not keys: + return None, None + if len(keys) == 1: + detail = value[keys[0]] + return keys[0], detail if isinstance(detail, (str, dict)) else None + details: list[dict[str, Any]] = [] + for key in keys: + detail = value[key] + if isinstance(detail, str): + details.append({"summary": detail}) + elif isinstance(detail, dict): + details.append(detail) + elif detail is not None: + continue + if not details: + return keys[0], None + + merged: dict[str, Any] = {} + text_fields = { + "cause", + "code", + "description", + "detail", + "explanation", + "rationale", + "summary", + "why", + } + for detail in details: + for field, item in detail.items(): + if field in ( + "evidenceRefs", + "evidence_refs", + "codeEvidence", + "code_evidence", + "language", + ): + continue + if field in text_fields and ( + not isinstance(item, str) or not item.strip() + ): + continue + current = merged.get(field) + if field not in merged or current in (None, "", [], {}) or ( + isinstance(current, str) and not current.strip() + ): + merged[field] = item + + evidence_values = [ + detail[field] + for detail in details + for field in ("evidenceRefs", "evidence_refs") + if field in detail + ] + if evidence_values: + merged["evidenceRefs"] = list( + dict.fromkeys( + item + for evidence in evidence_values + for item in (evidence if isinstance(evidence, list) else [evidence]) + if isinstance(item, str) + ) + ) + + embedded_evidence = [ + item + for detail in details + for field in ("codeEvidence", "code_evidence") + for item in ( + detail.get(field, []) if isinstance(detail.get(field), list) else [] + ) + ] + if embedded_evidence: + merged["codeEvidence"] = embedded_evidence + + code = merged.get("code") + matching_details = ( + details + if not isinstance(code, str) or not code.strip() + else [detail for detail in details if detail.get("code") == code] + ) + language = next( + ( + detail["language"] + for detail in matching_details + if isinstance(detail.get("language"), str) and detail["language"].strip() + ), + None, + ) + if language is not None: + merged["language"] = language + return keys[0], merged + + def _root_cause_code_evidence( finding: dict[str, Any], root_cause: dict[str, Any] ) -> list[dict[str, Any]]: @@ -419,34 +527,90 @@ def _surface_notes(surface: dict[str, Any]) -> str: 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 = finding.get("rootCause") + _, raw_root_cause = merged_root_cause(finding) root_cause = raw_root_cause if isinstance(raw_root_cause, dict) else {} attack_path = finding.get("attackPath") if isinstance(finding.get("attackPath"), dict) else {} - dataflow = attack_path.get("dataflow") if isinstance(attack_path.get("dataflow"), dict) else {} + dataflow_sections = [ + {"summary": value} if isinstance(value, str) else value + for key in ("dataFlow", "dataflow", "data_flow") + if isinstance((value := attack_path.get(key)), (str, dict)) + ] + dataflow: dict[str, Any] = {} + for key in ("summary", "source", "sink", "outcome"): + value = next( + ( + section[key] + for section in dataflow_sections + if key in section + and isinstance(section[key], str) + and section[key].strip() + ), + None, + ) + if value is not None: + dataflow[key] = value + transformations = list( + dict.fromkeys( + transformation + for section in dataflow_sections + for transformation in ( + [section.get("transformations")] + if isinstance(section.get("transformations"), str) + else section.get("transformations") + if isinstance(section.get("transformations"), list) + else [] + ) + if isinstance(transformation, str) and transformation.strip() + ) + ) + if transformations: + dataflow["transformations"] = transformations + raw_reachability = attack_path.get("reachability") reachability = ( - attack_path.get("reachability") if isinstance(attack_path.get("reachability"), dict) else {} + {"summary": raw_reachability} + if isinstance(raw_reachability, str) + else raw_reachability + if isinstance(raw_reachability, dict) + else {} ) severity = finding["severity"] + validation_outcomes = [ + (label, text) + for label, key in ( + ("Status", "status"), + ("Disposition", "disposition"), + ("Result", "result"), + ) + if (text := _text(validation.get(key), "")) + ] validation_summary = _text( validation.get("summary"), - f"{finding['confidence']['rationale']} Validation details were not recorded separately.", + "Validation outcomes are recorded below." + if validation_outcomes + else f"{finding['confidence']['rationale']} Validation details were not recorded separately.", ) validation_evidence = _strings(validation.get("evidence")) + validation_assertions = _strings(validation.get("assertions")) validation_counterevidence = _strings(validation.get("counterEvidence")) + validation_limitations = _strings(validation.get("limitations")) root_cause_summary = _text( raw_root_cause if isinstance(raw_root_cause, str) else root_cause.get("summary"), "", ) root_cause_code_evidence = _root_cause_code_evidence(finding, root_cause) validation_code_evidence = _section_code_evidence(finding, validation) - attack_path_code_evidence = _section_code_evidence(finding, attack_path) + dataflow_code_evidence = _section_code_evidence(finding, attack_path, *dataflow_sections) + reachability_code_evidence = _section_code_evidence(finding, reachability) dataflow_summary = _text( dataflow.get("summary"), f"The canonical finding records the affected path at {_locations(finding)}, but no expanded source-to-sink narrative was recorded.", ) reachability_summary = _text( reachability.get("summary"), - "Reachability was not recorded beyond the canonical finding summary and affected locations.", + _text( + attack_path.get("summary"), + "Reachability was not recorded beyond the canonical finding summary and affected locations.", + ), ) severity_rationale = _text( severity.get("rationale"), @@ -458,6 +622,7 @@ def _finding_section(number: int, finding: dict[str, Any]) -> list[str]: ) 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") lines = [ @@ -486,7 +651,13 @@ def _finding_section(number: int, finding: dict[str, Any]) -> list[str]: lines.extend(["", "#### Validation", "", validation_summary]) if validation.get("method"): lines.extend(["", f"Validation method: {_text(validation['method'], 'not recorded')}"]) + if validation_outcomes: + lines.extend( + ["", *(f"- **{label}:** {value}" for label, value in validation_outcomes)] + ) lines.extend(_code_evidence_lines(validation_code_evidence)) + if validation_assertions: + lines.extend(["", "Assertions:", *_bullets(validation_assertions, "None recorded.")]) if validation_evidence: lines.extend(["", "Evidence:", *_bullets(validation_evidence, "No evidence recorded.")]) if validation_counterevidence: @@ -497,25 +668,48 @@ def _finding_section(number: int, finding: dict[str, Any]) -> list[str]: *_bullets(validation_counterevidence, "None recorded."), ] ) + if validation_limitations: + lines.extend(["", "Limitations:", *_bullets(validation_limitations, "None recorded.")]) lines.extend(["", "#### Dataflow", "", dataflow_summary]) + if attack_steps: + lines.extend(["", "Attack steps:", *_bullets(attack_steps, "None recorded.")]) for label, key in (("Source", "source"), ("Sink", "sink"), ("Outcome", "outcome")): if dataflow.get(key): lines.extend(["", f"- **{label}:** {_text(dataflow[key], 'not recorded')}"]) transformations = _strings(dataflow.get("transformations")) if transformations: lines.extend(["", "Transformations:", *_bullets(transformations, "None recorded.")]) - lines.extend(_code_evidence_lines(attack_path_code_evidence)) + lines.extend(_code_evidence_lines(dataflow_code_evidence)) lines.extend(["", "#### Reachability", "", reachability_summary]) for label, key in ( ("Attacker", "attacker"), ("Entry point", "entrypoint"), + ("Source", "source"), + ("Sink", "sink"), ("Outcome", "outcome"), ): if reachability.get(key): lines.extend(["", f"- **{label}:** {_text(reachability[key], 'not recorded')}"]) - preconditions = _strings(reachability.get("preconditions")) + preconditions = list( + dict.fromkeys( + [ + *_strings(attack_path.get("preconditions")), + *_strings(reachability.get("preconditions")), + ] + ) + ) if preconditions: lines.extend(["", "Preconditions:", *_bullets(preconditions, "None recorded.")]) + for label, key in ( + ("Assumptions", "assumptions"), + ("Existing controls", "controls"), + ("Blind spots", "blindspots"), + ("Limitations", "limitations"), + ): + values = _strings(attack_path.get(key)) + if values: + lines.extend(["", f"{label}:", *_bullets(values, "None recorded.")]) + lines.extend(_code_evidence_lines(reachability_code_evidence)) lines.extend( [ "", @@ -524,6 +718,32 @@ def _finding_section(number: int, finding: dict[str, Any]) -> list[str]: f"**{severity['level'].capitalize()}** — {severity_rationale}", "", severity_change, + ] + ) + for label, key in (("Impact", "impact"), ("Likelihood", "likelihood")): + assessment = attack_path.get(key) + if isinstance(assessment, str): + rendered = _text(assessment, "") + if rendered: + lines.extend(["", f"**{label} assessment:** {rendered}"]) + continue + if not isinstance(assessment, dict): + continue + details = [ + (detail_label, text) + for detail_label, detail_key in ( + ("Level", "level"), + ("Rationale", "rationale"), + ("Why", "why"), + ) + if (text := _text(assessment.get(detail_key), "")) + ] + if details: + lines.extend( + ["", f"{label} assessment:", *(f"- **{name}:** {value}" for name, value in details)] + ) + lines.extend( + [ "", "#### Remediation", "", diff --git a/sdk/typescript/_bundled_plugin/scripts/validate_scan_contract.py b/sdk/typescript/_bundled_plugin/scripts/validate_scan_contract.py index 257a78a29..abcf313fe 100644 --- a/sdk/typescript/_bundled_plugin/scripts/validate_scan_contract.py +++ b/sdk/typescript/_bundled_plugin/scripts/validate_scan_contract.py @@ -59,11 +59,14 @@ def validate_contract(scan_dir: Path) -> dict[str, Any]: coverage_ref: coverage_bytes, }, ) - FINALIZER._validate_findings(manifest, findings) + findings_for_validation = FINALIZER._legacy_sealed_findings_for_validation(findings) + FINALIZER._validate_findings(manifest, findings_for_validation) FINALIZER._validate_derived_finding_identities(manifest, findings) FINALIZER._validate_coverage(manifest, coverage, scan_dir) FINALIZER._validate_sealed_coverage_receipts(scan, coverage) - FINALIZER.validate_against_schema(findings, schema_dir / "findings.schema.json") + FINALIZER.validate_against_schema( + findings_for_validation, schema_dir / "findings.schema.json" + ) FINALIZER.validate_against_schema(coverage, schema_dir / "coverage.schema.json") FINALIZER._require_scan_local_file(scan_dir, "report.md", "report.md") return { diff --git a/sdk/typescript/_bundled_plugin/scripts/workbench_cli.py b/sdk/typescript/_bundled_plugin/scripts/workbench_cli.py index 77e9d56ac..2fc1c8f86 100644 --- a/sdk/typescript/_bundled_plugin/scripts/workbench_cli.py +++ b/sdk/typescript/_bundled_plugin/scripts/workbench_cli.py @@ -309,7 +309,13 @@ def parse_args(description: str) -> argparse.Namespace: publication.add_argument("--input-file", required=True) subparsers.add_parser("database-info") - return parser.parse_args() + arguments = sys.argv[1:] + if "--user-context-stdin" in arguments: + if arguments.count("--user-context-stdin") != 1 or "--user-context" in arguments: + parser.error("pass exactly one user-context transport") + index = arguments.index("--user-context-stdin") + arguments[index : index + 1] = ["--user-context", sys.stdin.read()] + return parser.parse_args(arguments) def non_negative_int(value: str) -> int: diff --git a/sdk/typescript/scripts/generate-models.cjs b/sdk/typescript/scripts/generate-models.cjs index 03f3dd247..f0b798e1c 100644 --- a/sdk/typescript/scripts/generate-models.cjs +++ b/sdk/typescript/scripts/generate-models.cjs @@ -70,8 +70,8 @@ async function generate() { 'export type FindingRootCause = Extract;', 'export type FindingValidation = NonNullable;', 'export type FindingAttackPath = NonNullable;', - "export type AttackPathDataflow = ContractObject;", - "export type AttackPathReachability = ContractObject;", + 'export type AttackPathDataflow = Extract, object>;', + 'export type AttackPathReachability = Extract, object>;', 'export type FindingProvenance = Finding["provenance"];', 'export type CoverageMode = CoverageDocument["mode"];', 'export type CoverageCompleteness = CoverageDocument["completeness"];', diff --git a/sdk/typescript/src/contract.ts b/sdk/typescript/src/contract.ts index 559f17a1e..7e80a8fa4 100644 --- a/sdk/typescript/src/contract.ts +++ b/sdk/typescript/src/contract.ts @@ -99,6 +99,7 @@ export async function loadContract( ), }; throwIfAborted(options.signal); + let findingsPayload: unknown = payloads["findings.json"]; const ajv = createValidator(); for (const [filename, schemaName] of Object.entries(DOCUMENTS)) { @@ -112,7 +113,10 @@ export async function loadContract( } catch { throw new ContractValidationError(`${schemaName}: invalid JSON Schema.`); } - const payload = payloads[filename as keyof typeof payloads]; + let payload: unknown = + filename === "findings.json" + ? findingsPayload + : payloads[filename as keyof typeof payloads]; let valid: boolean; try { const result = validate(payload); @@ -120,16 +124,25 @@ export async function loadContract( throw new Error("asynchronous JSON Schema validation is unsupported"); } valid = result; + if (!valid && filename === "findings.json") { + payload = legacySealedFindingsForValidation(payload); + const compatibleResult = validate(payload); + if (typeof compatibleResult !== "boolean") { + throw new Error("asynchronous JSON Schema validation is unsupported"); + } + valid = compatibleResult; + } } catch { throw new ContractValidationError(`${schemaName}: invalid JSON Schema.`); } if (!valid) { throw schemaError(filename, validate.errors ?? []); } + if (filename === "findings.json") findingsPayload = payload; throwIfAborted(options.signal); } const manifest = payloads["scan-manifest.json"] as unknown as ScanManifest; - const findings = payloads["findings.json"] as unknown as FindingsDocument; + const findings = findingsPayload as FindingsDocument; const coverage = payloads["coverage.json"] as unknown as CoverageDocument; if ( findings.scanId !== manifest.scan.id || @@ -173,6 +186,210 @@ export async function loadContract( return { manifest, findings, coverage }; } +type JsonRecord = Record; + +function legacySealedFindingsForValidation(payload: unknown): unknown { + const compatible = structuredClone(payload); + if (!isJsonRecord(compatible) || !Array.isArray(compatible["findings"])) { + return compatible; + } + for (const finding of compatible["findings"]) { + if (!isJsonRecord(finding)) continue; + const legacyEvidence = finding["code_evidence"]; + if (Array.isArray(legacyEvidence)) { + const compatibleEvidence: JsonRecord[] = []; + for (const evidence of legacyEvidence) { + if (!isJsonRecord(evidence)) continue; + const id = evidence["id"]; + const code = evidence["code"]; + if ( + typeof id !== "string" || + id.length === 0 || + typeof code !== "string" || + code.length === 0 + ) { + continue; + } + compatibleEvidence.push(evidence); + } + finding["code_evidence"] = compatibleEvidence; + } else if ("code_evidence" in finding && legacyEvidence !== null) { + delete finding["code_evidence"]; + } + + for (const [sectionName, listFields] of [ + ["root_cause", ["evidenceRefs", "evidence_refs"]], + [ + "validation", + [ + "assertions", + "counterEvidence", + "evidenceRefs", + "evidence_refs", + "limitations", + ], + ], + [ + "attackPath", + [ + "assumptions", + "blindspots", + "controls", + "evidenceRefs", + "evidence_refs", + "limitations", + "preconditions", + "steps", + ], + ], + ] satisfies Array<[string, string[]]>) { + const section = finding[sectionName]; + if (!isJsonRecord(section)) continue; + normalizeLegacyStringLists(section, listFields); + } + + const legacyRootCause = finding["root_cause"]; + if (isJsonRecord(legacyRootCause)) { + removeUnsupportedLegacyStrings(legacyRootCause, [ + "summary", + "code", + "language", + ]); + } else if ( + "root_cause" in finding && + legacyRootCause !== null && + typeof legacyRootCause !== "string" + ) { + delete finding["root_cause"]; + } + + const validation = finding["validation"]; + if (isJsonRecord(validation)) { + normalizeLegacyStringOrList(validation, "evidence"); + removeUnsupportedLegacyStrings(validation, ["method", "summary"]); + removeUnsupportedLegacyNullableStrings(validation, [ + "status", + "disposition", + "result", + ]); + } + + const attackPath = finding["attackPath"]; + if (!isJsonRecord(attackPath)) continue; + removeUnsupportedLegacyStrings(attackPath, ["summary"]); + for (const field of ["dataFlow", "data_flow", "dataflow", "reachability"]) { + const detail = attackPath[field]; + if (detail === null) { + delete attackPath[field]; + continue; + } + if (typeof detail === "string") { + if (detail.length === 0) delete attackPath[field]; + continue; + } + if (!isJsonRecord(detail)) { + if (field in attackPath) delete attackPath[field]; + continue; + } + removeUnsupportedLegacyStrings(detail, [ + "summary", + "source", + "sink", + "outcome", + ...(field === "reachability" ? ["attacker", "entrypoint"] : []), + ]); + normalizeLegacyStringLists(detail, [ + "evidenceRefs", + "evidence_refs", + "transformations", + ...(field === "reachability" ? ["preconditions"] : []), + ]); + } + for (const field of ["impact", "likelihood"]) { + const detail = attackPath[field]; + if (isJsonRecord(detail)) { + removeUnsupportedLegacyStrings(detail, ["level", "rationale", "why"]); + } else if ( + detail !== undefined && + detail !== null && + (typeof detail !== "string" || detail.length === 0) + ) { + delete attackPath[field]; + } + } + } + return compatible; +} + +function isJsonRecord(value: unknown): value is JsonRecord { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function normalizeLegacyStringLists( + section: JsonRecord, + fields: string[], +): void { + for (const field of fields) { + if (!(field in section)) continue; + const value = section[field]; + if (Array.isArray(value)) { + section[field] = value.filter( + (item): item is string => typeof item === "string" && item.length > 0, + ); + } else if (typeof value === "string" && value.length > 0) { + section[field] = [value]; + } else { + delete section[field]; + } + } +} + +function normalizeLegacyStringOrList(section: JsonRecord, field: string): void { + if (!(field in section)) return; + const value = section[field]; + if (typeof value === "string") { + if (value.length === 0) delete section[field]; + return; + } + if (!Array.isArray(value)) { + delete section[field]; + return; + } + const normalized = value.filter( + (item): item is string => typeof item === "string" && item.length > 0, + ); + section[field] = normalized; +} + +function removeUnsupportedLegacyStrings( + section: JsonRecord, + fields: string[], +): void { + for (const field of fields) { + if ( + field in section && + (typeof section[field] !== "string" || section[field].length === 0) + ) { + delete section[field]; + } + } +} + +function removeUnsupportedLegacyNullableStrings( + section: JsonRecord, + fields: string[], +): void { + for (const field of fields) { + if ( + field in section && + section[field] !== null && + (typeof section[field] !== "string" || section[field].length === 0) + ) { + delete section[field]; + } + } +} + function validateCanonicalContract( manifest: ScanManifest, findings: FindingsDocument, diff --git a/sdk/typescript/src/models.ts b/sdk/typescript/src/models.ts index 1cfd25050..b9d9fefd6 100644 --- a/sdk/typescript/src/models.ts +++ b/sdk/typescript/src/models.ts @@ -130,6 +130,13 @@ export interface FindingsDocument { explanation: string; [k: string]: unknown; }[]; + code_evidence?: + | { + id: string; + code: string; + [k: string]: unknown; + }[] + | null; rootCause?: | { summary: string; @@ -139,11 +146,110 @@ export interface FindingsDocument { [k: string]: unknown; } | string; + root_cause?: + | { + summary?: string; + evidenceRefs?: string[]; + evidence_refs?: string[]; + code?: string; + language?: string; + [k: string]: unknown; + } + | string + | null; remediation: string; validation?: { + assertions?: string[]; + counterEvidence?: string[]; + evidence?: string | string[]; + evidenceRefs?: string[]; + evidence_refs?: string[]; + limitations?: string[]; + method?: string; + status?: string | null; + summary?: string; + disposition?: string | null; + result?: string | null; [k: string]: unknown; } | null; attackPath?: { + assumptions?: string[]; + blindspots?: string[]; + controls?: string[]; + dataFlow?: + | string + | { + summary?: string; + source?: string; + sink?: string; + outcome?: string; + transformations?: string[]; + evidenceRefs?: string[]; + evidence_refs?: string[]; + [k: string]: unknown; + }; + data_flow?: + | string + | { + summary?: string; + source?: string; + sink?: string; + outcome?: string; + transformations?: string[]; + evidenceRefs?: string[]; + evidence_refs?: string[]; + [k: string]: unknown; + }; + dataflow?: + | string + | { + summary?: string; + source?: string; + sink?: string; + outcome?: string; + transformations?: string[]; + evidenceRefs?: string[]; + evidence_refs?: string[]; + [k: string]: unknown; + }; + evidenceRefs?: string[]; + evidence_refs?: string[]; + impact?: + | string + | { + level?: string; + rationale?: string; + why?: string; + [k: string]: unknown; + } + | null; + likelihood?: + | string + | { + level?: string; + rationale?: string; + why?: string; + [k: string]: unknown; + } + | null; + limitations?: string[]; + preconditions?: string[]; + reachability?: + | string + | { + summary?: string; + attacker?: string; + entrypoint?: string; + source?: string; + sink?: string; + outcome?: string; + preconditions?: string[]; + evidenceRefs?: string[]; + evidence_refs?: string[]; + [k: string]: unknown; + }; + steps?: string[]; + summary?: string; [k: string]: unknown; } | null; remediationTests?: string[]; @@ -264,9 +370,15 @@ export type FindingValidation = NonNullable; export type FindingAttackPath = NonNullable; -export type AttackPathDataflow = ContractObject; +export type AttackPathDataflow = Extract< + NonNullable, + object +>; -export type AttackPathReachability = ContractObject; +export type AttackPathReachability = Extract< + NonNullable, + object +>; export type FindingProvenance = Finding["provenance"]; diff --git a/sdk/typescript/tests-ts/compact-diff-scan.test.ts b/sdk/typescript/tests-ts/compact-diff-scan.test.ts index c37c126ef..5407c0cd8 100644 --- a/sdk/typescript/tests-ts/compact-diff-scan.test.ts +++ b/sdk/typescript/tests-ts/compact-diff-scan.test.ts @@ -447,10 +447,10 @@ describe("compact diff scan", () => { expect(tools.map((tool) => tool.name)).toContain( "record_codex_security_discovery_candidates", ); - expect( - tools.find((tool) => tool.name === "start_codex_security_standard_scan") - ?.inputSchema.properties["userContext"]?.maxLength, - ).toBe(2400); + const preservedContextMaxLength = tools.find( + (tool) => tool.name === "start_codex_security_standard_scan", + )?.inputSchema.properties["userContext"]?.maxLength; + expect(preservedContextMaxLength).toBeUndefined(); const selection = { targetPath: repository, @@ -561,11 +561,150 @@ describe("compact diff scan", () => { ]; const coverageNote = "The handler does not grant access to another caller's state (src/handler.py:1)."; + const finding = { + ruleId: "path-traversal.archive-extraction", + title: "Unsafe archive extraction", + summary: "An untrusted archive entry reaches a filesystem write.", + severity: { level: "high" }, + confidence: { + level: "high", + rationale: "Source evidence establishes reachability.", + }, + taxonomy: { category: "path-traversal", cwe: ["CWE-22"] }, + locations: [{ path: "src/handler.py", startLine: 1 }], + remediation: "Validate each output path before writing.", + provenance: { source: "local_plugin" }, + }; + const invalidRootCauseReference = await client.request("tools/call", { + name: "record_codex_security_scan_draft", + arguments: { + scanId, + handoffClaimToken, + findings: [ + { + ...finding, + root_cause: { + evidenceRefs: ["missing-root-cause-evidence"], + }, + }, + ], + coverage: { + completeness: "complete", + surfaces: [{ label: "Changed files", disposition: "rejected" }], + explicitExclusions: [], + deferred: [], + }, + }, + _meta: { "openai/threadId": owner }, + }); + expect(invalidRootCauseReference["isError"]).toBe(true); + expect(JSON.stringify(invalidRootCauseReference)).toContain( + "root_cause.evidenceRefs", + ); + await call("record_codex_security_scan_draft", { + scanId, + handoffClaimToken, + findings: [ + { + ...finding, + identity: { + anchor: "candidate-duplicate-instance", + instance: "dss-147-a", + }, + }, + { + ...finding, + extensions: { + candidateId: "candidate-duplicate-instance", + reportId: "DSS-147-A", + }, + }, + ], + coverage: { + completeness: "complete", + surfaces: [{ label: "Changed files", disposition: "rejected" }], + explicitExclusions: [], + deferred: [], + }, + }); + expect( + ( + JSON.parse(readFileSync(join(scanDir, "findings.json"), "utf8")) as { + findings: JsonObject[]; + } + ).findings.map((draftFinding) => draftFinding["identity"]), + ).toEqual([ + { anchor: "candidate-duplicate-instance", instance: "dss-147-a" }, + { anchor: "candidate-duplicate-instance", instance: "dss-147-a" }, + ]); await call("record_codex_security_scan_draft", { scanId, handoffClaimToken, + findings: [ + { + ...finding, + extensions: { + candidateId: "candidate-singleton", + reportId: "DSS-144-A", + }, + }, + { + ...finding, + code_evidence: [ + { + code: "value = 2", + id: "legacy-source", + }, + ], + attackPath: { + dataflow: { evidence_refs: ["legacy-source"] }, + }, + }, + { + ...finding, + ruleId: "path-traversal.archive-upload", + identity: { + anchor: "candidate-cross-rule", + instance: "shared-report", + }, + }, + { + ...finding, + extensions: { + candidateId: "candidate-cross-rule", + reportId: "shared-report", + }, + }, + { + ...finding, + extensions: { + candidateId: "candidate-cross-rule", + reportId: "second-report", + }, + }, + { + ...finding, + identity: { + anchor: "candidate-authored-instance", + instance: "dss-147-a", + }, + }, + { + ...finding, + extensions: { + candidateId: "candidate-authored-instance", + reportId: "DSS-147-B", + }, + }, + { + ...finding, + extensions: { + candidateId: "candidate-authored-instance", + ledgerRowId: "ledger-row-c", + }, + }, + ], threatModel, - findings: [], coverage: { completeness: "complete", surfaces: [ @@ -605,6 +744,32 @@ describe("compact diff scan", () => { expect((completed["coverage"] as JsonObject)["inventoryStrategy"]).toBe( "diff", ); + expect( + ((completed["findings"] as JsonObject)["findings"] as JsonObject[]).map( + (completedFinding) => completedFinding["identity"], + ), + ).toEqual([ + { anchor: "candidate-singleton", instance: "dss-144-a" }, + { anchor: "unsafe-archive-extraction" }, + { anchor: "candidate-cross-rule", instance: "shared-report" }, + { anchor: "candidate-cross-rule", instance: "shared-report" }, + { anchor: "candidate-cross-rule", instance: "second-report" }, + { anchor: "candidate-authored-instance", instance: "dss-147-a" }, + { + anchor: "candidate-authored-instance", + instance: "dss-147-b", + }, + { anchor: "candidate-authored-instance", instance: "ledger-row-c" }, + ]); + const legacyFinding = ( + (completed["findings"] as JsonObject)["findings"] as JsonObject[] + )[1]; + expect(legacyFinding?.["code_evidence"]).toEqual([ + { code: "value = 2", id: "legacy-source" }, + ]); + expect(legacyFinding?.["attackPath"]).toEqual({ + dataflow: { evidence_refs: ["legacy-source"] }, + }); expect( ((completed["manifest"] as JsonObject)["scan"] as JsonObject)[ "threatModel" @@ -613,7 +778,6 @@ describe("compact diff scan", () => { expect((completed["coverage"] as JsonObject)["openQuestions"]).toEqual( openQuestions, ); - expect((completed["findings"] as JsonObject)["findings"]).toEqual([]); const contract = await loadContract(scanDir, { pluginRoot: PLUGIN_ROOT }); expect(contract.manifest.scan.threatModel).toEqual(threatModel); expect(contract.coverage.openQuestions).toEqual(openQuestions); diff --git a/sdk/typescript/tests-ts/contract.test.ts b/sdk/typescript/tests-ts/contract.test.ts index 4fdfcd05f..4d817a541 100644 --- a/sdk/typescript/tests-ts/contract.test.ts +++ b/sdk/typescript/tests-ts/contract.test.ts @@ -221,6 +221,137 @@ describe("canonical scan contract", () => { expect(contract.findings.scanId).toBe(contract.manifest.scan.id); }); + test("preserves schema-valid sealed finding details", async () => { + const scanDir = await copyExample(); + const findingsPath = join(scanDir, "findings.json"); + const findings = await readJson(findingsPath); + const validation = { + evidence: "single proof", + counterEvidence: [], + status: null, + disposition: null, + result: null, + }; + findings["findings"][0]["validation"] = validation; + findings["findings"][0]["code_evidence"] = null; + await writeJson(findingsPath, findings); + await reseal(scanDir); + + const loaded = await loadContract(scanDir, { pluginRoot: PLUGIN_ROOT }); + + expect(loaded.findings.findings[0]?.validation).toEqual(validation); + expect(loaded.findings.findings[0]?.code_evidence).toBeNull(); + expect(await readJson(findingsPath)).toEqual(findings); + }); + + test("preserves valid details while normalizing malformed legacy siblings", async () => { + const scanDir = await copyExample(); + const findingsPath = join(scanDir, "findings.json"); + const findings = await readJson(findingsPath); + const validation = { + evidence: "single proof", + counterEvidence: [], + status: null, + disposition: null, + result: null, + }; + findings["findings"][0]["validation"] = validation; + const codeEvidence = [{ id: " ", code: " " }]; + findings["findings"][0]["code_evidence"] = codeEvidence; + findings["findings"][0]["attackPath"] = { + steps: { first: "upload" }, + }; + await writeJson(findingsPath, findings); + await reseal(scanDir); + + const loaded = await loadContract(scanDir, { pluginRoot: PLUGIN_ROOT }); + + expect(loaded.findings.findings[0]?.validation).toEqual(validation); + expect(loaded.findings.findings[0]?.code_evidence).toEqual(codeEvidence); + expect(loaded.findings.findings[0]?.attackPath?.steps).toBeUndefined(); + expect(await readJson(findingsPath)).toEqual(findings); + }); + + test("preserves empty union lists while normalizing malformed siblings", async () => { + const scanDir = await copyExample(); + const findingsPath = join(scanDir, "findings.json"); + const findings = await readJson(findingsPath); + findings["findings"][0]["validation"] = { evidence: [] }; + findings["findings"][0]["attackPath"] = { + steps: { first: "upload" }, + }; + await writeJson(findingsPath, findings); + await reseal(scanDir); + + const loaded = await loadContract(scanDir, { pluginRoot: PLUGIN_ROOT }); + + expect(loaded.findings.findings[0]?.validation?.evidence).toEqual([]); + expect(loaded.findings.findings[0]?.attackPath?.steps).toBeUndefined(); + expect(await readJson(findingsPath)).toEqual(findings); + }); + + test("normalizes pre-typed sealed finding details without changing the artifact", async () => { + const scanDir = await copyExample(); + const findingsPath = join(scanDir, "findings.json"); + const findings = await readJson(findingsPath); + const legacyValidation = { + evidence: { kind: "trace" }, + counterEvidence: [null, "The mitigation was checked."], + }; + const legacyAttackPath = { steps: { first: "upload" } }; + findings["findings"][0]["validation"] = legacyValidation; + findings["findings"][0]["attackPath"] = legacyAttackPath; + findings["findings"][0]["root_cause"] = null; + findings["findings"][0]["code_evidence"] = [ + { id: "legacy-source", code: "legacy_source()" }, + ]; + await writeJson(findingsPath, findings); + await reseal(scanDir); + + const loaded = await loadContract(scanDir, { pluginRoot: PLUGIN_ROOT }); + expect(loaded.findings.findings[0]?.code_evidence?.[0]?.id).toBe( + "legacy-source", + ); + const loadedFinding = loaded.findings.findings[0]; + expect(loadedFinding?.validation?.evidence).toBeUndefined(); + expect( + loadedFinding?.validation?.counterEvidence?.map((item) => + item.toUpperCase(), + ), + ).toEqual(["THE MITIGATION WAS CHECKED."]); + expect(loadedFinding?.attackPath?.steps).toBeUndefined(); + expect(loadedFinding?.root_cause).toBeNull(); + expect(await readJson(findingsPath)).toEqual(findings); + }); + + test("rejects malformed canonical root-cause details", async () => { + const scanDir = await copyExample(); + const findingsPath = join(scanDir, "findings.json"); + const findings = await readJson(findingsPath); + findings["findings"][0]["rootCause"] = { summary: [] }; + await writeJson(findingsPath, findings); + await reseal(scanDir); + + await expect( + loadContract(scanDir, { pluginRoot: PLUGIN_ROOT }), + ).rejects.toThrow("findings.json"); + expect(await readJson(findingsPath)).toEqual(findings); + }); + + test("loads an empty legacy root cause without changing the artifact", async () => { + const scanDir = await copyExample(); + const findingsPath = join(scanDir, "findings.json"); + const findings = await readJson(findingsPath); + findings["findings"][0]["root_cause"] = ""; + await writeJson(findingsPath, findings); + await reseal(scanDir); + + const loaded = await loadContract(scanDir, { pluginRoot: PLUGIN_ROOT }); + + expect(loaded.findings.findings[0]?.root_cause).toBe(""); + expect(await readJson(findingsPath)).toEqual(findings); + }); + test("honors cancellation during contract validation", async () => { const scanDir = await copyExample(); const controller = new AbortController(); diff --git a/sdk/typescript/tests-ts/finding-preview.test.ts b/sdk/typescript/tests-ts/finding-preview.test.ts index 974ab037d..161c366bb 100644 --- a/sdk/typescript/tests-ts/finding-preview.test.ts +++ b/sdk/typescript/tests-ts/finding-preview.test.ts @@ -2,15 +2,39 @@ import { join } from "node:path"; import { describe, expect, test } from "bun:test"; import { PLUGIN_ROOT } from "./plugin-root.js"; +function projectFindingDetails(original: Record) { + const python = Bun.which("python3") ?? Bun.which("python") ?? Bun.which("py"); + expect(python).not.toBeNull(); + + const program = [ + "import json, sys", + "sys.path.insert(0, sys.argv[1])", + "from finding_preview import bounded_finding_details", + "original = json.loads(sys.stdin.read())", + "projected = {name: bounded_finding_details(details) for name, details in original.items()}", + "print(json.dumps({'projected': projected, 'original': original}))", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", program, join(PLUGIN_ROOT, "scripts")], + { + stdin: new TextEncoder().encode(JSON.stringify(original)), + stdout: "pipe", + stderr: "pipe", + }, + ); + + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + return JSON.parse(new TextDecoder().decode(result.stdout)); +} + describe("bundled finding previews", () => { test("normalizes attack-path assessments without changing stored finding details", () => { - const python = - Bun.which("python3") ?? Bun.which("python") ?? Bun.which("py"); - expect(python).not.toBeNull(); - const original = { scalar: { - attackPath: { impact: "high", likelihood: "medium" }, + attackPath: { + impact: "Native memory corruption is possible.", + likelihood: "medium", + }, }, structured: { attackPath: { @@ -24,42 +48,151 @@ describe("bundled finding previews", () => { absentAttackPath: { rootCause: { summary: "Synthetic root cause." }, }, + bothEvidenceAliases: { + codeEvidence: [{ id: "canonical", code: "canonical_source()" }], + code_evidence: [{ id: "legacy", code: "legacy_source()" }], + rootCause: { summary: "Synthetic root cause." }, + }, + bothRootCauseAliases: { + rootCause: { code: "SELECT * FROM users" }, + root_cause: { + code: "os.system(user_input)", + evidence_refs: ["legacy-source"], + language: "python", + summary: "The destination is not contained.", + }, + }, + invalidLegacyEvidenceFields: { + code_evidence: [ + { + id: "legacy-source", + code: "dangerous_call()", + startLine: 0, + endLine: "12", + label: 7, + role: { kind: "sink" }, + }, + ], + }, + malformedCanonicalRootCause: { + rootCause: { summary: 42 }, + root_cause: { + summary: "The valid legacy root cause.", + evidence_refs: ["legacy-root"], + }, + code_evidence: [{ id: "legacy-root", code: "legacy_root()" }], + }, }; - const program = [ - "import json, sys", - "sys.path.insert(0, sys.argv[1])", - "from finding_preview import bounded_finding_details", - "original = json.loads(sys.argv[2])", - "projected = {name: bounded_finding_details(details) for name, details in original.items()}", - "print(json.dumps({'projected': projected, 'original': original}))", - ].join("\n"); - const result = Bun.spawnSync( - [ - python!, - "-I", - "-B", - "-c", - program, - join(PLUGIN_ROOT, "scripts"), - JSON.stringify(original), - ], - { stdout: "pipe", stderr: "pipe" }, - ); - - expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); - expect(JSON.parse(new TextDecoder().decode(result.stdout))).toEqual({ + expect(projectFindingDetails(original)).toEqual({ projected: { scalar: { attackPath: { - impact: { level: "high" }, + impact: { rationale: "Native memory corruption is possible." }, likelihood: { level: "medium" }, }, }, structured: original.structured, absentAssessments: original.absentAssessments, absentAttackPath: original.absentAttackPath, + bothEvidenceAliases: { + codeEvidence: [ + { id: "canonical", code: "canonical_source()" }, + { id: "legacy", code: "legacy_source()" }, + ], + rootCause: { summary: "Synthetic root cause." }, + }, + bothRootCauseAliases: { + rootCause: { + code: "SELECT * FROM users", + evidenceRefs: ["legacy-source"], + summary: "The destination is not contained.", + }, + }, + invalidLegacyEvidenceFields: { + code_evidence: [{ id: "legacy-source", code: "dangerous_call()" }], + }, + malformedCanonicalRootCause: { + rootCause: { + summary: "The valid legacy root cause.", + evidenceRefs: ["legacy-root"], + }, + code_evidence: [{ id: "legacy-root", code: "legacy_root()" }], + }, }, original, }); }); + + test("preserves counter-evidence under the validation preview budget", () => { + const original = { + finding: { + validation: { + evidence: ["x".repeat(20_000)], + summary: `The traversal was reproduced. ${"x".repeat(20_000)}`, + method: "focused extraction test", + evidenceRefs: ["evidence-0"], + futureMetadata: "x".repeat(20_000), + counterEvidence: ["Known mitigations remain unverified."], + }, + }, + }; + + const result = projectFindingDetails(original); + + expect(result.projected.finding.validation.counterEvidence).toEqual([ + "Known mitigations remain unverified.", + ]); + expect(result.original).toEqual(original); + }); + + test("deduplicates evidence before applying the preview limit", () => { + const original = { + finding: { + codeEvidence: [ + { id: "shared", code: "canonical_shared()" }, + { id: "shared", code: "duplicate_shared()" }, + { id: "canonical-two", code: "canonical_two()" }, + { id: "canonical-three", code: "canonical_three()" }, + ], + code_evidence: [ + { id: "shared", code: "legacy_shared()" }, + { id: "legacy-four", code: "legacy_four()" }, + { id: "legacy-five", code: "legacy_five()" }, + ], + }, + }; + + const result = projectFindingDetails(original); + + expect( + result.projected.finding.codeEvidence.map( + (item: { id: string }) => item.id, + ), + ).toEqual(["shared", "canonical-two", "canonical-three", "legacy-four"]); + expect(result.projected.finding.codeEvidence[0].code).toBe( + "canonical_shared()", + ); + expect(result.original).toEqual(original); + }); + + test("filters malformed evidence before applying the preview limit", () => { + const original = { + finding: { + code_evidence: [ + null, + "junk", + {}, + { id: "empty", code: "" }, + { id: "valid", code: "valid_source()" }, + ], + }, + }; + + const result = projectFindingDetails(original); + + expect(result.projected.finding.code_evidence).toEqual([ + { id: "valid", code: "valid_source()" }, + ]); + expect(result.original).toEqual(original); + }); }); diff --git a/sdk/typescript/tests-ts/plugin-finding-detail-contract.test.ts b/sdk/typescript/tests-ts/plugin-finding-detail-contract.test.ts new file mode 100644 index 000000000..063a82086 --- /dev/null +++ b/sdk/typescript/tests-ts/plugin-finding-detail-contract.test.ts @@ -0,0 +1,1078 @@ +import { spawn } from "node:child_process"; +import { readFile } from "node:fs/promises"; +import { join } from "node:path"; +import { createInterface } from "node:readline"; +import { describe, expect, test } from "bun:test"; +import Ajv from "ajv"; +import Ajv2020 from "ajv/dist/2020.js"; +import { PLUGIN_ROOT } from "./plugin-root.js"; + +type JsonObject = Record; + +const invalidFindingDetails: Array<{ + section: "attackPath" | "rootCause" | "root_cause" | "validation"; + detail: JsonObject; +}> = [ + { + section: "rootCause", + detail: { summary: "Root cause.", code: ["not a string"] }, + }, + { + section: "rootCause", + detail: { summary: "Root cause.", language: 42 }, + }, + { section: "root_cause", detail: { summary: ["not a string"] } }, + { section: "root_cause", detail: { code: ["not a string"] } }, + { section: "root_cause", detail: { language: 42 } }, + { section: "attackPath", detail: { steps: "upload, then extract" } }, + { section: "attackPath", detail: { preconditions: "upload access" } }, + { + section: "attackPath", + detail: { reachability: { attacker: {} } }, + }, + { + section: "attackPath", + detail: { reachability: { entrypoint: [] } }, + }, + { + section: "attackPath", + detail: { reachability: { preconditions: "upload access" } }, + }, + { + section: "attackPath", + detail: { dataflow: { transformations: "decode, then dispatch" } }, + }, + { section: "validation", detail: { assertions: "sink reached" } }, + { section: "validation", detail: { counterEvidence: "none" } }, + { section: "validation", detail: { evidence: { kind: "trace" } } }, +]; + +const scanDraftFinding = { + ruleId: "path-traversal.archive-extraction", + title: "Unsafe archive extraction", + summary: "An untrusted archive entry reaches a filesystem write.", + severity: { level: "high" }, + confidence: { + level: "high", + rationale: "Source evidence establishes reachability.", + }, + taxonomy: { category: "path-traversal", cwe: ["CWE-22"] }, + locations: [{ path: "src/extract.py", startLine: 41 }], + remediation: "Validate each output path before writing.", + provenance: { source: "local_plugin" }, +}; + +const scanDraftInput = { + scanId: "7b95abf2-dc04-47a9-9950-53b5c2057f49", + findings: [scanDraftFinding], + coverage: { + completeness: "complete", + surfaces: [], + explicitExclusions: [], + deferred: [], + }, +}; + +const stringAssessmentInput = { + ...scanDraftInput, + findings: [ + { + ...scanDraftFinding, + attackPath: { + impact: "high", + likelihood: "medium", + reachability: { + summary: "A repository contributor can trigger archive extraction.", + attacker: "repository contributor", + entrypoint: "archive extraction", + outcome: "a file is written outside the extraction root", + preconditions: ["The service processes the uploaded archive."], + }, + }, + }, + ], +}; + +function projectFindingDetails(details: JsonObject): string { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "examples = plugin / 'examples' / 'completed-scan'", + "manifest, findings, coverage = [json.loads((examples / name).read_text()) for name in ('scan-manifest.json', 'findings.json', 'coverage.json')]", + "findings['findings'][0].update(json.loads(sys.argv[2]))", + "projection = runpy.run_path(str(plugin / 'scripts' / 'report_projection.py'))", + "print(projection['build_report_markdown'](manifest, findings, coverage))", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT, JSON.stringify(details)], + { stdout: "pipe", stderr: "pipe" }, + ); + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + return new TextDecoder().decode(result.stdout); +} + +async function readJson(path: string): Promise { + return JSON.parse(await readFile(path, "utf8")) as JsonObject; +} + +async function startMcp() { + const child = spawn( + process.execPath, + [join(PLUGIN_ROOT, "mcp", "server.mjs"), "--stdio"], + { stdio: ["pipe", "pipe", "pipe"] }, + ); + const messages = createInterface({ input: child.stdout })[ + Symbol.asyncIterator + ](); + let stderr = ""; + child.stderr.setEncoding("utf8"); + child.stderr.on("data", (chunk: string) => { + stderr += chunk; + }); + let nextId = 0; + + async function request( + method: string, + params: JsonObject, + ): Promise { + const id = ++nextId; + child.stdin.write( + `${JSON.stringify({ jsonrpc: "2.0", id, method, params })}\n`, + ); + while (true) { + const message = await messages.next(); + if (message.done) { + throw new Error(`MCP server exited before replying: ${stderr}`); + } + const response = JSON.parse(message.value) as JsonObject; + if (response["id"] !== id) continue; + if (response["error"] !== undefined) { + throw new Error(JSON.stringify(response["error"])); + } + return response["result"] as JsonObject; + } + } + + await request("initialize", { + protocolVersion: "2025-11-25", + capabilities: {}, + clientInfo: { name: "finding-detail-contract-test", version: "1.0.0" }, + }); + child.stdin.write( + `${JSON.stringify({ + jsonrpc: "2.0", + method: "notifications/initialized", + params: {}, + })}\n`, + ); + + return { + request, + async close(): Promise { + child.stdin.end(); + await new Promise((resolve) => { + child.once("close", () => resolve()); + }); + }, + }; +} + +describe("bundled plugin finding detail contracts", () => { + test("rejects malformed known fields in scan drafts", async () => { + const schemaRoot = join(PLUGIN_ROOT, "schemas"); + const [commonSchema, scanDraftSchema] = await Promise.all([ + readJson(join(schemaRoot, "definitions", "artifact-common.schema.json")), + readJson(join(schemaRoot, "tools", "scan-draft.schema.json")), + ]); + const validator = new Ajv2020({ strict: false }); + validator.addFormat("uuid", /^[0-9a-f-]{36}$/iu); + validator.addSchema(commonSchema); + const validate = validator.compile(scanDraftSchema); + + expect(validate(scanDraftInput), JSON.stringify(validate.errors)).toBe( + true, + ); + expect( + validate(stringAssessmentInput), + JSON.stringify(validate.errors), + ).toBe(true); + expect( + validate({ + ...scanDraftInput, + findings: [{ ...scanDraftFinding, code_evidence: null }], + }), + JSON.stringify(validate.errors), + ).toBe(false); + for (const { section, detail } of invalidFindingDetails) { + expect( + validate({ + ...scanDraftInput, + findings: [{ ...scanDraftFinding, [section]: detail }], + }), + `${section}: ${JSON.stringify(detail)}`, + ).toBe(false); + } + }); + + test("publishes the strict scan-draft contract through MCP", async () => { + const client = await startMcp(); + try { + const result = await client.request("tools/list", {}); + const tools = result["tools"] as Array; + const tool = tools.find( + (candidate) => candidate["name"] === "record_codex_security_scan_draft", + ); + expect(tool).toBeDefined(); + + const validator = new Ajv({ strict: false }); + validator.addFormat("uuid", /^[0-9a-f-]{36}$/iu); + const validate = validator.compile(tool!["inputSchema"] as JsonObject); + + expect(validate(scanDraftInput), JSON.stringify(validate.errors)).toBe( + true, + ); + expect( + validate(stringAssessmentInput), + JSON.stringify(validate.errors), + ).toBe(true); + expect( + validate({ + ...scanDraftInput, + findings: [{ ...scanDraftFinding, code_evidence: null }], + }), + JSON.stringify(validate.errors), + ).toBe(false); + for (const { section, detail } of invalidFindingDetails) { + expect( + validate({ + ...scanDraftInput, + findings: [{ ...scanDraftFinding, [section]: detail }], + }), + `${section}: ${JSON.stringify(detail)}`, + ).toBe(false); + } + } finally { + await client.close(); + } + }); + + test("rejects malformed known fields in canonical findings", async () => { + const [schema, example] = await Promise.all([ + readJson(join(PLUGIN_ROOT, "schemas", "findings.schema.json")), + readJson( + join(PLUGIN_ROOT, "examples", "completed-scan", "findings.json"), + ), + ]); + const validate = new Ajv2020({ strict: false }).compile(schema); + expect(validate(example), JSON.stringify(validate.errors)).toBe(true); + + const compatibleDocument = structuredClone(example) as { + findings: Array; + }; + compatibleDocument.findings[0]!["attackPath"] = + stringAssessmentInput.findings[0]!.attackPath; + expect(validate(compatibleDocument), JSON.stringify(validate.errors)).toBe( + true, + ); + + for (const { section, detail } of invalidFindingDetails) { + const document = structuredClone(example) as { + findings: Array; + }; + document.findings[0]![section] = detail; + expect(validate(document), `${section}: ${JSON.stringify(detail)}`).toBe( + false, + ); + } + }); + + test.each(["dataFlow", "dataflow", "data_flow"] as const)( + "projects scalar finding details for the %s alias", + (dataFlowKey) => { + const report = projectFindingDetails({ + attackPath: { + [dataFlowKey]: "request -> archive extraction -> filesystem write", + reachability: "An authenticated uploader can trigger extraction.", + }, + validation: { + assertions: ["The destination escapes the extraction root."], + evidence: "The archive entry path is not contained.", + limitations: ["The upload route was not exercised dynamically."], + }, + }); + + expect(report).toContain( + "request -\\> archive extraction -\\> filesystem write", + ); + expect(report).toContain( + "An authenticated uploader can trigger extraction.", + ); + expect(report).toContain("The destination escapes the extraction root."); + expect(report).toContain("The archive entry path is not contained."); + expect(report).toContain( + "The upload route was not exercised dynamically.", + ); + }, + ); + + test("projects the populated data-flow alias", () => { + const report = projectFindingDetails({ + attackPath: { + dataFlow: { producerExtension: true }, + dataflow: { + summary: + "request -> populated lowercase dataflow -> filesystem write", + }, + }, + }); + + expect(report).toContain( + "request -\\> populated lowercase dataflow -\\> filesystem write", + ); + }); + + test("ignores malformed data-flow alias values", () => { + const report = projectFindingDetails({ + attackPath: { + dataFlow: { summary: ["invalid canonical value"] }, + dataflow: { summary: "request -> validated sink" }, + }, + }); + + expect(report).toContain("request -\\> validated sink"); + }); + + test("merges transformations across data-flow aliases", () => { + const report = projectFindingDetails({ + attackPath: { + dataFlow: { + transformations: ["decode archive entry", "parse *input*"], + }, + dataflow: { + summary: "request -> archive extraction -> filesystem write", + transformations: ["dispatch extraction", "decode archive entry"], + }, + data_flow: { transformations: null }, + }, + }); + + expect(report).toContain("- decode archive entry"); + expect(report).toContain("- dispatch extraction"); + expect(report.match(/- decode archive entry/gu)).toHaveLength(1); + expect(report.split(/\r?\n/u)).toContain("- parse \\*input\\*"); + }); + + test("merges top-level and reachability preconditions", () => { + const report = projectFindingDetails({ + attackPath: { + preconditions: [ + "The service processes uploaded archives.", + "The attacker can upload an archive.", + ], + reachability: { + summary: "An authenticated uploader can trigger extraction.", + preconditions: [ + "The attacker can upload an archive.", + "Automatic extraction is enabled.", + ], + }, + }, + }); + + expect( + report.match(/- The service processes uploaded archives\./gu), + ).toHaveLength(1); + expect( + report.match(/- The attacker can upload an archive\./gu), + ).toHaveLength(1); + expect(report.match(/- Automatic extraction is enabled\./gu)).toHaveLength( + 1, + ); + }); + + test("projects the top-level attack-path summary as reachability", () => { + const report = projectFindingDetails({ + attackPath: { + summary: "An authenticated uploader can trigger archive extraction.", + }, + }); + + const reachability = report + .split("#### Reachability", 2)[1]! + .split("#### Severity", 1)[0]!; + expect(reachability).toContain( + "An authenticated uploader can trigger archive extraction.", + ); + expect(reachability).not.toContain("Reachability was not recorded"); + }); + + test("projects typed attack-path steps", () => { + const report = projectFindingDetails({ + attackPath: { + steps: [ + "Upload an archive with a traversal entry.", + "Trigger automatic extraction.", + ], + }, + }); + + const dataflow = report + .split("#### Dataflow", 2)[1]! + .split("#### Reachability", 1)[0]!; + expect(dataflow).toContain("Attack steps:"); + expect(dataflow).toContain("- Upload an archive with a traversal entry."); + expect(dataflow).toContain("- Trigger automatic extraction."); + }); + + test("projects typed assessments and validation outcomes", () => { + const report = projectFindingDetails({ + validation: { + status: "validated", + disposition: "reported", + result: "The traversal write was confirmed.", + }, + attackPath: { + impact: { + level: "high", + rationale: "The write can overwrite application files.", + why: "The destination escapes the extraction root.", + }, + likelihood: "Likely for authenticated uploaders.", + reachability: { + source: "Attacker-controlled archive entry.", + sink: "Unchecked filesystem write.", + }, + }, + }); + + expect(report).toContain("- **Status:** validated"); + expect(report).toContain("- **Disposition:** reported"); + expect(report).toContain( + "- **Result:** The traversal write was confirmed.", + ); + expect(report).toContain( + "- **Source:** Attacker-controlled archive entry.", + ); + expect(report).toContain("- **Sink:** Unchecked filesystem write."); + expect(report).toContain("Impact assessment:"); + expect(report).toContain("- **Level:** high"); + expect(report).toContain( + "- **Rationale:** The write can overwrite application files.", + ); + expect(report).toContain( + "- **Why:** The destination escapes the extraction root.", + ); + expect(report).toContain( + "**Likelihood assessment:** Likely for authenticated uploaders.", + ); + }); + + test("projects attack-path context lists", () => { + const report = projectFindingDetails({ + attackPath: { + assumptions: ["Automatic extraction is enabled."], + blindspots: ["A downstream sandbox was not exercised."], + controls: ["Archive uploads require authentication."], + limitations: ["The exploit was validated statically."], + }, + }); + + const reachability = report + .split("#### Reachability", 2)[1]! + .split("#### Severity", 1)[0]!; + expect(reachability).toContain("Assumptions:"); + expect(reachability).toContain("- Automatic extraction is enabled."); + expect(reachability).toContain("Existing controls:"); + expect(reachability).toContain("- Archive uploads require authentication."); + expect(reachability).toContain("Blind spots:"); + expect(reachability).toContain("- A downstream sandbox was not exercised."); + expect(reachability).toContain("Limitations:"); + expect(reachability).toContain("- The exploit was validated statically."); + }); + + test("merges root-cause aliases in generated reports", () => { + const report = projectFindingDetails({ + rootCause: { summary: "" }, + root_cause: { + summary: "The destination is not contained before the write.", + code: "destination.write_bytes(payload)", + language: "python", + }, + }).replaceAll("\r\n", "\n"); + + expect(report).toContain( + "The destination is not contained before the write.", + ); + expect(report).toContain( + "```python\ndestination.write_bytes(payload)\n```", + ); + }); + + test("merges scalar and list root-cause evidence references", () => { + const report = projectFindingDetails({ + rootCause: { + evidenceRefs: ["canonical-root-source"], + summary: "The source reaches the write.", + }, + root_cause: { evidence_refs: "legacy-root-source" }, + codeEvidence: [ + { id: "canonical-root-source", code: "canonical_source()" }, + ], + code_evidence: [{ id: "legacy-root-source", code: "legacy_source()" }], + }); + + expect(report).toContain("canonical_source()"); + expect(report).toContain("legacy_source()"); + }); + + test("merges embedded evidence across root-cause aliases", () => { + const report = projectFindingDetails({ + rootCause: { + summary: "The canonical root cause.", + codeEvidence: [{ id: "canonical-root", code: "canonical_evidence()" }], + }, + root_cause: { + codeEvidence: [{ id: "legacy-root", code: "legacy_evidence()" }], + }, + }); + + expect(report).toContain("canonical_evidence()"); + expect(report).toContain("legacy_evidence()"); + }); + + test("treats whitespace root-cause fields as unpopulated", () => { + const report = projectFindingDetails({ + rootCause: { summary: " ", code: "\t", language: "text" }, + root_cause: { + summary: "The destination is not contained before the write.", + code: "destination.write_bytes(payload)", + language: "python", + }, + }).replaceAll("\r\n", "\n"); + + expect(report).toContain( + "The destination is not contained before the write.", + ); + expect(report).toContain( + "```python\ndestination.write_bytes(payload)\n```", + ); + }); + + test("projects code evidence referenced by nested attack-path details", () => { + const report = projectFindingDetails({ + attackPath: { + dataflow: { + evidenceRefs: [], + evidence_refs: ["archive-source"], + summary: "An archive entry path reaches a filesystem write.", + }, + reachability: { + evidenceRefs: ["archive-sink"], + summary: "An authenticated uploader can trigger extraction.", + }, + }, + codeEvidence: [ + { + code: "entry_path = archive_entry.name", + explanation: "The archive controls the path.", + id: "archive-source", + label: "Attacker-controlled archive path", + path: "src/archive.py", + startLine: 20, + }, + { + code: "destination.write_bytes(entry.read())", + explanation: "The unchecked path reaches the write.", + id: "archive-sink", + label: "Unchecked filesystem write", + path: "src/archive.py", + startLine: 41, + }, + ], + }); + + expect(report).toContain("entry_path = archive_entry.name"); + expect(report).toContain("destination.write_bytes(entry.read())"); + }); + + test("normalizes scalar code-evidence references in generated reports", () => { + const report = projectFindingDetails({ + root_cause: { + summary: "The authorization check occurs after the write.", + evidence_refs: "root-source", + }, + attackPath: { + dataflow: { + summary: "An attacker-controlled value reaches the write.", + evidence_refs: "dataflow-source", + }, + reachability: { + summary: "An authenticated caller can reach the handler.", + evidence_refs: "reachability-source", + }, + }, + codeEvidence: [ + { id: "root-source", code: "root_source()" }, + { id: "dataflow-source", code: "dataflow_source()" }, + { id: "reachability-source", code: "reachability_source()" }, + ], + }); + + expect(report).toContain("root_source()"); + expect(report).toContain("dataflow_source()"); + expect(report).toContain("reachability_source()"); + }); + + test("rejects unknown code evidence referenced by nested attack-path details", () => { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "finding = json.loads((plugin / 'examples' / 'completed-scan' / 'findings.json').read_text())['findings'][0]", + "finding['attackPath'] = {'dataflow': {'evidenceRefs': ['missing-evidence']}}", + "finalizer = runpy.run_path(str(plugin / 'scripts' / 'finalize_scan_contract.py'))", + "try:", + " finalizer['_validate_finding'](finding, 'findings[0]')", + "except finalizer['ContractError'] as error:", + " print(error)", + "else:", + " print('accepted')", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT], + { stdout: "pipe", stderr: "pipe" }, + ); + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + expect(new TextDecoder().decode(result.stdout)).toContain( + "attackPath.dataflow.evidenceRefs: unknown code-evidence ids: missing-evidence", + ); + }); + + test("accepts nested references to the legacy code evidence catalog", () => { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "finding = json.loads((plugin / 'examples' / 'completed-scan' / 'findings.json').read_text())['findings'][0]", + "finding.pop('codeEvidence', None)", + "finding['code_evidence'] = [{'id': 'legacy-source', 'code': 'entry_path = archive_entry.name'}]", + "finding['attackPath'] = {'dataflow': {'evidence_refs': ['legacy-source']}}", + "finalizer = runpy.run_path(str(plugin / 'scripts' / 'finalize_scan_contract.py'))", + "finalizer['_validate_finding'](finding, 'findings[0]')", + "print('accepted')", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT], + { stdout: "pipe", stderr: "pipe" }, + ); + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + expect(new TextDecoder().decode(result.stdout)).toContain("accepted"); + }); + + test("rejects duplicate IDs across code evidence catalogs", () => { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "finding = json.loads((plugin / 'examples' / 'completed-scan' / 'findings.json').read_text())['findings'][0]", + "finding['codeEvidence'] = [{'id': 'shared-source', 'code': 'canonical_source()'}]", + "finding['code_evidence'] = [{'id': 'shared-source', 'code': 'conflicting_legacy_source()'}]", + "finalizer = runpy.run_path(str(plugin / 'scripts' / 'finalize_scan_contract.py'))", + "try:", + " finalizer['_validate_finding'](finding, 'findings[0]')", + "except finalizer['ContractError'] as error:", + " print(error)", + "else:", + " print('accepted')", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT], + { stdout: "pipe", stderr: "pipe" }, + ); + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + expect(new TextDecoder().decode(result.stdout)).toContain( + "code_evidence[0].id: duplicate code-evidence id", + ); + }); + + test("keeps legacy sealed evidence references compatible", () => { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "findings = json.loads((plugin / 'examples' / 'completed-scan' / 'findings.json').read_text())", + "findings['findings'][0]['codeEvidence'] = [{'id': 'shared-source', 'label': 'Canonical source', 'path': 'src/extract.py', 'startLine': 41, 'code': 'canonical_source()', 'explanation': 'Canonical snippet.'}]", + "findings['findings'][0]['code_evidence'] = [", + " {'id': 'shared-source', 'code': 'legacy_source()'},", + " {'id': 'legacy-duplicate', 'code': 'first_legacy_source()'},", + " {'id': 'legacy-duplicate', 'code': 'second_legacy_source()'},", + "]", + "findings['findings'][0]['validation'] = {'evidence_refs': ['legacy-validation-evidence'], 'counterEvidence': None, 'limitations': '', 'method': [], 'status': '', 'summary': {}, 'disposition': '', 'result': ''}", + "findings['findings'][0]['rootCause'] = {'summary': 'Canonical root cause.', 'code': 'canonical_root()', 'language': 'python'}", + "findings['findings'][0]['attackPath'] = {'evidence_refs': ['legacy-attack-evidence'], 'dataFlow': '', 'dataflow': {'summary': '', 'source': None, 'sink': None, 'outcome': None, 'evidenceRefs': ['legacy-missing-evidence']}, 'impact': {'level': '', 'rationale': None, 'why': None}, 'likelihood': {'level': None, 'rationale': None, 'why': None}, 'reachability': None, 'summary': ''}", + "finalizer = runpy.run_path(str(plugin / 'scripts' / 'finalize_scan_contract.py'))", + "compatible = finalizer['_legacy_sealed_findings_for_validation'](findings)", + "finalizer['_validate_finding'](compatible['findings'][0], 'findings[0]')", + "finalizer['validate_against_schema'](compatible, plugin / 'schemas' / 'findings.schema.json')", + "print(json.dumps({'originalDataFlow': findings['findings'][0]['attackPath']['dataFlow'], 'compatibleHasDataFlow': 'dataFlow' in compatible['findings'][0]['attackPath'], 'originalNested': findings['findings'][0]['attackPath']['dataflow']['evidenceRefs'], 'compatibleNested': compatible['findings'][0]['attackPath']['dataflow']['evidenceRefs'], 'originalReachability': findings['findings'][0]['attackPath']['reachability'], 'compatibleHasReachability': 'reachability' in compatible['findings'][0]['attackPath'], 'originalAttack': findings['findings'][0]['attackPath']['evidence_refs'], 'compatibleAttack': compatible['findings'][0]['attackPath']['evidence_refs'], 'originalValidation': findings['findings'][0]['validation']['evidence_refs'], 'compatibleValidation': compatible['findings'][0]['validation']['evidence_refs'], 'compatibleHasCounterEvidence': 'counterEvidence' in compatible['findings'][0]['validation'], 'compatibleHasDisposition': 'disposition' in compatible['findings'][0]['validation'], 'compatibleHasLimitations': 'limitations' in compatible['findings'][0]['validation'], 'compatibleHasMethod': 'method' in compatible['findings'][0]['validation'], 'compatibleHasResult': 'result' in compatible['findings'][0]['validation'], 'compatibleHasRootCause': 'rootCause' in compatible['findings'][0], 'compatibleHasStatus': 'status' in compatible['findings'][0]['validation'], 'compatibleHasSummary': 'summary' in compatible['findings'][0]['validation'], 'originalMethod': findings['findings'][0]['validation']['method'], 'originalRootCauseSummary': findings['findings'][0]['rootCause']['summary'], 'originalSummary': findings['findings'][0]['validation']['summary'], 'originalLegacyCatalog': findings['findings'][0]['code_evidence'], 'compatibleLegacyCatalog': compatible['findings'][0]['code_evidence']}))", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT], + { stdout: "pipe", stderr: "pipe" }, + ); + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + expect(JSON.parse(new TextDecoder().decode(result.stdout))).toEqual({ + compatibleAttack: [], + compatibleLegacyCatalog: [ + { code: "first_legacy_source()", id: "legacy-duplicate" }, + ], + compatibleHasDataFlow: false, + compatibleHasReachability: false, + compatibleNested: [], + compatibleValidation: [], + compatibleHasCounterEvidence: false, + compatibleHasDisposition: false, + compatibleHasLimitations: false, + compatibleHasMethod: false, + compatibleHasResult: false, + compatibleHasRootCause: true, + compatibleHasStatus: false, + compatibleHasSummary: false, + originalAttack: ["legacy-attack-evidence"], + originalLegacyCatalog: [ + { code: "legacy_source()", id: "shared-source" }, + { code: "first_legacy_source()", id: "legacy-duplicate" }, + { code: "second_legacy_source()", id: "legacy-duplicate" }, + ], + originalMethod: [], + originalDataFlow: "", + originalNested: ["legacy-missing-evidence"], + originalReachability: null, + originalRootCauseSummary: "Canonical root cause.", + originalSummary: {}, + originalValidation: ["legacy-validation-evidence"], + }); + }); + + test("normalizes formerly free-form sealed details for validation", () => { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "findings = json.loads((plugin / 'examples' / 'completed-scan' / 'findings.json').read_text())", + "finding = findings['findings'][0]", + "finding['validation'] = {'evidence': {'kind': 'trace'}, 'counterEvidence': [None, 'The mitigation was checked.']}", + "finding['attackPath'] = {'steps': {'first': 'upload'}}", + "finalizer = runpy.run_path(str(plugin / 'scripts' / 'finalize_scan_contract.py'))", + "compatible = finalizer['_legacy_sealed_findings_for_validation'](findings)", + "finalizer['_validate_finding'](compatible['findings'][0], 'findings[0]')", + "finalizer['validate_against_schema'](compatible, plugin / 'schemas' / 'findings.schema.json')", + "print(json.dumps({'originalValidation': finding['validation'], 'originalAttackPath': finding['attackPath'], 'compatibleValidation': compatible['findings'][0]['validation'], 'compatibleAttackPath': compatible['findings'][0]['attackPath']}))", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT], + { stdout: "pipe", stderr: "pipe" }, + ); + + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + expect(JSON.parse(new TextDecoder().decode(result.stdout))).toEqual({ + originalValidation: { + evidence: { kind: "trace" }, + counterEvidence: [null, "The mitigation was checked."], + }, + originalAttackPath: { steps: { first: "upload" } }, + compatibleValidation: { + counterEvidence: ["The mitigation was checked."], + }, + compatibleAttackPath: {}, + }); + }); + + test("rejects malformed canonical root causes during sealed validation", () => { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "findings = json.loads((plugin / 'examples' / 'completed-scan' / 'findings.json').read_text())", + "findings['findings'][0]['rootCause'] = {'summary': []}", + "finalizer = runpy.run_path(str(plugin / 'scripts' / 'finalize_scan_contract.py'))", + "compatible = finalizer['_legacy_sealed_findings_for_validation'](findings)", + "try:", + " finalizer['validate_against_schema'](compatible, plugin / 'schemas' / 'findings.schema.json')", + "except finalizer['ContractError'] as error:", + " print(error)", + "else:", + " print('accepted')", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT], + { stdout: "pipe", stderr: "pipe" }, + ); + + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + expect(new TextDecoder().decode(result.stdout)).toContain( + "rootCause.summary", + ); + }); + + test("ranks legacy code evidence during interrupted recovery", () => { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import copy, json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "examples = plugin / 'examples' / 'completed-scan'", + "manifest = json.loads((examples / 'scan-manifest.json').read_text())", + "example = json.loads((examples / 'findings.json').read_text())['findings'][0]", + "first = copy.deepcopy(example)", + "first.pop('codeEvidence', None)", + "first['summary'] = 'FIRST'", + "second = copy.deepcopy(first)", + "second['summary'] = 'SECOND'", + "second['code_evidence'] = [{'id': 'legacy-source', 'code': 'legacy_source()'}]", + "second['root_cause'] = None", + "findings = {'scanId': manifest['scan']['id'], 'findings': [first, second]}", + "warnings = []", + "finalizer = runpy.run_path(str(plugin / 'scripts' / 'finalize_scan_contract.py'))", + "finalizer['_recover_unsealed_findings'](manifest, findings, plugin / 'schemas', examples, warnings)", + "print(json.dumps({'summary': findings['findings'][0]['summary'], 'evidence': findings['findings'][0].get('code_evidence'), 'rootCause': findings['findings'][0].get('root_cause'), 'warnings': warnings}))", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT], + { stdout: "pipe", stderr: "pipe" }, + ); + + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + expect(JSON.parse(new TextDecoder().decode(result.stdout))).toEqual({ + summary: "SECOND", + evidence: [{ id: "legacy-source", code: "legacy_source()" }], + rootCause: null, + warnings: [ + "Recovered finding 2: retained stronger duplicate logical finding.", + ], + }); + }); + + test("ranks embedded root-cause evidence during interrupted recovery", () => { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import copy, json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "examples = plugin / 'examples' / 'completed-scan'", + "manifest = json.loads((examples / 'scan-manifest.json').read_text())", + "example = json.loads((examples / 'findings.json').read_text())['findings'][0]", + "finalizer = runpy.run_path(str(plugin / 'scripts' / 'finalize_scan_contract.py'))", + "results = {}", + "for name, detail in {'embedded': {'codeEvidence': [{'id': 'embedded-root', 'code': 'embedded_root()'}]}, 'legacy-code': {'code': 'legacy_root()'}}.items():", + " first = copy.deepcopy(example)", + " first.pop('codeEvidence', None)", + " first['summary'] = 'FIRST'", + " second = copy.deepcopy(first)", + " second['summary'] = 'SECOND'", + " second['rootCause'] = {'summary': 'Richer root cause', **detail}", + " findings = {'scanId': manifest['scan']['id'], 'findings': [first, second]}", + " warnings = []", + " finalizer['_recover_unsealed_findings'](manifest, findings, plugin / 'schemas', examples, warnings)", + " results[name] = {'summary': findings['findings'][0]['summary'], 'warnings': warnings}", + "print(json.dumps(results))", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT], + { stdout: "pipe", stderr: "pipe" }, + ); + + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + expect(JSON.parse(new TextDecoder().decode(result.stdout))).toEqual({ + embedded: { + summary: "SECOND", + warnings: [ + "Recovered finding 2: retained stronger duplicate logical finding.", + ], + }, + "legacy-code": { + summary: "SECOND", + warnings: [ + "Recovered finding 2: retained stronger duplicate logical finding.", + ], + }, + }); + }); + + test("ignores malformed legacy evidence references in sealed validation", () => { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "findings = json.loads((plugin / 'examples' / 'completed-scan' / 'findings.json').read_text())", + "finding = findings['findings'][0]", + "finding['code_evidence'] = [{'id': 'legacy-source', 'code': 'legacy_source()'}]", + "finding['validation'] = {'evidenceRefs': [None, '', 42, 'legacy-source', 'missing-source']}", + "finding['attackPath'] = {'dataflow': {'evidence_refs': [None, '', 42, 'legacy-source', 'missing-source']}}", + "finalizer = runpy.run_path(str(plugin / 'scripts' / 'finalize_scan_contract.py'))", + "compatible = finalizer['_legacy_sealed_findings_for_validation'](findings)", + "finalizer['_validate_finding'](compatible['findings'][0], 'findings[0]')", + "print(json.dumps({'originalValidation': finding['validation']['evidenceRefs'], 'compatibleValidation': compatible['findings'][0]['validation']['evidenceRefs'], 'originalDataflow': finding['attackPath']['dataflow']['evidence_refs'], 'compatibleDataflow': compatible['findings'][0]['attackPath']['dataflow']['evidence_refs']}))", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT], + { stdout: "pipe", stderr: "pipe" }, + ); + + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + expect(JSON.parse(new TextDecoder().decode(result.stdout))).toEqual({ + originalValidation: [null, "", 42, "legacy-source", "missing-source"], + compatibleValidation: ["legacy-source"], + originalDataflow: [null, "", 42, "legacy-source", "missing-source"], + compatibleDataflow: ["legacy-source"], + }); + }); + + test("drops unsupported legacy attack-path sequences from sealed validation", () => { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "findings = json.loads((plugin / 'examples' / 'completed-scan' / 'findings.json').read_text())", + "findings['findings'][0]['attackPath'] = {'dataflow': ['source', 'sink'], 'reachability': ['authenticated uploader']}", + "finalizer = runpy.run_path(str(plugin / 'scripts' / 'finalize_scan_contract.py'))", + "compatible = finalizer['_legacy_sealed_findings_for_validation'](findings)", + "finalizer['validate_against_schema'](compatible, plugin / 'schemas' / 'findings.schema.json')", + "print(json.dumps({'original': findings['findings'][0]['attackPath'], 'compatible': compatible['findings'][0]['attackPath']}))", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT], + { stdout: "pipe", stderr: "pipe" }, + ); + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + expect(JSON.parse(new TextDecoder().decode(result.stdout))).toEqual({ + original: { + dataflow: ["source", "sink"], + reachability: ["authenticated uploader"], + }, + compatible: {}, + }); + }); + + test("keeps nullable legacy sealed evidence catalogs compatible", () => { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import copy, json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "example = json.loads((plugin / 'examples' / 'completed-scan' / 'findings.json').read_text())['findings'][0]", + "finalizer = runpy.run_path(str(plugin / 'scripts' / 'finalize_scan_contract.py'))", + "results = {}", + "for field in ('codeEvidence', 'code_evidence'):", + " finding = copy.deepcopy(example)", + " finding[field] = None", + " compatible = finalizer['_legacy_sealed_findings_for_validation']({'findings': [finding]})", + " try:", + " finalizer['_validate_finding'](compatible['findings'][0], 'findings[0]')", + " except finalizer['ContractError'] as error:", + " outcome = str(error)", + " else:", + " outcome = 'accepted'", + " results[field] = {'outcome': outcome, 'present': field in compatible['findings'][0]} ", + "print(json.dumps(results))", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT], + { stdout: "pipe", stderr: "pipe" }, + ); + + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + expect(JSON.parse(new TextDecoder().decode(result.stdout))).toEqual({ + codeEvidence: { + outcome: "findings[0].codeEvidence: expected an array", + present: true, + }, + code_evidence: { outcome: "accepted", present: false }, + }); + }); + + test("ignores malformed legacy evidence rows in sealed validation", () => { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "findings = json.loads((plugin / 'examples' / 'completed-scan' / 'findings.json').read_text())", + "findings['findings'][0]['code_evidence'] = [None, 'legacy source', {}, {'id': '', 'code': 'empty_id()'}, {'id': 'missing-code'}, {'id': 'empty-code', 'code': ''}, {'id': 'legacy-source', 'code': 'legacy_source()'}]", + "finalizer = runpy.run_path(str(plugin / 'scripts' / 'finalize_scan_contract.py'))", + "compatible = finalizer['_legacy_sealed_findings_for_validation'](findings)", + "finalizer['_validate_finding'](compatible['findings'][0], 'findings[0]')", + "print(json.dumps({'original': findings['findings'][0]['code_evidence'], 'compatible': compatible['findings'][0]['code_evidence']}))", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT], + { stdout: "pipe", stderr: "pipe" }, + ); + + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + expect(JSON.parse(new TextDecoder().decode(result.stdout))).toEqual({ + original: [ + null, + "legacy source", + {}, + { id: "", code: "empty_id()" }, + { id: "missing-code" }, + { id: "empty-code", code: "" }, + { id: "legacy-source", code: "legacy_source()" }, + ], + compatible: [{ id: "legacy-source", code: "legacy_source()" }], + }); + }); + + test("projects canonical and legacy code evidence into safe SARIF locations", () => { + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const script = [ + "import json, pathlib, runpy, sys", + "plugin = pathlib.Path(sys.argv[1])", + "finding = json.loads((plugin / 'examples' / 'completed-scan' / 'findings.json').read_text())['findings'][0]", + "finding['codeEvidence'] = [{'id': 'canonical-source', 'path': 'src/canonical.py', 'startLine': 12, 'code': 'canonical_source()'}]", + "finding['code_evidence'] = [", + " {'id': 'legacy-sink', 'path': 'src/legacy.py', 'startLine': 37, 'endLine': 39, 'code': 'legacy_sink()'},", + " {'id': 'legacy-null-end', 'path': 'src/null_end.py', 'startLine': 48, 'endLine': None, 'code': 'null_end()'},", + " {'id': 'legacy-reversed-end', 'path': 'src/reversed_end.py', 'startLine': 59, 'endLine': 58, 'code': 'reversed_end()'},", + " {'id': 'legacy-text-end', 'path': 'src/text_end.py', 'startLine': 70, 'endLine': '71', 'code': 'text_end()'},", + " {'id': 'legacy-zero-start', 'path': 'src/zero.py', 'startLine': 0, 'code': 'zero_start()'},", + " {'id': 'legacy-unsafe-path', 'path': '../outside.py', 'startLine': 81, 'code': 'unsafe_path()'},", + "]", + "finalizer = runpy.run_path(str(plugin / 'scripts' / 'finalize_scan_contract.py'))", + "result = finalizer['_sarif_result'](finding, 0)", + "regions = {location['physicalLocation']['artifactLocation']['uri']: location['physicalLocation']['region'] for location in result['locations']}", + "print(json.dumps(regions, sort_keys=True))", + ].join("\n"); + const result = Bun.spawnSync( + [python!, "-I", "-B", "-c", script, PLUGIN_ROOT], + { stdout: "pipe", stderr: "pipe" }, + ); + + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + expect(JSON.parse(new TextDecoder().decode(result.stdout))).toMatchObject({ + "src/canonical.py": { startLine: 12, endLine: 12 }, + "src/legacy.py": { startLine: 37, endLine: 39 }, + "src/null_end.py": { startLine: 48, endLine: 48 }, + "src/reversed_end.py": { startLine: 59, endLine: 59 }, + "src/text_end.py": { startLine: 70, endLine: 70 }, + }); + expect( + JSON.parse(new TextDecoder().decode(result.stdout)), + ).not.toHaveProperty("src/zero.py"); + expect( + JSON.parse(new TextDecoder().decode(result.stdout)), + ).not.toHaveProperty("../outside.py"); + }); +}); diff --git a/sdk/typescript/tests-ts/runtime.test.ts b/sdk/typescript/tests-ts/runtime.test.ts index 20e2035bb..69792433b 100644 --- a/sdk/typescript/tests-ts/runtime.test.ts +++ b/sdk/typescript/tests-ts/runtime.test.ts @@ -461,21 +461,57 @@ describe("plugin runtime preparation", () => { ]); }); - test("bounds preserved context before starting a headless scan", async () => { - const parts = await Promise.all( - ["000", "001"].map((part) => - readFile(join(PLUGIN_ROOT, "mcp", `server.mjs.br.part-${part}`)), - ), + test("accepts preserved context before starting a headless scan", () => { + const messages = [ + { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-11-25", + capabilities: {}, + clientInfo: { name: "codex-security-test", version: "1.0.0" }, + }, + }, + { jsonrpc: "2.0", method: "notifications/initialized", params: {} }, + { jsonrpc: "2.0", id: 2, method: "tools/list", params: {} }, + ]; + const server = spawnSync( + process.execPath, + [join(PLUGIN_ROOT, "mcp", "server.mjs"), "--stdio"], + { + input: `${messages.map((message) => JSON.stringify(message)).join("\n")}\n`, + encoding: "utf8", + timeout: 10_000, + }, ); - const runtime = brotliDecompressSync(Buffer.concat(parts)).toString("utf8"); - const schema = - /var startHeadlessStandardScanSchema = \{[\s\S]*?\n\};/u.exec( - runtime, - )?.[0]; + expect(server.status, server.stderr).toBe(0); + const responses = server.stdout + .trim() + .split("\n") + .map( + (line) => + JSON.parse(line) as { + id: number; + result: { + tools?: Array<{ + name: string; + inputSchema: { + properties: { userContext?: { maxLength?: number } }; + }; + }>; + }; + }, + ); + const tool = responses + .find((response) => response.id === 2) + ?.result.tools?.find( + (candidate) => candidate.name === "start_codex_security_standard_scan", + ); + const userContext = "Assess the HTTP boundary. ".repeat(320); - expect(schema).toContain( - "userContext: editableUserContextSchema.max(2400).optional()", - ); + expect(userContext.length).toBeGreaterThan(2400); + expect(tool?.inputSchema.properties.userContext?.maxLength).toBeUndefined(); }); test("keeps native scan tools without the obsolete setup widget", async () => { diff --git a/sdk/typescript/tests-ts/skeleton.test.ts b/sdk/typescript/tests-ts/skeleton.test.ts index 807ecc753..0cdec62e6 100644 --- a/sdk/typescript/tests-ts/skeleton.test.ts +++ b/sdk/typescript/tests-ts/skeleton.test.ts @@ -1,7 +1,13 @@ import { readFile } from "node:fs/promises"; import { describe, expect, test } from "bun:test"; import { parse } from "smol-toml"; -import { CodexSecurity, CodexSecurityError, VERSION } from "../src/index.js"; +import { + type AttackPathDataflow, + type AttackPathReachability, + CodexSecurity, + CodexSecurityError, + VERSION, +} from "../src/index.js"; import { main } from "../src/cli.js"; function capture(): { @@ -21,6 +27,22 @@ function capture(): { } describe("TypeScript package skeleton", () => { + test("exports typed attack-path aliases", () => { + const dataflow: AttackPathDataflow = { + transformations: ["decode archive entry"], + }; + const reachability: AttackPathReachability = { + attacker: "authenticated uploader", + entrypoint: "archive upload endpoint", + preconditions: ["archive extraction is enabled"], + }; + const transformations: string[] | undefined = dataflow.transformations; + const attacker: string | undefined = reachability.attacker; + + expect(transformations).toEqual(["decode archive entry"]); + expect(attacker).toBe("authenticated uploader"); + }); + test("advertises the tested Node.js 22, 24, and 26 release lines", async () => { const packageJson = JSON.parse( await readFile(new URL("../package.json", import.meta.url), "utf8"),