bambuddy/backend/app/services/ldap_service.py

"""LDAP authentication service for BamBuddy (#794).

Supports:
- LDAP bind authentication (simple bind with user's credentials)
- StartTLS, LDAPS, and plaintext connections
- User search with configurable filter
- Group membership resolution for role mapping
"""

from __future__ import annotations

import json
import logging
from dataclasses import dataclass

from ldap3 import ALL, SUBTREE, Connection, Server, Tls
from ldap3.core.exceptions import LDAPException, LDAPObjectClassError, LDAPOperationResult

logger = logging.getLogger(__name__)


@dataclass
class LDAPUserInfo:
    """User information retrieved from LDAP after successful authentication."""

    username: str
    email: str | None
    display_name: str | None
    groups: list[str]  # List of group DNs the user belongs to


@dataclass
class LDAPSearchResult:
    """A directory user returned by the admin search endpoint (no auth performed)."""

    username: str
    email: str | None
    display_name: str | None
    dn: str


@dataclass
class LDAPConfig:
    """LDAP configuration parsed from settings."""

    server_url: str
    bind_dn: str
    bind_password: str
    search_base: str
    user_filter: str  # e.g. "(sAMAccountName={username})"
    security: str  # "none", "starttls", "ldaps"
    group_mapping: dict[str, str]  # LDAP group DN -> BamBuddy group name
    auto_provision: bool
    ca_cert_path: str  # Path to CA certificate file (empty = skip verification)
    default_group: str  # Fallback BamBuddy group assigned when user has no mapped groups (empty = no fallback)


def parse_ldap_config(settings: dict[str, str]) -> LDAPConfig | None:
    """Parse LDAP config from settings key-value pairs. Returns None if LDAP not enabled."""
    if settings.get("ldap_enabled", "false").lower() != "true":
        return None

    server_url = settings.get("ldap_server_url", "").strip()
    if not server_url:
        return None

    group_mapping_raw = settings.get("ldap_group_mapping", "")
    try:
        group_mapping = json.loads(group_mapping_raw) if group_mapping_raw else {}
    except json.JSONDecodeError:
        group_mapping = {}

    return LDAPConfig(
        server_url=server_url,
        bind_dn=settings.get("ldap_bind_dn", "").strip(),
        bind_password=settings.get("ldap_bind_password", ""),
        search_base=settings.get("ldap_search_base", "").strip(),
        user_filter=settings.get("ldap_user_filter", "(sAMAccountName={username})").strip(),
        security=settings.get("ldap_security", "starttls").strip(),
        group_mapping=group_mapping if isinstance(group_mapping, dict) else {},
        auto_provision=settings.get("ldap_auto_provision", "false").lower() == "true",
        ca_cert_path=settings.get("ldap_ca_cert_path", "").strip(),
        default_group=settings.get("ldap_default_group", "").strip(),
    )


def _create_server(config: LDAPConfig) -> Server:
    """Create an ldap3 Server instance from config.

    Always uses TLS — either LDAPS (TLS from start) or StartTLS (upgrade after connect).
    Plaintext LDAP is not supported.
    """
    import ssl

    use_ssl = config.security == "ldaps" or config.server_url.startswith("ldaps://")

    if config.ca_cert_path:
        tls = Tls(validate=ssl.CERT_REQUIRED, ca_certs_file=config.ca_cert_path)
    else:
        tls = Tls(validate=ssl.CERT_NONE)

    return Server(config.server_url, use_ssl=use_ssl, tls=tls, get_info=ALL, connect_timeout=10)


class LDAPStartTLSRefusedError(Exception):
    """The server answered the StartTLS request with an error result."""


def _start_tls_if_configured(conn: Connection, config: LDAPConfig) -> None:
    """Upgrade `conn` with StartTLS when the config asks for it.

    A server that doesn't offer StartTLS (lldap, for one, only does LDAPS)
    rejects the request with a bare "Unsupported extended operation" and the
    StartTLS OID, which doesn't tell the admin what to change (#3197). Only a
    rejection by the server is reworded; TLS handshake and certificate failures
    keep their own messages.

    ldap3 re-reads the server's info and schema straight after StartTLS by
    default, still unauthenticated. A directory that refuses anonymous
    searches (Active Directory, Samba AD) answers that read with
    "operationsError", which failed every StartTLS connection to it and would
    otherwise be reported as a refused StartTLS here. bind() reads the same
    info once authenticated, so the early read is skipped.
    """
    if config.security != "starttls" or config.server_url.startswith("ldaps://"):
        return
    try:
        conn.start_tls(read_server_info=False)
    except LDAPOperationResult as e:
        raise LDAPStartTLSRefusedError(
            f"the server refused StartTLS ({e.description}). If it only offers LDAPS, "
            "choose LDAPS and use its ldaps:// URL and port"
        ) from e


