Project-SDE-WP-Suite/server/ldap_auth.py

"""LDAPS authentication against the Windows domain — D13, built in T10.1.

The suite stores no passwords. A sign-in is a **simple bind** to
`ldaps://prime.local:636` as `<sAMAccountName>@prime.local` using the password the
person typed; a successful bind IS the authentication. This module owns that
conversation and nothing else — it does not touch the database, does not issue
sessions, and does not decide what anyone is allowed to do. `server/auth.py` and
the `login` route in `server/app.py` do that.

WHY `prime.local` AND NOT A DC NAME OR AN IP (this is load-bearing, do not "fix" it):
every domain controller's certificate carries `prime.local` in its SAN alongside its
own hostname, and the domain name round-robins in DNS across all six DCs published
in `_ldap._tcp.prime.local`. So connecting to the domain name both passes hostname
validation and gives failover in one move. Connecting to `192.168.3.37` instead
fails with `hostname mismatch` — the DC certificates carry no IP SAN — and the only
way to "fix" that is to disable the check, which is the one thing that must not
happen here (see below).

THE CLIENT PRESENTS NO CERTIFICATE. This module is the TLS *client*; clients verify,
they do not present. What it needs is a trust anchor: `PRIME CONTROLS ROOT CA` plus
`PRIME CONTROLS ISSUING CA 1`, shipped as a PEM bundle at `server/certs/`. Those are
public certificates — no private key, nothing secret, nothing issued to this app.

TWO WAYS THIS GOES CATASTROPHICALLY WRONG, both guarded here:

  1. An EMPTY PASSWORD. In LDAP a simple bind with an empty password is an
     *anonymous* bind, and it SUCCEEDS. Without an explicit guard a blank password
     authenticates as whatever username was submitted. `verify()` therefore rejects
     an empty or whitespace-only password BEFORE `bind()` is ever called. If you are
     refactoring this file and that check looks redundant, it is not.

  2. `validate=ssl.CERT_NONE`. It still encrypts, so it fails silently — what it
     loses is the ability to tell the real DC from someone who terminates the TLS
     session, harvests the domain password and relays the bind onward. Domain
     credentials cross this channel, so that turns an app compromise into a Windows
     compromise. `CERT_REQUIRED` with an explicit CA file is the only mode here, and
     the system trust store is deliberately NOT used: it currently trusts five other
     self-signed CAs on this estate, any of which could issue a DC-shaped cert.

FAILED BINDS COUNT AGAINST THE DOMAIN LOCKOUT POLICY. That is why nothing in here
retries a rejected credential — only genuine network failures are retried, and a
`bind()` that returns False is final. The caller throttles before it gets this far
(T10.2); this module's job is not to make the problem worse.

Unconfigured is a first-class state, as it is for `MICRON_DB_URL` in `assets_db.py`:
with no CA bundle or no `ldap3` installed, `is_configured()` is False and `verify()`
returns `UNCONFIGURED` rather than raising. Since D13 leaves no local password
fallback, the caller must surface that state loudly — an unconfigured deploy and a
mistyped password look identical at the login box otherwise.
"""
import logging
import os
import re
import ssl
from dataclasses import dataclass
from typing import Optional

log = logging.getLogger("wpsuite.ldap")

# ldap3 is pinned in requirements.txt. The import is guarded rather than assumed so
# that this module — and the test suite that imports it — still loads on a checkout
# where the dependency has not been installed. `is_configured()` reports the truth,
# and the startup line from `describe()` says so out loud.
try:
    from ldap3 import Server, Connection, Tls, SIMPLE, SUBTREE, NONE
    from ldap3.core.exceptions import (
        LDAPException,
        LDAPSocketOpenError,
        LDAPSessionTerminatedByServerError,
        LDAPCertificateError,
    )
    from ldap3.utils.conv import escape_filter_chars
    HAVE_LDAP3 = True
except ImportError:                                    # pragma: no cover
    HAVE_LDAP3 = False
    LDAPException = LDAPSocketOpenError = Exception
    LDAPSessionTerminatedByServerError = LDAPCertificateError = Exception

    def escape_filter_chars(x, encoding=None):         # type: ignore[misc]
        raise RuntimeError("ldap3 is not installed")


