CLI and API reference

CLI

throughline --help (or python -m throughline --help); every subcommand has its own --help.

throughline [--seed SEED] [--false-flag] [--version] COMMAND

demo                                   investigate the synthetic supply-chain intrusion (no downloads)
investigate QUERY [--min-conf X] [--as-of TIME]
                                       one-query investigation of a synthetic-world entity, as JSON
explain CLAIM_ID                       explain one claim's confidence (sources, corroboration, conflict)
export [--format json|cypher]          export the synthetic graph as JSON or Cypher
engines                                pinned sibling releases and what is installed (version, commit)
capture ID_OR_PATH [--data DIR] [--engines LIST] [--sigma core|all] [--top N] [--json]
                                       run every engine on a real OTRF capture and investigate it
captures [--data DIR]                  list the labelled OTRF captures under the dataset root
store {ingest,verify,head} DIR [FILES] [--allow-empty] [--expect-head HASH]
                                       append-only raw event store with a hash-chained ledger
serve [--host H] [--port P] [--capture ID] [--data DIR]
                                       REST API and console (needs the api extra)

Errors (an unknown entity or claim, missing datasets, a missing extra) end with a one-line message and exit code 2. serve binds to 127.0.0.1; on any other address without THROUGHLINE_API_TOKEN it generates a token and prints the console URL.

REST API

method	path	parameters	returns
GET	/health		mode and graph size (never needs a token)
GET	/connectors		registered connectors
GET	/engines		pinned and installed siblings, timings of the last run
POST	/ingest	JSON body {"records": [{"connector", "raw", "reliability"}]}	normalise and rebuild; at most 1,000 records per batch, 10 MB per body (413), 64 KB per raw record (422), 250,000 retained records (413); application/json only (415)
GET	/incidents	as_of	correlated incidents with fused technique confidence and cross-host cluster
GET	/investigator/{entity}	budget (1-20), as_of	the bounded, deterministic investigation: trace and cited report
GET	/investigate/{entity}	min_conf, as_of	facts about an entity or incident, each citing claims
GET	/graph/{key}	radius (1-2)	neighbourhood subgraph
GET	/claims/{id}/explain		how one claim's confidence was computed
POST	/neo4j/sync		mirror the graph, claims included, into Neo4j (THROUGHLINE_NEO4J_URI, THROUGHLINE_NEO4J_USER, THROUGHLINE_NEO4J_PASSWORD); 404 when no URI is set, 503 when the server is unreachable
GET	/ui		the investigation console (never needs a token)