def _open_service_connection(config: LDAPConfig, server: Server, *, check_names: bool = True) -> Connection:
    """Open and bind a service-account LDAP connection. Raises on failure.

    `check_names` toggles ldap3's client-side attribute-name validation. The
    default keeps it on so typos in `user_filter` fail loudly. The fuzzy
    directory search disables it because its fixed OR filter spans both AD-only
    (sAMAccountName, displayName) and OpenLDAP-only attribute names — without
    this bypass ldap3 throws `LDAPAttributeError` before any request is sent
    on a directory whose schema doesn't define one of the names.
    """
    conn = Connection(
        server,
        user=config.bind_dn,
        password=config.bind_password,
        auto_bind=False,
        raise_exceptions=True,
        read_only=True,
        check_names=check_names,
    )
    conn.open()
    _start_tls_if_configured(conn, config)
    conn.bind()
    return conn


# Group classes that list their members by DN, and the attribute each uses.
_MEMBER_DN_GROUP_CLASSES = (("groupOfNames", "member"), ("groupOfUniqueNames", "uniqueMember"))


def _schema_defines(names, name: str) -> bool | None:
    """Whether a schema dict (attribute types or object classes) has `name`.

    None when the server published no schema. ldap3 then skips its client-side
    name checks, so the caller can request the name without it raising.
    """
    if not names:
        return None
    return name in names


def _user_search_attributes(server: Server) -> list[str]:
    """Attributes to request for a user entry: all user attributes plus memberOf.

    `*` alone does not return memberOf on every directory. OpenLDAP's memberof
    overlay makes it operational and lldap only returns it when asked by name
    (#3197), so it has to be listed. But ldap3 rejects a requested name that the
    server's schema doesn't define before anything is sent, even with
    check_names off, so it is only listed when the schema has it (or publishes
    no schema at all).
    """
    schema = server.schema
    has_member_of = _schema_defines(schema.attribute_types if schema else None, "memberOf")
    return ["*"] if has_member_of is False else ["*", "memberOf"]


def _groups_base(server: Server, search_base: str) -> str:
    """The naming context that contains `search_base`, else `search_base`.

    Groups usually live beside the users rather than under them (ou=groups next
    to ou=people), so a group search from the user search base finds nothing.
    """
    base_lower = search_base.lower()
    contexts = server.info.naming_contexts if server.info and server.info.naming_contexts else []
    for context in contexts:
        context_lower = str(context).lower()
        if base_lower == context_lower or base_lower.endswith("," + context_lower):
            return str(context)
    return search_base


# Root DSE capability that Active Directory (and Samba AD) advertises.
_ACTIVE_DIRECTORY_CAPABILITY = "1.2.840.113556.1.4.800"


def _is_active_directory(server: Server) -> bool:
    info = server.info
    features = info.supported_features if info and info.supported_features else []
    return any(feature[0] == _ACTIVE_DIRECTORY_CAPABILITY for feature in features)


def _member_dn_groups(service_conn: Connection, config: LDAPConfig, user_dn: str) -> list[str]:
    """Find groupOfNames / groupOfUniqueNames entries that list `user_dn` as a member.

    memberOf alone is not enough outside Active Directory. Plain OpenLDAP has
    none without the memberof overlay; with it, the overlay tracks only the
    group class it was configured for (osixia's image: groupOfUniqueNames, so
    groupOfNames groups are missing) and only groups changed after it was
    loaded. The groups themselves are the authority, so they are asked too.

    Active Directory is skipped: it keeps memberOf complete itself, and its
    groups are objectClass=group, not either class searched here, so the
    search would cost a subtree walk from the domain root and find nothing.

    Only the classes the schema defines go into the filter, for the same
    client-side check the POSIX lookup trips over.
    """
    schema = service_conn.server.schema
    object_classes = schema.object_classes if schema else None
    clauses = [
        f"(&(objectClass={object_class})({attribute}={_ldap_escape(user_dn)}))"
        for object_class, attribute in _MEMBER_DN_GROUP_CLASSES
        if _schema_defines(object_classes, object_class) is not False
    ]
    if not clauses:
        return []
    search_filter = clauses[0] if len(clauses) == 1 else f"(|{''.join(clauses)})"
    try:
        service_conn.search(
            search_base=_groups_base(service_conn.server, config.search_base),
            search_filter=search_filter,
            search_scope=SUBTREE,
            attributes=["cn"],
        )
    except LDAPException as e:
        # The exception text can carry the user's DN (PII, #2681), so only its
        # type is logged. The user still logs in, with memberOf and POSIX groups.
        logger.warning("LDAP group membership lookup failed (%s); mapping without it", type(e).__name__)
        return []
    return [str(entry.entry_dn) for entry in service_conn.entries]


