6c1123201e
A quoted, space-bearing #include path failed the include regex and fell through to the comment branch, leaving a second authority file unenumerated. Detect any include directive before the comment rule and fail closed on every form except the exact bare includedir into the audited /etc/sudoers.d. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
268 lines
11 KiB
Python
268 lines
11 KiB
Python
#!/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)
|