feat(auth): LDAP and SSO users keep their source; directory TLS is real (#5036)

#5036 "SSO and LDAP: users keep their auth source, no merging into local
accounts, is_active checked, LDAP TLS real and filter escaped".

- users gain auth_source (local/ldap/oidc) and external_id (migration 0027).
  A provider account is found by (source, subject) only: the LDAP entry DN
  or the OIDC sub. A first sign-in whose username or email another account
  holds is refused with a message naming why; a disabled user stays out.
- Local passwords sign in local accounts only.
- LDAP: port 636 is LDAPS, otherwise StartTLS before any bind; the server
  certificate is checked against the system CAs or a CA pasted in
  Settings -> Auth. The login name is escaped into the filter (RFC 4515),
  and an empty password is refused before the directory sees it.
- OIDC: the ID token must carry this sign-in's nonce and the userinfo sub.

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

steward/auth/ldap_auth.py

@@ -18,6 +18,34 @@ def _ldap_available() -> bool:
         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.
 
@@ -27,22 +55,23 @@ def _ldap_authenticate_sync(cfg: dict, username: str, password: str) -> dict | N
 
     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)
+    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")
 
-    tls = ldap3.Tls() if use_tls else None
+    use_ssl, start_tls = transport(cfg)
-    server = ldap3.Server(host, port=port, tls=tls, get_info=ldap3.ALL)
+    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=True)
+        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
@@ -69,7 +98,7 @@ def _ldap_authenticate_sync(cfg: dict, username: str, password: str) -> dict | N
 
     # Bind as the user to verify password
     try:
