enodia-sentinal/enodia_sentinel/config.py

# SPDX-License-Identifier: GPL-3.0-or-later
"""Runtime configuration.

Defaults live here; overrides come from a TOML file (default
``/etc/enodia-sentinel.toml``, or ``$ENODIA_CONFIG``) and a couple of env vars
useful for testing without root (``$ENODIA_LOG_DIR``).
"""
from __future__ import annotations

import os
import tomllib
from dataclasses import dataclass, field, fields
from pathlib import Path

_DEFAULT_INTERPRETERS = (
    "bash sh dash zsh ksh ash nc ncat netcat socat telnet "
    "python python2 python3 perl ruby php lua awk"
).split()

_DEFAULT_WATCH = (
    "/etc/cron.d", "/etc/crontab", "/etc/cron.daily", "/etc/cron.hourly",
    "/etc/cron.weekly", "/var/spool/cron", "/etc/systemd/system",
    "/etc/ld.so.preload", "/etc/passwd", "/etc/sudoers", "/etc/sudoers.d",
    "/root/.ssh/authorized_keys", "/root/.bashrc", "/root/.profile",
)

_ALL_DETECTORS = (
    "reverse_shell", "ld_preload", "deleted_exe",
    "new_listener", "new_suid", "persistence", "egress",
)


@dataclass
class Config:
    # timing
    sample_interval: float = 4.0
    cooldown: int = 60
    baseline_grace: int = 10
    suid_refresh: int = 3600
    suid_scan_interval: int = 60

    # detector toggles
    detectors: frozenset[str] = frozenset(_ALL_DETECTORS)

    # tuning
    interpreters: frozenset[str] = frozenset(_DEFAULT_INTERPRETERS)
    egress_allow_cidrs: tuple[str, ...] = ()
    listener_allow_ports: frozenset[str] = frozenset()
    # Listeners owned by these process names never alert (e.g. P2P clients).
    listener_allow_comms: frozenset[str] = frozenset()
    # Suppress new_listener when the listening binary is package-owned (strong
    # provenance signal; off by default so the detector stays honest).
    suppress_package_owned_listeners: bool = False
    suid_hot_dirs: tuple[str, ...] = (
        "/tmp", "/dev/shm", "/var/tmp", "/home", "/run/user",
    )
    # Writable dirs always scanned for SUID even if on a separate filesystem
    # (tmpfs /tmp etc.). Kept small so the gated scan stays cheap.
    suid_scan_extra_dirs: tuple[str, ...] = (
        "/tmp", "/dev/shm", "/var/tmp", "/run/user",
    )
    watch_persistence: tuple[str, ...] = _DEFAULT_WATCH

    # file integrity monitoring (FIM)
    fim_enabled: bool = True
    fim_paths: tuple[str, ...] = ()          # populated from fim.DEFAULT_FIM_PATHS
    fim_scan_interval: int = 300             # seconds between hash scans
    fim_pkg_verify: bool = False             # run `pacman -Qkk` (slow; opt-in)
    fim_pkg_verify_interval: int = 21600     # seconds between package verifications
    # tamper-evidence
    pkgdb_verify: bool = True                # detect out-of-band package-DB edits
    pkgdb_interval: int = 600                # seconds between DB integrity checks
    heartbeat_max_age: int = 120             # heartbeat older than this = stale

    # eBPF on-ramp
    capture_execve_bpftrace: bool = False
    # Event-driven eBPF execve monitor (catches short-lived processes the poll
    # loop misses). Degrades gracefully to polling if bcc/root/BTF unavailable.
    ebpf_exec_monitor: bool = True
    exec_rules_file: str = ""   # optional extra Snort-style rules (TOML)

    # retention
    max_snapshots: int = 300
    max_snapshot_age_days: int = 60

    # notifications — desktop
    notify_users: tuple[str, ...] = ()
    notify_urgency: str = "critical"
    # notifications — phone push (a backend is enabled when its required keys
    # are set). Only alerts at/above notify_min_severity are pushed.
    notify_min_severity: str = "HIGH"
    notify_ntfy_url: str = ""        # e.g. "https://ntfy.sh" or self-hosted base
    notify_ntfy_topic: str = ""      # topic name (required to enable ntfy)
    notify_ntfy_token: str = ""      # optional Bearer token for protected topics
    notify_pushover_token: str = ""  # Pushover application token
    notify_pushover_user: str = ""   # Pushover user/group key
    notify_webhook_url: str = ""     # generic JSON POST target
    dashboard_url: str = ""          # optional base URL embedded in pushes

    # web dashboard (read-only)
    web_bind: str = ""               # "" = auto-detect Tailscale IP, else explicit
    web_port: int = 8787
    web_token: str = ""              # bearer token; auto-generated if empty + non-local

    # paths
    log_dir: Path = Path("/var/log/enodia-sentinel")

    # ---- derived paths ---------------------------------------------------
    @property
    def events_log(self) -> Path:
        return self.log_dir / "events.log"

    @property
    def listener_baseline(self) -> Path:
        return self.log_dir / "listener-baseline.json"

    @property
    def suid_baseline(self) -> Path:
        return self.log_dir / "suid-baseline.json"

    @property
    def fim_baseline(self) -> Path:
        return self.log_dir / "fim-baseline.json"

    def fim_path_list(self) -> tuple[str, ...]:
        """FIM paths: the configured/critical set, always plus Sentinel's own
        footprint so tampering with the watchdog trips the watchdog."""
        from .fim import DEFAULT_FIM_PATHS
        from .selfprotect import SELF_PATHS
        base = self.fim_paths or DEFAULT_FIM_PATHS
        return tuple(base) + SELF_PATHS

    # ---- loading ---------------------------------------------------------
    @classmethod
    def load(cls, path: str | os.PathLike | None = None) -> "Config":
        cfg = cls()

        env_log = os.environ.get("ENODIA_LOG_DIR")
        if env_log:
            cfg.log_dir = Path(env_log)

        cfg_path = Path(path or os.environ.get("ENODIA_CONFIG")
                        or "/etc/enodia-sentinel.toml")
        if cfg_path.is_file():
            cfg._apply_toml(cfg_path)
        return cfg

    def _apply_toml(self, path: Path) -> None:
        with open(path, "rb") as fh:
            data = tomllib.load(fh)
        known = {f.name for f in fields(self)}
        for key, value in data.items():
            if key not in known:
                continue
            current = getattr(self, key)
            if isinstance(current, frozenset):
                value = frozenset(value)
            elif isinstance(current, tuple):
                value = tuple(value)
            elif isinstance(current, Path):
                value = Path(value)
            setattr(self, key, value)

    def enabled(self, detector: str) -> bool:
        return detector in self.detectors