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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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) —
Expand Down
44 changes: 31 additions & 13 deletions src/taintline/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]:
Expand All @@ -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)

Expand All @@ -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

Expand Down
133 changes: 133 additions & 0 deletions src/taintline/report.py
Original file line number Diff line number Diff line change
@@ -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)
Loading
Loading