Oliver Hofmann 361ba93afc docs: add auth part 2 LDAP implementation plan

2026-04-27 18:32:26 +02:00

24 KiB

Raw Blame History

Auth Part 2: LDAP Integration Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: LDAP-Authentifizierung gegen das Active Directory der TH Nürnberg als Fallback zur lokalen Auth, mit login-getrigentem Background-Sync aller AD-User.

Architecture: Neue Datei app/modules/auth/ldap.py enthält alle ldap3-Aufrufe (kein FastAPI, keine DB-Abhängigkeiten außer in sync_all_users). service.py füllt den bestehenden LDAP-Stub und legt User bei LDAP-Erfolg an. router.py startet den Sync als FastAPI BackgroundTask. sync_all_users erzeugt seine eigene DB-Session (long-running task).

Tech Stack: ldap3, FastAPI BackgroundTasks, SQLAlchemy, pytest mit unittest.mock

Task 1: Config-Erweiterungen

Files:

Modify: app/core/config.py
Modify: .env.example
Modify: tests/test_config.py
Step 1: Update app/core/config.py

from functools import lru_cache

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

    APP_ENV: str = "development"
    APP_PREFIX: str = ""
    DATABASE_URL: str = "sqlite:///./app.db"
    SECRET_KEY: str = "changeme-replace-in-production"
    LDAP_ENABLED: bool = False
    LDAP_SERVER: str = "gso1.ads1.fh-nuernberg.de"
    LDAP_DOMAIN: str = "ADS1"
    LDAP_SEARCH_BASE: str = "OU=users,OU=EFI,OU=Faculties,DC=ADS1,DC=fh-nuernberg,DC=de"
    LDAP_SYNC_MIN_INTERVAL_HOURS: int = 12
    LDAP_SYNC_LETTER_DELAY_SECONDS: float = 5.0
    ADMIN_USERNAME: str = "admin"
    ADMIN_PASSWORD: str = "change_me"


@lru_cache
def get_settings() -> Settings:
    return Settings()

Step 2: Append to .env.example

Add these two lines in the development section (before the Produktion comment):

LDAP_SYNC_MIN_INTERVAL_HOURS=12
LDAP_SYNC_LETTER_DELAY_SECONDS=5.0

Step 3: Add assertions to tests/test_config.py

Append to the existing test_new_config_fields_have_defaults function:

    assert s.LDAP_SYNC_MIN_INTERVAL_HOURS == 12
    assert s.LDAP_SYNC_LETTER_DELAY_SECONDS == 5.0

Step 4: Run test

.venv/bin/pytest tests/test_config.py -v

Expected: PASS.

Step 5: Commit

git add app/core/config.py .env.example tests/test_config.py
git commit -m "feat: add LDAP sync interval config fields"

Task 2: `app/modules/auth/ldap.py`

Files:

Create: app/modules/auth/ldap.py
Create: tests/test_ldap.py
Step 1: Write failing tests

Create tests/test_ldap.py:

import time
from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock, patch

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.core.database import Base
from app.modules.auth.ldap import (
    _has_to_update,
    _parse_entry,
    _upsert_from_attrs,
    ldap_authenticate,
)
from app.modules.auth.models import User


# --- Helpers ---

def _make_entry(**kwargs):
    defaults = {
        "sAMAccountName": "hofmannol",
        "givenName": "Oliver",
        "sn": "Hofmann",
        "department": "EFI",
        "description": "PF",
        "accountExpires": None,
    }
    defaults.update(kwargs)

    class _Attr:
        def __init__(self, value):
            self.value = value

    class _Entry:
        def __getitem__(self, key):
            return _Attr(defaults.get(key))

    return _Entry()


@pytest.fixture
def db():
    engine = create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.close()
    Base.metadata.drop_all(bind=engine)


# --- _parse_entry ---

def test_parse_entry_basic():
    result = _parse_entry(_make_entry())
    assert result["username"] == "hofmannol"
    assert result["full_name"] == "Oliver Hofmann"
    assert result["department"] == "EFI"
    assert result["role"] == "PF"
    assert result["account_expires"] is None


