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")