soc_ai/config.py
Rama main · réplica de lectura del repositorio original.
"""Application configuration via pydantic-settings.
Settings are loaded from environment variables (case-insensitive) and from
a ``.env`` file in the working directory. See ``.env.example`` for the full
surface and inline documentation of each knob.
"""
from __future__ import annotations
import logging
from functools import lru_cache
from ipaddress import IPv4Address, IPv4Network, IPv6Address, IPv6Network, ip_address
from pathlib import Path
from typing import Annotated, Any, Literal
from pydantic import (
AliasChoices,
AnyHttpUrl,
Field,
IPvAnyNetwork,
SecretStr,
field_validator,
model_validator,
)
from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
from soc_ai.dossier.policy import (
DEFAULT_CONFLICT_MIN_OBSERVATIONS,
DEFAULT_CONFLICT_PROMPT_INTERVAL_HOURS,
DEFAULT_MIN_CONFIDENCE,
DEFAULT_STALENESS_HOURS,
)
# The labels the shipped alerts-feed filter unions, and the filter itself. Named
# here rather than written as a literal on the field below because two surfaces
# have to recommend it back to the operator (the doctor's alerts-feed-filter row
# and the alerts console's empty-queue explanation), and a recommendation that
# drifts from the shipped default is worse than no recommendation. Lives in the
# config module, not next to ALERT_LABEL_CANDIDATES in soc_ai.webui.alerts_query,
# only because that module imports Settings from here.
#
# ``tags:alert`` is Security Onion's own convention; ``event.kind:alert`` is the
# ECS field Elastic Defend endpoint alerts carry instead. Neither is a superset
# of the other on a measured grid, which is why the default is the union.
DEFAULT_ALERT_LABELS: tuple[str, ...] = ("tags:alert", "event.kind:alert")
DEFAULT_ALERTS_QUERY: str = " OR ".join(DEFAULT_ALERT_LABELS)
# The Security Onion login flows soc-ai can run, in display order. Named here
# because three places must agree on the exact strings: the Settings validator,
# the Config console option list, and the login client that picks a flow.
SO_LOGIN_FLOWS: tuple[str, ...] = ("auto", "browser", "api")
# ``soc-ai serve`` writes the address it hands to uvicorn into this variable, so
# the app can name its own bind at start. It is not a setting. Under a bare
# uvicorn command (the systemd unit, the container) the bind is a uvicorn
# argument, the variable is unset, and the app does not claim to know the bind.
SERVE_BIND_ENV = "SOC_AI_SERVE_BIND"
class Settings(BaseSettings):
"""Top-level configuration for soc-ai."""
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
case_sensitive=False,
# The admin Danger Zone persists connection/secret overrides that are
# re-applied onto this singleton at startup via setattr. validate_assignment
# makes those assignments coerce + validate exactly like construction (a
# str → AnyHttpUrl/SecretStr, a list → typed list), so a DB-stored override
# lands as the right type. The ONLY setattr site is
# config_overrides.apply_to_settings.
validate_assignment=True,
)
# --- Security Onion grid -------------------------------------------
so_host: AnyHttpUrl
so_username: str
so_password: SecretStr
so_verify_ssl: bool = True
so_ca_bundle: Path | None = None
# SO 3.0.0 mounts Kratos under /auth/... (the older path is /self-service/...).
# Default matches SO 3.0.0; older grids can override to "" to skip the prefix.
so_kratos_path_prefix: str = "/auth"
so_login_flow: str = "auto"
"""How soc-ai logs in to Security Onion.
``auto`` runs the Kratos browser flow first. It falls back to the Kratos API
flow if the browser flow cannot complete. ``browser`` forces the browser
flow, which SO 3.3 and later need. ``api`` forces the API flow, which SO 2.4
and SO 3.0 to 3.2 accept. soc-ai keeps the flow that works for the life of
the process. A change applies at the next restart."""
@field_validator("so_login_flow", mode="before")
@classmethod
def _validate_so_login_flow(cls, v: Any) -> Any:
"""Lowercase + validate so_login_flow in {auto, browser, api}."""
if not isinstance(v, str):
raise ValueError(
f"so_login_flow must be a string. The value has type {type(v).__name__}."
)
lowered = v.strip().lower()
if lowered not in SO_LOGIN_FLOWS:
raise ValueError(
f"so_login_flow must be one of: {', '.join(SO_LOGIN_FLOWS)}. The value is {v!r}."
)
return lowered
# Default timezone for ack/escalate calls (matches the SO web UI default).
so_timezone: str = "America/New_York"
# --- Connect API (Pro feature, optional fallback) ------------------
so_client_id: str | None = None
so_client_secret: SecretStr | None = None
# --- Elasticsearch (typically the same cluster as SO ES) -----------
es_hosts: Annotated[list[AnyHttpUrl], NoDecode]
es_username: str | None = None
es_password: SecretStr | None = None
es_verify_ssl: bool = True
# Optional path to a CA bundle for the ES connection specifically — mirrors
# so_ca_bundle/misp_ca_bundle. Without this, a self-hosted ES behind a
# private CA has no option between cert failures and ES_VERIFY_SSL=false
# (which disables validation on the channel carrying the ES basic-auth
# credential and every raw alert/event document).
es_ca_bundle: Path | None = None
# Per-request timeout in seconds. 30s balances a typical grid under moderate
# concurrency (several runs * ~6 pivot queries each = dozens of simultaneous
# ES queries) against failing fast on a wedged cluster. Lower for fast
# production clusters; raise for shared/contended ones that see
# `ConnectionTimeout` under batch load.
es_request_timeout_s: int = 30
# Transport-layer retries for transient ConnectionTimeout / 5xx from
# ES under heavy concurrency. The elasticsearch-py client retries
# internally, so we don't need to wrap call sites by hand. Bounded so a
# fully-unreachable grid can't stall a call for request_timeout x (1+retries):
# 30s x 3 = 90s worst case (was 60 x 4 = 240s).
es_max_retries: int = 2
# Elasticsearch defaults `allow_partial_search_results=true`: a search over
# failed or unassigned shards comes back HTTP 200 with PARTIAL — often zero
# — hits and no error, so one dead data node reads as a quiet network right
# across the console. When true (default) soc-ai detects that locally and
# raises, which every route already maps to the standard 503 "grid
# unavailable" and the agent already reads as a failed tool call. Set false
# only if you knowingly run with a chronically red shard and prefer partial
# data to an error: failures are then logged at WARNING and the hits used
# as-is.
# Scope: this governs what a QUERY does about a partial read. It does not
# govern whether the operator is told there was one. The health probe reads
# with `require_complete=True` and so do the callers whose answer is a claim
# about absence, so /health, the degraded banner and the bell keep reporting
# a half-read grid whichever way this is set.
es_fail_on_partial_results: bool = True
# Orchestrator-level retries around the WHOLE Phase-A prefetch call
# (`get_enriched_alert_context`), on top of `es_max_retries` above. The ES
# client's own retry budget covers a single pivot query hiccup; this covers
# a longer blip (a restarting node, a few seconds of network flap) that
# exhausts that budget and would otherwise raise straight through the
# pipeline's one hard-fail path (no evidence was ever gathered, so unlike
# every synthesis-phase failure there is no fallback report — a manual
# re-hunt is the only recovery). Only ConnectionError / elasticsearch-py
# TransportError (transient-infra) are retried; SoNotFoundError and
# validation errors fail through immediately (see `_prefetch_retry.py`).
prefetch_max_retries: int = 2
# Base seconds for the prefetch retry's exponential backoff (jittered,
# same formula as `litellm_max_retries`'s gateway transport). Attempt N
# waits random() * min(30s, base * 2**N).
prefetch_retry_base_delay_s: float = 2.0
# Hard wall-clock bound for INTERACTIVE console grid queries (alerts list +
# group events). Below the worst-case ES retry budget so the UI fails fast
# with a clean "grid unavailable" error instead of hanging while the client
# retries a down/slow Elasticsearch. Background hunts are not bounded by this.
webui_grid_timeout_s: int = 12
# --- LiteLLM gateway -----------------------------------------------
litellm_base_url: AnyHttpUrl
litellm_api_key: SecretStr | None = None
# Disable for self-signed certs in homelab deployments. Defaults true.
litellm_verify_ssl: bool = True
# Per-request HTTP read timeout in seconds. The synthesizer (heavy 120B)
# can take 60-150s for a single TriageReport call on a busy GPU; under
# batch concurrency=2+ that crosses the old 120s default. 300s gives
# headroom for slow-but-progressing calls without masking genuine hangs
# (the harness layer enforces a separate per-run wall-clock cap).
litellm_request_timeout_s: float = 300.0
chat_turn_timeout_s: int = 300
"""Wall-clock timeout for a single chat turn (seconds). A chat turn may
issue several tool calls, so this should be larger than
litellm_request_timeout_s. When the turn exceeds the limit it is bounded by
an ``asyncio.timeout`` block inside the turn (raising ``TimeoutError``, a
normal Exception), so the turn's error handler writes a terminal error row
and the pending status never gets stuck. Bumped 180→300 after
a legitimately-deep follow-up needed >180s; raise further for slower GPUs."""
hunt_chat_turn_timeout_s: int = 600
"""Wall-clock timeout for a single HUNT-chat follow-up turn (seconds).
Larger than ``chat_turn_timeout_s`` (the investigation chat): a hunt chat
turn reasons over a broad HuntReport (many findings/hosts) and can issue
wider OQL pivots, so it legitimately runs longer than an alert-scoped
follow-up. Bounded by the same ``asyncio.timeout`` block inside the turn
(raising ``TimeoutError``, caught by the turn's error handler → terminal
error row, never a stuck ``pending``). Raise on a slower stack."""
hunt_run_timeout_s: int = 1800
"""Whole-hunt wall-clock safety net for a slow stack (seconds).
A hunt's own request/tool budget (``hunt_request_limit`` /
``hunt_tool_calls_limit``) governs the NORMAL stopping point; this is the
backstop for a HUNG LLM stream that would otherwise stall the background
hunt task indefinitely (no wall-clock limit). On expiry the exploration
run's ``asyncio.timeout`` raises ``TimeoutError``, which falls through to
the SAME partial-report synthesis path used for budget exhaustion — so the
hunt lands a grounded PARTIAL report from what it gathered, not an empty
error. Generous (30 min) so a legitimately-broad hunt on a slow model is
never cut short; raise for an even slower stack."""
investigation_run_timeout_s: int = 900
"""Whole interactive-investigation wall-clock safety net (seconds).
Bounds a single on-demand ("Investigate") background run end to end. Like
the hunt/chat backstops, this is NOT the normal-path bound (the agent's
request budget governs that) — it is the safety net for a wedged LLM stream
that would otherwise leave the background task (and the investigation row)
running forever. On expiry the run's ``asyncio.timeout`` cancels the drain,
which the recorder's terminal-state handler lands as ``status='error'``
(the interrupted-run path), so the row never sticks in ``running``. Larger
than a per-turn cap but tighter than a hunt (a single investigation is
narrower than a broad hunt)."""
investigation_turn_timeout_s: int = 600
"""Per-primary-agent-run wall-clock backstop for an investigation (seconds).
Bounds a SINGLE investigator/synthesizer agent run inside the pipeline (the
per-turn analogue of ``chat_turn_timeout_s``), so one wedged model call can't
consume the whole ``investigation_run_timeout_s`` budget on its own. Consumed
by the orchestrator (it wraps the agent run in an ``asyncio.timeout`` and
surfaces a ``TimeoutError`` to the existing per-run handler, which now
concludes GRACEFULLY with the round-1 verdict rather than erroring out).
Deliberately larger than ``litellm_request_timeout_s`` (300) so it is a true
BACKSTOP for a genuinely hung stream, not a racer against a single turn that
is legitimately retrying the gateway (``litellm_max_retries`` with jittered
backoff) on a slow stack — cutting such a turn mid-retry would throw away work
that was about to succeed. Raise further for very slow GPUs."""
sigma_draft_timeout_s: float = 150.0
"""Server wall-clock budget for a single Sigma-draft model call; kept below
the SPA's client draft timeout so the server returns an honest 504 before
the client gives up."""
auto_triage_per_target_timeout_s: int = 1200
"""Wall-clock backstop for a single auto-triage investigation (seconds).
An investigation is heavier than a chat turn — several tool calls plus the
synthesis pass — so this is larger than ``chat_turn_timeout_s``. It is NOT a
normal-path bound (the investigation's own request budget governs that); it
is the safety net for a HUNG LLM stream, which otherwise has no wall-clock
limit and stalls the entire sequential sweep behind it (the exact failure
that wedged a batch mid-run). On expiry the per-target ``asyncio.timeout``
raises ``TimeoutError`` — caught by the loop's per-target handler, counted as
a failure — and the sweep proceeds to the next target.
MUST stay strictly greater than ``investigation_run_timeout_s`` — enforced at
use by ``_effective_per_target_timeout``. This is the OUTER of two nested
whole-run guards, and they do not report equally well. The inner one
(``recorded_run``'s ``investigation_run_timeout_s``) knows WHY the run died:
it records a ``whole_run_backstop`` error naming the wall clock and pointing
at the model backend, and it lets the orchestrator's own per-turn timeout
(``investigation_turn_timeout_s``) conclude gracefully with the round-1
verdict first. The outer cap can only cancel the generator from the consumer
side, which records the far less useful ``run_cancelled``. Whichever fires
first decides what the operator gets to see, so the useful one must win.
Raised 600 -> 1200 (2026-08-03). At 600 the outer cap was TIGHTER than the
inner 900s backstop, so it pre-empted both graceful paths on every slow run.
In the preceding 14 days that produced 49 of 64 error rows with no error
event at all, every one ``auto-triage:scheduler``, 47 finishing at exactly
10.0 minutes. Completed runs in the same window ran p50 2.1min / p95 5.7min /
p99 8.6min, so 600s was already clipping the live tail; anything that slows
runs down (a degraded backend, a slower model) pushes more of the
distribution into it. NOTE: the silent-row half of that failure is fixed
independently in ``recorded_run``, which now persists a terminal error event
on the timeout, cancel and crash paths — this setting governs which cause
gets reported, not whether one is reported at all."""
litellm_max_retries: int = 5
"""How many times the OpenAI client retries a gateway request on a transient
error (connection drop, 429, 5xx). The openai default is 2, which can't ride
out a brief LiteLLM-gateway blip (observed ~15s) — the investigator then
fails and the run collapses to a misleading needs_more_info. 5 with the
client's exponential backoff covers a short outage."""
# --- Model aliases (defined in the LiteLLM config) -----------------
# The single model the analyst agent uses for every triage. (There is no
# separate "fast" tier; the optional cloud second opinion is the Oracle,
# configured under `oracle_*` below and OFF by default.) ``HEAVY_MODEL`` is
# still accepted as a deprecated alias so older .env files keep working.
analyst_model: str = Field(
default="soc-ai-analyst",
validation_alias=AliasChoices("ANALYST_MODEL", "HEAVY_MODEL"),
)
analyst_cloud_redaction: bool = False
"""Redact internal identifiers from EVERYTHING sent to the analyst model.
Opt-in privacy gate for deployments that point ``analyst_model`` at a
CLOUD provider. When True, every payload sent to the analyst model —
the enriched alert context, all tool results, and the composed prompts
(investigation, hunt, and chat) — has internal IPs, hostnames, usernames,
MACs, and internal-domain emails replaced with stable opaque labels
(``IP_01``, ``HOST_02``, …) via the same reversible redaction tunnel the
Oracle path uses (:class:`soc_ai.agent.egress_guard.EgressGuard`). Every
model OUTPUT — verdicts, rationales, reasoning traces, hunt reports, chat
replies — has those labels restored to the real values before
storage/display, and tool arguments coming FROM the model (e.g. a query
string citing ``HOST_01``) are label-restored before they hit
Elasticsearch, so the agent loop still works end to end.
COST: some verdict quality. The model reasons over opaque labels, so it
cannot use identity knowledge it would otherwise infer — e.g. it can't
recognise ``dc01`` as a domain controller, or that ``HOST_03`` is the CEO's
laptop. Cross-references between labels are preserved, so behavioural
reasoning (beaconing, lateral movement patterns) is unaffected.
Leave False (the default) for a local model — redaction is pure overhead
when the analyst model never leaves your network."""
analyst_redaction_fail_closed: bool = False
"""Fail CLOSED on residual internal identifiers in the analyst egress path.
Only meaningful when ``analyst_cloud_redaction`` is on. The redaction
tunnel above is best-effort — it sanitizes each payload but has no
independent check that the sanitize pass actually removed everything. When
this is True, the FINAL composed outbound string for each analyst-model call
is swept by :func:`soc_ai.oracle.sanitize.unsafe_residue` (an INDEPENDENT
detector, re-implemented from scratch so a sanitize bug cannot blind it); if
any internal identifier survived, the model is NOT called and the run lands
a pipeline error (``resolution.provenance='pipeline_fallback'``, the same
honest-fallback shape a synth crash produces) naming only the leaked
identifier COUNT — never the values.
Leave False (the default) to keep the current best-effort behavior: a
sanitize miss is logged but the redacted payload still egresses. Turn on
for a deployment where a leak is worse than a blocked investigation. No
effect at all when ``analyst_cloud_redaction`` is off (no guard is built)."""
# --- Audit logging -------------------------------------------------
audit_index_alias: str = "soc-ai-audit"
audit_redact: bool = True
"""Redact secret-shaped strings from audit records before the ES write.
Default True: soc-ai's audit log lands in a *shared* ES cluster, so any
credential that leaks into audit content would be readable by everyone with
cluster access. Redaction fires only on secret *shapes* — ``scai_`` tokens,
``Bearer`` headers, ``X-Session-Token``, ``password=``, AWS/GitHub keys, and
similar patterns — so normal audit content (verdicts, tool calls, alert ids,
reasoning) is untouched. Set False only if you need verbatim audit bodies and
accept the leak risk."""
audit_fail_closed: bool = True
"""Abort SO-mutating actions (ack/escalate/comment/auto-ack) when their audit
record cannot be written. Default True: no state change without an audit
trail — a write tool returns an error instead of silently succeeding without
a record. This trades *availability* for *accountability*: if the audit ES
index is unreachable, mutating actions fail until it recovers. Read/triage/
enrichment audit writes stay fail-open regardless (audit loss never crashes a
read). Set False to revert mutating writes to fail-open (the pre-1.x
behaviour) if availability matters more than a guaranteed audit record."""
audit_verify_schedule_enabled: bool = True
"""Verify the tamper-evident audit chain on a schedule.
ON by default, unlike every other background job here, and for a reason
that is specific to this one. A tamper-evident log nobody verifies is a
log: the product's claim is that its record of a decision cannot be edited
afterwards, and until 2026-09 the only thing that ever tested that claim
was an operator pressing a button nobody presses. Live deployments carried
a broken chain for weeks with nothing saying so. The other schedulers
default off because they spend money (model calls) or change state (acks);
this one is a bounded read of one index, so the case for making the
operator opt in to checking their own audit trail is weak. Turn it off if
the scan is unwelcome; nothing else depends on it."""
audit_verify_schedule_interval_hours: int = 24
"""Hours between scheduled audit-chain verifications. Daily by default: a
break is a standing condition rather than a fast-moving one, and the scan
reads every record in its window. Only used when the schedule is on."""
audit_verify_days: int = 7
"""How many days back a scheduled verification reads.
A window, not the whole index, for two reasons. The scan cost grows with
the trail, and a deployment's OLD history can carry frozen scars that no
longer say anything about today — the current epoch is what "am I sound
right now" turns on. Set higher to widen the daily check; the full-index
scan stays available on demand (``soc-ai audit verify`` with no
``--days``). A windowed scan cannot verify the link across its own start
boundary, which the verifier accounts for rather than reporting as a
break."""
# --- Local enrichment ----------------------------------------------
misp_url: AnyHttpUrl | None = None
misp_api_key: SecretStr | None = None
# Verify the MISP TLS cert. Default True (secure). Homelab MISP often uses a
# self-signed cert — set MISP_CA_BUNDLE to its CA, or MISP_VERIFY_SSL=false to
# disable verification entirely (insecure; the API key transits this channel).
misp_verify_ssl: bool = True
misp_ca_bundle: Path | None = None
internal_cidrs: Annotated[list[IPvAnyNetwork], NoDecode] = [
IPv4Network("10.0.0.0/8"),
IPv4Network("172.16.0.0/12"),
IPv4Network("192.168.0.0/16"),
]
# --- Index patterns ------------------------------------------------
# Patterns that match the SO indices. On Security Onion 3.x the Suricata/Zeek
# event + alert data lives in Elasticsearch DATA STREAMS named `logs-*`, while
# cases/detections/playbooks use the older `so-*` indices. The defaults below
# target a SINGLE-NODE grid.
#
# WHY THE DEFAULT IS `logs-*` AND NOT `.ds-logs-...`
# A data stream's documents physically sit in hidden backing indices named
# `.ds-<stream>-<date>-<generation>` (e.g.
# `.ds-logs-suricata.alerts-so-2026.08.05-000001`). A search pattern never has
# to name those: `logs-*` matches the DATA-STREAM name and Elasticsearch
# expands it to every backing index underneath. (The leading dot is why the
# two forms are not interchangeable — dot-prefixed indices are hidden and a
# plain `logs-*` would never match a `.ds-…` name directly.) Writing the
# pattern against backing indices instead forces you to spell out the Elastic
# Agent NAMESPACE segment, and that segment is the trap:
# `.ds-logs-*-so-*` SO's own integrations — suricata, zeek, soc,
# kratos, strelka, import, detections.
# `.ds-logs-*-default-*` Elastic's stock integrations — system.auth,
# system.syslog, endpoint, winlog. This is the
# login/syslog evidence an investigation leans on.
# `logs-synth-*` soc-ai's own synthetic / eval data.
# On 2026-08-05 a deployment ran `.ds-logs-*-so-*,logs-synth-*` and lost the
# entire `-default-` namespace: ~117K auth and ~48M syslog records were
# invisible to every query, with no error — the pattern still matched 139M
# documents, so the doctor's "matched no documents" check stayed green — and
# an investigation was wrong because of it. `logs-*` covers all three groups
# above plus any namespace an operator adds later. Narrow it only for a
# concrete reason, and never to a single namespace.
#
# MULTI-NODE / distributed grids reach the data through cross-cluster search,
# so prefix every pattern with the remote-cluster wildcard, e.g.
# `*:logs-*`, `*:so-case*`. `setup.sh` AUTO-DETECTS the right prefix (it
# probes the grid during the ES check) and writes the concrete values to
# `.env`, so the guided install gets this right on either shape.
#
# NOTE: the legacy `*:so-*` default matched the old `so-*` indices, NOT the
# `logs-*` data streams where alerts live — it left the console empty on a
# healthy grid. Do not reintroduce it.
events_index_pattern: str = "logs-*"
cases_index_pattern: str = "so-case*"
detections_index_pattern: str = "so-detection*"
playbooks_index_pattern: str = "so-playbook*"
# --- Internal-identifier discovery ---------------------------------
# Auto-discovery of internal domain suffixes + bare internal hostnames from
# Security Onion data, for the Oracle egress sanitizer (so it redacts a
# deployment's own internal hosts before cloud egress without hand-config).
# See soc_ai/enrichment/discovery.py and the managed list in
# soc_ai/store/internal_identifiers.py.
discovery_enabled: bool = True
"""Master switch for the internal-identifier discovery job (the daily timer
and the `discover-internal-identifiers` CLI / scan-now endpoint). Off skips
the scan entirely — operator-set identifiers still apply."""
discovery_lookback_days: int = 7
"""ES query window (days) the discovery scan aggregates over to learn
internal domain suffixes + hostnames."""
discovery_min_hosts: int = 3
"""Distinct-internal-host count at/above which a clearly-internal candidate
auto-activates as a redaction rule. Below it, the candidate lands `muted`
(a suggestion). A public registrable domain NEVER auto-activates regardless
of this count — it is always muted."""
discovery_schedule_enabled: bool = False
"""Run the internal-identifier discovery scan automatically on a schedule
(an in-process background loop in the API server). Off by default — the
scan still runs on demand via the 'Scan now' button / the
`discover-internal-identifiers` CLI. When on, the loop honors
`discovery_enabled` (the master switch) and the interval below. Toggling
this in the config console takes effect live (no restart)."""
discovery_schedule_interval_hours: int = 24
"""Hours between automatic discovery scans when `discovery_schedule_enabled`
is on. Bounded 1..168 (hourly to weekly). The scheduler checks elapsed time
against the last completed scan, so a freshly-toggled-on schedule that has
never run scans on the next wake."""
# --- Host dossier ---------------------------------------------------
# Durable, system-inferred asset context: what a host IS, as opposed to what
# it did in the last 24h. A deterministic rule-based sweep aggregates ES per
# internal host into an inference lane; an operator overlay writes a
# physically separate lane that always wins at read time. See
# soc_ai/dossier/ and soc_ai/enrichment/host_dossier.py.
dossier_enabled: bool = True
"""Master switch for the host dossier — the sweep, the tool and the prompt
block. Off, `t_host_dossier` answers "host dossier disabled" rather than
disappearing from the tool surface: an agent that can see a tool and be told
it is off will not invent the answer, and one that never sees it may."""
dossier_schedule_enabled: bool = False
"""Run the network sweep automatically on a schedule (an in-process
background loop in the API server). Off by default, matching every other
background job here even though the sweep spends no tokens: it is a few
hundred Elasticsearch aggregations against a live production grid, and the
operator who installed soc-ai an hour ago should decide when that starts.
On demand via 'Rebuild now' meanwhile. Toggling takes effect live."""
dossier_schedule_interval_hours: int = 24
"""Hours between automatic dossier sweeps when `dossier_schedule_enabled` is
on. Bounded 1..168 (hourly to weekly). Daily by default because the thing
being measured — what a host is for — changes on the timescale of a
provisioning ticket, not a shift. The due-check reads the durable
`dossier_run` table rather than an in-memory stamp, so a restart loop cannot
re-sweep the whole network on every boot."""
dossier_lookback_days: int = 14
"""ES window (days) each host's dossier observations are aggregated over.
Wider than `discovery_lookback_days` (7) on purpose: the claim that carries
weight in an investigation is "this host has never initiated outbound SSH",
and seven days spans one weekend and one patch cycle; fourteen spans two, so
a Monday-only backup job or a fortnightly maintenance window reads as routine
instead of as a first-ever event.
This is the DOSSIER's window, for identity and role. The behavioural
BASELINE the hunting layer scores departures against is a separate thing
with its own window (`entity_profile_window_days`, 30) -- two knobs that
used to share one name, so lowering this one looked like it tightened the
baseline and did not."""
entity_profiles_enabled: bool = False
"""Whether the dossier sweep builds behavioural profiles.
OFF by default, and it stays off until a shadow week has been read. The
design is explicit that nothing in this layer goes live before the replay
gate passes and the observation distribution has been looked at — a
baseline that has never been measured against a real network is a source
of confident nonsense, not of findings."""
entity_profile_window_days: int = 30
"""ES window (days) a behavioural profile is built over.
Wider than `dossier_lookback_days` (14) because a profile is a claim about
what is ORDINARY, and the three time cells it buckets into need enough
samples each to have a dispersion at all. Thirty days spans four weekends
and a monthly cycle, so a month-end job reads as routine rather than as a
first-ever event."""
entity_profile_lag_hours: int = 24
"""How far before the present a behavioural baseline stops.
Must match the prior sweep's recent window. A baseline built right up to
now CONTAINS the window it will be compared against, so every member the
sweep observes is already in it and novelty is structurally impossible —
on the range that produced 656 evaluations, zero findings, and no visible
bug anywhere. The window ends where the recent window begins."""
profile_build_workers: int = 2
"""Profile batches the build reads at once.
The build splits the estate into batches of up to 500 hosts. Each batch
makes one search per dimension, one after the other. This many batches run
at the same time. Two keeps the grid's search queue short on a large
estate and still halves the wall time against one."""
profile_estate_rare_hosts: int = 3
"""Below this many hosts holding a member, a new member is estate-rare.
A new member that fewer hosts in the estate hold says more than one that
other hosts hold, so its observation is born heavier. The same number is
the floor of an estate-common member: one host is not a trait of the
estate, however small the estate is."""
profile_estate_common_share: float = 0.2
"""Above this share of the profiled hosts, a new member is estate-common.
An estate-common member is a trait of the estate. It forms no
observation, and the prior note counts it. A member must also clear the
rare bar of hosts to be common."""
estate_model_enabled: bool = False
"""Whether the estate model fits once a day, in shadow.
The model groups the hosts that act alike and scores each host against
the estate (soc_ai.hunting.estate_model). Its observations are always
shadow. When it is on, a host with no confident role reads its learned
group as its peer group. OFF by default: it needs the ``ml`` extra, and a
learned group changes which members the profile tests call a trait of the
peers."""
dossier_max_hosts_per_run: int = 200
"""Hosts built per sweep. Each host costs up to seven ES round trips, built
sequentially — 200 is roughly 3.5 minutes of wall clock. Overflow is not
dropped: rows are selected staleness-first (`last_built_at ASC NULLS FIRST`)
from a durable column, not walked with a cursor, so the hosts skipped this
sweep are exactly the ones at the front of the queue on the next."""
dossier_max_hosts: int = 5000
"""Cap on stored dossier rows. A scanned /16 puts 65k source IPs through the
network aggregation in one afternoon; without a cap the table (and the ~12
field rows per host behind it) grows to match. Pruning deletes oldest-
`last_seen` first and NEVER touches a row carrying an operator override —
losing hand-entered criticality to a port scan would be the worse outcome."""
dossier_stale_address_days: int = 30
"""Days after which an address no census has seen leaves the dossier table.
The cap above prunes only when the table is full, so an address that left
the network stayed for months: production held 122 rows that no census had
found in weeks. A row that holds an operator declaration stays. The machine
that held the address keeps it in its history. A sweep whose census failed
prunes nothing, because a grid outage is not evidence that a host left."""
dossier_min_events: int = 20
"""Events in the window below which the classifier refuses to guess a role.
Identity facts (hostname, MAC, domain) still get emitted — a single DHCP
lease is enough to name a host — but `role` comes back `unknown` with
evidence saying how thin the window was. A host with three packets is not a
workstation; it is a host with three packets, and asserting otherwise is the
exact confident-wrongness this feature exists to remove."""
dossier_min_confidence: float = DEFAULT_MIN_CONFIDENCE
"""Floor (0-1) an inferred value must clear to be used. Two jobs: the
resolver returns "unknown" below it rather than asserting a weak guess, and
the conflict detector will not prod an operator about a disagreement it is
not itself confident in. The classifier emits 0.9 (strong) / 0.5 (weak) /
0.0 (none), so 0.6 admits strong evidence only — deliberately, since the
output of this system lands in a prompt as ambient ground truth."""
dossier_staleness_hours: int = DEFAULT_STALENESS_HOURS
"""Age (hours) past which an inferred value is treated as unknown rather than
reasserted. Three days = three missed daily sweeps, so a single failed run
does not blank the network, but a builder that has been broken or switched off
for a week stops quietly feeding week-old asset facts into live
investigations. Staleness is measured from `inferred_last_run_at` — the last
build that EVALUATED the field, not the last one that found something —
so a host that legitimately went quiet does not read as stale."""
dossier_context_enabled: bool = True
"""Inject the dossier block into investigation, chat and hunt prompts.
Separate from `dossier_enabled` so the data can keep building while the
prompt injection is switched off — the knob to reach for when diagnosing a
prompt-size or model-behaviour regression, without throwing away the
accumulated inference and the operator overrides on top of it."""
dossier_conflict_min_observations: int = DEFAULT_CONFLICT_MIN_OBSERVATIONS
"""Consecutive disagreeing builds before the system prods the operator that
its inference still disagrees with their override. One disagreement is
noise: a transient banner, a DHCP lease mid-renewal, a scan. Three
consecutive builds is a standing signal. The counter resets to zero the
moment a build agrees again, so this is genuinely "continued evidence" and
not a running total."""
dossier_conflict_prompt_interval_hours: int = DEFAULT_CONFLICT_PROMPT_INTERVAL_HOURS
"""Minimum hours between prods about the same field. 336 = 14 days. The
override exists because the operator knows something the telemetry does not,
so the default posture is to shut up; a fortnight is long enough that the
reminder reads as a periodic review rather than nagging. 0 disables prodding
entirely. A 'keep mine' snooze doubles from here per prompt already sent,
capped at 90 days — the nag decays instead of repeating."""
# --- Server --------------------------------------------------------
soc_ai_host: str = "127.0.0.1"
soc_ai_port: int = 8443
soc_ai_tls_cert: Path | None = None
soc_ai_tls_key: Path | None = None
# The source commit of this build. The image stamps it at build time, and
# both deploy scripts write it into .env. Blank means unknown, never a guess.
# GET /api/v1/about and the console About section show it.
soc_ai_commit: str | None = None
log_level: str = "INFO"
@field_validator("log_level", mode="before")
@classmethod
def _validate_log_level(cls, v: Any) -> Any:
"""Uppercase + validate log_level against the stdlib level names.
``logging.basicConfig`` only knows the upper-case spellings while uvicorn
takes lowercase ones, so an operator who writes ``LOG_LEVEL=debug`` the
uvicorn way would boot the server and then lose it in lifespan with a
logging traceback. Normalise here and reject typos at load time instead.
"""
if not isinstance(v, str):
raise ValueError(f"LOG_LEVEL must be a string. The value has type {type(v).__name__}.")
names = logging.getLevelNamesMapping()
upper = v.strip().upper()
if upper not in names:
raise ValueError(
f"LOG_LEVEL must be one of: {', '.join(sorted(names))}. The value is {v!r}."
)
return upper
# --- Web UI / local store -------------------------------------------
soc_ai_data_dir: Path = Path("data")
soc_ai_database_url: SecretStr | None = None
"""The store URL. Empty keeps the store in the SQLite file ``soc-ai.db`` in
``SOC_AI_DATA_DIR``. A ``postgresql+asyncpg://user:password@host:5432/soc_ai``
URL keeps the store in PostgreSQL. ``postgresql://`` means the same driver.
The URL holds a password, so the value is a secret. Env-only. A change takes
effect at the next start. The engine checks the URL at start and names the
driver in the error, never the URL. See docs/DEPLOYMENT.md, "PostgreSQL"."""
soc_ai_database_pool_size: int = Field(default=10, ge=1, le=200)
"""The PostgreSQL connections soc-ai keeps open. Under load the pool opens
the same number again, and closes the extra connections when they go idle.
SQLite ignores the value. Env-only."""
soc_ai_demo: bool = False
"""Read-only public demo mode. Seeds the sanitized fixture set at startup,
refuses all outbound network clients (Elasticsearch restricted to loopback,
for the bundled mock), and turns every mutating endpoint into a structured
refusal except the two replay triggers. Env-only (``SOC_AI_DEMO``) — never
added to the config-console whitelist, so it cannot be flipped from the UI."""
# Secure default: require a login/token for the API. The admin console and
# cross-origin API clients both authenticate; only flip this off for an
# isolated, trusted-network demo where you understand the exposure.
api_auth_required: bool = True
# Cross-origin origins allowed to call the API (e.g. an external automation
# or integration hosted on another origin). CSV; empty = scope to so_host;
# "*" = allow all (NOT recommended for a public deployment). The React /app
# is same-origin and needs none of this.
cors_allow_origins: str = ""
# Extra origins (CSV) that cookie-authenticated mutating requests may carry in
# their Origin/Referer header, folded into the CSRF allowlist alongside the
# app's own origin and ``cors_allow_origins``. Default empty. Same-origin
# requests from the React SPA need nothing here; this is an escape hatch for
# a reverse-proxy hostname or an extra UI host. (Distinct from CORS: CORS
# governs cross-origin *browser fetches*; this governs CSRF Origin checks.)
csrf_trusted_origins: str = ""
# Expose the interactive API docs (/docs, /redoc) and the raw schema
# (/openapi.json). Off by default — a security product shouldn't publish its
# full admin API surface unauthenticated. Turn on for local development.
expose_api_docs: bool = False
# Content-Security-Policy sent on every response. The default is safe for the
# bundled Vite SPA (self-hosted scripts; Tailwind needs inline styles). Set to
# an empty string to disable, or override for a custom deployment.
content_security_policy: str = (
"default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; "
"img-src 'self' data:; font-src 'self' data:; connect-src 'self'; "
"object-src 'none'; base-uri 'self'; frame-ancestors 'none'"
)
# Coarse per-client-IP request ceiling (fixed 60s window) as flood/DoS
# backpressure. Generous by default so normal UI polling never trips it; set
# 0 to disable (e.g. when a reverse proxy already rate-limits). /healthz is
# always exempt so container health checks keep working under load.
api_rate_limit_per_min: int = 1200
# Reverse-proxy IPs whose ``X-Forwarded-For`` may be trusted for client
# attribution (login throttle + rate limiter). EMPTY by default: the socket
# peer IP is used. Set to your proxy's IP(s) when fronting soc-ai with a
# reverse proxy, else per-IP throttling buckets ALL clients under the proxy
# IP. XFF is never trusted from a peer not in this list (it can be forged).
proxy_trusted_ips: Annotated[list[str], NoDecode] = []
session_ttl_hours: int = 12
bootstrap_admin_password: SecretStr | None = None
config_secret_key: SecretStr | None = None
"""Fernet key (44-char url-safe base64) for encrypting secret config
overrides at rest in the config DB. Generate with
``python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"``
and set ``CONFIG_SECRET_KEY`` in ``.env``. When unset, the Danger Zone can
still edit connection identity but secret VALUES (passwords/keys/tokens) are
not editable via the UI — they stay env-managed."""
webui_alerts_query: str = DEFAULT_ALERTS_QUERY
"""OQL filter selecting which documents the alerts feed treats as alerts.
Two labels, because no single one covers a Security Onion grid. SO's own
ingest pipelines derive ``tags:alert`` for Suricata, Sigma/ElastAlert,
Wazuh and Strelka detections, and never set ``event.kind`` at all. It is
the filter SO's own Alerts page uses, and an ECS-only default would empty
the queue on a stock grid. Elastic Defend endpoint alerts are written by
Elastic's package pipeline instead, which never reaches SO's tag-deriving
pipeline, so they carry ECS ``event.kind:alert`` and no SO tag; the tag
alone cannot see them. The union is a superset of both, so no grid loses
anything it saw before.
The queue is only ever as good as this one line: it feeds the alerts
console, auto-triage, the host activity panels and every hunt that reads
the alert plane. ``soc-ai doctor``'s "alerts feed filter" row measures it
against the grid, because a filter that matches nothing looks exactly like
a quiet network."""
webui_inherit_window_days: int = 7
webui_extra_detections: bool = True
"""Broaden the alerts feed beyond Suricata to SO's other detection outputs:
union ``sigma.alert`` (Sigma hits) and ``zeek.notice`` ATTACK::*
(behavioral threat notices) with the configured Suricata primary, each tagged
by ``kind``. SO catches more than Suricata (e.g. Zerologon fires Zeek dce_rpc,
not a Suricata sig); this surfaces what SO flags by non-Suricata means. Off
restores the Suricata-only feed."""
auto_triage_max_targets: int = 25
"""Safety cap on how many investigation targets a single ⚡ auto-triage run
will queue. On a busy grid the planner can cluster many distinct (rule,
src, dst) flows; without a cap one click could spawn dozens of sequential
hunts. Overflow targets are simply not queued this run (they have no verdict
yet, so the next run picks them up). Set 0 to disable the cap."""
auto_triage_retry_limit: int = Field(default=2, ge=0, le=10)
"""The quick automatic retries of one alert after a failed run.
A run fails when it ends in an error or a pipeline fallback. The first
retry waits 15 minutes, and each next retry waits twice as long. After
the limit, the alert runs again once each 6 hours. A real verdict on the
alert resets the count. An explicit selection and Re-run are not limited.
On 2026-10-27 the model gateway refused the triage schema for 11 hours.
The scheduler ran two alerts 109 times each, one run about every 5
minutes. Environment variable: ``AUTO_TRIAGE_RETRY_LIMIT``."""
auto_triage_inheritance_enabled: bool = True
"""Inherit verdicts across similar alerts. When ON (default), an auto-investigate
sweep skips any (rule, source, destination) cluster that already has a verdict
within ``webui_inherit_window_days`` — it inherits that sibling's verdict instead
of opening a fresh investigation, so triaging every incoming alert stays tenable.
Turn OFF to investigate every cluster independently. Editable live."""
rule_prior_mode: Literal["off", "shadow", "live"] = "shadow"
"""The rule prior rung of the tier 1 ladder (soc_ai.agent.rule_prior).
The rule prior covers a scheduled alert with the verdict of its rule's
latest model-backed run when every safeguard holds: both endpoints inside
the estate, at least ``rule_prior_min_runs`` model-backed false positives
of the rule in the last 7 days and one in the last 24 hours, no open lead
and no fresh observation on either host, not critical, a rule the
detection tuning panel nominates, and no analyst override in the rule's
history. It never acknowledges in Security Onion.
``off`` evaluates nothing. ``shadow`` (the default) runs the model as
before and records what the prior would have decided beside the real
verdict. ``live`` skips the model on a covered alert, except a random
``rule_prior_sample_rate`` share. A sampled run that disagrees suspends
the prior for its rule until an analyst clears it. Editable live."""
rule_prior_min_runs: int = Field(default=5, ge=5)
"""How many model-backed false-positive runs of a rule in the last 7 days
the rule prior needs. Five or more."""
rule_prior_sample_rate: float = Field(default=0.02, ge=0.0, le=1.0)
"""The share of prior-covered alerts that still get a real run in live
mode, chosen at random. A sampled run that disagrees with the prior
suspends the prior for that rule."""
auto_triage_schedule_enabled: bool = False
"""Continuously auto-triage the backlog. When on, a background scheduler
periodically sweeps every untriaged detection at/above ``auto_triage_min_
severity`` (the same scope the ⚡ button uses) so the queue drains itself with
no operator action — the alerts feed lands verdicts on its own. OFF by default
(continuous LLM calls); editable live in the config console. Note: the floor
setting is the SCOPE of a sweep; THIS flag is what makes sweeps run on their
own."""
auto_triage_schedule_interval_minutes: int = 5
"""Minimum minutes between scheduled auto-triage sweeps (each sweep drains up
to ``auto_triage_max_targets``). Lower = the backlog drains faster but more
LLM calls; 5 is a calm default. Only used when the schedule is enabled."""
hunt_schedules_enabled: bool = False
"""Master switch for recurring (scheduled) hunts. When on, a background loop
wakes every ~60s and spawns a hunt for every DUE ``hunt_schedules`` row (its
interval has elapsed), landing it tagged ``kind="scheduled"``. Each schedule
carries its own objective + interval-minutes (managed in the Hunt Console);
THIS flag is the global on/off — with it off no scheduled hunt ever fires,
regardless of the per-row ``enabled``. OFF by default (recurring LLM calls);
editable live in the config console. Single uvicorn worker only (workers>1
would double-fire — Epoch 6.2 territory)."""
hunt_spec_sweeps_enabled: bool = False
"""Run the declarative hunt catalog on a loop.
OFF by default, and the switch to flip LAST. A spec sweep writes findings
with no human in the loop, so the order that earns the flip is: run it by
hand (``soc-ai spec-sweep``), then in shadow for a week
(``--shadow``, which counts what each spec WOULD surface without spending
the fire-once budget), then read the counts, and only then turn this on.
Unlike the LLM hunt schedules above, a sweep costs no model call at all —
two Elasticsearch queries per spec — so the interval floor is minutes rather
than the hour those need."""
hunt_spec_sweep_interval_minutes: int = 60
"""Minutes between catalog sweeps. Floor of 5.
The floor is far below ``hunt_schedules``' 60-minute one because the cost
profile is not comparable: a sweep is two bounded aggregations per spec and
no inference. It is not zero, though — each sweep is a real query against
the analyst's grid — so a floor exists at all."""
hunt_spec_sweep_window_minutes: int = 1440
"""How far back each sweep looks.
Wider than the interval on purpose. A spec that only ever looked back as far
as the last sweep would miss anything that landed during an outage, a
restart, or an ingest lag — the exact hole that makes a live-tailing rule
engine miss an attack whose telemetry arrives while it is not watching. The
fire-once gate is what stops the overlap becoming repeat findings."""
catalog_hunt_rows: bool = False
"""Record a hunt row for each analytic hit, as the sweep did before 1.5.0.
OFF by default. A catalog hit is an observation. It used to be a hunt row
as well, so one hit made three records: the row, a finding inside its
report, and the observation. The hunt list showed those rows beside the
hunts an agent ran, and the Hunt Console subtitle counted them as hunts.
ON restores the old write for one release. The observation is written
either way, so nothing downstream of the observation changes. A
visibility-gap row is not a hit and is not affected: a gap says the
analytic could not see, and silence there is a false all-clear."""
hunting_prior_sweep_enabled: bool = True
"""Run the profile sweep in the app, on its own interval.
ON by default, and the opposite posture to ``hunt_spec_sweeps_enabled``
for the opposite reason. A catalog sweep writes findings unattended, so it
is opt-in. A profile sweep writes observations: it compares each host with
its own baseline, records what departs, and raises nothing. That is the
shadow posture the release ships with, and a deployment that records no
profile observation has no hunting layer at all.
The sweep is the code path ``soc-ai priors --record`` runs. On the test
range a host timer runs it every hour. A container has no timer, so
without this loop the layer shipped and never ran.
OFF stops the sweep. The profile lane in the host dossier still builds the
baselines, and ``soc-ai priors`` still reads them by hand."""
analytic_self_heal_enabled: bool = True
"""Move a live analytic back to shadow when it breaches its own budget.
ON by default. The check runs after each analytic sweep and each profile
sweep. It reads the store only: the live hits of the last 24 hours against
the spec's ``fire_budget_per_day``, and the precision of the hunted leads
of the last 30 days against the spec's ``precision_floor``. A breach writes
a system version row with the numbers and raises a bell entry.
An analytic that declares neither field is never moved, so the default
changes nothing until a spec declares a budget. soc-ai never approves an
analytic to live and never retires one. OFF stops the check. A hold that
is already written stays until an analyst approves or retires the
analytic."""
grid_gap_settle_minutes: int = 60
"""Minutes the sweeps wait after a grid gap ends before they score its hours.
A grid gap is a span of hours the grid held no readable data for soc-ai.
The agents send the events they kept after the grid comes back. The hours
right after the recovery are not final until they do. The profile sweep
and the baseline treat the gap hours and this margin as unknown."""
hunting_prior_sweep_interval_minutes: int = 60
"""Minutes between profile sweeps. Floor of 15.
A sweep makes no model call. It makes several Elasticsearch aggregations
per dimension, so the cost is query load and the floor exists for the same
reason the catalog sweep has one. The default matches the hourly timer the
range has run since the profile lane shipped."""
lead_auto_hunt: bool = True
"""A lead that has never had a hunt starts one when it forms.
ON by default. A lead is already the product of several observations that
crossed a threshold together, so the answer to "is this worth a look?" is
yes by the time the lead exists. Before this the lead waited for a click,
and a lead that formed overnight was still waiting in the morning.
The in-process loop starts the hunt, not the sweep, because leads also form
in the timer process that runs the priors, and that process cannot run an
agent. OFF restores the older behaviour: a lead waits for an analyst to
start the hunt. A dismissed lead is never hunted, and a reopened lead is
not hunted again by the loop."""
lead_auto_hunt_concurrency: int = 2
"""How many lead hunts the loop runs at once.
The count is of running hunts the loop itself started, so a hunt an analyst
started by hand never blocks the loop and the loop never blocks the
analyst. The floor is 1: an operator who wants no lead hunts turns
``lead_auto_hunt`` off, which says so on the Config screen."""
hunt_findings_investigate: bool = True
"""A hunt opens one investigation for each subject finding.
ON by default. The owner's rule: a hunt that finds a sign of an attack
also checks it. A subject finding is a threat finding that names an
entity (soc_ai.hunting.findings.finding_is_subject). The hunt reads
``investigating`` while the investigations run and ``complete`` when the
last one ends. OFF restores the older behaviour: the hunt completes when
its report lands, and the analyst opens each investigation by hand."""
hunt_findings_investigate_max: int = 20
"""The most subject findings one hunt investigates by itself.
A finding past the cap is recorded as over the cap, and the analyst opens
its investigation by hand. The cap bounds the model time one hunt can
claim. Two investigations of one hunt run at a time."""
general_chat_enabled: bool = True
"""Master switch for the general chat: the general scope of the Ask soc-ai panel.
Governs whether the landing screen offers a one-turn agent thread that
answers with the read tools and PROPOSES a hunt when a question needs a
sweep. ON by default because it replaces a surface that already ships (the
Ask box that hands off to the Hunt Console), and because it costs nothing
until an analyst types — the idle page issues one transcript GET and the
poll only re-arms while a turn is pending.
It is a kill switch for the one risk the surface introduces: unlike the
investigation chat (reached by opening an alert) or a hunt (an explicit
launch), this agent sits on the screen every analyst lands on, so it is the
easiest place in the product to spend model capacity by accident — a room of
analysts idly asking it things while the backlog waits behind the same
backend. Off hides the box in the SPA entirely (the flag is published on
``/api/v1/about``) and refuses the endpoints, so a saturated deployment can
reclaim the capacity without a restart. Editable live in the config console;
turning it off does not delete stored threads."""
backtest_max_sample: int = 50
"""Hard ceiling on how many already-dispositioned alerts a single Backtest
run will replay through the agent. Each sampled alert is a FULL LLM
investigation, so a backtest is expensive: the request-level default is 20
and this is the absolute cap the endpoint clamps to (and logs when it does).
Raise only if you know you can afford the LLM cost of the larger replay."""
# --- Agent execution limits ----------------------------------------
# Caps the number of tool calls + LLM requests per investigation so a
# runaway loop can't burn the budget (or the operator's patience). Tuned
# down from 100/50 after a thorough investigation was observed to run
# ~12 tool calls (web_search + host pivots + oracle); 25/18 leaves
# generous headroom while bounding worst-case latency. Raise if a
# deployment legitimately needs deeper loops.
agent_tool_calls_limit: int = 25
agent_request_limit: int = 18
# Schema-retry budget for the investigation-loop agent. 10 was sized for
# Nemotron-30B's schema wobble; stronger models may need far less.
investigator_retries: int = 10
# Max Phase-D targeted-dispatch rounds per investigation. 1 = the original
# hard cap (synth names one tool, one round-2, done). 2 lets the synth
# chain e.g. t_get_event_raw -> t_decode_payload.
phase_d_max_rounds: int = 1
# Hunts explore FAR more broadly than a single-alert investigation (many hosts,
# many queries, cross-time correlation), so they get a bigger budget — otherwise
# the agent runs out of requests before it can synthesize its findings report.
# 60 → 90 on 2026-07-10: the 2026-07-08 inference-engine change moved measured
# per-hunt tool appetite from a median of ~29 calls to ~55-60, so at 60 the
# budget was a wall EVERY hunt hit (4/4 prod hunts capped that night). 90
# restores the pre-swap headroom ratio (~1.5x the median appetite); a hunt
# that still hits it lands a partial report via the synthesizer, not an error.
hunt_tool_calls_limit: int = 90
hunt_request_limit: int = 45
synthesizer_temperature: float = 0.2
"""Sampling temperature for the SYNTHESIZER (the verdict decision). Low for
determinism — the same evidence should yield the same verdict, run to run.
Reduces the FP/TP verdict swing observed on repeated hunts."""
investigator_temperature: float = 0.4
"""Sampling temperature for the INVESTIGATOR (tool selection / pivots).
Moderate — keep some exploration so it doesn't tunnel on one path, but low
enough that the investigation is broadly reproducible."""
verdict_consistency_samples: int = 1
"""Number of independent final-verdict synthesis samples for a
self-consistency majority vote; 1 disables the vote (the default —
a single synthesis call, no vote, and the ``inconclusive`` verdict is
never produced). When >1, the synth-first pipeline re-runs the FINAL
verdict synthesis (same inputs/prompt) that many times and majority-votes
the verdict: a strict majority wins (confidence = mean of the agreeing
samples); a split lands ``inconclusive`` (confidence capped at 0.5).
Bounded [1, 5] — each extra sample is a full synthesizer LLM call."""
@field_validator("verdict_consistency_samples", mode="before")
@classmethod
def _clamp_verdict_consistency_samples(cls, v: Any) -> Any:
"""Validate verdict_consistency_samples is an integer in [1, 5]."""
try:
i = int(v)
except (TypeError, ValueError) as exc:
raise ValueError("verdict_consistency_samples must be an integer in [1, 5]") from exc
if i < 1 or i > 5:
raise ValueError(f"verdict_consistency_samples must be in [1, 5]. The value is {i}.")
return i
# --- Swarm triage experiment (2026-10-08) ----------------------------
# docs/dev/specs/2026-10-08-swarm-triage-experiment-design.md. An
# experiment knob, off the Config console groups on purpose: the arms
# harness sets it per arm, and production keeps ``single``.
investigator_strategy: Literal["single", "spend", "ensemble", "advocates", "planes"] = "single"
"""Which strategy runs the investigation loop. Env: ``INVESTIGATOR_STRATEGY``.
``single`` is the loop as it is. ``spend`` and ``ensemble`` run the same
loop. They name an arm whose limits come from its environment. The
``spend`` arm raises ``agent_tool_calls_limit`` and ``agent_request_limit``.
The ``ensemble`` arm sets ``verdict_consistency_samples``. ``advocates`` runs a
threat advocate and a benign advocate in parallel, and a judge with no
tools writes the report from their briefs. ``planes`` runs one worker per
evidence plane of the alert, and a lead with no tools writes the report.
The setting swaps the loop only. Every step before and after it is the
same for each strategy (soc_ai.agent.swarm)."""
advocate_tool_calls: int = Field(default=10, ge=0, le=200)
"""Tool calls for each advocate of the ``advocates`` strategy. The two
advocates share ``agent_tool_calls_limit``: each one gets the smaller of
this value and half of that limit."""
advocate_request_limit: int = Field(default=8, ge=1, le=100)
"""Model requests for each advocate of the ``advocates`` strategy."""
plane_worker_tool_calls: int = Field(default=6, ge=0, le=200)
"""Tool calls for each plane worker of the ``planes`` strategy. The workers
share ``agent_tool_calls_limit``: each one gets the smaller of this value
and that limit divided by the worker count."""
plane_worker_request_limit: int = Field(default=5, ge=1, le=100)
"""Model requests for each plane worker of the ``planes`` strategy."""
judge_request_limit: int = Field(default=3, ge=1, le=20)
"""Model requests for the judge of ``advocates`` and the lead of ``planes``.
The judge has no tools, so the requests cover the schema retries."""
# --- Synthesis confidence floor -------------------------------------
# A synthesized verdict whose confidence comes back below this floor is
# treated as not-actionable: the post-validators rewrite the verdict to
# needs_more_info when it also lacks semantic citation coverage.
synthesis_confidence_floor: float = 0.6
# --- Investigator response-token cap -------------------------------
# Single investigator turns were observed producing ~13.6K
# reasoning-trace tokens for 2 tool calls — pure waste, and a dominant
# contributor to p95 investigation latency. The cap bounds the response
# (reasoning + content) per turn, so a chatty turn can't dominate the
# wall-clock.
#
# Calibration: measured reasoning_trace sizes — p50 5.6K chars (≈1.4K
# tokens), p95 23.7K chars (≈5.9K tokens), max 23.7K chars. With
# tool-call args on top, the actual response total p95 is closer to
# 6-7K tokens. Iterative calibration:
# - 2500 → rejected immediately ("token limit exceeded before any
# response was generated")
# - 8000 → rejected on first outlier turn
# - 16000 → hit the cap on ~10% of turns, each retry was another
# 16K-token attempt, and in-flight alerts stuck waiting for
# retry-exhaust → per-run timeout → effective error budget exceeded
# - 32000 → covers the worst-observed reasoning trace (~6K tokens)
# with 5x headroom, while still preventing the truly pathological
# "model writes a 50K-token essay before calling final_result"
# degenerate case. Per-turn, not per-investigation.
investigator_max_response_tokens: int = 32000
synthesizer_max_response_tokens: int = 32000
"""Per-call response cap (reasoning + TriageReport combined) for the
synthesizer paths, sent as OpenAI ``max_completion_tokens``.
Without an explicit value the gateway/route default applies — and a
REASONING model can spend that entire accidental budget thinking, so the
call truncates before any structured output is generated
("Model token limit (provider default) exceeded…" → fallback
needs_more_info verdict). 32000 covers the worst observed reasoning trace
with ample headroom; lower it for latency, raise it for very verbose
reasoning models."""
chat_regrounding_attempts: int = 2
"""How many times a chat turn may be re-run to fix ungrounded claims.
The narrative-grounding validator names the per-event facts an answer
asserted that appear in neither this turn's tool results nor the seeded
evidence. Historically that finding only produced a caveat appended to the
published answer — detection without correction. On 2026-08-05 the chat
asserted ``auth.success=true`` to justify overturning a correct
true_positive verdict; the validator flagged that exact claim and the answer
shipped regardless.
With this > 0 the finding is fed back as a correction prompt ("cite it with
a tool call, or remove it") and the turn re-runs. 2 is the default (raised
from 1 in the 2026-08-20 answer-quality batch): the terminal fallback is no
longer a caveat but a mechanical redaction of whatever stays ungrounded, so
a second attempt buys another real chance at a clean, unredacted answer
before that happens — each attempt costs a full turn against
``chat_turn_timeout_s``. 0 restores the old warn-only behavior. Ground-or-
strip redaction remains the terminal fallback when the agent will not
comply."""
analyst_tool_choice_required: bool = False
"""Allow ``tool_choice='required'`` for the analyst model's structured output.
False (the default) keeps today's behavior: pydantic-ai falls through to
``tool_choice='auto'``, the workaround for the vLLM qwen3_coder-parser bug
on the original Nemotron endpoint (see ``_analyst_profile``). Under
``auto`` a weaker model is free to answer in prose instead of calling the
output tool — burning schema retries or exhausting them outright — while
``required`` forces the tool call at the serving layer.
Whether flipping this helps is a PER-BACKEND fact, not a per-size fact:
the llama.cpp CPU tier handles ``auto`` fine, while the short-lived
laguna-s21 backend appeared to need ``required`` (small-N evidence only,
backend since decommissioned). When a new analyst backend lands, measure
with ``soc-ai model-probe --tool-choice required`` before flipping."""
synthesizer_output_mode: Literal["tool", "native", "prompted"] = "tool"
"""How the no-tools synthesizer agents obtain the structured TriageReport.
- ``tool`` (default, today's behavior): pydantic-ai's synthetic
``final_result`` tool call. Depends on the backend's tool-call parser
working — the component that produced the DSML-markup-leak failure class
(tool-call markup surfacing as prose content).
- ``native``: OpenAI ``response_format`` json_schema — server-side guided
decoding. The strongest fix for schema wobble on lesser models: the
server constrains generation to the schema, and the tool-call parser
leaves the path entirely. Requires backend support (vLLM guided
decoding, llama.cpp json_schema).
- ``prompted``: schema goes in the prompt, JSON parsed from plain text.
Weakest guarantees; the escape hatch for backends where BOTH the tool
parser and response_format are broken.
Applies only to the synthesizer agents (synth-first, loop-synth, partial) —
the investigator keeps tool mode because it interleaves real tool calls.
Validate a candidate mode with ``soc-ai model-probe --output-mode`` before
flipping in prod."""
model_context_window_tokens: int = 0
"""The analyst model's input window in tokens, for proactive context
budgeting (``soc_ai.agent.context_budget``).
0 (the default) = auto-discover from the LiteLLM gateway's ``/model/info``
(``max_input_tokens``), fail-soft to no budgeting when the gateway doesn't
publish it. Set explicitly to override discovery (e.g. a gateway that
reports nothing, or to force a smaller budget). When a window is known,
an oversized enriched alert context is trimmed tail-first (oldest pivot
events dropped, ``context_trimmed`` event emitted) before the first model
call instead of blowing the window mid-investigation."""
# --- Synth-first pipeline ------------------------------------------
fast_triage_enabled: bool = True
"""Allow the fast path: finalize a confident first-pass verdict without the
full tool-driven investigation loop. Saves time but can yield shallower
results (a verdict may land with few or no tool calls). Turn OFF to always
investigate with tools. Exposed in the admin config console."""
investigate_when_unsure: bool = True
"""Run a real bounded investigation loop when the synth-first round-1
verdict is not evidence-backed (Theme-1 Task 1).
The synth-first round-1 call is a NO-tools structured-output guess that
rationalizes the prefetch. When its verdict's citations resolve only to
self-referential ``alert.*`` fields (not to actual tool/enrichment/pivot
evidence), and the alert is non-trivial (not a clean-internal benign),
the pipeline runs the tool-bound investigator on the HEAVY model so the
model itself chooses which read tools to call, then re-synthesizes over
the gathered transcript. Trivially-benign alerts keep the fast path.
Flag so we can A/B it on the synth-9 and revert instantly. Set ``False``
to restore the pure zero-tool synth-first behavior."""
investigator_emits_report: bool = True
"""Let the investigation loop write the ``TriageReport`` itself, and skip the
round-2 synthesis call.
Today the loop emits an ``InvestigationTranscript`` (it has no verdict
field), and a second model call re-reads that transcript to write the
report. The 2026-09-19 turn audit measured what the second call adds: no new
evidence in 40 of 41 sampled runs, an empty ``citations`` list in 78% of
recent production runs, and two range verdict changes, both wrong (a correct
``true_positive`` 0.72 became ``needs_more_info`` 0.55 with no disconfirming
fact). It costs 8 s on production and 52 s on the range.
With this ON the loop's own output IS the report. Every gate that ran on the
round-2 report still runs on it: the citation validation and cap, the
confidence floor, the deterministic downgrades, the hard evidence gate, the
self-consistency vote and the Oracle escalation. The egress guard sanitizes
the prompt and restores the report exactly as before.
ON by default since 2026-09-19. The accuracy mission measured both paths on
eleven range alerts with ground truth: this path landed 31 of 31 correct
verdicts against 30 of 31, 42 s faster per investigation. Set it off to
restore the round-2 synthesis. The stored ``triage_report`` event carries
``report_path`` (``investigator`` or ``synth_round2``) on both paths, so the
two stay comparable in the event stream."""
synth_round1_always: bool = False
"""Run the round-1 synthesis even when it cannot settle the alert.
Round 1 is a no-tools call. Its verdict stands only when a dispositive
decision template already cleared the alert ``false_positive``; on every
other alert the investigation loop runs and overwrites it. The pipeline
therefore skips the call on those alerts and emits
``synth_round1_skipped`` with reason ``cannot_settle``. On production the
skipped call cost 18 s and about 8K tokens per alert, and its verdict was
discarded on 83% of the runs that made it (2026-09-19 turn audit, W2).
Set ``True`` to restore that call. The round-1 verdict then reappears in
the timeline and in the crash fallback. The older
``definitely_investigate`` skip (a malware or exploit signal) is not
affected. This is a revert switch for an A/B, not a quality knob: the loop
never reads the round-1 report."""
host_risk_window_hours: int = 24
"""Look-back/forward window (hours, each side) for the host-risk profile.
The 5 tight pivots key on fields (``network.community_id`` / ``host.name`` /
``user.name``) that are absent on network-sensor / so-import-pcap alerts, and
they only span ±``window_seconds`` (5 min) — far too narrow to notice that the
alert's HOST is compromised. For example, an internal SMB leg from a
NetSupport-RAT victim was cleared zero-tool because the C2 check-ins were
~12h away and keyed on fields the pivot never queried. The host-risk profile
is a wide ±N-hour aggregation over the alert's source/destination IPs that
surfaces the endpoint's recent alert histogram so the agent can see "this host
is also firing RAT/C2/malware signatures." Set to 0 to disable."""
investigation_reaper_minutes: int = 30
"""Age (minutes) past which a still-``running`` investigation is considered
stale and marked ``error`` by the periodic reaper. Generous so a legitimately
in-flight hunt is never killed; on startup ALL ``running`` rows are reaped
regardless of age (their background tasks died with the previous process)."""
investigation_reaper_interval_minutes: int = 10
"""How often the background reaper sweeps for stale ``running`` rows. Set the
interval or the age above to 0 to effectively disable the periodic sweep
(startup reap still runs)."""
# --- Local enrichment data directories (no runtime egress) ---------
blocklist_data_dir: Path = Path("/var/lib/soc-ai/blocklists")
"""Directory holding refreshed blocklist files (URLhaus CSV, Feodo Tracker
JSON, Tor exit list, internal_seed.yaml, etc.). Refreshed by the
`soc-ai blocklists refresh` CLI subcommand."""
maxmind_data_dir: Path = Path("/var/lib/soc-ai/maxmind")
"""Directory holding MaxMind GeoLite2 .mmdb files (ASN + City). No command
downloads them: copy GeoLite2-City.mmdb and GeoLite2-ASN.mmdb here by
hand. `maxmind_license_key` only marks the source as configured."""
cloud_prefix_data_dir: Path = Path("/var/lib/soc-ai/cloud_prefixes")
"""Directory holding vendored cloud-provider prefix JSON files (AWS, GCP,
Azure, Cloudflare). Refreshed weekly by the CLI subcommand."""
blocklist_sources: Annotated[list[str], NoDecode] = [
"urlhaus",
"threatfox",
"feodo",
"tor",
"internal_seed",
]
"""Which BlocklistDB sources to load at startup. Spamhaus DROP/EDROP is
OFF by default — it requires a commercial license for paid deployments
and the operator must explicitly opt in via this list AND set
`spamhaus_license_acknowledged=True`."""
spamhaus_license_acknowledged: bool = False
"""Operator acknowledges Spamhaus license terms (free for non-commercial
use; commercial use requires a paid license). Required to enable the
`spamhaus_drop` blocklist source."""
maxmind_license_key: SecretStr | None = None
"""Free MaxMind GeoLite2 license key. Register at maxmind.com/en/geolite2/signup
and put the key in `.env` as `MAXMIND_LICENSE_KEY=...`. Without it,
GeoIP/ASN enrichment is disabled (everything else still works)."""
blocklist_stale_threshold_days: int = 7
"""How many days a blocklist file can be without refresh before the
DB emits a warning event in the audit log. Triage continues to work
with stale data — fail-open."""
abuse_ch_auth_key: SecretStr | None = None
"""abuse.ch Auth-Key for the URLhaus / ThreatFox / Feodo Tracker downloads.
As of the 2024 abuse.ch policy, the CSV/JSON data exports are gated behind
a free Auth-Key sent as the ``Auth-Key`` HTTP header. Register at
https://auth.abuse.ch/ and put the key in ``.env`` as
``ABUSE_CH_AUTH_KEY=...``. When unset, ``soc-ai blocklists refresh`` SKIPS
the abuse.ch feeds with a clear message (the Tor exit list needs no key and
still refreshes). The key is sent only by the refresh job — never during
triage — and is never logged. See ``docs/BLOCKLISTS.md``."""
azure_service_tags_url: AnyHttpUrl = AnyHttpUrl(
"https://download.microsoft.com/download/7/1/D/71D86715-5596-4529-9B13-DA13A5DE5B63/"
"ServiceTags_Public_20241125.json"
)
"""Download URL for the Azure Service Tags prefix JSON. Override to point
at a newer dated snapshot without redeploying (the filename encodes the
publish date). Default matches the file that was current when this
setting was introduced (2026-06-10)."""
cloud_prefix_stale_threshold_days: int = 45
"""How many days the cloud-prefix data can be without a successful refresh
before enrichment appends a staleness warning to ``errors``. 45 days
is generous for a weekly-refresh schedule with a few missed runs."""
# --- PCAP retrieval (SSH + suripcap, behind explicit opt-in) ------
pcap_enabled: bool = False
"""Opt-in to live PCAP retrieval via SSH + Suricata ring-buffer pcap logs.
Requires ``so_ssh_key`` to point at a private key that has been
provisioned on the sensor. Off by default so the package deploys
safely without an SSH key. Set ``PCAP_ENABLED=true`` + the SSH
variables below to enable."""
so_ssh_host: str = ""
"""Hostname or IP of the Security Onion sensor running Suricata pcap-log.
Empty by default — the package ships no environment-specific host. REQUIRED
when ``pcap_enabled`` is True (enforced at startup). Set via ``SO_SSH_HOST``
or in the config console (Danger Zone)."""
so_ssh_user: str = "soc-ai"
"""Remote user to SSH in as. Must have passwordless sudo tcpdump or be
in the ``socore`` group with read access to ``so_suripcap_dir``."""
so_ssh_key: Path | None = None
"""Path to the SSH private key file. Passed as ``ssh -i <key>``. If
``None`` (the default), ssh falls back to its own key discovery — but
the service unit's ``ProtectHome=read-only`` means it won't find
``~/.ssh``, so you almost always need to set this explicitly."""
so_ssh_known_hosts: Path | None = None
"""Path to a persistent SSH ``known_hosts`` file for the sensor.
``None`` (the default) derives ``<soc_ai_data_dir>/known_hosts``. Used as
``UserKnownHostsFile`` so the sensor's host key is accepted on first contact
(``accept-new``) and *remembered* — a later key swap is then rejected rather
than silently trusted. Set via ``SO_SSH_KNOWN_HOSTS`` to override the path."""
so_ssh_sudo: str = "sudo"
"""``sudo`` prefix for the remote tcpdump command. Set to ``""`` if the
SSH user is already in ``socore`` and has direct read access."""
so_suripcap_dir: str = "/nsm/suripcap"
"""Directory on the sensor where Suricata's pcap-log files live."""
so_ssh_timeout_s: int = 120
"""Per-SSH-invocation timeout in seconds (connect + data transfer)."""
pcap_max_packets: int = 50000
"""Maximum packets to decode from a merged PCAP (``decode_pcap`` cap).
Prevents OOM on very large flows; truncation is noted in ``PcapFacts``."""
# --- Web search (SearXNG) -----------------------------------------
web_search_enabled: bool = False
"""Enable the ``web_search`` investigator tool (SearXNG). When False the
tool returns a disabled error without any network I/O. Editable in the
config console."""
searxng_url: str = ""
"""Base URL of the self-hosted SearXNG instance (e.g. ``https://search.example.com``).
Empty disables web search. Requires SearXNG's JSON API
(``search.formats: [json]``). Editable in the config console."""
searxng_verify_ssl: bool = True
"""Verify the SearXNG TLS cert. Set False for a self-signed homelab cert."""
searxng_timeout_s: int = 10
"""Per-request timeout for a web search."""
web_search_max_results: int = 5
"""How many SearXNG results to return to the agent per query."""
# --- Online enrichment (opt-in, runtime egress) -------------------
# Everything else in this app is zero-egress (local-mirror feeds). These
# tools reach OUT to third-party reputation/asset APIs, so they are OFF by
# default and gated by this master flag. Per-provider keys live below (set in
# .env; never stored in the DB). The master flag is hot-editable in the
# config console; keys are display-only there.
allow_online_enrichment: bool = False
"""Master switch for the opt-in online-enrichment tools (GreyNoise, Shodan
InternetDB, …). OFF by default to preserve the zero-egress posture. When off,
each online tool returns a clean 'disabled' result with no network I/O."""
online_enrichment_timeout_s: int = 8
"""Per-request timeout (seconds) for an online-enrichment lookup."""
online_enrichment_verify_ssl: bool = True
"""Verify TLS for online-enrichment HTTP calls (these reach the public
internet, so leave True; only a transparent-proxy setup would need False)."""
update_check_enabled: bool = False
"""Opt-in switch for the About page's 'check for updates' button. OFF by
default to preserve the zero-egress posture. When on, an admin can manually
compare the running version against the latest GitHub release; the check is
manual only (no polling), reuses the online-enrichment timeout/TLS policy,
and sends nothing about the environment — the version is compared locally.
When off, the check makes no network call at all. Editable live in the
config console."""
greynoise_api_key: SecretStr | None = None
"""GreyNoise API key (free Community tier available). Set in .env. Without it,
t_greynoise reports 'not configured' and performs no network I/O."""
shodan_api_key: SecretStr | None = None
"""Shodan API key (paid). Set in .env. Powers t_shodan_host (the full,
authenticated /shodan/host lookup — banners, services, vulns). Without it
that tool reports 'not configured' and makes no request; the free,
keyless t_shodan_internetdb and t_cve_lookup still work (master flag on)."""
# --- Notification routing (opt-in outbound webhook) ---------------
# The ONLY new *outbound* egress path in soc-ai (everything else is local-
# feed / zero-egress). Treated like the Oracle: OFF by default, the webhook
# URL is a SECRET, and every send is audited. When disabled, soc_ai.notify.fire
# returns before constructing any HTTP client — NO httpx call is ever made.
notify_enabled: bool = False
"""Master switch for outbound notification webhooks. OFF by default to
preserve the zero-egress posture. When off, ``soc_ai.notify.fire`` is a hard
no-op (returns before any network I/O). All the per-trigger toggles below are
inert unless this is on. Editable live in the config console."""
notify_webhook_url: SecretStr | None = None
"""Destination webhook URL for notifications (a SECRET — Fernet-encrypted at
rest, never rendered back). No sends happen until this is set. Point it at a
generic JSON receiver, a Slack incoming-webhook, or a Matrix hook (pick the
body shape with ``notify_format``). Set via the config console's Notifications
section or ``NOTIFY_WEBHOOK_URL`` in ``.env``."""
notify_format: str = "json"
"""Webhook body shape: ``json`` (a compact generic dict), ``slack``
(``{"text": ...}``), or ``matrix`` (``{"msgtype":"m.text","body":...}``).
An unknown value falls back to ``json``."""
@field_validator("notify_format", mode="before")
@classmethod
def _validate_notify_format(cls, v: Any) -> Any:
"""Lowercase + validate notify_format ∈ {json, slack, matrix}."""
if not isinstance(v, str):
raise ValueError(
f"notify_format must be a string. The value has type {type(v).__name__}."
)
lowered = v.strip().lower()
if lowered not in ("json", "slack", "matrix"):
raise ValueError(
f"notify_format must be one of: json, slack, matrix. The value is {v!r}."
)
return lowered
notify_verify_ssl: bool = True
"""Verify the webhook's TLS certificate. Set False only for a self-signed
internal receiver (e.g. a homelab collector on an internal CA)."""
notify_tp_confidence_threshold: float = 0.9
"""Minimum verdict confidence (0-1) for a true-positive to ping on-call. A TP
below this fires no notification — high-confidence TPs are the ones worth
waking someone for. Only used when ``notify_on_tp`` (and the master switch)
is on."""
@field_validator("notify_tp_confidence_threshold", mode="before")
@classmethod
def _clamp_notify_tp_threshold(cls, v: Any) -> Any:
"""Clamp notify_tp_confidence_threshold to [0.0, 1.0]."""
try:
f = float(v)
except (TypeError, ValueError) as exc:
raise ValueError(
"notify_tp_confidence_threshold must be a number in [0.0, 1.0]"
) from exc
if f < 0.0 or f > 1.0:
raise ValueError(
f"notify_tp_confidence_threshold must be in [0.0, 1.0]. The value is {f}."
)
return f
notify_on_tp: bool = True
"""Notify when an investigation finalizes verdict=true_positive at/above
``notify_tp_confidence_threshold``. Inert unless ``notify_enabled`` is on."""
notify_on_hunt_threat: bool = True
"""Notify when a hunt lands a report containing a finding with
category='threat'. Inert unless ``notify_enabled`` is on."""
notify_on_model_fitness_fail: bool = True
"""Notify when the model-fitness probe grades the analyst model FAIL (an unfit
model silently ruins triage). Inert unless ``notify_enabled`` is on."""
notify_on_quality_regression: bool = True
"""Notify when the nightly quality micro-eval (``soc-ai eval-nightly``)
detects a regression against its own trailing history — an agreement drop,
an error-rate spike, or a fallback-rate jump (see
:func:`soc_ai.eval.quality.detect_regression`). This is the alarm that
catches silent verdict degradation after an inference-engine swap. Inert
unless ``notify_enabled`` is on."""
notify_on_audit_chain_break: bool = True
"""Notify when the scheduled verification finds the tamper-evident audit
chain broken — either a position claimed twice by two writers or a record
whose content no longer matches its own hash. This is the only alarm here
that can mean someone edited the record of a decision, so it is sent at
critical severity. Inert unless ``notify_enabled`` is on; the bell in the
app carries it either way."""
notify_on_tls_expiry: bool = True
"""Notify when the certificate soc-ai serves with enters the 30, 14 or 7 day
band before expiry, or expires. Inert unless ``notify_enabled`` is on. The
bell in the app carries it either way."""
notify_on_oracle_failure: bool = True
"""Put one row in the bell when the Oracle route pauses: it answered with a
usage limit, or with three server errors in a row
(:mod:`soc_ai.oracle.breaker`). The webhook gets the same message once when
``notify_enabled`` is on. The doctor and the preflight report the pause
whatever this setting is."""
# --- crawl4ai (deep page read) ------------------------------------
crawl4ai_enabled: bool = False
"""Enable the ``crawl_page`` investigator tool (crawl4ai). When False the
tool returns a disabled error without any network I/O. Editable in the
config console. Complements web_search: search finds pages, crawl_page reads
them."""
crawl4ai_url: str = ""
"""Base URL of the self-hosted crawl4ai service (e.g. ``https://crawl.example.com``).
Empty disables page reads. Editable in the config console."""
crawl4ai_token: SecretStr | None = None
"""Optional Bearer token for the crawl4ai service. Stored Fernet-encrypted,
write-only, editable live in the config console's API keys panel (never
rendered back once set)."""
crawl4ai_verify_ssl: bool = True
"""Verify the crawl4ai TLS cert. Set False for a self-signed homelab cert."""
crawl4ai_timeout_s: int = 30
"""Per-request timeout for a page crawl (crawling can be slower than search)."""
crawl_max_chars: int = 6000
"""Cap on extracted page content (chars) returned to the agent per crawl."""
# --- Detection authoring (Sigma bridge) -----------------------------
sigma_authoring_enabled: bool = False
"""Enable the detection-engineering bridge: draft a Sigma rule from a
confirmed hunt finding, validate it (schema + would-have-fired dry run),
and EXPORT it for the analyst to paste into Security Onion. Export-only —
never writes to SO. Console-editable, hot."""
analytic_drafting_enabled: bool = True
"""Enable "draft an analytic from this finding" on a threat hunt finding.
The draft is stored in the local tier as a candidate; a candidate never
runs until an analyst moves it to shadow. Console-editable, hot."""
# --- Runbook retrieval (RAG) — opt-in gateway tier -----------------
# The DEFAULT runbook retrieval is SQLite FTS5 BM25 (migration 0017): zero
# new dependencies, zero egress, always on. These two knobs add an OPTIONAL
# semantic tier via the SAME operator-configured gateway the analyst model
# uses (litellm_base_url) — both default EMPTY = tier off, no gateway call
# is ever made for retrieval. Editable live in the config console.
rag_embed_model: str = ""
"""OpenAI-compatible ``/v1/embeddings`` model id on the configured gateway
used to embed runbooks + search queries for the semantic retrieval tier
(e.g. a LiteLLM embeddings alias). EMPTY (default) disables the tier:
runbook retrieval stays pure-local FTS5/keyword and issues no gateway call.
When set, runbook writes embed fail-soft (a down gateway just leaves the
row un-embedded until the next write or ``POST /config/rag/reembed``) and
``lookup_runbook`` unions semantic hits into its ranking. Changing the model
marks existing vectors stale — run "Re-embed runbooks" after switching."""
rag_rerank_model: str = ""
"""Cohere-shape ``/rerank`` model id on the configured gateway used to
rerank the merged keyword+semantic candidates (e.g. a LiteLLM rerank
alias). EMPTY (default) disables reranking — the weighted merge order
stands. Only consulted when a search actually has candidates; a rerank
failure is fail-soft (the merged order is returned, never an error)."""
# --- Investigation memory (prior outcomes) -------------------------
memory_enabled: bool = False
"""Deterministic prior-outcome context ("investigation memory") for the
synth-first round-1 prompt.
When on, the round-1 synthesis user message gains a small, clearly-framed
block of the most relevant PRIOR verdicts for similar alerts — matched by
deterministic SQL feature tiers over the local ``investigations`` table
(exact rule+src+dest triple, then same rule + one shared endpoint, then
same rule only; newest first within a tier). NO embeddings, NO new tables,
NO extra model calls — one indexed query per investigation (see
:func:`soc_ai.store.investigations.prior_outcomes`).
The block is presented to the model as CONTEXT ONLY, never evidence: its
header instructs the model not to cite it, and the citation gate
independently guarantees a citation quoting prior-outcome text never
resolves (priors are prompt context, not part of the evidence bundle).
Pipeline-fallback verdicts are excluded, so memory reflects model/analyst
conclusions rather than failure noise.
OFF by default: surfacing prior verdicts risks ANCHORING BIAS (the model
repeating an old wrong verdict instead of weighing the fresh evidence) —
keep it off until an anchoring-bias A/B validates default-on. Editable
live in the config console."""
memory_window_days: int = 90
"""How far back (days) the prior-outcome lookup searches for similar
COMPLETED investigations. Bounded [1, 365]. 90 balances institutional
memory against replaying stale conclusions from a since-changed network.
Only consulted when ``memory_enabled`` is on."""
memory_max_items: int = 3
"""Maximum prior-outcome digests injected into the round-1 prompt. Bounded
[1, 5] — the block must stay small (context budget + anchoring surface),
and the deterministic tiers already put the most-similar verdicts first.
Only consulted when ``memory_enabled`` is on."""
memory_include_chat: bool = True
"""Include past chat-transcript excerpts in the investigation-memory block.
When on (the default), the round-1 memory context ALSO recalls relevant
snippets from past analyst↔AI chat threads (investigation follow-up chats
+ hunt chats), retrieved by FTS5 BM25 over the ``chat_memory`` projection
(migration 0018) using the alert's rule-name words and endpoint IPs as
query terms. Past chats carry real institutional knowledge ("we know that
host, it's the vuln scanner") that never lands in a stored verdict.
HARD RULE — context, NEVER evidence: the user in a transcript is not
always right, so the block frames USER lines as unverified operator
opinion (labeled per-line) and ASSISTANT lines as statements about
different alerts; the citation gate independently refuses to resolve
citations against any of it. Nothing from a transcript can ground a
verdict.
**Only takes effect when ``memory_enabled`` is on** — this is a
sub-switch of investigation memory, not an independent feature: with
``memory_enabled`` off (the shipped default) no memory of any kind is
injected regardless of this value. It defaults True (vs memory's
default-off) so that enabling memory brings the full context in one
flip; turn this off to keep prior VERDICTS while excluding chatter.
Shares ``memory_window_days`` / ``memory_max_items`` (snippets are
capped at 5 like prior outcomes). Editable live in the config console."""
@field_validator("memory_window_days", mode="before")
@classmethod
def _clamp_memory_window_days(cls, v: Any) -> Any:
"""Validate memory_window_days is an integer in [1, 365]."""
try:
i = int(v)
except (TypeError, ValueError) as exc:
raise ValueError("memory_window_days must be an integer in [1, 365]") from exc
if i < 1 or i > 365:
raise ValueError(f"memory_window_days must be in [1, 365]. The value is {i}.")
return i
@field_validator("memory_max_items", mode="before")
@classmethod
def _clamp_memory_max_items(cls, v: Any) -> Any:
"""Validate memory_max_items is an integer in [1, 5]."""
try:
i = int(v)
except (TypeError, ValueError) as exc:
raise ValueError("memory_max_items must be an integer in [1, 5]") from exc
if i < 1 or i > 5:
raise ValueError(f"memory_max_items must be in [1, 5]. The value is {i}.")
return i
# --- Oracle frontier adjudication ---------------------------------
oracle_enabled: bool = False
"""Explicit cloud opt-in. When False, _should_escalate_to_oracle() is
always False and no case ever reaches a frontier model."""
oracle_model: str = "claude-opus-5"
"""LiteLLM model alias for the frontier adjudicator. Must be reachable
via litellm_base_url (typically an oracle alias on the gateway)."""
oracle_timeout_s: float = 120.0
"""Per-call HTTP timeout for the Oracle adjudication request (seconds)."""
oracle_rule_mode: Literal["classic", "shadow", "uncertainty"] = "shadow"
"""Which rule sends a local verdict to the Oracle.
``classic`` is the verdict-class rule that ran before stage 1: a
needs_more_info verdict, a verdict other than true_positive on a malware or
attack rule, or a confidence below 0.6
(``soc_ai.agent.orchestrator.classic_oracle_escalation_reason``).
``uncertainty`` is the stage 1 rule
(``soc_ai.agent.orchestrator.oracle_escalation_reason``). ``shadow`` lets
the classic rule decide and records on an ``oracle_shadow`` event what the
uncertainty rule would have done, so a week of production shows the
disagreement before the switch. The default is shadow. A replay of 30 days
of production (2026-10-04) found that the uncertainty rule sends 38 cases a
month the Oracle has never seen."""
oracle_escalate_needs_more_info: bool = True
"""Allow the Oracle to review a needs_more_info verdict.
Since stage 1 the Oracle escalates on uncertainty: a confidence in the gate
band, a split between the decision template and the model, or a deep run
that ended needs_more_info (``soc_ai.agent.orchestrator.
oracle_escalation_reason``). This opt-in narrows that rule: off, no
needs_more_info verdict reaches the Oracle. It never adds an escalation."""
oracle_escalate_malware_non_tp: bool = True
"""Allow the Oracle to review a verdict other than true_positive on a
malware, exploit or attack-class rule. Off, such a verdict stays local. It
narrows the uncertainty rule and never adds an escalation. A true positive
on such a rule always stays local."""
oracle_skip_after_confident_loop: float = 0.8
"""Keep a malware or attack-class verdict local when the investigation loop
ran and reached at least this confidence. It narrows the uncertainty rule.
It matters for a split between the decision template and the model, the one
trigger that can fire above the gate band."""
oracle_escalate_below_confidence: float = 0.7
"""The highest confidence at which an in-band verdict goes to the Oracle.
The gate band is ``[0.4, 0.7)`` (``soc_ai.agent.gates.gate_band``). The
default 0.7 is the top of the band, so the whole band escalates. A lower
value narrows it. A template split and a deep needs_more_info verdict
escalate whatever this value."""
# --- Oracle read-only tool loop (opt-in inside oracle_enabled) ------
oracle_tools_enabled: bool = False
"""Give the Oracle its own bounded, read-only investigation loop.
An opt-in INSIDE the ``oracle_enabled`` opt-in: with this False (the
default) the Oracle stays single-shot — it reasons over the pre-assembled,
sanitized case payload and nothing else, exactly as it shipped. With it
True, an adjudicator that doubts a claim can check it: between sanitize and
verdict the Oracle runs the read-only ``oracle`` tool surface
(:func:`soc_ai.agent.toolset.register_read_tools`, no write / online /
pcap tool), its tool arguments desanitized before execution and its tool
results re-sanitized on the way back, every outbound request body swept for
residue at a single wire-level choke point. Off by default so shipped
behaviour is unchanged until the owner flips it (see the 2026-08-27 design
``docs/dev/design/2026-08-27-oracle-tool-use.md``)."""
oracle_tool_calls_limit: int = 8
"""Max tool calls in ONE Oracle adjudication loop. The Oracle's own budget,
never the investigator's (``agent_tool_calls_limit``): escalation fires
precisely when the local loop is exhausted, so sharing its budget would hand
the Oracle a spent one in exactly the cases it exists for."""
oracle_request_limit: int = 6
"""Max model requests in ONE Oracle adjudication loop (its own budget)."""
oracle_adjudication_timeout_s: float = 300.0
"""Overall wall-clock for one tool-running Oracle adjudication (the loop, not
a single call — ``oracle_timeout_s`` still bounds the per-request HTTP wait)."""
# --- Oracle privacy gate -------------------------------------------
oracle_internal_suffixes: Annotated[tuple[str, ...], NoDecode] = (
".lan",
".local",
".internal",
".corp",
)
"""DNS suffixes that identify internal hostnames.
Any FQDN ending in one of these suffixes is redacted to an opaque
``HOST_NN`` label before the payload is sent to the frontier model.
Public domains pass through untouched. Extend via a comma-separated
env var: ``ORACLE_INTERNAL_SUFFIXES=.lan,.local,.myco.internal``
(pydantic-settings' :class:`NoDecode` + the ``_parse_suffixes`` before-validator
handle the parsing).
"""
oracle_extra_hosts: Annotated[list[str], NoDecode] = []
"""Bare internal hostnames (without a suffix) to redact before the Oracle.
The redacter already catches a lot automatically: private IPs/MACs, FQDNs on
an internal suffix, NetBIOS-shaped computer names (``DESKTOP-AB12``,
``FINANCE-PC``), every identifier learned from a structured host/user field
(propagated into free text), and usernames in an explicit credential context
(``user=jdoe``, ``DOMAIN\\jdoe``).
What it CANNOT know without being told: an internal FQDN on a public-looking
suffix (``dc01.ad.acme.com``) or an arbitrary bare codename (``WIN11-01``,
``APPSERVER01``) — these are shape-indistinguishable from public threat infra
the Oracle must see, so they egress verbatim unless enumerated here (bare
names) or via ``oracle_internal_suffixes`` (internal domains). Comma-separated:
``ORACLE_EXTRA_HOSTS=WIN11-01,APPSERVER01,dbserver``.
When ``oracle_enabled`` is true and neither this nor a custom
``oracle_internal_suffixes`` is set, the Oracle client logs a one-time warning
(see ``soc_ai.oracle.client._warn_if_privacy_gate_unconfigured``). Negligible
on the synthetic lab grid (all addresses are IPs); relevant on real home/work
grids where hostnames appear in alert fields and Zeek pivot records.
"""
@field_validator("oracle_internal_suffixes", mode="before")
@classmethod
def _parse_suffixes(cls, v: Any) -> Any:
# Canonicalise each entry to lowercase with a leading dot, the same form
# the DB identifier path produces. Every consumer assumes that form: the
# sanitizer's suffix-FQDN regex embeds the suffix verbatim after a label
# that cannot end in ".", and its email/domain rules lowercase the value
# but not the suffix, so "acme.example" or ".ACME.EXAMPLE" would match
# nothing and internal names would egress on the paths that read this
# tuple directly (CLI, eval, DB-less fallbacks).
if isinstance(v, str):
v = v.strip()
if not v:
return ()
out: list[str] = []
for raw in v.split(","):
lowered = raw.strip().lower().lstrip(".")
if lowered and "." + lowered not in out:
out.append("." + lowered)
return tuple(out)
return v
@field_validator("oracle_extra_hosts", mode="before")
@classmethod
def _parse_extra_hosts(cls, v: Any) -> Any:
if isinstance(v, str):
v = v.strip()
if not v:
return []
return [h.strip() for h in v.split(",") if h.strip()]
return v
# --- Auto-acknowledge high-confidence false positives ---------------
auto_ack_fp_enabled: bool = False
"""When True, alerts the system is confident are false positives are
automatically acknowledged in Security Onion. Two paths:
- an investigation that finalises with verdict=false_positive at
confidence >= auto_ack_fp_threshold auto-acks its alert;
- an auto-triage sweep that SKIPS a cluster because it inherits a
qualifying FP verdict (``auto_triage_inheritance_enabled``) acks the
cluster's events too — before this, an inherited verdict never reached
SO and those alerts lingered unacked forever.
OFF by default: an auto-ack writes state into Security Onion with no
analyst in the loop, so it stays opt-in — an operator must set this True
to accept that (mirroring ``auto_triage_schedule_enabled``'s default-off
posture). Even then both paths are gated by the confidence threshold AND
the high-stakes guard (``_is_high_stakes_alert`` — a critical/high-severity
or malware/exploit-class alert is never auto-acked, whatever the verdict),
and every unattended write is audited.
The direct path additionally requires that the investigation RETRIEVED
something: a successful tool call, a Phase-D targeted dispatch, or a tool
call in the Oracle's own loop. Confidence is a statement about the model's
certainty, not about whether anybody checked, and a decision template
supplies confidence without supplying evidence. A false positive nothing was
looked up for keeps its verdict and waits for a person; the run records
``auto_ack_skipped`` with reason ``no_investigation`` so the console can say
why. The inheritance path does not apply this test, because the inherited
row does not record whether the original run retrieved anything.
Note the severity interaction: the high-stakes guard never auto-acks a
critical/high-severity (or malware/exploit-class) alert, while
``auto_triage_min_severity`` defaults to "high". If you want auto-ack to
actually clear a backlog, lower the auto-triage floor to "medium"/"low" so
the sweep investigates the low-severity FPs auto-ack is allowed to write.
"""
auto_ack_fp_threshold: float = 0.7
"""Minimum confidence required for automatic FP acknowledgement.
Must be in [0.0, 1.0]. Recommended: 0.7 or higher. At 0.7, the model
is confident in the FP verdict; at 0.9+ it is very confident.
"""
@field_validator("auto_ack_fp_threshold", mode="before")
@classmethod
def _clamp_auto_ack_threshold(cls, v: Any) -> Any:
"""Clamp auto_ack_fp_threshold to [0.0, 1.0]."""
try:
f = float(v)
except (TypeError, ValueError) as exc:
raise ValueError("auto_ack_fp_threshold must be a number in [0.0, 1.0]") from exc
if f < 0.0 or f > 1.0:
raise ValueError(f"auto_ack_fp_threshold must be in [0.0, 1.0]. The value is {f}.")
return f
auto_triage_min_severity: str = "high"
"""Minimum severity included in a ⚡ auto-triage sweep — this level and
above. One of: critical, high, medium, low.
Example: "high" triages critical + high; "medium" adds medium too.
A sweep additionally covers every alert whose document carries no
``event.severity_label``, at every floor including "critical". A floor is a
comparison and an absent label has nothing to compare against, so the choice
is between sweeping those alerts and dropping them silently. "unknown" is
therefore not a legal value here: it is not a floor, it rides along with
every one. See :func:`soc_ai.webui.autotriage.config_severity_band`.
"""
@field_validator("auto_triage_min_severity", mode="before")
@classmethod
def _validate_auto_triage_min_severity(cls, v: Any) -> Any:
"""Lowercase and validate auto_triage_min_severity."""
if not isinstance(v, str):
raise ValueError(
f"auto_triage_min_severity must be a string. The value has type {type(v).__name__}."
)
lowered = v.strip().lower()
if lowered not in {"critical", "high", "medium", "low"}:
raise ValueError(
"auto_triage_min_severity must be one of: critical, high, medium, low. "
f"The value is {v!r}."
)
return lowered
# --- Eval harness (cloud oracle via LiteLLM) -----------------------
# The oracle path runs through LiteLLM (which forwards to an OAuth
# proxy that holds the cloud credential). This reuses
# `litellm_base_url` / `litellm_api_key` / `litellm_verify_ssl` —
# only the model alias and max_tokens are eval-specific.
claude_oracle_model: str = "claude-opus-5"
claude_oracle_max_tokens: int = 8192
# --- Nightly quality micro-eval (trend + alarm) ---------------------
# `soc-ai eval-nightly` runs a tiny batch of REAL investigations on a
# schedule and trends the results locally (quality_snapshots table), so
# "the verdicts are honest" is measured continuously — not once. The two
# knobs below are the whole tuning surface; both are admin-editable live
# in the config console (Quality section).
quality_nightly_n: int = 5
"""How many alerts each ``soc-ai eval-nightly`` run investigates. Small by
design — the nightly is a smoke-trend, not a benchmark: 5 real
investigations is enough to catch an engine swap that broke verdict
quality, while staying cheap enough to run unattended every night. The
CLI clamps the effective value to [1, 10] (the config console enforces
the same bounds), so a stray env value can't turn the nightly into an
hour-long batch."""
quality_alarm_drop: float = 0.15
"""Absolute agreement-rate drop (0-1) below the trailing 7-point same-mode
median that trips the ``quality_regression`` alarm. 0.15 default: on the
n=5 nightly a single flipped grade moves agreement by 0.2, so anything
tighter alarms on one-alert noise while anything looser sleeps through a
real regression until the median itself decays. Bounded [0.05, 0.5] in
the config console; the CLI clamps the same way."""
eval_nightly_enabled: bool = False
"""Master switch for the IN-APP nightly eval scheduler: run the quality
micro-eval once a day at ``eval_nightly_hour_utc`` without any host cron
(schedulable from the UI — Config → Quality). Off by default (recurring
LLM cost is an explicit opt-in, like scheduled hunts); host cron remains
a supported alternative — both paths share the same single-flight slot
logic through :func:`soc_ai.eval.nightly.run_eval_nightly`."""
eval_nightly_hour_utc: int = 3
"""UTC hour (0-23) the in-app scheduler runs the nightly eval at. The run
fires the first wake at/after this hour, at most once per UTC day."""
# ---- validators ---------------------------------------------------
@field_validator(
"es_hosts", "internal_cidrs", "blocklist_sources", "proxy_trusted_ips", mode="before"
)
@classmethod
def _split_csv(cls, v: Any) -> Any:
"""Accept both JSON-style and comma-separated env values for list fields."""
if isinstance(v, str):
v = v.strip()
if not v:
return []
if v.startswith("["):
return v # let pydantic parse as JSON
return [item.strip() for item in v.split(",") if item.strip()]
return v
@field_validator(
"so_ca_bundle",
"so_client_id",
"so_client_secret",
"es_username",
"es_password",
"litellm_api_key",
"misp_url",
"misp_api_key",
"misp_ca_bundle",
"es_ca_bundle",
"bootstrap_admin_password",
"config_secret_key",
"so_ssh_key",
# A blank SO_SSH_KNOWN_HOSTS must mean "unset" (derive the default path),
# not Path("") == Path(".") which would point SSH at the CWD.
"so_ssh_known_hosts",
# Same Path("") == Path(".") trap: .env.example ships both bare for the
# plain-HTTP-behind-a-proxy case, and a truthy "." would make `serve` hand
# uvicorn ssl_certfile="." and the CLI pick https:// for healthz/triage.
"soc_ai_tls_cert",
"soc_ai_tls_key",
# Optional provider keys / webhook URL, also shipped bare. SecretStr("")
# is not None, so a blank would read as "configured" on the data-sources
# page and in the doctor while every consumer refuses to use it.
"greynoise_api_key",
"shodan_api_key",
"maxmind_license_key",
"abuse_ch_auth_key",
"notify_webhook_url",
"crawl4ai_token",
mode="before",
)
@classmethod
def _blank_optional_to_none(cls, v: Any) -> Any:
"""Treat an empty/whitespace env value as unset.
``.env.example`` ships optional integrations as bare keys (e.g.
``MISP_URL=``), and a verbatim copy would otherwise fail URL/path
validation or wrongly trip an ``is not None`` feature gate (Connect API
OAuth). For these optional fields an empty string always means "unset".
"""
if isinstance(v, str) and not v.strip():
return None
return v
@model_validator(mode="after")
def _require_ssh_host_when_pcap(self) -> Settings:
"""Fail fast if PCAP retrieval is enabled without a sensor SSH host.
``so_ssh_host`` defaults to empty (we ship no environment-specific IP);
enabling PCAP without pointing it at a sensor would otherwise surface as
confusing per-tool SSH errors at investigation time.
"""
if self.pcap_enabled and not self.so_ssh_host.strip():
raise ValueError(
"PCAP_ENABLED is true and SO_SSH_HOST is empty. Set SO_SSH_HOST to "
"the Security Onion sensor host, or set PCAP_ENABLED=false."
)
return self
# ---- derived ------------------------------------------------------
@property
def use_connect_api(self) -> bool:
"""True iff Connect API OAuth client credentials are configured."""
return self.so_client_id is not None and self.so_client_secret is not None
def network_is_internal(self, ip: str) -> bool:
"""True iff ``ip`` belongs to one of the configured internal CIDRs."""
try:
addr = ip_address(ip)
except ValueError:
return False
for net in self.internal_cidrs:
if isinstance(addr, IPv4Address) and isinstance(net, IPv4Network) and addr in net:
return True
if isinstance(addr, IPv6Address) and isinstance(net, IPv6Network) and addr in net:
return True
return False
@lru_cache(maxsize=1)
def get_settings() -> Settings:
"""Cached singleton accessor for FastAPI dependency injection."""
return Settings() # type: ignore[call-arg]