refactor(auth): extract _ensure_tz_aware helper, use validate_session in logout
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
+2
-2
@@ -1116,9 +1116,9 @@ async def logout(request: Request, db: Session = Depends(get_db)):
|
|||||||
session_token = request.session.get("_session_token")
|
session_token = request.session.get("_session_token")
|
||||||
if session_token:
|
if session_token:
|
||||||
try:
|
try:
|
||||||
from app.models import UserSession
|
from app.utils.session_manager import validate_session
|
||||||
|
|
||||||
user_session = db.query(UserSession).filter(UserSession.session_token == session_token).first()
|
user_session = validate_session(db, session_token)
|
||||||
if user_session:
|
if user_session:
|
||||||
user_session.is_revoked = True
|
user_session.is_revoked = True
|
||||||
user_session.revoked_at = datetime.now(timezone.utc)
|
user_session.revoked_at = datetime.now(timezone.utc)
|
||||||
|
|||||||
@@ -20,6 +20,18 @@ from app.models import ApiToken, QRLoginChallenge, UserSession
|
|||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def _ensure_tz_aware(dt: datetime | None) -> datetime | None:
|
||||||
|
"""Return *dt* with UTC tzinfo if it is naive, or unchanged if already aware.
|
||||||
|
|
||||||
|
SQLite does not persist timezone information, so datetimes read back from
|
||||||
|
the database are offset-naive. This helper normalises them for safe
|
||||||
|
comparison with ``datetime.now(timezone.utc)``.
|
||||||
|
"""
|
||||||
|
if dt is not None and dt.tzinfo is None:
|
||||||
|
return dt.replace(tzinfo=timezone.utc)
|
||||||
|
return dt
|
||||||
|
|
||||||
|
|
||||||
def get_session_lifetime_days() -> int:
|
def get_session_lifetime_days() -> int:
|
||||||
"""Return the effective session lifetime in days.
|
"""Return the effective session lifetime in days.
|
||||||
|
|
||||||
@@ -118,18 +130,13 @@ def validate_session(db: Session, session_token: str) -> UserSession | None:
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
if user_session.expires_at:
|
if user_session.expires_at:
|
||||||
# Ensure timezone-aware comparison (SQLite returns naive datetimes)
|
expires = _ensure_tz_aware(user_session.expires_at)
|
||||||
expires = user_session.expires_at
|
|
||||||
if expires.tzinfo is None:
|
|
||||||
expires = expires.replace(tzinfo=timezone.utc)
|
|
||||||
if expires < now:
|
if expires < now:
|
||||||
logger.debug("[SESSION] Session id=%s has expired", user_session.id)
|
logger.debug("[SESSION] Session id=%s has expired", user_session.id)
|
||||||
return None
|
return None
|
||||||
|
|
||||||
# Update last_active_at (throttled to avoid excessive writes)
|
# Update last_active_at (throttled to avoid excessive writes)
|
||||||
last_active = user_session.last_active_at
|
last_active = _ensure_tz_aware(user_session.last_active_at)
|
||||||
if last_active and last_active.tzinfo is None:
|
|
||||||
last_active = last_active.replace(tzinfo=timezone.utc)
|
|
||||||
if not last_active or (now - last_active).total_seconds() > 60:
|
if not last_active or (now - last_active).total_seconds() > 60:
|
||||||
try:
|
try:
|
||||||
user_session.last_active_at = now
|
user_session.last_active_at = now
|
||||||
@@ -250,9 +257,7 @@ def list_user_sessions(db: Session, user_id: str) -> list[UserSession]:
|
|||||||
# Filter expired sessions in Python to handle timezone-naive datetimes (SQLite)
|
# Filter expired sessions in Python to handle timezone-naive datetimes (SQLite)
|
||||||
result = []
|
result = []
|
||||||
for s in sessions:
|
for s in sessions:
|
||||||
expires = s.expires_at
|
expires = _ensure_tz_aware(s.expires_at)
|
||||||
if expires and expires.tzinfo is None:
|
|
||||||
expires = expires.replace(tzinfo=timezone.utc)
|
|
||||||
if expires and expires > now:
|
if expires and expires > now:
|
||||||
result.append(s)
|
result.append(s)
|
||||||
return result
|
return result
|
||||||
@@ -337,10 +342,7 @@ def validate_qr_challenge(db: Session, challenge_token: str) -> QRLoginChallenge
|
|||||||
if challenge.is_claimed or challenge.is_cancelled:
|
if challenge.is_claimed or challenge.is_cancelled:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
# Ensure timezone-aware comparison (SQLite returns naive datetimes)
|
expires = _ensure_tz_aware(challenge.expires_at)
|
||||||
expires = challenge.expires_at
|
|
||||||
if expires and expires.tzinfo is None:
|
|
||||||
expires = expires.replace(tzinfo=timezone.utc)
|
|
||||||
if expires and expires < now:
|
if expires and expires < now:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@@ -437,10 +439,7 @@ def get_challenge_status(db: Session, challenge_id: int, user_id: str) -> dict |
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
now = datetime.now(timezone.utc)
|
now = datetime.now(timezone.utc)
|
||||||
# Ensure timezone-aware comparison (SQLite returns naive datetimes)
|
expires = _ensure_tz_aware(challenge.expires_at)
|
||||||
expires = challenge.expires_at
|
|
||||||
if expires and expires.tzinfo is None:
|
|
||||||
expires = expires.replace(tzinfo=timezone.utc)
|
|
||||||
|
|
||||||
if challenge.is_claimed:
|
if challenge.is_claimed:
|
||||||
status = "claimed"
|
status = "claimed"
|
||||||
|
|||||||
Reference in New Issue
Block a user