PacketStation/server/packetstation/auth/directory.py

"""The cached directory: `users` and `user_roles` rows written from what AD reports.

Packet Station grants no roles. `user_roles` is a cache of the roles AD groups give, rebuilt at every
sign-in, on the session's roles refresh and by the AD refresh.
"""

from __future__ import annotations

import logging
import uuid
from collections import Counter
from dataclasses import dataclass, field
from typing import Any

from psycopg.rows import dict_row
from psycopg.types.json import Jsonb

from .. import audit
from . import groups as groupmap
from . import sessions
from .ldap import AdUser, DirectoryProtocol

log = logging.getLogger(__name__)


def upsert_user(conn: Any, u: AdUser, *, groups: list[str], signin: bool = False) -> bool:
    """Insert or refresh the users row for an AD account. True when the row is new.

    A different, older row holding the same sAMAccountName (a deleted and re-created AD account) is renamed
    and marked inactive so the unique index holds and its history keeps its name.
    """
    with conn.transaction():
        conn.execute(
            "UPDATE users SET sam_account = sam_account || '~' || left(id::text, 8), active = false"
            " WHERE lower(sam_account) = lower(%s) AND id <> %s",
            (u.sam_account, u.guid),
        )
        flags = {"disabled": u.disabled, "emp_missing": not u.emp_no, "groups": groups}
        row = conn.execute(
            "INSERT INTO users (id, sam_account, upn, display_name, emp_no, email, active, directory_flags,"
            " ad_synced_at, last_signin_at)"
            " VALUES (%s, %s, %s, %s, %s, %s, %s, %s, now(), CASE WHEN %s THEN now() END)"
            " ON CONFLICT (id) DO UPDATE SET sam_account = EXCLUDED.sam_account, upn = EXCLUDED.upn,"
            " display_name = EXCLUDED.display_name, emp_no = EXCLUDED.emp_no, email = EXCLUDED.email,"
            " active = EXCLUDED.active, directory_flags = users.directory_flags || EXCLUDED.directory_flags,"
            " ad_synced_at = now(), last_signin_at = coalesce(EXCLUDED.last_signin_at, users.last_signin_at)"
            " RETURNING (xmax = 0) AS inserted",
            (
                u.guid,
                u.sam_account,
                u.upn,
                u.display_name,
                u.emp_no,
                u.email,
                not u.disabled,
                Jsonb(flags),
                signin,
            ),
        ).fetchone()
    return bool(row[0])


def set_roles(conn: Any, user_id: Any, roles: list[str]) -> None:
    """Rebuild a user's `user_roles` rows from role strings."""
    with conn.transaction():
        conn.execute("DELETE FROM user_roles WHERE user_id = %s", (user_id,))
        for text in roles:
            r = groupmap.parse_role(text)
            conn.execute(
                "INSERT INTO user_roles (user_id, role, dept_code) VALUES (%s, %s::role_kind, %s)",
                (user_id, r.kind, r.dept_code),
            )


def roles_of(conn: Any, user_id: Any) -> list[str]:
    rows = conn.execute("SELECT role::text, dept_code FROM user_roles WHERE user_id = %s", (user_id,)).fetchall()
    return groupmap.current().sort_roles(groupmap.role_from_row(r[0], r[1]) for r in rows)


def get_user(conn: Any, user_id: Any) -> dict[str, Any] | None:
    with conn.cursor(row_factory=dict_row) as cur:
        row: dict[str, Any] | None = cur.execute("SELECT * FROM users WHERE id = %s", (user_id,)).fetchone()
    return row


def mark_inactive(conn: Any, user_id: Any, *, reason: str, gone: bool) -> None:
    conn.execute(
        "UPDATE users SET active = false, directory_flags = directory_flags || %s, ad_synced_at = now() WHERE id = %s",
        (Jsonb({"disabled": not gone, "gone": gone}), user_id),
    )
    conn.execute("DELETE FROM user_roles WHERE user_id = %s", (user_id,))
    sessions.revoke_user(conn, user_id, reason)


def public_user(row: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": str(row["id"]),
        "sam_account": row["sam_account"],
        "upn": row.get("upn"),
        "display_name": row["display_name"],
        "emp_no": row.get("emp_no"),
        "email": row.get("email"),
    }


def ad_user_public(u: AdUser) -> dict[str, Any]:
    return {
        "id": str(u.guid),
        "dn": u.dn,
        "sam_account": u.sam_account,
        "upn": u.upn,
        "display_name": u.display_name,
        "emp_no": u.emp_no,
        "email": u.email,
    }


# ---------------------------------------------------------------- AD refresh