def test_parse_entry_truncates_long_fields():
    result = _parse_entry(_make_entry(department="A" * 100))
    assert len(result["department"]) == 64


def test_parse_entry_uses_nn_for_none_fields():
    result = _parse_entry(_make_entry(department=None, description=None))
    assert result["department"] == "N.N."
    assert result["role"] == "N.N."


def test_parse_entry_handles_none_account_expires():
    result = _parse_entry(_make_entry(accountExpires=None))
    assert result["account_expires"] is None


# --- _has_to_update ---

def test_has_to_update_true_when_no_ldap_users(db):
    assert _has_to_update(db, 12) is True


def test_has_to_update_false_when_recent_ldap_user(db):
    user = User(
        username="u", full_name="U", pw_hash=None,
        updated_at=datetime.now(timezone.utc),
    )
    db.add(user)
    db.commit()
    assert _has_to_update(db, 12) is False


def test_has_to_update_true_when_old_ldap_user(db):
    old_time = datetime.now(timezone.utc) - timedelta(hours=25)
    user = User(username="u", full_name="U", pw_hash=None, updated_at=old_time)
    db.add(user)
    db.commit()
    assert _has_to_update(db, 12) is True


def test_has_to_update_ignores_local_users(db):
    # Local user (pw_hash set) should NOT count for sync timing
    user = User(
        username="admin", full_name="Admin", pw_hash="hash",
        updated_at=datetime.now(timezone.utc),
    )
    db.add(user)
    db.commit()
    assert _has_to_update(db, 12) is True


# --- _upsert_from_attrs ---

def test_upsert_creates_new_user(db):
    attrs = {
        "username": "newuser",
        "full_name": "New User",
        "department": "EFI",
        "role": "ST",
        "account_expires": None,
    }
    _upsert_from_attrs(db, attrs)
    db.commit()
    user = db.query(User).filter(User.username == "newuser").first()
    assert user is not None
    assert user.full_name == "New User"
    assert user.pw_hash is None


def test_upsert_updates_existing_user(db):
    user = User(username="existing", full_name="Old Name", pw_hash=None)
    db.add(user)
    db.commit()
    attrs = {
        "username": "existing",
        "full_name": "New Name",
        "department": "EFI",
        "role": "PF",
        "account_expires": None,
    }
    _upsert_from_attrs(db, attrs)
    db.commit()
    db.refresh(user)
    assert user.full_name == "New Name"
    assert user.role == "PF"


# --- ldap_authenticate ---

@patch("app.modules.auth.ldap.Server")
@patch("app.modules.auth.ldap.Connection")
def test_ldap_authenticate_success(mock_conn_cls, mock_server_cls):
    mock_conn = MagicMock()
    mock_conn_cls.return_value = mock_conn
    mock_conn.extend.standard.who_am_i.return_value = "u:ADS1\\hofmannol"
    mock_conn.entries = [_make_entry()]

    result = ldap_authenticate("hofmannol", "password", "server.example", "ADS1")

    assert result is not None
    assert result["username"] == "hofmannol"
    assert result["full_name"] == "Oliver Hofmann"
    mock_conn.bind.assert_called_once()
    mock_conn.unbind.assert_called_once()


@patch("app.modules.auth.ldap.Server")
@patch("app.modules.auth.ldap.Connection")
def test_ldap_authenticate_returns_none_on_socket_error(mock_conn_cls, mock_server_cls):
    from ldap3.core.exceptions import LDAPSocketOpenError
    mock_conn = MagicMock()
    mock_conn_cls.return_value = mock_conn
    mock_conn.bind.side_effect = LDAPSocketOpenError("unreachable")

    result = ldap_authenticate("hofmannol", "pw", "server.example", "ADS1")
    assert result is None


