refactor: rename package directory fabledscryer → roundtable

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

roundtable/auth/ldap_auth.py

@@ -0,0 +1,98 @@
+"""fabledscryer/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 _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))
+    use_tls = cfg.get("tls", False)
+    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})").format(username=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")
+
+    tls = ldap3.Tls() if use_tls else None
+    server = ldap3.Server(host, port=port, tls=tls, get_info=ldap3.ALL)
+
+    # Service account bind to search for user DN
+    try:
+        svc_conn = ldap3.Connection(server, user=bind_dn, password=bind_password, auto_bind=True)
+    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=True)
+        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 {
+        "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."""
+    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)