-        user_conn = ldap3.Connection(server, user=user_dn, password=password, auto_bind=True)
+        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)
@@ -83,6 +112,7 @@ def _ldap_authenticate_sync(cfg: dict, username: str, password: str) -> dict | N
         role = "operator"
 
     return {
+        "subject": user_dn,
         "username": ldap_username,
         "email": ldap_email,
         "role": role,
@@ -90,7 +120,12 @@ def _ldap_authenticate_sync(cfg: dict, username: str, password: str) -> dict | N
 
 
 async def ldap_authenticate(cfg: dict, username: str, password: str) -> dict | None:
-    """Async wrapper around synchronous LDAP auth. Returns user info or 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

steward/auth/oidc.py

@@ -95,6 +95,25 @@ def decode_id_token_payload(id_token: str) -> dict:
         return {}
 
 
+def id_token_refusal(id_token: str, nonce: str | None, userinfo_sub: str) -> str | None:
+    """Why this sign-in's ID token does not belong to it, or None when it does.
+
+    The token must carry the nonce this browser was sent off with, so a token
+    from another sign-in cannot be replayed into this one, and name the same
+    subject as the userinfo response (OIDC Core 5.3.2). The signature is not
+    checked (see the module note)."""
+    if not userinfo_sub:
+        return "SSO returned no subject (sub) for this user."
+    claims = decode_id_token_payload(id_token) if id_token else {}
+    if not claims:
+        return "SSO returned no ID token. Check that the scopes include openid."
+    if not nonce or claims.get("nonce") != nonce:
+        return "This SSO sign-in does not match the one Steward started. Try again."
+    if claims.get("sub") != userinfo_sub:
+        return "SSO returned an ID token for a different user. Try again."
+    return None
+
+
 def map_role(groups: list[str], admin_group: str, operator_group: str) -> str:
     """Map IdP groups to a local role string."""
     if admin_group and admin_group in groups:

steward/auth/routes.py

@@ -10,31 +10,63 @@ from steward.models.users import User, UserRole
 auth_bp = Blueprint("auth", __name__)
 
 
-async def _provision_external_user(app, username: str, email: str, role_str: str):
+_SOURCE_LABEL = {"ldap": "LDAP", "oidc": "SSO"}
-    """Create or update a user from an external auth provider (OIDC/LDAP).
 
-    External users get a randomised unusable password so local login is blocked.
+
-    Their role is updated on every login to reflect current IdP group membership.
+async def _provision_external_user(app, source: str, subject: str, username: str,
+                                   email: str, role_str: str) -> tuple[User | None, str | None]:
+    """The Steward user for an identity a provider vouched for: (user, None),
+    or (None, the reason to show at the login page).
+
+    Found by (source, subject) only. A first sign-in creates the user with an
+    unusable password; a username or email another account already holds is
+    refused rather than merged, so a directory entry named `admin` cannot sign
+    in as the local admin. The role follows the provider's groups on every
+    sign-in; a user disabled in Steward stays out.
     """
     import secrets
-    from steward.models.users import UserRole
+    from sqlalchemy.exc import IntegrityError
-    async with app.db_sessionmaker() as db:
+    label = _SOURCE_LABEL[source]
-        result = await db.execute(select(User).where(User.username == username))
+    if not subject or len(subject) > 512 or not username or len(username) > 64:
-        user = result.scalar_one_or_none()
+        return None, f"{label} returned a username or id Steward cannot store (at most 64 characters)."
     new_role = UserRole(role_str)
-        async with db.begin():
-            if user is None:
-                unusable_hash = bcrypt.hashpw(secrets.token_bytes(32), bcrypt.gensalt()).decode()
-                user = User(username=username, email=email or f"{username}@external",
-                            password_hash=unusable_hash, role=new_role)
-                db.add(user)
-            else:
-                if user.role != new_role:
-                    user.role = new_role
-    # Re-fetch committed user
     async with app.db_sessionmaker() as db:
-        result = await db.execute(select(User).where(User.username == username))
+        try:
-        return result.scalar_one()
+            async with db.begin():
+                user = (await db.execute(select(User).where(
+                    User.auth_source == source, User.external_id == subject))).scalar_one_or_none()
+                if user is not None:
+                    if not user.is_active:
+                        return None, "This account is disabled in Steward."
+                    user.role = new_role
+                else:
+                    email = email or f"{username}@{source}"
+                    clash = (await db.execute(select(User.username).where(
+                        (User.username == username) | (User.email == email)))).first()
+                    if clash is not None:
+                        return None, (
+                            f"Steward already has an account named {clash[0]} with this username "
+                            f"or email, and it does not sign in with {label}. An admin must rename "
+                            f"or remove it before this {label} account can sign in.")
+                    user = User(username=username, email=email, role=new_role,
+                                password_hash=bcrypt.hashpw(secrets.token_bytes(32),
+                                                            bcrypt.gensalt()).decode(),
+                                auth_source=source, external_id=subject)
+                    db.add(user)
+        except IntegrityError:
+            # Another sign-in created the same name between our check and insert.
+            return None, f"Could not create the {label} account; try signing in again."
+        db.expunge(user)
+    return user, None
+
+
+async def _login_refused(message: str, next_url: str | None = None):
+    """The login page with why a provider-vouched sign-in was still refused."""
+    from quart import current_app
+    return await render_template(
+        "auth/login.html", error=message,
+        oidc_enabled=current_app.config.get("OIDC", {}).get("enabled", False),
+        next_url=safe_next_url(next_url)), 403
 
 
 async def get_user_count(app) -> int:
@@ -89,7 +121,8 @@ async def login_post():
 
     # Try local auth first
     user = await get_user_by_username(current_app, username)
-    if user and user.is_active and bcrypt.checkpw(password, user.password_hash.encode()):
+    if (user and user.auth_source == "local" and user.is_active
+            and bcrypt.checkpw(password, user.password_hash.encode())):
         login_throttle.succeeded(ip)
         login_user(user)
         await log_audit(current_app, user.id, user.username, "auth.login",
@@ -102,9 +135,11 @@ async def login_post():
         from steward.auth.ldap_auth import ldap_authenticate
         ldap_info = await ldap_authenticate(ldap_cfg, username, password_str)
         if ldap_info:
-            user = await _provision_external_user(
+            user, refusal = await _provision_external_user(
-                current_app, ldap_info["username"], ldap_info["email"], ldap_info["role"]
+                current_app, "ldap", ldap_info["subject"], ldap_info["username"],
-            )
+                ldap_info["email"], ldap_info["role"])
+            if refusal:
+                return await _login_refused(refusal, form.get("next"))
             login_throttle.succeeded(ip)
             login_user(user)
             await log_audit(current_app, user.id, user.username, "auth.login",
@@ -153,7 +188,8 @@ async def login_oidc():
 @public_route("SSO login callback")
 async def login_oidc_callback():
     from quart import current_app
-    from steward.auth.oidc import get_discovery, exchange_code, get_userinfo, map_role
+    from steward.auth.oidc import (get_discovery, exchange_code, get_userinfo, map_role,
+                                    id_token_refusal)
     from steward.core.audit import log_audit
 
     oidc_cfg = current_app.config.get("OIDC", {})
@@ -169,6 +205,7 @@ async def login_oidc_callback():
         return await render_template("auth/login.html", error="Invalid SSO state — try again.",
                                      oidc_enabled=True), 400
 
+    nonce = session.pop("oidc_nonce", None)
     try:
         doc = await get_discovery(oidc_cfg["discovery_url"])
         redirect_uri = url_for("auth.login_oidc_callback", _external=True)
@@ -180,23 +217,25 @@ async def login_oidc_callback():
         return await render_template("auth/login.html", error=f"SSO login failed: {exc}",
                                      oidc_enabled=True), 502
 
+    refusal = id_token_refusal(tokens.get("id_token", ""), nonce, userinfo.get("sub", ""))
+    if refusal:
+        return await render_template("auth/login.html", error=refusal, oidc_enabled=True), 400
+
     username_claim = oidc_cfg.get("username_claim", "preferred_username")
     email_claim = oidc_cfg.get("email_claim", "email")
     groups_claim = oidc_cfg.get("groups_claim", "groups")
 
-    username = userinfo.get(username_claim) or userinfo.get("sub", "")
+    subject = userinfo["sub"]
-    email = userinfo.get(email_claim, "") or f"{username}@oidc"
+    username = userinfo.get(username_claim) or subject
+    email = userinfo.get(email_claim, "")
     groups = userinfo.get(groups_claim, [])
     if not isinstance(groups, list):
         groups = []
     role = map_role(groups, oidc_cfg.get("admin_group", ""), oidc_cfg.get("operator_group", ""))
 
-    if not username:
+    user, refusal = await _provision_external_user(current_app, "oidc", subject, username, email, role)
-        return await render_template("auth/login.html",
+    if refusal:
-                                     error="SSO returned no username.",
+        return await _login_refused(refusal, session.pop("oidc_next", None))
-                                     oidc_enabled=True), 400
-
-    user = await _provision_external_user(current_app, username, email, role)
     login_user(user)
     await log_audit(current_app, user.id, user.username, "auth.login",
                     detail={"method": "oidc", "role": role})

steward/core/settings.py

@@ -134,6 +134,7 @@ DEFAULTS: dict[str, Any] = {
     "ldap.host": "",
     "ldap.port": 389,
     "ldap.tls": False,
+    "ldap.ca_cert": "",
     "ldap.bind_dn": "",
     "ldap.bind_password": "",
     "ldap.base_dn": "",

steward/migrations/versions/0027_user_auth_source.py

@@ -0,0 +1,30 @@
+"""Users keep the auth source they came from
+
+Revision ID: 0027_user_auth_source
+Revises: 0026_api_tokens
+Create Date: 2026-10-06
+"""
+from typing import Sequence, Union
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "0027_user_auth_source"
+down_revision: Union[str, None] = "0026_api_tokens"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Existing rows are local: nothing recorded which provider made them, and
+    # an SSO user who finds their name taken is told so at sign-in.
+    op.add_column("users", sa.Column("auth_source", sa.String(16), nullable=False,
+                                     server_default="local"))
+    op.add_column("users", sa.Column("external_id", sa.String(512), nullable=True))
+    op.create_unique_constraint("uq_users_auth_source_external_id", "users",
+                                ["auth_source", "external_id"])
+
+
+def downgrade() -> None:
+    op.drop_constraint("uq_users_auth_source_external_id", "users", type_="unique")
+    op.drop_column("users", "external_id")
+    op.drop_column("users", "auth_source")

steward/models/users.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 import enum
 import uuid
 from datetime import datetime, timezone
-from sqlalchemy import Boolean, DateTime, Enum, String
+from sqlalchemy import Boolean, DateTime, Enum, String, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column
 from .base import Base
 
@@ -13,8 +13,16 @@ class UserRole(str, enum.Enum):
     viewer = "viewer"
 
 
+# Where a user's identity lives. A local user signs in with a Steward
+# password; an ldap or oidc user is created at their first sign-in, holds an
+# unusable password, and is found again only by (auth_source, external_id) —
+# never by a matching username, so a directory cannot sign in as a local account.
+AUTH_SOURCES = ("local", "ldap", "oidc")
+
+
 class User(Base):
     __tablename__ = "users"
+    __table_args__ = (UniqueConstraint("auth_source", "external_id", name="uq_users_auth_source_external_id"),)
 
     id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
     username: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
@@ -25,3 +33,6 @@ class User(Base):
         DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)
     )
     is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    auth_source: Mapped[str] = mapped_column(String(16), nullable=False, default="local", server_default="local")
+    # The provider's stable id: the LDAP entry DN, or the OIDC `sub`. None for local users.
+    external_id: Mapped[str | None] = mapped_column(String(512), nullable=True)

steward/settings/routes.py

@@ -466,6 +466,7 @@ async def save_auth_settings():
             await set_setting(db, "ldap.host", form.get("ldap.host", "").strip())
             await set_setting(db, "ldap.port", int(form.get("ldap.port") or 389))
             await set_setting(db, "ldap.tls", "ldap.tls" in form)
+            await set_setting(db, "ldap.ca_cert", form.get("ldap.ca_cert", "").strip())
             await set_setting(db, "ldap.bind_dn", form.get("ldap.bind_dn", "").strip())
             if form.get("ldap.bind_password"):
                 await set_setting(db, "ldap.bind_password", form.get("ldap.bind_password").strip())

steward/templates/settings/auth.html

@@ -117,6 +117,18 @@
         <label for="ldap-tls" style="margin:0;white-space:nowrap;">StartTLS</label>
       </div>
     </div>
+    <p style="font-size:0.75rem;color:var(--text-muted);margin:-0.4rem 0 1rem;">Port 636 connects with LDAPS; any other port with StartTLS ticked
+      upgrades the connection before any password is sent. Without either, passwords cross
+      the network in the clear. The server's certificate is checked.</p>
+
+    <div class="form-group">
+      <label>CA certificate (PEM, optional)</label>
+      <textarea name="ldap.ca_cert" rows="4" placeholder="-----BEGIN CERTIFICATE-----"
+                style="font-family:monospace;font-size:0.8rem;">{{ settings['ldap.ca_cert'] }}</textarea>
+      <p style="font-size:0.75rem;color:var(--text-muted);margin-top:0.2rem;">The certificate
+        authority that signed the directory's certificate, when it is not one the system already
+        trusts (a private or self-signed CA).</p>
+    </div>
 
     <div class="form-group">
       <label>Service account bind DN</label>

tests/integration/test_external_users.py

@@ -0,0 +1,64 @@
+"""Integration: provider accounts are found by (source, subject) against real
+Postgres, never merged into a local account by a matching name."""
+from __future__ import annotations
+
+import asyncio
+import os
+import uuid
+
+import pytest
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def app():
+    if not os.environ.get("STEWARD_DATABASE_URL"):
+        pytest.skip("needs Postgres")
+    from steward.app import create_app
+    return create_app(testing=False)
+
+
+def test_provider_accounts_keep_their_source(app):
+    from sqlalchemy import select, update
+    from steward.auth.routes import _provision_external_user
+    from steward.models.users import User, UserRole
+
+    async def _go():
+        tag = uuid.uuid4().hex[:8]
+        local = f"loc-{tag}"
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                s.add(User(username=local, email=f"{local}@test", password_hash="x",
+                           role=UserRole.admin))
+
+        # A directory entry with a local user's name is refused, not merged.
+        user, refusal = await _provision_external_user(
+            app, "ldap", f"uid={local},dc=x", local, "", "admin")
+        assert user is None and f"named {local}" in refusal and "LDAP" in refusal
+
+        # First sign-in creates the account; the next finds it by subject and
+        # follows the provider's role.
+        name = f"sso-{tag}"
+        first, refusal = await _provision_external_user(app, "oidc", f"sub-{tag}", name, "", "viewer")
+        assert refusal is None
+        assert (first.auth_source, first.external_id, first.role) == ("oidc", f"sub-{tag}", UserRole.viewer)
+        again, _ = await _provision_external_user(app, "oidc", f"sub-{tag}", name, "", "operator")
+        assert again.id == first.id and again.role == UserRole.operator
+
+        # Another subject claiming the same username is someone else.
+        other, refusal = await _provision_external_user(app, "oidc", f"sub2-{tag}", name, "", "admin")
+        assert other is None and refusal
+
+        # Disabled in Steward stays out, whatever the provider says.
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await s.execute(update(User).where(User.id == first.id).values(is_active=False))
+        gone, refusal = await _provision_external_user(app, "oidc", f"sub-{tag}", name, "", "admin")
+        assert gone is None and "disabled" in refusal
+
+        async with app.db_sessionmaker() as s:
+            row = (await s.execute(select(User).where(User.username == local))).scalar_one()
+        assert (row.auth_source, row.external_id, row.role) == ("local", None, UserRole.admin)
+
+    asyncio.run(_go())

tests/test_auth.py

@@ -12,6 +12,7 @@ def make_user(role=UserRole.admin):
         password_hash=bcrypt.hashpw(b"password", bcrypt.gensalt()).decode(),
         role=role,
         is_active=True,
+        auth_source="local",
     )
 
 

tests/test_external_auth.py

@@ -0,0 +1,151 @@
+"""LDAP and SSO sign-in: the directory is reached over checked TLS with the
+login name escaped, an SSO token must belong to the sign-in it ends, and a
+provider's account never signs in as a local one."""
+import base64
+import json
+import sys
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from steward.auth import ldap_auth
+from steward.auth.oidc import id_token_refusal
+from tests.test_auth import make_user, with_csrf
+
+
+@pytest.mark.parametrize("raw,escaped", [
+    ("bob", "bob"),
+    ("*", "\\2a"),
+    ("*)(uid=*", "\\2a\\29\\28uid=\\2a"),
+    ("a\\b", "a\\5cb"),
+    ("nul\x00", "nul\\00"),
+])
+def test_filter_value_escapes_filter_syntax(raw, escaped):
+    assert ldap_auth.filter_value(raw) == escaped
+
+
+@pytest.mark.parametrize("cfg,want", [
+    ({"port": 636}, (True, False)),
+    ({"port": 636, "tls": True}, (True, False)),
+    ({"port": 389, "tls": True}, (False, True)),
+    ({"port": 389}, (False, False)),
+])
+def test_transport(cfg, want):
+    assert ldap_auth.transport(cfg) == want
+
+
+class _FakeLdap:
+    """Just enough of ldap3 to record what Steward asks the directory."""
+
+    ALL = "ALL"
+    AUTO_BIND_TLS_BEFORE_BIND = "tls-before-bind"
+    AUTO_BIND_NO_TLS = "no-tls"
+
+    def __init__(self, password="pw"):
+        self.password = password
+        self.servers, self.binds, self.filters, self.tls = [], [], [], []
+        fake = self
+
+        class Connection:
+            def __init__(self, server, user, password, auto_bind):
+                fake.binds.append((user, auto_bind))
+                if user.startswith("uid=") and password != fake.password:
+                    raise RuntimeError("invalid credentials")
+                self.entries = []
+
+            def search(self, search_base, search_filter, attributes):
+                fake.filters.append(search_filter)
+                self.entries = [SimpleNamespace(entry_dn="uid=bob,ou=people,dc=x", uid=["bob"],
+                                                mail=["bob@x"], memberOf=["cn=ops,dc=x"])]
+
+            def unbind(self):
+                pass
+
+        self.Connection = Connection
+
+    def Server(self, host, port, use_ssl, tls, get_info):
+        self.servers.append({"port": port, "use_ssl": use_ssl, "tls": tls})
+        return object()
+
+    def Tls(self, validate, ca_certs_data):
+        self.tls.append((validate, ca_certs_data))
+        return "tls"
+
+
+_CFG = {"host": "ldap", "port": 389, "tls": True, "bind_dn": "cn=svc", "base_dn": "dc=x",
+        "user_filter": "(&(objectClass=person)(uid={username}))",
+        "operator_group_dn": "cn=ops,dc=x", "ca_cert": "PEM"}
+
+
+def test_ldap_starts_tls_checks_the_certificate_and_escapes_the_name(monkeypatch):
+    import ssl
+    fake = _FakeLdap()
+    monkeypatch.setitem(sys.modules, "ldap3", fake)
+    info = ldap_auth._ldap_authenticate_sync(_CFG, "bob*", "pw")
+    assert fake.filters == ["(&(objectClass=person)(uid=bob\\2a))"]
+    assert fake.servers == [{"port": 389, "use_ssl": False, "tls": "tls"}]
+    assert fake.tls == [(ssl.CERT_REQUIRED, "PEM")]
+    assert {b[1] for b in fake.binds} == {"tls-before-bind"}
+    assert info == {"subject": "uid=bob,ou=people,dc=x", "username": "bob",
+                    "email": "bob@x", "role": "operator"}
+
+
+def test_ldap_wrong_password_is_refused(monkeypatch):
+    monkeypatch.setitem(sys.modules, "ldap3", _FakeLdap(password="right"))
+    assert ldap_auth._ldap_authenticate_sync(_CFG, "bob", "wrong") is None
+
+
+@pytest.mark.asyncio
+async def test_ldap_empty_password_never_reaches_the_directory(monkeypatch):
+    fake = _FakeLdap(password="")
+    monkeypatch.setitem(sys.modules, "ldap3", fake)
+    assert await ldap_auth.ldap_authenticate(_CFG, "bob", "") is None
+    assert fake.binds == []
+
+
+def _jwt(claims: dict) -> str:
+    body = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip("=")
+    return f"e30.{body}.sig"
+
+
+@pytest.mark.parametrize("token,nonce,sub,ok", [
+    (_jwt({"nonce": "n1", "sub": "u1"}), "n1", "u1", True),
+    (_jwt({"nonce": "n2", "sub": "u1"}), "n1", "u1", False),   # another sign-in's token
+    (_jwt({"nonce": "n1", "sub": "u1"}), None, "u1", False),   # no sign-in was started
+    (_jwt({"nonce": "n1", "sub": "u2"}), "n1", "u1", False),   # token names someone else
+    ("", "n1", "u1", False),
+    (_jwt({"nonce": "n1", "sub": "u1"}), "n1", "", False),
+])
+def test_id_token_must_belong_to_this_sign_in(token, nonce, sub, ok):
+    assert (id_token_refusal(token, nonce, sub) is None) is ok
+
+
+@pytest.mark.asyncio
+async def test_local_password_does_not_sign_in_a_directory_account(client):
+    user = make_user()
+    user.auth_source = "ldap"
+    headers = await with_csrf(client)
+    with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=1)), \
+         patch("steward.auth.routes.get_user_by_username", new=AsyncMock(return_value=user)):
+        response = await client.post("/login", form={"username": "admin", "password": "password"},
+                                     headers=headers)
+    assert response.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_ldap_sign_in_shows_why_it_was_refused(app, client):
+    app.config["LDAP"] = {"enabled": True}
+    headers = await with_csrf(client)
+    info = {"subject": "uid=admin,dc=x", "username": "admin", "email": "", "role": "viewer"}
+    with patch("steward.auth.routes.get_user_count", new=AsyncMock(return_value=1)), \
+         patch("steward.auth.routes.get_user_by_username", new=AsyncMock(return_value=make_user())), \
+         patch("steward.auth.ldap_auth.ldap_authenticate", new=AsyncMock(return_value=info)), \
+         patch("steward.auth.routes._provision_external_user",
+               new=AsyncMock(return_value=(None, "Steward already has an account named admin"))):
+        response = await client.post("/login", form={"username": "admin", "password": "dir-pw"},
+                                     headers=headers)
+    assert response.status_code == 403
+    assert "already has an account named admin" in await response.get_data(as_text=True)
+    async with client.session_transaction() as sess:
+        assert "user_id" not in sess