diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index fe60bef9..fc39cc66 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -108,6 +108,7 @@ configMapGenerator: - node_account_hardening.py=scripts/node_account_hardening.py - node_account_audit.py=scripts/node_account_audit.py - node_account_io.py=scripts/node_account_io.py + - node_polkit_audit.py=scripts/node_polkit_audit.py options: disableNameSuffixHash: true - name: hermes-auto-router-plugin diff --git a/services/hermes/scripts/node_account_audit.py b/services/hermes/scripts/node_account_audit.py index b1795e99..31b54302 100644 --- a/services/hermes/scripts/node_account_audit.py +++ b/services/hermes/scripts/node_account_audit.py @@ -9,6 +9,7 @@ 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]: @@ -108,6 +109,87 @@ def _mentions_dedicated_identity(text: str, account: str, account_uid: int) -> b 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, @@ -160,10 +242,7 @@ def audit_privilege_policies( active = _active_sudo_policy(text) if _mentions_dedicated_identity(active, account, account_uid): raise HardeningError("dedicated Hermes account has sudo authority") - for line in active.splitlines(): - fields = line.split() - if fields and fields[0] in {"ALL", "%ALL"} and "=" in line: - raise HardeningError("broad sudo authority includes Hermes") + _audit_sudo_grants(active) continue active = "\n".join( line for line in text.splitlines() if not line.lstrip().startswith("#") @@ -176,3 +255,4 @@ def audit_privilege_policies( 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) diff --git a/services/hermes/scripts/node_polkit_audit.py b/services/hermes/scripts/node_polkit_audit.py new file mode 100644 index 00000000..281f1484 --- /dev/null +++ b/services/hermes/scripts/node_polkit_audit.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""Literal-identity proof for polkit policies on hardened Atlas nodes. + +Polkit rules are JavaScript, so a grant can name its subject indirectly: +concatenated strings, escape sequences, bracket lookups, or arithmetic can +compute an identity that plain text matching never sees. Rather than model +the language, every grant must scope itself through exact literal identity +comparisons. Anything the audit cannot prove literal fails closed. +""" + +from __future__ import annotations + +import re +from pathlib import Path + +from node_account_io import HardeningError + +_ALLOWED_ESCAPES = {'"', "'", "\\", "/", "n", "r", "t"} +_STRING_MARK = r"\x00[0-9]+\x00" +_GRANT_RE = re.compile(r"\bResult\s*\.\s*YES\b") +_PKLA_GRANT_RE = re.compile(r"(?im)^\s*Result(?:Any|Inactive|Active)\s*=\s*yes\s*$") +_PKLA_IDENTITY_RE = re.compile(r"unix-(?:user|group):[A-Za-z0-9][A-Za-z0-9._-]{0,99}\Z") +_IDENTITY_USE_RE = re.compile(r"subject\s*(\[|\.\s*(?:user|uid|isInGroup)\b)") +_COMPUTED_TOKEN_RE = re.compile( + r"\b(?:eval|Function|arguments|fromCharCode|fromCodePoint|atob|btoa|unescape" + r"|escape|decodeURI[A-Za-z]*|encodeURI[A-Za-z]*|charCodeAt|codePointAt" + r"|normalize|constructor|prototype|globalThis|Reflect|Proxy)\b" +) +# Logical && and || stay legal; every value-building operator next to a +# string or number literal is rejected so comparisons stay whole literals. +_OPERATOR_ADJACENT_RE = re.compile( + rf"(?:{_STRING_MARK}|[0-9])\s*[+\-*/%<>^~](?![&|=])" + rf"|(?])[+\-*/%<>^~]\s*(?:{_STRING_MARK}|[0-9])" +) +_USER_BEFORE_RE = re.compile(rf"{_STRING_MARK}\s*[=!]==?\s*\Z") +_USER_AFTER_RE = re.compile(rf"\s*[=!]==?\s*{_STRING_MARK}") +_UID_BEFORE_RE = re.compile(r"(? tuple[str, list[str]]: + """Replace string literals with markers and drop JavaScript comments.""" + code: list[str] = [] + strings: list[str] = [] + index = 0 + length = len(text) + while index < length: + character = text[index] + if character == "`": + raise HardeningError("polkit rule uses computed template strings") + if character in {'"', "'"}: + quote = character + index += 1 + value: list[str] = [] + while True: + if index >= length or text[index] in "\r\n": + raise HardeningError("polkit rule string is unterminated") + character = text[index] + if character == "\\": + if index + 1 >= length or text[index + 1] not in _ALLOWED_ESCAPES: + raise HardeningError("polkit rule string uses computed escapes") + escaped = text[index + 1] + value.append({"n": "\n", "r": "\r", "t": "\t"}.get(escaped, escaped)) + index += 2 + continue + index += 1 + if character == quote: + break + value.append(character) + strings.append("".join(value)) + code.append(f"\x00{len(strings) - 1}\x00") + continue + if text.startswith("//", index): + newline = text.find("\n", index) + index = length if newline < 0 else newline + continue + if text.startswith("/*", index): + closing = text.find("*/", index + 2) + if closing < 0: + raise HardeningError("polkit rule comment is unterminated") + code.append(" ") + index = closing + 2 + continue + code.append(character) + index += 1 + return "".join(code), strings + + +def _require_literal_identity_use(code: str, match: re.Match[str]) -> None: + """Reject one identity reference unless it is an exact literal comparison.""" + form = match.group(1) + if form.startswith("["): + raise HardeningError("polkit rule reads identity through a computed lookup") + member = form.removeprefix(".").strip() + before = code[max(match.start() - 48, 0) : match.start()] + after = code[match.end() : match.end() + 48] + if member == "user": + literal = _USER_BEFORE_RE.search(before) or _USER_AFTER_RE.match(after) + elif member == "uid": + literal = _UID_BEFORE_RE.search(before) or _UID_AFTER_RE.match(after) + else: + literal = _GROUP_AFTER_RE.match(after) + if not literal: + raise HardeningError("polkit rule identity test is not an exact literal") + + +def _audit_rules_javascript(text: str) -> None: + """Require granting rules to gate on exact literal subject identities. + + With computed tokens, bracket lookups, and subject aliasing rejected in + every rule file, a YES decision can only be written as the literal grant + token, so the grant test below cannot be evaded by construction. + """ + code, _strings = _split_strings(text) + if _COMPUTED_TOKEN_RE.search(code): + raise HardeningError("polkit rule computes values the audit cannot prove") + if _RESULT_USE_RE.search(code): + raise HardeningError("polkit rule reads results through a computed lookup") + for bare in _BARE_SUBJECT_RE.finditer(code): + if not _PARAMETER_BEFORE_RE.search(code[: bare.start()]): + raise HardeningError("polkit rule passes the subject to computed code") + if not _GRANT_RE.search(code): + return + if _OPERATOR_ADJACENT_RE.search(code): + raise HardeningError("polkit grant builds values with operators") + uses = 0 + for match in _IDENTITY_USE_RE.finditer(code): + uses += 1 + _require_literal_identity_use(code, match) + if not uses: + raise HardeningError("polkit grant is not scoped to an explicit identity") + + +def _audit_pkla(text: str) -> None: + """Require granting localauthority entries to name exact literal identities.""" + if not _PKLA_GRANT_RE.search(text): + return + identities: list[str] = [] + for line in text.splitlines(): + name, separator, value = line.strip().partition("=") + if name.strip() != "Identity" or not separator: + continue + entries = [entry.strip() for entry in value.split(";") if entry.strip()] + identities.extend(entries) + if not identities: + raise HardeningError("polkit authority grant lacks an identity scope") + for entry in identities: + if not _PKLA_IDENTITY_RE.fullmatch(entry): + raise HardeningError("polkit authority identity is not an exact literal") + + +def audit_polkit_policy(path: Path, text: str) -> None: + """Prove one polkit policy file cannot grant through a computed identity.""" + if "localauthority" in path.parts or path.suffix == ".pkla": + _audit_pkla(text) + else: + _audit_rules_javascript(text) diff --git a/testing/tests/test_hermes_node_polkit_identity.py b/testing/tests/test_hermes_node_polkit_identity.py new file mode 100644 index 00000000..059f05f2 --- /dev/null +++ b/testing/tests/test_hermes_node_polkit_identity.py @@ -0,0 +1,211 @@ +"""Literal-identity denial proofs for polkit grants near the Hermes account.""" + +from __future__ import annotations + +import importlib.util +import os +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).parents[2] +SCRIPTS = ROOT / "services/hermes/scripts" +sys.path.insert(0, str(SCRIPTS)) + + +def _load(): + spec = importlib.util.spec_from_file_location( + "node_polkit_audit_test", SCRIPTS / "node_polkit_audit.py" + ) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def _rules(module, text: str): + module.audit_polkit_policy(Path("rules.d/50-test.rules"), text) + + +def test_literal_distribution_rule_passes(): + module = _load() + _rules( + module, + "// allow the network daemon\n" + "polkit.addRule(function(action, subject) {\n" + ' if (action.id == "org.freedesktop.network1.set-dns" &&\n' + ' subject.user == "systemd-network") {\n' + " return polkit.Result.YES;\n" + " }\n" + "});\n", + ) + _rules( + module, + "polkit.addRule(function(action, subject) {\n" + ' if (subject.isInGroup("wheel") && subject.uid === 4242) {\n' + " return polkit.Result.YES;\n" + " }\n" + "});\n", + ) + _rules( + module, + "/* deny-only rules may use constructs the audit cannot prove */\n" + "polkit.addRule(function(action, subject) {\n" + " if (subject.user.indexOf(\"x\") == 0) { return polkit.Result.NO; }\n" + "});\n", + ) + + +def test_reversed_literal_comparisons_pass(): + module = _load() + _rules( + module, + "polkit.addRule(function(action, subject) {\n" + ' if ("atlas" == subject.user || 4242 === subject.uid) {\n' + " return polkit.Result.YES;\n" + " }\n" + "});\n", + ) + + +def test_unconditional_grant_fails_closed(): + module = _load() + with pytest.raises(module.HardeningError, match="explicit identity"): + _rules(module, "polkit.addRule(function() { return polkit.Result.YES; });\n") + + +@pytest.mark.parametrize( + ("snippet", "match"), + [ + ('subject.user == "herme" + "s-agent"', "operators"), + ("subject.uid == 1199 + 1", "operators"), + ("subject.uid == 0x4B0", "exact literal"), + ("subject.uid >= 1000", "exact literal"), + ("subject.uid == -1", "operators"), + ("subject.user == name", "exact literal"), + ("subject.user.indexOf(\"h\") == 0", "exact literal"), + ("subject.isInGroup(group)", "exact literal"), + ('subject["user"] == "atlas"', "computed lookup"), + ], +) +def test_computed_identity_grants_fail_closed(snippet: str, match: str): + module = _load() + text = ( + "polkit.addRule(function(action, subject) {\n" + f" if ({snippet}) {{ return polkit.Result.YES; }}\n" + "});\n" + ) + with pytest.raises(module.HardeningError, match=match): + _rules(module, text) + + +@pytest.mark.parametrize( + ("text", "match"), + [ + ("var x = `hermes`;\n", "template strings"), + ('var x = "unterminated\n', "unterminated"), + ('var x = "bad\\x2descape";\n', "computed escapes"), + ('var x = "dangling\\', "computed escapes"), + ("/* never closed\n", "comment is unterminated"), + ("var f = eval;\n", "cannot prove"), + ('var s = String.fromCharCode(104);\n', "cannot prove"), + ("var a = arguments;\n", "cannot prove"), + ('var r = polkit.Result["YES"];\n', "computed lookup"), + ("var r = polkit.Result;\n", "computed lookup"), + ("var s = subject;\n", "computed code"), + ("check(subject);\n", "computed code"), + ], +) +def test_unprovable_rule_constructs_fail_closed(text: str, match: str): + module = _load() + with pytest.raises(module.HardeningError, match=match): + _rules(module, text) + + +def test_string_and_comment_handling_keeps_literal_rules_auditable(): + module = _load() + code, strings = module._split_strings( + 'var a = "with // not a comment"; // trailing\n' + "/* block */ var b = 'quo\\'te\\n';\n" + ) + assert strings == ["with // not a comment", "quo'te\n"] + assert "//" not in code and "block" not in code + assert "\x000\x00" in code and "\x001\x00" in code + + +def test_grant_detection_survives_spacing_and_strings(): + module = _load() + with pytest.raises(module.HardeningError, match="explicit identity"): + _rules(module, "polkit.addRule(function() { return polkit.Result . YES; });\n") + # A grant token inside a string is inert and needs no identity scope. + _rules(module, 'var label = "polkit.Result.YES";\n') + + +@pytest.mark.parametrize( + ("text", "match"), + [ + ( + "[All]\nIdentity=unix-user:*\nAction=org.example\nResultAny=yes\n", + "exact literal", + ), + ( + "[N]\nIdentity=unix-netgroup:ops\nAction=org.example\nResultActive=yes\n", + "exact literal", + ), + ("[N]\nAction=org.example\nResultInactive=yes\n", "identity scope"), + ("[N]\nIdentity=\nAction=org.example\nResultAny=yes\n", "identity scope"), + ], +) +def test_pkla_grants_require_exact_literal_identities(text: str, match: str): + module = _load() + with pytest.raises(module.HardeningError, match=match): + module.audit_polkit_policy(Path("localauthority/50-local.d/a.pkla"), text) + + +def test_pkla_literal_grants_and_deny_entries_pass(): + module = _load() + module.audit_polkit_policy( + Path("localauthority/50-local.d/a.pkla"), + "[Mount]\nIdentity=unix-user:atlas;unix-group:operators\n" + "Action=org.freedesktop.udisks2.filesystem-mount\nResultAny=yes\n", + ) + module.audit_polkit_policy( + Path("etc/b.pkla"), + "[Deny]\nIdentity=unix-user:*\nAction=org.example\nResultAny=no\n", + ) + + +def test_integration_rejects_previously_invisible_grants(tmp_path: Path): + spec = importlib.util.spec_from_file_location( + "node_account_audit_polkit_test", SCRIPTS / "node_account_audit.py" + ) + assert spec and spec.loader + audit = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = audit + spec.loader.exec_module(audit) + + host_etc = tmp_path / "etc" + rules = host_etc / "polkit-1/rules.d" + rules.mkdir(parents=True) + (rules / "90-open.rules").write_text( + "polkit.addRule(function() { return polkit.Result.YES; });\n", + encoding="utf-8", + ) + with pytest.raises(audit.HardeningError, match="explicit identity"): + audit.audit_privilege_policies( + "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "missing" + ) + + (rules / "90-open.rules").unlink() + local = host_etc / "polkit-1/localauthority/50-local.d" + local.mkdir(parents=True) + (local / "wide.pkla").write_text( + "[W]\nIdentity=unix-user:*\nAction=org.example.benign\nResultAny=yes\n", + encoding="utf-8", + ) + with pytest.raises(audit.HardeningError, match="exact literal"): + audit.audit_privilege_policies( + "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "missing" + ) \ No newline at end of file diff --git a/testing/tests/test_hermes_node_sudo_alias_audit.py b/testing/tests/test_hermes_node_sudo_alias_audit.py new file mode 100644 index 00000000..e62b82bf --- /dev/null +++ b/testing/tests/test_hermes_node_sudo_alias_audit.py @@ -0,0 +1,134 @@ +"""Sudo alias and computed-identity denial proofs for the Hermes account.""" + +from __future__ import annotations + +import importlib.util +import os +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).parents[2] +SCRIPTS = ROOT / "services/hermes/scripts" +sys.path.insert(0, str(SCRIPTS)) + + +def _load(): + spec = importlib.util.spec_from_file_location( + "node_account_audit_sudo_test", SCRIPTS / "node_account_audit.py" + ) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def _audit(tmp_path: Path, sudoers: str): + module = _load() + host_etc = tmp_path / "etc" + (host_etc / "sudoers.d").mkdir(parents=True, exist_ok=True) + (host_etc / "sudoers.d/atlas").write_text(sudoers, encoding="utf-8") + module.audit_privilege_policies( + "hermes-agent", 1200, os.getuid(), host_etc, tmp_path / "missing-share" + ) + + +def test_benign_distribution_policy_passes(tmp_path: Path): + _audit( + tmp_path, + "Defaults secure_path=/usr/bin\n" + "root ALL=(ALL:ALL) ALL\n" + "%sudo ALL=(ALL:ALL) ALL\n" + "#999 ALL=(root) /usr/bin/uptime\n" + "alice, bob ALL=(root) NOPASSWD: /usr/bin/systemctl status\n", + ) + + +def test_alias_reaching_hermes_directly_is_denied(tmp_path: Path): + module = _load() + with pytest.raises(module.HardeningError, match="sudo authority"): + _audit( + tmp_path, + "User_Alias OPERATORS = atlas, hermes-agent\n" + "OPERATORS ALL=(ALL) ALL\n", + ) + + +def test_alias_reaching_hermes_by_numeric_uid_is_denied(tmp_path: Path): + module = _load() + with pytest.raises(module.HardeningError, match="sudo authority"): + _audit(tmp_path, "User_Alias OPERATORS = #1200\nOPERATORS ALL=(ALL) ALL\n") + + +def test_alias_expanding_to_all_is_denied(tmp_path: Path): + module = _load() + with pytest.raises(module.HardeningError, match="broad sudo authority"): + _audit(tmp_path, "User_Alias ADMINS = ALL\nADMINS ALL=(ALL) ALL\n") + + +def test_nested_alias_expanding_to_all_via_continuation_is_denied(tmp_path: Path): + module = _load() + with pytest.raises(module.HardeningError, match="broad sudo authority"): + _audit( + tmp_path, + "User_Alias INNER = atlas, \\\n ALL\n" + "User_Alias OUTER = INNER\n" + "OUTER ALL=(ALL) ALL\n", + ) + + +@pytest.mark.parametrize( + ("sudoers", "match"), + [ + ("+operators ALL=(ALL) ALL\n", "netgroup"), + ("%:S-1-5-32 ALL=(ALL) ALL\n", "netgroup"), + ("GHOSTS ALL=(ALL) ALL\n", "alias is undefined"), + ("User_Alias A = B\nUser_Alias B = A\nA ALL=(ALL) ALL\n", "cyclic"), + ("User_Alias broken\nroot ALL=(ALL) ALL\n", "not auditable"), + ("User_Alias lower = atlas\n", "not auditable"), + ("User_Alias A = atlas\nUser_Alias A = bob\nA ALL=(ALL) ALL\n", "not auditable"), + ("User_Alias A = atlas,,bob\nA ALL=(ALL) ALL\n", "not auditable"), + ("stray-line-without-equals\n", "not auditable"), + ("=orphan (ALL) ALL\n", "not auditable"), + ("! ALL=(ALL) ALL\n", "malformed"), + ("User_Alias A = atlas : B = ALL\nB ALL=(ALL) ALL\n", "broad sudo"), + ], +) +def test_unauditable_or_broad_sudo_policies_fail_closed( + tmp_path: Path, sudoers: str, match: str +): + module = _load() + with pytest.raises(module.HardeningError, match=match): + _audit(tmp_path, sudoers) + + +def test_negated_and_quoted_principals_are_expanded_before_judging(tmp_path: Path): + module = _load() + with pytest.raises(module.HardeningError, match="broad sudo authority"): + _audit(tmp_path, '!"ALL" ALL=(ALL) ALL\n') + _audit(tmp_path, "!alice ALL=(ALL) ALL\n") + + +def test_continuation_join_requires_a_complete_final_line(): + module = _load() + with pytest.raises(module.HardeningError, match="line continuation"): + module._joined_sudo_lines("root ALL=(ALL) ALL \\") + assert module._joined_sudo_lines("a \\\nb\nc") == ["a b", "c"] + + +def test_alias_definitions_parse_multiple_groups_per_line(): + module = _load() + aliases = module._sudo_user_aliases( + ["User_Alias A = atlas, bob : B = carol", "Runas_Alias R = root"] + ) + assert aliases == {"A": ["atlas", "bob"], "B": ["carol"]} + + +def test_expansion_accepts_nested_literal_users(): + module = _load() + aliases = {"A": ["atlas", "B"], "B": ["carol"]} + module._expand_sudo_principal("A", aliases, frozenset()) + module._expand_sudo_principal("%wheel", aliases, frozenset()) + module._expand_sudo_principal("#999", aliases, frozenset()) \ No newline at end of file