From 24ed4d221389bd5ec0504600556d7f666382bbeb Mon Sep 17 00:00:00 2001 From: Bamdad Dashtban Date: Wed, 22 Jul 2026 16:19:18 +0100 Subject: [PATCH 1/4] Add deterministic report serializers: text, JSON, SARIF 2.1.0 New taintline.report module. to_text is the existing human report; to_json is a stable sorted view of the findings; to_sarif emits SARIF 2.1.0 for GitHub code scanning (one result per finding, detector->ruleId, severity->level, span ids in the message + properties + per-location logicalLocations, trace path as the artifactLocation). All three are pure, offline and byte-deterministic: object keys emitted sorted and findings already contract-sorted, so nothing depends on PYTHONHASHSEED. Typed SARIF nodes, no loose dict assembly. --- src/taintline/report.py | 133 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 133 insertions(+) create mode 100644 src/taintline/report.py diff --git a/src/taintline/report.py b/src/taintline/report.py new file mode 100644 index 0000000..3a89dfd --- /dev/null +++ b/src/taintline/report.py @@ -0,0 +1,133 @@ +"""Deterministic report serializers for a :class:`~taintline.model.Verdict`. + +Three formats share one contract: **pure, offline, byte-deterministic**. No LLM, +no network, no timestamps or rng. Given the same ``Verdict`` (and, for SARIF, the +same trace uri) the bytes are identical across runs and independent of +``PYTHONHASHSEED`` — object keys are emitted sorted and the findings are already +contract-sorted upstream, so nothing depends on set/dict iteration order. + +* ``to_text`` — the human report (unchanged default; see :func:`taintline.cli.render`). +* ``to_json`` — a stable JSON view of the findings for scripting. +* ``to_sarif`` — SARIF 2.1.0 for GitHub code scanning (``upload-sarif``): one + ``result`` per finding, ``detector`` -> ``ruleId``, ``severity`` -> ``level``, + the span ids in the message, ``properties`` and per-location ``logicalLocations``. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any + +from taintline.model import Finding, Verdict + +SARIF_SCHEMA = "https://json.schemastore.org/sarif-2.1.0.json" +SARIF_VERSION = "2.1.0" +TOOL_NAME = "taintline" +TOOL_URI = "https://github.com/bamdadd/taintline" + +#: taintline ``Finding.severity`` -> SARIF ``result.level``. +_SARIF_LEVEL = {"error": "error", "warn": "warning"} + +#: One-line, deterministic rule descriptions for the SARIF tool driver. +_DETECTOR_DESCRIPTIONS = { + "taint": "Untrusted tool output reaches a sensitive sink without sanitization.", + "loops": "A tool-call cycle repeats without making progress.", + "retry-storm": "A span is retried far more often than its siblings.", + "swallowed-error": "An ERROR span is followed by success as if nothing failed.", + "context-truncation": "Context or tool output is truncated below a safe budget.", + "budget-overrun": "A token or step budget is exceeded.", +} + + +def _level(severity: str) -> str: + return _SARIF_LEVEL.get(severity, "warning") + + +def to_text(verdict: Verdict) -> str: + """The human report. Kept identical to the historical default output.""" + if not verdict.findings: + return "[taintline] no findings" + lines = [ + f"{f.severity.upper()} {f.detector}: {f.message} [{','.join(f.span_ids)}]" + for f in verdict.findings + ] + lines.append(f"[taintline] {len(verdict.findings)} finding(s)") + return "\n".join(lines) + + +def _finding_json(finding: Finding) -> dict[str, Any]: + return { + "detector": finding.detector, + "severity": finding.severity, + "message": finding.message, + "span_ids": list(finding.span_ids), + } + + +def to_json(verdict: Verdict) -> str: + """A stable JSON view of the verdict. ``sort_keys`` makes it byte-identical + for a given trace regardless of ``PYTHONHASHSEED``.""" + payload = { + "tool": TOOL_NAME, + "count": len(verdict.findings), + "findings": [_finding_json(f) for f in verdict.findings], + } + return json.dumps(payload, sort_keys=True, indent=2) + + +@dataclass(frozen=True) +class _Rule: + """One SARIF ``reportingDescriptor`` (a rule) for a detector.""" + + detector: str + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.detector, + "name": self.detector, + "shortDescription": {"text": _DETECTOR_DESCRIPTIONS.get(self.detector, self.detector)}, + } + + +def _result(finding: Finding, trace_uri: str) -> dict[str, Any]: + """One SARIF ``result`` for a finding, located at the trace + its spans.""" + spans = ", ".join(finding.span_ids) + location: dict[str, Any] = { + "physicalLocation": {"artifactLocation": {"uri": trace_uri or "trace"}}, + "logicalLocations": [{"name": sid, "kind": "span"} for sid in finding.span_ids], + } + return { + "ruleId": finding.detector, + "level": _level(finding.severity), + "message": {"text": f"{finding.message} [spans: {spans}]"}, + "locations": [location], + "properties": {"span_ids": list(finding.span_ids)}, + } + + +def to_sarif(verdict: Verdict, trace_uri: str = "") -> str: + """Serialize the verdict as SARIF 2.1.0 for ``github/codeql-action/upload-sarif``. + + ``trace_uri`` is the checked trace's path, used as the result's + ``artifactLocation.uri``. Deterministic: rules are sorted by id, results follow + the already-sorted findings, and all object keys are emitted sorted. + """ + detectors = sorted({f.detector for f in verdict.findings}) + log = { + "$schema": SARIF_SCHEMA, + "version": SARIF_VERSION, + "runs": [ + { + "tool": { + "driver": { + "name": TOOL_NAME, + "informationUri": TOOL_URI, + "rules": [_Rule(d).to_dict() for d in detectors], + } + }, + "results": [_result(f, trace_uri) for f in verdict.findings], + } + ], + } + return json.dumps(log, sort_keys=True, indent=2) From b8452cc9e7fd998e0ed90832361302002ab0a730 Mon Sep 17 00:00:00 2001 From: Bamdad Dashtban Date: Wed, 22 Jul 2026 16:19:18 +0100 Subject: [PATCH 2/4] cli: add --format {text,json,sarif} to 'taintline check' Thread a keyword-only fmt + trace_uri through run_check and dispatch to the new serializers; default stays text and byte-identical to today (render now delegates to report.to_text). Exit still gates on --fail-on regardless of format, so a SARIF run both surfaces findings and fails the build. main passes the trace path as the SARIF artifact uri. --- src/taintline/cli.py | 44 +++++++++++++++++++++++++++++++------------- 1 file changed, 31 insertions(+), 13 deletions(-) diff --git a/src/taintline/cli.py b/src/taintline/cli.py index 1330d0a..01b6a39 100644 --- a/src/taintline/cli.py +++ b/src/taintline/cli.py @@ -7,31 +7,41 @@ from taintline.detectors import ALL, run_all from taintline.ingest import load_trace from taintline.model import Trace, Verdict +from taintline.report import to_json, to_sarif, to_text DETECTORS = ["taint", "loops", "retry-storm", "swallowed-error", "context-truncation"] +FORMATS = ("text", "json", "sarif") def render(verdict: Verdict) -> str: """Deterministic textual report. Findings are already contract-sorted.""" - if not verdict.findings: - return "[taintline] no findings" - lines = [ - f"{f.severity.upper()} {f.detector}: {f.message} [{','.join(f.span_ids)}]" - for f in verdict.findings - ] - lines.append(f"[taintline] {len(verdict.findings)} finding(s)") - return "\n".join(lines) + return to_text(verdict) -def run_check(trace: Trace, fail_on: frozenset[str]) -> tuple[int, str]: - """Pure check core: run all detectors, render, gate exit on ``fail_on``. +def run_check( + trace: Trace, + fail_on: frozenset[str], + *, + fmt: str = "text", + trace_uri: str = "", +) -> tuple[int, str]: + """Pure check core: run all detectors, serialize, gate exit on ``fail_on``. Returns ``(exit_code, output)``. Exit is non-zero iff a finding's detector is - in ``fail_on``. Deterministic: same trace + fail_on => identical output. + in ``fail_on`` — independent of ``fmt``, so a SARIF run still gates the build. + ``fmt`` selects the serializer (``text``/``json``/``sarif``); ``trace_uri`` is + the checked trace's path, recorded in SARIF locations. Deterministic: same + trace + fail_on + fmt => byte-identical output. """ verdict = run_all(trace) exit_code = 1 if verdict.failed(fail_on) else 0 - return exit_code, render(verdict) + if fmt == "json": + output = to_json(verdict) + elif fmt == "sarif": + output = to_sarif(verdict, trace_uri) + else: + output = to_text(verdict) + return exit_code, output def _parse_fail_on(raw: str) -> frozenset[str]: @@ -45,6 +55,12 @@ def main(argv: list[str] | None = None) -> int: chk.add_argument("trace", help="OTel trace file (or Strands AgentResult export)") detectors = ",".join(sorted(ALL)) chk.add_argument("--fail-on", default="taint", help=f"comma list of: {detectors}") + chk.add_argument( + "--format", + choices=FORMATS, + default="text", + help="output format: text (default), json, or sarif (SARIF 2.1.0 for code scanning)", + ) sub.add_parser("view", help="render a local static HTML report") args = parser.parse_args(argv) @@ -54,7 +70,9 @@ def main(argv: list[str] | None = None) -> int: return 0 trace = load_trace(args.trace) - exit_code, output = run_check(trace, _parse_fail_on(args.fail_on)) + exit_code, output = run_check( + trace, _parse_fail_on(args.fail_on), fmt=args.format, trace_uri=args.trace + ) print(output) return exit_code From 33b08d1e8bca5ae172963abfaf97a0f316e3ad20 Mon Sep 17 00:00:00 2001 From: Bamdad Dashtban Date: Wed, 22 Jul 2026 16:19:18 +0100 Subject: [PATCH 3/4] test: SARIF 2.1.0 validity + JSON/SARIF hash-seed stability A seeded taint trace: assert the SARIF has the expected taint result (ruleId/level=error/message/logicalLocations/span_ids) and all required SARIF 2.1.0 fields ($schema, version, runs[].tool.driver, results[] with ruleId in the declared rules, valid level, message.text, artifactLocation). JSON + SARIF are byte-identical across five PYTHONHASHSEED values (subprocess). Default output confirmed unchanged (text == to_text, not JSON). --- tests/test_report.py | 159 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 159 insertions(+) create mode 100644 tests/test_report.py diff --git a/tests/test_report.py b/tests/test_report.py new file mode 100644 index 0000000..4920a02 --- /dev/null +++ b/tests/test_report.py @@ -0,0 +1,159 @@ +"""JSON + SARIF serializers: deterministic, hash-seed-stable, SARIF 2.1.0 valid.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +from taintline.cli import run_check +from taintline.detectors import run_all +from taintline.model import Span, Trace +from taintline.report import to_json, to_sarif, to_text +from tests._helpers import PAYLOAD, mk_span + +FAIL_ON = frozenset({"taint"}) +_VALID_LEVELS = {"error", "warning", "note", "none"} + + +def _fixture() -> Trace: + """A trace with a firing taint flow (a-src -> a-sink) plus clean spans.""" + return Trace( + spans=( + mk_span("a-src", start_ns=10, tool_name="web_fetch", tool_output=f"intro\n{PAYLOAD}\n"), + mk_span( + "a-sink", start_ns=20, tool_name="transfer", tool_input={"instruction": PAYLOAD} + ), + mk_span("c", start_ns=50, tool_name="web_search", tool_output="weather is fine"), + ) + ) + + +def _dump_otlp(trace: Trace) -> list[dict[str, object]]: + """Serialize spans in the OTLP-ish shape ``ingest.load_trace`` parses.""" + + def as_dict(s: Span) -> dict[str, object]: + attrs: dict[str, object] = {} + if s.tool_name is not None: + attrs["gen_ai.tool.name"] = s.tool_name + if s.tool_input is not None: + attrs["gen_ai.tool.input"] = s.tool_input + if s.tool_output is not None: + attrs["gen_ai.tool.output"] = s.tool_output + return { + "spanId": s.span_id, + "name": s.name, + "startTimeUnixNano": s.start_ns, + "attributes": attrs, + } + + return [as_dict(s) for s in trace.spans] + + +def _write_fixture(tmp_path: Path) -> Path: + fixture = tmp_path / "trace.json" + fixture.write_text(json.dumps({"spans": _dump_otlp(_fixture())}), encoding="utf-8") + return fixture + + +def _run_cli(fixture: Path, fmt: str, seed: str) -> tuple[int, str]: + code = "from taintline.cli import main; raise SystemExit(main())" + argv = [ + sys.executable, + "-c", + code, + "check", + str(fixture), + "--fail-on", + "taint", + "--format", + fmt, + ] + env = {**os.environ, "PYTHONHASHSEED": seed} + proc = subprocess.run(argv, capture_output=True, text=True, env=env) + return proc.returncode, proc.stdout + + +# --- SARIF 2.1.0 validity ------------------------------------------------------- # + + +def test_sarif_is_structurally_valid_2_1_0() -> None: + verdict = run_all(_fixture()) + log = json.loads(to_sarif(verdict, trace_uri="demo/traces/taint_flow.otlp.json")) + + assert log["$schema"].endswith("sarif-2.1.0.json") + assert log["version"] == "2.1.0" + assert isinstance(log["runs"], list) and len(log["runs"]) == 1 + + driver = log["runs"][0]["tool"]["driver"] + assert driver["name"] == "taintline" + rule_ids = {r["id"] for r in driver["rules"]} + assert all("shortDescription" in r for r in driver["rules"]) + + results = log["runs"][0]["results"] + assert results, "the taint flow must produce at least one result" + for res in results: + assert res["ruleId"] in rule_ids # every result references a declared rule + assert res["level"] in _VALID_LEVELS + assert res["message"]["text"] + loc = res["locations"][0]["physicalLocation"]["artifactLocation"]["uri"] + assert loc == "demo/traces/taint_flow.otlp.json" + + +def test_sarif_carries_the_taint_finding() -> None: + verdict = run_all(_fixture()) + log = json.loads(to_sarif(verdict, trace_uri="trace.json")) + taint = next(r for r in log["runs"][0]["results"] if r["ruleId"] == "taint") + + assert taint["level"] == "error" + assert taint["properties"]["span_ids"] == ["a-src", "a-sink"] + assert "a-src" in taint["message"]["text"] and "a-sink" in taint["message"]["text"] + names = {loc["name"] for loc in taint["locations"][0]["logicalLocations"]} + assert names == {"a-src", "a-sink"} + + +def test_sarif_byte_identical_across_hash_seeds(tmp_path: Path) -> None: + fixture = _write_fixture(tmp_path) + outputs = set() + for seed in ("0", "1", "42", "1000", "31337"): + code, out = _run_cli(fixture, "sarif", seed) + assert code == 1 # taint fires -> the gate still trips under --format sarif + json.loads(out) # valid JSON every time + outputs.add(out) + assert len(outputs) == 1 + + +# --- JSON stability ------------------------------------------------------------- # + + +def test_json_round_trips_and_is_stable_across_hash_seeds(tmp_path: Path) -> None: + fixture = _write_fixture(tmp_path) + outputs = set() + for seed in ("0", "1", "42", "1000", "31337"): + code, out = _run_cli(fixture, "json", seed) + assert code == 1 + payload = json.loads(out) + assert payload["tool"] == "taintline" + assert payload["count"] == len(payload["findings"]) + assert any(f["detector"] == "taint" for f in payload["findings"]) + outputs.add(out) + assert len(outputs) == 1 + + +# --- default text is unchanged -------------------------------------------------- # + + +def test_default_output_is_text_and_unchanged() -> None: + trace = _fixture() + verdict = run_all(trace) + default_code, default_out = run_check(trace, FAIL_ON) + text_code, text_out = run_check(trace, FAIL_ON, fmt="text") + + assert default_out == text_out == to_text(verdict) + assert default_code == text_code == 1 + # historical shape: an uppercase-severity line, then the count footer. + assert default_out.splitlines()[0].startswith("ERROR taint:") + assert default_out.endswith("finding(s)") + assert default_out != to_json(verdict) # default is not JSON From 73338b664eef7f60c32ebeede9d9c470f81eca8c Mon Sep 17 00:00:00 2001 From: Bamdad Dashtban Date: Wed, 22 Jul 2026 16:19:18 +0100 Subject: [PATCH 4/4] docs: document --format and the code-scanning (upload-sarif) wiring --- README.md | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/README.md b/README.md index d7e861d..b6e8746 100644 --- a/README.md +++ b/README.md @@ -82,6 +82,30 @@ Gate any repo on a captured agent trace: (comma-separated detectors; default `taint,loops`). A non-zero exit fails the step. Full runnable CI gate: [`demo/`](demo/). +## Output formats & GitHub code scanning + +`check` takes `--format {text,json,sarif}` (default `text`, byte-identical to +before). `json` is a stable, sorted view of the findings for scripting; `sarif` +emits **SARIF 2.1.0** for GitHub's Security tab. Every format is deterministic and +LLM-free — the same trace yields byte-identical bytes across runs. + +Wire the findings into code scanning with +[`github/codeql-action/upload-sarif`](https://github.com/github/codeql-action): + +```yaml +- name: taintline (SARIF) + run: uv run taintline check trace.otlp.json --format sarif > taintline.sarif + continue-on-error: true # still upload even when the gate trips +- name: Upload to code scanning + if: always() + uses: github/codeql-action/upload-sarif@v3 + with: + sarif_file: taintline.sarif +``` + +`check` still exits non-zero on a `--fail-on` hit regardless of format, so the +build can both **surface** findings in the Security tab and **gate** on them. + ## vs Strands Evals (read this first) Strands Evals scores agent quality and diagnoses failures; its **detectors** are LLM judges (default Bedrock Claude Sonnet 4.6, `ConfidenceLevel` threshold) —