ping-mee-self-service-platform/portal/tests/test_ldap_directory.py

"""The LDAPS adapter against a stand-in `ldap` module: what it sends and how it maps errors."""

from datetime import UTC, datetime, timedelta
from types import SimpleNamespace

import pytest

from portal.adapters.directory import (
    DirectoryUnavailable,
    Forbidden,
    LdapDirectory,
    PasswordRejected,
    ValidationFailed,
)
from portal.tests import ldap_stub

OU = "OU=Users,DC=example,DC=test"
DOMAIN = "DC=example,DC=test"
USER_DN = "CN=Alice A,OU=Users,DC=example,DC=test"
SVC = "svc@example.test"
EPOCH = datetime(1601, 1, 1, tzinfo=UTC)


def filetime(moment):
    return (moment - EPOCH) // timedelta(microseconds=1) * 10


def b(**attrs):
    return {k: [v.encode()] for k, v in attrs.items()}


def config(tmp_path, server="dc.example.test"):
    ca = tmp_path / "ca.pem"
    ca.write_text("ca")
    return SimpleNamespace(
        server=server,
        user_ou_dn=OU,
        bind_dn=SVC,
        bind_password="service-password",
        ca_file=ca,
        publisher_group_dn="CN=Publishers,OU=Users,DC=example,DC=test",
        admin_group_dn="CN=Admins,OU=Users,DC=example,DC=test",
    )


@pytest.fixture
def world():
    w = ldap_stub.World()
    w.passwords[USER_DN] = "Alice-Pass-1!"
    w.entries[(OU, 2)] = [
        (
            USER_DN,
            {
                **b(
                    sAMAccountName="Alice",
                    givenName="Alice",
                    sn="A",
                    mail="a@example.test",
                    userAccountControl="512",
                    pwdLastSet=str(filetime(datetime(2026, 1, 1, tzinfo=UTC))),
                ),
                "memberOf": [
                    b"cn=publishers, OU=Users,DC=example,DC=test",
                    b"CN=Other,DC=example,DC=test",
                ],
            },
        )
    ]
    w.entries[("", 0)] = [("", b(defaultNamingContext=DOMAIN))]
    w.entries[(DOMAIN, 0)] = [
        (
            DOMAIN,
            b(
                minPwdLength="12",
                pwdProperties="1",
                pwdHistoryLength="24",
                minPwdAge=str(-864000000000),
                maxPwdAge=str(-77760000000000),
                lockoutDuration=str(-18000000000),
                lockoutThreshold="5",
                lockoutObservationWindow=str(-18000000000),
            ),
        )
    ]
    return w


@pytest.fixture
def directory(world, tmp_path):
    return LdapDirectory(config(tmp_path), ldap_module=ldap_stub.build(world))


def user_attrs(world):
    return world.entries[(OU, 2)][0][1]


def test_connection_is_ldaps_with_certificate_verification(directory, world, tmp_path):
    directory.find_user("alice")
    conn = world.connections[0]
    assert conn.url == "ldaps://dc.example.test:636"
    assert conn.options["require_cert"] == "demand"
    assert conn.options["cacertfile"] == str(tmp_path / "ca.pem")
    assert conn.options["newctx"] == 0


def test_an_explicit_port_is_kept(world, tmp_path):
    d = LdapDirectory(config(tmp_path, "dc.example.test:3269"), ldap_module=ldap_stub.build(world))
    d.find_user("alice")
    assert world.connections[0].url == "ldaps://dc.example.test:3269"


def test_find_user_reads_state_and_roles(directory):
    user = directory.find_user("alice")
    assert (user.sam, user.given_name, user.surname, user.mail) == (
        "Alice",
        "Alice",
        "A",
        "a@example.test",
    )
    assert user.enabled and not user.locked
    assert user.is_publisher and not user.is_admin
    assert user.password_last_set == datetime(2026, 1, 1, tzinfo=UTC)


