Skip to content

API reference

This page links to the primary public Python APIs currently implemented in the repository.

Core

satark.core.engine.AnalysisEngine 

AnalysisEngine(
    plugins: Sequence[Plugin] | None = None,
    *,
    store: EventStore | None = None,
    settings: SatarkSettings | None = None
)

Domain-agnostic engine for security analytics.

The engine does not understand vendor formats. Plugins normalize data into Events; the engine stores them, runs pipelines, and aggregates findings.

Source code in src/satark/core/engine.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init__(
    self,
    plugins: Sequence[Plugin] | None = None,
    *,
    store: EventStore | None = None,
    settings: SatarkSettings | None = None,
) -> None:
    self.settings = settings or load_settings()
    self.store: EventStore = store or InMemoryEventStore()
    self._plugins: dict[str, Plugin] = {}
    if plugins:
        for plugin in plugins:
            self.register(plugin)

register 

register(plugin: Plugin) -> None

Register a plugin by its metadata name.

Source code in src/satark/core/engine.py 
52
53
54
55
56
57
58
def register(self, plugin: Plugin) -> None:
    """Register a plugin by its metadata name."""
    name = plugin.meta.name
    if name in self._plugins:
        msg = f"Plugin already registered: {name}"
        raise ValueError(msg)
    self._plugins[name] = plugin

list_plugins 

list_plugins() -> list[str]

Return registered plugin names.

Source code in src/satark/core/engine.py 
60
61
62
def list_plugins(self) -> list[str]:
    """Return registered plugin names."""
    return sorted(self._plugins)

get_plugin 

get_plugin(name: str) -> Plugin

Fetch a registered plugin or raise KeyError.

Source code in src/satark/core/engine.py 
64
65
66
67
68
69
70
def get_plugin(self, name: str) -> Plugin:
    """Fetch a registered plugin or raise KeyError."""
    try:
        return self._plugins[name]
    except KeyError as exc:
        msg = f"Plugin not found: {name}"
        raise KeyError(msg) from exc

ingest 

ingest(events: Sequence[Event]) -> int

Store normalized events and return the count ingested.

Source code in src/satark/core/engine.py 
72
73
74
75
def ingest(self, events: Sequence[Event]) -> int:
    """Store normalized events and return the count ingested."""
    self.store.put(events)
    return len(events)

ingest_raw 

ingest_raw(
    plugin_name: str,
    records: Sequence[dict[str, Any]],
    context: PluginContext | None = None,
) -> list[Event]

Normalize raw records via a plugin and store the resulting events.

Source code in src/satark/core/engine.py 
77
78
79
80
81
82
83
84
85
86
87
88
def ingest_raw(
    self,
    plugin_name: str,
    records: Sequence[dict[str, Any]],
    context: PluginContext | None = None,
) -> list[Event]:
    """Normalize raw records via a plugin and store the resulting events."""
    plugin = self.get_plugin(plugin_name)
    ctx = context or PluginContext()
    events = plugin.normalize(records, ctx)
    self.ingest(events)
    return events

analyze 

analyze(
    *,
    plugin_name: str | None = None,
    events: Sequence[Event] | None = None,
    context: PluginContext | None = None
) -> AnalysisResult

Analyze events with one or all plugins.

Source code in src/satark/core/engine.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def analyze(
    self,
    *,
    plugin_name: str | None = None,
    events: Sequence[Event] | None = None,
    context: PluginContext | None = None,
) -> AnalysisResult:
    """Analyze events with one or all plugins."""
    source_events = list(events) if events is not None else self.store.list_events()
    if not self._plugins:
        return AnalysisResult(events_processed=len(source_events))

    pipeline = AnalysisPipeline(list(self._plugins.values()))
    ctx = context or PluginContext(config={"risk_threshold": self.settings.risk_threshold})

    result: PipelineResult
    if plugin_name is not None:
        result = pipeline.run_events(source_events, ctx, plugin_name=plugin_name)
    else:
        result = pipeline.run_all(source_events, ctx)

    elevated = [f for f in result.findings if f.score.value >= self.settings.risk_threshold]
    return AnalysisResult(
        findings=result.findings,
        events_processed=len(source_events),
        elevated=elevated,
    )

run_plugin 