def _pick_canonical_username(entry, fallback: str) -> str:
    """Prefer sAMAccountName, then uid, then the supplied fallback."""
    if hasattr(entry, "sAMAccountName") and entry.sAMAccountName:
        return str(entry.sAMAccountName)
    if hasattr(entry, "uid") and entry.uid:
        return str(entry.uid)
    return fallback


def _extract_user_info(
    service_conn: Connection, config: LDAPConfig, user_entry, fallback_username: str
) -> LDAPUserInfo:
    """Build an LDAPUserInfo from an already-fetched directory entry.

    Collects memberOf groups, the groupOfNames / groupOfUniqueNames entries
    that list the user (except on Active Directory), POSIX
    memberUid groups, and the primary gidNumber group; dedups DNs
    case-insensitively. Uses the supplied service-bound connection for the
    group searches. The entry must have been fetched with
    `_user_search_attributes`, or memberOf may be missing from it.
    """
    email = str(user_entry.mail) if hasattr(user_entry, "mail") and user_entry.mail else None
    display_name = (
        str(user_entry.displayName) if hasattr(user_entry, "displayName") and user_entry.displayName else None
    )

    # Collect groups from the memberOf attribute (Active Directory, lldap,
    # OpenLDAP with the memberof overlay, 389-DS), then ask the groups too.
    groups = [str(g) for g in user_entry.memberOf] if hasattr(user_entry, "memberOf") and user_entry.memberOf else []
    if not _is_active_directory(service_conn.server):
        groups.extend(_member_dn_groups(service_conn, config, str(user_entry.entry_dn)))

    canonical_username = _pick_canonical_username(user_entry, fallback_username)

    # Also search for POSIX groups, both the memberUid kind and the primary
    # gidNumber kind. Both filters name the posixGroup object class, and ldap3
    # validates that name against the schema it fetched at connect time
    # (get_info=ALL) before it builds the request — so on a directory that
    # publishes a schema without posixGroup it raises client-side and nothing is
    # ever sent. A directory with no posixGroup class has no posixGroup entries,
    # which is exactly the answer the searches would have returned, so the
    # correct response is to carry on with the memberOf groups collected above.
    #
    # Left uncaught, that exception escaped authenticate_ldap_user, and the login
    # route reports any LDAP error as "Incorrect username or password" — so an
    # lldap user, whose accounts carry posixAccount but whose directory defines
    # no group classes beyond groupOfNames, could never log in and had nothing
    # but a wrong-password message to go on (#2769). This predates the primary
    # gidNumber lookup: the memberUid filter has named the class since #794.
    try:
        posix_filter = f"(&(objectClass=posixGroup)(memberUid={_ldap_escape(canonical_username)}))"
        service_conn.search(
            search_base=config.search_base,
            search_filter=posix_filter,
            search_scope=SUBTREE,
            attributes=["cn"],
        )
        for entry in service_conn.entries:
            groups.append(str(entry.entry_dn))

        # POSIX primary group: user's gidNumber matches a posixGroup's gidNumber.
        # Standard Unix semantics treat this as full group membership, so we need
        # to resolve it to a group DN alongside the memberUid results.
        if hasattr(user_entry, "gidNumber") and user_entry.gidNumber:
            primary_gid = str(user_entry.gidNumber)
            primary_filter = f"(&(objectClass=posixGroup)(gidNumber={_ldap_escape(primary_gid)}))"
            service_conn.search(
                search_base=config.search_base,
                search_filter=primary_filter,
                search_scope=SUBTREE,
                attributes=["cn"],
            )
            for entry in service_conn.entries:
                groups.append(str(entry.entry_dn))
    except LDAPObjectClassError:
        # Logged once per authentication, at info: it is the explanation for a
        # user's POSIX groups being absent from their mapping, and it is not an
        # error the operator can or should act on.
        logger.info(
            "Directory publishes no posixGroup object class; skipping POSIX group lookup "
            "(memberOf and groupOfNames groups are unaffected)"
        )

    # Dedupe group DNs (user may be in a group via both memberUid and primary gidNumber).
    # Case-insensitive comparison — LDAP DNs are case-insensitive by spec.
    seen_lower: set[str] = set()
    deduped_groups: list[str] = []
    for g in groups:
        key = g.lower()
        if key not in seen_lower:
            seen_lower.add(key)
            deduped_groups.append(g)

    return LDAPUserInfo(
        username=canonical_username,
        email=email,
        display_name=display_name,
        groups=deduped_groups,
    )


