enodia-sentinal/enodia_sentinel/fim.py

# SPDX-License-Identifier: GPL-3.0-or-later
"""File integrity monitoring — the Tripwire/OSSEC-syscheck capability.

Two engines:

* **hash baseline** — SHA-256 (plus mode/uid/gid/size) of a curated set of
  security-critical files that the package manager does *not* track
  (``/usr/local``, ``/etc`` configs, systemd units, SSH keys). We own this
  baseline and refresh it on demand (a pacman PostTransaction hook calls
  ``fim-update`` after every upgrade, so legitimate package changes never
  alert).

* **package verification** — ``pacman -Qkk`` checks package-owned binaries
  against the distro's own signed checksums. No baseline to maintain; it is
  implicitly current because the package DB updates on every upgrade.

Content hashing is the upgrade over mtime checks: it catches a file swapped out
for a malicious copy even when the timestamp is preserved.
"""
from __future__ import annotations

import hashlib
import os
import stat
import subprocess
from collections.abc import Iterable, Iterator

from .alert import Alert, Severity

# SIDs 100017+ are FIM signatures.
SID_MODIFIED = 100017
SID_ADDED = 100018
SID_REMOVED = 100019
SID_PKG = 100020

DEFAULT_FIM_PATHS = (
    "/usr/local/bin", "/usr/local/sbin", "/usr/local/lib",
    "/etc/systemd/system", "/etc/ld.so.preload", "/etc/ld.so.conf",
    "/etc/ld.so.conf.d", "/etc/pam.d", "/etc/ssh/sshd_config",
    "/etc/sudoers", "/etc/sudoers.d", "/etc/profile", "/etc/profile.d",
    "/etc/bash.bashrc", "/etc/crontab", "/etc/cron.d", "/etc/cron.daily",
    "/etc/cron.hourly", "/etc/passwd", "/etc/shadow", "/etc/group",
    "/root/.ssh/authorized_keys", "/root/.bashrc", "/root/.profile",
)


# --- hashing & scanning ---------------------------------------------------

def hash_file(path: str, _bufsize: int = 1 << 16) -> str | None:
    h = hashlib.sha256()
    try:
        with open(path, "rb") as fh:
            while chunk := fh.read(_bufsize):
                h.update(chunk)
    except OSError:
        return None
    return h.hexdigest()


def _entry(path: str) -> dict | None:
    try:
        st = os.lstat(path)
    except OSError:
        return None
    e = {"mode": stat.S_IMODE(st.st_mode), "uid": st.st_uid,
         "gid": st.st_gid, "size": st.st_size}
    if stat.S_ISLNK(st.st_mode):
        try:
            e["link"] = os.readlink(path)
        except OSError:
            e["link"] = ""
    elif stat.S_ISREG(st.st_mode):
        e["sha256"] = hash_file(path)
    else:
        return None  # skip sockets/devices/fifos
    return e


def scan_paths(paths: Iterable[str]) -> dict[str, dict]:
    """Map path -> integrity entry for every regular file / symlink under paths."""
    out: dict[str, dict] = {}
    for root in paths:
        if os.path.isfile(root) or os.path.islink(root):
            e = _entry(root)
            if e:
                out[root] = e
        elif os.path.isdir(root):
            for dirpath, _dirs, files in os.walk(root, onerror=lambda e: None):
                for f in files:
                    p = os.path.join(dirpath, f)
                    e = _entry(p)
                    if e:
                        out[p] = e
    return out


# --- diffing --------------------------------------------------------------

def diff(old: dict[str, dict], new: dict[str, dict]) -> dict[str, list]:
    """Return {'added':[...], 'removed':[...], 'modified':[(path, changes)]}."""
    added = sorted(set(new) - set(old))
    removed = sorted(set(old) - set(new))
    modified = []
    for path in sorted(set(old) & set(new)):
        o, n = old[path], new[path]
        changes = [k for k in ("sha256", "mode", "uid", "gid", "link")
                   if o.get(k) != n.get(k)]
        if changes:
            modified.append((path, changes))
    return {"added": added, "removed": removed, "modified": modified}


def diff_alerts(d: dict[str, list]) -> Iterator[Alert]:
    for path, changes in d["modified"]:
        crit = "sha256" in changes or "mode" in changes
        yield Alert(
            severity=Severity.HIGH if crit else Severity.MEDIUM,
            signature="fim_modified",
            key=f"fim:mod:{path}",
            detail=f"integrity change ({', '.join(changes)}): {path}",
            sid=SID_MODIFIED, classtype="integrity-violation",
        )
    for path in d["added"]:
        yield Alert(Severity.MEDIUM, "fim_added", f"fim:add:{path}",
                    f"new file in monitored path: {path}",
                    sid=SID_ADDED, classtype="integrity-violation")
    for path in d["removed"]:
        yield Alert(Severity.MEDIUM, "fim_removed", f"fim:del:{path}",
                    f"monitored file removed: {path}",
                    sid=SID_REMOVED, classtype="integrity-violation")


# --- package verification (engine A) --------------------------------------

def parse_pacman_verify(output: str) -> list[tuple[str, str, str]]:
    """Parse ``pacman -Qkk`` diagnostics into (package, path, reason).

    Only lines reporting an actual content/permission/ownership change are
    returned; 'total files' summaries and 'No package owns' lines are ignored.
    """
    out: list[tuple[str, str, str]] = []
    for line in output.splitlines():
        line = line.strip()
        if "(" not in line or ")" not in line:
            continue
        reason = line[line.rfind("(") + 1:line.rfind(")")]
        rl = reason.lower()
        # Content/permission/ownership changes matter; a modification-time-only
        # mismatch is benign noise (a `touch` doesn't alter the file).
        if not any(k in rl for k in ("checksum", "sha", "size",
                                     "permission", "ownership", "uid", "gid")):
            continue
        head = line[:line.rfind("(")].strip()
        # head looks like "pkgname: /path/to/file"
        if ": /" not in head:
            continue
        pkg, path = head.split(": /", 1)
        pkg = pkg.replace("warning:", "").strip()
        out.append((pkg, "/" + path.strip(), reason.strip()))
    return out


def pacman_verify(timeout: int = 600) -> list[tuple[str, str, str]]:
    """Run ``pacman -Qkk`` and return altered package-owned files (best-effort)."""
    try:
        r = subprocess.run(["pacman", "-Qkk"], capture_output=True,
                           text=True, timeout=timeout)
    except (OSError, subprocess.SubprocessError):
        return []
    return parse_pacman_verify(r.stdout + "\n" + r.stderr)


def pacman_verify_alerts(timeout: int = 600) -> Iterator[Alert]:
    for pkg, path, reason in pacman_verify(timeout):
        yield Alert(
            severity=Severity.CRITICAL,
            signature="fim_pkg_modified",
            key=f"fim:pkg:{path}",
            detail=f"package file altered ({pkg}: {reason}): {path}",
            sid=SID_PKG, classtype="integrity-violation",
        )