From cb956b9429cc991b8afbc2745bc1fc3bab2080b9 Mon Sep 17 00:00:00 2001 From: Eddi Sanchez Date: Tue, 10 Nov 2026 19:37:10 +0000 Subject: [PATCH] Add SARIF 2.1.0 export for a retained pull-request review. Project a review the caller already holds with the same closed rules as the CLI findings --sarif command, including locations, gaps, and refusals. Co-authored-by: Edmundo Sanchez --- README.md | 27 ++ proofline/__init__.py | 3 + proofline/sarif.py | 585 ++++++++++++++++++++++++++++++++++++++++++ tests/test_sarif.py | 509 ++++++++++++++++++++++++++++++++++++ 4 files changed, 1124 insertions(+) create mode 100644 proofline/sarif.py create mode 100644 tests/test_sarif.py diff --git a/README.md b/README.md index be99a7f..8496b69 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,33 @@ with ProoflineClient() as client: The response names the recorded report and its covered and instrumented line counts. An identical upload returns that report with `created=False`. Refused reports raise `ProoflineError`. Uploading a report does not establish a check verdict. +## SARIF reviews + +`review_sarif` projects one retained pull-request review to SARIF 2.1.0, the same closed projection as `proofline findings --sarif`. Upload the document to GitHub code scanning or any other SARIF 2.1.0 consumer. It keeps each finding's identity, severity, and verified file locations at the reviewed head, and it leaves out World evidence values and the raw API document. + +```python +import json +from pathlib import Path +from proofline import review_sarif + +outcome = client.wait_for_review(organization_id, review_id) +if outcome.review is None: + raise SystemExit(outcome.reason) +Path("proofline.sarif").write_text( + json.dumps( + review_sarif( + outcome.review, + organization_id, + review_id=review_id, + review_url=outcome.review_url, + ) + ), + encoding="utf-8", +) +``` + +A review that is still running, failed, stopped, or incomplete is still projected. Its invocation has `executionSuccessful` false and a notification naming the gap. `SUPERSEDED` is an incomplete result, matching the CLI projection. A deployment or other subject, a head that is not a full commit SHA, an unrecognized severity, or an unrecognized risk domain raises `SarifExportError` and produces no document. + ## MCP organization selection With an MCP OAuth credential approved for several organizations, discover its currently authorized targets with `client.mcp.list_organizations()`. It returns a `KnowledgeResult`; inspect `isError` before reading `structuredContent["data"]["organizations"]`. Each entry contains `organization_id`, `github_login`, and `scopes`. diff --git a/proofline/__init__.py b/proofline/__init__.py index 61bd629..9f7bbe3 100644 --- a/proofline/__init__.py +++ b/proofline/__init__.py @@ -34,6 +34,7 @@ from .client import AsyncProoflineClient, ProoflineClient, ReviewFilters from .exports import ExportDataset, ExportFormat, ExportPage from .outcomes import AssessmentGap, CheckGap, ReviewOutcome +from .sarif import SarifExportError, review_sarif __all__ = [ "AgentSession", @@ -64,10 +65,12 @@ "ReviewFilters", "ReviewOutcome", "ReviewPage", + "review_sarif", "RunNowMode", "RunNowRefusal", "RunNowRequest", "RunNowResponse", + "SarifExportError", "SecretStr", "StartSession", "StopResponse", diff --git a/proofline/sarif.py b/proofline/sarif.py new file mode 100644 index 0000000..70b4cbc --- /dev/null +++ b/proofline/sarif.py @@ -0,0 +1,585 @@ +"""Closed SARIF 2.1.0 projection of one retained pull-request review. + +This matches the CLI's `findings --sarif` projection. It keeps finding identity, +severity, and verified locations at the reviewed head. It does not copy World +evidence values or the raw API document. +""" + +from __future__ import annotations + +import unicodedata +from typing import Any +from uuid import UUID + +from ._generated.models.customer_assessment import CustomerAssessment +from ._generated.models.customer_check import CustomerCheck +from ._generated.models.customer_code_assessment import CustomerCodeAssessment +from ._generated.models.customer_evidence_ref import CustomerEvidenceRef +from ._generated.models.customer_finding import CustomerFinding +from ._generated.models.customer_review_detail import CustomerReviewDetail +from ._generated.types import Unset +from ._transport import VERSION + +_SCHEMA = ( + "https://docs.oasis-open.org/sarif/sarif/v2.1.0/errata01/os/schemas/sarif-schema-2.1.0.json" +) +# domains-v5, the public environment catalog. +_DOMAINS: tuple[str, ...] = ( + "artifact_identity", + "deployment_workflow", + "runtime_configuration", + "network_reachability", + "iam_permissions", + "datastore_compatibility", + "resource_capacity", + "rollout_topology", + "runtime_architecture", + "environment_binding", + "dependency_reachability", + "data_integrity", + "external_side_effects", + "application_security", + "scalability", + "availability_resilience", + "observability", + "source_coverage", +) +# code-categories-v2, the deterministic CODE categories, and committed_secret, +# which recorded findings still carry. +_CODE_DOMAINS: tuple[str, ...] = ( + "logic", + "edge_case", + "error_handling", + "state_transition", + "concurrency", + "resource_lifecycle", + "credential_exposure", + "test_marker", + "removed_assertion", + "debug_residue", + "committed_secret", +) +_COMPLETE_CHECKS = frozenset(("READY", "BLOCKED", "NOT_APPLICABLE")) +_COMPLETE_LIFECYCLES = frozenset(("GATING", "SHADOW", "QUARANTINED", "RETIRED")) +_KNOWN_ASSESSMENTS = frozenset(("PENDING", "RUNNING", "COMPLETED", "INCOMPLETE", "FAILED")) +_WITHHELD_MARKER = "[withheld: environment value]" +_HEX = frozenset("0123456789abcdefABCDEF") +_FORBIDDEN_PATH_CHARACTERS = frozenset( + "\\:\u200e\u200f\u202a\u202b\u202c\u202d\u202e\u2066\u2067\u2068\u2069" +) +_URI_HEX = "0123456789ABCDEF" + + +class SarifExportError(Exception): + """The review cannot be projected to SARIF.""" + + +class _Location: + def __init__( + self, status: str, path: str | None = None, start: int | None = None, end: int | None = None + ) -> None: + self.status = status + self.path = path + self.start = start + self.end = end + + +def review_sarif( + review: CustomerReviewDetail, + organization_id: UUID, + *, + review_id: UUID, + review_url: str, +) -> dict[str, Any]: + """Project one retained pull-request review to a SARIF 2.1.0 document. + + ``review_id`` is the review the caller requested. A different record is + refused, as are a non-pull-request subject, a head that is not a full + commit SHA, an unrecognized severity, and an unrecognized risk domain. + An incomplete or unfinished review is still projected: its invocation is + unsuccessful and names each gap. ``SUPERSEDED`` is an incomplete result, + matching the CLI projection. + """ + + if not isinstance(review, CustomerReviewDetail): + raise TypeError("review must be a CustomerReviewDetail.") + if not isinstance(organization_id, UUID) or not isinstance(review_id, UUID): + raise TypeError("organization_id and review_id must be UUIDs.") + if not isinstance(review_url, str) or not review_url: + raise ValueError("review_url must be the review's address.") + if not isinstance(review.id, UUID): + raise _invalid() + if review.id != review_id: + raise SarifExportError( + f"requested review {review_id}, but the server returned review {review.id}; " + "SARIF export refused" + ) + if not isinstance(review.kind, str): + raise _invalid() + if review.kind != "PULL_REQUEST": + raise SarifExportError( + f"review {review.id} has kind {_one_line(review.kind)}; " + "SARIF export requires a retained pull-request review" + ) + head = _recorded_text(review.head_commit_sha) + if head is None or not _hex_sha(head): + raise SarifExportError( + f"review {review.id} does not record a full commit SHA; " + "SARIF export requires an immutable pull-request head" + ) + if not isinstance(review.status, str): + raise _invalid() + head = head.lower() + if type(review.assessment_count) is not int or not isinstance(review.assessments, list): + raise _invalid() + state = _terminal(review) + code = _usable_code(review) + results: list[dict[str, Any]] = [] + notifications: list[dict[str, Any]] = [] + code_properties: dict[str, Any] | None = None + if code is not None: + if not isinstance(code.id, UUID): + raise _invalid() + for finding in code.findings: + results.append(_finding_result(finding, code, head, code=True)) + reason = _code_gap(code) + reasons = [reason] if reason is not None else [] + for item in reasons: + notifications.append(_notification(f"Code-only: {item}", item, code.id)) + code_properties = { + "id": str(code.id), + "status": _plain_text(code.status), + "baseCommitSha": code.base_commit_sha, + "headCommitSha": code.head_commit_sha, + "gapReasons": reasons, + } + assessments: list[dict[str, Any]] = [] + for assessment in review.assessments: + if not isinstance(assessment, CustomerAssessment): + raise _invalid() + if not isinstance(assessment.id, UUID) or not isinstance( + assessment.environment_display_name, str + ): + raise _invalid() + if not isinstance(assessment.status, str): + raise _invalid() + for finding in assessment.findings: + results.append(_finding_result(finding, assessment, head, code=False)) + reasons = _assessment_gap_reasons(assessment) + for reason in reasons: + notifications.append( + _notification( + f"{assessment.environment_display_name}: {reason}", reason, assessment.id + ) + ) + assessments.append( + { + "id": str(assessment.id), + "environment": assessment.environment_display_name, + "status": assessment.status, + "gapReasons": reasons, + } + ) + if state != "COMPLETED": + reason = _review_reason(state) + notifications.append(_notification(f"Review {review.id}: {reason}", reason, None)) + if _code_unavailable(review): + notifications.append( + _notification( + "Code-only: CODE_ASSESSMENT_UNAVAILABLE", "CODE_ASSESSMENT_UNAVAILABLE", None + ) + ) + rules = [{"id": domain} for domain in _DOMAINS] + if code is not None: + rules.extend({"id": domain} for domain in _CODE_DOMAINS) + properties: dict[str, Any] = { + "reviewId": str(review.id), + "organizationId": str(organization_id), + "reviewUrl": review_url, + "headCommitSha": head, + "repository": _recorded_text(review.repository_full_name), + "pullRequestNumber": _recorded_int(review.pull_request_number), + "runNumber": _recorded_int(review.run_number), + "reviewStatus": review.status, + "outcome": state, + "assessmentCount": review.assessment_count, + "assessments": assessments, + } + if code_properties is not None: + properties["codeAssessment"] = code_properties + return { + "$schema": _SCHEMA, + "version": "2.1.0", + "runs": [ + { + "tool": { + "driver": {"name": "Proofline", "version": VERSION, "rules": rules}, + }, + "invocations": [ + { + "executionSuccessful": state == "COMPLETED" and not notifications, + "toolExecutionNotifications": notifications, + } + ], + "results": results, + "properties": properties, + } + ], + } + + +def _notification(message: str, reason: str, assessment_id: UUID | None) -> dict[str, Any]: + properties: dict[str, Any] = {} + if assessment_id is not None: + properties["assessmentId"] = str(assessment_id) + properties["reasonCode"] = reason + return { + "level": "error", + "message": {"text": _sarif_text(message)}, + "properties": properties, + } + + +def _finding_result( + finding: CustomerFinding, + source: CustomerAssessment | CustomerCodeAssessment, + head: str, + *, + code: bool, +) -> dict[str, Any]: + if not isinstance(finding, CustomerFinding) or not isinstance(finding.id, UUID): + raise _invalid() + if not isinstance(finding.severity, str) or not isinstance(finding.risk_domain, str): + raise _invalid() + if not isinstance(finding.claim, str) or not isinstance(finding.impact, str): + raise _invalid() + if not isinstance(finding.lifecycle_state, str) or not isinstance(finding.evidence_refs, list): + raise _invalid() + if finding.severity == "SERIOUS": + level = "error" + elif finding.severity == "WARNING": + level = "warning" + else: + raise SarifExportError( + f"finding {finding.id} has unsupported severity {_one_line(finding.severity)}; " + "SARIF export refused" + ) + domains = _CODE_DOMAINS if code else _DOMAINS + try: + domain_index = domains.index(finding.risk_domain) + except ValueError: + raise SarifExportError( + f"finding {finding.id} has unsupported risk domain {_one_line(finding.risk_domain)}; " + "SARIF export refused" + ) from None + locations: list[dict[str, Any]] = [] + seen: set[tuple[str, int, int]] = set() + evidence: list[dict[str, Any]] = [] + for reference in finding.evidence_refs: + if not isinstance(reference, CustomerEvidenceRef): + raise _invalid() + located = _code_location(reference, head) + if ( + located.status == "LOCATED" + and located.path is not None + and located.start is not None + and located.end is not None + and (located.path, located.start, located.end) not in seen + ): + seen.add((located.path, located.start, located.end)) + locations.append( + { + "physicalLocation": { + "artifactLocation": { + "uri": _path_uri(located.path), + "uriBaseId": "%SRCROOT%", + }, + "region": {"startLine": located.start, "endLine": located.end}, + } + } + ) + evidence.append(_evidence_properties(reference, located.status)) + message = f"{finding.claim}\n\n{finding.impact}" + if code: + message = f"Code-only finding\n\n{message}" + result: dict[str, Any] = { + "ruleId": domains[domain_index], + "ruleIndex": (len(_DOMAINS) if code else 0) + domain_index, + "level": level, + "message": {"text": _sarif_text(message)}, + "locations": locations, + } + fingerprint = _recorded_text(finding.finding_fingerprint) + if fingerprint is not None: + result["partialFingerprints"] = {"prooflineFinding/v1": fingerprint} + if len(locations) > 1: + result["relatedLocations"] = locations[1:] + result["locations"] = locations[:1] + environment = ( + source.environment_display_name if isinstance(source, CustomerAssessment) else None + ) + properties: dict[str, Any] = { + "findingId": str(finding.id), + "assessmentId": str(source.id), + "lane": "CODE" if code else "ENV", + } + if environment is not None: + properties["environment"] = environment + properties["lifecycleState"] = finding.lifecycle_state + properties["locationStatus"] = "LOCATED" if locations else "NO_VERIFIED_REVIEWED_LOCATION" + remediation = _recorded_text(finding.remediation) + if remediation is not None: + properties["remediation"] = remediation + properties["evidenceReferences"] = evidence + result["properties"] = properties + return result + + +def _evidence_properties(reference: CustomerEvidenceRef, status: str) -> dict[str, Any]: + return { + "kind": reference.kind, + "provenanceStatus": reference.provenance_status + if isinstance(reference.provenance_status, str) + else None, + "commitSha": _recorded_text(reference.commit_sha), + "companionRepository": _recorded_text(reference.companion_repository), + "repositoryPath": _recorded_text(reference.repository_path), + "startLine": reference.start_line if type(reference.start_line) is int else None, + "endLine": reference.end_line if type(reference.end_line) is int else None, + "locationStatus": status, + } + + +def _code_location(reference: CustomerEvidenceRef, head: str) -> _Location: + if not isinstance(reference.kind, str) or not isinstance(reference.provenance_status, str): + raise _invalid() + if ( + reference.kind == "companion_repository_file" + or _recorded_text(reference.companion_repository) is not None + ): + return _Location("COMPANION_REPOSITORY") + if reference.kind != "repository_file": + return _Location("NON_CODE_REFERENCE") + if reference.provenance_status != "VERIFIED_REPOSITORY_READ": + return _Location("UNVERIFIED_PROVENANCE") + commit = _recorded_text(reference.commit_sha) + if commit is None: + return _Location("COMMIT_UNAVAILABLE") + if not _hex_sha(commit): + return _Location("INVALID_COMMIT") + if commit.lower() != head.lower(): + return _Location("DIFFERENT_COMMIT") + path = _recorded_text(reference.repository_path) + if path is None or not _valid_path(path): + return _Location("INVALID_PATH") + start = reference.start_line + if type(start) is not int or start <= 0: + return _Location("INVALID_REGION") + end = reference.end_line + if isinstance(end, Unset) or end is None: + end_line = start + elif type(end) is not int or end < start: + return _Location("INVALID_REGION") + else: + end_line = end + return _Location("LOCATED", path, start, end_line) + + +def _assessment_gap_reasons(assessment: CustomerAssessment) -> list[str]: + checks = assessment.checks + if ( + not isinstance(checks.families, list) + or type(checks.receipt_recorded) is not bool + or not isinstance(checks.status, str) + ): + raise _invalid() + check_gap = not checks.receipt_recorded or checks.status not in _COMPLETE_CHECKS + for family in checks.families: + if ( + not isinstance(family, CustomerCheck) + or not isinstance(family.outcome, str) + or not isinstance(family.lifecycle, str) + ): + raise _invalid() + if family.outcome not in _COMPLETE_CHECKS or family.lifecycle not in _COMPLETE_LIFECYCLES: + check_gap = True + recorded = assessment.reason_codes + if isinstance(recorded, Unset) or recorded is None: + reasons: list[str] = [] + elif isinstance(recorded, list) and all(isinstance(reason, str) for reason in recorded): + reasons = list(recorded) + else: + raise _invalid() + kind = _recorded_text(assessment.incompleteness_kind) + evidence = _recorded_text(assessment.release_evidence_gap) + if ( + assessment.status == "COMPLETED" + and not check_gap + and not reasons + and kind is None + and evidence is None + ): + return [] + why = list(reasons) + if kind is not None: + why.append(kind) + if evidence is not None: + why.append(evidence) + if check_gap: + why.append("CHECK_RECEIPT_ABSENT" if not checks.receipt_recorded else "CHECKS_UNRESOLVED") + if not why: + why.append( + "ASSESSMENT_RESULT_UNAVAILABLE" + if assessment.status in _KNOWN_ASSESSMENTS + else "UNKNOWN_ASSESSMENT_STATUS" + ) + return sorted(set(why)) + + +def _code_gap(code: CustomerCodeAssessment) -> str | None: + kind = _recorded_text(code.incompleteness_kind) + if kind: + return kind + if _plain_text(code.status) == "COMPLETED": + return None + return "CODE_ASSESSMENT_STATE_UNAVAILABLE" + + +def _usable_code(review: CustomerReviewDetail) -> CustomerCodeAssessment | None: + code = review.code_assessment + if not isinstance(code, CustomerCodeAssessment): + return None + if ( + not isinstance(code.head_commit_sha, str) + or not isinstance(code.base_commit_sha, str) + or not isinstance(code.findings, list) + ): + return None + + def same(recorded: object, actual: str) -> bool: + text = _recorded_text(recorded) + return text is None or text.lower() == actual.lower() + + if not same(review.head_commit_sha, code.head_commit_sha) or not same( + review.base_commit_sha, code.base_commit_sha + ): + return None + if any( + isinstance(finding, CustomerFinding) and finding.risk_domain == _WITHHELD_MARKER + for finding in code.findings + ): + return None + return code + + +def _code_unavailable(review: CustomerReviewDetail) -> bool: + if _usable_code(review) is not None: + return False + recorded = review.code_assessment + if recorded is not None and not isinstance(recorded, Unset): + return True + return review.assessment_count == 0 and not review.assessments + + +def _terminal(review: CustomerReviewDetail) -> str | None: + if not isinstance(review.status, str): + raise _invalid() + if review.status in ("PENDING", "RUNNING"): + return None + if review.status == "FAILED": + return "FAILED" + if review.status == "STOPPED": + return "STOPPED" + if review.status == "COMPLETED" and review.assessment_count == len(review.assessments): + if not review.assessments: + code = _usable_code(review) + complete = code is not None and _code_gap(code) is None + else: + complete = all( + not _assessment_gap_reasons(assessment) for assessment in review.assessments + ) + return "COMPLETED" if complete else "INCOMPLETE" + return "INCOMPLETE" + + +def _review_reason(state: str | None) -> str: + if state is None: + return "REVIEW_IN_PROGRESS" + if state == "FAILED": + return "REVIEW_FAILED" + if state == "STOPPED": + return "REVIEW_STOPPED" + return "REVIEW_RESULT_INCOMPLETE" + + +def _recorded_text(value: object) -> str | None: + if isinstance(value, Unset) or value is None: + return None + if isinstance(value, str): + return value + raise _invalid() + + +def _recorded_int(value: object) -> int | None: + if isinstance(value, Unset) or value is None: + return None + if type(value) is int: + return value + raise _invalid() + + +def _plain_text(value: object) -> str: + if isinstance(value, str): + return str(value) + raise _invalid() + + +def _hex_sha(value: str) -> bool: + return len(value) in (40, 64) and all(character in _HEX for character in value) + + +def _valid_path(path: str) -> bool: + if any(part in ("", ".", "..") for part in path.split("/")): + return False + return not any( + unicodedata.category(character) == "Cc" or character in _FORBIDDEN_PATH_CHARACTERS + for character in path + ) + + +def _path_uri(path: str) -> str: + encoded: list[str] = [] + for byte in path.encode("utf-8"): + if byte in b"-._~/" or 48 <= byte <= 57 or 65 <= byte <= 90 or 97 <= byte <= 122: + encoded.append(chr(byte)) + else: + encoded.append(f"%{_URI_HEX[byte >> 4]}{_URI_HEX[byte & 15]}") + return "".join(encoded) + + +def _sarif_text(raw: str) -> str: + # SARIF message text interprets braces and embedded links. Escape once. + text: list[str] = [] + for character in raw: + if character == "{": + text.append("{{") + elif character == "}": + text.append("}}") + elif character in "\\[]": + text.append("\\") + text.append(character) + else: + text.append(character) + return "".join(text) + + +def _one_line(value: str) -> str: + cleaned = "".join( + character if character in "\n\t" or unicodedata.category(character) != "Cc" else "\ufffd" + for character in value + ) + return cleaned.replace("\n", " ").replace("\t", " ") + + +def _invalid() -> SarifExportError: + return SarifExportError("The review has invalid identity or finding metadata.") diff --git a/tests/test_sarif.py b/tests/test_sarif.py new file mode 100644 index 0000000..af16c5c --- /dev/null +++ b/tests/test_sarif.py @@ -0,0 +1,509 @@ +"""SARIF projection of a retained pull-request review.""" + +from __future__ import annotations + +import json +from typing import Any +from uuid import UUID + +import pytest + +from proofline import ( + CustomerReviewDetail, + ProoflineClient, + SarifExportError, + review_sarif, +) +from proofline import __version__ as SDK_VERSION + +from .fixtures import ORG, REVIEW, review +from .http_server import Reply, Request, Server + +HEAD = "a" * 40 +BASE = "b" * 40 +OTHER = "9876543210abcdef0123456789abcdef01234567" +CLAIM = "The order receipt is applied twice." +IMPACT = "The customer can be charged twice." +CANARY = "world-evidence-canary" +TOKEN = "pln_test_sdk_credential_canary" +URL = f"https://proofline.sh/orgs/by-id/{ORG}/reviews/{REVIEW}" + + +def finding( + number: int, + severity: str = "WARNING", + refs: list[dict[str, Any]] | None = None, + **extra: Any, +) -> dict[str, Any]: + body: dict[str, Any] = { + "id": str(UUID(int=0x0190F3A0000070008000000000000100 + number)), + "severity": severity, + "risk_domain": "data_integrity", + "claim": CLAIM, + "impact": IMPACT, + "symbols": [], + "lifecycle_state": "ACTIVE", + "evidence_refs": [] if refs is None else refs, + "rule_refs": [], + "comment_reactions": [], + "remediation": "Make the order receipt idempotent.", + "evidence_values": CANARY, + } + body.update(extra) + return body + + +def code_ref(path: str, start: int, end: int | None = None, **extra: Any) -> dict[str, Any]: + body: dict[str, Any] = { + "kind": "repository_file", + "label": CANARY, + "provenance_status": "VERIFIED_REPOSITORY_READ", + "commit_sha": HEAD, + "repository_path": path, + "start_line": start, + "url": f"https://elsewhere.example/{CANARY}", + } + if end is not None: + body["end_line"] = end + body.update(extra) + return body + + +def pull_request(findings: list[dict[str, Any]], **extra: Any) -> dict[str, Any]: + document = review() + document["kind"] = "PULL_REQUEST" + document["head_commit_sha"] = HEAD + document["repository_full_name"] = "acme/web" + document["pull_request_number"] = 42 + document["run_number"] = 7 + document["operator_diagnostics"] = CANARY + document["assessments"] = [ + { + "id": document["assessments"][0]["id"], + "environment_display_name": "production", + "status": "COMPLETED", + "domains": [], + "findings": findings, + "checks": {"families": [], "receipt_recorded": True, "status": "READY"}, + "world_evidence_values": CANARY, + } + ] + document.update(extra) + return document + + +def project(document: dict[str, Any], *, review_id: UUID = REVIEW) -> dict[str, Any]: + return review_sarif( + CustomerReviewDetail.from_dict(document), + ORG, + review_id=review_id, + review_url=URL, + ) + + +def notice_codes(run: dict[str, Any]) -> list[str]: + notes = run["invocations"][0]["toolExecutionNotifications"] + return [note["properties"]["reasonCode"] for note in notes] + + +def run_of(document: dict[str, Any]) -> dict[str, Any]: + report = project(document) + assert report["version"] == "2.1.0" + assert report["$schema"].endswith("sarif-schema-2.1.0.json") + assert CANARY not in json.dumps(report) + assert TOKEN not in json.dumps(report) + runs = report["runs"] + assert len(runs) == 1 + for result in runs[0]["results"]: + rules = runs[0]["tool"]["driver"]["rules"] + assert rules[result["ruleIndex"]]["id"] == result["ruleId"] + assert "guid" not in result + assert "fingerprints" not in result + assert "correlationGuid" not in result + assert "markdown" not in result["message"] + assert "arguments" not in result["message"] + return runs[0] + + +def test_completed_review_keeps_identity_locations_and_fingerprint() -> None: + markup = "charge {twice} \\[note]" + references = [ + code_ref("src/orders #100%.rs", 10, 14), + code_ref("src/orders #100%.rs", 10, 14), + code_ref("src/\u00e9.rs", 20), + code_ref("src/orders.rs", 1, 2, commit_sha=OTHER), + code_ref("src/orders.rs", 1, 2, commit_sha="main"), + code_ref("src/orders.rs", 1, 2, commit_sha=None), + code_ref("src/orders.rs", 1, 2, provenance_status="UNVERIFIED"), + code_ref( + "src/orders.rs", + 1, + 2, + kind="companion_repository_file", + companion_repository="acme/contracts", + ), + code_ref("src/orders.rs", 1, 2, companion_repository="acme/contracts"), + ] + for path in ( + "", + "/etc/passwd", + "../src/orders.rs", + "src/../orders.rs", + "src/./orders.rs", + "src//orders.rs", + "src\\orders.rs", + "C:/orders.rs", + "https://elsewhere.example/orders.rs", + "src/\u001b[2J.rs", + "src/\u202eorders.rs", + ): + references.append(code_ref(path, 1, 2)) + references.extend( + ( + code_ref("src/orders.rs", 0, 2), + code_ref("src/orders.rs", 2, 1), + code_ref("src/orders.rs", 1, 2, start_line=None), + { + "kind": "world_entity", + "label": CANARY, + "provenance_status": "PINNED_WORLD_ENTITY", + "world_entity_id": CANARY, + "values": CANARY, + }, + ) + ) + retained = finding( + 1, + "SERIOUS", + references, + claim=markup, + finding_fingerprint="fingerprint-v1", + ) + older = finding(2, "WARNING", [], remediation=None) + run = run_of(pull_request([retained, older])) + assert run["tool"]["driver"]["name"] == "Proofline" + assert run["tool"]["driver"]["version"] == SDK_VERSION + assert [rule["id"] for rule in run["tool"]["driver"]["rules"]][11] == "data_integrity" + assert len(run["tool"]["driver"]["rules"]) == 18 + assert run["invocations"][0]["executionSuccessful"] is True + assert run["invocations"][0]["toolExecutionNotifications"] == [] + properties = run["properties"] + assert properties["reviewId"] == str(REVIEW) + assert properties["organizationId"] == str(ORG) + assert properties["reviewUrl"] == URL + assert properties["headCommitSha"] == HEAD + assert properties["repository"] == "acme/web" + assert properties["pullRequestNumber"] == 42 + assert properties["runNumber"] == 7 + assert properties["reviewStatus"] == "COMPLETED" + assert properties["outcome"] == "COMPLETED" + assert properties["assessmentCount"] == 1 + assert "codeAssessment" not in properties + assert len(run["results"]) == 2 + serious = run["results"][0] + assert serious["level"] == "error" + assert serious["ruleId"] == "data_integrity" + assert serious["ruleIndex"] == 11 + assert serious["message"]["text"] == "charge {{twice}} \\\\\\[note\\]\n\n" + IMPACT + assert serious["partialFingerprints"] == {"prooflineFinding/v1": "fingerprint-v1"} + assert serious["properties"]["findingId"] == retained["id"] + assert serious["properties"]["environment"] == "production" + assert serious["properties"]["lane"] == "ENV" + assert serious["properties"]["lifecycleState"] == "ACTIVE" + assert serious["properties"]["remediation"] == "Make the order receipt idempotent." + assert serious["properties"]["locationStatus"] == "LOCATED" + locations = serious["locations"] + assert len(locations) == 1 + artifact = locations[0]["physicalLocation"]["artifactLocation"] + assert artifact["uri"] == "src/orders%20%23100%25.rs" + assert artifact["uriBaseId"] == "%SRCROOT%" + assert locations[0]["physicalLocation"]["region"] == {"startLine": 10, "endLine": 14} + related = serious["relatedLocations"] + assert len(related) == 1 + assert related[0]["physicalLocation"]["artifactLocation"]["uri"] == "src/%C3%A9.rs" + assert related[0]["physicalLocation"]["region"] == {"startLine": 20, "endLine": 20} + refs = serious["properties"]["evidenceReferences"] + assert [ref["locationStatus"] for ref in refs] == [ + "LOCATED", + "LOCATED", + "LOCATED", + "DIFFERENT_COMMIT", + "INVALID_COMMIT", + "COMMIT_UNAVAILABLE", + "UNVERIFIED_PROVENANCE", + "COMPANION_REPOSITORY", + "COMPANION_REPOSITORY", + *["INVALID_PATH"] * 11, + *["INVALID_REGION"] * 3, + "NON_CODE_REFERENCE", + ] + assert refs[5]["commitSha"] is None + assert refs[7]["companionRepository"] == "acme/contracts" + assert refs[2]["endLine"] is None + assert "label" not in refs[23] + warning = run["results"][1] + assert warning["level"] == "warning" + assert warning["locations"] == [] + assert "relatedLocations" not in warning + assert "partialFingerprints" not in warning + assert "remediation" not in warning["properties"] + assert warning["properties"]["locationStatus"] == "NO_VERIFIED_REVIEWED_LOCATION" + again = project(pull_request([retained, older])) + assert again == project(pull_request([retained, older])) + + +def test_uppercase_head_matches_its_evidence_and_is_stored_lowercase() -> None: + head = "A" * 40 + document = pull_request( + [finding(1, "WARNING", [code_ref("src/orders.rs", 3, 3, commit_sha=HEAD)])], + head_commit_sha=head, + ) + run = run_of(document) + assert run["properties"]["headCommitSha"] == HEAD + assert run["results"][0]["properties"]["locationStatus"] == "LOCATED" + long_head = "ab" * 32 + document = pull_request([finding(1)], head_commit_sha=long_head.upper()) + assert run_of(document)["properties"]["headCommitSha"] == long_head + + +def test_code_findings_keep_their_lane_and_do_not_replace_environments() -> None: + recorded = finding(1, "SERIOUS", [code_ref("src/orders.rs", 10, 14)], risk_domain="logic") + code = { + "id": "00000000-0000-4000-8000-000000000006", + "status": "COMPLETED", + "base_commit_sha": BASE, + "head_commit_sha": HEAD, + "incompleteness_kind": None, + "findings": [recorded], + "compatible_addition": CANARY, + } + code_only = pull_request( + [], assessment_count=0, assessments=[], base_commit_sha=BASE, code_assessment=code + ) + run = run_of(code_only) + rules = [rule["id"] for rule in run["tool"]["driver"]["rules"]] + assert rules[11] == "data_integrity" + assert rules[18:] == [ + "logic", + "edge_case", + "error_handling", + "state_transition", + "concurrency", + "resource_lifecycle", + "credential_exposure", + "test_marker", + "removed_assertion", + "debug_residue", + "committed_secret", + ] + assert run["invocations"][0]["executionSuccessful"] is True + assert run["properties"]["outcome"] == "COMPLETED" + assert run["properties"]["codeAssessment"]["status"] == "COMPLETED" + assert run["properties"]["codeAssessment"]["gapReasons"] == [] + result = run["results"][0] + assert result["ruleId"] == "logic" + assert result["ruleIndex"] == 18 + assert result["properties"]["lane"] == "CODE" + assert "environment" not in result["properties"] + assert result["message"]["text"] == f"Code-only finding\n\n{CLAIM}\n\n{IMPACT}" + assert result["properties"]["locationStatus"] == "LOCATED" + + beside = pull_request( + [finding(2, "WARNING", [], risk_domain="observability")], + base_commit_sha=BASE, + code_assessment=code, + ) + run = run_of(beside) + assert [item["properties"]["lane"] for item in run["results"]] == ["CODE", "ENV"] + assert run["results"][1]["ruleIndex"] == 16 + assert run["invocations"][0]["executionSuccessful"] is True + + partial = dict(code) + partial["status"] = "INCOMPLETE" + partial["incompleteness_kind"] = "MODEL_BUDGET_EXHAUSTED" + run = run_of(pull_request([finding(2)], base_commit_sha=BASE, code_assessment=partial)) + assert run["properties"]["outcome"] == "COMPLETED" + assert run["invocations"][0]["executionSuccessful"] is False + assert run["results"][0]["properties"]["lane"] == "CODE" + notices = run["invocations"][0]["toolExecutionNotifications"] + assert notices[0]["properties"]["reasonCode"] == "MODEL_BUDGET_EXHAUSTED" + assert notices[0]["message"]["text"] == "Code-only: MODEL_BUDGET_EXHAUSTED" + assert "codeAssessment" in run["properties"] + + withheld = dict(code) + withheld["findings"] = [finding(1, risk_domain="[withheld: environment value]")] + withheld["status"] = "COMPLETED" + withheld["incompleteness_kind"] = None + run = run_of(pull_request([finding(2)], base_commit_sha=BASE, code_assessment=withheld)) + assert [item["properties"]["lane"] for item in run["results"]] == ["ENV"] + assert "codeAssessment" not in run["properties"] + assert run["invocations"][0]["executionSuccessful"] is False + assert run["invocations"][0]["toolExecutionNotifications"][-1]["properties"]["reasonCode"] == ( + "CODE_ASSESSMENT_UNAVAILABLE" + ) + + moved = dict(code) + moved["head_commit_sha"] = OTHER + run = run_of( + pull_request( + [], assessment_count=0, assessments=[], base_commit_sha=BASE, code_assessment=moved + ) + ) + assert run["results"] == [] + assert notice_codes(run) == ["REVIEW_RESULT_INCOMPLETE", "CODE_ASSESSMENT_UNAVAILABLE"] + + +def test_gaps_are_named_and_a_partial_review_still_projects_findings() -> None: + recorded = finding(1, "WARNING", [code_ref("src/orders.rs", 5)]) + checks = { + "families": [ + {"name": "migrations", "outcome": "UNRESOLVED", "lifecycle": "GATING", "files": []} + ], + "receipt_recorded": False, + "status": "READY", + } + for status, reason in ( + ("COMPLETED", "REVIEW_RESULT_INCOMPLETE"), + ("PENDING", "REVIEW_IN_PROGRESS"), + ("RUNNING", "REVIEW_IN_PROGRESS"), + ("FAILED", "REVIEW_FAILED"), + ("STOPPED", "REVIEW_STOPPED"), + ("UNRECOGNIZED", "REVIEW_RESULT_INCOMPLETE"), + ("SUPERSEDED", "REVIEW_RESULT_INCOMPLETE"), + ): + document = pull_request([recorded], status=status) + document["assessments"][0]["status"] = "INCOMPLETE" + document["assessments"][0]["incompleteness_kind"] = "WORLD_UNAVAILABLE" + document["assessments"][0]["reason_codes"] = [ + "WORLD_UNAVAILABLE", + "WORLD_UNAVAILABLE", + "LITERAL_REASON", + ] + document["assessments"][0]["release_evidence_gap"] = "RELEASE_EVIDENCE_ABSENT" + document["assessments"][0]["checks"] = checks + document["assessments"][0]["environment_display_name"] = "prod {name}" + run = run_of(document) + assert run["properties"]["reviewStatus"] == status + assert run["properties"]["outcome"] == { + "PENDING": None, + "RUNNING": None, + "FAILED": "FAILED", + "STOPPED": "STOPPED", + }.get(status, "INCOMPLETE") + assert run["invocations"][0]["executionSuccessful"] is False + assert run["results"][0]["properties"]["findingId"] == recorded["id"] + assert run["results"][0]["properties"]["locationStatus"] == "LOCATED" + notices = run["invocations"][0]["toolExecutionNotifications"] + assert [notice["properties"]["reasonCode"] for notice in notices] == [ + "CHECK_RECEIPT_ABSENT", + "LITERAL_REASON", + "RELEASE_EVIDENCE_ABSENT", + "WORLD_UNAVAILABLE", + reason, + ] + literal = next( + notice for notice in notices if notice["properties"]["reasonCode"] == "LITERAL_REASON" + ) + assert literal["message"]["text"] == "prod {{name}}: LITERAL_REASON" + assert literal["level"] == "error" + assert "markdown" not in literal["message"] + assert run["properties"]["assessments"][0]["gapReasons"] == [ + "CHECK_RECEIPT_ABSENT", + "LITERAL_REASON", + "RELEASE_EVIDENCE_ABSENT", + "WORLD_UNAVAILABLE", + ] + + pending = pull_request([]) + pending["assessments"][0]["status"] = "PENDING" + assert run_of(pending)["properties"]["assessments"][0]["gapReasons"] == [ + "ASSESSMENT_RESULT_UNAVAILABLE" + ] + stopped = pull_request([]) + stopped["assessments"][0]["status"] = "STOPPED" + assert run_of(stopped)["properties"]["assessments"][0]["gapReasons"] == [ + "UNKNOWN_ASSESSMENT_STATUS" + ] + + unresolved = pull_request([recorded]) + unresolved["assessments"][0]["checks"] = { + "families": [ + {"name": "migrations", "outcome": "UNRESOLVED", "lifecycle": "GATING", "files": []} + ], + "receipt_recorded": True, + "status": "READY", + } + run = run_of(unresolved) + assert run["properties"]["assessments"][0]["gapReasons"] == ["CHECKS_UNRESOLVED"] + assert run["properties"]["outcome"] == "INCOMPLETE" + + missing = pull_request([]) + missing["assessment_count"] = 2 + empty = pull_request([], assessment_count=0, assessments=[]) + for document in (missing, empty): + run = run_of(document) + assert run["invocations"][0]["executionSuccessful"] is False + assert notice_codes(run)[0] == "REVIEW_RESULT_INCOMPLETE" + assert notice_codes(run_of(empty))[1] == "CODE_ASSESSMENT_UNAVAILABLE" + assert len(run_of(missing)["invocations"][0]["toolExecutionNotifications"]) == 1 + + +def test_refuses_unpinned_or_unrecognized_records_without_a_document() -> None: + wrong = pull_request([]) + with pytest.raises(SarifExportError, match="server returned review"): + project(wrong, review_id=UUID("00000000-0000-4000-8000-000000000004")) + for head in (None, "main", f" {HEAD}"): + with pytest.raises(SarifExportError, match="full commit SHA"): + project(pull_request([], head_commit_sha=head)) + with pytest.raises(SarifExportError, match="retained pull-request review"): + project(pull_request([], kind="DEPLOYMENT")) + with pytest.raises(SarifExportError, match="unsupported severity"): + project(pull_request([finding(1, "CRITICAL")])) + injected = finding(1, "BAD\x1b[2J", claim=CANARY) + with pytest.raises(SarifExportError, match="unsupported severity") as raised: + project(pull_request([injected])) + assert "\x1b" not in str(raised.value) + assert CANARY not in str(raised.value) + assert "BAD\ufffd[2J" in str(raised.value) + with pytest.raises(SarifExportError, match="unsupported risk domain"): + project(pull_request([finding(1, risk_domain="unrecognized_domain")])) + code = { + "id": "00000000-0000-4000-8000-000000000006", + "status": "COMPLETED", + "base_commit_sha": BASE, + "head_commit_sha": HEAD, + "findings": [finding(1, risk_domain="unrecognized_domain")], + } + with pytest.raises(SarifExportError, match="unsupported risk domain"): + project( + pull_request( + [], assessment_count=0, assessments=[], base_commit_sha=BASE, code_assessment=code + ) + ) + + +def test_caller_arguments_are_rejected_before_a_document_is_built() -> None: + parsed = CustomerReviewDetail.from_dict(pull_request([])) + with pytest.raises(TypeError): + review_sarif(parsed, "org", review_id=REVIEW, review_url=URL) # type: ignore[arg-type] + with pytest.raises(ValueError): + review_sarif(parsed, ORG, review_id=REVIEW, review_url="") + + +def test_get_review_projects_the_parsed_record() -> None: + document = pull_request([finding(1, "SERIOUS", [code_ref("src/orders.rs", 8, 9)])]) + + def respond(request: Request, index: int) -> Reply: + assert request.method == "GET" + assert request.headers["Authorization"] == f"Bearer {TOKEN}" + assert request.path.split("?", 1)[0] == f"/api/v1/organizations/{ORG}/reviews/{REVIEW}" + return Reply(body=document) + + with Server(respond) as server: + with ProoflineClient(TOKEN, base_url=server.url) as client: + parsed = client.get_review(ORG, REVIEW) + report = review_sarif(parsed, ORG, review_id=REVIEW, review_url=URL) + result = report["runs"][0]["results"][0] + assert result["level"] == "error" + assert result["locations"][0]["physicalLocation"]["region"] == {"startLine": 8, "endLine": 9} + assert CANARY not in json.dumps(report)