Skip to content

API reference

This page generates reference material from the implementation. The stable conceptual API is the event → detection → score → finding flow; individual classes remain subject to alpha-version changes.

Typical imports

from satark.core.engine import AnalysisEngine
from satark.core.events import Event, EventCategory
from satark.core.plugin import Plugin, PluginContext, PluginMeta
from satark.core.models import Detection, Evidence, Finding, ScoreBreakdown
from satark.plugins import create_plugin

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 
48
49
50
51
52
53
54
55
56
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()

satark.plugins.insider.lstm.LstmInsiderDetector 

LstmInsiderDetector(
    *,
    sequence_length: int = 20,
    validation_fraction: float = 0.2,
    percentile: float = 99.0,
    epochs: int = 25,
    batch_size: int = 32
)

Per-actor LSTM reconstruction-error detector.

Call :meth:fit with known-normal baseline events, then call :meth:detect on later events. Scaling and threshold calibration use only the baseline; the held-out tail of that baseline determines the threshold. TensorFlow is imported only when fit is called.

Source code in src/satark/plugins/insider/lstm.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def __init__(
    self,
    *,
    sequence_length: int = 20,
    validation_fraction: float = 0.2,
    percentile: float = 99.0,
    epochs: int = 25,
    batch_size: int = 32,
) -> None:
    if sequence_length < 2:
        raise ValueError("sequence_length must be at least 2")
    if epochs < 1 or batch_size < 1:
        raise ValueError("epochs and batch_size must be positive")
    if not 0.0 < validation_fraction < 1.0:
        raise ValueError("validation_fraction must be between 0 and 1")
    if not 0.0 < percentile < 100.0:
        raise ValueError("percentile must be between 0 and 100")
    self.sequence_length = sequence_length
    self.validation_fraction = validation_fraction
    self.percentile = percentile
    self.epochs = epochs
    self.batch_size = batch_size
    self._mean: np.ndarray | None = None
    self._scale: np.ndarray | None = None
    self._threshold: float | None = None
    self._model: Any = None

threshold property 

threshold: float

Calibrated reconstruction-error threshold.

fit 

fit(baseline_events: Sequence[Event]) -> None

Fit on known-normal events and calibrate on a held-out tail.

Source code in src/satark/plugins/insider/lstm.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 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
117
118
119
120
121
122
123
124
125
def fit(self, baseline_events: Sequence[Event]) -> None:
    """Fit on known-normal events and calibrate on a held-out tail."""
    # Split raw buckets within each actor BEFORE creating overlapping windows.
    # No observation is shared between fitting and threshold calibration.
    training_parts = []
    validation_parts = []
    for actor, buckets in self._buckets(baseline_events).items():
        vectors = np.array([row[0] for row in buckets], dtype=float)
        split = int(len(vectors) * (1.0 - self.validation_fraction))
        if min(split, len(vectors) - split) < self.sequence_length:
            raise ValueError(
                f"Actor {actor!r} needs at least {self.sequence_length} buckets "
                "in BOTH training and validation partitions."
            )
        training_parts.append(vectors[:split])
        validation_parts.append(vectors[split:])
    if not training_parts:
        raise ValueError("No supported baseline events")
    raw_training = np.concatenate(training_parts)
    mean = raw_training.mean(axis=0)
    scale = raw_training.std(axis=0)
    scale[scale == 0] = 1.0
    training = (np.concatenate([self._windows(p) for p in training_parts]) - mean) / scale
    validation = (np.concatenate([self._windows(p) for p in validation_parts]) - mean) / scale
    tensorflow: Any = _tensorflow()

    model = tensorflow.keras.Sequential(
        [
            tensorflow.keras.layers.Input(shape=(self.sequence_length, len(_FEATURES))),
            tensorflow.keras.layers.LSTM(64, return_sequences=True),
            tensorflow.keras.layers.Dropout(0.2),
            tensorflow.keras.layers.LSTM(32),
            tensorflow.keras.layers.Dense(len(_FEATURES)),
        ]
    )
    model.compile(optimizer="adam", loss="mse")
    model.fit(
        training,
        training[:, -1, :],
        epochs=self.epochs,
        batch_size=self.batch_size,
        shuffle=False,
        verbose=0,
    )
    predicted = model.predict(validation, verbose=0)
    errors = np.mean(np.square(predicted - validation[:, -1, :]), axis=1)
    if not np.all(np.isfinite(errors)):
        raise ValueError("Model produced non-finite calibration errors")
    # Publish state only after a successful fit, including on refits.
    self._mean, self._scale = mean, scale
    self._threshold = float(np.percentile(errors, self.percentile))
    self._model = model