@patch("app.modules.auth.ldap.Server")
@patch("app.modules.auth.ldap.Connection")
def test_ldap_authenticate_returns_none_when_who_am_i_empty(mock_conn_cls, mock_server_cls):
    mock_conn = MagicMock()
    mock_conn_cls.return_value = mock_conn
    mock_conn.extend.standard.who_am_i.return_value = None

    result = ldap_authenticate("hofmannol", "wrong", "server.example", "ADS1")
    assert result is None
    mock_conn.unbind.assert_called_once()


@patch("app.modules.auth.ldap.Server")
@patch("app.modules.auth.ldap.Connection")
def test_ldap_authenticate_returns_none_when_no_entries(mock_conn_cls, mock_server_cls):
    mock_conn = MagicMock()
    mock_conn_cls.return_value = mock_conn
    mock_conn.extend.standard.who_am_i.return_value = "u:ADS1\\hofmannol"
    mock_conn.entries = []

    result = ldap_authenticate("hofmannol", "pw", "server.example", "ADS1")
    assert result is None

Step 2: Run to confirm failure

.venv/bin/pytest tests/test_ldap.py -v

Expected: FAIL — ModuleNotFoundError: No module named 'app.modules.auth.ldap'

Step 3: Create app/modules/auth/ldap.py

import logging
import time
from datetime import datetime, timedelta, timezone
from typing import Optional

from ldap3 import ALL, ALL_ATTRIBUTES, NTLM, Connection, Server
from ldap3.core.exceptions import LDAPSocketOpenError
from sqlalchemy.orm import Session

from app.modules.auth.models import User

logger = logging.getLogger(__name__)


def ldap_authenticate(
    username: str,
    password: str,
    ldap_server: str,
    ldap_domain: str,
) -> Optional[dict]:
    """NTLM bind + fetch user attributes. Returns attrs dict on success, None on failure."""
    server = Server(ldap_server, get_info=ALL, connect_timeout=8)
    conn = Connection(
        server,
        user=f"{ldap_domain}\\{username}",
        password=password,
        authentication=NTLM,
    )
    try:
        conn.bind()
    except LDAPSocketOpenError:
        logger.warning("LDAP server %s not reachable", ldap_server)
        return None
    except Exception:
        logger.warning("LDAP bind failed for user %s", username, exc_info=True)
        return None

    if not conn.extend.standard.who_am_i():
        conn.unbind()
        return None

    conn.search(
        search_base=f"DC={ldap_domain},DC=fh-nuernberg,DC=de",
        search_filter=f"(&(objectclass=user)(CN={username}))",
        attributes=ALL_ATTRIBUTES,
    )

    if not conn.entries:
        conn.unbind()
        logger.warning("LDAP: user %s authenticated but no entry found", username)
        return None

    attrs = _parse_entry(conn.entries[0])
    conn.unbind()
    return attrs


def _parse_entry(entry) -> dict:
    def val(field: str, default: str = "N.N.") -> str:
        try:
            v = entry[field].value
            return str(v)[:64] if v else default
        except Exception:
            return default

    def expire_date(field: str) -> Optional[datetime]:
        try:
            v = entry[field].value
            if v is None:
                return None
            if isinstance(v, datetime):
                dt = v
            else:
                dt = datetime(v.year, v.month, v.day, tzinfo=timezone.utc)
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=timezone.utc)
            return None if dt.year >= 2099 else dt
        except Exception:
            return None

    return {
        "username": val("sAMAccountName"),
        "full_name": (f"{val('givenName', '')} {val('sn', '')}".strip() or "N.N."),
        "department": val("department"),
        "role": val("description"),
        "account_expires": expire_date("accountExpires"),
    }


def _has_to_update(db: Session, min_interval_hours: int) -> bool:
    """True if a background sync is due."""
    deadline = datetime.now(timezone.utc) - timedelta(hours=min_interval_hours)
    latest = (
        db.query(User)
        .filter(User.pw_hash.is_(None))
        .order_by(User.updated_at.desc())
        .first()
    )
    if latest is None:
        return True
    last = latest.updated_at
    if last.tzinfo is None:
        last = last.replace(tzinfo=timezone.utc)
    return last < deadline