as_of (ISO-8601) answers from the claims known at that time. With THROUGHLINE_API_TOKEN set, every path except /health and /ui requires Authorization: Bearer <token> (the console reads #token= from its URL). Only loopback Host headers are served (add names with THROUGHLINE_ALLOWED_HOSTS), and state-changing requests from another origin get 403.

Python API

throughline.contracts

Frozen contracts: canonical event schema, graph schema, claim model.

Everything else in THROUGHLINE is pluggable; these types are the only inter-component API (see docs/adr/0001-frozen-contracts.md).

Method

Bases: StrEnum

How a claim was obtained; each method has a prior in the confidence model.

Reliability

Bases: StrEnum

Admiralty-style source reliability (CTI A-F).

Layer

Bases: StrEnum

Which layer of the estate an event belongs to (code, build, runtime, network, identity, intel).

ContractError

Bases: ValueError

Input violates the frozen contracts (unknown type, bad method or grade, oversized attributes).

CanonicalEvent dataclass

One normalized event: actor -action-> object, with source, method, reliability, timestamp and a raw_ref back to the hashed raw record. Every event becomes one claim.

validate()

Check the event against the closed node/edge types, methods, grades and attribute caps.

Raises:

Type	Description
ContractError	on the first violation.

to_dict()

Plain-dict form (JSON-serialisable).

Claim dataclass

claim = (assertion, source, method, timestamp, confidence, corroboration[])

to_dict()

Plain-dict form (JSON-serialisable).

canonical_hash(obj)

SHA-256 of the canonical JSON form of a raw record (sorted keys, no whitespace).

node_key(node_type, node_id)

Graph key of a node: <Type>:<id> (e.g. Process:ws5/4200/rundll32.exe).

throughline.graph.KnowledgeGraph

Temporal security knowledge graph in which every node and edge is backed by claims.

All claims about one assertion are kept; confidence is recomputed by the confidence engine (noisy-OR over independent sources, discounted by contradictions) whenever evidence changes. Backed by a networkx MultiDiGraph; :mod:throughline.neo4j_adapter can mirror it.

add_claim(subj_type, subj_id, predicate, obj_type, obj_id, *, source, method, reliability, timestamp, evidence_ref=None, corroboration=None, score=None)

Record one claim. score lets an engine report its own probability for the assertion; the base confidence is then score x reliability weight instead of method prior x reliability weight (ADR-0006).

blast_radius(key, max_nodes=10000)

Everything causally downstream of key (annotation edges are not followed).

claims_for(s, predicate, o)

Every claim about the assertion (subject, predicate, object).

edge_confidence(s, o, predicate)

Current fused confidence of one edge.

explain_claim(cid)

How a claim's confidence was computed (see :func:throughline.confidence.explain).

finalize()

Recompute confidence for everything touched while defer_confidence was on.

ingest(ev)

Validate a canonical event, record it as a claim and attach its display attributes.

Returns:

Type	Description
Claim	The new claim.

neighborhood(key, radius=2)

Subgraph within radius hops of key (both directions).

root_cause_chain(key, max_hops=12)

Walk causal predecessors back to the earliest origin (longest ancestor chain).

Only causal predicates are followed (who spawned / built / introduced what); annotations such as ALERTED_ON, PART_OF or EXHIBITS are not causes.

set_attrs(key, attrs)

Attach display metadata (image, command line, ...) to a node. Not a claim.

stats()

Node, edge, claim and event counts.

tag_technique(subj_key, technique, reason, *, source, ts, corroboration=None, method='inferred', reliability='C', score=None)

Record <subject> EXHIBITS Technique:<id> (the id is canonicalised, revoked ids forwarded).

to_json()

Nodes, edges and claims as a JSON-serialisable dict.

throughline.confidence

Transparent, rule-based confidence & evidence engine.

Rules (documented in docs/adr/0003-confidence-model.md): base = METHOD_PRIOR[method] * RELIABILITY_WEIGHT[reliability] combined = 1 - prod(1 - c_i) over the claim and corroborating claims from independent sources (one best claim per source counts) final = combined * (1 - max(contradicting confidences) * CONFLICT_WEIGHT) Every number is explainable via explain().

base_confidence(method, reliability)

Base confidence of a claim from its method prior and Admiralty reliability weight.

brier(predictions)

Calibration metric for benchmarking claim confidence vs ground truth.

combine(primary, supporting, contradicting)

Final confidence of primary: noisy-OR over the best claim of each independent source, discounted by the strongest contradicting claim (ADR-0003).

combine_best(best, extra, worst_contra)

combine with the per-source maxima of an assertion's claims precomputed.

Every claim about one assertion shares the same independent-source maxima, so the graph computes them once per assertion (linear) instead of once per claim (quadratic when one process repeats the same action thousands of times).

explain(primary, supporting, contradicting)

Decompose a claim's confidence into base, independent sources, corroboration and conflict.

scored_confidence(score, reliability)

Base confidence for an engine that reports its own probability (ADR-0006).

throughline.reasoning.correlation

Correlation: many alerts -> few incidents (one incident = one graph cluster).

Reads only the graph. Every process carrying a technique claim (from any engine) or an alert is a signal. Signals are grouped by their story root: walk SPAWNED predecessors until the parent is a session/service boundary process (services.exe, explorer.exe, wmiprvse.exe, ...). Everything a user or service launched under one boundary child is one incident; this is the same notion of "story" REVENANT and ROOTLINE use, applied across engines.

Each incident gets

• <process> PART_OF Incident:<host>/<root> for every signal and every process on the path from the root down to it (source correlation),
• Incident EXHIBITS Technique roll-ups: the best claim of each independent engine across all members, combined by noisy-OR (the member claims are cited in evidence_ref, not re-counted), so intel engines read "what this incident did" in one hop and cross-engine agreement raises confidence, and
• display attributes: score (its most confident technique), breadth, pivot (the strongest member), alert/technique counts.

Cross-host stitching (:func:stitch) joins incidents on different hosts into one cluster through two kinds of evidence:

• shared rare destination - members of both incidents contacted the same remote endpoint (CONNECTED_TO an IPAddress or RESOLVED a Domain), so a beacon to one C2 from several hosts reads as one intrusion. Only connections made by non-system processes count (lsass.exe talking to the domain controller is not evidence), non-routable addresses (loopback, link-local, multicast) and localhost are ignored, as are Windows infrastructure ports (DNS, Kerberos, LDAP, RPC endpoint mapper and the dynamic RPC range, NetBIOS, SMB). A destination is rare if at most max_fanout incidents share it and at most half of the hosts (minimum 2) contacted it at all, counted over every process's traffic, not just incidents.
• lateral movement - a member on host A connected to host B on a remote administration port (SMB, RPC, WinRM, RDP, SSH), and an incident on B starts under a remote-execution service (wsmprovhost.exe, psexesvc.exe, wmiprvse.exe, winrshost.exe) within an hour of that connection. Addresses are mapped to hosts from the local address of each host's own outbound connections.

Each link is a proposal with its evidence (cluster_via), not a proven edge.

CorrelationEngine

Groups signal processes into incidents by story root, fuses each incident's technique claims across independent engines and stitches incidents across hosts (see module docs).

run(kg, context)

Write PART_OF and incident EXHIBITS claims; put incidents and clusters in context.

destinations(kg, members)

Remote endpoints (IPAddress/Domain nodes) contacted by non-system members that can count as stitching evidence.

evidence_destination(key)

Whether contacting key can count as shared-destination evidence at all.

host_addresses(kg)

local address -> host, from the local_ip of each host's outbound connections (an address seen on two hosts, e.g. a NAT or VPN address, is dropped).

host_fanout(kg)

destination -> hosts whose processes contacted it (all traffic), and every host seen.

host_of(key)

Host part of a Process: / Incident: key (host/pid/image or host/...).

image(key)

Lower-case image name of a Process: key (empty for other nodes).

lateral_links(kg, incidents, members)

(incident on A, incident on B, evidence) for every admin-port connection A -> B that is followed by an incident on B rooted under a remote-execution service.

noisy_or(values)

1 - prod(1 - v) over the values, capped at 0.99.

remote_entry(kg, root)

The story root is (or was spawned by) a remote-execution service.

routable(addr)

False for loopback, link-local, multicast, unspecified and broadcast addresses and localhost.

signal_techniques(kg, key)

technique -> {source: best base confidence} over the signal claims on one node.

split_endpoint(key)

IPAddress:10.0.0.4:88 -> ("10.0.0.4", 88); IPv6 keeps its colons; Domain:x -> ("x", None).

stitch(incidents, dests, max_fanout=3, host_fanout_of=None, n_hosts=0, links=None)

Union incidents on different hosts that share a rare destination or a lateral link.

dests maps incident key -> destination node keys. A destination joins incidents only if at most max_fanout incidents share it and (when host_fanout_of is given) at most max(2, n_hosts // 2) hosts contacted it in all. links are extra (incident, incident, evidence) joins. Returns incident key -> {"cluster": id, "hosts": [...], "via": [...]}.

story_root(kg, key, max_hops=64)

(root, path root->key) following SPAWNED parents up to a boundary process.

throughline.reasoning.investigator.Investigator

Bounded, deterministic, read-only investigation over graph queries.

It runs a fixed sequence of tools (incident, techniques, root cause, corroboration, attribution, posture, blast radius); every finding cites claim ids, and nothing is written to the graph.

investigate(query)

Run every tool on an entity or incident and return the cited report and the trace.

Parameters:

Name	Type	Description	Default
query	str	a node key, an id or a key suffix.	required

Returns:

Type	Description
dict	{"target", "incident", "steps", "hypotheses", "report"}; every step cites claim ids.

t_attribution(inc)

Competing ATTRIBUTED_TO hypotheses (UNKNOWN included) and their confidence.

t_blast_radius(key)

Everything reachable from the entity through causal edges.

t_corroboration(inc)

Which independent engines back the incident's strongest technique (read from the member claims the roll-up was fused from).

t_incident(inc)

Incident membership: how many processes, from which engines, under which story root.

t_posture(inc)

What should have detected the observed techniques and what was missed.

t_root_cause(key)

The causal chain back to the root cause, with the command lines along it.

t_techniques(inc)

The incident's strongest fused technique claims.

throughline.reasoning.calibration

Confidence calibration: measure it (Brier, ECE, reliability curve) and fit it.

ADR-0003 left the confidence constants as uncalibrated guesses. This module closes that TODO without hiding the model: calibration is a separate, fitted map from the transparent fused confidence to an empirical probability,

p = sigmoid(a * logit(confidence) + b)            (Platt scaling)

fitted on labelled data (OTRF captures with known emulated techniques) and evaluated only on held-out folds. The raw rule-based confidence stays in every claim; the calibrated value is reported next to it. Pure Python on purpose.

Platt dataclass

Platt scaling: p = sigmoid(a * logit(confidence) + b), fitted on labelled claims.

__call__(p)

Calibrated probability for a raw confidence.

fit(pairs, iters=100, l2=0.001) classmethod

Newton-Raphson logistic regression on logit(confidence), with Platt's smoothed targets to avoid overconfidence on small samples.

load(path) classmethod

Read a map written by :meth:save.

save(path)

Write the map as JSON.

auc(pairs)

ROC AUC via the rank statistic (ties count half).

brier(pairs)

Mean squared error between stated probabilities and outcomes (lower is better).

default_map()

The Platt map fitted by benchmarks/bench_otrf.py (shipped in the package), or None.

ece(pairs, bins=10)

Expected calibration error with equal-width bins.

reliability(pairs, bins=10)

Reliability-diagram bins: claims, mean stated confidence and accuracy per bin.

summary(pairs)

Count, positives, Brier score, ECE and AUC of (probability, outcome) pairs.

throughline.temporal

Temporal views of the claim graph: what the platform knew as of a point in time.

Every claim carries the event time of the evidence it rests on, so the graph is temporal by construction. Two kinds of claims are distinguished:

• first-order claims are made from one event at the time it happened: the normalised telemetry itself (observed) and per-event engine claims such as a Sigma alert or a provenance heuristic tag;
• derived claims are conclusions over many events (correlation incidents and their fused roll-ups, story membership, attribution, posture). Their meaning depends on everything seen so far, so they are not filtered by timestamp but re-derived.

:func:as_of keeps the first-order claims with timestamp <= ts (confidence recomputed from exactly that evidence) and drops the derived ones; :func:replay then re-runs the derived engines (by default correlation) on that view. The result is the investigation an analyst would have seen at ts.

as_of(kg, ts)

A new graph holding the first-order claims of kg made at or before ts.

Node display attributes are copied for the nodes that remain; confidence is recomputed from the remaining evidence only (corroboration links to claims that are not yet known are dropped).

parse_ts(ts)

Parse the ISO-8601 timestamps connectors emit (Z or offset, 0-9 fractional digits).

replay(kg, ts, engines=None)

:func:as_of plus the derived engines re-run on it (default: correlation).

throughline.stack

The full real-data stack: every installed sibling engine, in dependency order.

stack = Stack.from_data_dir()            # $THROUGHLINE_DATA or ../../datasets/throughline
kg, summary, ctx = stack.run(records)    # records = [("windows", raw, "B"), ...]

Order matters only through the graph: detection (ANVIL) and provenance (REVENANT) write technique claims -> correlation clusters them into incidents -> ROOTLINE re-derives incident membership from its own provenance graph -> intel (DRAGNET, OCCAM) attributes incidents -> posture (VANTAGE) checks what should have fired -> simulation (GAUNTLET) plans re-emulation of the gaps. Engines whose sibling package is not installed are skipped and reported.

DataPaths dataclass

Where each public dataset lives under the data root.

attack property

ATT&CK Enterprise v19.2 STIX bundle.

attack_old property

ATT&CK Enterprise v10.1 STIX bundle (temporal hold-out profiles).

baseline property

Optional benign Windows baseline (evtx-baseline).

cis property

CIS Controls v8 -> ATT&CK mapping workbook.

misp property

MISP galaxy threat-actor cluster.

otrf property

OTRF Security-Datasets root (atomic and compound captures).

sigma property

SigmaHQ checkout at the pinned commit.

missing()

Names of the required datasets that are absent.

Stack dataclass

Every installed sibling engine wired in dependency order over one data root.

Heavy objects (ATT&CK, the compiled Sigma library, DRAGNET/OCCAM knowledge bases, the VANTAGE catalog) are loaded once and reused across runs.

attack()

The ATT&CK catalog (loaded once, made the active catalog).

from_data_dir(root=None, **kw) classmethod

A stack over root (default: :func:data_dir).

registry()

A fresh engine registry; engines whose sibling is missing are recorded in skipped.

run(records, registry=None)

Build a graph from raw records and run every engine.

Returns:

Type	Description
tuple[KnowledgeGraph, dict, dict]	(graph, summary, context); the context holds incidents, clusters and engine objects.

sigma_library()

ANVIL's compiled Sigma library for the configured subset (loaded once).

capture_records(capture, reliability='B')

A capture's records in the (connector, raw, reliability) form :meth:Stack.run takes.

data_dir()

Dataset root: $THROUGHLINE_DATA; in a source checkout ../../datasets/throughline if that folder exists, else <checkout>/data; for an installed package ~/.throughline/data (never inside site-packages).

throughline.engines

Sibling portfolio projects wired in as capability engines.

Every engine implements the frozen Engine protocol (name, reads, writes, run(kg, context) -> list[Claim]) and talks to the rest of the platform only through the knowledge graph (ADR-0001). The sibling packages are optional extras installed from pinned release tags (pip install -e .[engines] in a checkout); an engine whose package is missing reports itself as unavailable instead of failing, so the core stays dependency-light.

The table below is the single source of truth for which sibling fills which slot, at which release tag, and which commit that tag pointed to when it was validated. pyproject.toml's engines/network extras must list exactly :attr:Sibling.pip_spec for each row (tests/test_engines.py checks), and scripts/check_sibling_tags.py fails when a newer release exists or a pinned tag has been moved to another commit.

EngineUnavailable

Bases: RuntimeError

A sibling package is not installed (install the engines extra).

Sibling dataclass

One sibling project and the release it is pinned to.

installed property

Whether the sibling's import package can be found.

pip_spec property

The pinned direct-URL requirement, exactly as the pyproject extra lists it.

repo property

GitHub URL of the sibling project.

repo_name property

GitHub repository name of the sibling project.

installed_commit()

Commit the installed distribution was built from (PEP 610 direct_url.json).

installed_version()

Version of the installed distribution, or None when it is not installed.

installed_versions()

Installed sibling versions and commits, recorded in every benchmark result file.

require(project)

Import a sibling's top-level package or raise :class:EngineUnavailable with the install hint.

sibling(project)

The :class:Sibling row for a project name (case-insensitive); KeyError if unknown.

status()

Installed / missing status of every sibling engine (for the CLI and API).

pinned is the release tag the table validates; installed_version and installed_commit describe what is actually installed (they differ from the pin when someone installed another ref by hand).

throughline.eventstore.EventStore

Append-only JSONL shards plus a hash-chained ledger.

Parameters:

Name	Type	Description	Default
root	str | Path	store directory.	required
shard_size	int	records per events-<n>.jsonl shard.	100000
create	bool	create root if it does not exist (False for read-only use such as verify; a missing directory then raises :class:StoreError).	True
key	bytes | str | None	HMAC key for ledger records (default $THROUGHLINE_LEDGER_KEY).	None

head property

Hash of the newest ledger record; record it off-host to detect tail changes later.

append(records, note='')

Store (connector, raw) records; return their raw_refs (store://<seq>#<sha256>).

get(ref)

The stored record a raw_ref points at; KeyError if absent or its digest differs.

iter()

Every stored record, shard by shard.

verify(allow_empty=False, expect_head=None)

Re-hash every record, replay the ledger chain and check its coverage.

ok requires: every record's hash matches its content, every ledger record chains to the previous one and digests exactly its batch, the batches cover sequence numbers 0..n-1 contiguously with no duplicates and no unledgered record, HMACs verify when a key is configured, the head equals expect_head when given, and (unless allow_empty) the store holds at least one record.

throughline.api.create_app(seed_demo=True, capture=None, data=None, token=None, hosts=None)

Build the API app.

Parameters:

Name	Type	Description	Default
seed_demo	bool	start from the synthetic supply-chain intrusion (ignored with capture).	True
capture	str | None	an OTRF capture id or path to serve through every installed engine.	None
data	str | None	dataset root for capture (default $THROUGHLINE_DATA).	None
token	str | None	bearer token (default $THROUGHLINE_API_TOKEN; None/empty disables auth).	None
hosts	list[str] | None	extra Host header names to accept (besides loopback and THROUGHLINE_ALLOWED_HOSTS).	None