35 lines
1.1 KiB
Python
35 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import delete, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.security import SessionPayload, session_token_hash
|
|
from app.models.revoked_session import RevokedSession
|
|
|
|
settings = get_settings()
|
|
|
|
|
|
def is_session_revoked(db: Session, token: str) -> bool:
|
|
token_hash = session_token_hash(token)
|
|
revoked_id = db.scalar(select(RevokedSession.id).where(RevokedSession.token_hash == token_hash))
|
|
return revoked_id is not None
|
|
|
|
|
|
def revoke_session(db: Session, token: str, payload: SessionPayload) -> None:
|
|
db.execute(delete(RevokedSession).where(RevokedSession.expires_at < datetime.now(UTC)))
|
|
token_hash = session_token_hash(token)
|
|
existing = db.scalar(select(RevokedSession.id).where(RevokedSession.token_hash == token_hash))
|
|
if existing is not None:
|
|
return
|
|
db.add(
|
|
RevokedSession(
|
|
token_hash=token_hash,
|
|
expires_at=datetime.fromtimestamp(
|
|
payload.issued_at + settings.session_ttl_seconds, UTC
|
|
),
|
|
)
|
|
)
|