def _upsert_from_attrs(db: Session, attrs: dict) -> None:
    """Create or update a User from AD attribute dict (no commit)."""
    user = db.query(User).filter(User.username == attrs["username"]).first()
    if user is None:
        user = User(username=attrs["username"])
        db.add(user)
    user.full_name = attrs["full_name"]
    user.department = attrs["department"]
    user.role = attrs["role"]
    user.account_expires = attrs["account_expires"]


def sync_all_users(
    username: str,
    password: str,
    ldap_server: str,
    ldap_domain: str,
    ldap_search_base: str,
    min_interval_hours: int,
    letter_delay_seconds: float,
) -> None:
    """Login-triggered background sync of all AD users. Creates its own DB session."""
    from app.core.database import SessionLocal

    with SessionLocal() as db:
        if not _has_to_update(db, min_interval_hours):
            logger.info("LDAP sync skipped: last sync is recent")
            return

        server = Server(ldap_server, get_info=ALL, connect_timeout=8)
        conn = Connection(
            server,
            user=f"{ldap_domain}\\{username}",
            password=password,
            authentication=NTLM,
        )
        try:
            conn.bind()
        except Exception:
            logger.warning("LDAP sync: bind failed", exc_info=True)
            return

        found_usernames: set[str] = set()
        chars = "abcdefghijklmnopqrstuvwxyz_"

        for char in chars:
            try:
                conn.search(
                    search_base=ldap_search_base,
                    search_filter=f"(&(objectclass=user)(CN={char}*))",
                    attributes=ALL_ATTRIBUTES,
                )
                for entry in conn.entries:
                    try:
                        attrs = _parse_entry(entry)
                        found_usernames.add(attrs["username"])
                        _upsert_from_attrs(db, attrs)
                    except Exception:
                        logger.warning("LDAP sync: failed to parse entry", exc_info=True)
                db.commit()
            except Exception:
                logger.warning("LDAP sync: search failed for prefix '%s'", char, exc_info=True)

            if letter_delay_seconds > 0:
                time.sleep(letter_delay_seconds)

        # Deactivate LDAP users no longer in AD
        ldap_users = db.query(User).filter(User.pw_hash.is_(None)).all()
        for user_obj in ldap_users:
            if user_obj.username not in found_usernames:
                logger.info("LDAP sync: deactivating %s (not in AD)", user_obj.username)
                user_obj.is_active = False
        db.commit()
        conn.unbind()
        logger.info("LDAP sync complete — %d users found", len(found_usernames))

Step 4: Run to confirm pass

.venv/bin/pytest tests/test_ldap.py -v

Expected: All 14 tests PASS.

Step 5: Run full suite

.venv/bin/pytest -v

Expected: All tests PASS.

Step 6: Commit

git add app/modules/auth/ldap.py tests/test_ldap.py
git commit -m "feat: add LDAP functions (authenticate, sync, parse, upsert)"

Task 3: service.py — LDAP-Auth-Integration

Files:

Modify: app/modules/auth/service.py
Modify: tests/test_auth_service.py
Step 1: Append LDAP tests to tests/test_auth_service.py

# --- LDAP auth integration (uses mocked ldap_authenticate) ---

from unittest.mock import patch as mock_patch


def test_authenticate_creates_user_on_ldap_success(db):
    ldap_attrs = {
        "username": "newldap",
        "full_name": "New LDAP User",
        "department": "EFI",
        "role": "ST",
        "account_expires": None,
    }
    with mock_patch("app.modules.auth.ldap.ldap_authenticate", return_value=ldap_attrs):
        user = authenticate_user(db, "newldap", "password", ldap_enabled=True)
    assert user is not None
    assert user.username == "newldap"
    assert user.full_name == "New LDAP User"
    assert user.last_login is not None


