skeleton-api/app/services/ldap.py

"""LDAP / Active Directory authentication.

Enabled and configured via ``settings.ldap_*`` (see ``app/core/config.py``).
When enabled, ``authenticate_ldap`` validates a login by binding to the
directory. On success it returns the directory attributes needed to provision
or update a local ``User`` row; the local row is what RBAC, tokens, and the
admin UI operate on.

``ldap3`` is synchronous, so the blocking bind/search runs in a worker thread
via ``run_in_threadpool`` to avoid stalling the async event loop. ``ldap3`` is
imported lazily so the rest of the app runs even when it is not installed and
LDAP is disabled.
"""

from dataclasses import dataclass

from starlette.concurrency import run_in_threadpool

from app.core.config import settings


@dataclass(frozen=True)
class LdapUser:
    """Identity resolved from the directory for a successfully-bound user."""

    email: str
    full_name: str | None = None


async def authenticate_ldap(username: str, password: str) -> LdapUser | None:
    """Return the directory identity if ``username``/``password`` bind, else None.

    Never raises for an ordinary auth failure (bad credentials, user not found,
    server unreachable) — those all return ``None``.
    """
    if not settings.ldap_enabled:
        return None
    if not username or not password:
        # An empty password would trigger an anonymous bind that "succeeds"
        # without proving anything — treat it as a failed login.
        return None
    return await run_in_threadpool(_authenticate_sync, username, password)


def _build_server():
    from ldap3 import Server, Tls

    tls = None
    if settings.ldap_use_ssl or settings.ldap_start_tls:
        import ssl

        validate = ssl.CERT_REQUIRED if settings.ldap_tls_validate else ssl.CERT_NONE
        tls = Tls(
            validate=validate,
            ca_certs_file=settings.ldap_ca_certs_file or None,
        )
    return Server(
        settings.ldap_server_uri,
        use_ssl=settings.ldap_use_ssl,
        tls=tls,
        get_info="NONE",
    )


def _authenticate_sync(username: str, password: str) -> LdapUser | None:
    """Blocking bind + attribute lookup. Runs in a worker thread."""
    from ldap3 import Connection
    from ldap3.core.exceptions import LDAPException
    from ldap3.utils.conv import escape_filter_chars
    from ldap3.utils.dn import escape_rdn

    server = _build_server()

    try:
        if settings.ldap_bind_dn_template:
            # Direct bind: construct the DN and bind straight as the user.
            user_dn = settings.ldap_bind_dn_template.format(
                username=escape_rdn(username)
            )
            return _bind_and_read(server, user_dn, password, username)

        # Search + bind: locate the entry first, then bind as its DN.
        search_filter = settings.ldap_user_search_filter.format(
            username=escape_filter_chars(username)
        )
        finder = Connection(
            server,
            user=settings.ldap_bind_dn or None,
            password=settings.ldap_bind_password or None,
            auto_bind=True,
        )
        try:
            attrs = [settings.ldap_attr_email, settings.ldap_attr_full_name]
            found = finder.search(
                search_base=settings.ldap_user_search_base,
                search_filter=search_filter,
                attributes=attrs,
            )
            if not found or not finder.entries:
                return None
            entry = finder.entries[0]
            user_dn = entry.entry_dn
        finally:
            finder.unbind()

        return _bind_and_read(server, user_dn, password, username, entry=entry)
    except LDAPException:
        # Network error, invalid DN template, etc. Fail closed.
        return None
    except (KeyError, IndexError, ValueError):
        # e.g. a bad ``{username}`` template placeholder.
        return None


def _bind_and_read(server, user_dn, password, username, entry=None):
    """Bind as ``user_dn`` to verify the password, then extract attributes."""
    from ldap3 import Connection
    from ldap3.core.exceptions import LDAPException

    try:
        conn = Connection(server, user=user_dn, password=password, auto_bind=True)
    except LDAPException:
        # Wrong password / bind rejected.
        return None

    try:
        # Direct-bind mode has no prior search entry; read the attributes now.
        if entry is None:
            conn.search(
                search_base=user_dn,
                search_filter="(objectClass=*)",
                search_scope="BASE",
                attributes=[settings.ldap_attr_email, settings.ldap_attr_full_name],
            )
            entry = conn.entries[0] if conn.entries else None
    finally:
        conn.unbind()

    email = _read_attr(entry, settings.ldap_attr_email) or _default_email(username)
    full_name = _read_attr(entry, settings.ldap_attr_full_name)
    return LdapUser(email=email.lower(), full_name=full_name)


def _read_attr(entry, name: str) -> str | None:
    if entry is None:
        return None
    try:
        value = entry[name].value
    except (KeyError, LookupError):
        return None
    if isinstance(value, (list, tuple)):
        value = value[0] if value else None
    return str(value) if value else None


def _default_email(username: str) -> str:
    """Fallback identity when the directory entry exposes no email attribute."""
    return username if "@" in username else f"{username}@ldap.local"