CLI and API reference

CLI

Command	Purpose
anvil lint [--profile sigma] [--attack STIX] [--summary]	Validate rules; SigmaHQ-style or ANVIL-style
anvil test	Per-rule fixtures + benign FP corpus + SOC budget gate (exit 1 on failure)
anvil scan --rules DIR... --corpus PATH...	Route a whole library over real telemetry (EVTX, JSON, OTRF zip, JSONL.gz, auditd, CloudTrail)
anvil regress --sigma PATH	Replay SigmaHQ regression captures (exit 1 on a failing case, 2 on a missing or empty checkout)
anvil ingest SRC... --out DIR / --benign	Normalise telemetry into sharded JSONL.gz
anvil draft REPORT [--backend heuristic/llm/keywords]	Draft candidate rules from CTI (review required)
anvil convert RULE --target splunk/elastic/kusto/sqlite	Deployable SIEM queries via pySigma
anvil coverage [--attack STIX --platform Windows] [--navigator out.json]	ATT&CK coverage, claimed vs validated
anvil decay [--routed] [--baseline b.json]	Schema drift, symbolic breakage (broken / source-missing count as issues), regressions; runs weekly in .github/workflows/decay.yml
anvil score	0-100 quality score per rule
anvil synth [--schema v1/v2] [-n N] [--seed S]	Synthetic benign telemetry for the demo
anvil report	Static health dashboard from results/*.json

Python API

anvil.engine

Sigma matching engine.

Rules are compiled once into a tree of small predicate closures, then applied to many events. Supported (covers >95% of the SigmaHQ Windows/Linux rule corpus):

• selections: field maps (AND), lists of maps (OR), value lists (OR), keyword lists (full-text search over all event values), null values
• wildcards */? with Sigma escaping (\\*, \\?, \\\\)
• modifiers: contains, startswith, endswith, all, exists, cased, re (+ i/m/s), windash, base64, base64offset, wide/utf16le/utf16be/utf16, cidr, lt/lte/gt/gte, fieldref, neq
• conditions: and/or/not, parentheses, 1 of x*, all of x*, them

Anything else (aggregations | count(), expand placeholders, unknown modifiers) raises :class:UnsupportedRule at compile time, so callers can report coverage honestly instead of silently mis-evaluating a rule.

ConditionError

Bases: ValueError

Raised when a Sigma condition string cannot be parsed.

UnsupportedRule

Bases: ValueError

The rule uses a Sigma feature this engine deliberately does not evaluate.

CompiledRule

A rule compiled to a single predicate, with an optional logsource guard.

Parameters:

Name	Type	Description	Default
rule	Rule	The parsed rule.	required

Raises:

Type	Description
UnsupportedRule	If the rule uses an unsupported feature.

matches(event)

True if the event satisfies the guard and the rule condition.

get_field(event, field)

Look up a field, falling back to dotted-path traversal of nested dicts.

compile_selection(sel)

Compile one Sigma selection (map or list) to a predicate.

Raises:

Type	Description
UnsupportedRule	If the selection uses an unsupported feature.

match_selection(event, selection)

Evaluate one selection against one event (compiles on each call).

match_field(event, key, expected)

Evaluate one field|modifiers: value pair against an event.

tokenize(cond)

Split a Sigma condition into tokens.

parse_condition(cond, names)

Parse a Sigma condition into an AST.

Parameters:

Name	Type	Description	Default
cond	str	Condition string, e.g. selection and not 1 of filter_*.	required
names	list[str]	Selection names defined in the rule.	required

Returns:

Type	Description
tuple	A nested-tuple AST.

Raises:

Type	Description
UnsupportedRule	For aggregation conditions.
ConditionError	For malformed conditions.

referenced_selections(node)

Return the selection names a condition AST refers to.

required_event_ids(node, sels_raw)

Static analysis: the EventIDs an event must have to possibly match (None = any).

Used by the runner to index rules by (channel, EventID) instead of testing every rule for a channel against every event of that channel.

compile_rule(rule)

Compile a rule; see CompiledRule.

run(rule, events)

Return the indices of events that the rule matches.

rule_fields(rule)

All event field names a rule references (for schema-drift checks).

anvil.runner

Evaluate a whole rule library against a telemetry stream.

The runner compiles every rule once, resolves its logsource to (channel, event id) routes and builds an index so that each event is only tested against the rules that apply to it. On the SigmaHQ corpus this cuts the work per event from ~3,000 rule evaluations to a few hundred at most.

LoadReport dataclass

Counts of loaded, unparsable and skipped rule files.

Library dataclass

A compiled, routed rule library.

active_ids property

IDs of compiled, routable rules.

build(rules, include_deprecated=False) classmethod

Compile rules and build the (channel, EventID) index.

Parameters:

Name	Type	Description	Default
rules	Iterable[Rule]	Parsed rules.	required
include_deprecated	bool	Keep deprecated rules.	False

Returns:

Type	Description
Library	The Library; unsupported and unroutable rules are recorded, not raised.

candidates(ev)

Rule IDs whose logsource covers this event.

match_event(ev)

Rule IDs that match this event.

ScanResult dataclass

Hit counts and throughput of one scan.

events_per_second property

Scan throughput.

load_rule_dir(paths)

Load Sigma YAML rules from one or more directories (recursive).

scan(lib, events, keep_keys=True, on_hit=None)

Run a library over an event stream.

Parameters:

Name	Type	Description	Default
lib	Library	Compiled rule library.	required
events	Iterable[Event]	Events to scan.	required
keep_keys	bool	Record routing keys of hits.	True
on_hit	Callable[[str, Event], None] | None	Callback per (rule_id, event) hit.	None

Returns:

Type	Description
ScanResult	The ScanResult.

anvil.lint

