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
|
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: |
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: |
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: |
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[].callswith per-call timestamps,behavior.processtreeandnetworksections; - reduced report (Avast-CTU public dataset) - only
behavior.summaryandstatic.pe; there is no timing, so events receive a deterministic synthetic order and are flaggedsynthetic_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 |
None
|
raw
|
bytes | None
|
original bytes, hashed into |
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 |
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 |
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:xmlorGet-WinEvent ... | ForEach-Object { $_.ToXml() }(one or many<Event>elements, with or without a wrapping root); - JSON lines: one object per event with
EventIDplus the Sysmon field names (Image,TargetFilename, ...) at the top level or underEventData(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 |
''
|
Raises:
| Type | Description |
|---|---|
ValueError
|
on oversized or malformed input, or when the export's process
hashes do not include |
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 |
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 |
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 |
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:
- extracts candidate selections from every host-visible action of one run (process creation, registry write, file write),
- builds a ladder of increasingly general wildcard patterns for each one (exact -> numbers/hex wildcarded -> file stem wildcarded -> parent dir wildcarded),
- 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]]
|
|
required |
negatives
|
Sequence[dict[str, str]]
|
per-category blobs (see :func: |
required |
max_rules
|
int
|
rule budget per run (most general rules first). |
MAX_RULES
|
max_rung
|
int | None
|
if set, only rungs |
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
|
|
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 |
required |
Returns:
| Type | Description |
|---|---|
tuple[str, str, float]
|
|
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. |