@dataclass
class RefreshResult:
    groups_checked: int = 0
    groups_missing: list[str] = field(default_factory=list)
    users_seen: int = 0
    created: int = 0
    updated: int = 0
    deactivated: int = 0
    emp_missing: list[str] = field(default_factory=list)
    emp_duplicate: list[str] = field(default_factory=list)

    def as_dict(self) -> dict[str, Any]:
        return {
            "groups_checked": self.groups_checked,
            "groups_missing": self.groups_missing,
            "users_seen": self.users_seen,
            "created": self.created,
            "updated": self.updated,
            "deactivated": self.deactivated,
            "emp_missing": self.emp_missing,
            "emp_duplicate": self.emp_duplicate,
        }


def refresh_from_ad(
    conn: Any, directory: DirectoryProtocol, *, actor: sessions.Actor | None = None, ip: str | None = None
) -> RefreshResult:
    """Enumerate every mapped group, upsert its members and rebuild their roles; deactivate gone accounts.

    All AD reads happen first, then the database writes in one transaction, so no lock is held while AD
    answers. Raises LdapUnavailable when AD cannot be read.
    """
    res = RefreshResult()
    names = groupmap.group_names()
    in_ad = {g.lower() for g in directory.list_groups()}
    res.groups_missing = [n for n in names if n.lower() not in in_ad]
    present = [n for n in names if n.lower() in in_ad]
    seen: dict[uuid.UUID, AdUser] = {}
    member_groups: dict[uuid.UUID, set[str]] = {}
    for name in present:
        for u in directory.members_of(name):
            seen.setdefault(u.guid, u)
            member_groups.setdefault(u.guid, set()).add(name)
    res.groups_checked = len(present)
    res.users_seen = len(seen)

    known = conn.execute("SELECT id FROM users WHERE active").fetchall()
    unseen = [r[0] for r in known if r[0] not in seen]
    # Accounts not in any group: still in AD (keep, no roles), disabled, or gone
    unseen_state: dict[uuid.UUID, AdUser | None] = {uid: directory.get_user_by_guid(uid) for uid in unseen}

    enabled = [u for u in seen.values() if not u.disabled]
    res.emp_missing = sorted(u.sam_account for u in enabled if not u.emp_no)
    counts = Counter(u.emp_no for u in enabled if u.emp_no)
    res.emp_duplicate = sorted(e for e, n in counts.items() if n > 1)
    dup = set(res.emp_duplicate)

    with conn.transaction():
        for guid, u in seen.items():
            grp = groupmap.mapped_groups(member_groups[guid])
            was_active_row = conn.execute("SELECT active FROM users WHERE id = %s", (guid,)).fetchone()
            created = upsert_user(conn, u, groups=grp)
            conn.execute(
                "UPDATE users SET directory_flags = directory_flags || %s WHERE id = %s",
                (Jsonb({"emp_duplicate": bool(u.emp_no and u.emp_no in dup), "gone": False}), guid),
            )
            if created:
                res.created += 1
            else:
                res.updated += 1
            if u.disabled:
                set_roles(conn, guid, [])
                sessions.revoke_user(conn, guid, "ad_disabled")
                if was_active_row is not None and was_active_row[0]:
                    res.deactivated += 1
                    audit.append(
                        conn,
                        action="ad.user_deactivated",
                        user_id=actor.user_id if actor else None,
                        session_id=actor.session_id if actor else None,
                        role_snapshot=actor.role_snapshot if actor else None,
                        subject_type="user",
                        subject_id=guid,
                        detail={"reason": "disabled", "sam_account": u.sam_account},
                        ip=ip,
                    )
            else:
                set_roles(conn, guid, groupmap.roles_from_groups(grp))
        for uid, found in unseen_state.items():
            if found is None or found.disabled:
                mark_inactive(conn, uid, reason="ad_disabled" if found else "ad_gone", gone=found is None)
                res.deactivated += 1
                audit.append(
                    conn,
                    action="ad.user_deactivated",
                    user_id=actor.user_id if actor else None,
                    session_id=actor.session_id if actor else None,
                    role_snapshot=actor.role_snapshot if actor else None,
                    subject_type="user",
                    subject_id=uid,
                    detail={"reason": "gone" if found is None else "disabled"},
                    ip=ip,
                )
            else:
                upsert_user(conn, found, groups=[])
                set_roles(conn, uid, [])
                sessions.revoke_user(conn, uid, "roles_lost")
        audit.append(
            conn,
            action="ad.refresh",
            user_id=actor.user_id if actor else None,
            session_id=actor.session_id if actor else None,
            role_snapshot=actor.role_snapshot if actor else None,
            detail={
                "groups_checked": res.groups_checked,
                "groups_missing": len(res.groups_missing),
                "users_seen": res.users_seen,
                "created": res.created,
                "updated": res.updated,
                "deactivated": res.deactivated,
                "emp_missing": len(res.emp_missing),
                "emp_duplicate": len(res.emp_duplicate),
            },
            ip=ip,
        )
    log.info("AD refresh: %s", {k: v for k, v in res.as_dict().items() if isinstance(v, int)})
    return res