def authenticate_ldap_user(config: LDAPConfig, username: str, password: str) -> LDAPUserInfo | None:
    """Authenticate a user via LDAP bind.

    1. Bind with service account to search for the user DN
    2. Attempt bind with the user's DN and provided password
    3. On success, retrieve user attributes and group memberships

    Returns LDAPUserInfo on success, None on failure.
    """
    if not password:
        return None

    server = _create_server(config)

    try:
        service_conn = _open_service_connection(config, server)
    except Exception as e:
        logger.warning("LDAP service account bind failed: %s", e)
        return None

    try:
        # Search for the user
        search_filter = config.user_filter.replace("{username}", _ldap_escape(username))
        service_conn.search(
            search_base=config.search_base,
            search_filter=search_filter,
            search_scope=SUBTREE,
            attributes=_user_search_attributes(server),
        )

        if not service_conn.entries:
            logger.info("LDAP user not found: %s", username)
            return None

        user_entry = service_conn.entries[0]
        user_dn = str(user_entry.entry_dn)

        # Step 2: Bind as the user to verify password
        try:
            user_conn = Connection(
                server,
                user=user_dn,
                password=password,
                auto_bind=False,
                raise_exceptions=True,
                read_only=True,
            )
            user_conn.open()
            _start_tls_if_configured(user_conn, config)
            user_conn.bind()
            user_conn.unbind()
        except Exception as e:
            logger.info("LDAP bind failed for user %s: %s", username, e)
            return None

        info = _extract_user_info(service_conn, config, user_entry, username)
        # Don't log the raw DN — its leaf CN is the user's real name (PII, #2681).
        # The username + group count is enough to confirm a successful auth; the
        # support-bundle sanitizer also redacts any DN that slips through (e.g. an
        # ldap3 exception string), but keeping it out of the log at the source is
        # the primary hygiene per the "no private data in logs" rule.
        logger.info(
            "LDAP authentication successful for user: %s (groups: %d)",
            info.username,
            len(info.groups),
        )
        return info
    finally:
        service_conn.unbind()


def lookup_ldap_user(config: LDAPConfig, username: str) -> LDAPUserInfo | None:
    """Look up a directory user by exact username via the service-account bind.

    Performs no password verification — intended for the admin manual-provision
    flow, where the caller has already been authenticated as a BamBuddy admin
    and now needs the directory attributes (email, display name, group DNs)
    to create the user.

    Uses the same `user_filter` template that the login path uses, so anything
    that logs in successfully via auto-provision is also resolvable here.
    """
    server = _create_server(config)

    try:
        service_conn = _open_service_connection(config, server)
    except Exception as e:
        logger.warning("LDAP service account bind failed during lookup: %s", e)
        raise

    try:
        search_filter = config.user_filter.replace("{username}", _ldap_escape(username))
        service_conn.search(
            search_base=config.search_base,
            search_filter=search_filter,
            search_scope=SUBTREE,
            attributes=_user_search_attributes(server),
        )
        if not service_conn.entries:
            logger.info("LDAP lookup: user not found: %s", username)
            return None
        return _extract_user_info(service_conn, config, service_conn.entries[0], username)
    finally:
        service_conn.unbind()


