171 lines
4.6 KiB
Python
171 lines
4.6 KiB
Python
"""
|
|
Refresh Token Model - ServiceManagerWeb
|
|
|
|
Modelo para persistencia de refresh tokens con revocaci├│n y tracking.
|
|
"""
|
|
|
|
from sqlalchemy import String, Boolean, DateTime, ForeignKey, Index, Integer
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
from typing import Optional, TYPE_CHECKING
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from app.core.database import Base, GUID
|
|
|
|
if TYPE_CHECKING:
|
|
from app.models.user import User
|
|
|
|
|
|
class RefreshToken(Base):
|
|
"""
|
|
Refresh Token persistente para gesti├│n de sesiones.
|
|
|
|
Almacena refresh tokens con informaci├│n de dispositivo y permite
|
|
revocaci├│n para mejorar la seguridad.
|
|
|
|
Características:
|
|
- Token hasheado (no se guarda en texto plano)
|
|
- Device fingerprinting
|
|
- Revocaci├│n individual con tracking
|
|
- Auto-expiraci├│n
|
|
- Tracking de IP y uso
|
|
"""
|
|
|
|
__tablename__ = "refresh_tokens"
|
|
|
|
# User relationship
|
|
user_id: Mapped[uuid.UUID] = mapped_column(
|
|
GUID(),
|
|
ForeignKey("users.id", ondelete="CASCADE"),
|
|
nullable=False,
|
|
index=True
|
|
)
|
|
|
|
# Token JWT (almacenado directamente - firmado y verificable)
|
|
# VARCHAR(500) para acomodar JWTs con payload extenso
|
|
token: Mapped[str] = mapped_column(
|
|
String(500),
|
|
nullable=False,
|
|
unique=True
|
|
)
|
|
|
|
# Device information
|
|
device_id: Mapped[Optional[str]] = mapped_column(
|
|
String(100),
|
|
nullable=True
|
|
)
|
|
|
|
device_name: Mapped[Optional[str]] = mapped_column(
|
|
String(200),
|
|
nullable=True
|
|
)
|
|
|
|
user_agent: Mapped[Optional[str]] = mapped_column(
|
|
String(500),
|
|
nullable=True
|
|
)
|
|
|
|
# IP address del cliente (varchar(45) para IPv6)
|
|
ip_address: Mapped[Optional[str]] = mapped_column(
|
|
String(45),
|
|
nullable=True
|
|
)
|
|
|
|
# Expiraci├│n del token
|
|
expires_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=False,
|
|
index=True
|
|
)
|
|
|
|
# Estado de revocaci├│n
|
|
revoked: Mapped[bool] = mapped_column(
|
|
Boolean,
|
|
default=False,
|
|
nullable=False
|
|
)
|
|
|
|
revoked_at: Mapped[Optional[datetime]] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=True
|
|
)
|
|
|
|
revoked_by: Mapped[Optional[uuid.UUID]] = mapped_column(
|
|
GUID(),
|
|
ForeignKey("users.id", ondelete="SET NULL"),
|
|
nullable=True
|
|
)
|
|
|
|
# Tracking de uso
|
|
last_used_at: Mapped[Optional[datetime]] = mapped_column(
|
|
DateTime(timezone=True),
|
|
nullable=True
|
|
)
|
|
|
|
usage_count: Mapped[int] = mapped_column(
|
|
Integer,
|
|
default=0,
|
|
nullable=False
|
|
)
|
|
|
|
# Timestamps
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
default=datetime.utcnow,
|
|
nullable=False,
|
|
server_default="NOW()"
|
|
)
|
|
|
|
updated_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
default=datetime.utcnow,
|
|
onupdate=datetime.utcnow,
|
|
nullable=False,
|
|
server_default="NOW()"
|
|
)
|
|
|
|
# Relaci├│n con usuario
|
|
user: Mapped["User"] = relationship("User", foreign_keys=[user_id], back_populates="refresh_tokens")
|
|
revoker: Mapped[Optional["User"]] = relationship("User", foreign_keys=[revoked_by])
|
|
|
|
# Índices compuestos
|
|
__table_args__ = (
|
|
Index('idx_refresh_tokens_user_expires', 'user_id', 'expires_at'),
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<RefreshToken(user_id='{self.user_id}', revoked={self.revoked}, expires={self.expires_at})>"
|
|
|
|
@property
|
|
def is_valid(self) -> bool:
|
|
"""
|
|
Verificar si el token es válido.
|
|
|
|
Un token es válido si:
|
|
- No está revocado
|
|
- No ha expirado
|
|
"""
|
|
return not self.revoked and self.expires_at > datetime.now(timezone.utc)
|
|
|
|
@property
|
|
def is_expired(self) -> bool:
|
|
"""Verificar si el token ha expirado."""
|
|
return datetime.now(timezone.utc) >= self.expires_at
|
|
|
|
def revoke(self, revoked_by: Optional[uuid.UUID] = None) -> None:
|
|
"""
|
|
Marcar el token como revocado.
|
|
|
|
Args:
|
|
revoked_by: ID del usuario que revoc├│ el token
|
|
"""
|
|
self.revoked = True
|
|
self.revoked_at = datetime.now(timezone.utc)
|
|
if revoked_by:
|
|
self.revoked_by = revoked_by
|
|
|
|
def track_usage(self) -> None:
|
|
"""Registrar uso del token."""
|
|
self.last_used_at = datetime.now(timezone.utc)
|
|
self.usage_count += 1
|