# ── configuration ─────────────────────────────────────────────────────────────
# Module level, matching how server/auth.py reads AUTH_SESSION_HOURS. Tests patch
# these attributes directly rather than re-importing.
DOMAIN = os.getenv("LDAP_DOMAIN", "prime.local")
# Defaults to the domain name on purpose — see the module docstring. Override only
# if you have a reason, and never with an IP address.
HOST = os.getenv("LDAP_HOST", "") or DOMAIN
PORT = int(os.getenv("LDAP_PORT", "636"))
CA_FILE = os.getenv("LDAP_CA_FILE", "") or os.path.join(
    os.path.dirname(os.path.abspath(__file__)), "certs", "prime-ca-chain.pem"
)
TIMEOUT_SECONDS = int(os.getenv("LDAP_TIMEOUT_SECONDS", "8"))
# Retries apply to CONNECT failures only, never to a rejected credential. Two extra
# attempts covers the realistic case: DNS round-robin handed out a DC that is
# rebooting for patching.
CONNECT_RETRIES = int(os.getenv("LDAP_CONNECT_RETRIES", "2"))
# The initial value and the fallback for the admin-console setting added in T10.5.
REQUIRED_GROUP = os.getenv("LDAP_REQUIRED_GROUP", "")

# AD's "member of, transitively" extensible match. Plain `memberOf` is DIRECT
# membership only and would wrongly refuse anyone who is in a nested child group,
# which is how most estates actually organise access.
NESTED_MEMBER_RULE = "1.2.840.113556.1.4.1941"

_USER_ATTRS = ["sAMAccountName", "mail", "displayName", "userPrincipalName"]

# Reason codes. Machine-readable, for logs and for the caller's branching — never
# for a response body, because several of them would leak whether an account exists.
OK = "ok"
EMPTY_INPUT = "empty_input"
UNCONFIGURED = "unconfigured"
UNREACHABLE = "unreachable"
UNTRUSTED = "untrusted"
BAD_CREDENTIALS = "bad_credentials"
NOT_IN_GROUP = "not_in_group"
NO_DIRECTORY_ENTRY = "no_directory_entry"
GROUP_NOT_FOUND = "group_not_found"
GROUP_CHECK_FAILED = "group_check_failed"

# AD returns these as `data <code>` inside an error-49 message. Kept for the log
# only: telling an unauthenticated caller "your password expired" confirms the
# account exists, which `login()` goes out of its way not to do.
_ERR49 = {
    "525": "no such user",
    "52e": "bad password",
    "530": "not permitted at this time",
    "531": "not permitted at this workstation",
    "532": "password expired",
    "533": "account disabled",
    "701": "account expired",
    "773": "must change password",
    "775": "account locked out",
}
_ERR49_RE = re.compile(r"data\s+([0-9a-fA-F]{3})")


@dataclass(frozen=True)
class LdapResult:
    """Outcome of a bind attempt.

    `ok` is the only field the caller should branch on for allow/deny. `reason` and
    `detail` are for logs. `sam`/`mail`/`full_name` are populated only on success and
    are what T10.4 uses to match or provision the local account.
    """
    ok: bool
    reason: str = ""
    detail: str = ""
    sam: str = ""
    mail: str = ""
    full_name: str = ""
    upn: str = ""

    @property
    def is_config_problem(self) -> bool:
        """True when the failure is ours, not the user's. With no local password
        fallback (D13) these must be logged as errors and surfaced to an operator —
        otherwise a broken deploy is indistinguishable from a forgotten password."""
        return self.reason in (UNCONFIGURED, UNREACHABLE, UNTRUSTED, GROUP_NOT_FOUND,
                               GROUP_CHECK_FAILED)


def base_dn(domain: Optional[str] = None) -> str:
    """`prime.local` -> `DC=prime,DC=local`."""
    d = (domain or DOMAIN or "").strip().strip(".")
    return ",".join(f"DC={part}" for part in d.split(".") if part)


def is_configured() -> bool:
    """Whether a bind could even be attempted. Deliberately does not touch the
    network — `selftest()` does that."""
    return bool(HAVE_LDAP3 and HOST and CA_FILE and os.path.isfile(CA_FILE))


