atlas-iac/services/hermes/scripts/node_account_audit.py

#!/usr/bin/env python3
"""Fail-closed supplementary-group and host privilege-policy audits."""

from __future__ import annotations

import re
import stat
import struct
from pathlib import Path

from node_account_io import HardeningError, read_regular
from node_polkit_audit import audit_polkit_policy


def _members(field: str, context: str) -> set[str]:
    if not field:
        return set()
    values = field.split(",")
    if any(not value or value.strip() != value for value in values):
        raise HardeningError(f"{context} has a malformed member list")
    if len(values) != len(set(values)):
        raise HardeningError(f"{context} has duplicate members")
    return set(values)


def audit_membership(
    account: str, groups: list[list[str]], gshadow: list[list[str]]
) -> None:
    """Reject the dedicated account in any supplementary group/admin list."""
    for record in groups:
        if account in _members(record[3], f"group {record[0]}"):
            raise HardeningError("dedicated Hermes account has supplementary group access")
    for record in gshadow:
        principals = _members(record[2], f"gshadow {record[0]} admins")
        principals |= _members(record[3], f"gshadow {record[0]} members")
        if account in principals:
            raise HardeningError("dedicated Hermes account has gshadow group access")


def _policy_files(host_etc: Path, polkit_share: Path, root_uid: int) -> list[Path]:
    files: list[Path] = []
    sudoers = host_etc / "sudoers"
    if sudoers.exists():
        files.append(sudoers)
    roots = (
        host_etc / "sudoers.d",
        host_etc / "polkit-1/rules.d",
        host_etc / "polkit-1/localauthority",
        polkit_share / "rules.d",
        polkit_share / "localauthority",
    )
    for root in roots:
        if not root.exists():
            continue
        metadata = root.lstat()
        if (
            not stat.S_ISDIR(metadata.st_mode)
            or stat.S_ISLNK(metadata.st_mode)
            or metadata.st_uid != root_uid
            or stat.S_IMODE(metadata.st_mode) & 0o022
        ):
            raise HardeningError("unsafe sudo/polkit policy directory")
        files.extend(path for path in root.rglob("*") if path.is_file())
    if len(files) > 256:
        raise HardeningError("too many sudo/polkit policy files")
    return sorted(files)


def _audit_policy_metadata(snapshot, root_uid: int, account_uid: int) -> None:
    if snapshot.uid != root_uid or snapshot.mode & 0o022:
        raise HardeningError("sudo/polkit policy metadata permits unsafe mutation")
    acl = dict(snapshot.xattrs).get("system.posix_acl_access", b"")
    if not acl:
        return
    header = struct.Struct("<I")
    entry = struct.Struct("<HHI")
    if len(acl) < header.size or (len(acl) - header.size) % entry.size:
        raise HardeningError("sudo/polkit policy ACL is malformed")
    for offset in range(header.size, len(acl), entry.size):
        tag, permissions, identifier = entry.unpack_from(acl, offset)
        if tag == 0x02 and identifier == account_uid and permissions & 0o2:
            raise HardeningError("sudo/polkit policy ACL grants Hermes write access")


def _active_sudo_policy(text: str) -> str:
    """Remove comments without discarding sudoers' numeric-UID principals."""
    active: list[str] = []
    for line in text.splitlines():
        stripped = line.lstrip()
        if not stripped:
            continue
        # Detect any include spelling before the comment rule below.  sudo
        # honours both #include and @include; a quoted or space-bearing path
        # must never fall through the "#" comment branch and leave a second
        # authority file unaudited.  Only the exact bare includedir into the
        # audited /etc/sudoers.d passes; everything else fails closed.
        if re.match(r"(?i)[#@]include(?:dir)?(?=\s|$)", stripped):
            include = re.fullmatch(
                r"(?i)([#@]include(?:dir)?)\s+([^\s\"'\\]+)", stripped
            )
            if include:
                directive, target = include.groups()
                if (
                    directive.lower().endswith("includedir")
                    and target == "/etc/sudoers.d"
                ):
                    continue
            raise HardeningError("sudo policy includes an unaudited authority source")
        if stripped.startswith("#") and not re.match(r"#\d+(?:\s|$)", stripped):
            continue
        active.append(line)
    return "\n".join(active)


def _mentions_dedicated_identity(text: str, account: str, account_uid: int) -> bool:
    account_pattern = rf"(?<![A-Za-z0-9_.-])%?{re.escape(account)}(?![A-Za-z0-9_.-])"
    numeric_pattern = rf"(?<![0-9])#?{account_uid}(?![0-9])"
    return bool(re.search(account_pattern, text) or re.search(numeric_pattern, text))


SUDO_ALIAS_NAME_RE = re.compile(r"[A-Z][A-Z0-9_]*\Z")


def _joined_sudo_lines(active: str) -> list[str]:
    """Merge sudoers backslash continuations into whole logical lines."""
    lines: list[str] = []
    pending = ""
    for line in active.splitlines():
        if line.endswith("\\"):
            pending += line[:-1] + " "
            continue
        lines.append(pending + line)
        pending = ""
    if pending:
        raise HardeningError("sudo policy ends inside a line continuation")
    return lines


