enodia-sentinal/enodia_sentinel/posture.py

# SPDX-License-Identifier: GPL-3.0-or-later
"""Host posture checks — config hygiene that *precedes* an incident.

Where the detectors answer "is something attacking right now?", posture answers
"is this host set up to be easy to attack?": root SSH logins, password auth,
passwordless sudo, world-writable ``PATH`` directories, loose permissions on
sensitive files, and a downgraded package-signature policy.

These are advisory findings, not live detections, so posture is a command
(``enodia-sentinel posture check``), not a daemon detector — it never blocks
startup. Findings reuse the ``Alert`` type so they flow through the same JSON
and triage machinery as everything else.

Every check is split into a pure evaluator (operates on already-read content or
``stat`` facts, so tests need no root and no live ``/etc``) plus a thin system
reader that gathers that content.
"""
from __future__ import annotations

import glob
import os
import re
import stat
from collections.abc import Iterable, Iterator
from typing import NamedTuple

from .alert import Alert, Severity
from .config import Config

# --- SID allocation (host-posture: 100040-100047, 100050-100051;
#     100048-100049 belong to the memory_obfuscation detector) -------------
SID_SSH_ROOT_LOGIN = 100040
SID_SSH_PASSWORD_AUTH = 100041
SID_SSH_EMPTY_PASSWORD = 100042
SID_SUDO_NOPASSWD = 100043
SID_SUDO_NOAUTH = 100044
SID_SUDOERS_WRITABLE = 100045
SID_PATH_WRITABLE = 100046
SID_SENSITIVE_FILE_PERM = 100047
SID_SYSTEMD_UNIT_WRITABLE = 100050
SID_SYSTEMD_EXEC_UNSAFE = 100051

CLASSTYPE = "host-posture"

_DEFAULT_PATH_DIRS = (
    "/usr/local/sbin", "/usr/local/bin", "/usr/sbin", "/usr/bin",
    "/sbin", "/bin", "/root/bin",
)

# An enabled service whose ExecStart binary lives under one of these is a
# persistence/hijack red flag — these are user- or world-writable locations a
# system service should never run from.
_UNSAFE_EXEC_PREFIXES = ("/tmp/", "/dev/shm/", "/var/tmp/", "/home/", "/run/user/")

# (path, world-readable severity, group/world-writable severity). None = skip
# that rule for this file. Keyed by the kind of secret each file holds.
_SENSITIVE_FILES: tuple[tuple[str, Severity | None, Severity | None], ...] = (
    ("/etc/shadow", Severity.HIGH, Severity.CRITICAL),
    ("/etc/gshadow", Severity.HIGH, Severity.CRITICAL),
    ("/etc/passwd", None, Severity.CRITICAL),
    ("/etc/group", None, Severity.CRITICAL),
    ("/etc/ssh/sshd_config", None, Severity.HIGH),
)


# --- SSH ------------------------------------------------------------------

def resolve_sshd(lines: Iterable[str]) -> dict[str, str]:
    """Reduce already-Include-expanded sshd_config lines to effective values.

    sshd uses the *first* obtained value for each keyword, so we keep the first
    occurrence and ignore later ones. Keys are lowercased; ``Include`` lines are
    assumed already expanded by the reader and are skipped here.
    """
    effective: dict[str, str] = {}
    for raw in lines:
        line = raw.split("#", 1)[0].strip()
        if not line:
            continue
        parts = line.split(None, 1)
        key = parts[0].lower()
        if key == "include":
            continue
        value = parts[1].strip() if len(parts) > 1 else ""
        effective.setdefault(key, value)
    return effective


