← Files Codex SecurityARCHIVED FILE
scripts/finalize_scan_contract.py
119 KB · Oct 2, 2026 · 00:04 UTC
#!/usr/bin/env python3
"""Validate and seal additive Codex Security scan-contract artifacts."""
from __future__ import annotations
import argparse
import copy
import csv
import errno
import hashlib
import importlib.util
import io
import json
import math
import os
import re
import secrets
import stat
import sys
import time
from collections.abc import Iterator
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Any, TextIO
from urllib.parse import quote, urlsplit
SCHEMA_VERSION = "1.0"
PRODUCER_NAME = "codex-security-plugin"
FINGERPRINT_ALGORITHM = "codex-security/v1"
SARIF_SCHEMA = "https://docs.oasis-open.org/sarif/sarif/v2.1.0/os/schemas/sarif-schema-2.1.0.json"
SEVERITIES = {"critical", "high", "medium", "low", "informational"}
CONFIDENCES = {"high", "medium", "low"}
TARGET_KINDS = {"git_revision", "git_worktree", "git_diff", "directory_snapshot"}
TARGET_COORDINATE_FIELDS = {
"revision",
"baseRevision",
"headRevision",
"snapshotDigest",
}
TARGET_REQUIRED_COORDINATE_FIELDS = {
"git_revision": {"revision"},
"git_worktree": {"snapshotDigest"},
"git_diff": {"snapshotDigest"},
"directory_snapshot": {"snapshotDigest"},
}
DISPOSITIONS = {"reported", "no_issue_found", "rejected", "not_applicable", "needs_follow_up"}
SARIF_LEVELS = {
"critical": "error",
"high": "error",
"medium": "warning",
"low": "note",
"informational": "note",
}
SARIF_SECURITY_SCORES = {
"critical": 9.5,
"high": 8.0,
"medium": 5.0,
"low": 2.0,
"informational": 0.0,
}
SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9._/-]*$")
RFC3339_RE = re.compile(
r"^\d{4}-\d{2}-\d{2}[Tt]\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:[Zz]|[+-]\d{2}:\d{2})$"
)
GITHUB_HASH_BLOCK_SIZE = 100
GITHUB_HASH_MOD = 37
GITHUB_HASH_MASK = (1 << 64) - 1
GITHUB_HASH_EOF = 65535
SOURCE_READ_CHUNK_SIZE = 64 * 1024
MAX_JSON_INTEGER = (1 << 53) - 1
EXPORT_PATHS = {
"csv": "exports/findings.csv",
"json": "exports/findings.json",
"sarif": "exports/results.sarif",
}
WINDOWS_UNSAFE_PATH_COMPONENT_RE = re.compile(
r'[<>:"|?*\x00-\x1f]|[ .]$|^(?:con|prn|aux|nul|conin\$|conout\$|com[1-9¹²³]|lpt[1-9¹²³])(?:\..*)?$',
re.IGNORECASE,
)
class ContractError(ValueError):
"""Raised when a completed scan does not satisfy the additive contract."""
class RecoverableContractError(ContractError):
"""Raised when report projection can safely be retried before publication."""
def _reject_non_finite_json(value: str) -> None:
raise ValueError(f"non-finite JSON number {value!r} is not supported")
def _loads_json(value: str | bytes) -> Any:
return json.loads(value, parse_constant=_reject_non_finite_json)
def _read_json(path: Path) -> dict[str, Any]:
try:
payload = _loads_json(path.read_text(encoding="utf-8"))
_require_safe_json_value(payload, str(path))
except FileNotFoundError as exc:
raise ContractError(f"missing required contract artifact: {path}") from exc
except ContractError:
raise
except ValueError as exc:
raise ContractError(f"{path}: invalid JSON: {exc}") from exc
if not isinstance(payload, dict):
raise ContractError(f"{path}: expected a JSON object")
return payload
def _generate_report_projection(
manifest: dict[str, Any],
findings: dict[str, Any],
coverage: dict[str, Any],
) -> bytes:
script = Path(__file__).resolve().parent / "report_projection.py"
spec = importlib.util.spec_from_file_location("codex_security_report_projection", script)
if spec is None or spec.loader is None:
raise ContractError(f"could not load report projection helper: {script}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
attempts = (
getattr(sys.modules.get("workbench_constants"), "SQLITE_RETRY_ATTEMPTS", 1)
if coverage.get("mode") == "deep_repository"
else 1
)
for attempt in range(attempts):
try:
return module.generate_report_markdown(manifest, findings, coverage)
except OSError as exc:
if attempt == attempts - 1:
raise RecoverableContractError(f"report projection failed: {exc}") from exc
time.sleep(0.05 * (2**attempt))
except ValueError as exc:
raise ContractError(f"report projection failed: {exc}") from exc
raise AssertionError("Report projection retry loop exhausted unexpectedly.")
def _validate_report_output_paths(scan_dir: Path) -> None:
_validate_scan_local_output_path(scan_dir, scan_dir / "report.md", "report.md")
def _json_bytes(payload: Any) -> bytes:
try:
encoded = json.dumps(payload, allow_nan=False, indent=2, sort_keys=True)
except ValueError as exc:
raise ContractError(f"cannot encode canonical JSON: {exc}") from exc
return (encoded + "\n").encode("utf-8")
def _contract_json_bytes(relative_path: str, payload: Any) -> bytes:
_require_safe_json_value(payload, relative_path)
return _json_bytes(payload)
def _require_safe_json_value(value: Any, context: str, *, validate_strings: bool = True) -> None:
def visit(item: Any, location: str) -> None:
if isinstance(item, dict):
for key, child in item.items():
if not isinstance(key, str):
raise ContractError(f"{location}: expected string JSON property names")
if validate_strings:
_require_safe_json_string(key, location)
visit(child, f"{location}.<property>")
elif isinstance(item, list):
for index, child in enumerate(item):
visit(child, f"{location}[{index}]")
elif isinstance(item, str) and validate_strings:
_require_safe_json_string(item, location)
elif isinstance(item, int) and not isinstance(item, bool):
if abs(item) > MAX_JSON_INTEGER:
raise ContractError(
f"{location}: unsafe integer-valued JSON numbers are not supported"
)
elif isinstance(item, float):
if not math.isfinite(item):
raise ContractError(f"{location}: non-finite JSON numbers are not supported")
if item.is_integer() and abs(item) > MAX_JSON_INTEGER:
raise ContractError(
f"{location}: unsafe integer-valued JSON numbers are not supported"
)
visit(value, context)
def _require_safe_json_string(value: str, context: str) -> None:
try:
value.encode("utf-8")
except UnicodeEncodeError as exc:
raise ContractError(f"{context}: expected well-formed Unicode JSON strings") from exc
def _sha256_text(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()
def _sha256_bytes(value: bytes) -> str:
return hashlib.sha256(value).hexdigest()
def _sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _require_dict(payload: dict[str, Any], key: str, context: str) -> dict[str, Any]:
value = payload.get(key)
if not isinstance(value, dict):
raise ContractError(f"{context}.{key}: expected an object")
return value
def _require_list(payload: dict[str, Any], key: str, context: str) -> list[Any]:
value = payload.get(key)
if not isinstance(value, list):
raise ContractError(f"{context}.{key}: expected an array")
return value
def _require_str(payload: dict[str, Any], key: str, context: str) -> str:
value = payload.get(key)
if not isinstance(value, str) or not value.strip():
raise ContractError(f"{context}.{key}: expected a non-empty string")
return value
def _require_safe_relative_path(value: str, context: str, *, allow_dot: bool = False) -> str:
try:
value.encode("utf-8")
except UnicodeEncodeError as exc:
raise ContractError(f"{context}: expected a safe repository-relative POSIX path") from exc
path = PurePosixPath(value)
normalized = path.as_posix()
if (
not value.strip()
or (normalized == "." and not allow_dot)
or (len(value) >= 2 and value[0].isalpha() and value[1] == ":")
or "\\" in value
or "\0" in value
or any(ord(character) < 32 for character in value)
or path.is_absolute()
or ".." in path.parts
):
raise ContractError(f"{context}: expected a safe repository-relative POSIX path")
return normalized
def _require_portable_relative_path(value: str, context: str, *, allow_dot: bool = False) -> str:
normalized = _require_safe_relative_path(value, context, allow_dot=allow_dot)
if any(_windows_unsafe_path_component(part) for part in value.split("/")):
raise ContractError(f"{context}: expected a safe scan-relative POSIX path")
return normalized
def _windows_unsafe_path_component(value: str) -> bool:
if not value or value == ".":
return False
return WINDOWS_UNSAFE_PATH_COMPONENT_RE.search(value) is not None
def _require_scan_directory(scan_dir: Path) -> Path:
scan_dir = scan_dir.absolute()
try:
metadata = scan_dir.lstat()
except OSError as exc:
raise ContractError("scan directory: expected an existing non-symlink directory") from exc
if not stat.S_ISDIR(metadata.st_mode):
raise ContractError("scan directory: expected an existing non-symlink directory")
try:
resolved = scan_dir.resolve(strict=True)
except OSError as exc:
raise ContractError("scan directory: expected an existing non-symlink directory") from exc
if os.path.normcase(resolved) != os.path.normcase(scan_dir):
raise ContractError("scan directory: expected a canonical non-symlink directory")
return resolved
def _validate_scan_local_output_path(scan_dir: Path, path: Path, relative_path: str) -> None:
try:
resolved_parent = path.parent.resolve(strict=True)
resolved_parent.relative_to(scan_dir)
except (OSError, RuntimeError, ValueError) as exc:
raise ContractError(f"{relative_path}: expected a path inside the scan directory") from exc
if os.path.normcase(resolved_parent) != os.path.normcase(path.parent) or path.is_symlink():
raise ContractError(
f"{relative_path}: expected a non-symlink path inside the scan directory"
)
if path.exists() and not path.is_file():
raise ContractError(f"{relative_path}: expected a regular file")
def _descriptor_relative_reads_available() -> bool:
return os.open in os.supports_dir_fd and hasattr(os, "O_NOFOLLOW")
def _is_windows() -> bool:
return os.name == "nt"
def _descriptor_relative_writes_available() -> bool:
# os.replace accepts src_dir_fd/dst_dir_fd wherever descriptor-relative
# os.rename is supported, but Python lists only os.rename in supports_dir_fd.
required_operations = (os.mkdir, os.open, os.rename, os.stat, os.unlink)
return hasattr(os, "O_NOFOLLOW") and all(
operation in os.supports_dir_fd for operation in required_operations
)
_WINDOWS_SCAN_LOCAL_FILES: Any | None = None
def _windows_scan_local_files() -> Any:
"""Load the Win32 backend and shared stream comparison lazily."""
global _WINDOWS_SCAN_LOCAL_FILES
if _WINDOWS_SCAN_LOCAL_FILES is None:
script = Path(__file__).resolve().with_name("windows_scan_local_files.py")
spec = importlib.util.spec_from_file_location("codex_security_windows_scan_files", script)
if spec is None or spec.loader is None:
raise ContractError(f"could not load Windows scan-local file helper: {script}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
_WINDOWS_SCAN_LOCAL_FILES = module
return _WINDOWS_SCAN_LOCAL_FILES
def _open_verified_scan_directory(
scan_dir: Path, expected_root_identity: tuple[int, int] | None = None
) -> int:
scan_dir = scan_dir.absolute()
try:
expected = scan_dir.lstat()
observed_identity = (expected.st_dev, expected.st_ino)
if expected_root_identity is not None and observed_identity != expected_root_identity:
raise ContractError("scan directory: changed after artifact restoration setup")
canonical = _require_scan_directory(scan_dir)
descriptor = os.open(
canonical,
os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0),
)
except OSError as exc:
raise ContractError("scan directory: expected an existing non-symlink directory") from exc
opened = os.fstat(descriptor)
opened_identity = (opened.st_dev, opened.st_ino)
if opened_identity != observed_identity or (
expected_root_identity is not None and opened_identity != expected_root_identity
):
os.close(descriptor)
raise ContractError("scan directory: changed while it was being opened")
return descriptor
def scan_root_identity(scan_dir: Path) -> tuple[Path, tuple[int, int]]:
"""Return a canonical scan root and its identity from a held handle."""
scan_dir = _require_scan_directory(scan_dir)
if not _descriptor_relative_writes_available():
if not _is_windows():
raise ContractError("scan-local output requires descriptor-relative file operations")
return _windows_scan_local_files().scan_root_identity(scan_dir)
descriptor = _open_verified_scan_directory(scan_dir)
try:
metadata = os.fstat(descriptor)
return scan_dir, (metadata.st_dev, metadata.st_ino)
finally:
os.close(descriptor)
def _open_scan_local_directory(root_fd: int, parts: tuple[str, ...], *, create: bool) -> int:
descriptor = os.dup(root_fd)
try:
for part in parts:
if create:
try:
os.mkdir(part, mode=0o700, dir_fd=descriptor)
except FileExistsError:
pass
next_descriptor = os.open(
part,
os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0),
dir_fd=descriptor,
)
os.close(descriptor)
descriptor = next_descriptor
return descriptor
except BaseException:
os.close(descriptor)
raise
def open_scan_local_file_descriptor(scan_dir: Path, relative_path: str, context: str) -> int:
scan_dir = _require_scan_directory(scan_dir)
relative_path = _require_portable_relative_path(relative_path, context)
if not _descriptor_relative_reads_available():
if not _is_windows():
raise ContractError("scan-local input requires descriptor-relative file operations")
try:
return _windows_scan_local_files().open_read_fd(scan_dir, relative_path, context)
except OSError as exc:
raise ContractError(str(exc)) from exc
root_fd: int | None = None
parent_fd: int | None = None
descriptor: int | None = None
try:
root_fd = _open_verified_scan_directory(scan_dir)
parts = PurePosixPath(relative_path).parts
try:
parent_fd = _open_scan_local_directory(root_fd, parts[:-1], create=False)
descriptor = os.open(
parts[-1],
os.O_RDONLY | os.O_NOFOLLOW | getattr(os, "O_NONBLOCK", 0),
dir_fd=parent_fd,
)
except OSError as exc:
if exc.errno == errno.ELOOP:
try:
link_target = Path(os.readlink(parts[-1], dir_fd=parent_fd))
if not link_target.is_absolute():
link_target = scan_dir.joinpath(*parts[:-1], link_target)
link_target.resolve(strict=False).relative_to(scan_dir)
except (OSError, RuntimeError, ValueError):
raise ContractError(
f"{context}: expected a file inside the scan directory"
) from exc
raise ContractError(f"{context}: expected a regular non-symlink file") from exc
raise ContractError(f"{context}: expected a file inside the scan directory") from exc
metadata = os.fstat(descriptor)
if not stat.S_ISREG(metadata.st_mode):
raise ContractError(f"{context}: expected a regular non-symlink file")
result = descriptor
descriptor = None
return result
finally:
if descriptor is not None:
os.close(descriptor)
if parent_fd is not None:
os.close(parent_fd)
if root_fd is not None:
os.close(root_fd)
def _require_scan_local_file(scan_dir: Path, relative_path: str, context: str) -> None:
descriptor = open_scan_local_file_descriptor(scan_dir, relative_path, context)
os.close(descriptor)
def _require_derived_writeup_files(scan_dir: Path, findings: dict[str, Any]) -> None:
for index, finding in enumerate(findings.get("findings", [])):
if not isinstance(finding, dict):
continue
writeup = finding.get("writeup")
if not isinstance(writeup, dict):
continue
report_path = writeup.get("reportPath")
if isinstance(report_path, str):
_require_scan_local_file(
scan_dir,
report_path,
f"findings[{index}].writeup.reportPath",
)
def _require_hardening_portfolio_file(scan_dir: Path, scan: dict[str, Any]) -> None:
hardening = scan.get("hardening")
if not isinstance(hardening, dict):
return
portfolio_path = hardening.get("portfolioPath")
if isinstance(portfolio_path, str):
_require_scan_local_file(
scan_dir,
portfolio_path,
"manifest.scan.hardening.portfolioPath",
)
def _read_scan_local_json_bytes(
scan_dir: Path, relative_path: str, context: str
) -> tuple[dict[str, Any], bytes]:
descriptor = open_scan_local_file_descriptor(scan_dir, relative_path, context)
try:
with os.fdopen(descriptor, "rb") as handle:
descriptor = -1
raw = handle.read()
try:
payload = _loads_json(raw.decode("utf-8"))
except (UnicodeDecodeError, ValueError) as exc:
raise ContractError(f"{context}: invalid JSON: {exc}") from exc
if not isinstance(payload, dict):
raise ContractError(f"{context}: expected a JSON object")
_require_safe_json_value(payload, context, validate_strings=False)
finally:
if descriptor >= 0:
os.close(descriptor)
return payload, raw
def _read_scan_local_json(scan_dir: Path, relative_path: str, context: str) -> dict[str, Any]:
payload, _ = _read_scan_local_json_bytes(scan_dir, relative_path, context)
return payload
def _sha256_scan_local_file(scan_dir: Path, relative_path: str, context: str) -> str:
descriptor = open_scan_local_file_descriptor(scan_dir, relative_path, context)
digest = hashlib.sha256()
try:
with os.fdopen(descriptor, "rb") as handle:
descriptor = -1
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
finally:
if descriptor >= 0:
os.close(descriptor)
return digest.hexdigest()
def write_scan_local_bytes(
scan_dir: Path,
relative_path: str,
payload: bytes,
*,
external_name: bool = False,
expected_root_identity: tuple[int, int] | None = None,
) -> None:
scan_dir = _require_scan_directory(scan_dir)
if external_name:
if relative_path in {"", ".", ".."} or "/" in relative_path or "\0" in relative_path:
raise ContractError("external output path: expected a safe file name")
else:
relative_path = _require_portable_relative_path(relative_path, "scan-local output path")
path = scan_dir / relative_path
if not _descriptor_relative_writes_available():
if not _is_windows():
raise ContractError("scan-local output requires descriptor-relative file operations")
try:
_windows_scan_local_files().atomic_write(
scan_dir,
relative_path,
payload,
expected_root_identity=expected_root_identity,
)
except OSError as exc:
raise ContractError(f"{relative_path}: {exc}") from exc
return
root_fd: int | None = None
parent_fd: int | None = None
temp_name: str | None = None
try:
root_fd = _open_verified_scan_directory(scan_dir, expected_root_identity)
parts = PurePosixPath(relative_path).parts
try:
parent_fd = _open_scan_local_directory(root_fd, parts[:-1], create=True)
except OSError as exc:
raise ContractError(
f"{relative_path}: expected a path inside the scan directory"
) from exc
# The held descriptor is the authority for the validated parent. A
# concurrent rename cannot redirect later operations through a
# replacement path or link.
try:
metadata = os.stat(parts[-1], dir_fd=parent_fd, follow_symlinks=False)
except FileNotFoundError:
pass
else:
if not stat.S_ISREG(metadata.st_mode):
raise ContractError(f"{relative_path}: expected a regular non-symlink file")
try:
existing_fd = os.open(
parts[-1],
os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0),
dir_fd=parent_fd,
)
except OSError as exc:
if exc.errno not in {errno.ENOENT, errno.EACCES, errno.EPERM}:
raise
else:
try:
opened = os.fstat(existing_fd)
if not stat.S_ISREG(opened.st_mode):
raise ContractError(f"{relative_path}: expected a regular non-symlink file")
if (opened.st_dev, opened.st_ino) != (
metadata.st_dev,
metadata.st_ino,
):
raise ContractError(f"{relative_path}: changed while it was being opened")
if expected_root_identity is not None and opened.st_size == len(payload):
try:
with os.fdopen(existing_fd, "rb") as handle:
existing_fd = -1
if _windows_scan_local_files().stream_matches_payload(
handle, payload
):
return
except OSError:
pass
finally:
if existing_fd >= 0:
os.close(existing_fd)
temp_name = f".{path.name}.{secrets.token_hex(8)}.tmp"
temp_fd = os.open(temp_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600, dir_fd=parent_fd)
with os.fdopen(temp_fd, "wb") as handle:
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
os.replace(temp_name, parts[-1], src_dir_fd=parent_fd, dst_dir_fd=parent_fd)
temp_name = None
finally:
if temp_name is not None and parent_fd is not None:
try:
os.unlink(temp_name, dir_fd=parent_fd)
except FileNotFoundError:
pass
if parent_fd is not None:
os.close(parent_fd)
if root_fd is not None:
os.close(root_fd)
def _remove_scan_local_file_if_exists(scan_dir: Path, relative_path: str) -> None:
scan_dir = _require_scan_directory(scan_dir)
relative_path = _require_portable_relative_path(relative_path, "scan-local cleanup path")
if not _descriptor_relative_writes_available():
if not _is_windows():
raise ContractError("scan-local cleanup requires descriptor-relative file operations")
try:
_windows_scan_local_files().unlink_if_exists(scan_dir, relative_path)
except OSError as exc:
raise ContractError(f"{relative_path}: {exc}") from exc
return
root_fd: int | None = None
parent_fd: int | None = None
try:
root_fd = _open_verified_scan_directory(scan_dir)
parts = PurePosixPath(relative_path).parts
parent_fd = _open_scan_local_directory(root_fd, parts[:-1], create=False)
try:
metadata = os.stat(parts[-1], dir_fd=parent_fd, follow_symlinks=False)
except FileNotFoundError:
return
if not (stat.S_ISREG(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode)):
raise ContractError(f"{relative_path}: expected a regular file or symlink")
os.unlink(parts[-1], dir_fd=parent_fd)
finally:
if parent_fd is not None:
os.close(parent_fd)
if root_fd is not None:
os.close(root_fd)
def _write_scan_local_json(scan_dir: Path, relative_path: str, payload: Any) -> None:
write_scan_local_bytes(scan_dir, relative_path, _contract_json_bytes(relative_path, payload))
def _validate_remote(remote: str, context: str) -> None:
parsed = urlsplit(remote)
if not parsed.scheme or not parsed.netloc:
raise ContractError(f"{context}: expected a sanitized canonical absolute URL")
if parsed.username or parsed.password or parsed.query or parsed.fragment:
raise ContractError(
f"{context}: remote URL must not contain credentials, query, or fragment"
)
def _validate_date_time(value: str, context: str) -> None:
if not RFC3339_RE.fullmatch(value):
raise ContractError(f"{context}: expected an RFC 3339 timestamp")
try:
parsed = datetime.fromisoformat(value[:-1] + "+00:00" if value[-1] in "Zz" else value)
except ValueError as exc:
raise ContractError(f"{context}: expected an RFC 3339 timestamp") from exc
if parsed.tzinfo is None:
raise ContractError(f"{context}: expected an RFC 3339 timestamp")
def _validate_target(target: dict[str, Any]) -> None:
kind = _require_str(target, "kind", "scan.target")
if kind not in TARGET_KINDS:
raise ContractError(f"scan.target.kind: unsupported target kind: {kind}")
_require_str(target, "targetId", "scan.target")
_require_str(target, "displayName", "scan.target")
remote = target.get("remote")
if remote is not None:
if not isinstance(remote, str):
raise ContractError("scan.target.remote: expected a string")
_validate_remote(remote, "scan.target.remote")
if kind == "git_revision":
_require_str(target, "revision", "scan.target")
elif kind == "git_worktree":
_require_str(target, "snapshotDigest", "scan.target")
elif kind == "git_diff":
_require_str(target, "snapshotDigest", "scan.target")
elif kind == "directory_snapshot":
_require_str(target, "snapshotDigest", "scan.target")
def _fingerprint(target_id: str, finding: dict[str, Any]) -> str:
identity = _require_dict(finding, "identity", "finding")
anchor = _require_str(identity, "anchor", "finding.identity")
if not SLUG_RE.fullmatch(anchor):
raise ContractError("finding.identity.anchor: expected a stable lowercase semantic slug")
instance = identity.get("instance", "")
if not isinstance(instance, str):
raise ContractError("finding.identity.instance: expected a string")
if instance and not SLUG_RE.fullmatch(instance):
raise ContractError("finding.identity.instance: expected a stable lowercase semantic slug")
rule_id = _require_str(finding, "ruleId", "finding")
if not SLUG_RE.fullmatch(rule_id):
raise ContractError("finding.ruleId: expected a stable lowercase rule slug")
material = "\0".join((FINGERPRINT_ALGORITHM, target_id, rule_id, anchor, instance))
return f"{FINGERPRINT_ALGORITHM}:sha256:{_sha256_text(material)}"
def _stable_id(prefix: str, *parts: str) -> str:
return f"{prefix}_{_sha256_text(chr(0).join(parts))[:24]}"
def _validate_location(location: dict[str, Any], context: str) -> None:
_require_safe_relative_path(_require_str(location, "path", context), f"{context}.path")
start = location.get("startLine")
end = location.get("endLine", start)
if not isinstance(start, int) or start < 1:
raise ContractError(f"{context}.startLine: expected a positive integer")
if not isinstance(end, int) or end < start:
raise ContractError(f"{context}.endLine: expected an integer >= startLine")
role = location.get("role")
if role is not None and (not isinstance(role, str) or not role):
raise ContractError(f"{context}.role: expected a non-empty string")
def _derived_finding_identity_rows(
manifest: dict[str, Any],
findings: dict[str, Any],
) -> list[tuple[str, dict[str, Any], str, str, dict[str, str]]]:
scan = _require_dict(manifest, "scan", "manifest")
scan_id = _require_str(scan, "id", "manifest.scan")
target_id = _require_str(
_require_dict(scan, "target", "manifest.scan"), "targetId", "scan.target"
)
if findings.get("scanId") != scan_id:
raise ContractError("findings.scanId: must match manifest scan id")
finding_ids: set[str] = set()
occurrence_ids: set[str] = set()
rows: list[tuple[str, dict[str, Any], str, str, dict[str, str]]] = []
for index, finding in enumerate(_require_list(findings, "findings", "findings")):
context = f"findings.findings[{index}]"
if not isinstance(finding, dict):
raise ContractError(f"{context}: expected an object")
fingerprint = _fingerprint(target_id, finding)
expected_finding_id = _stable_id("csf", fingerprint)
expected_occurrence_id = _stable_id("occ", scan_id, fingerprint)
expected_fingerprints = {"algorithm": FINGERPRINT_ALGORITHM, "primary": fingerprint}
rows.append(
(
context,
finding,
expected_finding_id,
expected_occurrence_id,
expected_fingerprints,
)
)
finding_ids.add(expected_finding_id)
if expected_occurrence_id in occurrence_ids:
raise ContractError(
f"{context}: duplicate occurrence identity; use identity.instance to split siblings"
)
occurrence_ids.add(expected_occurrence_id)
if len(finding_ids) != len(occurrence_ids):
raise ContractError("findings: duplicate logical findings in one scan")
return rows
def _populate_unsealed_finding_identities(
manifest: dict[str, Any],
findings: dict[str, Any],
) -> None:
"""Replace draft-owned finding identities with deterministic finalizer values."""
for _, finding, finding_id, occurrence_id, fingerprints in _derived_finding_identity_rows(
manifest,
findings,
):
finding["findingId"] = finding_id
finding["occurrenceId"] = occurrence_id
finding["fingerprints"] = fingerprints
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"]),
_finding_evidence_strength(finding),
)
def _recover_unsealed_findings(
manifest: dict[str, Any],
findings: dict[str, Any],
schema_dir: Path,
scan_dir: Path,
warnings: list[str],
) -> list[str]:
schema = _read_json(schema_dir / "findings.schema.json")
properties = _require_dict(schema, "properties", "findings.schema")
finding_array = _require_dict(properties, "findings", "findings.schema.properties")
finding_schema = _require_dict(finding_array, "items", "findings.schema.properties.findings")
finding_properties = _require_dict(
finding_schema, "properties", "findings.schema.properties.findings.items"
)
writeup_schema = _require_dict(
finding_properties, "writeup", "findings.schema.properties.findings.items.properties"
)
auxiliary_schemas = {
name: _require_dict(
finding_properties, name, "findings.schema.properties.findings.items.properties"
)
for name in ("remediationTests", "preventiveControls")
}
scan = _require_dict(manifest, "scan", "manifest")
scan_id = _require_str(scan, "id", "manifest.scan")
if findings.get("scanId") != scan_id:
raise ContractError("findings.scanId: must match manifest scan id")
recovered: list[dict[str, Any]] = []
discarded: list[str] = []
finding_positions: dict[str, int] = {}
writeup_paths: set[str] = set()
for index, finding in enumerate(_require_list(findings, "findings", "findings")):
context = f"findings.findings[{index}]"
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"),
(identity, "anchor", f"{context}.identity", "semantic anchor"),
]
if "instance" in identity:
fields.append((identity, "instance", f"{context}.identity", "instance"))
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):
continue
normalized = re.sub(r"[^a-z0-9._/-]+", "-", value.lower()).strip("._/-")
if not SLUG_RE.fullmatch(normalized):
raise ContractError(
f"{field_context}.{field}: expected a stable lowercase semantic slug"
)
parent[field] = normalized
normalized_fields.append(label)
severity = finding.get("severity")
if isinstance(severity, dict):
change_conditions = severity.get("changeConditions")
if (
isinstance(change_conditions, list)
and change_conditions
and all(
isinstance(condition, str) and condition.strip()
for condition in change_conditions
)
):
for condition_index, condition in enumerate(change_conditions):
_require_safe_json_string(
condition,
f"{context}.severity.changeConditions[{condition_index}]",
)
severity["changeConditions"] = " ".join(
condition.strip() for condition in change_conditions
)
normalized_fields.append("severity change conditions")
_populate_unsealed_finding_identities(
manifest,
{"scanId": scan_id, "findings": [finding]},
)
finding_id = finding["findingId"]
previous_position = finding_positions.get(finding_id)
_validate_finding(finding, context)
if "writeup" in finding:
try:
_validate_schema_node(finding["writeup"], writeup_schema, f"{context}.writeup")
report_path = finding["writeup"]["reportPath"]
previous_writeup = (
recovered[previous_position].get("writeup")
if previous_position is not None
else None
)
if report_path in writeup_paths and (
previous_writeup is None or previous_writeup["reportPath"] != report_path
):
raise ContractError(f"{context}.writeup.reportPath: duplicate report path")
_require_scan_local_file(scan_dir, report_path, f"{context}.writeup.reportPath")
except ContractError as exc:
finding.pop("writeup")
warnings.append(f"Skipped malformed writeup for finding {index + 1}: {exc}.")
for auxiliary, auxiliary_schema in auxiliary_schemas.items():
if auxiliary not in finding:
continue
try:
_validate_schema_node(
finding[auxiliary], auxiliary_schema, f"{context}.{auxiliary}"
)
except ContractError as exc:
finding.pop(auxiliary)
warnings.append(
f"Skipped malformed {auxiliary} for finding {index + 1}: {exc}."
)
_validate_schema_node(finding, finding_schema, context)
except ContractError as exc:
warning = f"Skipped malformed finding {index + 1}: {exc}."
warnings.append(warning)
discarded.append(warning)
continue
if previous_position is not None:
previous = recovered[previous_position]
strongest, earlier = (
(previous, finding)
if _finding_strength(finding) <= _finding_strength(previous)
else (finding, previous)
)
if strongest != earlier:
original = copy.deepcopy(earlier)
older = original["provenance"].pop("previousFindings", [])
if not isinstance(strongest["provenance"].get("previousFindings"), list):
strongest["provenance"]["previousFindings"] = []
history = strongest["provenance"]["previousFindings"]
for record in [*(older if isinstance(older, list) else []), original]:
if record not in history:
history.append(record)
if _finding_strength(finding) <= _finding_strength(previous):
warnings.append(
f"Skipped malformed finding {index + 1}: duplicate logical finding."
)
continue
previous_writeup = previous.get("writeup")
if previous_writeup is not None:
writeup_paths.discard(previous_writeup["reportPath"])
recovered[previous_position] = finding
warnings.append(
f"Recovered finding {index + 1}: retained stronger duplicate logical finding."
)
else:
finding_positions[finding_id] = len(recovered)
recovered.append(finding)
if "writeup" in finding:
writeup_paths.add(finding["writeup"]["reportPath"])
if normalized_fields:
warnings.append(
f"Recovered finding {index + 1}: normalized {', '.join(normalized_fields)}."
)
findings["findings"] = recovered
return discarded
def _recover_unsealed_coverage(
coverage: dict[str, Any],
schema_dir: Path,
scan_dir: Path,
warnings: list[str],
discarded_findings: list[str],
) -> None:
schema = _read_json(schema_dir / "coverage.schema.json")
properties = _require_dict(schema, "properties", "coverage.schema")
completeness = coverage.get("completeness")
partial = completeness not in ("complete", "partial", "unknown")
if partial:
warnings.append("Recovered malformed coverage completeness; marked coverage as partial.")
if (
coverage.get("mode") == "deep_repository"
and coverage.get("inventoryStrategy") != "repository"
):
coverage["inventoryStrategy"] = "repository"
warnings.append(
"Recovered malformed Deep Scan inventory strategy; marked coverage as partial."
)
partial = True
surface_ids: set[str] = set()
for field, label in (
("surfaces", "coverage surface"),
("explicitExclusions", "coverage exclusion"),
("deferred", "deferred coverage item"),
):
array_schema = _require_dict(properties, field, "coverage.schema.properties")
item_schema = _require_dict(array_schema, "items", f"coverage.schema.properties.{field}")
items = coverage.get(field)
if not isinstance(items, list):
warnings.append(f"Skipped malformed {label} records: expected an array.")
coverage[field] = []
partial = True
continue
recovered: list[dict[str, Any]] = []
for index, item in enumerate(items):
context = f"coverage.{field}[{index}]"
try:
if not isinstance(item, dict):
raise ContractError(f"{context}: expected an object")
if field == "surfaces":
surface_id = _require_str(item, "id", context)
if surface_id in surface_ids:
raise ContractError(f"{context}.id: duplicate surface id")
disposition = item.get("disposition")
surface_recovered = False
if not isinstance(disposition, str) or disposition not in DISPOSITIONS:
warnings.append(
f"Recovered coverage surface {index + 1}: "
"the review disposition could not be verified."
)
item["disposition"] = "needs_follow_up"
surface_recovered = True
receipt_refs = item.get("receiptRefs")
if not isinstance(receipt_refs, list):
warnings.append(
f"Skipped malformed receipt references for coverage surface "
f"{index + 1}: expected an array."
)
receipt_refs = []
surface_recovered = True
recovered_receipts: list[str] = []
for ref_index, ref in enumerate(receipt_refs):
ref_context = f"{context}.receiptRefs[{ref_index}]"
try:
if not isinstance(ref, str):
raise ContractError(f"{ref_context}: expected a string")
normalized_ref = _require_portable_relative_path(ref, ref_context)
if not normalized_ref.startswith("artifacts/"):
raise ContractError(
f"{ref_context}: expected a file under artifacts/"
)
_require_scan_local_file(scan_dir, normalized_ref, ref_context)
except ContractError as exc:
warnings.append(
f"Skipped malformed coverage receipt "
f"{index + 1}.{ref_index + 1}: {exc}."
)
surface_recovered = True
continue
recovered_receipts.append(normalized_ref)
item["receiptRefs"] = recovered_receipts
if surface_recovered or item["disposition"] == "needs_follow_up":
if not surface_recovered and completeness != "partial":
warnings.append(
f"Coverage surface {index + 1} requires follow-up; "
"marked coverage as partial."
)
item["disposition"] = "needs_follow_up"
partial = True
_validate_schema_node(item, item_schema, context)
except ContractError as exc:
warnings.append(f"Skipped malformed {label} {index + 1}: {exc}.")
partial = True
continue
if field == "surfaces":
surface_ids.add(surface_id)
recovered.append(item)
coverage[field] = recovered
if discarded_findings:
for surface in coverage["surfaces"]:
surface["disposition"] = "needs_follow_up"
coverage["deferred"].extend(
{"id": f"discarded-finding-{index}", "reason": warning}
for index, warning in enumerate(discarded_findings, 1)
)
partial = True
if coverage["deferred"] and completeness != "partial":
if not discarded_findings:
warnings.append("Coverage has deferred review work; marked coverage as partial.")
partial = True
if partial:
coverage["completeness"] = "partial"
def _recover_unsealed_hardening(
manifest: dict[str, Any],
scan_dir: Path,
warnings: list[str],
) -> None:
scan = _require_dict(manifest, "scan", "manifest")
if "hardening" not in scan:
return
try:
hardening = _require_dict(scan, "hardening", "manifest.scan")
portfolio_path = _require_str(hardening, "portfolioPath", "manifest.scan.hardening")
if portfolio_path != "hardening/hardening.md":
raise ContractError(
"manifest.scan.hardening.portfolioPath: expected hardening/hardening.md"
)
_require_hardening_portfolio_file(scan_dir, scan)
except ContractError as exc:
scan.pop("hardening")
warnings.append(f"Skipped malformed hardening portfolio: {exc}.")
def _validate_derived_finding_identities(
manifest: dict[str, Any],
findings: dict[str, Any],
) -> None:
"""Require sealed finding identities to equal their deterministic values."""
for context, finding, finding_id, occurrence_id, fingerprints in _derived_finding_identity_rows(
manifest,
findings,
):
if finding.get("findingId") != finding_id:
raise ContractError(f"{context}.findingId: does not match derived fingerprint identity")
if finding.get("occurrenceId") != occurrence_id:
raise ContractError(f"{context}.occurrenceId: does not match scan occurrence identity")
if finding.get("fingerprints") != fingerprints:
raise ContractError(f"{context}.fingerprints: does not match derived fingerprint")
def _populate_unsealed_manifest_envelope(
manifest: dict[str, Any],
scan: dict[str, Any],
completion_binding: dict[str, Any] | None,
) -> None:
"""Populate non-semantic draft fields owned by finalization or the workbench."""
manifest["documentType"] = "codex-security.scan-manifest"
manifest["schemaVersion"] = SCHEMA_VERSION
scan["status"] = (
completion_binding.get("status", "completed") if completion_binding else "completed"
)
scan["coverageRef"] = "coverage.json"
scan["findingsRef"] = "findings.json"
if completion_binding is None:
started_at = os.environ.get("CODEX_SECURITY_STARTED_AT")
if started_at is not None:
_validate_date_time(started_at, "CODEX_SECURITY_STARTED_AT")
scan["startedAt"] = started_at
scan["completedAt"] = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
return
scan["id"] = completion_binding["scanId"]
scan["startedAt"] = completion_binding["startedAt"]
scan["completedAt"] = completion_binding["completedAt"]
scan["producer"] = copy.deepcopy(completion_binding["producer"])
target = scan.get("target")
if isinstance(target, dict):
_populate_unsealed_target_binding(target, completion_binding["target"])
scope = scan.get("scope")
if isinstance(scope, dict):
scope.update(copy.deepcopy(completion_binding["scope"]))
def _populate_unsealed_target_binding(
target: dict[str, Any],
target_binding: dict[str, Any],
) -> None:
"""Replace workbench-owned target coordinates without retaining incompatible drafts."""
target_kind = target.get("kind")
required_coordinates = (
TARGET_REQUIRED_COORDINATE_FIELDS.get(target_kind, set())
if isinstance(target_kind, str)
else set()
)
for field in TARGET_COORDINATE_FIELDS:
if field not in target_binding and field not in required_coordinates:
target.pop(field, None)
target.update(copy.deepcopy(target_binding))
def _populate_unsealed_artifact_envelope(
manifest: dict[str, Any],
findings: dict[str, Any],
coverage: dict[str, Any],
completion_binding: dict[str, Any] | None,
) -> None:
"""Populate deterministic top-level draft fields after refs have been resolved."""
findings["documentType"] = "codex-security.findings"
findings["schemaVersion"] = SCHEMA_VERSION
coverage["documentType"] = "codex-security.coverage"
coverage["schemaVersion"] = SCHEMA_VERSION
if completion_binding is None:
return
scan_id = completion_binding["scanId"]
findings["scanId"] = scan_id
coverage["scanId"] = scan_id
coverage["mode"] = completion_binding["coverageMode"]
scan = _require_dict(manifest, "scan", "manifest")
scope = _require_dict(scan, "scope", "manifest.scan")
if "includePaths" in scope:
coverage["includePaths"] = copy.deepcopy(scope["includePaths"])
if "excludePaths" in scope:
coverage["excludePaths"] = copy.deepcopy(scope["excludePaths"])
def _normalize_unsealed_open_questions(coverage: dict[str, Any]) -> None:
"""Keep only schema-valid optional open-question rows without inventing content."""
open_questions = coverage.get("openQuestions")
if not isinstance(open_questions, list):
coverage.pop("openQuestions", None)
return
normalized: list[Any] = []
for item in open_questions:
if isinstance(item, str):
question = item.strip()
if question:
normalized.append({"question": question})
continue
if not isinstance(item, dict):
continue
question = item.get("question")
if not isinstance(question, str) or not question.strip():
continue
row = dict(item)
row["question"] = question.strip()
follow_up = row.get("followUpPrompt")
if not isinstance(follow_up, str) or not follow_up.strip():
row.pop("followUpPrompt", None)
normalized.append(row)
coverage["openQuestions"] = normalized
def _normalize_unsealed_deep_repository_inventory_strategy(
coverage: dict[str, Any],
*,
expected_coverage_mode: str | None,
) -> None:
"""Label whole-repository Deep scans as using the repository inventory."""
if expected_coverage_mode == "deep_repository":
coverage["inventoryStrategy"] = "repository"
def _validate_completion_binding(
manifest: dict[str, Any],
findings: dict[str, Any],
coverage: dict[str, Any],
completion_binding: dict[str, Any] | None,
) -> None:
"""Verify populated workbench-owned fields before an unsealed draft is written."""
if completion_binding is None:
return
scan = _require_dict(manifest, "scan", "manifest")
if scan.get("id") != completion_binding["scanId"]:
raise ContractError("manifest.scan.id: must match the workbench scan")
if scan.get("status") != completion_binding.get("status", "completed"):
raise ContractError("manifest.scan.status: must match the workbench outcome")
if scan.get("startedAt") != completion_binding["startedAt"]:
raise ContractError("manifest.scan.startedAt: must match the workbench scan")
if scan.get("completedAt") != completion_binding["completedAt"]:
raise ContractError("manifest.scan.completedAt: must match the workbench completion")
if scan.get("producer") != completion_binding["producer"]:
raise ContractError("manifest.scan.producer: must match the workbench producer")
target = _require_dict(scan, "target", "manifest.scan")
allowed_target_kinds = completion_binding["allowedTargetKinds"]
if target.get("kind") not in allowed_target_kinds:
raise ContractError("scan.target.kind: must match the workbench target")
for key, expected in completion_binding["target"].items():
if target.get(key) != expected:
raise ContractError(f"scan.target.{key}: must match the workbench target")
scope = _require_dict(scan, "scope", "manifest.scan")
for key, expected in completion_binding["scope"].items():
if scope.get(key) != expected:
raise ContractError(f"manifest.scan.scope.{key}: must match the workbench scan")
if findings.get("scanId") != completion_binding["scanId"]:
raise ContractError("findings.scanId: must match the workbench scan")
if coverage.get("scanId") != completion_binding["scanId"]:
raise ContractError("coverage.scanId: must match the workbench scan")
if coverage.get("mode") != completion_binding["coverageMode"]:
raise ContractError("coverage.mode: must match the workbench scan")
for key, expected in completion_binding["scope"].items():
if coverage.get(key) != expected:
raise ContractError(f"coverage.{key}: must match the workbench scan")
def _validate_finding(finding: dict[str, Any], context: str) -> None:
for key in ("findingId", "occurrenceId", "ruleId", "title", "summary", "remediation"):
_require_str(finding, key, context)
_require_dict(finding, "identity", context)
fingerprints = _require_dict(finding, "fingerprints", context)
if fingerprints.get("algorithm") != FINGERPRINT_ALGORITHM:
raise ContractError(f"{context}.fingerprints.algorithm: unsupported algorithm")
_require_str(fingerprints, "primary", f"{context}.fingerprints")
severity = _require_dict(finding, "severity", context)
level = _require_str(severity, "level", f"{context}.severity")
if level not in SEVERITIES:
raise ContractError(f"{context}.severity.level: unsupported severity: {level}")
score = severity.get("score")
if score is not None:
if not isinstance(score, (int, float)) or isinstance(score, bool) or not 0 <= score <= 10:
raise ContractError(f"{context}.severity.score: expected a number from 0 through 10")
_require_str(severity, "scoringSystem", f"{context}.severity")
confidence = _require_dict(finding, "confidence", context)
confidence_level = _require_str(confidence, "level", f"{context}.confidence")
if confidence_level not in CONFIDENCES:
raise ContractError(
f"{context}.confidence.level: unsupported confidence: {confidence_level}"
)
_require_str(confidence, "rationale", f"{context}.confidence")
taxonomy = _require_dict(finding, "taxonomy", context)
_require_str(taxonomy, "category", f"{context}.taxonomy")
cwe = taxonomy.get("cwe", [])
if not isinstance(cwe, list) or any(not isinstance(item, str) or not item for item in cwe):
raise ContractError(f"{context}.taxonomy.cwe: expected an array of strings")
locations = _require_list(finding, "locations", context)
if not locations:
raise ContractError(f"{context}.locations: expected at least one location")
for index, location in enumerate(locations):
if not isinstance(location, dict):
raise ContractError(f"{context}.locations[{index}]: expected an object")
_validate_location(location, f"{context}.locations[{index}]")
evidence_ids: set[str] = set()
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}.{evidence_key}: expected an array")
for index, evidence in enumerate(code_evidence):
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)
if evidence_id in evidence_ids:
raise ContractError(f"{evidence_context}.id: duplicate code-evidence id")
evidence_ids.add(evidence_id)
_require_str(evidence, "code", evidence_context)
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
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")
extensions = finding.get("extensions")
if extensions is not None and not isinstance(extensions, dict):
raise ContractError(f"{context}.extensions: expected an object")
def _validate_coverage(manifest: dict[str, Any], coverage: dict[str, Any], scan_dir: Path) -> None:
scan = _require_dict(manifest, "scan", "manifest")
scan_id = _require_str(scan, "id", "manifest.scan")
if coverage.get("scanId") != scan_id:
raise ContractError("coverage.scanId: must match manifest scan id")
_require_str(coverage, "mode", "coverage")
completeness = _require_str(coverage, "completeness", "coverage")
_require_str(coverage, "inventoryStrategy", "coverage")
scope = _require_dict(scan, "scope", "manifest.scan")
if coverage.get("includePaths") != scope.get("includePaths"):
raise ContractError("coverage.includePaths: must match manifest scope")
if coverage.get("excludePaths") != scope.get("excludePaths"):
raise ContractError("coverage.excludePaths: must match manifest scope")
surface_ids: set[str] = set()
has_needs_follow_up = False
for index, surface in enumerate(_require_list(coverage, "surfaces", "coverage")):
context = f"coverage.surfaces[{index}]"
if not isinstance(surface, dict):
raise ContractError(f"{context}: expected an object")
surface_id = _require_str(surface, "id", context)
if surface_id in surface_ids:
raise ContractError(f"{context}.id: duplicate surface id")
surface_ids.add(surface_id)
_require_str(surface, "label", context)
disposition = _require_str(surface, "disposition", context)
if disposition not in DISPOSITIONS:
raise ContractError(f"{context}.disposition: unsupported disposition: {disposition}")
has_needs_follow_up = has_needs_follow_up or disposition == "needs_follow_up"
receipt_refs = surface.get("receiptRefs", [])
if not isinstance(receipt_refs, list):
raise ContractError(f"{context}.receiptRefs: expected an array")
for ref_index, ref in enumerate(receipt_refs):
if not isinstance(ref, str):
raise ContractError(f"{context}.receiptRefs[{ref_index}]: expected a string")
normalized_ref = _require_portable_relative_path(
ref, f"{context}.receiptRefs[{ref_index}]"
)
if not normalized_ref.startswith("artifacts/"):
raise ContractError(
f"{context}.receiptRefs[{ref_index}]: expected a file under artifacts/"
)
receipt_refs[ref_index] = normalized_ref
_require_scan_local_file(
scan_dir, normalized_ref, f"{context}.receiptRefs[{ref_index}]"
)
for field in ("explicitExclusions", "deferred"):
if not isinstance(coverage.get(field, []), list):
raise ContractError(f"coverage.{field}: expected an array")
if completeness == "complete" and (has_needs_follow_up or coverage.get("deferred")):
raise ContractError("coverage.completeness: complete coverage cannot have deferred work")
_require_safe_json_value(coverage, "coverage.json")
def _validate_manifest(manifest: dict[str, Any]) -> None:
if manifest.get("documentType") != "codex-security.scan-manifest":
raise ContractError("manifest.documentType: expected codex-security.scan-manifest")
if manifest.get("schemaVersion") != SCHEMA_VERSION:
raise ContractError(f"manifest.schemaVersion: expected {SCHEMA_VERSION}")
scan = _require_dict(manifest, "scan", "manifest")
for key in ("id", "startedAt", "completedAt", "sealedAt"):
_require_str(scan, key, "manifest.scan")
if scan.get("status") not in {"completed", "failed", "canceled", "interrupted"}:
raise ContractError("manifest.scan.status: expected a terminal scan outcome")
producer = _require_dict(scan, "producer", "manifest.scan")
_require_str(producer, "name", "manifest.scan.producer")
_require_str(producer, "version", "manifest.scan.producer")
_validate_target(_require_dict(scan, "target", "manifest.scan"))
scope = _require_dict(scan, "scope", "manifest.scan")
for field in ("includePaths", "excludePaths"):
values = _require_list(scope, field, "manifest.scan.scope")
for index, value in enumerate(values):
if not isinstance(value, str):
raise ContractError(f"manifest.scan.scope.{field}[{index}]: expected a string")
_require_safe_relative_path(
value, f"manifest.scan.scope.{field}[{index}]", allow_dot=True
)
_validate_contract_refs(scan)
artifacts = _require_list(scan, "artifacts", "manifest.scan")
if not artifacts:
raise ContractError("manifest.scan.artifacts: expected generated artifact records")
artifact_paths: set[str] = set()
artifact_collision_keys: set[str] = set()
for index, artifact in enumerate(artifacts):
context = f"manifest.scan.artifacts[{index}]"
if not isinstance(artifact, dict):
raise ContractError(f"{context}: expected an object")
path = _require_portable_relative_path(
_require_str(artifact, "path", context), f"{context}.path"
)
collision_key = path.lower()
if collision_key in artifact_collision_keys:
raise ContractError(f"{context}.path: duplicate artifact path")
artifact_paths.add(path)
artifact_collision_keys.add(collision_key)
_require_str(artifact, "sha256", context)
_require_str(artifact, "mediaType", context)
for required_path in ("findings.json", "coverage.json"):
if required_path not in artifact_paths:
raise ContractError(
f"manifest.scan.artifacts: missing required artifact: {required_path}"
)
_require_safe_json_value(manifest, "scan-manifest.json")
def _validate_findings(manifest: dict[str, Any], findings: dict[str, Any]) -> None:
if findings.get("documentType") != "codex-security.findings":
raise ContractError("findings.documentType: expected codex-security.findings")
if findings.get("schemaVersion") != SCHEMA_VERSION:
raise ContractError(f"findings.schemaVersion: expected {SCHEMA_VERSION}")
scan_id = _require_str(_require_dict(manifest, "scan", "manifest"), "id", "manifest.scan")
if findings.get("scanId") != scan_id:
raise ContractError("findings.scanId: must match manifest scan id")
finding_ids: set[str] = set()
occurrence_ids: set[str] = set()
for index, finding in enumerate(_require_list(findings, "findings", "findings")):
context = f"findings.findings[{index}]"
if not isinstance(finding, dict):
raise ContractError(f"{context}: expected an object")
_validate_finding(finding, context)
finding_id = str(finding["findingId"])
occurrence_id = str(finding["occurrenceId"])
if finding_id in finding_ids or occurrence_id in occurrence_ids:
raise ContractError(f"{context}: duplicate finding or occurrence id")
finding_ids.add(finding_id)
occurrence_ids.add(occurrence_id)
_require_safe_json_value(findings, "findings.json")
def _schema_type_matches(value: Any, expected: str) -> bool:
return {
"array": isinstance(value, list),
"boolean": isinstance(value, bool),
"integer": isinstance(value, int) and not isinstance(value, bool),
"number": isinstance(value, (int, float)) and not isinstance(value, bool),
"object": isinstance(value, dict),
"string": isinstance(value, str),
"null": value is None,
}[expected]
def _schema_values_equal(left: Any, right: Any) -> bool:
if (
isinstance(left, (int, float))
and not isinstance(left, bool)
and isinstance(right, (int, float))
and not isinstance(right, bool)
):
return left == right
if type(left) is not type(right):
return False
if isinstance(left, dict):
return left.keys() == right.keys() and all(
_schema_values_equal(left[key], right[key]) for key in left
)
if isinstance(left, list):
return len(left) == len(right) and all(
_schema_values_equal(left_item, right_item)
for left_item, right_item in zip(left, right, strict=True)
)
return left == right
def _resolve_schema_reference(
root_schema: dict[str, Any], reference: str, context: str
) -> dict[str, Any]:
if reference == "#":
return root_schema
if not reference.startswith("#/"):
raise ContractError(f"{context}: unsupported schema reference {reference!r}")
target: Any = root_schema
for raw_part in reference[2:].split("/"):
part = raw_part.replace("~1", "/").replace("~0", "~")
if not isinstance(target, dict) or part not in target:
raise ContractError(f"{context}: unresolved schema reference {reference!r}")
target = target[part]
if not isinstance(target, dict):
raise ContractError(f"{context}: schema reference {reference!r} is not an object")
return target
def _validate_schema_node(
value: Any,
schema: dict[str, Any],
context: str,
root_schema: dict[str, Any] | None = None,
) -> None:
root_schema = schema if root_schema is None else root_schema
reference = schema.get("$ref")
if reference is not None:
if not isinstance(reference, str):
raise ContractError(f"{context}: schema reference must be a string")
_validate_schema_node(
value,
_resolve_schema_reference(root_schema, reference, context),
context,
root_schema,
)
expected = schema.get("type")
if isinstance(expected, list):
if not any(_schema_type_matches(value, item) for item in expected):
raise ContractError(f"{context}: does not match schema type {expected}")
elif isinstance(expected, str) and not _schema_type_matches(value, expected):
raise ContractError(f"{context}: expected schema type {expected}")
if "const" in schema and not _schema_values_equal(value, schema["const"]):
raise ContractError(f"{context}: expected {schema['const']!r}")
if "enum" in schema and not any(
_schema_values_equal(value, candidate) for candidate in schema["enum"]
):
raise ContractError(f"{context}: unsupported value {value!r}")
if isinstance(value, str):
if schema.get("minLength", 0) and len(value) < schema["minLength"]:
raise ContractError(f"{context}: string is too short")
if "pattern" in schema and not re.fullmatch(schema["pattern"], value):
raise ContractError(f"{context}: string does not match schema pattern")
if schema.get("format") == "date-time":
_validate_date_time(value, context)
if isinstance(value, (int, float)) and not isinstance(value, bool):
if "minimum" in schema and value < schema["minimum"]:
raise ContractError(f"{context}: value is below schema minimum")
if "maximum" in schema and value > schema["maximum"]:
raise ContractError(f"{context}: value is above schema maximum")
if isinstance(value, list):
if "minItems" in schema and len(value) < schema["minItems"]:
raise ContractError(f"{context}: array has too few items")
if "maxItems" in schema and len(value) > schema["maxItems"]:
raise ContractError(f"{context}: array has too many items")
if schema.get("uniqueItems") is True:
for index, item in enumerate(value):
if any(_schema_values_equal(item, candidate) for candidate in value[:index]):
raise ContractError(f"{context}: array items must be unique")
contains = schema.get("contains")
if isinstance(contains, dict):
matches = 0
for item in value:
try:
_validate_schema_node(item, contains, context, root_schema)
except ContractError:
pass
else:
matches += 1
if matches < schema.get("minContains", 1):
raise ContractError(f"{context}: array contains too few matching items")
if "maxContains" in schema and matches > schema["maxContains"]:
raise ContractError(f"{context}: array contains too many matching items")
item_schema = schema.get("items")
if isinstance(item_schema, dict):
for index, item in enumerate(value):
_validate_schema_node(item, item_schema, f"{context}[{index}]", root_schema)
if isinstance(value, dict):
for item_schema in schema.get("allOf", []):
_validate_schema_node(value, item_schema, context, root_schema)
condition = schema.get("if")
if isinstance(condition, dict):
try:
_validate_schema_node(value, condition, context, root_schema)
except ContractError:
pass
else:
then_schema = schema.get("then")
if isinstance(then_schema, dict):
_validate_schema_node(value, then_schema, context, root_schema)
for key in schema.get("required", []):
if key not in value:
raise ContractError(f"{context}.{key}: missing required schema property")
if "minProperties" in schema and len(value) < schema["minProperties"]:
raise ContractError(f"{context}: object has too few properties")
properties = schema.get("properties", {})
additional_properties = schema.get("additionalProperties", True)
for key, item in value.items():
item_schema = properties.get(key)
if isinstance(item_schema, dict):
_validate_schema_node(item, item_schema, f"{context}.{key}", root_schema)
elif additional_properties is False:
raise ContractError(f"{context}.{key}: unexpected schema property")
elif isinstance(additional_properties, dict):
_validate_schema_node(
item,
additional_properties,
f"{context}.{key}",
root_schema,
)
def validate_against_schema(payload: dict[str, Any], schema_path: Path) -> None:
schema = _read_json(schema_path)
_validate_schema_node(payload, schema, schema_path.stem, schema)
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 (
("rootCause", ("evidenceRefs", "evidence_refs")),
("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],
coverage: dict[str, Any],
schema_dir: Path,
) -> None:
provisional_manifest = copy.deepcopy(manifest)
provisional_scan = _require_dict(provisional_manifest, "scan", "manifest")
provisional_scan["artifacts"] = [
{"path": "findings.json", "sha256": "0" * 64, "mediaType": "application/json"},
{"path": "coverage.json", "sha256": "0" * 64, "mediaType": "application/json"},
]
validate_against_schema(provisional_manifest, schema_dir / "scan-manifest.schema.json")
validate_against_schema(findings, schema_dir / "findings.schema.json")
validate_against_schema(coverage, schema_dir / "coverage.schema.json")
def _validate_contract_refs(scan: dict[str, Any]) -> None:
for field, expected in (
("coverageRef", "coverage.json"),
("findingsRef", "findings.json"),
):
actual = _require_str(scan, field, "manifest.scan")
if actual != expected:
raise ContractError(f"manifest.scan.{field}: expected {expected!r}")
def _sarif_label(value: str) -> str:
words = re.sub(r"[-_./]+", " ", value).split()
acronyms = {"api", "csrf", "html", "http", "id", "rce", "sql", "ssrf", "url", "xml", "xss"}
label = " ".join(word.upper() if word.lower() in acronyms else word for word in words)
return label[:1].upper() + label[1:]
def _sarif_rule(rule_id: str, findings: list[dict[str, Any]]) -> dict[str, Any]:
name = ": ".join(_sarif_label(part) for part in rule_id.split("."))
categories = sorted({finding["taxonomy"]["category"] for finding in findings})
cwes = sorted({cwe for finding in findings for cwe in finding["taxonomy"]["cwe"]})
tags = {"security", *categories}
for cwe in cwes:
match = re.fullmatch(r"CWE-([0-9]+)", cwe, re.IGNORECASE)
if match and int(match[1]) > 0:
tags.add(f"external/cwe/cwe-{int(match[1]):03d}")
description = f"{name}. Categories: {', '.join(map(_sarif_label, categories))}."
if cwes:
description += f" Weaknesses: {', '.join(cwes)}."
remediation = "\n\n".join(sorted({finding["remediation"] for finding in findings}))
properties: dict[str, Any] = {"tags": sorted(tags)}
# GitHub assigns security severity to the shared rule, not each result.
score = max(
finding["severity"].get("score", SARIF_SECURITY_SCORES[finding["severity"]["level"]])
for finding in findings
)
if score > 0:
properties["security-severity"] = str(score)
return {
"id": rule_id,
"name": name,
"shortDescription": {"text": name},
"fullDescription": {"text": description},
"help": {
"text": f"{description}\n\nRemediation:\n\n{remediation}",
"markdown": f"{description}\n\n## Remediation\n\n{remediation}",
},
"properties": properties,
}
def _sarif_finding_message(finding: dict[str, Any]) -> str:
taxonomy = finding["taxonomy"]
details = [
finding["title"],
finding["summary"],
f"Severity: {finding['severity']['level']}",
f"Category: {_sarif_label(taxonomy['category'])}",
]
if taxonomy["cwe"]:
details.append(f"Weaknesses: {', '.join(taxonomy['cwe'])}")
details.append(f"Remediation:\n{finding['remediation']}")
for key, label in (
("remediationTests", "Remediation tests"),
("preventiveControls", "Preventive controls"),
):
if finding.get(key):
details.append(f"{label}:\n" + "\n".join(f"- {item}" for item in finding[key]))
return "\n\n".join(details)
def _utf16_code_units(value: str) -> Iterator[int]:
encoded = value.encode("utf-16-le")
for index in range(0, len(encoded), 2):
yield int.from_bytes(encoded[index : index + 2], "little")
def _github_line_hashes(
handle: TextIO,
requested_lines: set[int] | None = None,
) -> dict[int, str]:
window = [0] * GITHUB_HASH_BLOCK_SIZE
line_numbers = [-1] * GITHUB_HASH_BLOCK_SIZE
hash_counts: dict[str, int] = {}
hashes: dict[int, str] = {}
first_mod = pow(GITHUB_HASH_MOD, GITHUB_HASH_BLOCK_SIZE, 1 << 64)
hash_raw = 0
index = 0
line_number = 0
line_start = True
previous_was_cr = False
def output_hash() -> None:
nonlocal index
hash_value = format(hash_raw & GITHUB_HASH_MASK, "x")
hash_counts[hash_value] = hash_counts.get(hash_value, 0) + 1
line_number = line_numbers[index]
if requested_lines is None or line_number in requested_lines:
hashes[line_number] = f"{hash_value}:{hash_counts[hash_value]}"
line_numbers[index] = -1
def update_hash(current: int) -> None:
nonlocal hash_raw, index
beginning = window[index]
window[index] = current
hash_raw = (GITHUB_HASH_MOD * hash_raw + current - first_mod * beginning) & GITHUB_HASH_MASK
index = (index + 1) % GITHUB_HASH_BLOCK_SIZE
def process_character(current: int) -> None:
nonlocal line_number, line_start, previous_was_cr
if current in {ord(" "), ord("\t")} or (previous_was_cr and current == ord("\n")):
previous_was_cr = False
return
if current == ord("\r"):
current = ord("\n")
previous_was_cr = True
else:
previous_was_cr = False
if line_numbers[index] != -1:
output_hash()
if line_start:
line_start = False
line_number += 1
line_numbers[index] = line_number
if current == ord("\n"):
line_start = True
update_hash(current)
while chunk := handle.read(SOURCE_READ_CHUNK_SIZE):
for code_unit in _utf16_code_units(chunk):
process_character(code_unit)
process_character(GITHUB_HASH_EOF)
for _ in range(GITHUB_HASH_BLOCK_SIZE):
if line_numbers[index] != -1:
output_hash()
update_hash(0)
return hashes
def _open_source_file(source_root: Path, relative_path: str) -> TextIO | None:
file_fd: int | None = None
try:
file_fd = open_scan_local_file_descriptor(
source_root, relative_path, f"source file {relative_path}"
)
handle = os.fdopen(file_fd, "r", encoding="utf-8", errors="replace")
file_fd = None
return handle
except (ContractError, OSError, ValueError):
return None
finally:
if file_fd is not None:
os.close(file_fd)
def _github_line_hashes_for_source(
source_root: Path,
relative_path: str,
requested_lines: set[int] | None = None,
) -> dict[int, str] | None:
handle = _open_source_file(source_root, relative_path)
if handle is None:
return None
try:
with handle:
return _github_line_hashes(handle, requested_lines)
except OSError:
return None
def _sarif_primary_location(finding: dict[str, Any]) -> dict[str, Any]:
return next(
(location for location in finding["locations"] if location.get("role") == "root_control"),
finding["locations"][0],
)
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),
]
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 = (
location["path"],
location["startLine"],
location.get("endLine", location["startLine"]),
)
unique.setdefault(key, location)
return list(unique.values())
def _github_primary_location_line_hash(
finding: dict[str, Any],
source_root: Path | None,
line_hash_cache: dict[tuple[Path, int], str | None] | None = None,
) -> str | None:
if source_root is None:
return None
primary_location = _sarif_primary_location(finding)
try:
source_root = source_root.resolve(strict=True)
except (OSError, RuntimeError):
return None
relative_path = _require_safe_relative_path(primary_location["path"], "SARIF source location")
source_path = source_root / relative_path
start_line = primary_location["startLine"]
cache_key = (source_path, start_line)
if line_hash_cache is not None and cache_key in line_hash_cache:
return line_hash_cache[cache_key]
line_hashes = _github_line_hashes_for_source(source_root, relative_path, {start_line})
line_hash = None if line_hashes is None else line_hashes.get(start_line)
if line_hash_cache is not None:
line_hash_cache[cache_key] = line_hash
return line_hash
def _github_line_hash_cache(
findings: list[dict[str, Any]], source_root: Path | None
) -> dict[tuple[Path, int], str | None]:
if source_root is None:
return {}
try:
source_root = source_root.resolve(strict=True)
except (OSError, RuntimeError):
return {}
requested_lines_by_path: dict[str, set[int]] = {}
for finding in findings:
primary_location = _sarif_primary_location(finding)
relative_path = _require_safe_relative_path(
primary_location["path"], "SARIF source location"
)
requested_lines_by_path.setdefault(relative_path, set()).add(primary_location["startLine"])
line_hash_cache: dict[tuple[Path, int], str | None] = {}
for relative_path, requested_lines in requested_lines_by_path.items():
line_hashes = _github_line_hashes_for_source(source_root, relative_path, requested_lines)
source_path = source_root / relative_path
for line_number in requested_lines:
line_hash_cache[(source_path, line_number)] = (
None if line_hashes is None else line_hashes.get(line_number)
)
return line_hash_cache
def _sarif_location(location: dict[str, Any], location_id: int | None = None) -> dict[str, Any]:
sarif_location: dict[str, Any] = {
"physicalLocation": {
"artifactLocation": {
"uri": quote(location["path"], safe="/"),
},
"region": {
"startLine": location["startLine"],
"endLine": location.get("endLine", location["startLine"]),
},
}
}
if location_id is not None:
sarif_location["id"] = location_id
if location.get("role"):
sarif_location["message"] = {"text": location["role"]}
return sarif_location
def _sarif_result(
finding: dict[str, Any],
rule_index: int,
source_root: Path | None = None,
line_hash_cache: dict[tuple[Path, int], str | None] | None = None,
) -> dict[str, Any]:
properties = {
"category": finding["taxonomy"]["category"],
"confidence": finding["confidence"]["level"],
"findingId": finding["findingId"],
"occurrenceId": finding["occurrenceId"],
"severity": finding["severity"]["level"],
}
extensions = finding.get("extensions")
candidate_id = extensions.get("candidateId") if isinstance(extensions, dict) else None
if isinstance(candidate_id, str) and candidate_id:
properties["candidateId"] = candidate_id
partial_fingerprints = {
"codexSecurity/v1": finding["fingerprints"]["primary"],
}
line_hash = _github_primary_location_line_hash(finding, source_root, line_hash_cache)
if line_hash is not None:
partial_fingerprints["primaryLocationLineHash"] = line_hash
result = {
"ruleId": finding["ruleId"],
"ruleIndex": rule_index,
"level": SARIF_LEVELS[finding["severity"]["level"]],
"message": {"text": _sarif_finding_message(finding)},
"locations": [_sarif_location(location) for location in _sarif_locations(finding)],
"partialFingerprints": partial_fingerprints,
"properties": properties,
}
return result
def build_sarif(
manifest: dict[str, Any], findings: dict[str, Any], source_root: Path | None = None
) -> dict[str, Any]:
scan = manifest["scan"]
target = scan["target"]
ordered_findings = sorted(findings["findings"], key=lambda finding: finding["occurrenceId"])
findings_by_rule: dict[str, list[dict[str, Any]]] = {}
for finding in ordered_findings:
findings_by_rule.setdefault(finding["ruleId"], []).append(finding)
ordered_rule_ids = sorted(findings_by_rule)
rule_index = {rule_id: index for index, rule_id in enumerate(ordered_rule_ids)}
line_hash_cache = _github_line_hash_cache(ordered_findings, source_root)
run: dict[str, Any] = {
"tool": {
"driver": {
"name": "Codex Security",
"version": scan["producer"]["version"],
"rules": [
_sarif_rule(rule_id, findings_by_rule[rule_id]) for rule_id in ordered_rule_ids
],
}
},
"automationDetails": {"id": scan["id"]},
"results": [
_sarif_result(finding, rule_index[finding["ruleId"]], source_root, line_hash_cache)
for finding in ordered_findings
],
"properties": {
"codexSecuritySchemaVersion": manifest["schemaVersion"],
"codexSecurityTargetKind": target["kind"],
},
}
if target["kind"] == "git_revision" and target.get("remote") and target.get("revision"):
run["versionControlProvenance"] = [
{
"repositoryUri": target["remote"],
"revisionId": target["revision"],
}
]
return {
"$schema": SARIF_SCHEMA,
"version": "2.1.0",
"runs": [run],
}
def _validate_sarif(sarif: dict[str, Any]) -> None:
if sarif.get("version") != "2.1.0":
raise ContractError("SARIF: expected version 2.1.0")
runs = sarif.get("runs")
if not isinstance(runs, list) or len(runs) != 1:
raise ContractError("SARIF: expected exactly one run")
run = runs[0]
if not isinstance(run, dict):
raise ContractError("SARIF: expected a run object")
rule_ids = [rule["id"] for rule in run["tool"]["driver"]["rules"]]
for result in run["results"]:
if result["ruleId"] not in rule_ids:
raise ContractError("SARIF: result references an unknown rule")
if not result.get("partialFingerprints"):
raise ContractError("SARIF: result is missing partialFingerprints")
def _artifact_record(
scan_dir: Path, relative_path: str, media_type: str, contents: bytes | None = None
) -> dict[str, str]:
relative_path = _require_portable_relative_path(relative_path, "artifact path")
if contents is not None:
_validate_scan_local_output_path(scan_dir, scan_dir / relative_path, relative_path)
return {
"mediaType": media_type,
"path": relative_path,
"sha256": (
_sha256_bytes(contents)
if contents is not None
else _sha256_scan_local_file(scan_dir, relative_path, relative_path)
),
}
def _coverage_receipt_refs(coverage: dict[str, Any]) -> list[str]:
refs = {ref for surface in coverage["surfaces"] for ref in surface.get("receiptRefs", [])}
return sorted(refs)
def _validate_sealed_coverage_receipts(scan: dict[str, Any], coverage: dict[str, Any]) -> None:
artifact_paths = {
_require_portable_relative_path(artifact["path"], "sealed artifact path")
for artifact in scan["artifacts"]
}
for ref in _coverage_receipt_refs(coverage):
if ref not in artifact_paths:
raise ContractError(f"coverage receipt is missing from sealed artifacts: {ref}")
def _validate_existing_seal(
scan_dir: Path,
scan: dict[str, Any],
*,
artifact_contents: dict[str, bytes] | None = None,
) -> None:
sealed_at = scan.get("sealedAt")
artifacts = scan.get("artifacts")
if sealed_at is None and artifacts is None:
return
if sealed_at != scan.get("completedAt"):
raise ContractError("manifest.scan.sealedAt: must match completedAt")
if not isinstance(artifacts, list) or not artifacts:
raise ContractError("manifest.scan.artifacts: sealed manifest requires artifact records")
artifact_paths: set[str] = set()
artifact_collision_keys: set[str] = set()
for index, artifact in enumerate(artifacts):
context = f"manifest.scan.artifacts[{index}]"
if not isinstance(artifact, dict):
raise ContractError(f"{context}: expected an object")
path = _require_portable_relative_path(
_require_str(artifact, "path", context), f"{context}.path"
)
collision_key = path.lower()
if collision_key in artifact_collision_keys:
raise ContractError(f"{context}.path: duplicate artifact path")
artifact_paths.add(path)
artifact_collision_keys.add(collision_key)
expected_sha256 = _require_str(artifact, "sha256", context)
contents = (artifact_contents or {}).get(path)
actual_sha256 = (
_sha256_bytes(contents)
if contents is not None
else _sha256_scan_local_file(scan_dir, path, context)
)
if actual_sha256 != expected_sha256:
raise ContractError(f"{context}: sealed artifact changed or is missing")
def _read_sealed_scan(
scan_dir: Path, schema_dir: Path | None, required_for: str
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], bytes]:
scan_dir = _require_scan_directory(scan_dir)
schema_dir = schema_dir or Path(__file__).resolve().parent.parent / "schemas"
manifest = _read_scan_local_json(scan_dir, "scan-manifest.json", "scan-manifest.json")
scan = _require_dict(manifest, "scan", "manifest")
_validate_contract_refs(scan)
if scan.get("sealedAt") is None or scan.get("artifacts") is None:
raise ContractError(f"manifest.scan: {required_for} requires a sealed scan")
findings, findings_bytes = _read_scan_local_json_bytes(
scan_dir, scan["findingsRef"], scan["findingsRef"]
)
coverage, coverage_bytes = _read_scan_local_json_bytes(
scan_dir, scan["coverageRef"], scan["coverageRef"]
)
_validate_existing_seal(
scan_dir,
scan,
artifact_contents={
scan["findingsRef"]: findings_bytes,
scan["coverageRef"]: coverage_bytes,
},
)
_validate_manifest(manifest)
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_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
def build_sarif_projection(
scan_dir: Path, source_root: Path | None = None, schema_dir: Path | None = None
) -> dict[str, Any]:
if source_root is not None:
try:
source_root = source_root.resolve(strict=True)
source_root_is_directory = source_root.is_dir()
except (OSError, RuntimeError):
source_root_is_directory = False
if not source_root_is_directory:
raise ContractError("source root: expected an existing directory")
manifest, findings, coverage, _ = _read_sealed_scan(scan_dir, schema_dir, "SARIF projection")
sarif = build_sarif(manifest, findings, source_root)
execution_successful = manifest["scan"]["status"] == "completed"
if not execution_successful or coverage["completeness"] != "complete":
run = sarif["runs"][0]
run["properties"]["codexSecurityCoverageCompleteness"] = coverage["completeness"]
run["invocations"] = [
{
"executionSuccessful": execution_successful,
"toolExecutionNotifications": [
{"level": "warning", "message": {"text": item["reason"]}}
for item in coverage["deferred"]
],
}
]
_validate_sarif(sarif)
return sarif
def write_sarif_projection(
scan_dir: Path, source_root: Path | None = None, schema_dir: Path | None = None
) -> None:
sarif = build_sarif_projection(scan_dir, source_root, schema_dir)
_write_scan_local_json(scan_dir, "exports/results.sarif", sarif)
def write_sarif_output(scan_dir: Path, output: Path, sarif: dict[str, Any]) -> None:
write_export_output(scan_dir, output, "sarif", _json_bytes(sarif))
def csv_cell(value: Any) -> Any:
if isinstance(value, str) and (
value.startswith(("\t", "\r", "\n"))
or value.lstrip().startswith(("=", "+", "-", "@", "＝", "＋", "－", "＠"))
):
return f"'{value}"
return value
def finding_candidate_id(finding: dict[str, Any]) -> str | None:
provenance = finding.get("provenance")
if (
isinstance(provenance, dict)
and isinstance(value := provenance.get("candidateId"), str)
and value.strip()
):
return value
extensions = finding.get("extensions")
if not isinstance(extensions, dict):
return None
return next(
(
value
for field in ("candidateId", "reportId", "ledgerRowId")
if isinstance(value := extensions.get(field), str) and value.strip()
),
None,
)
def build_csv_projection(findings: dict[str, Any], coverage: dict[str, Any]) -> bytes:
output = io.StringIO(newline="")
writer = csv.writer(output)
deep_scan = coverage.get("mode") == "deep_repository" or (
coverage.get("mode") == "scoped_path"
and any(
isinstance(finding.get("extensions"), dict)
and any(
isinstance(finding["extensions"].get(field), str)
and finding["extensions"][field].strip()
for field in ("candidateId", "reportId")
)
for finding in findings["findings"]
)
)
writer.writerow(
(
"occurrence_id",
"finding_id",
*(("candidate_id",) if deep_scan else ()),
"title",
"summary",
"severity",
"confidence",
"status",
"close_reason",
"note",
"remediation",
"path",
"start_line",
"end_line",
)
)
for finding in findings["findings"]:
locations = finding["locations"]
location = next(
(candidate for candidate in locations if candidate.get("role") == "root_control"),
locations[0],
)
writer.writerow(
(
csv_cell(finding["occurrenceId"]),
csv_cell(finding["findingId"]),
*((csv_cell(finding_candidate_id(finding)),) if deep_scan else ()),
csv_cell(finding["title"]),
csv_cell(finding["summary"]),
csv_cell(finding["severity"]["level"]),
csv_cell(finding["confidence"]["level"]),
"open",
"",
"",
csv_cell(finding["remediation"]),
csv_cell(location["path"]),
location["startLine"],
location.get("endLine", location["startLine"]),
)
)
return output.getvalue().encode("utf-8")
def build_findings_export(
scan_dir: Path,
export_format: str,
source_root: Path | None = None,
schema_dir: Path | None = None,
) -> bytes:
if export_format not in EXPORT_PATHS:
raise ContractError(f"unsupported export format: {export_format}")
if export_format == "sarif":
return _json_bytes(build_sarif_projection(scan_dir, source_root, schema_dir))
if source_root is not None:
raise ContractError("source-root is only supported for SARIF exports")
_, findings, coverage, findings_bytes = _read_sealed_scan(
scan_dir, schema_dir, f"{export_format.upper()} export"
)
if export_format == "json":
return findings_bytes
return build_csv_projection(findings, coverage)
def write_export_output(scan_dir: Path, output: Path, export_format: str, contents: bytes) -> None:
if export_format not in EXPORT_PATHS:
raise ContractError(f"unsupported export format: {export_format}")
scan_dir = _require_scan_directory(scan_dir)
output = Path(os.path.abspath(output))
try:
relative_output = output.relative_to(scan_dir).as_posix()
except ValueError:
for ancestor in output.parents:
try:
inside_scan = ancestor.samefile(scan_dir)
except FileNotFoundError:
continue
except OSError as exc:
raise ContractError(
"export output path: unable to inspect output directory"
) from exc
if inside_scan:
if any(parent.is_symlink() for parent in (ancestor, *ancestor.parents)):
raise ContractError(
"export output path: symbolic links cannot alias the scan directory"
) from None
relative_output = output.relative_to(ancestor).as_posix()
break
else:
write_scan_local_bytes(output.parent, output.name, contents, external_name=True)
return
if relative_output != EXPORT_PATHS[export_format]:
raise ContractError(f"{export_format.upper()} output path cannot overwrite a scan artifact")
manifest = _read_scan_local_json(scan_dir, "scan-manifest.json", "scan-manifest.json")
scan = _require_dict(manifest, "scan", "manifest")
artifacts = _require_list(scan, "artifacts", "manifest.scan")
artifact_paths = [
_require_portable_relative_path(
_require_str(artifact, "path", f"manifest.scan.artifacts[{index}]"),
f"manifest.scan.artifacts[{index}].path",
)
for index, artifact in enumerate(artifacts)
if isinstance(artifact, dict)
]
try:
output_metadata = output.stat(follow_symlinks=False)
except FileNotFoundError:
output_metadata = None
except OSError as exc:
raise ContractError(f"{relative_output}: unable to inspect export output") from exc
for artifact_path in artifact_paths:
if artifact_path == relative_output:
raise ContractError(
f"{export_format.upper()} output path cannot overwrite a sealed scan artifact"
)
if output_metadata is None:
continue
descriptor = open_scan_local_file_descriptor(
scan_dir, artifact_path, f"sealed artifact {artifact_path}"
)
try:
artifact_metadata = os.fstat(descriptor)
finally:
os.close(descriptor)
if os.path.samestat(output_metadata, artifact_metadata):
raise ContractError(
f"{export_format.upper()} output path cannot overwrite a sealed scan artifact"
)
write_scan_local_bytes(scan_dir, relative_output, contents)
def _write_sarif_projection_if_possible(
scan_dir: Path, source_root: Path | None = None, schema_dir: Path | None = None
) -> None:
try:
write_sarif_projection(scan_dir, source_root, schema_dir)
except (ContractError, OSError) as error:
print(
f"codex-security: warning: automatic SARIF export failed: {error}. "
"Run `codex-security export <scan-dir> --export-format sarif` to retry.",
file=sys.stderr,
)
PreparedScanFinalization = tuple[
Path,
Path,
dict[str, Any],
dict[str, Any],
dict[str, Any],
bool,
bytes,
]
def _prepare_scan_finalization(
scan_dir: Path,
schema_dir: Path | None = None,
*,
expected_coverage_mode: str | None = None,
completion_binding: dict[str, Any] | None = None,
completion_warnings: list[str] | None = None,
draft_documents: tuple[dict[str, Any], dict[str, Any], dict[str, Any]] | None = None,
) -> PreparedScanFinalization:
"""Read, populate, and validate a scan without writing any output files."""
scan_dir = _require_scan_directory(scan_dir)
schema_dir = schema_dir or Path(__file__).resolve().parent.parent / "schemas"
manifest = (
copy.deepcopy(draft_documents[0])
if draft_documents is not None
else _read_scan_local_json(scan_dir, "scan-manifest.json", "scan-manifest.json")
)
scan = _require_dict(manifest, "scan", "manifest")
if scan.get("sealedAt") is None and scan.get("artifacts") == []:
del scan["artifacts"]
was_sealed = scan.get("sealedAt") is not None or scan.get("artifacts") is not None
if not was_sealed:
_populate_unsealed_manifest_envelope(manifest, scan, completion_binding)
_validate_contract_refs(scan)
if draft_documents is None:
findings, findings_input_bytes = _read_scan_local_json_bytes(
scan_dir, scan["findingsRef"], scan["findingsRef"]
)
coverage, coverage_input_bytes = _read_scan_local_json_bytes(
scan_dir, scan["coverageRef"], scan["coverageRef"]
)
else:
findings, coverage = copy.deepcopy(draft_documents[1:])
findings_input_bytes, coverage_input_bytes = _json_bytes(findings), _json_bytes(coverage)
if not was_sealed:
_populate_unsealed_artifact_envelope(manifest, findings, coverage, completion_binding)
_normalize_unsealed_deep_repository_inventory_strategy(
coverage,
expected_coverage_mode=expected_coverage_mode,
)
_normalize_unsealed_open_questions(coverage)
if manifest.get("schemaVersion") != SCHEMA_VERSION:
raise ContractError(f"manifest.schemaVersion: expected {SCHEMA_VERSION}")
if scan.get("status") not in {"completed", "failed", "canceled", "interrupted"}:
raise ContractError("manifest.scan.status: expected a terminal scan outcome before sealing")
if (
expected_coverage_mode is not None
and completion_binding is not None
and completion_binding["coverageMode"] != expected_coverage_mode
):
raise ContractError("completion binding coverage mode does not match expected mode")
if expected_coverage_mode is not None and coverage.get("mode") != expected_coverage_mode:
raise ContractError(
f"coverage.mode: must match selected scan mode {expected_coverage_mode}"
)
_validate_existing_seal(
scan_dir,
scan,
artifact_contents={
scan["findingsRef"]: findings_input_bytes,
scan["coverageRef"]: coverage_input_bytes,
},
)
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_for_validation)
_validate_derived_finding_identities(manifest, findings)
elif completion_warnings is not None:
discarded_findings = _recover_unsealed_findings(
manifest, findings, schema_dir, scan_dir, completion_warnings
)
_recover_unsealed_coverage(
coverage, schema_dir, scan_dir, completion_warnings, discarded_findings
)
_recover_unsealed_hardening(manifest, scan_dir, completion_warnings)
else:
_populate_unsealed_finding_identities(manifest, findings)
_validate_findings(manifest, findings_for_validation)
_validate_coverage(manifest, coverage, scan_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_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)
return (
scan_dir,
schema_dir,
manifest,
findings,
coverage,
was_sealed,
report_markdown_bytes,
)
findings_bytes = _contract_json_bytes("findings.json", findings)
coverage_bytes = _contract_json_bytes("coverage.json", coverage)
report_markdown_bytes = _generate_report_projection(manifest, findings, coverage)
_validate_report_output_paths(scan_dir)
scan["artifacts"] = [
_artifact_record(scan_dir, "findings.json", "application/json", findings_bytes),
_artifact_record(scan_dir, "coverage.json", "application/json", coverage_bytes),
*[
_artifact_record(scan_dir, ref, "application/octet-stream")
for ref in _coverage_receipt_refs(coverage)
],
]
_validate_sealed_coverage_receipts(scan, coverage)
_validate_manifest(manifest)
validate_against_schema(manifest, schema_dir / "scan-manifest.schema.json")
validate_against_schema(findings_for_validation, schema_dir / "findings.schema.json")
validate_against_schema(coverage, schema_dir / "coverage.schema.json")
_contract_json_bytes("scan-manifest.json", manifest)
return (
scan_dir,
schema_dir,
manifest,
findings,
coverage,
was_sealed,
report_markdown_bytes,
)
def _write_prepared_scan_finalization(
prepared: PreparedScanFinalization,
source_root: Path | None = None,
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
"""Write a previously validated scan finalization result."""
(
scan_dir,
schema_dir,
manifest,
findings,
coverage,
was_sealed,
report_markdown_bytes,
) = prepared
scan = _require_dict(manifest, "scan", "manifest")
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)
return manifest, findings, coverage
_write_scan_local_json(scan_dir, "findings.json", findings)
_write_scan_local_json(scan_dir, "coverage.json", coverage)
write_scan_local_bytes(scan_dir, "report.md", report_markdown_bytes)
_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)
return manifest, findings, coverage
def finalize_scan(
scan_dir: Path,
schema_dir: Path | None = None,
source_root: Path | None = None,
*,
expected_coverage_mode: str | None = None,
completion_binding: dict[str, Any] | 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)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--scan-dir", required=True, type=Path)
parser.add_argument("--schema-dir", type=Path)
parser.add_argument("--source-root", type=Path)
parser.add_argument("--sarif-only", action="store_true")
parser.add_argument("--sarif-output", type=Path)
parser.add_argument("--export-format", choices=sorted(EXPORT_PATHS))
parser.add_argument("--export-output", type=Path)
args = parser.parse_args()
try:
if args.sarif_only and args.export_format is not None:
parser.error("--sarif-only cannot be combined with --export-format")
if args.export_output is not None and args.export_format is None:
parser.error("--export-output requires --export-format")
if args.sarif_output is not None and not args.sarif_only:
parser.error("--sarif-output requires --sarif-only")
if args.export_format is not None:
contents = build_findings_export(
args.scan_dir, args.export_format, args.source_root, args.schema_dir
)
if args.export_output is None:
sys.stdout.buffer.write(contents)
else:
write_export_output(args.scan_dir, args.export_output, args.export_format, contents)
elif args.sarif_only:
sarif = build_sarif_projection(args.scan_dir, args.source_root, args.schema_dir)
if args.sarif_output is None:
sys.stdout.buffer.write(_json_bytes(sarif))
else:
write_sarif_output(args.scan_dir, args.sarif_output, sarif)
else:
finalize_scan(args.scan_dir, args.schema_dir, args.source_root)
except ContractError as exc:
parser.error(str(exc))
return 0
if __name__ == "__main__":
raise SystemExit(main())
SHA-256: 7f191c50ded50600b0a70be1e11b724f0355f166a690312226990d88e895d159