def describe() -> str:
    """One line for the startup log. D13 removed the local password path, so an
    operator needs to see this in `docker compose logs api` rather than discovering
    it when the first person cannot sign in."""
    from . import ldap_fake
    if ldap_fake.is_active():
        return ("*** FAKE DIRECTORY ACTIVE — passwords come from "
                f"{ldap_fake.ENV_VAR}, NOT from the domain. Tests only. ***")
    if not HAVE_LDAP3:
        return "LDAP auth DISABLED — ldap3 is not installed. No one can sign in."
    if not CA_FILE or not os.path.isfile(CA_FILE):
        return (f"LDAP auth DISABLED — CA bundle not found at {CA_FILE!r}. "
                f"No one can sign in. Set LDAP_CA_FILE.")
    group = REQUIRED_GROUP or "(none configured — every domain account may sign in)"
    return (f"LDAP auth enabled — ldaps://{HOST}:{PORT}, domain {DOMAIN}, "
            f"CA {CA_FILE}, required group: {group}")


def _tls() -> "Tls":
    """The only TLS configuration in this module.

    `CERT_REQUIRED` plus an explicit `ca_certs_file`. ldap3 performs the hostname
    check against the certificate's SAN itself whenever validate is not CERT_NONE,
    which is what makes connecting by IP fail instead of quietly succeeding.
    """
    # No `version=` pin: ldap3's default negotiates the highest version both ends
    # support, which against these DCs is TLS 1.3. Pinning PROTOCOL_TLSv1_2 here
    # would silently DOWNGRADE every connection to 1.2.
    return Tls(
        ca_certs_file=CA_FILE,
        validate=ssl.CERT_REQUIRED,
    )


def _server() -> "Server":
    return Server(
        HOST, port=PORT, use_ssl=True, tls=_tls(),
        connect_timeout=TIMEOUT_SECONDS, get_info=NONE,
    )


def _err49(result: Optional[dict]) -> str:
    """Human-readable AD sub-code from a bind failure, for the log only."""
    msg = ((result or {}).get("message") or "")
    m = _ERR49_RE.search(msg)
    if not m:
        return ((result or {}).get("description") or "invalid credentials")
    code = m.group(1).lower()
    return f"{code} ({_ERR49.get(code, 'unrecognised sub-code')})"


def normalize_username(raw: str) -> str:
    """Reduce whatever was typed in the login box to a `sAMAccountName`.

    People type their email address. The mail domain here (`prime-controls.com`) is
    not the AD domain (`prime.local`), so an address is never a valid bind string —
    the local part is used instead. This assumes the mail local part equals the
    sAMAccountName, which is the norm but not guaranteed; where it differs the person
    must type their short logon name, and we log it so that is diagnosable.

    ONE candidate is produced, never a list to try in turn: every rejected bind
    counts against the domain lockout policy, so guessing would let a handful of
    login attempts lock a real account out of Windows.
    """
    name = (raw or "").strip()
    if not name:
        return ""
    # DOMAIN\user, as typed by anyone used to a Windows logon prompt.
    if "\\" in name:
        name = name.rsplit("\\", 1)[1].strip()
    if "@" in name:
        local, _, dom = name.partition("@")
        log.info("login input %r looks like an address; binding as sAMAccountName %r "
                 "(mail domain %r is not the AD domain)", name, local.strip(), dom)
        name = local.strip()
    return name


def _resolve_group_dn(conn, group: str) -> Optional[str]:
    """Accept either a distinguished name or a plain group name, return a DN."""
    g = (group or "").strip()
    if not g:
        return None
    if "," in g and "=" in g:
        return g                     # already a DN
    esc = escape_filter_chars(g)
    conn.search(base_dn(), f"(&(objectClass=group)(|(cn={esc})(sAMAccountName={esc})))",
                search_scope=SUBTREE, attributes=["cn"], size_limit=2)
    if not conn.entries:
        return None
    if len(conn.entries) > 1:
        log.warning("group %r is ambiguous in the directory (%d matches); using %s",
                    g, len(conn.entries), conn.entries[0].entry_dn)
    return conn.entries[0].entry_dn