def sshd_findings(effective: dict[str, str]) -> Iterator[Alert]:
    """Posture findings from effective sshd directives (OpenSSH defaults aware)."""
    # PermitRootLogin: modern default is prohibit-password, which is fine; only
    # an explicit `yes` (password root login) is a finding.
    root_login = effective.get("permitrootlogin", "prohibit-password").lower()
    if root_login == "yes":
        yield Alert(
            severity=Severity.HIGH, signature="ssh_root_login",
            key="posture:ssh:rootlogin",
            detail="sshd PermitRootLogin=yes — root can log in over SSH with a password",
            sid=SID_SSH_ROOT_LOGIN, classtype=CLASSTYPE)

    # PasswordAuthentication default is yes, so an unset value is still a finding.
    pw_auth = effective.get("passwordauthentication", "yes").lower()
    if pw_auth == "yes":
        how = "enabled" if "passwordauthentication" in effective else "enabled (compiled-in default)"
        yield Alert(
            severity=Severity.MEDIUM, signature="ssh_password_auth",
            key="posture:ssh:passwordauth",
            detail=f"sshd PasswordAuthentication {how} — keys-only login is safer "
                   "against credential brute force",
            sid=SID_SSH_PASSWORD_AUTH, classtype=CLASSTYPE)

    if effective.get("permitemptypasswords", "no").lower() == "yes":
        yield Alert(
            severity=Severity.CRITICAL, signature="ssh_empty_password",
            key="posture:ssh:emptypw",
            detail="sshd PermitEmptyPasswords=yes — accounts with empty passwords "
                   "can log in over SSH",
            sid=SID_SSH_EMPTY_PASSWORD, classtype=CLASSTYPE)


def read_sshd_lines(path: str, _depth: int = 0, _seen: set[str] | None = None) -> list[str]:
    """Read sshd_config, expanding ``Include`` globs in place (sshd order).

    Drop-in files are read in lexical order at the point the Include appears.
    A visited set and depth cap guard against include loops.
    """
    seen = _seen if _seen is not None else set()
    real = os.path.realpath(path)
    if _depth > 16 or real in seen:
        return []
    seen.add(real)
    out: list[str] = []
    try:
        with open(path, encoding="utf-8", errors="replace") as fh:
            raw_lines = fh.readlines()
    except OSError:
        return out
    for raw in raw_lines:
        stripped = raw.split("#", 1)[0].strip()
        if stripped and stripped.split(None, 1)[0].lower() == "include":
            pattern = stripped.split(None, 1)[1].strip() if " " in stripped else ""
            if pattern and not os.path.isabs(pattern):
                pattern = os.path.join("/etc/ssh", pattern)
            for inc in sorted(glob.glob(pattern)):
                out.extend(read_sshd_lines(inc, _depth + 1, seen))
        else:
            out.append(raw)
    return out


# --- sudoers --------------------------------------------------------------

def sudoers_findings(files: Iterable[tuple[str, str, int]]) -> Iterator[Alert]:
    """Findings from sudoers files, each given as (name, text, st_mode).

    Flags passwordless sudo (NOPASSWD), disabled auth (!authenticate), and any
    sudoers file that is group/world-writable (which sudo itself would refuse,
    making it a strong tamper signal).
    """
    for name, text, mode in files:
        if mode & 0o022:
            yield Alert(
                severity=Severity.HIGH, signature="sudoers_writable",
                key=f"posture:sudo:writable:{name}",
                detail=f"sudoers file is group/world-writable ({stat.filemode(mode)}): {name}",
                sid=SID_SUDOERS_WRITABLE, classtype=CLASSTYPE)
        for raw in text.splitlines():
            # In sudoers, '#include'/'#includedir' are directives, not comments;
            # everything else after '#' is a comment.
            token0 = raw.strip().split(None, 1)[0] if raw.strip() else ""
            line = raw if token0 in ("#include", "#includedir") else raw.split("#", 1)[0]
            line = line.strip()
            if not line:
                continue
            if "NOPASSWD" in line:
                yield Alert(
                    severity=Severity.MEDIUM, signature="sudo_nopasswd",
                    key=f"posture:sudo:nopasswd:{name}:{line}",
                    detail=f"passwordless sudo rule in {name}: {line}",
                    sid=SID_SUDO_NOPASSWD, classtype=CLASSTYPE)
            if line.lower().startswith("defaults") and "!authenticate" in line.lower():
                yield Alert(
                    severity=Severity.MEDIUM, signature="sudo_no_authenticate",
                    key=f"posture:sudo:noauth:{name}",
                    detail=f"sudo authentication disabled (!authenticate) in {name}",
                    sid=SID_SUDO_NOAUTH, classtype=CLASSTYPE)