def test_authenticate_updates_existing_user_on_ldap_success(db):
    existing = User(username="existing", full_name="Old Name", pw_hash=None)
    db.add(existing)
    db.commit()
    ldap_attrs = {
        "username": "existing",
        "full_name": "New Name",
        "department": "EFI",
        "role": "PF",
        "account_expires": None,
    }
    with mock_patch("app.modules.auth.ldap.ldap_authenticate", return_value=ldap_attrs):
        user = authenticate_user(db, "existing", "password", ldap_enabled=True)
    assert user.full_name == "New Name"


def test_authenticate_returns_none_on_ldap_failure(db):
    with mock_patch("app.modules.auth.ldap.ldap_authenticate", return_value=None):
        result = authenticate_user(db, "anyone", "wrong", ldap_enabled=True)
    assert result is None


def test_authenticate_blocked_user_not_bypassed_by_ldap(db):
    blocked = User(username="blocked", full_name="B", pw_hash=None, is_active=False)
    db.add(blocked)
    db.commit()
    ldap_attrs = {"username": "blocked", "full_name": "B", "department": "", "role": "", "account_expires": None}
    with mock_patch("app.modules.auth.ldap.ldap_authenticate", return_value=ldap_attrs):
        result = authenticate_user(db, "blocked", "any", ldap_enabled=True)
    assert result is None

Step 2: Run to confirm new tests fail

.venv/bin/pytest tests/test_auth_service.py -v -k "ldap"

Expected: FAIL — authenticate_user LDAP stub is still pass.

Step 3: Replace app/modules/auth/service.py

from datetime import datetime, timezone
from typing import Optional

from passlib.context import CryptContext
from sqlalchemy.orm import Session

from app.modules.auth.models import User

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def hash_password(plain: str) -> str:
    return pwd_context.hash(plain)


def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)


def get_user(db: Session, username: str) -> Optional[User]:
    return db.query(User).filter(User.username == username).first()


def upsert_ldap_user(db: Session, attrs: dict) -> User:
    """Create or update a User from LDAP attribute dict, commit and return."""
    from app.modules.auth.ldap import _upsert_from_attrs
    _upsert_from_attrs(db, attrs)
    db.commit()
    return get_user(db, attrs["username"])


def authenticate_user(
    db: Session, username: str, password: str, ldap_enabled: bool
) -> Optional[User]:
    user = get_user(db, username)

    # Explicitly deactivated users are always blocked
    if user is not None and not user.is_active:
        return None

    # Local auth
    if user is not None and user.pw_hash is not None:
        if verify_password(password, user.pw_hash):
            _touch_last_login(db, user)
            return user

    # LDAP auth
    if ldap_enabled:
        from app.core.config import get_settings
        from app.modules.auth.ldap import ldap_authenticate
        s = get_settings()
        attrs = ldap_authenticate(username, password, s.LDAP_SERVER, s.LDAP_DOMAIN)
        if attrs is None:
            return None
        user = upsert_ldap_user(db, attrs)
        _touch_last_login(db, user)
        return user

    return None


def _touch_last_login(db: Session, user: User) -> None:
    user.last_login = datetime.now(timezone.utc)
    db.commit()

Step 4: Run to confirm all service tests pass

.venv/bin/pytest tests/test_auth_service.py -v

Expected: All 12 tests PASS (8 existing + 4 new).

Step 5: Run full suite

.venv/bin/pytest -v

Expected: All tests PASS.

Step 6: Commit

git add app/modules/auth/service.py tests/test_auth_service.py
git commit -m "feat: implement LDAP auth fallback in authenticate_user"

Task 4: router.py — BackgroundTask für Sync

Files:

Modify: app/modules/auth/router.py
Modify: tests/test_auth_router.py
Step 1: Add LDAP login test to tests/test_auth_router.py

Append this test to the existing file:

from unittest.mock import patch as mock_patch