def member_of(conn, sam: str, group: str) -> bool:
    """Is `sam` in `group`, counting nested membership?

    THE RETURN VALUE OF conn.search() IS NOT OPTIONAL READING. The connection is
    built with raise_exceptions=False, so a search that FAILS returns False and
    leaves conn.entries empty — which is byte-for-byte indistinguishable from "no
    match" if you only look at conn.entries. An earlier version of this function did
    exactly that, and every failure of the extensible-match filter presented to the
    user as "you are not in the group" while they plainly were.

    Two searches, in order, and the second one exists to catch the first being wrong:

      1. AD's transitive matching rule (LDAP_MATCHING_RULE_IN_CHAIN). This is the
         correct query — it walks nested groups, which plain memberOf does not.
      2. If that matches nothing, a plain memberOf equality check for DIRECT
         membership.

    If (2) matches after (1) did not, the person IS a member and is let in — refusing
    a real member is the worse error — but it is logged as a WARNING, because it means
    the transitive rule is returning nothing and NESTED membership is silently not
    working on this connection. That needs a human; it must not pass unnoticed.
    """
    dn = _resolve_group_dn(conn, group)
    if not dn:
        log.error("required group %r does not resolve in %s — refusing the sign-in. "
                  "This is a configuration fault, not a bad password.", group, base_dn())
        raise LookupError(GROUP_NOT_FOUND)

    esc_sam, esc_dn = escape_filter_chars(sam), escape_filter_chars(dn)

    def _search(filt: str, label: str):
        ok = conn.search(base_dn(), filt, search_scope=SUBTREE,
                         attributes=["sAMAccountName"], size_limit=1)
        if not ok:
            log.error("the %s membership search FAILED (not 'no match') for %r: %s | "
                      "filter=%s", label, sam, conn.result, filt)
            raise LookupError(GROUP_CHECK_FAILED)
        return bool(conn.entries)

    if _search(f"(&(sAMAccountName={esc_sam})(memberOf:{NESTED_MEMBER_RULE}:={esc_dn}))",
               "nested"):
        return True

    if _search(f"(&(sAMAccountName={esc_sam})(memberOf={esc_dn}))", "direct"):
        log.warning(
            "%r IS a direct member of %r, but AD's transitive matching rule "
            "(%s) returned nothing for them. Allowing the sign-in — refusing a real "
            "member is worse — but NESTED group membership is not working on this "
            "connection and needs investigating.", sam, dn, NESTED_MEMBER_RULE)
        return True

    return False