def read_sudoers(main: str, sudoers_dir: str) -> list[tuple[str, str, int]]:
    out: list[tuple[str, str, int]] = []
    paths = [main]
    try:
        # sudo ignores backup/dotfiles in sudoers.d (names with '.' or ending '~').
        for entry in sorted(os.listdir(sudoers_dir)):
            if entry.startswith(".") or entry.endswith("~") or "." in entry:
                continue
            paths.append(os.path.join(sudoers_dir, entry))
    except OSError:
        pass
    for p in paths:
        try:
            st = os.stat(p)
            if not stat.S_ISREG(st.st_mode):
                continue
            with open(p, encoding="utf-8", errors="replace") as fh:
                out.append((p, fh.read(), st.st_mode))
        except OSError:
            continue
    return out


# --- PATH directories -----------------------------------------------------

def path_findings(dirs: Iterable[tuple[str, int, int]]) -> Iterator[Alert]:
    """Findings from PATH directories, each given as (path, st_mode, st_uid).

    A world-writable ``PATH`` dir lets any user drop a binary that root may run;
    a non-root-owned one is a weaker but real hijack vector.
    """
    for path, mode, uid in dirs:
        if mode & 0o002:
            yield Alert(
                severity=Severity.HIGH, signature="path_world_writable",
                key=f"posture:path:writable:{path}",
                detail=f"PATH directory is world-writable ({stat.filemode(mode)}) — "
                       f"any user can plant a binary root may execute: {path}",
                sid=SID_PATH_WRITABLE, classtype=CLASSTYPE)
        elif uid != 0:
            yield Alert(
                severity=Severity.MEDIUM, signature="path_nonroot_owner",
                key=f"posture:path:owner:{path}",
                detail=f"PATH directory not owned by root (uid {uid}): {path}",
                sid=SID_PATH_WRITABLE, classtype=CLASSTYPE)


def read_path_dirs(dirs: Iterable[str]) -> list[tuple[str, int, int]]:
    out: list[tuple[str, int, int]] = []
    seen: set[str] = set()
    for d in dirs:
        if not d or d in seen:
            continue
        seen.add(d)
        try:
            st = os.stat(d)
        except OSError:
            continue
        if stat.S_ISDIR(st.st_mode):
            out.append((d, st.st_mode, st.st_uid))
    return out


# --- sensitive file permissions -------------------------------------------

def sensitive_file_findings(
    facts: Iterable[tuple[str, int]],
    policy: Iterable[tuple[str, Severity | None, Severity | None]] = _SENSITIVE_FILES,
) -> Iterator[Alert]:
    """Findings from sensitive files, each given as (path, st_mode), against a
    policy of (path, world_readable_sev, writable_sev)."""
    rules = {p: (rsev, wsev) for p, rsev, wsev in policy}
    for path, mode in facts:
        rule = rules.get(path)
        if rule is None:
            continue
        rsev, wsev = rule
        if wsev is not None and mode & 0o022:
            yield Alert(
                severity=wsev, signature="sensitive_file_writable",
                key=f"posture:file:writable:{path}",
                detail=f"sensitive file is group/world-writable ({stat.filemode(mode)}): {path}",
                sid=SID_SENSITIVE_FILE_PERM, classtype=CLASSTYPE)
        if rsev is not None and mode & 0o004:
            yield Alert(
                severity=rsev, signature="sensitive_file_readable",
                key=f"posture:file:readable:{path}",
                detail=f"sensitive file is world-readable ({stat.filemode(mode)}): {path}",
                sid=SID_SENSITIVE_FILE_PERM, classtype=CLASSTYPE)


def read_sensitive_files(
    policy: Iterable[tuple[str, Severity | None, Severity | None]] = _SENSITIVE_FILES,
) -> list[tuple[str, int]]:
    out: list[tuple[str, int]] = []
    for path, _r, _w in policy:
        try:
            st = os.stat(path)
        except OSError:
            continue
        if stat.S_ISREG(st.st_mode):
            out.append((path, st.st_mode))
    return out


# --- systemd units --------------------------------------------------------

class UnitFact(NamedTuple):
    name: str
    fragment_path: str
    fragment_mode: int          # st_mode of the unit file (0 if unknown)
    enabled: bool
    exec_paths: tuple[str, ...]  # ExecStart binary paths