def test_the_lookup_is_confined_to_the_ou_and_asks_only_for_listed_attributes(directory, world):
    directory.find_user("alice")
    base, scope, _flt, attrs = world.searches[0]
    assert (base, scope) == (OU, 2)
    assert "unicodePwd" not in attrs and "description" not in attrs


@pytest.mark.parametrize(
    ("evil", "escaped"),
    [
        ("*", r"\2a"),
        ("a)(objectClass=*", r"a\29\28objectClass=\2a"),
        ("a*)(|(sAMAccountName=*", r"a\2a\29\28|\28sAMAccountName=\2a"),
        ("back\\slash", r"back\5cslash"),
    ],
)
def test_filter_input_is_escaped(directory, world, evil, escaped):
    directory.find_user(evil)
    flt = world.searches[0][2]
    assert flt == f"(&(objectClass=user)(objectCategory=person)(sAMAccountName={escaped}))"


def test_a_disabled_account_and_a_lock_are_reported(directory, world):
    user_attrs(world)["userAccountControl"] = [b"514"]
    user_attrs(world)["msDS-User-Account-Control-Computed"] = [b"16"]
    user = directory.find_user("alice")
    assert not user.enabled and user.locked


def test_lockout_falls_back_to_the_time_when_the_computed_attribute_is_missing(directory, world):
    user_attrs(world)["lockoutTime"] = [str(filetime(datetime.now(UTC))).encode()]
    assert directory.find_user("alice").locked is True  # the lockout duration is 30 minutes
    user_attrs(world)["lockoutTime"] = [str(filetime(datetime(2026, 1, 1, tzinfo=UTC))).encode()]
    assert directory.find_user("alice").locked is False


def test_pwd_last_set_zero_means_the_user_must_change_it(directory, world):
    user_attrs(world)["pwdLastSet"] = [b"0"]
    assert directory.find_user("alice").must_change_password is True


def test_an_unknown_or_ambiguous_user_is_not_found(directory, world):
    world.entries[(OU, 2)] = world.entries[(OU, 2)] * 2
    assert directory.find_user("alice") is None
    world.entries[(OU, 2)] = []
    assert directory.find_user("alice") is None


def test_a_refused_service_account_is_an_outage(world, tmp_path):
    world.passwords[SVC] = "changed"
    d = LdapDirectory(config(tmp_path), ldap_module=ldap_stub.build(world))
    with pytest.raises(DirectoryUnavailable):
        d.find_user("alice")


def test_a_server_that_is_down_is_an_outage(directory, world):
    world.down = True
    with pytest.raises(DirectoryUnavailable):
        directory.find_user("alice")


def test_verify_password_binds_as_the_found_user(directory, world):
    assert directory.verify_password("alice", "Alice-Pass-1!") is True
    assert world.connections[-1].bound == USER_DN
    assert directory.verify_password("alice", "nope") is False
    assert directory.verify_password("nobody", "x") is False


def test_an_empty_password_is_refused_without_contacting_the_server(directory, world):
    assert directory.verify_password("alice", "") is False
    assert world.connections == []


def test_password_policy_converts_the_directory_units(directory):
    policy = directory.password_policy("alice")
    assert policy.min_length == 12
    assert policy.complexity is True
    assert policy.history_length == 24
    assert policy.min_age == timedelta(days=1)
    assert policy.max_age == timedelta(days=90)
    assert policy.fine_grained is False
    assert policy.lockout_threshold == 5
    assert policy.lockout_window == timedelta(minutes=30)


