87 lines
2.6 KiB
Python
87 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from hashlib import sha256
|
|
from secrets import token_urlsafe
|
|
|
|
from fastapi import Cookie, Depends, HTTPException, Response
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from . import models
|
|
from .db import get_session
|
|
|
|
SESSION_COOKIE = "mousehold_session"
|
|
SESSION_DAYS = 30
|
|
|
|
|
|
def hash_session_token(token: str) -> str:
|
|
return sha256(token.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def create_local_session(session: Session, user: models.User) -> tuple[models.LocalSession, str]:
|
|
token = token_urlsafe(32)
|
|
now = datetime.now(UTC)
|
|
local_session = models.LocalSession(
|
|
user_id=user.id,
|
|
household_id=user.household_id,
|
|
token_hash=hash_session_token(token),
|
|
created_at=now,
|
|
last_seen_at=now,
|
|
expires_at=now + timedelta(days=SESSION_DAYS),
|
|
)
|
|
session.add(local_session)
|
|
session.flush()
|
|
return local_session, token
|
|
|
|
|
|
def set_session_cookie(response: Response, token: str) -> None:
|
|
response.set_cookie(
|
|
SESSION_COOKIE,
|
|
token,
|
|
httponly=True,
|
|
samesite="lax",
|
|
secure=False,
|
|
max_age=SESSION_DAYS * 24 * 60 * 60,
|
|
path="/",
|
|
)
|
|
|
|
|
|
def clear_session_cookie(response: Response) -> None:
|
|
response.delete_cookie(SESSION_COOKIE, path="/")
|
|
|
|
|
|
def get_optional_local_session(
|
|
session_token: str | None = Cookie(default=None, alias=SESSION_COOKIE),
|
|
session: Session = Depends(get_session),
|
|
) -> models.LocalSession | None:
|
|
if not session_token:
|
|
return None
|
|
now = datetime.now(UTC)
|
|
local_session = session.scalars(
|
|
select(models.LocalSession)
|
|
.where(models.LocalSession.token_hash == hash_session_token(session_token))
|
|
.where(models.LocalSession.revoked_at.is_(None))
|
|
.where(models.LocalSession.expires_at > now)
|
|
.limit(1)
|
|
).first()
|
|
if local_session:
|
|
local_session.last_seen_at = now
|
|
session.add(local_session)
|
|
session.commit()
|
|
session.refresh(local_session)
|
|
return local_session
|
|
|
|
|
|
def require_local_session(
|
|
local_session: models.LocalSession | None = Depends(get_optional_local_session),
|
|
) -> models.LocalSession:
|
|
if not local_session:
|
|
raise HTTPException(status_code=401, detail="Local session required")
|
|
return local_session
|
|
|
|
|
|
def require_household_access(household_id: str, local_session: models.LocalSession | None) -> None:
|
|
if local_session and local_session.household_id != household_id:
|
|
raise HTTPException(status_code=403, detail="Session does not belong to this household")
|