def verify(username: str, password: str, required_group: Optional[str] = None) -> LdapResult:
    """Authenticate against the domain. The only entry point the app should call.

    `required_group` overrides the `LDAP_REQUIRED_GROUP` default so the admin-console
    setting (T10.5) wins. Pass an empty string to mean "no group gate"; pass None to
    use the configured default.
    """
    sam = normalize_username(username)

    # ── guard 1: no bind on empty input ──────────────────────────────────────
    # An empty password makes the bind below an ANONYMOUS bind, which SUCCEEDS and
    # would authenticate `sam` without proving anything at all. Must stay before
    # every return path that reaches bind().
    if not sam or not (password or "").strip():
        return LdapResult(False, EMPTY_INPUT, "empty username or password")

    group = REQUIRED_GROUP if required_group is None else required_group

    # Test seam (T10.7). Deliberately placed AFTER the empty-input guard above, so
    # the anonymous-bind guard covers the fake path too — a fake that re-implemented
    # it would let the real one rot without any test noticing. `is_active()` refuses
    # whenever a non-SQLite DATABASE_URL is configured; see server/ldap_fake.py.
    from . import ldap_fake
    if ldap_fake.is_active():
        ok, reason, attrs = ldap_fake.lookup(sam, password, group)
        if not ok:
            log.info("fake directory refused %r: %s", sam, reason)
            return LdapResult(False, reason, "fake directory")
        return LdapResult(True, OK, "fake directory", **attrs)

    if not is_configured():
        return LdapResult(False, UNCONFIGURED, describe())

    bind_user = f"{sam}@{DOMAIN}"
    last_network_error = ""

    # Retries cover CONNECT failures only: DNS round-robin across six DCs will
    # eventually hand out one that is rebooting. A rejected credential returns
    # immediately and is never retried — each attempt counts against AD lockout.
    for attempt in range(1, max(1, CONNECT_RETRIES + 1) + 1):
        conn = None
        try:
            conn = Connection(
                _server(), user=bind_user, password=password,
                authentication=SIMPLE, read_only=True,
                receive_timeout=TIMEOUT_SECONDS, raise_exceptions=False,
            )
            if not conn.bind():
                detail = _err49(conn.result)
                log.warning("bind refused for %r: %s", sam, detail)
                return LdapResult(False, BAD_CREDENTIALS, detail)

            # Bound as the user. AD lets an account read its own object, so no
            # service account is needed for either of the next two steps.
            conn.search(base_dn(),
                        f"(&(objectClass=user)(sAMAccountName={escape_filter_chars(sam)}))",
                        search_scope=SUBTREE, attributes=_USER_ATTRS, size_limit=1)
            if not conn.entries:
                log.error("bind succeeded for %r but the account has no readable "
                          "directory entry under %s", sam, base_dn())
                return LdapResult(False, NO_DIRECTORY_ENTRY, "no readable directory entry")
            e = conn.entries[0]

            def one(attr: str) -> str:
                v = getattr(e, attr, None)
                return str(v.value) if v is not None and v.value else ""

            if group:
                try:
                    if not member_of(conn, sam, group):
                        log.warning("bind succeeded for %r but the account is NOT in %r",
                                    sam, group)
                        return LdapResult(False, NOT_IN_GROUP, f"not in {group}")
                except LookupError as exc:
                    reason = str(exc) or GROUP_NOT_FOUND
                    return LdapResult(False, reason, f"group {group!r}: {reason}")

            return LdapResult(
                True, OK,
                sam=one("sAMAccountName") or sam,
                mail=one("mail"),
                full_name=one("displayName"),
                upn=one("userPrincipalName"),
            )

        except LDAPCertificateError as exc:
            # NOT retried and NOT downgraded. Either the CA bundle is wrong or
            # something is impersonating a DC; both need a human, and retrying with
            # relaxed validation is exactly the wrong instinct.
            log.error("LDAPS certificate validation FAILED against %s: %s. Refusing "
                      "to continue — check LDAP_CA_FILE, and never set CERT_NONE.",
                      HOST, exc)
            return LdapResult(False, UNTRUSTED, str(exc))
        except (LDAPSocketOpenError, LDAPSessionTerminatedByServerError) as exc:
            last_network_error = str(exc)
            log.warning("LDAPS connect to %s:%s failed (attempt %d): %s",
                        HOST, PORT, attempt, exc)
            continue
        except LDAPException as exc:
            log.error("LDAP error for %r: %s", sam, exc)
            return LdapResult(False, UNREACHABLE, str(exc))
        finally:
            if conn is not None:
                try:
                    conn.unbind()
                except Exception:
                    pass

    return LdapResult(False, UNREACHABLE,
                      last_network_error or f"no domain controller answered on {HOST}:{PORT}")


def selftest() -> LdapResult:
    """Open a TLS session to the domain and validate the certificate, WITHOUT binding.

    Used by the startup check and by the admin console's diagnostics. Touches no
    account, so it cannot contribute to a lockout. Proves the three things that
    actually break a deploy: DNS resolves, a DC answers on 636, and the presented
    certificate validates against our CA bundle.
    """
    if not is_configured():
        return LdapResult(False, UNCONFIGURED, describe())
    conn = None
    try:
        conn = Connection(_server(), receive_timeout=TIMEOUT_SECONDS, raise_exceptions=True)
        conn.open()
        return LdapResult(True, OK, f"ldaps://{HOST}:{PORT} certificate validates")
    except LDAPCertificateError as exc:
        return LdapResult(False, UNTRUSTED, str(exc))
    except LDAPException as exc:
        return LdapResult(False, UNREACHABLE, str(exc))
    finally:
        if conn is not None:
            try:
                conn.unbind()
            except Exception:
                pass