def search_ldap_users(config: LDAPConfig, query: str, limit: int = 25) -> list[LDAPSearchResult]:
    """Fuzzy search the directory for users matching `query`.

    Uses a fixed OR filter across sAMAccountName, uid, mail, displayName, and
    cn — covering both Active Directory and OpenLDAP layouts. The query is
    RFC-4515 escaped so a typed `*` doesn't enumerate the whole directory.
    Returns up to `limit` results (default 25). Service-bind failures raise so
    the caller can surface a 503; "no matches" returns an empty list.

    Callers should enforce a minimum query length (≥2 chars) — short queries
    against a large directory are wasteful and effectively unbounded.
    """
    query = query.strip()
    if len(query) < 2:
        return []

    escaped = _ldap_escape(query)
    search_filter = (
        f"(|(sAMAccountName=*{escaped}*)(uid=*{escaped}*)(mail=*{escaped}*)(displayName=*{escaped}*)(cn=*{escaped}*))"
    )

    server = _create_server(config)

    try:
        # check_names=False so OpenLDAP directories (no sAMAccountName/displayName
        # in schema) don't reject the cross-schema OR filter — see helper docstring.
        service_conn = _open_service_connection(config, server, check_names=False)
    except Exception as e:
        logger.warning("LDAP service account bind failed during search: %s", e)
        raise

    try:
        # attributes=["*"] requests all user attributes. We can't enumerate the
        # AD/OpenLDAP-specific names (sAMAccountName, displayName) explicitly
        # because ldap3 validates the attribute list against the server schema
        # even with check_names=False — and OpenLDAP rejects the AD names. The
        # `*` wildcard is hardcoded in ldap3's ATTRIBUTES_EXCLUDED_FROM_CHECK so
        # it bypasses that validation, and the server returns whatever it has.
        service_conn.search(
            search_base=config.search_base,
            search_filter=search_filter,
            search_scope=SUBTREE,
            attributes=["*"],
            size_limit=limit,
        )
        results: list[LDAPSearchResult] = []
        for entry in service_conn.entries:
            username = _pick_canonical_username(entry, "")
            if not username and hasattr(entry, "cn") and entry.cn:
                # Last resort — some OpenLDAP layouts only have cn
                username = str(entry.cn)
            if not username:
                continue
            email = str(entry.mail) if hasattr(entry, "mail") and entry.mail else None
            display_name = str(entry.displayName) if hasattr(entry, "displayName") and entry.displayName else None
            results.append(
                LDAPSearchResult(
                    username=username,
                    email=email,
                    display_name=display_name,
                    dn=str(entry.entry_dn),
                )
            )
        logger.info("LDAP directory search for %r returned %d result(s)", query, len(results))
        return results
    finally:
        service_conn.unbind()


def resolve_group_mapping(ldap_groups: list[str], group_mapping: dict[str, str]) -> list[str]:
    """Map LDAP group DNs to BamBuddy group names.

    Returns list of BamBuddy group names that the user should be added to.
    Comparison is case-insensitive on the LDAP group DN.
    """
    if not group_mapping:
        return []

    # Build case-insensitive lookup
    mapping_lower = {k.lower(): v for k, v in group_mapping.items()}
    result = []
    for ldap_group in ldap_groups:
        bambuddy_group = mapping_lower.get(ldap_group.lower())
        if bambuddy_group:
            result.append(bambuddy_group)
    return result


def test_ldap_connection(config: LDAPConfig) -> tuple[bool, str]:
    """Test LDAP connection and service account bind.

    Returns (success, message).
    """
    try:
        server = _create_server(config)
        conn = Connection(
            server,
            user=config.bind_dn,
            password=config.bind_password,
            auto_bind=False,
            raise_exceptions=True,
            read_only=True,
        )
        conn.open()
        _start_tls_if_configured(conn, config)
        conn.bind()

        # Try a search to verify search base
        conn.search(
            search_base=config.search_base,
            search_filter="(objectClass=*)",
            search_scope=SUBTREE,
            size_limit=1,
        )
        conn.unbind()
        return True, "LDAP connection successful"
    except Exception as e:
        return False, f"LDAP connection failed: {e}"


def _ldap_escape(value: str) -> str:
    """Escape special characters in LDAP search filter values (RFC 4515)."""
    replacements = {
        "\\": "\\5c",
        "*": "\\2a",
        "(": "\\28",
        ")": "\\29",
        "\x00": "\\00",
    }
    for char, escaped in replacements.items():
        value = value.replace(char, escaped)
    return value