Adding LDAP login

.claude/settings.local.json

@@ -6,7 +6,10 @@
       "Bash(git stash *)",
       "Bash(python3 *)",
       "Bash(curl -s -X POST http://localhost:8000/api/auth/login -d \"username=admin@armament.com&password=changethis\" -H \"Content-Type: application/x-www-form-urlencoded\")",
-      "Bash(PYTHONPATH=. uv run *)"
+      "Bash(PYTHONPATH=. uv run *)",
+      "Bash(python *)",
+      "Bash(DATABASE_URL=\"sqlite+aiosqlite:////tmp/claude-1000/-home-iroddis-dev-armament-fitguide/41e5e14d-88fe-45e7-ba3a-be5a2cbde2cd/scratchpad/mig_test.db\" uv run alembic upgrade head)",
+      "Bash(sqlite3 /tmp/claude-1000/-home-iroddis-dev-armament-fitguide/41e5e14d-88fe-45e7-ba3a-be5a2cbde2cd/scratchpad/mig_test.db '.schema users')"
     ]
   }
 }

.env.example

@@ -26,6 +26,21 @@ COOKIE_NAME=access_token
 # Set true when serving over HTTPS.
 COOKIE_SECURE=false
 
+# --- LDAP (optional; direct-bind) ---
+# When enabled, the login page and /api/auth/ldap-login accept a directory
+# username + password. On first login a local profile is provisioned using the
+# username and the default role (mark one in the Roles admin UI).
+LDAP_ENABLED=false
+LDAP_HOST=localhost
+LDAP_PORT=389
+LDAP_USE_SSL=false
+# "{username}" is substituted with the submitted username to form the bind DN.
+LDAP_BIND_DN_TEMPLATE="uid={username},ou=people,dc=example,dc=com"
+# Provisioned users get the email "<username>@<LDAP_EMAIL_DOMAIN>".
+LDAP_EMAIL_DOMAIN=example.com
+# LDAP attribute read for the provisioned user's display name.
+LDAP_FULL_NAME_ATTR=cn
+
 # --- CORS (JSON list) ---
 CORS_ORIGINS=["*"]
 

alembic/versions/c9f1a2b3d4e5_add_ldap_and_default_role.py