def test_a_fine_grained_policy_is_used_when_readable(directory, world):
    pso = "CN=PSO,CN=Password Settings Container,CN=System,DC=example,DC=test"
    user_attrs(world)["msDS-ResultantPSO"] = [pso.encode()]
    world.entries[(pso, 0)] = [
        (
            pso,
            b(
                **{
                    "msDS-MinimumPasswordLength": "20",
                    "msDS-PasswordComplexityEnabled": "FALSE",
                    "msDS-PasswordHistoryLength": "5",
                    "msDS-MinimumPasswordAge": "0",
                    "msDS-MaximumPasswordAge": str(-(2**63)),
                    "msDS-LockoutThreshold": "3",
                    "msDS-LockoutObservationWindow": str(-6000000000),
                }
            ),
        )
    ]
    policy = directory.password_policy("alice")
    assert (policy.min_length, policy.complexity, policy.history_length) == (20, False, 5)
    assert policy.max_age is None and policy.fine_grained is True
    assert policy.lockout_threshold == 3 and policy.lockout_window == timedelta(minutes=10)


def test_an_unreadable_fine_grained_policy_is_flagged_not_hidden(directory, world, monkeypatch):
    pso = "CN=PSO,CN=System,DC=example,DC=test"
    user_attrs(world)["msDS-ResultantPSO"] = [pso.encode()]
    original = world.respond

    def respond(base, scope, flt):
        if base == pso:
            raise ldap_stub.INSUFFICIENT_ACCESS()
        return original(base, scope, flt)

    monkeypatch.setattr(world, "respond", respond)
    policy = directory.password_policy("alice")
    assert policy.fine_grained is True and policy.fine_grained_readable is False
    assert policy.min_length == 12


def test_update_sends_a_replace_for_allowed_attributes(directory, world):
    directory.update_attributes("alice", {"mobile": "+49 1", "sn": None})
    dn, mods = world.modifications[0]
    assert dn == USER_DN
    assert mods == [(2, "mobile", [b"+49 1"]), (2, "sn", [])]


def test_update_refuses_other_attributes_without_connecting(directory, world):
    with pytest.raises(Forbidden):
        directory.update_attributes("alice", {"description": "x"})
    assert world.connections == [] and world.modifications == []


def test_update_maps_a_server_refusal_and_missing_rights(directory, world):
    world.modify_error = ldap_stub.CONSTRAINT_VIOLATION()
    with pytest.raises(ValidationFailed):
        directory.update_attributes("alice", {"mail": "x"})
    world.modify_error = ldap_stub.INSUFFICIENT_ACCESS()
    with pytest.raises(Forbidden):
        directory.update_attributes("alice", {"mail": "x"})


def test_change_password_sends_delete_old_and_add_new_as_utf16(directory, world):
    directory.change_password("alice", "Old-Pass-1!", "New-Pass-2@")
    dn, mods = world.modifications[0]
    assert dn == USER_DN
    assert mods == [
        (1, "unicodePwd", ['"Old-Pass-1!"'.encode("utf-16-le")]),
        (0, "unicodePwd", ['"New-Pass-2@"'.encode("utf-16-le")]),
    ]


@pytest.mark.parametrize(
    ("info", "reason"),
    [
        ("0000052D: AtrErr: DSID-03191077, problem 1005 (CONSTRAINT_ATT_TYPE)", "policy"),
        ("00000056: AtrErr: DSID-03190F80, problem 1005 (CONSTRAINT_ATT_TYPE)", "wrong_current"),
        ("something unexpected", "policy"),
    ],
)
def test_password_refusals_are_told_apart_by_the_directory_error_code(
    directory, world, info, reason
):
    world.modify_error = ldap_stub.CONSTRAINT_VIOLATION({"info": info})
    with pytest.raises(PasswordRejected) as exc:
        directory.change_password("alice", "Old-Pass-1!", "New-Pass-2@")
    assert exc.value.reason == reason


def test_a_refused_password_change_does_not_carry_the_passwords(directory, world):
    world.modify_error = ldap_stub.CONSTRAINT_VIOLATION({"info": "0000052D: x"})
    with pytest.raises(PasswordRejected) as exc:
        directory.change_password("alice", "Old-Pass-1!", "New-Pass-2@")
    text = repr(exc.value) + str(exc.value)
    assert "Old-Pass-1!" not in text and "New-Pass-2@" not in text