"""New-user account verification, chosen by the admin who creates the user (Users page → "Require verification"), or for every self-registered Public Portal account by the tenant's RegistrationSettings.verification: - "email": a link + a 6-digit one-time code are emailed right away (via the tenant's first enabled email_smtp Connector). The person opens the link (frontend /verify-email), enters the code, and only then can sign in — POST /auth/login refuses a pending email account with a "email_verification_required" code the login page offers a resend for. - "sms": nothing is sent up front. At first sign-in, a correct password doesn't yet return a session — a code is texted (via the tenant's first enabled "sms" Connector) and the login page asks for it; the session is issued only once it's entered (POST /auth/login/verify-otp). Later sign-ins are normal. - "both": the email step first (sign-in stays blocked until it's done), then the SMS step at the next sign-in. Each channel is tracked by its own User.email_verified_at / phone_verified_at. Codes and link tokens are never stored in plain text: the code as an HMAC-SHA256 keyed by the JWT secret, the link token as SHA-256. A code expires (EMAIL_TTL / SMS_TTL), allows MAX_ATTEMPTS wrong tries, and can be re-sent at most once per RESEND_COOLDOWN.""" import asyncio import hashlib import hmac import secrets import smtplib import uuid from datetime import datetime, timedelta, timezone import httpx from jose import JWTError, jwt from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import settings from app.crud.connectors import first_enabled_connector from app.crud.secrets import connector_config from app.crud.tasks import _send_email_sync, send_sms from app.models.identity import Tenant, User EMAIL_TTL = timedelta(hours=24) SMS_TTL = timedelta(minutes=10) CHALLENGE_TTL = timedelta(minutes=15) RESEND_COOLDOWN = timedelta(seconds=30) MAX_ATTEMPTS = 5 OTP_DIGITS = 6 _CHALLENGE_PURPOSE = "login_otp" CONNECTOR_TYPE = {"email": "email_smtp", "sms": "sms"} CONNECTOR_LABEL = {"email": "Email (SMTP)", "sms": "SMS"} def verification_channels(method: str | None) -> list[str]: """"both" -> ["email", "sms"]; "email"/"sms" -> itself; None/"none" -> [].""" if method == "both": return ["email", "sms"] return [method] if method in ("email", "sms") else [] class VerificationError(Exception): def __init__(self, message: str, status_code: int = 400): super().__init__(message) self.status_code = status_code def _now() -> datetime: return datetime.now(timezone.utc) def _hash_code(code: str) -> str: return hmac.new(settings.jwt_secret.encode(), code.encode(), hashlib.sha256).hexdigest() def _hash_token(token: str) -> str: return hashlib.sha256(token.encode()).hexdigest() def _new_code() -> str: return str(secrets.randbelow(10**OTP_DIGITS)).zfill(OTP_DIGITS) def public_base_url(origin: str | None, referer: str | None) -> str: """Where the verification link should point: settings.app_base_url, else the Origin (or Referer's origin) of the browser request that triggered the email.""" if settings.app_base_url.strip(): return settings.app_base_url.strip() if origin and origin != "null": return origin if referer: parts = referer.split("/") if len(parts) >= 3: return "/".join(parts[:3]) return "http://localhost:3001" def mask_phone(phone: str | None) -> str: digits = "".join(c for c in (phone or "") if c.isdigit()) return f"•••• {digits[-4:]}" if len(digits) >= 4 else "your phone" async def require_connector(db: AsyncSession, tenant_id: uuid.UUID, method: str): connector = await first_enabled_connector(db, tenant_id, CONNECTOR_TYPE[method]) if connector is None: raise VerificationError( f"{CONNECTOR_LABEL[method]} verification needs an enabled {CONNECTOR_LABEL[method]} connector — add one on the Connectors page first.", status_code=422, ) return connector def _check_cooldown(user: User) -> None: if user.verification_sent_at and _now() - user.verification_sent_at < RESEND_COOLDOWN: wait = int((RESEND_COOLDOWN - (_now() - user.verification_sent_at)).total_seconds()) + 1 raise VerificationError(f"A code was just sent — wait {wait}s before asking for another.", status_code=429) def _arm(user: User, ttl: timedelta) -> str: code = _new_code() user.verification_code_hash = _hash_code(code) user.verification_expires_at = _now() + ttl user.verification_sent_at = _now() user.verification_attempts = 0 return code async def send_email_verification(db: AsyncSession, user: User, tenant: Tenant, base_url: str, *, enforce_cooldown: bool = False) -> str: """Arms a new link token + code and emails them. Returns a short delivery note ("sent" or why not) instead of raising on an SMTP failure — the caller decides whether that's fatal. Raises VerificationError when there's no connector at all or the cooldown hasn't passed.""" if enforce_cooldown: _check_cooldown(user) connector = await require_connector(db, user.tenant_id, "email") code = _arm(user, EMAIL_TTL) token = secrets.token_urlsafe(32) user.verification_token_hash = _hash_token(token) await db.flush() link = f"{base_url.rstrip('/')}/verify-email?tenant={tenant.id}&token={token}" hours = int(EMAIL_TTL.total_seconds() // 3600) body = ( f"Hello {user.name},\n\n" f"An account has been created for you on {tenant.name}. To verify your email address, open this link:\n\n" f"{link}\n\n" f"and enter this one-time code:\n\n" f" {code}\n\n" f"The link and code expire in {hours} hours. If you weren't expecting this, you can ignore this email." ) try: await asyncio.to_thread(_send_email_sync, await connector_config(db, user.tenant_id, connector), user.email, f"Verify your {tenant.name} account", body) except (smtplib.SMTPException, OSError, ValueError) as exc: return f"failed: {exc}" return "sent" async def send_sms_code(db: AsyncSession, user: User, tenant: Tenant, *, enforce_cooldown: bool = False) -> None: """Arms a new code and texts it. Raises VerificationError if it can't be delivered — without the text the person has no way to continue.""" if enforce_cooldown: _check_cooldown(user) if not user.phone: raise VerificationError("This account needs SMS verification but has no phone number — contact your administrator.", 409) connector = await require_connector(db, user.tenant_id, "sms") code = _arm(user, SMS_TTL) await db.flush() minutes = int(SMS_TTL.total_seconds() // 60) message = f"{code} is your {tenant.name} verification code. It expires in {minutes} minutes." try: status_code = await send_sms(await connector_config(db, user.tenant_id, connector), user.phone, message) except (ValueError, httpx.HTTPError) as exc: raise VerificationError(f"Couldn't send the verification SMS ({exc}) — contact your administrator.", 503) from exc if status_code >= 400: raise VerificationError(f"The SMS gateway refused the verification message (HTTP {status_code}) — contact your administrator.", 503) def check_code(user: User, code: str, *, channel: str, token: str | None = None) -> None: """Raises VerificationError unless `code` (and, for email, the link's `token`) is right and still live; on success marks that channel ("email" or "sms") verified — and the account, once no channel is left — and clears every stored secret. A wrong code counts toward MAX_ATTEMPTS — the caller must still commit so that count sticks.""" still_needed = user.needs_email_verification if channel == "email" else user.needs_sms_verification if not still_needed: raise VerificationError("This is already verified — sign in.", 409) if token is not None and (not user.verification_token_hash or not hmac.compare_digest(_hash_token(token), user.verification_token_hash)): raise VerificationError("This verification link isn't valid any more — ask for a new one from the sign-in page.") if not user.verification_code_hash or not user.verification_expires_at or user.verification_expires_at < _now(): raise VerificationError("This code has expired — ask for a new one.") if user.verification_attempts >= MAX_ATTEMPTS: raise VerificationError("Too many wrong codes — ask for a new one.", 429) if not hmac.compare_digest(_hash_code(code.strip()), user.verification_code_hash): user.verification_attempts += 1 left = MAX_ATTEMPTS - user.verification_attempts raise VerificationError(f"That code isn't right — {left} attempt{'s' if left != 1 else ''} left." if left else "Too many wrong codes — ask for a new one.") if channel == "email": user.email_verified_at = _now() else: user.phone_verified_at = _now() if not user.verification_pending: user.verified_at = _now() user.verification_code_hash = None user.verification_token_hash = None user.verification_expires_at = None user.verification_attempts = 0 def create_challenge_token(user: User) -> str: """Short-lived proof that the password step passed, for the SMS code step. Carries "purpose" and no "tenant_id"/"user_id" keys, so get_auth_context rejects it as a session token.""" claims = { "purpose": _CHALLENGE_PURPOSE, "challenge_user_id": str(user.id), "challenge_tenant_id": str(user.tenant_id), "exp": _now() + CHALLENGE_TTL, } return jwt.encode(claims, settings.jwt_secret, algorithm=settings.jwt_algorithm) def decode_challenge_token(token: str, tenant_id: uuid.UUID) -> uuid.UUID: try: claims = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm]) # 400 rather than 401 throughout: these are pre-session endpoints, and # the frontend treats any 401 as "your session ended". except JWTError as exc: raise VerificationError("This sign-in attempt has expired — sign in again.") from exc if claims.get("purpose") != _CHALLENGE_PURPOSE or claims.get("challenge_tenant_id") != str(tenant_id): raise VerificationError("This sign-in attempt has expired — sign in again.") return uuid.UUID(claims["challenge_user_id"])