detect 

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

Return a detection for every sequence above the fitted threshold.

Source code in src/satark/plugins/insider/lstm.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def detect(self, events: Sequence[Event]) -> list[Detection]:
    """Return a detection for every sequence above the fitted threshold."""
    if self._model is None or self._mean is None or self._scale is None:
        raise RuntimeError("Call fit() before detect()")
    buckets = self._buckets(events)
    detections: list[Detection] = []
    for actor, actor_buckets in buckets.items():
        vectors = np.array([row[0] for row in actor_buckets], dtype=float)
        if len(vectors) < self.sequence_length:
            continue
        sequences = self._windows(vectors)
        scaled = self._transform(sequences)
        predicted = self._model.predict(scaled, verbose=0)
        errors = np.mean(np.square(predicted - scaled[:, -1, :]), axis=1)
        if not np.all(np.isfinite(errors)):
            raise ValueError("Model produced non-finite inference errors")
        for offset, error in enumerate(errors):
            if float(error) <= self.threshold:
                continue
            _, timestamp, event_ids = actor_buckets[offset + self.sequence_length - 1]
            detections.append(
                Detection(
                    plugin="insider-lstm",
                    rule_id="insider.lstm_reconstruction_error",
                    title=f"LSTM behavioral anomaly for {actor}",
                    description=(
                        "The per-actor feature sequence had reconstruction error "
                        f"{float(error):.4f}, above the calibrated threshold "
                        f"{self.threshold:.4f}."
                    ),
                    severity=DetectionSeverity.MEDIUM,
                    event_ids=[event.id for event in event_ids],
                    evidence=[
                        Evidence(
                            kind=EvidenceKind.BEHAVIORAL,
                            summary="Per-actor LSTM reconstruction anomaly",
                            details={
                                "actor": actor,
                                "timestamp": timestamp.isoformat(),
                                "reconstruction_error": float(error),
                                "threshold": self.threshold,
                                "features": [item.value for item in _FEATURES],
                            },
                            weight=0.7,
                        )
                    ],
                    tags=["insider", "lstm", "ml", "anomaly"],
                )
            )
    return detections

analyze 

analyze(events: Sequence[Event]) -> list[Finding]

Convert ML detections into transparent, reviewable SATARK findings.

Source code in src/satark/plugins/insider/lstm.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def analyze(self, events: Sequence[Event]) -> list[Finding]:
    """Convert ML detections into transparent, reviewable SATARK findings."""
    findings: list[Finding] = []
    for detection in self.detect(events):
        evidence = detection.evidence[0]
        error = float(evidence.details["reconstruction_error"])
        factor = ScoreFactor(
            name="lstm_reconstruction_error",
            contribution=(
                0.6 if self.threshold == 0 else min(0.6, 0.3 * error / self.threshold)
            ),
            description=(
                f"Reconstruction error {error:.4f} exceeds the calibrated "
                f"threshold {self.threshold:.4f}."
            ),
            evidence=list(detection.evidence),
        )
        score = aggregate_score(
            [factor],
            confidence=0.6,
            reasoning=(
                "Optional LSTM model flagged an unusual per-actor feature sequence. "
                "Review the listed telemetry before taking action."
            ),
            baseline=0.1,
        )
        explanation = (
            f"{detection.title}: reconstruction error {error:.4f} was above "
            f"the baseline threshold {self.threshold:.4f}."
        )
        findings.append(Finding(detection=detection, score=score, explanation=explanation))
    return findings

Supporting APIs

satark.core.storage provides InMemoryEventStore for experiments and JsonlEventStore for lightweight persistent event logs. satark.graph offers EntityGraph, build_timeline, and find_attack_paths. satark.rules contains regex, Sigma-like, STIX-like, and custom predicate rule engines; YARA is a placeholder that requires external integration.

satark.knowledge provides static, versioned lookup providers. The included catalogs are small seeds, not a replacement for a maintained upstream data-sync process.

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