30 lines
1.0 KiB
Python
30 lines
1.0 KiB
Python
from datetime import datetime
|
|
|
|
from sqlalchemy.dialects.mysql import INTEGER
|
|
|
|
from app import db
|
|
|
|
|
|
class TokenBlacklist(db.Model):
|
|
"""Revoked JWT identifiers. Access and refresh tokens are added on logout."""
|
|
__tablename__ = 'token_blacklist'
|
|
|
|
id = db.Column(INTEGER(unsigned=True), autoincrement=True, primary_key=True)
|
|
jti = db.Column(db.String(36), unique=True, nullable=False, index=True)
|
|
user_id = db.Column(INTEGER(unsigned=True), nullable=False)
|
|
expires_at = db.Column(db.DateTime, nullable=False)
|
|
|
|
@classmethod
|
|
def is_blacklisted(cls, jti: str) -> bool:
|
|
entry = cls.query.filter_by(jti=jti).first()
|
|
if not entry:
|
|
return False
|
|
# Automatically ignore expired entries (they can be cleaned up later)
|
|
return entry.expires_at > datetime.utcnow()
|
|
|
|
@classmethod
|
|
def cleanup_expired(cls):
|
|
"""Delete entries that have already expired — call occasionally to keep table small."""
|
|
cls.query.filter(cls.expires_at <= datetime.utcnow()).delete()
|
|
db.session.commit()
|