run_plugin(
    plugin_name: str, context: PluginContext | None = None
) -> AnalysisResult

Run a plugin's full collect→normalize→detect→score→explain pipeline.

Source code in src/satark/core/engine.py 
118
119
120
121
122
123
124
125
126
127
128
def run_plugin(self, plugin_name: str, context: PluginContext | None = None) -> AnalysisResult:
    """Run a plugin's full collect→normalize→detect→score→explain pipeline."""
    plugin = self.get_plugin(plugin_name)
    findings = plugin.run(context)
    events = self.store.list_events()
    elevated = [f for f in findings if f.score.value >= self.settings.risk_threshold]
    return AnalysisResult(
        findings=findings,
        events_processed=len(events),
        elevated=elevated,
    )

satark.core.plugin.Plugin

Bases: ABC

Abstract base for all SATARK analytics plugins.

Prefer composition for helpers; subclasses implement the stage methods. Detections produced by :meth:detect must be reproducible without AI.

meta abstractmethod property 

meta: PluginMeta

Return plugin metadata.

collect 

collect(context: PluginContext) -> Iterable[dict[str, Any]]

Collect raw vendor/source records.

Default implementation yields nothing; override when the plugin owns its own data sources.

Source code in src/satark/core/plugin.py 
61
62
63
64
65
66
67
def collect(self, context: PluginContext) -> Iterable[dict[str, Any]]:
    """Collect raw vendor/source records.

    Default implementation yields nothing; override when the plugin owns
    its own data sources.
    """
    return []

normalize abstractmethod 

normalize(
    records: Sequence[dict[str, Any]],
    context: PluginContext,
) -> list[Event]

Normalize raw records into canonical :class:Event objects.

Source code in src/satark/core/plugin.py 
69
70
71
72
73
74
75
@abstractmethod
def normalize(
    self,
    records: Sequence[dict[str, Any]],
    context: PluginContext,
) -> list[Event]:
    """Normalize raw records into canonical :class:`Event` objects."""

detect abstractmethod 

detect(
    events: Sequence[Event], context: PluginContext
) -> list[Detection]

Produce detections from normalized events (no AI required).

Source code in src/satark/core/plugin.py 
77
78
79
@abstractmethod
def detect(self, events: Sequence[Event], context: PluginContext) -> list[Detection]:
    """Produce detections from normalized events (no AI required)."""

score abstractmethod 

score(
    detection: Detection,
    events: Sequence[Event],
    context: PluginContext,
) -> ScoreBreakdown

Compute a transparent risk score for a detection.

Source code in src/satark/core/plugin.py 
81
82
83
84
85
86
87
88
@abstractmethod
def score(
    self,
    detection: Detection,
    events: Sequence[Event],
    context: PluginContext,
) -> ScoreBreakdown:
    """Compute a transparent risk score for a detection."""

explain 

explain(
    detection: Detection,
    score: ScoreBreakdown,
    events: Sequence[Event],
    context: PluginContext,
) -> str

Return a human-readable explanation for a scored detection.

Source code in src/satark/core/plugin.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def explain(
    self,
    detection: Detection,
    score: ScoreBreakdown,
    events: Sequence[Event],
    context: PluginContext,
) -> str:
    """Return a human-readable explanation for a scored detection."""
    factor_lines = "; ".join(
        f"{f.name} ({f.contribution:+.2f}): {f.description}" for f in score.factors
    )
    return (
        f"{detection.title}: risk={score.value:.2f}, confidence={score.confidence:.2f}. "
        f"{score.reasoning}" + (f" Factors: {factor_lines}." if factor_lines else "")
    )

run 

run(context: PluginContext | None = None) -> list[Finding]

Execute the full plugin pipeline: collect → normalize → detect → score → explain.

Source code in src/satark/core/plugin.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def run(self, context: PluginContext | None = None) -> list[Finding]:
    """Execute the full plugin pipeline: collect → normalize → detect → score → explain."""
    ctx = context or PluginContext()
    records = list(self.collect(ctx))
    events = self.normalize(records, ctx)
    detections = self.detect(events, ctx)
    findings: list[Finding] = []
    for detection in detections:
        score = self.score(detection, events, ctx)
        explanation = self.explain(detection, score, events, ctx)
        findings.append(
            Finding(
                detection=detection,
                score=score,
                explanation=explanation,
            )
        )
    return findings

