Skip to content

Python API

The core package is standard-library only; specimen.ml needs the [ml] extra.

Pipeline

specimen.pipeline

Orchestration: triage -> (trace replay) -> reconstruct -> score -> synthesize -> report.

run(sample_path, trace_path=None, benign_blobs=None, force_detonate=False)

Full pipeline on sample bytes plus an optional recorded run.

The trace may be a native SPECIMEN trace, a CAPE/Cuckoo JSON report or a Sysmon XML / JSON-lines export. Sigma rules come from the same specificity-checked v2 synthesizer as :func:run_report (packaged negative corpus minus the predicted family); YARA comes from the report's PE metadata when present, otherwise from strings in the sample bytes.

Parameters:

Name Type Description Default
sample_path str | Path

file read as bytes (never executed).

required
trace_path str | Path | None

recorded run to replay; None = static triage only.

None
benign_blobs list[bytes] | None

benign byte blobs that string-based YARA must not match.

None
force_detonate bool

replay the trace even if the static gate says skip.

False

Returns:

Type Description
dict[str, Any]

the report dict (see :func:specimen.report.build_report).

Raises:

Type Description
ValueError

on malformed input or when the trace is bound to a different sample.

run_report(report_path, negatives=None, use_packaged_negatives=True)

Report-only analysis of a CAPE/Cuckoo JSON report (no sample bytes).

Static gate from the report's PE metadata, provenance + timeline from the behaviour, family attribution (trained model if available), and specificity-checked Sigma (behaviour) + YARA (PE metadata) rules.

Parameters:

Name Type Description Default
report_path str | Path

CAPE/Cuckoo JSON report.

required
negatives list[dict[str, str]] | None

extra negative blobs (see :func:specimen.detect.blobs).

None
use_packaged_negatives bool

also check rules against the packaged Avast-CTU negative corpus (minus the predicted family) and rank YARA imports by its prevalence table.

True

Returns:

Type Description
dict[str, Any]

the report dict (see :func:specimen.report.build_report).

behaviour_score(trace)

Real-data API n-gram scorer when the trace has an API-call sequence, otherwise the MVP ATT&CK-feature scorer.

is_sysmon(raw)

Sysmon XML export, or JSON lines whose first object carries an EventID (UTF-8, UTF-16 or UTF-32, with or without a byte-order mark).

Adapters

specimen.adapters.cape

CAPEv2 / Cuckoo JSON report -> SPECIMEN Trace.

Handles three shapes of report, all reports only (no binaries involved):

  • full CAPEv2 / Cuckoo report - behavior.processes[].calls with per-call timestamps, behavior.processtree and network sections;
  • reduced report (Avast-CTU public dataset) - only behavior.summary and static.pe; there is no timing, so events receive a deterministic synthetic order and are flagged synthetic_ts;
  • any mix of the two (summary is used to fill gaps the call log misses).

The adapter never trusts the report: every string is coerced and truncated, list sizes are capped and unknown shapes are ignored rather than raised on.

cape_to_trace(report, run_id=None, raw=None)

Convert a CAPE/Cuckoo report (full call log or reduced summary) into a :class:Trace.

Parameters:

Name Type Description Default
report dict[str, Any]

parsed JSON report; unknown shapes are ignored, not raised on.

required
run_id str | None

