atlas-iac/testing/tests/test_hermes_node_polkit_identity.py

"""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_regex_literal_tokenizer_desync_grant_is_rejected():
    """A regex literal embedding a quote must not open a phantom string.

    Without regex awareness the two /'/ literals fool the string scanner into
    swallowing the unconditional Result.YES between their quotes; the grant
    then reads as unscoped and would slip past every later check.
    """
    module = _load()
    exploit = (
        "polkit.addRule(function(action, subject) { "
        "var a = /'/; "
        'if (action.id == "org.freedesktop.policykit.exec") '
        "{ return polkit.Result.YES; } "
        "var b = /'/; });\n"
    )
    with pytest.raises(module.HardeningError, match="regex or division"):
        _rules(module, exploit)


@pytest.mark.parametrize(
    "snippet",
    [
        "var a = /'/;",
        'var a = /"/;',
        "var a = /abc/;",
        "var a = /[/'\"]/;",
        "var a = /x\\/y/;",
        "var a = subject.user / 2;",
        "// a comment / with a slash then\nvar a = /'/;",
        '/* block */ var a = /"/;',
        "return polkit.Result.YES / 1;",
    ],
)
def test_every_slash_form_outside_a_comment_fails_closed(snippet: str):
    module = _load()
    text = f"polkit.addRule(function(action, subject) {{ {snippet} }});\n"
    with pytest.raises(module.HardeningError, match="regex or division"):
        _rules(module, text)


def test_slashes_confined_to_strings_and_comments_stay_legal():
    module = _load()
    # Slashes inside string literals and // or /* */ comments are inert.
    _rules(
        module,
        "// path-like /usr/bin comment\n"
        "/* another /etc/ comment */\n"
        "polkit.addRule(function(action, subject) {\n"
        '    if (action.id == "org.freedesktop.systemd1.manage-units" &&\n'
        '        subject.user == "systemd-network") {\n'
        "        return polkit.Result.YES;\n"
        "    }\n"
        "});\n",
    )


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"
        )