Spaces:
Running
Running
File size: 4,324 Bytes
09801ca | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | """
Auth Models — Sessions, refresh tokens, and password reset tokens.
"""
import uuid
from datetime import datetime
from typing import Optional
from sqlalchemy import String, Boolean, Text, Index, ForeignKey, DateTime
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base, TimestampMixin, UUIDPrimaryKeyMixin
class UserSession(UUIDPrimaryKeyMixin, TimestampMixin, Base):
"""
Active user sessions — tracks where a user is logged in.
Enables 'logout from all devices' and session management.
"""
__tablename__ = "user_sessions"
__table_args__ = (
Index("ix_user_sessions_user_id", "user_id"),
Index("ix_user_sessions_is_active", "is_active"),
)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
)
ip_address: Mapped[Optional[str]] = mapped_column(String(45), nullable=True)
user_agent: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
device_info: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
last_activity_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
# Relationships
user: Mapped["User"] = relationship(back_populates="sessions")
def __repr__(self) -> str:
return f"<UserSession id={self.id} user_id={self.user_id}>"
class RefreshToken(UUIDPrimaryKeyMixin, TimestampMixin, Base):
"""
Refresh token store — supports token rotation and family tracking.
Token Family: When a refresh token is used, a new one is issued in the
same family. If a revoked token is used again (replay attack), the entire
family is revoked.
"""
__tablename__ = "refresh_tokens"
__table_args__ = (
Index("ix_refresh_tokens_token_hash", "token_hash", unique=True),
Index("ix_refresh_tokens_user_id", "user_id"),
Index("ix_refresh_tokens_family_id", "family_id"),
)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
)
session_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("user_sessions.id", ondelete="CASCADE"),
nullable=False,
)
token_hash: Mapped[str] = mapped_column(
String(255), unique=True, nullable=False
)
family_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), default=uuid.uuid4, nullable=False
)
is_revoked: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
revoked_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
revoked_reason: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
def __repr__(self) -> str:
return f"<RefreshToken id={self.id} user_id={self.user_id} revoked={self.is_revoked}>"
class PasswordResetToken(UUIDPrimaryKeyMixin, TimestampMixin, Base):
"""Password reset token — single use, time-limited."""
__tablename__ = "password_reset_tokens"
__table_args__ = (
Index("ix_password_reset_tokens_token_hash", "token_hash", unique=True),
Index("ix_password_reset_tokens_user_id", "user_id"),
)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
)
token_hash: Mapped[str] = mapped_column(
String(255), unique=True, nullable=False
)
is_used: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
used_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
def __repr__(self) -> str:
return f"<PasswordResetToken id={self.id} user_id={self.user_id}>"
# Forward reference
from app.models.user import User # noqa: E402
|