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()