def systemd_findings(units: Iterable[UnitFact]) -> Iterator[Alert]:
    """Findings from enabled systemd units.

    Flags a unit file anyone can rewrite (group/world-writable — what root runs
    at boot becomes attacker-controlled) and an enabled service whose ExecStart
    binary lives in a writable/unsafe directory (a classic systemd persistence
    trick).
    """
    for u in units:
        if u.fragment_path and (u.fragment_mode & 0o022):
            yield Alert(
                severity=Severity.HIGH, signature="systemd_unit_writable",
                key=f"posture:systemd:writable:{u.name}",
                detail=f"systemd unit file is group/world-writable "
                       f"({stat.filemode(u.fragment_mode)}): {u.fragment_path} ({u.name})",
                sid=SID_SYSTEMD_UNIT_WRITABLE, classtype=CLASSTYPE)
        for path in u.exec_paths:
            if path.startswith(_UNSAFE_EXEC_PREFIXES):
                yield Alert(
                    severity=Severity.HIGH, signature="systemd_exec_unsafe_path",
                    key=f"posture:systemd:exec:{u.name}:{path}",
                    detail=f"enabled systemd unit {u.name} runs from a writable/unsafe "
                           f"path: {path}",
                    sid=SID_SYSTEMD_EXEC_UNSAFE, classtype=CLASSTYPE)


_EXEC_PATH_RE = re.compile(r"path=(\S+)")


def parse_systemctl_show(text: str) -> list[tuple[str, str, bool, tuple[str, ...]]]:
    """Parse ``systemctl show`` output into (name, fragment_path, enabled, execs).

    Pure: blocks are separated by blank lines; each carries ``Id``,
    ``FragmentPath``, ``UnitFileState``, and zero or more ``ExecStart`` lines
    (whose ``path=`` field is the resolved binary). No filesystem access.
    """
    out: list[tuple[str, str, bool, tuple[str, ...]]] = []
    for block in text.split("\n\n"):
        props: dict[str, str] = {}
        execs: list[str] = []
        for line in block.splitlines():
            key, _, val = line.partition("=")
            if key == "ExecStart":
                m = _EXEC_PATH_RE.search(val)
                if m:
                    execs.append(m.group(1))
            else:
                props.setdefault(key, val)
        name = props.get("Id", "")
        if not name:
            continue
        out.append((name, props.get("FragmentPath", ""),
                    props.get("UnitFileState", "") == "enabled", tuple(execs)))
    return out


def read_systemd_units(systemctl: str = "systemctl") -> list[UnitFact]:
    """Gather enabled systemd service units as UnitFacts (fails closed).

    Non-systemd or restricted hosts (no ``systemctl``) yield no facts.
    """
    import subprocess
    try:
        listing = subprocess.run(
            [systemctl, "list-unit-files", "--type=service",
             "--state=enabled", "--no-legend", "--no-pager"],
            capture_output=True, text=True, timeout=8).stdout
    except (OSError, subprocess.SubprocessError):
        return []
    names = [line.split()[0] for line in listing.splitlines() if line.split()]
    if not names:
        return []
    try:
        shown = subprocess.run(
            [systemctl, "show", "--no-pager", "--property=Id",
             "--property=FragmentPath", "--property=ExecStart",
             "--property=UnitFileState", *names],
            capture_output=True, text=True, timeout=12).stdout
    except (OSError, subprocess.SubprocessError):
        return []
    facts: list[UnitFact] = []
    for name, frag, enabled, execs in parse_systemctl_show(shown):
        mode = 0
        if frag:
            try:
                mode = os.stat(frag).st_mode
            except OSError:
                mode = 0
        facts.append(UnitFact(name, frag, mode, enabled, execs))
    return facts


# --- orchestration --------------------------------------------------------

def run(cfg: Config) -> Iterator[Alert]:
    """Run every posture check and yield findings. Each reader fails closed
    (missing file → no finding), so a non-Arch or minimal host degrades quietly."""
    yield from sshd_findings(resolve_sshd(read_sshd_lines(cfg.posture_sshd_config)))
    yield from sudoers_findings(read_sudoers(cfg.posture_sudoers, cfg.posture_sudoers_dir))

    path_dirs = cfg.posture_path or _DEFAULT_PATH_DIRS
    yield from path_findings(read_path_dirs(path_dirs))
    yield from sensitive_file_findings(read_sensitive_files())

    if cfg.posture_systemd:
        yield from systemd_findings(read_systemd_units())

    # Package-signature policy is a posture question too; reuse the tamper check.
    from . import pkgdb
    sl = pkgdb.siglevel_alert()
    if sl is not None:
        yield sl