kundenplattform/auth.py

import os
import uuid
from dataclasses import dataclass
from types import SimpleNamespace
from typing import Optional

from fastapi import APIRouter, Depends, Form, HTTPException, Request
from fastapi.responses import RedirectResponse
from fastapi_users import BaseUserManager, FastAPIUsers, UUIDIDMixin, schemas
from fastapi_users.authentication import AuthenticationBackend, CookieTransport, JWTStrategy
from fastapi_users.db import BaseUserDatabase

from db import get_async_database_connection
from templating import templates


SESSION_SECRET = os.environ.get("KUNDENPLATTFORM_SESSION_SECRET", "")

COOKIE_NAME = "kundenplattform_session"
COOKIE_MAX_AGE = 14 * 24 * 60 * 60  # 14 Tage, wie zuvor Starlettes SessionMiddleware-Default

BENUTZER_SPALTEN = "id, organization_id, email, hashed_password, is_active, is_superuser, is_verified"


@dataclass
class User:
    id: uuid.UUID
    email: str
    hashed_password: str
    organization_id: uuid.UUID
    is_active: bool = True
    is_superuser: bool = False
    is_verified: bool = True


class UserCreate(schemas.BaseUserCreate):
    organization_id: uuid.UUID
    is_verified: Optional[bool] = True


def _row_to_user(row) -> User:
    user_id, organization_id, email, hashed_password, is_active, is_superuser, is_verified = row

    # psycopg liefert TEXT-Spalten hier teils als bytes zurück, nicht als str.
    if isinstance(hashed_password, bytes):
        hashed_password = hashed_password.decode("utf-8")
    if isinstance(email, bytes):
        email = email.decode("utf-8")

    return User(
        id=user_id,
        organization_id=organization_id,
        email=email,
        hashed_password=hashed_password,
        is_active=is_active,
        is_superuser=is_superuser,
        is_verified=is_verified,
    )


class UserDatabase(BaseUserDatabase[User, uuid.UUID]):
    def __init__(self, connection):
        self.connection = connection

    async def get(self, id: uuid.UUID) -> Optional[User]:
        async with self.connection.cursor() as cur:
            await cur.execute(f"SELECT {BENUTZER_SPALTEN} FROM benutzer WHERE id = %s", (id,))
            row = await cur.fetchone()

        return _row_to_user(row) if row else None

    async def get_by_email(self, email: str) -> Optional[User]:
        async with self.connection.cursor() as cur:
            await cur.execute(f"SELECT {BENUTZER_SPALTEN} FROM benutzer WHERE email = %s", (email,))
            row = await cur.fetchone()

        return _row_to_user(row) if row else None

    async def get_by_oauth_account(self, oauth: str, account_id: str) -> Optional[User]:
        raise NotImplementedError("OAuth wird nicht unterstützt.")

    async def create(self, create_dict: dict) -> User:
        user_id = uuid.uuid4()

        async with self.connection.cursor() as cur:
            await cur.execute(
                """
                INSERT INTO benutzer
                    (id, organization_id, email, hashed_password, is_active, is_superuser, is_verified)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                """,
                (
                    user_id,
                    create_dict["organization_id"],
                    create_dict["email"],
                    create_dict["hashed_password"],
                    create_dict.get("is_active", True),
                    create_dict.get("is_superuser", False),
                    create_dict.get("is_verified", True),
                ),
            )

        return await self.get(user_id)

    async def update(self, user: User, update_dict: dict) -> User:
        felder = ", ".join(f"{spalte} = %s" for spalte in update_dict)
        werte = [*update_dict.values(), user.id]

        async with self.connection.cursor() as cur:
            await cur.execute(f"UPDATE benutzer SET {felder} WHERE id = %s", werte)

        return await self.get(user.id)

    async def delete(self, user: User) -> None:
        async with self.connection.cursor() as cur:
            await cur.execute("DELETE FROM benutzer WHERE id = %s", (user.id,))


async def get_user_db():
    async with await get_async_database_connection() as connection:
        yield UserDatabase(connection)


class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
    reset_password_token_secret = SESSION_SECRET
    verification_token_secret = SESSION_SECRET


async def get_user_manager(user_db: UserDatabase = Depends(get_user_db)):
    yield UserManager(user_db)


cookie_transport = CookieTransport(
    cookie_name=COOKIE_NAME,
    cookie_max_age=COOKIE_MAX_AGE,
    cookie_secure=True,
    cookie_httponly=True,
    cookie_samesite="lax",
)


def get_jwt_strategy() -> JWTStrategy:
    return JWTStrategy(secret=SESSION_SECRET, lifetime_seconds=COOKIE_MAX_AGE)


auth_backend = AuthenticationBackend(
    name="jwt-cookie",
    transport=cookie_transport,
    get_strategy=get_jwt_strategy,
)

fastapi_users = FastAPIUsers[User, uuid.UUID](get_user_manager, [auth_backend])

_current_user_optional = fastapi_users.current_user(active=True, optional=True)


async def get_current_user(user: Optional[User] = Depends(_current_user_optional)) -> User:
    if user is None:
        raise HTTPException(status_code=303, headers={"Location": "/login"})

    return user


async def require_admin(user: User = Depends(get_current_user)) -> User:
    if not user.is_superuser:
        raise HTTPException(status_code=403, detail="Kein Admin-Zugriff.")

    return user


router = APIRouter()


@router.get("/login")
async def login_form(request: Request, user: Optional[User] = Depends(_current_user_optional)):
    if user is not None:
        return RedirectResponse("/geraete", status_code=303)

    return templates.TemplateResponse(request, "login.html", {"error": None})


@router.post("/login")
async def login_submit(
    request: Request,
    email: str = Form(...),
    password: str = Form(...),
    user_manager: UserManager = Depends(get_user_manager),
):
    user = await user_manager.authenticate(SimpleNamespace(username=email, password=password))

    if user is None or not user.is_active:
        return templates.TemplateResponse(
            request, "login.html", {"error": "E-Mail oder Passwort ist falsch."}, status_code=401
        )

    strategy = get_jwt_strategy()
    token = await strategy.write_token(user)

    response = RedirectResponse("/geraete", status_code=303)
    response.set_cookie(
        COOKIE_NAME,
        token,
        max_age=COOKIE_MAX_AGE,
        path="/",
        secure=True,
        httponly=True,
        samesite="lax",
    )

    return response


@router.get("/logout")
def logout():
    response = RedirectResponse("/login", status_code=303)
    response.delete_cookie(COOKIE_NAME, path="/")

    return response