100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
"""LDAP authentication (direct-bind) and just-in-time user provisioning.
|
|
|
|
``ldap3`` is synchronous, so its network calls run in a threadpool to avoid
|
|
blocking the event loop. Login builds the user's DN from
|
|
``settings.ldap_bind_dn_template`` and binds directly with their password; a
|
|
successful profile is provisioned locally on first sign-in.
|
|
"""
|
|
|
|
import logging
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from fastapi.concurrency import run_in_threadpool
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import settings
|
|
from app.crud import role as role_crud
|
|
from app.crud import user as user_crud
|
|
from app.models.user import User
|
|
|
|
|
|
@dataclass
|
|
class LdapIdentity:
|
|
username: str
|
|
email: str
|
|
full_name: str | None
|
|
|
|
|
|
def _bind_and_fetch(username: str, password: str) -> LdapIdentity | None:
|
|
"""Bind as the user and read their display name. Returns ``None`` on failure.
|
|
|
|
Runs synchronously inside a threadpool.
|
|
"""
|
|
import ldap3
|
|
from ldap3.core.exceptions import LDAPException
|
|
|
|
user_dn = settings.ldap_bind_dn_template.format(username=username)
|
|
server = ldap3.Server(
|
|
settings.ldap_host,
|
|
port=settings.ldap_port,
|
|
use_ssl=settings.ldap_use_ssl,
|
|
get_info=ldap3.NONE,
|
|
)
|
|
try:
|
|
conn = ldap3.Connection(server, user=user_dn, password=password, auto_bind=True)
|
|
except LDAPException as e:
|
|
logging.exception("Failed to log into LDAP for user %s", user_dn)
|
|
# Wrong credentials, unreachable server, or a refused bind.
|
|
return None
|
|
|
|
full_name: str | None = None
|
|
attr = settings.ldap_full_name_attr
|
|
try:
|
|
found = conn.search(
|
|
user_dn,
|
|
"(objectClass=*)",
|
|
search_scope=ldap3.BASE,
|
|
attributes=[attr],
|
|
)
|
|
if found and conn.entries:
|
|
values = conn.entries[0].entry_attributes_as_dict.get(attr) or []
|
|
full_name = str(values[0]) if values else None
|
|
except LDAPException:
|
|
full_name = None
|
|
finally:
|
|
conn.unbind()
|
|
|
|
email = f"{username}@{settings.ldap_email_domain}"
|
|
return LdapIdentity(username=username, email=email, full_name=full_name)
|
|
|
|
|
|
async def authenticate(username: str, password: str) -> LdapIdentity | None:
|
|
"""Verify credentials against LDAP. Returns the identity or ``None``."""
|
|
if not settings.ldap_enabled:
|
|
return None
|
|
# A simple bind with an empty password can succeed as an anonymous bind;
|
|
# reject empties so that never counts as authentication.
|
|
if not username or not password:
|
|
return None
|
|
return await run_in_threadpool(_bind_and_fetch, username, password)
|
|
|
|
|
|
async def login(db: AsyncSession, username: str, password: str) -> User | None:
|
|
"""Authenticate against LDAP and provision/return the local user profile."""
|
|
identity = await authenticate(username, password)
|
|
if identity is None:
|
|
return None
|
|
default_role = await role_crud.get_default(db)
|
|
role_ids = [default_role.id] if default_role else []
|
|
user = await user_crud.get_or_create_ldap_user(
|
|
db,
|
|
username=identity.username,
|
|
email=identity.email,
|
|
full_name=identity.full_name,
|
|
role_ids=role_ids,
|
|
)
|
|
if not user.is_active:
|
|
return None
|
|
return user
|