satark.core.events.Event

Bases: BaseModel

Canonical security event consumed by the SATARK engine.

Examples

from datetime import datetime, UTC Event( ... category=EventCategory.USB_INSERTION, ... source="endpoint.agent", ... actor="alice", ... timestamp=datetime.now(UTC), ... attributes={"device_id": "USB-42"}, ... ) Event(...)

with_attribute 

with_attribute(key: str, value: Any) -> Event

Return a copy with an additional attribute (immutability-friendly).

Source code in src/satark/core/events/__init__.py 
77
78
79
80
def with_attribute(self, key: str, value: Any) -> Event:
    """Return a copy with an additional attribute (immutability-friendly)."""
    merged = {**self.attributes, key: value}
    return self.model_copy(update={"attributes": merged})

Scoring

satark.scoring.risk.aggregate_score 

aggregate_score(
    factors: Sequence[ScoreFactor],
    *,
    confidence: float,
    reasoning: str,
    evidence: Sequence[Evidence] | None = None,
    references: Sequence[KnowledgeReference] | None = None,
    baseline: float = 0.0
) -> ScoreBreakdown

Aggregate signed factor contributions into an explainable score.

Positive contributions increase risk; negative contributions reduce it. The result is always clamped to [0, 1].

Source code in src/satark/scoring/risk/__init__.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def aggregate_score(
    factors: Sequence[ScoreFactor],
    *,
    confidence: float,
    reasoning: str,
    evidence: Sequence[Evidence] | None = None,
    references: Sequence[KnowledgeReference] | None = None,
    baseline: float = 0.0,
) -> ScoreBreakdown:
    """Aggregate signed factor contributions into an explainable score.

    Positive contributions increase risk; negative contributions reduce it.
    The result is always clamped to [0, 1].
    """
    total = baseline + sum(f.contribution for f in factors)
    all_evidence: list[Evidence] = list(evidence or [])
    for factor in factors:
        all_evidence.extend(factor.evidence)
    return ScoreBreakdown(
        value=clamp01(total),
        confidence=clamp01(confidence),
        factors=list(factors),
        evidence=all_evidence,
        reasoning=reasoning,
        references=list(references or []),
    )

satark.scoring.explainability.why_malicious 

why_malicious(
    detection: Detection, score: ScoreBreakdown
) -> str

Short answer to: Why was this event classified as malicious?

Source code in src/satark/scoring/explainability/__init__.py 
33
34
35
36
37
38
39
def why_malicious(detection: Detection, score: ScoreBreakdown) -> str:
    """Short answer to: Why was this event classified as malicious?"""
    top = sorted(score.factors, key=lambda f: abs(f.contribution), reverse=True)[:3]
    if not top:
        return score.reasoning
    parts = [f"{f.name} ({f.contribution:+.2f})" for f in top]
    return f"{detection.title} scored {score.value:.2f} due to: " + ", ".join(parts) + "."

Plugins

satark.plugins.registry.create_plugin 

create_plugin(name: str) -> Plugin

Instantiate a built-in plugin by name.

Source code in src/satark/plugins/registry.py 
32
33
34
35
36
37
38
39
40
def create_plugin(name: str) -> Plugin:
    """Instantiate a built-in plugin by name."""
    try:
        factory = _REGISTRY[name]
    except KeyError as exc:
        available = ", ".join(builtin_plugins())
        msg = f"Unknown plugin '{name}'. Available: {available}"
        raise KeyError(msg) from exc
    return factory()

satark.plugins.insider.InsiderThreatPlugin 

InsiderThreatPlugin(
    *,
    usb_spike_threshold: float = 3.0,
    file_spike_threshold: float = 3.0
)

Bases: Plugin

Detect insider-threat patterns from normalized endpoint events.

Source code in src/satark/plugins/insider/__init__.py 
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    *,
    usb_spike_threshold: float = 3.0,
    file_spike_threshold: float = 3.0,
) -> None:
    self.usb_spike_threshold = usb_spike_threshold
    self.file_spike_threshold = file_spike_threshold
    self._attack = default_attack_provider()

For deeper module docs, browse the source under src/satark/.