Thomas Stallinger
ec11de1ee9
Neue Rolle ist_organisationsadmin (auth.py), bewusst getrennt von is_superuser: is_superuser bleibt exklusiv internes, organisationsuebergreifendes Tuxflotte-Personal mit Zugriff auf /admin/* - ist_organisationsadmin ist auf die eigene Organisation beschränkt und kann nie über Self-Service gesetzt werden (organization_id/is_superuser kommen bei POST /organisation/konten nie aus dem Formular, sondern immer aus der Session). /organisation zeigt jetzt Kontakt-/Stammdaten (bearbeitbar) und die Kolleg:innen-Liste der eigenen Organisation (für alle sichtbar, Verwaltung nur für Org-Admins/internes Personal via require_org_admin). Selbstlöschung des eigenen Kontos wird verweigert. fetch_benutzer_fuer_organisation() von admin_kunden.py nach auth.py verschoben (jetzt von beiden Routern importiert, kein Duplikat) - liefert zusätzlich id + ist_organisationsadmin. admin_kunden.py bekommt eine neue Organisationsdaten-Sektion (nutzt denselben PATCH-Endpoint) und die neue Checkbox beim Konto-Anlegen, um den ersten Org-Admin einer neuen Organisation zu bootstrappen. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
274 lines
8.7 KiB
Python
274 lines
8.7 KiB
Python
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, get_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, "
|
|
"ist_organisationsadmin, 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
|
|
ist_organisationsadmin: bool = False
|
|
is_verified: bool = True
|
|
|
|
|
|
class UserCreate(schemas.BaseUserCreate):
|
|
organization_id: uuid.UUID
|
|
ist_organisationsadmin: Optional[bool] = False
|
|
is_verified: Optional[bool] = True
|
|
|
|
|
|
def _row_to_user(row) -> User:
|
|
(
|
|
user_id, organization_id, email, hashed_password, is_active,
|
|
is_superuser, ist_organisationsadmin, 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,
|
|
ist_organisationsadmin=ist_organisationsadmin,
|
|
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,
|
|
ist_organisationsadmin, is_verified)
|
|
VALUES (%s, %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("ist_organisationsadmin", 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
|
|
|
|
|
|
async def require_org_admin(user: User = Depends(get_current_user)) -> User:
|
|
"""
|
|
Für Self-Service-Kontenverwaltung (/organisation/konten): erlaubt
|
|
entweder is_superuser (internes Tuxflotte-Personal, soll nicht
|
|
ausgesperrt sein, falls Support nötig ist) oder ist_organisationsadmin
|
|
(die neue, auf die eigene Organisation beschränkte Rolle). Bewusst kein
|
|
Ersatz für require_admin - is_superuser bleibt exklusiv für /admin/*.
|
|
"""
|
|
|
|
if not (user.is_superuser or user.ist_organisationsadmin):
|
|
raise HTTPException(status_code=403, detail="Kein Admin-Zugriff für diese Organisation.")
|
|
|
|
return user
|
|
|
|
|
|
def fetch_benutzer_fuer_organisation(organization_id: str) -> list[dict]:
|
|
"""
|
|
Kontenliste einer Organisation - gemeinsam genutzt vom internen
|
|
Admin-Bereich (admin_kunden.py) und der Self-Service-Seite
|
|
(routers/organisation.py), um die SQL-Abfrage nicht zu duplizieren
|
|
(siehe ADR-0017 fetch_gruppen()-Reuse-Pattern).
|
|
"""
|
|
|
|
with get_database_connection() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, email, is_superuser, ist_organisationsadmin, created_at
|
|
FROM benutzer WHERE organization_id = %s ORDER BY email
|
|
""",
|
|
(organization_id,),
|
|
)
|
|
return [
|
|
{
|
|
"id": str(benutzer_id),
|
|
"email": email,
|
|
"is_superuser": is_superuser,
|
|
"ist_organisationsadmin": ist_organisationsadmin,
|
|
"created_at": created_at,
|
|
}
|
|
for benutzer_id, email, is_superuser, ist_organisationsadmin, created_at in cur.fetchall()
|
|
]
|
|
|
|
|
|
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
|