Rule linting / schema validation.

Finding dataclass

One lint finding: rule ID, severity, code and message.

redos_risk(pattern)

Heuristic catastrophic-backtracking check for a rule regex.

Flags a quantified group that itself contains a repeat quantifier ((a+)+, (a*)*, (\w+\s?)*) and a quantified alternation with duplicate or prefix-overlapping branches ((a|a)*, (a|ab)+). It is a lint heuristic, not a proof: it can miss exotic cases and flag some safe ones.

Returns:

Type	Description
str | None	A short reason, or None if no risky construct was found.

lint_rule(rule, profile='anvil', catalog=None)

Lint one rule.

profile="anvil" requires in-rule TP/TN fixtures (ANVIL's own rules); profile="sigma" follows SigmaHQ conventions, where tests live in regression_data and are referenced by regression_tests_path. catalog (an :class:anvil.attack.Catalog) enables stale-tag checks.

lint_rules(rules, profile='anvil', catalog=None)

Lint a rule set, including cross-rule checks such as duplicate IDs.

Parameters:

Name	Type	Description	Default
rules	list[Rule]	Parsed rules.	required
profile	str	anvil (strict) or sigma (SigmaHQ conventions).	'anvil'
catalog	Any	Optional ATT&CK Catalog for tag validation.	None

Returns:

Type	Description
list[Finding]	All findings.

has_errors(findings)

True if any finding has error severity.

anvil.harness

Test harness: TP/TN fixtures + benign corpus FP measurement + CI gate.

GatePolicy dataclass

Thresholds for the CI gate: FP rate and per-rule share of SOC capacity.

RuleResult dataclass

Per-rule gate result: TP recall, benign hits, projected alerts/day, verdict.

to_dict()

Return the result as a plain dict.

evaluate(rule, corpus, policy=None)

Run one rule against its fixtures and a benign corpus.

Parameters:

Name	Type	Description	Default
rule	Rule	Rule with tests fixtures.	required
corpus	list[dict[str, Any]]	Benign events.	required
policy	GatePolicy | None	Gate thresholds; defaults to GatePolicy().	None

Returns:

Type	Description
RuleResult	The RuleResult with its verdict and reasons.

evaluate_all(rules, corpus, policy=None)

Evaluate every non-deprecated rule; see evaluate.

anvil.fpmodel

Predict whether a rule will be noisy before running it on a benign corpus.

Spec item "FP-rate prediction ML": static rule features -> P(rule fires on clean telemetry). The model is trained on labels produced by ANVIL itself (did the rule fire on the evtx-baseline corpus?), restricted to rules whose log source actually occurs in that corpus, and evaluated with stratified cross-validation against simple heuristics a reviewer would otherwise use.

Requires scikit-learn (pip install anvil-dac[ml]).

features(rule)

Return the static FP-prediction features of a rule.

matrix(rules)

Return (feature names, feature matrix) for a list of rules.

expected_top_k(y, s, k)

Expected positives among the top-k scores when ties are broken uniformly at random.

np.argsort breaks ties by row order, which flatters coarse scores (a 5-valued heuristic) whenever rows happen to be sorted by label.

bootstrap_auc(y, scores, reference, n_boot=1000, seed=7)

95% percentile intervals over rules (resampled with replacement) and paired deltas vs reference.

cross_validate(rules, labels, seed=7, folds=5, importance=True, bootstrap=False, gbdt_iter=300)

Stratified k-fold CV for the ML models vs. reviewer heuristics. Returns metrics.

gbdt_iter is the boosting-round count (300 in the benchmark; unit tests use a few).

repeated_cv(rules, labels, seeds=range(10), folds=5, gbdt_iter=300)

Repeat stratified CV over several seeds; report the mean and a 95% t-interval per metric.

The interval only measures fold-assignment (CV-seed) variation on the same rules, not sampling uncertainty over rules: use cross_validate(..., bootstrap=True) for that. The heuristics are deterministic, so their interval collapses to a point.

anvil.decay

Decay monitor: catch detections that silently stop working.

Two complementary checks:

• Static schema analysis (no attack data needed). For every rule, look up which fields its log source actually carries in current telemetry and evaluate the condition symbolically: a selection is satisfiable only if the fields it needs exist. Verdicts: ok, degraded (some referenced fields missing but a detection path survives, or an exclusion filter went blind - expect more FPs), broken (no path can fire any more) and source-missing (the log source itself disappeared).
• Regression vs. baseline (needs TP evidence). Re-run the rule on its stored true-positive samples and compare recall/FP with the saved baseline.

SCHEMA_CHANGES holds realistic telemetry changes used by the benchmark to measure how many decays each check catches.

FieldInventory

Which fields occur per (channel, EventID) in a telemetry stream.

add(ev)

Record the fields of one event under its (channel, EventID) key.

from_events(events) classmethod

Build an inventory from an event iterable.

fields_for(rule)

Union of fields over the keys the rule's log source routes to; None if absent.

observed_fields(events)

Return every field path (dotted for nested dicts) seen in the events.

schema_drift(rules, telemetry)

Rules whose referenced fields no longer appear anywhere in current telemetry (0.1 API).

analyse(rule, inventory)

Check whether a rule's fields still exist in the telemetry.

Parameters:

Name	Type	Description	Default
rule	Rule	The rule to check.	required
inventory	FieldInventory	Fields observed per log source.	required

Returns:

Type	Description
dict[str, Any]	A dict with status, the missing fields and blind_filters
dict[str, Any]	(filter selections whose fields are absent).

regressions(baseline, current, fp_spike_factor=3.0)

Compare current results to a baseline {rule_id: RuleResult.to_dict()}.

apply_change(events, name)

Apply the named simulated schema change from SCHEMA_CHANGES to events.