PacketStation/tests/server/test_ldap_dc.py

"""Against the CI Samba domain controller (deploy/samba). Skipped unless the TEST_LDAP_* values are set.

TEST_LDAP_URL, TEST_LDAP_BASE_DN, TEST_LDAP_BIND_DN, TEST_LDAP_PASSWORD_FILE, TEST_LDAP_GROUP_BASE_DN,
TEST_LDAP_CA_BUNDLE and TEST_DC_USERS_FILE (the users.json the DC seed writes, with per-run passwords).
"""

from __future__ import annotations

import json
import os
from pathlib import Path
from typing import Any

import pytest
from fastapi.testclient import TestClient
from pshelpers import ORIGIN_HEADERS, app_conn, make_settings, needs_pg

from packetstation.auth import groups
from packetstation.auth.directory import refresh_from_ad
from packetstation.auth.ldap import Directory
from packetstation.settings import Settings

ENV = (
    "TEST_LDAP_URL",
    "TEST_LDAP_BASE_DN",
    "TEST_LDAP_BIND_DN",
    "TEST_LDAP_PASSWORD_FILE",
    "TEST_LDAP_GROUP_BASE_DN",
    "TEST_LDAP_CA_BUNDLE",
    "TEST_DC_USERS_FILE",
)
MISSING = [k for k in ENV if not os.environ.get(k)]
pytestmark = pytest.mark.skipif(bool(MISSING), reason=f"no test domain controller ({', '.join(MISSING)} not set)")


def ldap_values() -> dict[str, Any]:
    return {
        "ldap_url": os.environ["TEST_LDAP_URL"],
        "ldap_base_dn": os.environ["TEST_LDAP_BASE_DN"],
        "ldap_bind_dn": os.environ["TEST_LDAP_BIND_DN"],
        "ldap_password_file": Path(os.environ["TEST_LDAP_PASSWORD_FILE"]),
        "ldap_group_base_dn": os.environ["TEST_LDAP_GROUP_BASE_DN"],
        "ldap_ca_bundle": Path(os.environ["TEST_LDAP_CA_BUNDLE"]),
        "ldap_starttls": os.environ.get("TEST_LDAP_STARTTLS", "").lower() == "true",
    }


@pytest.fixture(scope="module")
def dc_users() -> dict[str, dict[str, str]]:
    data = json.loads(Path(os.environ["TEST_DC_USERS_FILE"]).read_text(encoding="utf-8"))
    users: dict[str, dict[str, str]] = data["users"]
    return users


@pytest.fixture
def directory(tmp_path: Path) -> Directory:
    return Directory(make_settings(tmp_path, **ldap_values()))


def roles_of(d: Directory, login: str) -> list[str]:
    u = d.find_user(login)
    assert u is not None, login
    return groups.roles_from_groups(d.groups_of(u.dn))


def test_password_bind(directory: Directory, dc_users: dict[str, dict[str, str]]) -> None:
    u = directory.find_user("clerk.clara")
    assert u is not None and not u.disabled and u.emp_no == "4001" and u.display_name == "Clara Clerk"
    assert directory.verify_password(u.dn, dc_users["clerk.clara"]["password"]) is True
    assert directory.verify_password(u.dn, "not-the-password") is False
    assert directory.verify_password(u.dn, "") is False
    assert roles_of(directory, "clerk.clara") == ["clerical"]
    again = directory.get_user_by_guid(u.guid)
    assert again is not None and again.sam_account == "clerk.clara"


def test_nested_and_levels(directory: Directory) -> None:
    assert roles_of(directory, "nested.ned") == ["shop:10:worker"]
    assert roles_of(directory, "both.bo") == ["shop:03:supervisor"]
    assert roles_of(directory, "multi.mia") == ["clerical", "shop:06:supervisor"]
    assert roles_of(directory, "norole.nick") == []


def test_disabled_account(directory: Directory, dc_users: dict[str, dict[str, str]]) -> None:
    u = directory.find_user("disabled.dan")
    assert u is not None and u.disabled is True
    assert directory.verify_password(u.dn, dc_users["disabled.dan"]["password"]) is False


def test_group_listing_and_members(directory: Directory) -> None:
    found = directory.list_groups()
    lower = {g.lower() for g in found}
    assert "ps-14-electrical-lead" not in lower and "ps-extra-legacy" in lower
    assert sum(1 for n in groups.group_names() if n.lower() in lower) == 49
    sams = {u.sam_account for u in directory.members_of("PS-03-WeldParts-Worker")}
    assert {"weld.andy", "both.bo", "dup.dave"} <= sams
    assert directory.members_of("PS-No-Such-Group") == []


@needs_pg
def test_refresh_flags_duplicates(pg_settings: Settings, directory: Directory) -> None:
    with app_conn(pg_settings) as conn:
        res = refresh_from_ad(conn, directory)
    assert res.groups_missing == ["PS-14-Electrical-Lead"]
    assert "9001" in res.emp_duplicate and "noemp.nora" in res.emp_missing


@needs_pg
def test_signin_api_against_dc(
    pg_settings: Settings, tmp_path: Path, dev_visual: None, dc_users: dict[str, dict[str, str]]
) -> None:
    from packetstation.main import create_app

    s = make_settings(
        tmp_path / "dc",
        **ldap_values(),
        **{k: getattr(pg_settings, k) for k in ("pg_host", "pg_port", "pg_database")},
        pg_app_password_file=pg_settings.pg_app_password_file,
        pg_migrate_password_file=pg_settings.pg_migrate_password_file,
    )
    with TestClient(create_app(s), headers=ORIGIN_HEADERS) as c:

        def signin(sam: str, pw: str | None = None) -> Any:
            password = dc_users[sam]["password"] if pw is None else pw
            return c.post("/api/auth/signin/password", json={"login": sam, "password": password})

        r = signin("clerk.clara")
        assert r.status_code == 200 and r.json()["roles"] == ["clerical"]
        assert signin("clerk.clara", "wrong").status_code == 401
        assert signin("disabled.dan").status_code == 401
        r = signin("norole.nick")
        assert r.status_code == 403 and r.json()["code"] == "no_role"
        assert signin("nested.ned").json()["roles"] == ["shop:10:worker"]