Spaces:
Paused
Paused
File size: 4,800 Bytes
83bdb4a | 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 116 | import uuid
import hashlib
import secrets
import json
from datetime import datetime, timedelta
from fastapi_users_db_sqlalchemy import SQLAlchemyBaseUserTableUUID
from fastapi_users_db_sqlalchemy.generics import GUID
from sqlalchemy import Column, String, Integer, DateTime, Boolean, ForeignKey, Index, Text
from sqlalchemy.sql import func
from app.database import Base
# Use versioned encryption from utils
from app.utils.encryption import encrypt_mfa, decrypt_mfa, encrypt_backup_codes, decrypt_backup_codes
class User(SQLAlchemyBaseUserTableUUID, Base):
__tablename__ = "users"
full_name = Column(String(255), nullable=True)
credits = Column(Integer, default=1) # Limit to 1 free search
plan = Column(String(50), default="free")
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# Account lockout
failed_login_attempts = Column(Integer, default=0, nullable=False)
locked_until = Column(DateTime(timezone=True), nullable=True)
# Email verification
is_verified = Column(Boolean, default=False, nullable=False)
# ─── MFA / 2FA (encrypted) ───
mfa_enabled = Column(Boolean, default=False, nullable=False)
mfa_secret = Column(Text, nullable=True) # Encrypted TOTP secret (Fernet)
mfa_backup_codes = Column(Text, nullable=True) # Encrypted JSON array of backup codes
mfa_verified_at = Column(DateTime(timezone=True), nullable=True)
# OAuth / Social Login
oauth_provider = Column(String(50), nullable=True) # google, microsoft, github
oauth_provider_id = Column(String(255), nullable=True)
# Password strength
password_strength_score = Column(Integer, nullable=True) # zxcvbn score 0-4
@property
def mfa_secret_decrypted(self) -> str | None:
"""Get decrypted MFA secret for TOTP verification (v1 or v2)."""
if self.mfa_secret:
from app.utils.encryption import safe_decrypt_mfa
return safe_decrypt_mfa(self.mfa_secret)
return None
@mfa_secret_decrypted.setter
def mfa_secret_decrypted(self, value: str | None):
"""Set encrypted MFA secret (always writes v2)."""
from app.utils.encryption import encrypt_mfa
self.mfa_secret = encrypt_mfa(value) if value else None
@property
def mfa_backup_codes_decrypted(self) -> list[str]:
"""Get decrypted backup codes list (v1 or v2)."""
if self.mfa_backup_codes:
from app.utils.encryption import safe_decrypt_backup_codes
return safe_decrypt_backup_codes(self.mfa_backup_codes)
return []
@mfa_backup_codes_decrypted.setter
def mfa_backup_codes_decrypted(self, value: list[str] | None):
"""Set encrypted backup codes (always writes v2)."""
from app.utils.encryption import encrypt_backup_codes
self.mfa_backup_codes = encrypt_backup_codes(value) if value else None
class LoginHistory(Base):
__tablename__ = "login_history"
id = Column(Integer, primary_key=True, autoincrement=True)
user_id = Column(GUID, ForeignKey("users.id"), nullable=True)
email = Column(String(255), nullable=False)
success = Column(Boolean, default=False)
ip_address = Column(String(45), nullable=True)
user_agent = Column(String(500), nullable=True)
failure_reason = Column(String(255), nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
class RefreshToken(Base):
"""Refresh tokens para rotación segura (HttpOnly cookie)."""
__tablename__ = "refresh_tokens"
id = Column(Integer, primary_key=True, autoincrement=True)
user_id = Column(GUID, ForeignKey("users.id"), nullable=False, index=True)
token_hash = Column(String(64), nullable=False, unique=True, index=True) # SHA-256
expires_at = Column(DateTime(timezone=True), nullable=False)
revoked = Column(Boolean, default=False, nullable=False)
user_agent = Column(String(500), nullable=True)
ip_address = Column(String(45), nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
__table_args__ = (
Index("ix_refresh_tokens_user_revoked", "user_id", "revoked"),
)
@staticmethod
def hash_token(token: str) -> str:
"""Hash del token para almacenamiento seguro."""
return hashlib.sha256(token.encode()).hexdigest()
@staticmethod
def generate_token() -> str:
"""Generar token criptográficamente seguro."""
return secrets.token_urlsafe(32)
def verify(self, token: str) -> bool:
"""Verificar token contra hash almacenado."""
return self.token_hash == self.hash_token(token) and not self.revoked and self.expires_at > datetime.utcnow()
|