106 lines
3.7 KiB
Python
106 lines
3.7 KiB
Python
"""
|
|
Auth-lag for TeeCup.
|
|
|
|
Ekte autentisering: magic-link (se app/routers/auth.py for utstedelse) +
|
|
JWT-sesjon i en HttpOnly-cookie. `get_current_user` er IKKE lenger en stub —
|
|
den dekoder og verifiserer den innsendte sesjonscookien, og bekrefter at
|
|
brukeren fortsatt finnes.
|
|
|
|
`get_authorized_org` er fortsatt sikkerhetskritisk uendret: den verifiserer at
|
|
den innloggede brukeren faktisk er medlem av organisasjonen før org-konteksten
|
|
settes.
|
|
|
|
Hvorfor det er kritisk: RLS stoler blindt på `app.current_org`. Setter appen den
|
|
til en organisasjon brukeren ikke tilhører, gir RLS lydig tilgang til den
|
|
organisasjonens data. Isolasjonen står og faller altså på at denne sjekken skjer
|
|
FØR org_connection kalles — aldri sett konteksten fra en uverifisert kilde.
|
|
"""
|
|
|
|
import time
|
|
from dataclasses import dataclass
|
|
|
|
import jwt
|
|
from fastapi import Depends, Request
|
|
|
|
from .config import settings
|
|
from .db import plain_connection
|
|
from .errors import app_error
|
|
|
|
SESSION_COOKIE_NAME = "teecup_session"
|
|
_JWT_ALGORITHM = "HS256" # Eksplisitt både ved signering og dekoding -- blokkerer
|
|
# alg-forvirring/"alg:none"-angrep.
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CurrentUser:
|
|
user_id: str
|
|
|
|
|
|
def create_session_token(user_id: str) -> str:
|
|
now = int(time.time())
|
|
payload = {"sub": user_id, "iat": now, "exp": now + settings.SESSION_MAX_AGE_SECONDS}
|
|
return jwt.encode(payload, settings.SESSION_SECRET, algorithm=_JWT_ALGORITHM)
|
|
|
|
|
|
def should_use_secure_cookies(request: Request) -> bool:
|
|
"""Secure-flagget skal kun være sant over https.
|
|
|
|
Stoler på X-Forwarded-Proto -- dette er KUN trygt fordi appen ikke skal
|
|
være nåbar unntatt gjennom Caddy, som terminerer TLS og proxyer videre
|
|
over vanlig http internt.
|
|
"""
|
|
if request.url.scheme == "https":
|
|
return True
|
|
return request.headers.get("x-forwarded-proto", "").strip().lower() == "https"
|
|
|
|
|
|
async def get_current_user(request: Request) -> CurrentUser:
|
|
"""Dekoder sesjonscookien og bekrefter at brukeren fortsatt finnes.
|
|
|
|
Det siste steget (et ekte DB-oppslag, ikke bare å stole på JWT-en) gir
|
|
faktisk tilbakekalling: en slettet/deaktivert bruker kan ikke ri ut
|
|
resten av sesjonens levetid på en ellers gyldig token.
|
|
"""
|
|
token = request.cookies.get(SESSION_COOKIE_NAME)
|
|
if not token:
|
|
raise app_error(401, "NOT_AUTHENTICATED", "Ikke innlogget.")
|
|
try:
|
|
claims = jwt.decode(token, settings.SESSION_SECRET, algorithms=[_JWT_ALGORITHM])
|
|
except jwt.PyJWTError:
|
|
raise app_error(401, "NOT_AUTHENTICATED", "Ugyldig eller utløpt sesjon.")
|
|
|
|
user_id = claims.get("sub")
|
|
if not user_id:
|
|
raise app_error(401, "NOT_AUTHENTICATED", "Ugyldig sesjon.")
|
|
|
|
async with plain_connection() as conn:
|
|
exists = await conn.fetchval("SELECT EXISTS (SELECT 1 FROM app_user WHERE id = $1)", user_id)
|
|
if not exists:
|
|
raise app_error(401, "NOT_AUTHENTICATED", "Brukeren finnes ikke lenger.")
|
|
|
|
return CurrentUser(user_id=user_id)
|
|
|
|
|
|
async def get_authorized_org(
|
|
organization_id: str,
|
|
user: CurrentUser = Depends(get_current_user),
|
|
) -> str:
|
|
"""Returnerer organization_id KUN hvis brukeren er medlem — ellers 403.
|
|
|
|
Medlemskapstabellen er ikke organisasjonsavgrenset, så oppslaget gjøres på en
|
|
tilkobling uten org-kontekst.
|
|
"""
|
|
async with plain_connection() as conn:
|
|
is_member = await conn.fetchval(
|
|
"""
|
|
SELECT EXISTS (
|
|
SELECT 1 FROM organization_membership
|
|
WHERE user_id = $1 AND organization_id = $2
|
|
)
|
|
""",
|
|
user.user_id,
|
|
organization_id,
|
|
)
|
|
if not is_member:
|
|
raise app_error(403, "NOT_ORG_MEMBER", "Brukeren er ikke medlem av denne organisasjonen.")
|
|
return organization_id
|