Spaces:
Running
Running
| """SuperAdmin TOTP enrollment — mandatory before privileged operations.""" | |
| from __future__ import annotations | |
| import structlog | |
| from fastapi import HTTPException | |
| from redis.asyncio import Redis | |
| from sqlalchemy.ext.asyncio import AsyncSession | |
| from app.core.access.totp import generate_qr_code_base64, generate_totp_secret, verify_totp | |
| from app.core.crypto.credential_crypto import decrypt_secret, encrypt_secret | |
| from app.models.user import User | |
| from app.repositories.audit_repo import AuditRepository | |
| from app.repositories.user_repo import UserRepository | |
| logger = structlog.get_logger(__name__) | |
| _PENDING_PREFIX = "totp_pending:v1:" | |
| _PENDING_TTL = 600 | |
| class TotpSetupService: | |
| """Manage SuperAdmin two-factor enrollment lifecycle.""" | |
| def __init__(self, db: AsyncSession, redis: Redis | None = None) -> None: | |
| self._db = db | |
| self._redis = redis | |
| self._users = UserRepository(db) | |
| self._audit = AuditRepository(db) | |
| async def status(self, actor: User) -> dict: | |
| """Return whether TOTP is configured for the current SuperAdmin.""" | |
| return {"configured": bool(actor.totp_secret)} | |
| async def begin_setup(self, actor: User) -> dict: | |
| """Start enrollment — returns QR for authenticator app.""" | |
| if actor.totp_secret: | |
| raise HTTPException(400, "TOTP already configured for this account") | |
| secret = generate_totp_secret() | |
| await self._store_pending(actor.id, secret) | |
| logger.info("TOTP enrollment started", actor_email=actor.email) | |
| return { | |
| "qr_code_data_uri": generate_qr_code_base64(secret, actor.email), | |
| "manual_entry_key": secret, | |
| } | |
| async def confirm_setup(self, actor: User, code: str) -> dict: | |
| """Verify authenticator code and persist TOTP secret.""" | |
| if actor.totp_secret: | |
| raise HTTPException(400, "TOTP already configured for this account") | |
| pending = await self._load_pending(actor.id) | |
| if not pending: | |
| raise HTTPException(400, "Enrollment expired — scan QR again") | |
| if not verify_totp(pending, code): | |
| raise HTTPException(400, "Invalid code — check your authenticator app") | |
| await self._users.set_totp_secret(actor.id, pending) | |
| await self._clear_pending(actor.id) | |
| await self._audit.log( | |
| actor_id=actor.id, | |
| actor_email=actor.email, | |
| action="totp_enrolled", | |
| target_type="user", | |
| target_id=actor.id, | |
| details={}, | |
| ) | |
| await self._db.flush() | |
| logger.info("TOTP enrollment completed", actor_email=actor.email) | |
| return {"configured": True} | |
| async def _store_pending(self, user_id: str, secret: str) -> None: | |
| if self._redis is None: | |
| raise HTTPException(503, "Redis required for TOTP enrollment") | |
| await self._redis.set( | |
| _PENDING_PREFIX + user_id, | |
| encrypt_secret(secret), | |
| ex=_PENDING_TTL, | |
| ) | |
| async def _load_pending(self, user_id: str) -> str | None: | |
| if self._redis is None: | |
| return None | |
| raw = await self._redis.get(_PENDING_PREFIX + user_id) | |
| if not raw: | |
| return None | |
| return decrypt_secret(raw) | |
| async def _clear_pending(self, user_id: str) -> None: | |
| if self._redis is None: | |
| return | |
| await self._redis.delete(_PENDING_PREFIX + user_id) | |