""" Autentisering: magic-link-forespørsel/-verifisering, utlogging, "hvem er jeg". Sikkerhetsmønster (se plan/ADR-009): - request-link svarer ALLTID identisk, uansett om e-posten finnes eller nettopp fikk en lenke -- unngår at endepunktet kan brukes til å sjekke hvilke e-poster som har konto (enumerering). - Kun SHA-256-hashen av token lagres, aldri klarteksten. - app_user opprettes FØRST når en gyldig, uforbrukt token løses inn i verify-link -- IKKE når lenken bare forespørres (hindrer massopprettelse av kontoer for e-poster man ikke eier). - Token-forbruk er ÉN atomisk UPDATE ... RETURNING (ikke les-sjekk-skriv) for å hindre at to samtidige forsøk med samme lenke begge lykkes. - Dev-modus: siden ingen ekte SMTP-oppsett finnes for TeeCup ennå, logges rå token server-side i stedet for å sendes på e-post -- KUN når settings.DEV_LOG_MAGIC_LINKS er eksplisitt satt (aldri på som standard). # TODO: erstatt med ekte SMTP-utsending før produksjon. """ import hashlib import secrets from datetime import datetime, timedelta, timezone from fastapi import APIRouter, Depends, HTTPException, Request, Response, status from pydantic import BaseModel, EmailStr from ..auth import CurrentUser, SESSION_COOKIE_NAME, create_session_token, get_current_user, should_use_secure_cookies from ..config import settings from ..db import org_connection, plain_connection router = APIRouter(prefix="/auth", tags=["auth"]) _GENERIC_RESPONSE = { "status": "ok", "detail": "Hvis e-posten er gyldig, er en innloggingslenke sendt.", } def _hash_token(token: str) -> str: return hashlib.sha256(token.encode("utf-8")).hexdigest() class MagicLinkRequest(BaseModel): email: EmailStr @router.post("/request-link") async def request_magic_link(body: MagicLinkRequest) -> dict: email = body.email.lower() now = datetime.now(timezone.utc) cooldown_cutoff = now - timedelta(seconds=settings.MAGIC_LINK_COOLDOWN_SECONDS) async with plain_connection() as conn: recent = await conn.fetchval( """ SELECT EXISTS ( SELECT 1 FROM magic_link_token WHERE email = $1 AND consumed_at IS NULL AND created_at > $2 ) """, email, cooldown_cutoff, ) if not recent: # Ugyldiggjør tidligere uforbrukte lenker for denne e-posten -- # begrenser vinduet en eldre, lekket lenke kan misbrukes i. await conn.execute( "UPDATE magic_link_token SET consumed_at = now() WHERE email = $1 AND consumed_at IS NULL", email, ) raw_token = secrets.token_urlsafe(32) expires_at = now + timedelta(minutes=settings.MAGIC_LINK_MAX_AGE_MINUTES) await conn.execute( "INSERT INTO magic_link_token (email, token_hash, expires_at) VALUES ($1, $2, $3)", email, _hash_token(raw_token), expires_at, ) if settings.DEV_LOG_MAGIC_LINKS: print(f"[DEV] Magic link for {email}: {raw_token}", flush=True) return _GENERIC_RESPONSE class MagicLinkVerify(BaseModel): token: str display_name: str | None = None class SessionUser(BaseModel): id: str email: str display_name: str @router.post("/verify-link", response_model=SessionUser) async def verify_magic_link( body: MagicLinkVerify, response: Response, request: Request ) -> SessionUser: token_hash = _hash_token(body.token) async with plain_connection() as conn: # Atomisk forbruk: ÉN setning, ikke les-så-sjekk-så-skriv -- to # samtidige forsøk med samme lenke kan da ikke begge lykkes. email = await conn.fetchval( """ UPDATE magic_link_token SET consumed_at = now() WHERE token_hash = $1 AND consumed_at IS NULL AND expires_at > now() RETURNING email """, token_hash, ) if email is None: raise HTTPException( status.HTTP_401_UNAUTHORIZED, detail="Lenken er ugyldig, brukt eller utløpt." ) placeholder_name = email.split("@")[0] user_id = await conn.fetchval( """ INSERT INTO app_user (email, display_name) VALUES ($1, $2) ON CONFLICT (email) WHERE email IS NOT NULL DO NOTHING RETURNING id """, email, body.display_name or placeholder_name, ) if user_id is None: # En annen samtidig verifisering for samme e-post vant innsettingen. user_id = await conn.fetchval("SELECT id FROM app_user WHERE email = $1", email) user_row = await conn.fetchrow( "SELECT id::text AS id, email::text AS email, display_name FROM app_user WHERE id = $1", user_id, ) token = create_session_token(user_row["id"]) response.set_cookie( SESSION_COOKIE_NAME, token, max_age=settings.SESSION_MAX_AGE_SECONDS, httponly=True, samesite="lax", secure=should_use_secure_cookies(request), path="/", ) return SessionUser(**dict(user_row)) @router.post("/logout") async def logout(response: Response) -> dict: response.delete_cookie(SESSION_COOKIE_NAME, path="/") return {"status": "ok"} class MyOrg(BaseModel): organization_id: str name: str role: str class Me(BaseModel): id: str email: str display_name: str organizations: list[MyOrg] @router.get("/me", response_model=Me) async def me(user: CurrentUser = Depends(get_current_user)) -> Me: # MERK: `organization` (org_self-policyen) kan IKKE joines direkte her via # plain_connection() -- migrasjon 005 fikset kun tomstreng-krasjen, den # endret ikke at org_self krever en MATCHENDE app.current_org for å vise # en rad i det hele tatt (riktig RLS-oppførsel, ikke en bug). En bruker # kan tilhøre flere organisasjoner samtidig, så det finnes ingen ÉN # kontekst å sette for en tverr-org-spørring som denne. Løsningen er # derfor å slå opp hvert org-navn ETT OM GANGEN gjennom org_connection() # (som setter riktig kontekst for akkurat den ene raden) -- N+1 spørringer, # men N er antall organisasjoner brukeren tilhører (typisk 1-3), og dette # er den eneste måten å gjøre det på uten å endre selve RLS-modellen. async with plain_connection() as conn: user_row = await conn.fetchrow( "SELECT id::text AS id, email::text AS email, display_name FROM app_user WHERE id = $1", user.user_id, ) membership_rows = await conn.fetch( "SELECT organization_id::text AS organization_id, role FROM organization_membership WHERE user_id = $1", user.user_id, ) organizations = [] for m in membership_rows: async with org_connection(m["organization_id"]) as org_conn: name = await org_conn.fetchval( "SELECT name FROM organization WHERE id = $1", m["organization_id"] ) organizations.append(MyOrg(organization_id=m["organization_id"], name=name, role=m["role"])) return Me( id=user_row["id"], email=user_row["email"], display_name=user_row["display_name"], organizations=organizations, )