From f2b99089ea1ab989358c9294acfe9c021cef1198 Mon Sep 17 00:00:00 2001 From: Ian Roddis <31021769+iroddis@users.noreply.github.com> Date: Sun, 19 Jul 2026 12:02:36 -0300 Subject: [PATCH] Adding LDAP support --- .env.example | 26 ++++ CLAUDE.md | 7 + README.md | 49 +++++-- alembic/versions/0002_add_auth_provider.py | 33 +++++ app/api/auth.py | 11 +- app/core/config.py | 34 +++++ app/core/security.py | 11 ++ app/crud/user.py | 52 ++++++- app/models/user.py | 4 + app/schemas/user.py | 1 + app/services/auth.py | 34 ++++- app/services/ldap.py | 158 +++++++++++++++++++++ app/templates/login.html | 6 + app/web/pages.py | 15 +- pyproject.toml | 1 + tests/test_ldap.py | 99 +++++++++++++ uv.lock | 23 +++ 17 files changed, 541 insertions(+), 23 deletions(-) create mode 100644 alembic/versions/0002_add_auth_provider.py create mode 100644 app/services/ldap.py create mode 100644 tests/test_ldap.py diff --git a/.env.example b/.env.example index 1ae8726..51bef8c 100644 --- a/.env.example +++ b/.env.example @@ -29,6 +29,32 @@ COOKIE_SECURE=false # --- CORS (JSON list) --- CORS_ORIGINS=["*"] +# --- LDAP / Active Directory (optional) --- +# When enabled, login credentials are validated against the directory and a +# local user is provisioned on first login. Local DB auth still works as a +# fallback (so the seeded superuser can log in). Disabled by default. +LDAP_ENABLED=false +LDAP_SERVER_URI="ldap://ldap.example.com:389" +LDAP_USE_SSL=false +LDAP_START_TLS=false +LDAP_TLS_VALIDATE=true +# LDAP_CA_CERTS_FILE="/etc/ssl/certs/ca.pem" + +# Option 1 — direct bind: build the user DN from the login name. +# LDAP_BIND_DN_TEMPLATE="uid={username},ou=people,dc=example,dc=com" + +# Option 2 — search + bind: a service account finds the user, then we bind as +# the found DN. Leave the template above empty to use this mode. +# LDAP_BIND_DN="cn=svc-reader,dc=example,dc=com" +# LDAP_BIND_PASSWORD="service-account-password" +# LDAP_USER_SEARCH_BASE="ou=people,dc=example,dc=com" +# LDAP_USER_SEARCH_FILTER="(uid={username})" + +# Directory attributes copied onto the local user, and roles granted on first login. +LDAP_ATTR_EMAIL=mail +LDAP_ATTR_FULL_NAME=cn +LDAP_DEFAULT_ROLES=["user"] + # --- Bootstrap superuser (used by scripts/seed.py) --- FIRST_SUPERUSER_EMAIL=admin@example.com FIRST_SUPERUSER_PASSWORD=changethis diff --git a/CLAUDE.md b/CLAUDE.md index 0441cea..6dd91c3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -44,6 +44,13 @@ token logic in `app/services/auth.py`. Keep route handlers thin; put DB logic in - **Auth, two transports**: API uses `Authorization: Bearer`; the HTMX UI uses an httpOnly cookie (`settings.cookie_name`). `app/deps.py` reads both. Web auth failures raise `NotAuthenticated`, handled in `main.py` as a redirect to `/login`. +- **Credential verification** goes through `services/auth.authenticate(db, username, + password)` — both login routes call it, not `crud.user.authenticate` directly. + When `settings.ldap_enabled`, it binds against LDAP first (`services/ldap.py`, + sync `ldap3` offloaded via `run_in_threadpool`), provisions/refreshes the local + row via `crud.user.upsert_ldap_user` (unusable local password, `auth_provider="ldap"`, + `LDAP_DEFAULT_ROLES`), then falls back to local DB auth. LDAP config lives in + `settings.ldap_*`. Tests fake the blocking bind by patching `services.ldap._authenticate_sync`. - **Refresh tokens** are stored server-side (hash + `jti` + `revoked`) in the `refresh_tokens` table. `services/auth.rotate_refresh_token` revokes-then-reissues. Note: JWT payload `exp` is a tz-aware `datetime` (PyJWT only epoch-encodes inside diff --git a/README.md b/README.md index c2b9f85..0c05cec 100644 --- a/README.md +++ b/README.md @@ -4,19 +4,19 @@ A production-shaped starting point for a web app that serves both **HTML (HTMX)* ## Stack -| Concern | Choice | -|----------------|--------| -| Web framework | FastAPI | -| HTML / interactivity | Jinja2 + HTMX | -| ORM | SQLAlchemy 2.0 (async) | -| Migrations | Alembic | -| Database | SQLite (default) or Postgres — one `DATABASE_URL` | -| Auth | JWT access + refresh tokens, rotation, DB-backed revocation | -| Authorization | RBAC: users → roles → permissions | -| Admin UI | Custom HTMX dashboard (user & role CRUD) | -| Config | env vars **or** `.env` (pydantic-settings) | -| Tooling | uv, pytest, ruff | -| Packaging | Dockerfile + docker-compose | +| Concern | Choice | +| -------------------- | ----------------------------------------------------------- | +| Web framework | FastAPI | +| HTML / interactivity | Jinja2 + HTMX | +| ORM | SQLAlchemy 2.0 (async) | +| Migrations | Alembic | +| Database | SQLite (default) or Postgres — one `DATABASE_URL` | +| Auth | JWT access + refresh tokens, rotation, DB-backed revocation | +| Authorization | RBAC: users → roles → permissions | +| Admin UI | Custom HTMX dashboard (user & role CRUD) | +| Config | env vars **or** `.env` (pydantic-settings) | +| Tooling | uv, pytest, ruff | +| Packaging | Dockerfile + docker-compose | ## Layout @@ -154,6 +154,7 @@ Every setting is read from an environment variable first, then `.env` | `AUTO_CREATE_TABLES` | `true` | create tables on startup (dev); use Alembic in prod | | `COOKIE_SECURE` | `false` | set `true` behind HTTPS | | `FIRST_SUPERUSER_EMAIL` / `_PASSWORD` | admin@example.com / changethis | used by seed | +| `LDAP_ENABLED` | `false` | validate logins against LDAP/AD (see below) | ## Auth model @@ -168,6 +169,28 @@ stores the access token in an httpOnly cookie so pages work without client JS. Refresh tokens are stored server-side as a SHA-256 hash + `jti` with a `revoked` flag, so logout and rotation genuinely invalidate them. +### LDAP / Active Directory (optional) + +Set `LDAP_ENABLED=true` to validate logins against a directory. On a successful +bind, a local user is **provisioned on first login** (with an unusable local +password and the roles in `LDAP_DEFAULT_ROLES`) so RBAC, tokens, and the admin +UI work unchanged. Existing accounts still authenticate against the local +database as a fallback, so the seeded superuser keeps working. + +Two lookup modes (configured in `.env`, see `.env.example`): + +- **Direct bind** — set `LDAP_BIND_DN_TEMPLATE`, e.g. + `uid={username},ou=people,dc=example,dc=com`; the login name is substituted + and used to bind directly. +- **Search + bind** — leave the template empty and set `LDAP_BIND_DN` / + `LDAP_BIND_PASSWORD` (a service account) plus `LDAP_USER_SEARCH_BASE` / + `LDAP_USER_SEARCH_FILTER`; the user is located, then re-bound to verify the + password. + +The local user's email/name come from the `LDAP_ATTR_EMAIL` / `LDAP_ATTR_FULL_NAME` +attributes. Use `ldaps://` (`LDAP_USE_SSL=true`) or `LDAP_START_TLS=true` in +production so credentials aren't sent in cleartext. + ## RBAC `users → roles → permissions`. Permissions follow `resource:action` diff --git a/alembic/versions/0002_add_auth_provider.py b/alembic/versions/0002_add_auth_provider.py new file mode 100644 index 0000000..1519783 --- /dev/null +++ b/alembic/versions/0002_add_auth_provider.py @@ -0,0 +1,33 @@ +"""add auth_provider to users + +Revision ID: 0002_add_auth_provider +Revises: 0001_initial +Create Date: 2026-07-19 00:00:00.000000 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa + +from alembic import op + +revision: str = "0002_add_auth_provider" +down_revision: str | None = "0001_initial" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column( + "users", + sa.Column( + "auth_provider", + sa.String(length=20), + nullable=False, + server_default="local", + ), + ) + + +def downgrade() -> None: + op.drop_column("users", "auth_provider") diff --git a/app/api/auth.py b/app/api/auth.py index bda52bf..b66742d 100644 --- a/app/api/auth.py +++ b/app/api/auth.py @@ -10,7 +10,12 @@ from app.deps import get_current_user from app.models.user import User from app.schemas.token import LogoutRequest, Token, TokenRefreshRequest from app.schemas.user import RegisterRequest, UserRead -from app.services.auth import RefreshError, issue_token_pair, rotate_refresh_token +from app.services.auth import ( + RefreshError, + authenticate, + issue_token_pair, + rotate_refresh_token, +) router = APIRouter() @@ -40,8 +45,8 @@ async def login( form: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db), ): - # OAuth2 form uses ``username``; we treat it as the email. - user = await user_crud.authenticate(db, form.username, form.password) + # OAuth2 form uses ``username``; we treat it as the email (or LDAP login). + user = await authenticate(db, form.username, form.password) if user is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, diff --git a/app/core/config.py b/app/core/config.py index edb9c95..135ff6c 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -40,6 +40,40 @@ class Settings(BaseSettings): # CORS cors_origins: list[str] = ["*"] + # --- LDAP / Active Directory (optional) -------------------------------- + # When ``ldap_enabled`` is true, login credentials are validated by binding + # to the directory. A local ``User`` row is provisioned on first successful + # login so RBAC, tokens, and the admin UI keep working. Local database auth + # remains available as a fallback (so the seeded superuser can still log in). + ldap_enabled: bool = False + # e.g. "ldap://ldap.example.com:389" or "ldaps://ldap.example.com:636" + ldap_server_uri: str = "" + ldap_use_ssl: bool = False + ldap_start_tls: bool = False + # Verify the server's TLS certificate (ldaps/StartTLS). Disable only for dev. + ldap_tls_validate: bool = True + ldap_ca_certs_file: str = "" + + # Two ways to locate/authenticate the user (direct bind takes precedence): + # + # 1. Direct bind — build the user's DN from the login name, then bind as it: + # LDAP_BIND_DN_TEMPLATE="uid={username},ou=people,dc=example,dc=com" + ldap_bind_dn_template: str = "" + # + # 2. Search + bind — bind with a service account (or anonymously), search for + # the user, then re-bind as the found DN to verify the password: + ldap_bind_dn: str = "" + ldap_bind_password: str = "" + ldap_user_search_base: str = "" + ldap_user_search_filter: str = "(uid={username})" + + # Attributes read from the directory entry to populate the local user. + ldap_attr_email: str = "mail" + ldap_attr_full_name: str = "cn" + + # Roles (by name) granted to a user provisioned from LDAP on first login. + ldap_default_roles: list[str] = ["user"] + # Bootstrap superuser (used by scripts/seed.py) first_superuser_email: str = "admin@example.com" first_superuser_password: str = "changethis" diff --git a/app/core/security.py b/app/core/security.py index d4fb4e7..90bcc29 100644 --- a/app/core/security.py +++ b/app/core/security.py @@ -1,6 +1,7 @@ """Password hashing and JWT token helpers.""" import hashlib +import secrets import uuid from datetime import UTC, datetime, timedelta @@ -27,6 +28,16 @@ def verify_password(password: str, hashed: str) -> bool: return False +def make_unusable_password() -> str: + """A sentinel hash that no password can match. + + Used for externally-authenticated users (e.g. LDAP) who have no local + password. It is not a valid bcrypt hash, so ``verify_password`` always + returns ``False`` for it — local password auth can never succeed. + """ + return "!" + secrets.token_urlsafe(32) + + # --- JWT --------------------------------------------------------------------- diff --git a/app/crud/user.py b/app/crud/user.py index d371f0a..8695bff 100644 --- a/app/crud/user.py +++ b/app/crud/user.py @@ -1,7 +1,7 @@ from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession -from app.core.security import hash_password, verify_password +from app.core.security import hash_password, make_unusable_password, verify_password from app.models.role import Role from app.models.user import User @@ -32,6 +32,13 @@ async def _resolve_roles(db: AsyncSession, role_ids: list[int]) -> list[Role]: return list(result.scalars().all()) +async def _resolve_roles_by_name(db: AsyncSession, names: list[str]) -> list[Role]: + if not names: + return [] + result = await db.execute(select(Role).where(Role.name.in_(names))) + return list(result.scalars().all()) + + async def create( db: AsyncSession, *, @@ -40,12 +47,14 @@ async def create( full_name: str | None = None, is_superuser: bool = False, role_ids: list[int] | None = None, + auth_provider: str = "local", ) -> User: user = User( email=email, hashed_password=hash_password(password), full_name=full_name, is_superuser=is_superuser, + auth_provider=auth_provider, roles=await _resolve_roles(db, role_ids or []), ) db.add(user) @@ -54,6 +63,47 @@ async def create( return user +async def upsert_ldap_user( + db: AsyncSession, + *, + email: str, + full_name: str | None = None, + default_role_names: list[str] | None = None, +) -> User: + """Provision (or refresh) the local row for an LDAP-authenticated user. + + On first login the user is created with an unusable local password (so + password auth can never succeed) and the configured default roles. On later + logins the existing row is reused; only ``full_name`` is synced from the + directory. Roles assigned afterwards in the admin UI are preserved. + """ + user = await get_by_email(db, email) + if user is None: + user = User( + email=email, + hashed_password=make_unusable_password(), + full_name=full_name, + auth_provider="ldap", + roles=await _resolve_roles_by_name(db, default_role_names or []), + ) + db.add(user) + await db.commit() + await db.refresh(user) + return user + + changed = False + if user.auth_provider != "ldap": + user.auth_provider = "ldap" + changed = True + if full_name and user.full_name != full_name: + user.full_name = full_name + changed = True + if changed: + await db.commit() + await db.refresh(user) + return user + + async def update(db: AsyncSession, user: User, data: dict) -> User: if "password" in data and data["password"]: user.hashed_password = hash_password(data.pop("password")) diff --git a/app/models/user.py b/app/models/user.py index 7a30ff6..2aa9ae7 100644 --- a/app/models/user.py +++ b/app/models/user.py @@ -16,6 +16,10 @@ class User(Base): full_name: Mapped[str | None] = mapped_column(String(255), default=None) is_active: Mapped[bool] = mapped_column(Boolean, default=True) is_superuser: Mapped[bool] = mapped_column(Boolean, default=False) + # Where this account authenticates: "local" (password) or "ldap". + auth_provider: Mapped[str] = mapped_column( + String(20), default="local", server_default="local" + ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now() ) diff --git a/app/schemas/user.py b/app/schemas/user.py index 324a723..bcb8558 100644 --- a/app/schemas/user.py +++ b/app/schemas/user.py @@ -31,6 +31,7 @@ class UserRead(UserBase): id: int is_active: bool is_superuser: bool + auth_provider: str = "local" created_at: datetime roles: list[RoleRead] = [] diff --git a/app/services/auth.py b/app/services/auth.py index cb73f7e..23d4cfb 100644 --- a/app/services/auth.py +++ b/app/services/auth.py @@ -1,8 +1,12 @@ -"""Token issuance / rotation shared by the JSON API and the HTMX UI.""" +"""Auth orchestration shared by the JSON API and the HTMX UI. + +Covers credential verification (LDAP + local) and token issuance / rotation. +""" import jwt from sqlalchemy.ext.asyncio import AsyncSession +from app.core.config import settings from app.core.security import ( REFRESH_TOKEN_TYPE, create_access_token, @@ -10,7 +14,35 @@ from app.core.security import ( decode_token, ) from app.crud import token as token_crud +from app.crud import user as user_crud from app.models.user import User +from app.services import ldap as ldap_service + + +async def authenticate(db: AsyncSession, username: str, password: str) -> User | None: + """Verify credentials and return the matching active user, or ``None``. + + When LDAP is enabled the directory is tried first; on a successful bind the + local user row is provisioned/refreshed. Local database authentication is + always attempted as a fallback, so pre-existing accounts (e.g. the seeded + superuser) keep working alongside LDAP. + """ + if settings.ldap_enabled: + ldap_user = await ldap_service.authenticate_ldap(username, password) + if ldap_user is not None: + user = await user_crud.upsert_ldap_user( + db, + email=ldap_user.email, + full_name=ldap_user.full_name, + default_role_names=settings.ldap_default_roles, + ) + return user if user.is_active else None + + # Local password auth (the only path when LDAP is disabled). + user = await user_crud.authenticate(db, username, password) + if user is not None and not user.is_active: + return None + return user async def issue_token_pair(db: AsyncSession, user: User) -> tuple[str, str]: diff --git a/app/services/ldap.py b/app/services/ldap.py new file mode 100644 index 0000000..45cb672 --- /dev/null +++ b/app/services/ldap.py @@ -0,0 +1,158 @@ +"""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" diff --git a/app/templates/login.html b/app/templates/login.html index 3a82fc7..0b8134b 100644 --- a/app/templates/login.html +++ b/app/templates/login.html @@ -6,9 +6,15 @@

