FabledSteward/steward/auth/ldap_auth.py

"""steward/auth/ldap_auth.py

LDAP authentication helper. Requires the optional `ldap3` package.
Falls back gracefully if ldap3 is not installed.
"""
from __future__ import annotations
import asyncio
import logging

logger = logging.getLogger(__name__)


def _ldap_available() -> bool:
    try:
        import ldap3  # noqa: F401
        return True
    except ImportError:
        return False


def filter_value(value: str) -> str:
    """Escape a value for an LDAP search filter (RFC 4515), so a login name
    like `*)(uid=*` matches only itself."""
    out = []
    for ch in value:
        if ch in "\\*()\x00":
            out.append("\\%02x" % ord(ch))
        else:
            out.append(ch)
    return "".join(out)


def transport(cfg: dict) -> tuple[bool, bool]:
    """(use_ssl, start_tls) for the configured server: port 636 is LDAPS,
    otherwise the TLS setting means StartTLS before any bind."""
    if int(cfg.get("port", 389)) == 636:
        return True, False
    return False, bool(cfg.get("tls", False))


def _tls(ldap3, cfg: dict):
    """Certificate-checked TLS: against the CA the operator pasted in, else the
    system's CAs. A server whose certificate does not check is refused."""
    import ssl
    ca = (cfg.get("ca_cert") or "").strip()
    return ldap3.Tls(validate=ssl.CERT_REQUIRED, ca_certs_data=ca or None)


def _ldap_authenticate_sync(cfg: dict, username: str, password: str) -> dict | None:
    """Synchronous LDAP auth. Returns user info dict or None on failure.

    Called via run_in_executor so it doesn't block the event loop.
    """
    import ldap3

    host = cfg.get("host", "")
    port = int(cfg.get("port", 389))
    bind_dn = cfg.get("bind_dn", "")
    bind_password = cfg.get("bind_password", "")
    base_dn = cfg.get("base_dn", "")
    user_filter = cfg.get("user_filter", "(uid={username})").replace("{username}", filter_value(username))
    admin_group_dn = cfg.get("admin_group_dn", "")
    operator_group_dn = cfg.get("operator_group_dn", "")
    attr_username = cfg.get("attr_username", "uid")
    attr_email = cfg.get("attr_email", "mail")

    use_ssl, start_tls = transport(cfg)
    tls = _tls(ldap3, cfg) if (use_ssl or start_tls) else None
    server = ldap3.Server(host, port=port, use_ssl=use_ssl, tls=tls, get_info=ldap3.ALL)
    auto_bind = ldap3.AUTO_BIND_TLS_BEFORE_BIND if start_tls else ldap3.AUTO_BIND_NO_TLS

    # Service account bind to search for user DN
    try:
        svc_conn = ldap3.Connection(server, user=bind_dn, password=bind_password, auto_bind=auto_bind)
    except Exception as exc:
        logger.error("LDAP service bind failed: %s", exc)
        return None

    try:
        svc_conn.search(
            search_base=base_dn,
            search_filter=user_filter,
            attributes=[attr_username, attr_email, "memberOf", "dn"],
        )
        if not svc_conn.entries:
            logger.debug("LDAP user not found: %s", username)
            return None
        entry = svc_conn.entries[0]
        user_dn = str(entry.entry_dn)
        ldap_username = str(getattr(entry, attr_username, [username])[0]) if hasattr(entry, attr_username) else username
        ldap_email = str(getattr(entry, attr_email, [""])[0]) if hasattr(entry, attr_email) else ""
        member_of: list[str] = [str(g) for g in getattr(entry, "memberOf", [])]
    except Exception as exc:
        logger.error("LDAP search failed: %s", exc)
        return None
    finally:
        svc_conn.unbind()

    # Bind as the user to verify password
    try:
        user_conn = ldap3.Connection(server, user=user_dn, password=password, auto_bind=auto_bind)
        user_conn.unbind()
    except Exception:
        logger.debug("LDAP password verify failed for %s", username)
        return None

    # Map role from group membership
    role = "viewer"
    if admin_group_dn and admin_group_dn in member_of:
        role = "admin"
    elif operator_group_dn and operator_group_dn in member_of:
        role = "operator"

    return {
        "subject": user_dn,
        "username": ldap_username,
        "email": ldap_email,
        "role": role,
    }


async def ldap_authenticate(cfg: dict, username: str, password: str) -> dict | None:
    """Async wrapper around synchronous LDAP auth. Returns user info or None.

    An empty password is refused here: LDAP treats a simple bind with a DN
    and no password as an anonymous bind, which succeeds."""
    if not username or not password:
        return None
    if not _ldap_available():
        logger.warning("ldap3 not installed — LDAP auth unavailable")
        return None
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, _ldap_authenticate_sync, cfg, username, password)