def _sudo_user_aliases(lines: list[str]) -> dict[str, list[str]]:
    """Collect User_Alias definitions so grant principals can be expanded."""
    aliases: dict[str, list[str]] = {}
    for line in lines:
        stripped = line.strip()
        if not stripped.startswith("User_Alias"):
            continue
        for definition in stripped.removeprefix("User_Alias").split(":"):
            name, separator, members = definition.partition("=")
            name = name.strip()
            values = [item.strip() for item in members.split(",")]
            if (
                not separator
                or not SUDO_ALIAS_NAME_RE.fullmatch(name)
                or name in aliases
                or not all(values)
            ):
                raise HardeningError("sudo alias definition is not auditable")
            aliases[name] = values
    return aliases


def _expand_sudo_principal(
    principal: str, aliases: dict[str, list[str]], seen: frozenset[str]
) -> None:
    """Reject grant principals that reach Hermes through aliases or wildcards."""
    value = principal.strip()
    while value.startswith("!"):
        value = value[1:].strip()
    value = value.strip('"')
    if not value:
        raise HardeningError("sudo grant principal is malformed")
    if value in aliases:
        if value in seen:
            raise HardeningError("sudo alias expansion is cyclic")
        for member in aliases[value]:
            _expand_sudo_principal(member, aliases, seen | {value})
        return
    if value in {"ALL", "%ALL"}:
        raise HardeningError("broad sudo authority includes Hermes")
    if value.startswith("+") or value.startswith("%:"):
        raise HardeningError("sudo netgroup authority cannot be audited")
    if SUDO_ALIAS_NAME_RE.fullmatch(value):
        raise HardeningError("sudo alias is undefined")


def _audit_sudo_grants(active: str) -> None:
    """Expand every grant's user list; only literal non-Hermes principals pass."""
    lines = _joined_sudo_lines(active)
    aliases = _sudo_user_aliases(lines)
    skip = ("Defaults", "User_Alias", "Runas_Alias", "Host_Alias", "Cmnd_Alias")
    for line in lines:
        stripped = line.strip()
        if not stripped or stripped.startswith(skip):
            continue
        head, separator, _rest = stripped.partition("=")
        fields = head.split()
        if not separator or len(fields) < 2:
            raise HardeningError("sudo grant line is not auditable")
        for principal in " ".join(fields[:-1]).split(","):
            _expand_sudo_principal(principal, aliases, frozenset())


def audit_privilege_policies(
    account: str,
    account_uid: int,
    root_uid: int,
    host_etc: Path,
    polkit_share: Path,
) -> None:
    """Reject direct, group, wildcard sudo, or root-equivalent polkit grants."""
    dangerous_polkit = (
        "org.freedesktop.policykit.exec",
        "org.freedesktop.systemd1.manage-unit",
        "org.freedesktop.udisks2.modify-device",
        "org.freedesktop.packagekit",
    )
    nsswitch = host_etc / "nsswitch.conf"
    if nsswitch.exists():
        try:
            nss_snapshot = read_regular(nsswitch, 64 * 1024)
            _audit_policy_metadata(nss_snapshot, root_uid, account_uid)
            text = nss_snapshot.value.decode("utf-8")
        except UnicodeDecodeError as exc:
            raise HardeningError("nsswitch policy is not UTF-8") from exc
        for line in text.splitlines():
            name, separator, sources = line.partition(":")
            if name.strip() in {
                "passwd",
                "group",
                "initgroups",
                "shadow",
                "sudoers",
            } and separator:
                active_sources = [item for item in sources.split() if not item.startswith("[")]
                if not active_sources or any(
                    item not in {"files", "systemd"} for item in active_sources
                ):
                    raise HardeningError("external group/account/sudo authority source is unsafe")
    total = 0
    for path in _policy_files(host_etc, polkit_share, root_uid):
        snapshot = read_regular(path, 256 * 1024)
        _audit_policy_metadata(snapshot, root_uid, account_uid)
        value = snapshot.value
        total += len(value)
        if total > 2 * 1024 * 1024:
            raise HardeningError("sudo/polkit policy input exceeds safe limit")
        try:
            text = value.decode("utf-8")
        except UnicodeDecodeError as exc:
            raise HardeningError("sudo/polkit policy is not UTF-8") from exc
        if path.name == "sudoers" or "sudoers.d" in path.parts:
            active = _active_sudo_policy(text)
            if _mentions_dedicated_identity(active, account, account_uid):
                raise HardeningError("dedicated Hermes account has sudo authority")
            _audit_sudo_grants(active)
            continue
        active = "\n".join(
            line for line in text.splitlines() if not line.lstrip().startswith("#")
        )
        grants = "polkit.Result.YES" in active or re.search(
            r"(?im)^\s*Result(?:Any|Inactive|Active)\s*=\s*yes\s*$", active
        )
        if grants and _mentions_dedicated_identity(active, account, account_uid):
            raise HardeningError("dedicated Hermes account has polkit authority")
        broad = "unix-user:*" in active or "Identity=unix-user:*" in active
        if broad and grants and any(item in active for item in dangerous_polkit):
            raise HardeningError("root-equivalent broad polkit authority includes Hermes")
        audit_polkit_policy(path, text)