def test_ldap_login_sets_cookie_and_redirects(client, override_db):
    """LDAP login via mocked ldap_authenticate: cookie is set, redirects to /."""
    ldap_attrs = {
        "username": "ldapuser",
        "full_name": "LDAP User",
        "department": "EFI",
        "role": "ST",
        "account_expires": None,
    }
    with mock_patch("app.modules.auth.ldap.ldap_authenticate", return_value=ldap_attrs), \
         mock_patch("app.modules.auth.ldap.sync_all_users"), \
         mock_patch("app.modules.auth.router.settings") as mock_settings:
        mock_settings.LDAP_ENABLED = True
        mock_settings.LDAP_SERVER = "test"
        mock_settings.LDAP_DOMAIN = "ADS1"
        mock_settings.LDAP_SEARCH_BASE = "OU=test"
        mock_settings.LDAP_SYNC_MIN_INTERVAL_HOURS = 12
        mock_settings.LDAP_SYNC_LETTER_DELAY_SECONDS = 0.0
        response = client.post("/auth/login", data={"username": "ldapuser", "password": "any"})
    assert response.status_code in (302, 303, 307)
    assert "access_token" in response.cookies

Step 2: Run to confirm failure

.venv/bin/pytest tests/test_auth_router.py::test_ldap_login_sets_cookie_and_redirects -v

Expected: FAIL — Router importiert sync_all_users und BackgroundTasks noch nicht.

Step 3: Replace app/modules/auth/router.py

from fastapi import APIRouter, BackgroundTasks, Depends, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session

from app.core.auth import clear_auth_cookie, create_access_token, set_auth_cookie
from app.core.config import get_settings
from app.core.database import get_db
from app.modules.auth.dependencies import get_current_user
from app.modules.auth.ldap import sync_all_users
from app.modules.auth.schemas import UserOut
from app.modules.auth.service import authenticate_user

router = APIRouter(prefix="/auth", tags=["auth"])
templates = Jinja2Templates(directory="app/templates")
settings = get_settings()

_NAV: list[dict] = []


@router.get("/login", response_class=HTMLResponse)
async def login_page(request: Request):
    return templates.TemplateResponse(
        request, "auth/login.html", {"nav_items": _NAV, "app_version": "0.1.0"}
    )


@router.post("/login", response_class=HTMLResponse)
async def login(
    request: Request,
    background_tasks: BackgroundTasks,
    username: str = Form(...),
    password: str = Form(...),
    db: Session = Depends(get_db),
):
    user = authenticate_user(db, username, password, ldap_enabled=settings.LDAP_ENABLED)
    if user is None:
        return templates.TemplateResponse(
            request,
            "auth/login.html",
            {"nav_items": _NAV, "app_version": "0.1.0", "error": "Ungültige Zugangsdaten."},
            status_code=200,
        )
    if settings.LDAP_ENABLED:
        background_tasks.add_task(
            sync_all_users,
            username,
            password,
            settings.LDAP_SERVER,
            settings.LDAP_DOMAIN,
            settings.LDAP_SEARCH_BASE,
            settings.LDAP_SYNC_MIN_INTERVAL_HOURS,
            settings.LDAP_SYNC_LETTER_DELAY_SECONDS,
        )
    token = create_access_token(username=user.username, is_admin=user.is_admin)
    response = RedirectResponse(url="/", status_code=303)
    set_auth_cookie(response, token)
    return response


@router.get("/logout")
async def logout():
    response = RedirectResponse(url="/", status_code=303)
    clear_auth_cookie(response)
    return response


@router.get("/me", response_model=UserOut)
async def me(user=Depends(get_current_user)):
    return user

Step 4: Run all router tests

.venv/bin/pytest tests/test_auth_router.py -v

Expected: All tests PASS.

Step 5: Run full suite

.venv/bin/pytest -v

Expected: All tests PASS.

Step 6: Commit

git add app/modules/auth/router.py tests/test_auth_router.py
git commit -m "feat: trigger LDAP background sync on successful login"

24 KiB Raw Blame History

Auth Part 2: LDAP Integration Implementation Plan

Task 1: Config-Erweiterungen

Task 2: app/modules/auth/ldap.py

Task 3: service.py — LDAP-Auth-Integration

Task 4: router.py — BackgroundTask für Sync

24 KiB

Raw Blame History

Task 2: `app/modules/auth/ldap.py`