Python API¶
Generated from the docstrings by mkdocstrings. The frozen contract models are in linchpin.models; everything that
touches the attack graph goes through linchpin.graph.store.GraphStore.
Contracts and configuration¶
linchpin.models ¶
Frozen contract models (see contracts/).
Kind
module-attribute
¶
Kind = Literal['cve', 'service', 'credential', 'acl', 'config', 'reachability']
NormalizedFinding ¶
Bases: BaseModel
The one schema every connector emits (contracts/finding.schema.json).
AttackPath ¶
Bases: BaseModel
An entry-to-crown-jewel path: node and edge ids in order, total cost, kill-chain stage per hop.
Remediation ¶
Bases: BaseModel
One fix in a plan: target node, action, paths broken out of the total, rationale and evidence.
BuildStats ¶
NodeDetail ¶
Bases: BaseModel
A node's properties and its inbound / outbound attack edges.
PathStats ¶
Bases: BaseModel
What-if result: paths and reachable crown jewels before and after removing nodes.
GroundTruth ¶
Bases: BaseModel
Answer key of a synthetic topology: planted linchpin / cut, entry points, crown jewels, provenance.
planted_cut
class-attribute
instance-attribute
¶
planted_cut: list[str] = Field(default_factory=list)
make_finding_id ¶
make_finding_id(host_id: str, kind: str, key: str) -> str
Stable finding id: the first 16 hex digits of sha256(host_id|kind|key).
detail_problem ¶
detail_problem(f: NormalizedFinding) -> str | None
Why f.detail cannot be used to build the attack graph, or None when it is usable.
The schema leaves detail free-form; the graph builder needs these keys per kind:
credential -> principal (and a list valid_on if given); acl -> principal
plus target (AdminTo) or target_name (abusable ACE); reachability ->
from_segment and to_segment (and a list of integer ports if given); inventory
config -> datastores as a list of objects with a name.
linchpin.config ¶
Configuration model (contracts/config.schema.json) and its YAML loader.
Weights ¶
Config ¶
Bases: BaseModel
Everything that shapes the attack graph and the plan; every field has a default.
skill_penalty
class-attribute
instance-attribute
¶
skill_penalty: dict[str, float] = Field(default_factory=lambda: {'network_exploit': 0.1, 'cred_reuse': 0.2, 'client_side': 0.4, 'privesc': 0.3, 'acl_abuse': 0.25})
entrypoints
class-attribute
instance-attribute
¶
entrypoints: list[str] = Field(default_factory=lambda: ['auto:internet_facing'])
crown_jewels
class-attribute
instance-attribute
¶
crown_jewels: list[str] = Field(default_factory=lambda: ['auto:sensitivity=high'])
prerequisite_penalty
class-attribute
instance-attribute
¶
prerequisite_penalty: float = Field(0.5, ge=0.0, le=1.0)
fix_cost
class-attribute
instance-attribute
¶
fix_cost: dict[str, float] = Field(default_factory=lambda: {'Vuln': 1.0, 'Credential': 1.0, 'Ace': 1.0, 'Host': 3.0})
neo4j
class-attribute
instance-attribute
¶
neo4j: dict[str, str] = Field(default_factory=lambda: {'uri': 'bolt://localhost:7687', 'user': 'neo4j'})
load_config ¶
load_config(path: str | Path | None = None) -> Config
Read a YAML config (missing file or None -> defaults); unknown keys are ignored.
Loading data¶
linchpin.scenario ¶
Scenario loader: real exported findings + a declared topology overlay + public exploit intel.
A scenario YAML ties together::
name: composite-lab
sources: # exported files, relative to --data-dir
- scans/openvas/many_vuln.xml
- {path: bloodhound/v6, connector: bloodhound}
intel: derived/cve_intel.csv.gz # optional; built by `linchpin intel-build`
model: derived/exploit_model.npz # optional M11 model (benchmarks/ml_exploitability.py)
hosts: [...] # topology overlay (see connectors/inventory.py)
reachability: [...]
credentials: [...]
crown_jewels: ["auto:sensitivity=high"] # optional config overrides
entrypoints: ["auto:internet_facing"]
Scanner host ids listed under a host's match: are renamed to that host's id, so findings
from different tools about the same machine merge onto one node.
CONFIG_KEYS
module-attribute
¶
CONFIG_KEYS = ('crown_jewels', 'entrypoints', 'k_shortest', 'exploitability_source')
apply_aliases ¶
apply_aliases(findings: list[NormalizedFinding], alias: dict[str, str]) -> list[NormalizedFinding]
Rename scanner host ids to their canonical topology ids (match: lists), re-keying ids.
load_scenario ¶
load_scenario(path: str | Path, data_dir: str | Path | None = None, use_intel: bool = True, confine: bool = False) -> tuple[list[NormalizedFinding], Config, dict]
Load a scenario YAML: its exports, the topology overlay and (optionally) exploit intel.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str | Path
|
the scenario YAML. |
required |
data_dir
|
str | Path | None
|
base directory for the files it lists (default: the YAML's folder). |
None
|
use_intel
|
bool
|
enrich CVE findings from the |
True
|
confine
|
bool
|
refuse file references that are absolute or escape |
False
|
Returns:
| Type | Description |
|---|---|
list[NormalizedFinding]
|
|
Config
|
findings, unmatched |
Raises:
| Type | Description |
|---|---|
PathNotAllowed
|
with |
FileNotFoundError
|
a listed export does not exist. |
linchpin.connectors ¶
M1 connectors: raw tool exports -> NormalizedFinding[]. Offline file parsers only.
CONNECTORS
module-attribute
¶
CONNECTORS: dict[str, Callable[[str], list[NormalizedFinding]]] = {'json': native.parse, 'nmap': nmap.parse, 'openvas': openvas.parse, 'nessus': nessus.parse, 'bloodhound': bloodhound.parse, 'inventory': inventory.parse}
parse_any ¶
parse_any(path: str) -> list[NormalizedFinding]
Parse path with the connector :func:detect picks for it.
linchpin.connectors.bloodhound ¶
SharpHound / BloodHound CE JSON (v5/v6 ingest format) -> identity-layer findings.
Pass the path of computers.json (or a directory holding the collection); sibling
users.json / groups.json / domains.json are loaded from the same directory.
Only the attacker-relevant subset is modelled:
- local
Administratorsmembership (nested groups expanded) ->aclAdminTo Remote Desktop Users/Remote Management Users->aclCanRDP / CanPSRemote- Domain Admins / Enterprise Admins / Administrators members -> AdminTo every computer
- interactive / privileged / registry sessions ->
credentialcached on that computer (cred_type=hash),valid_on= hosts the user administers -
each computer ->
configinventory (segmentad; domain controllers flagged and given anntdsdatastore) -
abusable ACEs (
Aceson users / groups / computers / domains) ->aclfindings withace=True: control of a user (GenericAll, ForceChangePassword, ...) -> that user's credential; write access to a group -> the group's effective admin rights (and the NTDS if the group is privileged); control of a computer (GenericAll, AddAllowedToAct/RBCD, AllExtendedRights/LAPS ...) -> admin on it; DCSync (GetChanges + GetChangesAll) or full control of the domain -> the domain controllers' NTDS.
ADCS (ESC1-8), GPO links, shadow credentials and trust edges are not modelled.
LOCAL_GROUP_RIGHT
module-attribute
¶
LOCAL_GROUP_RIGHT = {'-544': 'AdminTo', '-555': 'CanRDP', '-580': 'CanPSRemote'}
ABUSABLE
module-attribute
¶
ABUSABLE = {'user': _CONTROL | {'GenericWrite', 'ForceChangePassword', 'AllExtendedRights'}, 'group': _CONTROL | {'GenericWrite', 'AddMember', 'AddSelf', 'AllExtendedRights'}, 'computer': _CONTROL | {'GenericWrite', 'AllExtendedRights', 'AddAllowedToAct', 'ReadLAPSPassword'}, 'domain': _CONTROL | {'AllExtendedRights', 'GetChangesAll'}}
is_bloodhound ¶
is_bloodhound(path: str | Path) -> bool
True if path is a SharpHound/BloodHound JSON file or a folder of them.
parse ¶
parse(path: str) -> list[NormalizedFinding]
Parse a SharpHound v5/v6 collection (file or folder) into admin rights, sessions, credentials and ACEs.
linchpin.connectors.inventory ¶
Host-inventory / topology overlay (YAML or JSON) -> inventory + reachability findings.
Scanner exports know hosts and services but not where a host sits, what it holds or what the firewall allows. This file supplies that context::
hosts:
- id: web-01
match: [192.168.1.10, testphp.vulnweb.com] # scanner host ids to rename to `id`
segment: dmz
internet_facing: true
os: linux
datastores: [{name: customer-db, sensitivity: high}]
reachability:
- {from: dmz, to: internal, ports: [445, 3389]} # empty ports = any
credentials:
- {principal: svc_backup, cred_type: password, stored_on: web-01, valid_on: [db-01]}
aliases(path) returns the scanner-id -> id rename map used by :mod:linchpin.scenario.
load ¶
load(path: str | Path) -> dict
Read a topology / inventory document (YAML, or JSON by extension).
Raises:
| Type | Description |
|---|---|
ValueError
|
the document is not valid YAML / JSON (message names the line and column). |
is_inventory ¶
is_inventory(path: str | Path) -> bool
True if path is a YAML document with a hosts list.
aliases ¶
aliases(path_or_doc) -> dict[str, str]
Scanner host id -> canonical host id, from every host's match: list.
parse ¶
parse(path: str) -> list[NormalizedFinding]
Inventory, reachability and credential findings of a topology file.
from_doc ¶
from_doc(doc: dict) -> list[NormalizedFinding]
Inventory, reachability and credential findings of an already-loaded topology document.
linchpin.intel.cpe ¶
Offline version -> CVE matching from scanner-detected services (no NSE scripts, no APIs).
A scanner's service detection (nmap -sV) reports a product, a version and usually a CPE
such as cpe:/a:apache:http_server:2.4.49. :class:CpeIndex matches that against the
version ranges NVD publishes for each CVE (configurations[].nodes[].cpeMatch), using an
index built offline from the local NVD feeds (scripts/build_cpe_index.py) and shipped as
package data with each CVE's CVSS / EPSS / KEV values. :func:match_services turns matched
services into cve findings, one per service (fixing it means upgrading the software), so
a scan can be analysed without any network lookup.
Limits, stated plainly: a banner version is not proof of vulnerability (backported fixes, build options and configuration are invisible to it), configurations that only apply on a specific platform are ignored by default, and products outside the index are not matched.
ALIASES
module-attribute
¶
ALIASES: dict[str, tuple[str, ...]] = {'apache:http_server': ('apache:http_server',), 'apache:tomcat': ('apache:tomcat',), 'igor_sysoev:nginx': ('f5:nginx', 'nginx:nginx'), 'nginx:nginx': ('f5:nginx', 'nginx:nginx'), 'f5:nginx': ('f5:nginx', 'nginx:nginx'), 'redislabs:redis': ('redis:redis', 'redislabs:redis'), 'redis:redis': ('redis:redis', 'redislabs:redis'), 'mysql:mysql': ('oracle:mysql', 'mysql:mysql', 'oracle:mysql_server'), 'oracle:mysql': ('oracle:mysql', 'mysql:mysql', 'oracle:mysql_server'), 'mariadb:mariadb': ('mariadb:mariadb',), 'postgresql:postgresql': ('postgresql:postgresql',), 'openbsd:openssh': ('openbsd:openssh',)}
PRODUCT_NAMES
module-attribute
¶
PRODUCT_NAMES: dict[str, str] = {'apache httpd': 'apache:http_server', 'apache tomcat': 'apache:tomcat', 'nginx': 'nginx:nginx', 'redis key-value store': 'redis:redis', 'mysql': 'mysql:mysql', 'mariadb': 'mariadb:mariadb', 'postgresql db': 'postgresql:postgresql', 'openssh': 'openbsd:openssh'}
CpeRow
dataclass
¶
One vulnerable cpeMatch of a CVE: a product, an exact version or a range, and the CVE's scores.
covers ¶
covers(version: str) -> bool
True if version satisfies this row's exact version or range bounds.
CpeIndex ¶
The packaged CPE -> CVE index (scripts/build_cpe_index.py).
compare_versions ¶
compare_versions(a: str, b: str) -> int
Compare two version strings: -1, 0 or 1.
Numbers compare numerically and trailing zeros do not matter (1.0 == 1.0.0); pre-release words (rc, beta, M) sort before the release they precede (1.0rc1 < 1.0); patch words (OpenSSH's p1) sort after it.
parse_cpe ¶
parse_cpe(cpe: str) -> tuple[str, str, str] | None
(vendor:product, version, part) from a CPE 2.2 URI (cpe:/a:v:p:1.2) or 2.3 string.
match_services ¶
match_services(findings: Iterable[NormalizedFinding], index: CpeIndex | None = None, include_conditional: bool = False) -> tuple[list[NormalizedFinding], dict]
cve findings for every service whose product version matches NVD version ranges.
One finding per (host, port): fixing it means upgrading the software. detail.cves lists
every matched CVE; cve_id and the CVSS / EPSS / KEV inputs of the edge cost are those
of the driver (:func:_driver: a code-execution CVE by vector first, then KEV, then
EPSS), so a KEV-listed denial-of-service does not floor a code-execution edge. All KEV
members are kept in detail.kev_cves; detail.match = "cpe-range".
linchpin.intel.store ¶
CveIntel: compact CVE -> (CVSS, EPSS, KEV) lookup, built from the public feeds.
COLUMNS
module-attribute
¶
COLUMNS = ['cve', 'published', 'cvss_version', 'cvss_base', 'cvss_vector', 'cvss_exploitability', 'av', 'ac', 'pr', 'ui', 'cwe', 'epss', 'kev', 'kev_date', 'ransomware', 'description']
CveRecord
dataclass
¶
One CVE's public intel: CVSS (version, vector, sub-score), EPSS, KEV status and date, text.
CveIntel ¶
In-memory lookup. CveIntel.load(path) reads the cache written by build.
build
classmethod
¶
build(data_dir: str | Path, out: str | Path | None = None) -> CveIntel
Build the lookup from the downloaded NVD / EPSS / KEV feeds and optionally write the cache.
load
classmethod
¶
load(path: str | Path, only: set[str] | None = None) -> CveIntel
Load the cache; only restricts to a CVE-id subset (much less memory).
linchpin.intel.enrich ¶
Enrich NormalizedFindings with real CVSS / EPSS / KEV data.
Fills gaps only; scanner-reported CVSS is kept unless prefer_intel=True.
enrich ¶
enrich(findings: list[NormalizedFinding], intel: CveIntel, prefer_intel: bool = False, model=None) -> tuple[list[NormalizedFinding], dict]
model (optional, linchpin.ml.exploitability.ExploitModel) adds detail.exploitability_learned.
Attack graph¶
linchpin.graph.store ¶
M2: GraphStore, the in-memory NetworkX attack graph (Neo4j mirror and GDS: neo4j_store.py).
This is the ONLY module that knows how the graph is stored; engines go through it.
STAGE_BY_REL
module-attribute
¶
STAGE_BY_REL = {'CAN_REACH': 'lateral', 'HAS_VULN': 'exploit', 'ENABLES': 'exploit', 'LEADS_TO': 'privesc', 'STORED_ON': 'privesc', 'GRANTS': 'lateral', 'HOLDS': 'objective', 'HAS_ACE': 'lateral', 'ABUSES': 'privesc'}
CLASS_BY_REL
module-attribute
¶
CLASS_BY_REL = {'ENABLES': 'network_exploit', 'GRANTS': 'cred_reuse', 'STORED_ON': 'privesc', 'ABUSES': 'acl_abuse'}
GraphTooLarge ¶
Bases: ValueError
Building the attack graph would exceed the caller's edge budget.
GraphStore ¶
In-memory attack graph (NetworkX DiGraph) built from :class:NormalizedFinding records.
Nodes carry a label from the frozen taxonomy (contracts/graph_model.md); every edge
carries rel, a stable id and the deterministic cost of contracts/edge_cost.md.
Engines (paths, cuts, optimizer, explain) only read g through this class.
upsert_findings ¶
upsert_findings(findings: Iterable[NormalizedFinding]) -> None
Add or replace findings by finding_id (the graph is rebuilt by :meth:build_attack_graph).
build_attack_graph ¶
build_attack_graph(max_edges: int | None = None) -> BuildStats
(Re)build the attack graph from the current findings, replacing self.g.
Findings whose detail lacks what their kind needs (:func:linchpin.models.detail_problem)
are skipped with a warning instead of failing the build.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
max_edges
|
int | None
|
when given, raise :class: |
None
|
Returns:
| Type | Description |
|---|---|
BuildStats
|
Node and edge counts and the build time in milliseconds. |
entrypoints ¶
entrypoints() -> list[str]
Entry nodes from cfg.entrypoints (auto:internet_facing -> the virtual internet node).
crown_jewels ¶
crown_jewels() -> list[str]
Crown-jewel nodes from cfg.crown_jewels (auto:sensitivity=high -> high datastores).
k_shortest_paths ¶
k_shortest_paths(sources=None, targets=None, k: int | None = None, exclude: Iterable[str] = ()) -> list[AttackPath]
Yen's k-shortest simple paths (by summed edge cost) from any source to any target.
reachable_crown_jewels ¶
reachable_crown_jewels(exclude: Iterable[str] = ()) -> list[str]
Crown jewels reachable from any entrypoint with exclude removed (no graph copy).
node ¶
node(node_id: str) -> NodeDetail
One node with its properties and inbound / outbound attack edges.
remove_nodes_view ¶
remove_nodes_view(ids: list[str], k: int = 100) -> PathStats
What-if: recompute path stats with nodes removed. Never mutates the store.
edge_id ¶
edge_id(u: str, rel: str, v: str) -> str
Stable relationship id used in AttackPath.edges and remediation evidence.
path_from_nodes ¶
path_from_nodes(g: DiGraph, nodes: list[str], total_cost: float | None = None) -> AttackPath
Build an :class:AttackPath for an entry -> crown-jewel node sequence of g.
The path id hashes the node sequence, so a path gets the same id whichever backend
(NetworkX or Neo4j GDS) enumerated it. total_cost defaults to the summed edge costs.
linchpin.graph.neo4j_store ¶
Neo4j adapter for the attack graph (optional: the [neo4j] extra, pip install ".[neo4j]").
Design (see docs/adr/0002-neo4j-adapter.md): the attack graph is built by the same code as
the in-memory store (so edge costs are identical and deterministic) and then mirrored into
Neo4j with idempotent MERGE batches, using exactly the frozen labels / relationship types
of contracts/graph_model.md. Neo4j is then the system of record for exploration (Browser,
Bloom, GDS); :meth:Neo4jGraphStore.pull reads a graph back into NetworkX so every engine
(paths, optimizer, cuts) runs unchanged on data that lives in Neo4j.
Without a server, :func:to_cypher renders the same statements as a .cypher script that
cypher-shell -f can load.
With the Graph Data Science plugin installed, :meth:Neo4jGraphStore.gds_k_shortest_paths
runs Yen's k-shortest paths inside Neo4j (gds.shortestPath.yens) over the same cost
property. CI cross-checks it against the in-memory engine (same path ids and costs) on
synthetic graphs with up to ~240k relationships and records both backends' timings.
LABELS
module-attribute
¶
LABELS = {'Host', 'Service', 'Vuln', 'Credential', 'Privilege', 'DataStore', 'Internet', 'Ace'}
GDS_PROJECT
module-attribute
¶
GDS_PROJECT = 'MATCH (s:LinchpinNode {graph: $graph}) OPTIONAL MATCH (s)-[r]->(t:LinchpinNode {graph: $graph}) WITH gds.graph.project($name, s, t, {relationshipProperties: r {.cost}}) AS g RETURN g.graphName AS name, g.nodeCount AS nodes, g.relationshipCount AS rels, g.projectMillis AS ms'
GDS_YENS
module-attribute
¶
GDS_YENS = "MATCH (a:LinchpinNode {graph: $graph, id: $src}), (b:LinchpinNode {graph: $graph, id: $dst}) CALL gds.shortestPath.yens.stream($name, {sourceNode: a, targetNode: b, k: $k, relationshipWeightProperty: 'cost'}) YIELD index, totalCost, nodeIds RETURN index, totalCost, [n IN gds.util.asNodes(nodeIds) | n.id] AS ids ORDER BY index"
Neo4jGraphStore ¶
Bases: GraphStore
GraphStore whose built graph is mirrored to (and can be re-read from) Neo4j.
connect
classmethod
¶
connect(cfg: Config, password: str, graph_name: str = 'default', **kw) -> Neo4jGraphStore
Open a driver for cfg.neo4j and verify that the server answers and accepts the login.
Raises:
| Type | Description |
|---|---|
ServiceUnavailable
|
nothing answers at the configured URI. |
AuthError
|
the credentials are rejected. |
build_attack_graph ¶
build_attack_graph(max_edges: int | None = None) -> BuildStats
Build in memory (same code as GraphStore), then mirror into Neo4j when connected.
push ¶
push() -> int
Mirror the current in-memory attack graph into Neo4j. Returns statements executed.
pull ¶
pull() -> nx.DiGraph
Replace the in-memory graph with the one stored in Neo4j (for engines to run on).
gds_version ¶
gds_version() -> str | None
Installed Graph Data Science version, or None when the server reports gds.* unknown.
Only "unknown function / procedure" errors mean the plugin is missing. Connection and authentication failures propagate, so a wrong URI or password is never reported as "GDS not installed".
gds_project ¶
gds_project(name: str | None = None) -> dict
(Re)create an in-memory GDS projection of this graph with cost on every relationship.
gds_k_shortest_paths ¶
gds_k_shortest_paths(sources: list[str] | None = None, targets: list[str] | None = None, k: int | None = None, name: str | None = None) -> list[AttackPath]
Yen's k-shortest paths computed by Neo4j GDS (same semantics as k_shortest_paths).
GDS Yen is single-source/single-target, so every (entrypoint, crown jewel) pair is solved
and the union is cut to the k cheapest -- the global k shortest are among each pair's k
shortest. Requires :meth:gds_project first; edge ids and kill-chain stages are filled in
from the in-memory graph (call :meth:pull first when the graph only lives in Neo4j).
graph_rows ¶
graph_rows(g: DiGraph) -> tuple[dict[str, list[dict]], dict[str, list[dict]]]
Group nodes by label and edges by relationship type, with Neo4j-safe properties.
statements ¶
statements(g: DiGraph, graph_name: str = 'default') -> list[tuple[str, dict]]
Parameterised, idempotent Cypher statements that mirror g into Neo4j.
to_cypher ¶
to_cypher(g: DiGraph, graph_name: str = 'default') -> str
Render the mirror statements as a standalone cypher-shell -f script (parameters inlined).
Parameters are substituted in one pass over the query template only, so data that happens
to contain $graph or $rows (e.g. a scanned service banner) is never re-substituted.
Engines¶
linchpin.engine.edge_cost ¶
M3: pure edge-cost function (contracts/edge_cost.md).
EdgeContext ¶
Bases: BaseModel
Everything the cost of one attacker transition depends on (contracts/edge_cost.md).
exploitability ¶
exploitability(ctx: EdgeContext) -> float
Exploitability in [0, 1] of an edge.
An explicit override (e.g. the learned model) wins; non-ENABLES edges are 1.0;
otherwise 0.6 x CVSS exploitability sub-score + 0.4 x EPSS, with whichever is known
alone, 0.5 when neither is, and a floor of 0.95 for CISA KEV entries.
edge_cost ¶
edge_cost(ctx: EdgeContext, cfg: Config) -> float
Cost in [0, 1] of one attacker transition; lower means easier.
(w1 (1 - exploitability) + w2 (1 - prerequisite_match) + w3 skill_penalty) / (w1 + w2 + w3),
clamped and rounded to 6 decimals. Pure: same inputs, same output.
linchpin.engine.paths ¶
M4: path ranking and per-node criticality.
rank_paths ¶
rank_paths(store, cfg: Config | None = None, k: int | None = None) -> list[AttackPath]
The k cheapest entry-to-crown-jewel paths, ordered by (cost, length, path id).
node_criticality ¶
node_criticality(paths: list[AttackPath]) -> dict[str, float]
0.7 * fraction-of-paths-through-node + 0.3 * normalised betweenness on the path subgraph.
Endpoints (entry and crown jewel) are excluded: they are trivially on every path.
linchpin.engine.cuts ¶
Exact graph-cut oracles over the attack graph.
- :func:
chokepoints-- every remediable node that, removed alone, disconnects all entrypoints from all crown jewels (dominators of a virtual sink, O(E alpha(E))). - :func:
min_remediation_cut-- the minimum number (or, weighted, total effort) of remediable nodes whose removal disconnects everything (vertex-split max-flow / min-cut). This is the optimum the greedy optimizer approximates, used by the benchmark to report an optimality gap.
Only remediable nodes (Vuln, Credential, non-endpoint Host -- see optimizer.candidates) may be cut; structural nodes (services, privileges, the internet) get infinite capacity.
chokepoints ¶
chokepoints(store, candidates: list[str] | None = None) -> list[str]
Remediable nodes lying on every entry -> crown-jewel route (single points of failure).
fix_cost ¶
fix_cost(store, node_id: str) -> float
Remediation effort of one candidate node (config fix_cost by label, default 1).
min_remediation_cut ¶
min_remediation_cut(store, candidates: list[str] | None = None, weighted: bool = False) -> list[str] | None
Minimum set of remediable nodes disconnecting all crown jewels.
weighted=False minimises the number of fixes; weighted=True minimises total
remediation effort (:func:fix_cost, e.g. a segmentation rule costs 3 patches).
Returns [] if nothing is reachable already, or None if no finite cut exists (e.g. an
entrypoint service leads straight to a crown jewel with nothing remediable in between).
linchpin.engine.optimizer ¶
M5: greedy set-cover remediation optimizer.
action_for ¶
action_for(store, node_id: str) -> str
The remediation verb for a node: patch / upgrade, rotate credential, remove ACE, segment host.
candidates ¶
candidates(store) -> list[str]
Remediable nodes: vulns, credentials, ACEs and hosts that are neither entry points nor crown jewels.
recommend ¶
recommend(store, paths: list[AttackPath] | None = None, budget: int = 5, cfg: Config | None = None, k: int = 100, exact: bool = True, restrict: list[str] | None = None) -> list[Remediation]
Pick <= budget remediations that break the most crown-jewel attack paths.
Greedy set cover: repeatedly pick the remediable node on the most still-viable paths.
With exact=True (default) the exact minimum vertex cut over remediable nodes is also
computed (max-flow, engine/cuts.py). If it fits in the budget and greedy does not already
disconnect everything with that many fixes, the greedy is re-run restricted to the cut, so
the plan provably disconnects every crown jewel with the fewest possible fixes.
Ties are broken by (fewest crown jewels still reachable after removal, node id), which
prefers true graph cuts over nodes that merely appear on every enumerated path.
When all enumerated paths are broken, paths are re-enumerated on the reduced graph.
restrict limits the plan to these remediable nodes (e.g. only vulnerabilities: a
patch-only plan); the exact cut is then the minimum cut over them.
linchpin.engine.explain ¶
M6: deterministic, template-based explanations (no LLM).
explain_remediation ¶
explain_remediation(r: Remediation, store, paths: list[AttackPath] | None = None, cuts_all: bool = False) -> str
Plain-English rationale for one remediation, from templates (no LLM).
Names what the fix removes (pivot host, credential cache, vuln, ACE), where it sits, and how many of the enumerated paths to which crown jewels it breaks.
explain_path ¶
explain_path(p: AttackPath, store) -> str
One-line hop list of an attack path with relation and kill-chain stage per hop.
linchpin.engine.interdiction ¶
Budgeted shortest-path interdiction: comparison planners from the network-interdiction literature.
LINCHPIN's optimizer cuts paths (exact vertex min cut, else greedy set cover). The
interdiction literature instead maximises the attacker's cheapest path cost after at most
budget removals, which is the right objective when no plan within budget can disconnect.
- :func:
greedy_interdiction-- each round, remove the remediable node on the current cheapest attack path whose removal raises that cost the most (disconnecting = infinitely much). This is the shape of the GREEDY blocking baseline used for Active Directory attack graphs by Guo et al. (AAAI 2022, 2023); it is a re-implementation of the idea on LINCHPIN's graph and cost model, not a reproduction of their system. - :func:
exact_interdiction-- the optimum, from the mixed-integer program of Israeli & Wood, "Shortest-path network interdiction", Networks 40(2), 2002 (doi:10.1002/net.10039): the inner shortest path is replaced by its LP dual (node potentials), node removal is an interdicted split arc with a big-M length, and the MILP is solved with HiGHS through :func:scipy.optimize.milp(needs the[ml]extra for scipy).
Both use the same remediable candidates as the optimizer and the same edge costs, so their
plans are scored with exactly the metrics of :mod:linchpin.benchmark.
cheapest_path ¶
cheapest_path(store, exclude: Iterable[str] = ()) -> tuple[float, list[str]]
Cost and node list of the cheapest entry -> crown-jewel path with exclude removed.
Returns (inf, []) when no crown jewel is reachable.
greedy_interdiction ¶
greedy_interdiction(store, budget: int, candidates: Iterable[str] | None = None) -> list[str]
Greedy shortest-path interdiction (Guo et al.-style GREEDY): up to budget node ids.
exact_interdiction ¶
exact_interdiction(store, budget: int, candidates: Iterable[str] | None = None, time_limit: float = 60.0) -> dict
Optimal budget-node interdiction of the cheapest attack path (Israeli & Wood MILP).
Returns:
| Type | Description |
|---|---|
dict
|
|
dict
|
|
dict
|
disconnect every crown jewel) and |
Synthetic data, benchmark and learned exploitability¶
linchpin.synth.topologies ¶
Benchmark topology families with REAL vulnerability parameters.
Every vuln planted here is drawn from synth/data/cve_pool.csv (shipped as package data):
a seeded, KEV-enriched stratified sample of 3,000 real CVEs (300 in KEV, i.e. 10% against
about 0.5% in the source population) with their NVD CVSS vector / exploitability sub-score,
FIRST EPSS and CISA KEV status. The CVSS / EPSS / KEV baselines therefore rank real score
values; the KEV share is deliberately inflated so KEV-first queues have KEV entries to pick.
Hosts, segments and credentials are synthetic. A missing pool is an error unless the caller
opts into placeholder CVE-2099-* ids with allow_synthetic=True.
Families (family argument):
single-- one planted pivot (bastion) between DMZ and everything else.multi-- 2-3 parallel bastions and two independent routes into the DB tier (vuln + cached credential): no single-node cut, minimum cut is 2-3 nodes.none-- flat network: several internet-facing hosts reach the data tier directly, the DB exposes several exploitable services and its admin credential is cached widely. There is no small cut; budgets cannot disconnect it.ad-- tiered Active-Directory estate: phished workstations, helpdesk local-admin reuse, server-admin and Domain-Admin sessions cached on lower tiers, DC holds NTDS. Chokepoint is usually the DA credential; half the seeds add an exploitable DC service (e.g. a KEV-listed RCE) that removes it.
All families add two kinds of planted distractors, and the benchmark reports them:
- decoys -- two isolated hosts (
legacy-00,legacy-01) in a segment nothing can reach, each with a KEV-listed remote-code-execution CVE of CVSS >= 9.0; - noise -- high-CVSS (>= 6.5) vulns without code execution on reachable hosts.
Both are what a score-sorted patch queue spends its budget on; the reachability-filtered
baselines in :mod:linchpin.benchmark remove that advantage.
pool_provenance ¶
pool_provenance(path: str | Path | None = None) -> str
sha256:<first 16 hex> of the CVE pool file, or synthetic-fallback when it is absent.
Line endings are normalised first, so a Windows checkout (CRLF) hashes like the wheel (LF).
load_pool
cached
¶
load_pool(path: str | None = None, allow_synthetic: bool = False) -> dict[str, list[dict]]
Real-CVE parameters by impact class (rce / info / local / rce_kev).
Raises:
| Type | Description |
|---|---|
FileNotFoundError
|
the pool file is missing and |
generate_family ¶
generate_family(family: str, n_hosts: int = 20, seed: int = 0, pool_path: str | None = None, allow_synthetic: bool = False) -> tuple[list[NormalizedFinding], GroundTruth]
Generate one seeded topology of family with about n_hosts hosts (minimum 8).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
family
|
str
|
one of :data: |
required |
n_hosts
|
int
|
approximate host count; decoys and the crown-jewel host come on top. |
20
|
seed
|
int
|
RNG seed; the same (family, n_hosts, seed, pool) always gives the same findings. |
0
|
pool_path
|
str | None
|
alternative CVE pool CSV (default: the packaged one). |
None
|
allow_synthetic
|
bool
|
permit placeholder CVEs when the pool is missing (warns). |
False
|
Returns:
| Type | Description |
|---|---|
tuple[list[NormalizedFinding], GroundTruth]
|
|
linchpin.benchmark ¶
Remediation-ordering benchmark.
For each topology, every strategy proposes an ordered list of remediations; we apply its top
budget fixes and measure what an attacker still has:
disconnected-- no crown jewel reachable from any entrypoint any moreresidual-- number of distinct attack paths still enumerated (capped atk)min_cost-- cheapest remaining attack path cost (inf when disconnected)
Strategies (:data:STRATEGIES):
linchpin-- exact minimum vertex cut when it fits the budget, else greedy path set cover (this project);greedy-- the set cover alone.milp_interdiction/greedy_interdiction-- published shortest-path interdiction planners (:mod:linchpin.engine.interdiction): the exact budgeted MILP of Israeli & Wood (2002) and a Guo et al.-style greedy. They maximise the attacker's cheapest-path cost.cvss/epss/kev_epss-- score-sorted patch queues over all vulns (highest CVSS; highest EPSS; CISA KEV first, then EPSS -- BOD 22-01 style).*_reach-- the same queues restricted to vulns that lie on some entry -> crown-jewel route. They ignore the planted unreachable decoys and non-code-execution noise, so they are the like-for-like test of whether path structure (not just reachability) matters.vuln_only-- LINCHPIN's planner restricted to vulnerabilities (exact cut over vulns when it fits the budget, else greedy over vulns): the best a patch-only plan can do, i.e. the like-for-like ceiling for the score queues, which may also only patch.betweenness(graph-aware, path-agnostic) andrandom(uniform over all remediable nodes, on a path or not).
The exact min cut (engine/cuts.py) gives the true minimum number of fixes that disconnect
everything, and the interdiction MILP the best achievable cheapest-path cost, so optimality
gaps are reported too. Topology sizes: :func:hosts_for_seed.
STRATEGIES
module-attribute
¶
STRATEGIES = ('linchpin', 'greedy', 'milp_interdiction', 'greedy_interdiction', 'vuln_only', 'cvss', 'cvss_reach', 'epss', 'epss_reach', 'kev_epss', 'kev_epss_reach', 'betweenness', 'random')
LABEL
module-attribute
¶
LABEL = {'linchpin': 'LINCHPIN (exact cut, else greedy)', 'greedy': 'LINCHPIN greedy set cover only', 'milp_interdiction': 'Exact interdiction MILP (Israeli & Wood 2002)', 'greedy_interdiction': 'Greedy interdiction (Guo et al.-style)', 'vuln_only': 'Best patch-only plan (LINCHPIN over vulns only)', 'cvss': 'CVSS-first', 'cvss_reach': 'CVSS-first, on-path vulns only', 'epss': 'EPSS-first', 'epss_reach': 'EPSS-first, on-path vulns only', 'kev_epss': 'KEV then EPSS', 'kev_epss_reach': 'KEV then EPSS, on-path vulns only', 'betweenness': 'Betweenness', 'random': 'Random'}
hosts_for_seed ¶
hosts_for_seed(seed: int) -> int
Topology size for a seed: 12..60 hosts, every size once per 49 seeds.
cvss_baseline ¶
cvss_baseline(store: GraphStore, budget: int, on_path_only: bool = False) -> list[str]
Highest CVSS base score first (ties: node id).
order ¶
order(store: GraphStore, strategy: str, budget: int, k: int = 100, seed: int = 0) -> list[str]
The first budget remediation targets (node ids) proposed by strategy.
evaluate ¶
evaluate(store: GraphStore, removed: list[str], k: int) -> dict
What an attacker still has after removed: disconnection, residual paths, cheapest cost.
fixes_to_disconnect ¶
fixes_to_disconnect(store: GraphStore, k: int, cap: int = 25) -> int | None
Fixes the greedy set cover needs to disconnect every crown jewel (None: not within cap).
run_one ¶
run_one(family: str, seed: int, n_hosts: int, budget: int, k: int = 100, strategies: tuple[str, ...] = STRATEGIES) -> dict
Build one seeded topology, run every strategy, and return one result row.
wilson_ci ¶
wilson_ci(k: int, n: int, z: float = 1.96, ndigits: int | None = 4) -> tuple[float, float]
95% Wilson score interval for a binomial proportion k/n (ndigits=None: unrounded).
bootstrap_ci ¶
bootstrap_ci(xs: list[float], reps: int = 2000, seed: int = 0) -> tuple[float, float] | None
Percentile-bootstrap 95% CI of the mean (seeded).
mcnemar_exact ¶
mcnemar_exact(b: int, c: int) -> float
Two-sided exact McNemar p-value for discordant pair counts b and c.
paired_diff ¶
paired_diff(a: list[float], b: list[float], reps: int = 2000, seed: int = 0) -> dict | None
Paired comparison of a and b on the same topologies.
Returns the mean of a - b with a percentile-bootstrap 95% interval over the pairs, how often each is larger, and an exact two-sided sign test (ties dropped).
format_p ¶
format_p(p: float) -> str
One format for every p-value in the result files: < 1e-4 or three significant digits.
summarise ¶
summarise(rows: list[dict], strategies: tuple[str, ...] = STRATEGIES) -> dict
Per-family aggregates, Wilson CIs for rates, bootstrap CIs for gains, paired tests.
run ¶
run(n_topologies: int = 100, budget: int = 2, k: int = 100) -> dict
Legacy entry point (single planted linchpin, placeholder CVEs).
linchpin.ml.exploitability ¶
M11: learned exploitability -- P(CVE becomes known-exploited) from public NVD data.
Label: membership in CISA KEV. Features: CVSS vector components, CWE ids, the base-score bucket and the English description, hashed into one sparse space (stateless, so there is no vocabulary to persist). Model: class-balanced L2 logistic regression. Deliberately simple, deterministic and cheap to retrain.
The model plugs into the frozen edge-cost formula only through EdgeContext.exploitability
(see contracts/edge_cost.md); the formula shape is unchanged.
Two leaks are closed by default (see benchmarks/ml_exploitability.py for their size):
- labels from the future -- training positives are only the CVEs already in KEV before
the cutoff date (
kev_date < cutoff), not today's catalog; - exploitation status in the text -- NVD descriptions are edited after the fact ("Apple is
aware of a report that this issue may have been actively exploited"); sentences reporting
exploitation status are removed before features are built (:func:
mask_exploitation_status) in training and at scoring time.
The features still come from today's NVD record (vectors and descriptions can be revised after publication), so the evaluation is a temporal split on publication date, not a replay of what was known on each CVE's first day. EPSS is reported as an external reference, not a baseline we claim to beat: EPSS is trained on exploitation telemetry this model does not have.
EXPLOITATION_STATUS
module-attribute
¶
EXPLOITATION_STATUS = re.compile('in the wild|actively exploited|active exploitation|under active|exploited in (an?|the) |(has|have|had) been exploited|(was|were|is|are) being exploited|may have been exploited|known to (have been|be) exploited|reports? of (active )?(exploitation|attacks?)|aware of (a |an )?(report|attack|exploit|instance|limited|targeted|active)|limited,? targeted attacks?|targeted attacks? (against|in)|exploitation has been (observed|detected)', re.IGNORECASE)
ExploitModel
dataclass
¶
Sparse logistic-regression weights over hashed features, plus how they were trained.
score ¶
score(rec) -> float
Interface used by the edge-cost call site: exploit likelihood in [0, 1].
save ¶
save(path: str | Path) -> None
Write the sparse weights and training metadata to a .npz file (no pickle).
load
classmethod
¶
load(path: str | Path) -> ExploitModel
Read a model written by :meth:save (allow_pickle=False).
mask_exploitation_status ¶
mask_exploitation_status(text: str) -> str
Drop the sentences of text that report exploitation status (see :data:EXPLOITATION_STATUS).
mentions_exploitation ¶
mentions_exploitation(text: str) -> bool
True when text reports exploitation status.
features ¶
features(rec, mask: bool = True) -> str
Flatten one :class:~linchpin.intel.CveRecord into a token string for the hashing vectorizer.
known_exploited ¶
known_exploited(rec, asof: str | None = None) -> bool
KEV label as known on asof (ISO date; None = the catalog snapshot).
train ¶
train(recs, C: float = 4.0, cutoff: str = '', labels=None, mask: bool = True) -> ExploitModel
Fit the model; labels defaults to KEV membership as of cutoff (or the snapshot without one).
temporal_split ¶
temporal_split(intel, cutoff: str = '2023-01-01', start: str = '2015-01-01') -> tuple[list, list]
Train on CVEs published in [start, cutoff), test on CVEs published from cutoff on (vector required).
evaluate ¶
evaluate(model: ExploitModel, test, n_boot: int = 0, seed: int = 0, labels=None) -> dict
ROC-AUC and average precision of every scorer on test.
With n_boot > 0, each metric also gets a seeded, class-stratified percentile-bootstrap
95% interval. labels defaults to KEV membership in the catalog snapshot.
Safety helpers¶
linchpin.paths_safe ¶
Confine client- or file-supplied relative paths to a base directory, by text first.
Used by the API (/demo/load) and by :func:linchpin.scenario.load_scenario when it runs
on behalf of a remote caller. The textual checks run before any filesystem call, so a
rejected path (absolute, drive-qualified, UNC, device name, ..) is never opened or
resolved -- resolving a UNC path on Windows would already start an SMB connection.
PathNotAllowed ¶
Bases: ValueError
A path was refused because it is not a plain relative path inside the allowed base.
relative_parts ¶
relative_parts(name: str) -> list[str]
Validate name as a plain relative path by its text alone and return its parts.
Raises:
| Type | Description |
|---|---|
PathNotAllowed
|
empty, too long, NUL byte, absolute, drive-qualified, UNC/device
( |
safe_join ¶
safe_join(base: str | PathLike[str], name: str) -> Path
base / name for a relative name that provably stays inside base.
The textual check (:func:relative_parts) and a string containment check come first;
only then is the joined path resolved, and containment is checked again so a symlink
inside base cannot point outside it.
Raises:
| Type | Description |
|---|---|
PathNotAllowed
|
the name is not a plain relative path, or it escapes |