run identifier (defaults to the report's info.id).

None
raw bytes | None

original bytes, hashed into Trace.source_sha256.

None

Raises:

Type Description
CapeFormatError

if the object is not a report or has no behaviour.

static_pe(report)

The static.pe block of a report (empty dict if absent).

specimen.adapters.api_seq

API-call-sequence datasets (e.g. MalbehavD-V1, Cuckoo-derived) -> Trace.

Such datasets keep only the ordered API names per sample, so every call becomes an api_call event with its sequence index as timestamp.

api_sequence_to_trace(apis, run_id, sample_sha256='', image='sample.exe')

Convert an ordered list of API names into a Trace of api_call events.

Parameters:

Name Type Description Default
apis Sequence[str]

API names in call order; empty entries are skipped and at most MAX_CALLS are kept.

required
run_id str

identifier stored on the trace.

required
sample_sha256 str

hash of the sample, if known.

''
image str

process image name given to every event.

'sample.exe'

Returns:

Type Description
Trace

a Trace whose event timestamps are the sequence indices.

specimen.adapters.sysmon

Sysmon events -> Trace (spec stage 5, capture adapter; offline only).

Accepts what you get out of a Windows lab box without extra tooling:

  • XML: wevtutil qe Microsoft-Windows-Sysmon/Operational /f:xml or Get-WinEvent ... | ForEach-Object { $_.ToXml() } (one or many <Event> elements, with or without a wrapping root);
  • JSON lines: one object per event with EventID plus the Sysmon field names (Image, TargetFilename, ...) at the top level or under EventData (winlogbeat/NXLog style exports).

Binary .evtx is not parsed here (it would need a third-party parser); export it to XML first. UTF-8, UTF-16 and UTF-32 exports are accepted (with or without a byte-order mark; Windows PowerShell 5.1 > writes UTF-16LE). Parsing uses xml.etree on untrusted input, so DTDs/entities are refused before parsing, sizes are capped and malformed XML becomes a ValueError.

Evidence binding: when process-creation events (ID 1) carry a Hashes field with SHA256=..., the export is bound to those hashes and :func:sysmon_to_trace refuses a sample whose SHA-256 is not among them.

sysmon_to_trace(text, run_id='sysmon', sample_sha256='')

Convert a Sysmon export (wevtutil /f:xml XML or JSON lines) into a :class:Trace.

Parameters:

Name Type Description Default
text str | bytes

export contents (bytes in UTF-8/16/32); DTDs are refused and size is capped.

required
run_id str

run identifier stored in the trace.

'sysmon'
sample_sha256 str

SHA-256 of the sample the run is claimed to belong to. If the export records process hashes (event 1 Hashes), it must be one of them and the trace is bound to it; if the export records no hashes the trace stays unbound (sample_sha256 == "").

''

Raises:

Type Description
ValueError

on oversized or malformed input, or when the export's process hashes do not include sample_sha256 (evidence mismatch).

decode_export(raw)

Decode a Sysmon export: UTF-32/UTF-16/UTF-8 byte-order marks, or BOM-less UTF-16 detected from NUL-interleaved ASCII; otherwise UTF-8 (errors replaced).

process_hashes(trace)

SHA-256 values recorded by Sysmon event 1 (Hashes field) in a trace.

specimen.adapters.speakeasy

Speakeasy emulation report -> Trace (offline; reports only).

Speakeasy (Mandiant) emulates a PE instead of running it. The Quo Vadis dataset (Trizna 2022, Apache-2.0) publishes one Speakeasy JSON report per sample; benign samples live under report_clean and report_windows_syswow64. SPECIMEN uses those benign reports to measure how often synthesised Sigma rules fire on benign software.

Accepted shapes: a list of entry points, or an object with an entry_points list. Each entry point may carry::

apis             [{api_name: "KERNEL32.CreateFileA", args: [...], ret_val}]
file_access      [{event: "create"|"open"|"read"|"write"|"delete", path}]
registry_access  [{event: "open_key"|"create_key"|"read_value"|"write_value"|..., path, value_name}]
network_events   {dns: [{query}], traffic: [{server, port, proto}]}

Mapping (same line format as CAPE and Sysmon traces, so the same Sigma matcher applies): process-creation APIs (CreateProcess*, WinExec, ShellExecute*) -> process_create; file create/write -> file_write; registry value writes -> registry_set; DNS/traffic -> dns_query/net_connect; every API call -> api_call.

The input is untrusted: strings are coerced and truncated, lists capped, unknown shapes ignored. An emulator records fewer side effects than a sandbox, so benign false-positive rates measured on these reports are a lower bound for sandbox traces.

speakeasy_to_trace(report, run_id='speakeasy', raw=None)

Convert one Speakeasy report (list of entry points or {"entry_points": [...]}) into a :class:Trace.

Parameters:

Name Type Description Default
report Any

parsed JSON (or raw bytes/str, parsed here).

required
run_id str

identifier stored in the trace (e.g. the report's file name).

'speakeasy'
raw bytes | None

original bytes, only used for the trace's source hash.

None

Returns:

Type Description
Trace

a trace with sandbox == "speakeasy" and synthetic, ordered timestamps.

Raises:

Type Description
ValueError

if the report is not JSON or has no entry points.

Hostile-input helpers

specimen.coerce

Coercion helpers for hostile report and trace values (standard library only).

Sandbox reports and traces come from the environment that ran the sample, so numbers may be NaN, Infinity, 1e999 or the wrong type, and JSON may be nested deeply enough to exhaust the parser's recursion limit. These helpers turn such values into defaults or a :class:ValueError with a clear message, never into an unhandled OverflowError / RecursionError traceback.

finite_float(v, default=0.0, lo=None, hi=None)

float(v) if it is a finite number (optionally clamped to [lo, hi]), else default.

safe_int(v, default=0)

int(v) for finite numbers and integer strings, else default (never raises).

loads(raw, what='input')

json.loads for untrusted input.

Raises:

Type Description
ValueError

on invalid JSON, undecodable bytes or nesting deep enough to exhaust the parser's recursion limit.

Behaviour scoring

specimen.api_behaviour

Real-data behaviour scorer: API unigram+bigram TF-IDF + logistic regression.

Trained on MalbehavD-V1 (benchmarks/bench_malbehavd.py) and stored as a plain JSON file (models/api_behaviour.json: per-token idf and weight, no pickle). Prediction is pure Python, so the core pipeline stays stdlib-only. It replicates scikit-learn's TfidfVectorizer(sublinear_tf=True) (L2 norm) followed by LogisticRegression.decision_function.

It only applies to traces that carry a real API-call sequence (full CAPE / Cuckoo call logs or API-sequence datasets); reduced reports fall back to the MVP scorer.

ApiBehaviourModel

TF-IDF + logistic-regression API n-gram scorer evaluated in pure Python.

Exported from scikit-learn by benchmarks/bench_malbehavd.py; stored as JSON (specimen/data/api_behaviour.json) so the core stays stdlib-only.

proba(apis)

Malicious probability of an API sequence.

Parameters:

Name Type Description Default
apis Sequence[str]

API names in call order.

required

Returns:

Type Description
float

probability in [0, 1].

explain(apis, k=8)

Top n-gram contributions to the score of an API sequence.

Parameters:

Name Type Description Default
apis Sequence[str]

API names in call order.

required
k int

number of contributions to return.

8

Returns:

Type Description
list[Contribution]

contributions sorted by absolute impact.

score(trace)

Score a trace; None when it holds fewer than MIN_CALLS API calls.

Parameters:

Name Type Description Default
trace Trace

normalised sandbox trace.

required

Returns:

Type Description
BehaviorScore | None

a BehaviorScore or None.

save(path)

Write the model as JSON into a directory.

Parameters:

Name Type Description Default
path Path

destination directory (created if missing).

required

Returns:

Type Description
Path

the model file written.

load(path) classmethod

Load a model saved with :meth:save.

Parameters:

Name Type Description Default
path Path

directory holding the model file.

required

Returns:

Type Description
ApiBehaviourModel

the loaded model.

api_sequence(trace)

Ordered API names of a trace (typed CAPE events keep theirs in extra['api']).

ngrams(apis)

Lower-case API unigrams plus adjacent bigrams (a>b), the model's token set.

Parameters:

Name Type Description Default
apis Sequence[str]

API call names in call order.

required

Returns:

Type Description
list[str]

unigram tokens followed by bigram tokens.

specimen.scoring

Behavior ML scoring: pure-Python logistic regression over explainable features, per-event rarity (anomaly), and nearest-family clustering.

featurize(trace)

MVP behaviour features: nine log-scaled ATT&CK-derived counts of a trace.

Parameters:

Name Type Description Default
trace Trace

validated trace.

required

Returns:

Type Description
list[float]

feature vector in the order of FEATURES.

score(trace, family_threshold=0.5)

Score a trace with the MVP synthetic-trained scorer and the Jaccard family matcher.

Parameters:

Name Type Description Default
trace Trace

validated trace.

required
family_threshold float

minimum Jaccard similarity for a family match.

0.5

Returns:

Type Description
BehaviorScore

probability, label, per-feature contributions and family match.

event_anomaly(event_type, technique)

Rarity of this action vs. the benign baseline (0 common .. 1 never seen).

Detection synthesis

specimen.detect

Specificity-constrained detection synthesis (Sigma + YARA), v2.

The MVP synthesizer (synth.py) emits exact-value rules only for events that map to an ATT&CK technique. On real sandbox data that produces rules that almost never fire on a sibling sample: dropped-file names, registry values and command lines are randomised per run.

This module instead:

  1. extracts candidate selections from every host-visible action of one run (process creation, registry write, file write),
  2. builds a ladder of increasingly general wildcard patterns for each one (exact -> numbers/hex wildcarded -> file stem wildcarded -> parent dir wildcarded),
  3. keeps the most general rung that still has zero hits on a negative corpus (benign traces and/or other families) and still carries enough literal characters to be meaningful.

Matching uses the same line format for rules and traces, so a rule is one compiled regex searched over a per-category text blob - fast enough to validate against thousands of traces.

Standard library only.

synthesize_sigma(events, negatives, max_rules=MAX_RULES, max_neg_hits=0, max_rung=None)

Pick, per candidate action, the most general rung with <= max_neg_hits matches in negatives (per-category blobs of negative traces).

Parameters:

Name Type Description Default
events Iterable[Sequence[str]]

[type, target, cmdline] rows of one run.

required
negatives Sequence[dict[str, str]]

per-category blobs (see :func:blobs); a merged blob of many traces is equivalent when max_neg_hits == 0.

required
max_rules int

rule budget per run (most general rules first).

MAX_RULES
max_rung int | None

if set, only rungs <= max_rung are considered (0 = exact values only; used for the ablation).

None

Returns:

Type Description
SynthesisResult

chosen rules and counts of candidates / rejected candidates.

synthesize_yara_pe(static_tokens, negatives, prevalence=None, k=8, frac=0.75, with_imphash=True)

imphash OR >= frac of the k rarest imports; dropped if it hits a negative.

Report

specimen.report

Unified confidence-graded report (JSON + Markdown) with run manifest.

build_report(sample, static, trace, graph, timeline, behavior, det, trace_binding=None)

Assemble the JSON report: fused verdict, static reasons, timeline, graph, IOCs, rules.

Parameters:

Name Type Description Default
trace_binding str | None

"bound: ..." / "unbound: ..." when a trace was replayed against sample bytes (analyze); an unbound trace lowers the verdict confidence and is recorded in the manifest. None for report-only analysis.

None

Returns:

Type Description
dict[str, Any]

a JSON-serialisable dict; the manifest is added by the pipeline.

render_markdown(rep)

Human-readable Markdown report; every report-derived string is escaped and IOCs are defanged.

fuse(static, behavior)

Fuse static and behaviour scores into a final verdict.

The final score is the maximum of both; confidence drops when they disagree.

Parameters:

Name Type Description Default
static StaticVerdict

static-gate verdict.

required
behavior BehaviorScore | None

behaviour score, or None when no trace was analysed.

required

Returns:

Type Description
tuple[str, str, float]

(label, confidence, final_score).

Data models

specimen.models

Typed data models shared across pipeline stages.

Sample dataclass

Identity of a submitted file: path, hashes, size and bytes analysed.

Contribution dataclass

One feature's share of a score: impact = value * weight.

StaticVerdict dataclass

Static-gate output: score in [0, 1], label, detonate decision, reasons and extracted strings/IOCs.

Event dataclass

One normalised sandbox event (type from EVENT_TYPES, timestamp, process, optional target).

Trace dataclass

Ordered events of one detonation run, with the sandbox name and source-file hash.

Node dataclass

Provenance-graph vertex (process, file, registry key, network endpoint, ...).

Edge dataclass

Directed, timestamped provenance-graph relation between two node ids.

TimelineEntry dataclass

Report timeline row: description, optional ATT&CK technique/tactic and anomaly score.

BehaviorScore dataclass

Behaviour-model output: malicious probability, label, contributions and family estimate.

Detections dataclass

Synthesised YARA and Sigma rules plus their hits on the packaged false-positive corpus.

to_dict(obj)

Recursively convert dataclasses (and lists of them) to plain dicts for JSON output.

Parameters:

Name Type Description Default
obj Any

a dataclass instance, a list, or any other value (returned unchanged).

required

Returns:

Type Description
Any

a JSON-serialisable structure.