Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion sdk/typescript/_bundled_plugin/references/sarif-adapter.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ The adapter:

Lifecycle, rich validation evidence, attack-path context, and coverage are lossy or omitted in SARIF. Preserve them in semantic JSON.

SARIF `executionSuccessful` is true only for complete coverage. Incomplete scans retain findings and warnings. Integrations must read the coverage file referenced by `scan.coverageRef`; manifest status `completed` alone does not mean success.
SARIF `executionSuccessful` comes from coverage and is true only for complete coverage. Notifications report incomplete-coverage reasons and recorded run warnings; a complete scan can therefore succeed while still carrying a warning. Integrations must read the coverage file referenced by `scan.coverageRef`; manifest status `completed` alone does not mean success.

Automatic SARIF export during finalization is best-effort so projection errors cannot invalidate a canonical seal. Use the strict adapter entry point when a consumer requires SARIF and should surface export errors.

Expand Down
78 changes: 66 additions & 12 deletions sdk/typescript/_bundled_plugin/scripts/finalize_scan_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
import stat
import sys
import time
from collections.abc import Iterator
from collections.abc import Iterator, Sequence
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Any, TextIO
Expand Down Expand Up @@ -2168,6 +2168,35 @@ def _validate_sarif(sarif: dict[str, Any]) -> None:
raise ContractError("SARIF: result references an unknown rule")
if not result.get("partialFingerprints"):
raise ContractError("SARIF: result is missing partialFingerprints")
_validate_sarif_invocations(run.get("invocations"))


def _validate_sarif_invocations(invocations: Any) -> None:
if invocations is None:
return
if not isinstance(invocations, list) or not invocations:
raise ContractError("SARIF: invocations must be a non-empty array when present")
for invocation in invocations:
if not isinstance(invocation, dict):
raise ContractError("SARIF: expected an invocation object")
if not isinstance(invocation.get("executionSuccessful"), bool):
raise ContractError("SARIF: invocation is missing executionSuccessful")
notifications = invocation.get("toolExecutionNotifications")
# A complete scan with no warnings reports success and notifies nothing.
if notifications is None:
continue
if not isinstance(notifications, list) or not notifications:
raise ContractError("SARIF: invocation has no toolExecutionNotifications")
for notification in notifications:
if not isinstance(notification, dict):
raise ContractError("SARIF: expected a notification object")
if notification.get("level") not in {"none", "note", "warning", "error"}:
raise ContractError("SARIF: notification has an unsupported level")
message = notification.get("message")
if not isinstance(message, dict) or not isinstance(message.get("text"), str):
raise ContractError("SARIF: notification is missing message text")
if not message["text"].strip():
raise ContractError("SARIF: notification message text is empty")


