# 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