@@ -0,0 +1,54 @@
+"""add LDAP user fields and default role flag
+
+Revision ID: c9f1a2b3d4e5
+Revises: 1b99be8a9b1e
+Create Date: 2026-07-16 00:00:00.000000
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "c9f1a2b3d4e5"
+down_revision: str | None = "1b99be8a9b1e"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # Batch mode so SQLite can alter the nullability of hashed_password.
+    with op.batch_alter_table("users") as batch:
+        batch.add_column(sa.Column("username", sa.String(length=255), nullable=True))
+        batch.add_column(
+            sa.Column(
+                "auth_source",
+                sa.String(length=16),
+                nullable=False,
+                server_default="local",
+            )
+        )
+        batch.alter_column(
+            "hashed_password", existing_type=sa.String(length=255), nullable=True
+        )
+        batch.create_index("ix_users_username", ["username"], unique=True)
+
+    op.add_column(
+        "roles",
+        sa.Column(
+            "is_default", sa.Boolean(), nullable=False, server_default=sa.false()
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("roles", "is_default")
+
+    with op.batch_alter_table("users") as batch:
+        batch.drop_index("ix_users_username")
+        batch.alter_column(
+            "hashed_password", existing_type=sa.String(length=255), nullable=False
+        )
+        batch.drop_column("auth_source")
+        batch.drop_column("username")

app/api/auth.py

@@ -2,6 +2,7 @@ from fastapi import APIRouter, Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordRequestForm
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.core.config import settings
 from app.core.database import get_db
 from app.crud import role as role_crud
 from app.crud import token as token_crud
@@ -10,27 +11,34 @@ 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 import ldap as ldap_service
 from app.services.auth import RefreshError, issue_token_pair, rotate_refresh_token
 
 router = APIRouter()
 
+# Fallback default role name if no role is flagged as default in the DB.
 DEFAULT_USER_ROLE = "user"
 
 
+async def _default_role_ids(db: AsyncSession) -> list[int]:
+    role = await role_crud.get_default(db)
+    if role is None:
+        role = await role_crud.get_by_name(db, DEFAULT_USER_ROLE)
+    return [role.id] if role else []
+
+
 @router.post("/register", response_model=UserRead, status_code=status.HTTP_201_CREATED)
 async def register(payload: RegisterRequest, db: AsyncSession = Depends(get_db)):
     if await user_crud.get_by_email(db, payload.email):
         raise HTTPException(
             status_code=status.HTTP_409_CONFLICT, detail="Email already registered"
         )
-    default_role = await role_crud.get_by_name(db, DEFAULT_USER_ROLE)
-    role_ids = [default_role.id] if default_role else []
     user = await user_crud.create(
         db,
         email=payload.email,
         password=payload.password,
         full_name=payload.full_name,
-        role_ids=role_ids,
+        role_ids=await _default_role_ids(db),
     )
     return user
 
@@ -52,6 +60,27 @@ async def login(
     return Token(access_token=access, refresh_token=refresh)
 
 
+@router.post("/ldap-login", response_model=Token)
+async def ldap_login(
+    form: OAuth2PasswordRequestForm = Depends(),
+    db: AsyncSession = Depends(get_db),
+):
+    """Authenticate against LDAP, provisioning the local profile on first login."""
+    if not settings.ldap_enabled:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="LDAP login is not enabled"
+        )
+    user = await ldap_service.login(db, form.username, form.password)
+    if user is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access, refresh = await issue_token_pair(db, user)
+    return Token(access_token=access, refresh_token=refresh)
+
+
 @router.post("/refresh", response_model=Token)
 async def refresh(payload: TokenRefreshRequest, db: AsyncSession = Depends(get_db)):
     try:

app/core/config.py

@@ -37,6 +37,19 @@ class Settings(BaseSettings):
     cookie_name: str = "access_token"
     cookie_secure: bool = False
 
+    # LDAP (direct-bind). When enabled, the login page offers an LDAP option.
+    # At login the user's DN is built from ``ldap_bind_dn_template`` (with
+    # ``{username}`` substituted) and bound directly with their password.
+    ldap_enabled: bool = False
+    ldap_host: str = "localhost"
+    ldap_port: int = 389
+    ldap_use_ssl: bool = False
+    ldap_bind_dn_template: str = "uid={username},ou=people,dc=example,dc=com"
+    # Provisioned LDAP users get a synthesized email: "<username>@<domain>".
+    ldap_email_domain: str = "example.com"
+    # LDAP attribute read for the provisioned user's display name.
+    ldap_full_name_attr: str = "cn"
+
     # CORS
     cors_origins: list[str] = ["*"]
 

app/crud/role.py

@@ -1,4 +1,5 @@
 from sqlalchemy import select
+from sqlalchemy import update as sa_update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.permission import Permission
@@ -14,6 +15,23 @@ async def get_by_name(db: AsyncSession, name: str) -> Role | None:
     return result.scalar_one_or_none()
 
 
+async def get_default(db: AsyncSession) -> Role | None:
+    """The role auto-assigned to new users, or ``None`` if none is marked."""
+    result = await db.execute(select(Role).where(Role.is_default.is_(True)))
+    return result.scalars().first()
+
+
+async def set_default(db: AsyncSession, role: Role) -> Role:
+    """Make ``role`` the sole default, clearing the flag on every other role."""
+    await db.execute(
+        sa_update(Role).where(Role.id != role.id).values(is_default=False)
+    )
+    role.is_default = True
+    await db.commit()
+    await db.refresh(role)
+    return role
+
+
 async def list_roles(db: AsyncSession) -> list[Role]:
     result = await db.execute(select(Role).order_by(Role.id))
     return list(result.scalars().all())
@@ -50,6 +68,16 @@ async def update(db: AsyncSession, role: Role, data: dict) -> Role:
     else:
         data.pop("permission_ids", None)
 
+    # Setting a role default must clear the flag on all others.
+    make_default = data.pop("is_default", None)
+    if make_default:
+        await db.execute(
+            sa_update(Role).where(Role.id != role.id).values(is_default=False)
+        )
+        role.is_default = True
+    elif make_default is False:
+        role.is_default = False
+
     for field, value in data.items():
         if value is not None:
             setattr(role, field, value)

app/crud/user.py

@@ -3,7 +3,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.security import hash_password, verify_password
 from app.models.role import Role
-from app.models.user import User
+from app.models.user import AUTH_SOURCE_LDAP, AUTH_SOURCE_LOCAL, User
 
 
 async def get(db: AsyncSession, user_id: int) -> User | None:
@@ -15,6 +15,11 @@ async def get_by_email(db: AsyncSession, email: str) -> User | None:
     return result.scalar_one_or_none()
 
 
+async def get_by_username(db: AsyncSession, username: str) -> User | None:
+    result = await db.execute(select(User).where(User.username == username))
+    return result.scalar_one_or_none()
+
+
 async def list_users(db: AsyncSession, skip: int = 0, limit: int = 100) -> list[User]:
     result = await db.execute(select(User).order_by(User.id).offset(skip).limit(limit))
     return list(result.scalars().all())
@@ -36,14 +41,18 @@ async def create(
     db: AsyncSession,
     *,
     email: str,
-    password: str,
+    password: str | None = None,
     full_name: str | None = None,
     is_superuser: bool = False,
     role_ids: list[int] | None = None,
+    username: str | None = None,
+    auth_source: str = AUTH_SOURCE_LOCAL,
 ) -> User:
     user = User(
         email=email,
-        hashed_password=hash_password(password),
+        username=username,
+        hashed_password=hash_password(password) if password else None,
+        auth_source=auth_source,
         full_name=full_name,
         is_superuser=is_superuser,
         roles=await _resolve_roles(db, role_ids or []),
@@ -54,6 +63,37 @@ async def create(
     return user
 
 
+async def get_or_create_ldap_user(
+    db: AsyncSession,
+    *,
+    username: str,
+    email: str,
+    full_name: str | None,
+    role_ids: list[int] | None = None,
+) -> User:
+    """Find an LDAP user by username, provisioning a local profile on first login.
+
+    The profile is keyed on the LDAP username; email is synthesized by the
+    caller. Newly provisioned users get ``role_ids`` (the default role).
+    """
+    user = await get_by_username(db, username)
+    if user is not None:
+        # Keep the display name in sync with LDAP on subsequent logins.
+        if full_name and user.full_name != full_name:
+            user.full_name = full_name
+            await db.commit()
+            await db.refresh(user)
+        return user
+    return await create(
+        db,
+        email=email,
+        username=username,
+        full_name=full_name,
+        auth_source=AUTH_SOURCE_LDAP,
+        role_ids=role_ids,
+    )
+
+
 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"))
@@ -81,6 +121,9 @@ async def delete(db: AsyncSession, user: User) -> None:
 
 async def authenticate(db: AsyncSession, email: str, password: str) -> User | None:
     user = await get_by_email(db, email)
-    if not user or not verify_password(password, user.hashed_password):
+    # LDAP users have no local password and must sign in via LDAP.
+    if not user or not user.hashed_password:
+        return None
+    if not verify_password(password, user.hashed_password):
         return None
     return user

app/models/role.py

@@ -1,4 +1,4 @@
-from sqlalchemy import String
+from sqlalchemy import Boolean, String
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.core.database import Base
@@ -11,6 +11,9 @@ class Role(Base):
     id: Mapped[int] = mapped_column(primary_key=True)
     name: Mapped[str] = mapped_column(String(64), unique=True, index=True)
     description: Mapped[str | None] = mapped_column(String(255), default=None)
+    # The role auto-assigned to newly provisioned/registered users. At most one
+    # role is the default; see ``crud.role.set_default``.
+    is_default: Mapped[bool] = mapped_column(Boolean, default=False)
 
     users: Mapped[list["User"]] = relationship(  # noqa: F821
         secondary=user_roles,

app/models/user.py

@@ -6,13 +6,23 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship
 from app.core.database import Base
 from app.models.associations import user_roles
 
+# Where a user's credentials are verified.
+AUTH_SOURCE_LOCAL = "local"
+AUTH_SOURCE_LDAP = "ldap"
+
 
 class User(Base):
     __tablename__ = "users"
 
     id: Mapped[int] = mapped_column(primary_key=True)
     email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
-    hashed_password: Mapped[str] = mapped_column(String(255))
+    # Set for users provisioned from LDAP; how they sign in. Null for local users.
+    username: Mapped[str | None] = mapped_column(
+        String(255), unique=True, index=True, default=None
+    )
+    # Null for LDAP users, who have no local password (verified against LDAP).
+    hashed_password: Mapped[str | None] = mapped_column(String(255), default=None)
+    auth_source: Mapped[str] = mapped_column(String(16), default=AUTH_SOURCE_LOCAL)
     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)

app/schemas/rbac.py

@@ -12,6 +12,7 @@ class PermissionRead(BaseModel):
 class RoleBase(BaseModel):
     name: str
     description: str | None = None
+    is_default: bool = False
 
 
 class RoleCreate(RoleBase):
@@ -21,6 +22,7 @@ class RoleCreate(RoleBase):
 class RoleUpdate(BaseModel):
     name: str | None = None
     description: str | None = None
+    is_default: bool | None = None
     permission_ids: list[int] | None = None
 
 

app/schemas/user.py

@@ -29,6 +29,8 @@ class UserRead(UserBase):
     model_config = ConfigDict(from_attributes=True)
 
     id: int
+    username: str | None = None
+    auth_source: str = "local"
     is_active: bool
     is_superuser: bool
     created_at: datetime

app/services/ldap.py

@@ -0,0 +1,98 @@
+"""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.
+"""
+
+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:
+        # 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

app/templates/admin/_role_row.html

@@ -28,6 +28,16 @@
       </tbody>
     </table>
   </td>
+  <td>
+    {% if r.is_default %}
+      <span class="tag">default</span>
+    {% else %}
+      <button class="btn btn-ghost"
+              hx-post="/admin/roles/{{ r.id }}/default"
+              hx-target="#role-list"
+              hx-swap="outerHTML">Make default</button>
+    {% endif %}
+  </td>
   <td class="row-actions">
     <button class="btn btn-ghost"
             hx-get="/admin/roles/{{ r.id }}/edit"

app/templates/admin/_role_row_edit.html

@@ -1,5 +1,5 @@
 <tr id="role-{{ r.id }}">
-  <td colspan="5">
+  <td colspan="6">
     <form hx-post="/admin/roles/{{ r.id }}"
           hx-target="#role-{{ r.id }}"
           hx-swap="outerHTML"

app/templates/admin/_roles_table.html

@@ -1,7 +1,7 @@
 <div id="role-list" class="card">
   <table class="table">
     <thead>
-      <tr><th>ID</th><th>Name</th><th>Description</th><th>Permissions</th><th></th></tr>
+      <tr><th>ID</th><th>Name</th><th>Description</th><th>Permissions</th><th>Default</th><th></th></tr>
     </thead>
     <tbody>
       {% for r in roles %}

app/templates/login.html

@@ -14,5 +14,19 @@
     </label>
     <button type="submit" class="btn btn-primary">Sign in</button>
   </form>
+
+  {% if ldap_enabled %}
+  <form class="card auth-card" method="post" action="/login/ldap">
+    <h2>Sign in with LDAP</h2>
+    <p class="muted">Use your directory username and password</p>
+    <label>Username
+      <input type="text" name="username" value="{{ username or '' }}" required>
+    </label>
+    <label>Password
+      <input type="password" name="password" required>
+    </label>
+    <button type="submit" class="btn btn-primary">Sign in with LDAP</button>
+  </form>
+  {% endif %}
 </div>
 {% endblock %}

app/web/admin.py

@@ -360,6 +360,33 @@ async def update_role(
     )
 
 
+@router.post("/roles/{role_id}/default", response_class=HTMLResponse)
+async def make_default_role(
+    request: Request,
+    role_id: int,
+    db: AsyncSession = Depends(get_db),
+    actor: User = Depends(web_require("roles:write")),
+):
+    role = await role_crud.get(db, role_id)
+    if role is None:
+        raise HTTPException(status_code=404, detail="Role not found")
+    await role_crud.set_default(db, role)
+    await audit_crud.log(
+        db,
+        actor=actor,
+        action="update",
+        resource_type="Role",
+        resource_id=role_id,
+        description=f"Set role {role.name} (#{role_id}) as the default role",
+    )
+    # Re-render the whole table so every row's default badge refreshes.
+    return templates.TemplateResponse(
+        request,
+        "admin/_roles_table.html",
+        {"roles": await role_crud.list_roles(db), **(await _role_perm_ctx(db))},
+    )
+
+
 @router.delete("/roles/{role_id}")
 async def delete_role(
     role_id: int,

app/web/pages.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, Depends, Form, Request, status
+from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
 from fastapi.responses import HTMLResponse, RedirectResponse, Response
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -7,6 +7,7 @@ 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 import ldap as ldap_service
 from app.services.auth import issue_token_pair
 from app.web.templates import templates
 
@@ -35,7 +36,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")
@@ -51,7 +54,38 @@ async def login_submit(
         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)
+    response = RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)
+    _set_auth_cookie(response, access)
+    return response
+
+
+@router.post("/login/ldap")
+async def login_ldap_submit(
+    request: Request,
+    username: str = Form(...),
+    password: str = Form(...),
+    db: AsyncSession = Depends(get_db),
+):
+    if not settings.ldap_enabled:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
+    user = await ldap_service.login(db, username, password)
+    if user is None:
+        return templates.TemplateResponse(
+            request,
+            "login.html",
+            {
+                "error": "LDAP sign-in failed",
+                "username": username,
+                "ldap_enabled": True,
+            },
             status_code=status.HTTP_401_UNAUTHORIZED,
         )
     access, _ = await issue_token_pair(db, user)

pyproject.toml

@@ -18,6 +18,7 @@ dependencies = [
     "pyjwt>=2.8",
     "bcrypt>=4.1",
     "httpx>=0.27",
+    "ldap3>=2.9",
 ]
 
 [project.optional-dependencies]

scripts/seed.py

@@ -77,6 +77,10 @@ PERMISSIONS = [
     ("audit_log:read", "View audit log"),
 ]
 
+# Role auto-assigned to new (registered/LDAP-provisioned) users unless an admin
+# marks a different one as default in the UI.
+DEFAULT_ROLE = "user"
+
 # Default roles -> permission names. "admin" gets everything.
 ROLES = {
     "admin": [name for name, _ in PERMISSIONS],  # includes audit_log:read
@@ -127,6 +131,12 @@ async def seed() -> None:
                 )
             roles[role_name] = role
 
+        # Ensure a default role is marked (idempotent).
+        if await role_crud.get_default(db) is None:
+            default_role = roles.get(DEFAULT_ROLE)
+            if default_role is not None:
+                await role_crud.set_default(db, default_role)
+
         # Attachment point names
         for apn_name in ATTACHMENT_POINT_NAMES:
             if await apn_crud.get_by_name(db, apn_name) is None:

tests/test_ldap_login.py

@@ -0,0 +1,148 @@
+import pytest
+
+from app.core.config import settings
+from app.crud import role as role_crud
+from app.crud import user as user_crud
+from app.services import ldap as ldap_service
+
+
+@pytest.fixture
+def ldap_enabled(monkeypatch):
+    monkeypatch.setattr(settings, "ldap_enabled", True)
+
+
+def _fake_identity(username="jdoe", full_name="Jane Doe"):
+    return ldap_service.LdapIdentity(
+        username=username,
+        email=f"{username}@example.com",
+        full_name=full_name,
+    )
+
+
+async def test_ldap_login_provisions_user_with_default_role(
+    client, seeded, db_sessionmaker, monkeypatch, ldap_enabled
+):
+    # An admin has marked the "user" role as default.
+    async with db_sessionmaker() as db:
+        role = await role_crud.get(db, seeded["roles"]["user"].id)
+        await role_crud.set_default(db, role)
+
+    async def fake_authenticate(username, password):
+        assert (username, password) == ("jdoe", "secret")
+        return _fake_identity()
+
+    monkeypatch.setattr(ldap_service, "authenticate", fake_authenticate)
+
+    resp = await client.post(
+        "/api/auth/ldap-login",
+        data={"username": "jdoe", "password": "secret"},
+    )
+    assert resp.status_code == 200, resp.text
+    access = resp.json()["access_token"]
+
+    # A local profile was provisioned from the username with the default role.
+    async with db_sessionmaker() as db:
+        user = await user_crud.get_by_username(db, "jdoe")
+        assert user is not None
+        assert user.email == "jdoe@example.com"
+        assert user.full_name == "Jane Doe"
+        assert user.auth_source == "ldap"
+        assert user.hashed_password is None
+        assert {r.name for r in user.roles} == {"user"}
+
+    # The issued token authenticates.
+    me = await client.get(
+        "/api/users/me", headers={"Authorization": f"Bearer {access}"}
+    )
+    assert me.status_code == 200
+    assert me.json()["username"] == "jdoe"
+
+
+async def test_ldap_login_is_idempotent(
+    client, seeded, db_sessionmaker, monkeypatch, ldap_enabled
+):
+    async with db_sessionmaker() as db:
+        role = await role_crud.get(db, seeded["roles"]["user"].id)
+        await role_crud.set_default(db, role)
+
+    async def fake_authenticate(username, password):
+        return _fake_identity()
+
+    monkeypatch.setattr(ldap_service, "authenticate", fake_authenticate)
+
+    for _ in range(2):
+        resp = await client.post(
+            "/api/auth/ldap-login", data={"username": "jdoe", "password": "secret"}
+        )
+        assert resp.status_code == 200
+
+    async with db_sessionmaker() as db:
+        assert await user_crud.count(db) == 1
+
+
+async def test_ldap_login_bad_credentials(
+    client, seeded, monkeypatch, ldap_enabled
+):
+    async def fake_authenticate(username, password):
+        return None
+
+    monkeypatch.setattr(ldap_service, "authenticate", fake_authenticate)
+
+    resp = await client.post(
+        "/api/auth/ldap-login", data={"username": "jdoe", "password": "nope"}
+    )
+    assert resp.status_code == 401
+
+
+async def test_ldap_login_disabled_returns_404(client, seeded):
+    # ldap_enabled defaults to False.
+    resp = await client.post(
+        "/api/auth/ldap-login", data={"username": "jdoe", "password": "secret"}
+    )
+    assert resp.status_code == 404
+
+
+async def test_ldap_authenticate_short_circuits_when_disabled():
+    # No network call: disabled or empty credentials return None immediately.
+    assert await ldap_service.authenticate("jdoe", "secret") is None
+
+
+async def test_set_default_role_is_exclusive(db_sessionmaker):
+    async with db_sessionmaker() as db:
+        a = await role_crud.create(db, name="alpha")
+        b = await role_crud.create(db, name="beta")
+
+        await role_crud.set_default(db, a)
+        assert (await role_crud.get_default(db)).id == a.id
+
+        # Marking another default clears the previous one.
+        await role_crud.set_default(db, b)
+        default = await role_crud.get_default(db)
+        assert default.id == b.id
+
+        await db.refresh(a)
+        assert a.is_default is False
+
+
+async def test_local_login_rejects_ldap_user(
+    client, seeded, db_sessionmaker, monkeypatch, ldap_enabled
+):
+    """An LDAP-provisioned user (no local password) can't use the local login."""
+    async with db_sessionmaker() as db:
+        role = await role_crud.get(db, seeded["roles"]["user"].id)
+        await role_crud.set_default(db, role)
+
+    async def fake_authenticate(username, password):
+        return _fake_identity()
+
+    monkeypatch.setattr(ldap_service, "authenticate", fake_authenticate)
+    await client.post(
+        "/api/auth/ldap-login", data={"username": "jdoe", "password": "secret"}
+    )
+
+    # Local login with the synthesized email must fail (no local password).
+    resp = await client.post(
+        "/api/auth/login",
+        data={"username": "jdoe@example.com", "password": "secret"},
+    )
+    assert resp.status_code == 401

uv.lock

@@ -291,6 +291,7 @@ dependencies = [
     { name = "fastapi" },
     { name = "httpx" },
     { name = "jinja2" },
+    { name = "ldap3" },
     { name = "pydantic", extra = ["email"] },
     { name = "pydantic-settings" },
     { name = "pyjwt" },
@@ -319,6 +320,7 @@ requires-dist = [
     { name = "httpx", specifier = ">=0.27" },
     { 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" },
@@ -518,6 +520,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"
@@ -649,6 +663,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"