def _artifact_record(
Expand Down Expand Up @@ -2278,7 +2307,10 @@ def _read_sealed_scan(


def build_sarif_projection(
scan_dir: Path, source_root: Path | None = None, schema_dir: Path | None = None
scan_dir: Path,
source_root: Path | None = None,
schema_dir: Path | None = None,
warnings: Sequence[str] | None = None,
) -> dict[str, Any]:
if source_root is not None:
try:
Expand All @@ -2293,22 +2325,39 @@ def build_sarif_projection(
run = sarif["runs"][0]
completeness = coverage["completeness"]
run["invocations"] = [{"executionSuccessful": completeness == "complete"}]
notifications: list[dict[str, Any]] = []
reported: set[str] = set()
if completeness != "complete":
run["properties"]["codexSecurityCoverageCompleteness"] = completeness
reasons = [item["reason"] for item in coverage["deferred"]] or [
f"Scan coverage is {completeness}; results may be incomplete."
]
run["invocations"][0]["toolExecutionNotifications"] = [
{"level": "warning", "message": {"text": reason}} for reason in reasons
]
for reason in reasons:
reported.add(reason)
notifications.append({"level": "warning", "message": {"text": reason}})
# Run warnings are reported whatever the completeness. A scan whose target drifted
# reviewed everything it set out to review, so it stays complete; the tree simply moved
# underneath it, and `toolExecutionNotifications` is where SARIF expects to read that.
# Deferred coverage already contributes its reason verbatim as a warning, so the same
# text is not notified twice.
for warning in warnings or ():
if not isinstance(warning, str) or not warning or warning in reported:
continue
reported.add(warning)
notifications.append({"level": "warning", "message": {"text": warning}})
if notifications:
run["invocations"][0]["toolExecutionNotifications"] = notifications
_validate_sarif(sarif)
return sarif


def write_sarif_projection(
scan_dir: Path, source_root: Path | None = None, schema_dir: Path | None = None
scan_dir: Path,
source_root: Path | None = None,
schema_dir: Path | None = None,
warnings: Sequence[str] | None = None,
) -> None:
sarif = build_sarif_projection(scan_dir, source_root, schema_dir)
sarif = build_sarif_projection(scan_dir, source_root, schema_dir, warnings)
_write_scan_local_json(scan_dir, "exports/results.sarif", sarif)


Expand Down Expand Up @@ -2487,10 +2536,13 @@ def write_export_output(scan_dir: Path, output: Path, export_format: str, conten


def _write_sarif_projection_if_possible(
scan_dir: Path, source_root: Path | None = None, schema_dir: Path | None = None
scan_dir: Path,
source_root: Path | None = None,
schema_dir: Path | None = None,
warnings: Sequence[str] | None = None,
) -> None:
try:
write_sarif_projection(scan_dir, source_root, schema_dir)
write_sarif_projection(scan_dir, source_root, schema_dir, warnings)
except (ContractError, OSError) as error:
print(
f"codex-security: warning: automatic SARIF export failed: {error}. "
Expand Down Expand Up @@ -2642,6 +2694,7 @@ def _prepare_scan_finalization(
def _write_prepared_scan_finalization(
prepared: PreparedScanFinalization,
source_root: Path | None = None,
warnings: Sequence[str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
"""Write a previously validated scan finalization result."""

Expand All @@ -2658,7 +2711,7 @@ def _write_prepared_scan_finalization(
if was_sealed:
write_scan_local_bytes(scan_dir, "report.md", report_markdown_bytes)
_remove_scan_local_file_if_exists(scan_dir, "report.html")
_write_sarif_projection_if_possible(scan_dir, source_root, schema_dir)
_write_sarif_projection_if_possible(scan_dir, source_root, schema_dir, warnings)
return manifest, findings, coverage

_write_scan_local_json(scan_dir, "findings.json", findings)
Expand All @@ -2667,7 +2720,7 @@ def _write_prepared_scan_finalization(
_remove_scan_local_file_if_exists(scan_dir, "report.html")
_write_scan_local_json(scan_dir, "scan-manifest.json", manifest)
_validate_existing_seal(scan_dir, scan)
_write_sarif_projection_if_possible(scan_dir, source_root, schema_dir)
_write_sarif_projection_if_possible(scan_dir, source_root, schema_dir, warnings)
return manifest, findings, coverage


Expand All @@ -2678,14 +2731,15 @@ def finalize_scan(
*,
expected_coverage_mode: str | None = None,
completion_binding: dict[str, Any] | None = None,
warnings: Sequence[str] | None = None,
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
prepared = _prepare_scan_finalization(
scan_dir,
schema_dir,
expected_coverage_mode=expected_coverage_mode,
completion_binding=completion_binding,
)
return _write_prepared_scan_finalization(prepared, source_root)
return _write_prepared_scan_finalization(prepared, source_root, warnings)


def main() -> int:
Expand Down
14 changes: 12 additions & 2 deletions sdk/typescript/_bundled_plugin/scripts/workbench_db.py
Original file line number Diff line number Diff line change
Expand Up @@ -1433,10 +1433,12 @@ def complete_scan_locked(
scan_dir = require_canonical_scan_directory(Path(scan["scan_dir"]))
require_recorded_manifest_digest(scan, scan_dir)
verify_manifest_binding(scan, read_json_object(scan_dir / ARTIFACTS["manifest"]))
completion_warnings = json.loads(scan["completion_warnings_json"])
try:
manifest, _, _ = finalize_scan(
scan_dir,
expected_coverage_mode=expected_coverage_mode(scan),
warnings=completion_warnings,
)
except ContractError as exc:
raise SystemExit(str(exc)) from exc
Expand Down Expand Up @@ -1502,7 +1504,9 @@ def add_warning() -> None:
completion_warnings=warnings,
)
add_warning()
manifest, findings, _ = _write_prepared_scan_finalization(prepared)
manifest, findings, _ = _write_prepared_scan_finalization(
prepared, warnings=warnings
)
except ContractError as exc:
raise SystemExit(str(exc)) from exc
artifacts = {
Expand Down Expand Up @@ -2672,10 +2676,14 @@ def export_findings(connection: sqlite3.Connection, args: argparse.Namespace) ->
scan_dir = require_canonical_scan_directory(Path(scan["scan_dir"]))
require_recorded_manifest_digest(scan, scan_dir)
verify_manifest_binding(scan, read_json_object(scan_dir / ARTIFACTS["manifest"]))
# Re-exporting has to reproduce the notifications the scan recorded at completion,
# so the warnings are read back rather than left behind with the completed scan.
completion_warnings = json.loads(scan["completion_warnings_json"])
try:
manifest, _, _ = finalize_scan(
scan_dir,
expected_coverage_mode=expected_coverage_mode(scan),
warnings=completion_warnings,
)
except ContractError as exc:
raise SystemExit(str(exc)) from exc
Expand All @@ -2686,7 +2694,7 @@ def export_findings(connection: sqlite3.Connection, args: argparse.Namespace) ->
path = artifact_path(scan_dir, ARTIFACTS["findings"], required=True)
elif args.format == "sarif":
try:
write_sarif_projection(scan_dir)
write_sarif_projection(scan_dir, warnings=completion_warnings)
except ContractError as exc:
raise SystemExit(str(exc)) from exc
path = artifact_path(scan_dir, "exports/results.sarif", required=True)
Expand Down Expand Up @@ -3369,9 +3377,11 @@ def backfill_legacy_finding_details(connection: sqlite3.Connection, scan: sqlite
scan_dir = require_canonical_scan_directory(Path(scan["scan_dir"]))
require_recorded_manifest_digest(scan, scan_dir)
verify_manifest_binding(scan, read_json_object(scan_dir / ARTIFACTS["manifest"]))
completion_warnings = json.loads(scan["completion_warnings_json"])
manifest, findings_document, _ = finalize_scan(
scan_dir,
expected_coverage_mode=expected_coverage_mode(scan),
warnings=completion_warnings,
)
verify_manifest_binding(scan, manifest)
manifest_digest = published_manifest_digest(scan_dir, manifest)
Expand Down
2 changes: 1 addition & 1 deletion sdk/typescript/tests-ts/cli-export.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -418,7 +418,7 @@ describe("CLI", () => {
completeness === "complete" ? undefined : completeness,
);
if (completeness === "complete") {
expect(invocation.toolExecutionNotifications).toBeUndefined();
expect(invocation).not.toHaveProperty("toolExecutionNotifications");
} else {
const reasons = hasDeferred
? coverage.deferred.map((item) => item.reason)
Expand Down
Loading