{{ app_name }}

Sign in to continue

{% if error %}
{{ error }}
{% endif %} + {% if ldap_enabled %} + + {% else %} + {% endif %} diff --git a/app/web/pages.py b/app/web/pages.py index f9fe769..165c7b3 100644 --- a/app/web/pages.py +++ b/app/web/pages.py @@ -4,10 +4,9 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import settings from app.core.database import get_db -from app.crud import user as user_crud from app.deps import get_current_web_user from app.models.user import User -from app.services.auth import issue_token_pair +from app.services.auth import authenticate, issue_token_pair from app.web.templates import templates router = APIRouter(include_in_schema=False) @@ -35,7 +34,9 @@ async def dashboard( @router.get("/login", response_class=HTMLResponse) async def login_page(request: Request): - return templates.TemplateResponse(request, "login.html", {}) + return templates.TemplateResponse( + request, "login.html", {"ldap_enabled": settings.ldap_enabled} + ) @router.post("/login") @@ -45,13 +46,17 @@ async def login_submit( password: str = Form(...), db: AsyncSession = Depends(get_db), ): - user = await user_crud.authenticate(db, email, password) + user = await authenticate(db, email, password) if user is None: # Re-render the form with an error (HTMX swaps the form partial). return templates.TemplateResponse( request, "login.html", - {"error": "Incorrect email or password", "email": email}, + { + "error": "Incorrect email or password", + "email": email, + "ldap_enabled": settings.ldap_enabled, + }, status_code=status.HTTP_401_UNAUTHORIZED, ) access, _ = await issue_token_pair(db, user) diff --git a/pyproject.toml b/pyproject.toml index 8c00649..9d0bb4f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,7 @@ dependencies = [ "pydantic-settings>=2.3", "pyjwt>=2.8", "bcrypt>=4.1", + "ldap3>=2.9", ] [project.optional-dependencies] diff --git a/tests/test_ldap.py b/tests/test_ldap.py new file mode 100644 index 0000000..65ccc9d --- /dev/null +++ b/tests/test_ldap.py @@ -0,0 +1,99 @@ +"""LDAP authentication tests. + +The blocking directory bind (``_authenticate_sync``) is replaced with a fake so +no real LDAP server is needed; everything above it — the enabled gate, the +threadpool hop, JIT provisioning, and local fallback — runs for real. +""" + +import pytest + +from app.core.config import settings +from app.crud import user as user_crud +from app.services import ldap as ldap_service + +DIRECTORY = { + # login name -> (password, email, full_name) + "jdoe": ("ldap-pass", "jdoe@corp.example", "Jane Doe"), +} + + +def _fake_sync(username: str, password: str): + record = DIRECTORY.get(username) + if record is None or record[0] != password: + return None + _, email, full_name = record + return ldap_service.LdapUser(email=email.lower(), full_name=full_name) + + +@pytest.fixture +def ldap_enabled(monkeypatch): + monkeypatch.setattr(settings, "ldap_enabled", True) + monkeypatch.setattr(settings, "ldap_default_roles", ["user"]) + monkeypatch.setattr(ldap_service, "_authenticate_sync", _fake_sync) + + +async def test_ldap_login_provisions_local_user( + client, seeded, db_sessionmaker, ldap_enabled +): + resp = await client.post( + "/api/auth/login", + data={"username": "jdoe", "password": "ldap-pass"}, + ) + assert resp.status_code == 200, resp.text + access = resp.json()["access_token"] + + # A local row was created from the directory entry. + async with db_sessionmaker() as db: + user = await user_crud.get_by_email(db, "jdoe@corp.example") + assert user is not None + assert user.auth_provider == "ldap" + assert user.full_name == "Jane Doe" + assert "user" in {r.name for r in user.roles} # default role applied + + # The issued token authenticates against the API. + me = await client.get( + "/api/users/me", headers={"Authorization": f"Bearer {access}"} + ) + assert me.status_code == 200 + assert me.json()["email"] == "jdoe@corp.example" + assert me.json()["auth_provider"] == "ldap" + + +async def test_ldap_wrong_password_rejected(client, seeded, ldap_enabled): + resp = await client.post( + "/api/auth/login", + data={"username": "jdoe", "password": "nope"}, + ) + assert resp.status_code == 401 + + +async def test_ldap_user_cannot_use_local_password( + client, seeded, db_sessionmaker, ldap_enabled +): + # Provision via LDAP first. + await client.post( + "/api/auth/login", data={"username": "jdoe", "password": "ldap-pass"} + ) + # The local password slot is unusable, so DB auth must fail for this user + # even though the account now exists. + async with db_sessionmaker() as db: + assert ( + await user_crud.authenticate(db, "jdoe@corp.example", "ldap-pass") is None + ) + + +async def test_local_user_still_logs_in_when_ldap_enabled( + client, seeded, make_user, ldap_enabled +): + # A local account that does NOT exist in the directory falls back to DB auth. + await make_user("local@example.com", password="secret123") + resp = await client.post( + "/api/auth/login", + data={"username": "local@example.com", "password": "secret123"}, + ) + assert resp.status_code == 200, resp.text + + +async def test_empty_password_is_rejected(ldap_enabled): + # Guards against an anonymous bind masquerading as a successful login. + assert await ldap_service.authenticate_ldap("jdoe", "") is None diff --git a/uv.lock b/uv.lock index 78996e0..e242127 100644 --- a/uv.lock +++ b/uv.lock @@ -290,6 +290,7 @@ dependencies = [ { name = "bcrypt" }, { name = "fastapi" }, { name = "jinja2" }, + { name = "ldap3" }, { name = "pydantic", extra = ["email"] }, { name = "pydantic-settings" }, { name = "pyjwt" }, @@ -317,6 +318,7 @@ requires-dist = [ { name = "fastapi", specifier = ">=0.111" }, { name = "httpx", marker = "extra == 'dev'", specifier = ">=0.27" }, { name = "jinja2", specifier = ">=3.1" }, + { name = "ldap3", specifier = ">=2.9" }, { name = "pydantic", extras = ["email"], specifier = ">=2.7" }, { name = "pydantic-settings", specifier = ">=2.3" }, { name = "pyjwt", specifier = ">=2.8" }, @@ -516,6 +518,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/62/a1/3d680cbfd5f4b8f15abc1d571870c5fc3e594bb582bc3b64ea099db13e56/jinja2-3.1.6-py3-none-any.whl", hash = "sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67", size = 134899, upload-time = "2025-03-05T20:05:00.369Z" }, ] +[[package]] +name = "ldap3" +version = "2.9.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pyasn1" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/43/ac/96bd5464e3edbc61595d0d69989f5d9969ae411866427b2500a8e5b812c0/ldap3-2.9.1.tar.gz", hash = "sha256:f3e7fc4718e3f09dda568b57100095e0ce58633bcabbed8667ce3f8fbaa4229f", size = 398830, upload-time = "2021-07-18T06:34:21.786Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/4e/f6/71d6ec9f18da0b2201287ce9db6afb1a1f637dedb3f0703409558981c723/ldap3-2.9.1-py2.py3-none-any.whl", hash = "sha256:5869596fc4948797020d3f03b7939da938778a0f9e2009f7a072ccf92b8e8d70", size = 432192, upload-time = "2021-07-18T06:34:12.905Z" }, +] + [[package]] name = "mako" version = "1.3.12" @@ -647,6 +661,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, ] +[[package]] +name = "pyasn1" +version = "0.6.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a4/9a/23310166d960def5897e91fe20e5b724601b02a22e84ba1f94232c0b7f67/pyasn1-0.6.4.tar.gz", hash = "sha256:9c447d8431c947fe4c8febc4ed9e760bc29011a5b01e5c74b67025bd9fb8ce81", size = 151262, upload-time = "2026-07-09T01:12:33.988Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9a/3b/6163796d69c3977d1e4287bea4a6979161cbbdd170ebb430511e8e1999ce/pyasn1-0.6.4-py3-none-any.whl", hash = "sha256:deda9277cfd454080ec40b207fb6df82206a3a2688735233cdcd8d3d565f088b", size = 84410, upload-time = "2026-07-09T01:12:32.92Z" }, +] + [[package]] name = "pydantic" version = "2.13.4"