"""Generic one-time-code verification framework. A single framework backs **every** verification purpose in the platform — email verification today, and password reset / email change / login verification / 2FA in future — with no second OTP table and no duplicated logic. Adding a new authentication feature means referencing a new :class:`~app.models.enums.VerificationPurpose`; the framework already handles generation, hashing, expiry, single-use, rate limiting, delivery and auditing. Design: * Codes are stored in the ``email_verifications`` table, isolated by ``purpose`` — a password-reset code can never satisfy an email-verification check (purpose isolation). * The code is **never persisted in plain text**: ``verification_token`` holds an HMAC-SHA256 digest (64 hex chars) keyed with ``SECRET_KEY`` and bound to the ``user_id``. * ``expires_at`` enforces the short lifetime, ``verified_at`` makes a code single-use, and retained rows drive per-purpose rate limiting from ``created_at`` — no extra columns. Public API:: issue_verification_code(user, purpose, *, email=None, enforce_cooldown=True) verify_code(user, purpose, code) invalidate_codes(user, purpose) Verification is deliberately generic: :func:`verify_code` validates and consumes a code but applies **no** purpose-specific side effect. Each feature applies its own consequence (e.g. the email-verification route calls ``mark_email_verified``), so the framework stays reusable. All limits (length, expiry, cooldown, hourly cap, verify attempts) come from configuration — see :class:`app.config.Config`. """ from __future__ import annotations import hashlib import hmac import secrets from datetime import timedelta from flask import current_app from flask_mail import Message from app.extensions import db, mail from app.models import EmailVerification, VerificationPurpose from app.models.base import utcnow # Default purpose keeps call sites that only care about email verification terse # and preserves the original behaviour exactly. DEFAULT_PURPOSE = VerificationPurpose.EMAIL_VERIFICATION class OtpError(Exception): """Base class for verification errors surfaced to the user.""" class OtpRateLimited(OtpError): """Raised when a code cannot be issued yet (cooldown or hourly cap).""" # --- Configuration helpers ----------------------------------------------------- def _cfg(key: str, default): return current_app.config.get(key, default) def _as_purpose(purpose) -> VerificationPurpose: """Accept an enum member or its string value; normalise to the enum.""" if isinstance(purpose, VerificationPurpose): return purpose return VerificationPurpose(purpose) # --- Audit logging (never logs the code) --------------------------------------- def audit_event(event: str, user, purpose, **extra) -> None: """Record a verification lifecycle event: issued, resent, verified, failed, expired, locked. The purpose is always included; the code value never is.""" purpose = _as_purpose(purpose) detail = "".join(f" {k}={v}" for k, v in extra.items()) logger = current_app.logger line = "verification.%s purpose=%s user=%s%s" args = (event, purpose.value, getattr(user, "id", None), detail) if event in ("failed", "expired", "locked"): logger.warning(line, *args) else: logger.info(line, *args) # --- Code generation & hashing (shared primitives) ----------------------------- # These are the single implementation of the OTP algorithm. Both the user-bound # flow below and the pre-user registration flow (app/auth/registration.py) call # them, so there is exactly one place that generates and hashes codes. def generate_code() -> str: """Cryptographically-random numeric code, zero-padded to the configured length.""" length = _cfg("OTP_LENGTH", 6) return str(secrets.randbelow(10 ** length)).zfill(length) def hash_code(identity, purpose, code: str) -> str: """Keyed HMAC-SHA256 digest (64 hex chars → fits CHAR(64)). Keyed with ``SECRET_KEY`` and bound to both an ``identity`` (a user id, or a ``"reg:"`` namespace for a pre-user registration) and the ``purpose`` so a leaked table can't be reversed without the app secret, and a digest is valid only for the identity *and* purpose it was issued for.""" purpose = _as_purpose(purpose) secret = current_app.config["SECRET_KEY"].encode("utf-8") message = f"{identity}:{purpose.value}:{code}".encode("utf-8") return hmac.new(secret, message, hashlib.sha256).hexdigest() # --- Rate-limit introspection (all scoped to user + purpose) ------------------- def _latest_request(user, purpose: VerificationPurpose) -> EmailVerification | None: return db.session.scalar( db.select(EmailVerification) .where(EmailVerification.user_id == user.id, EmailVerification.purpose == purpose) .order_by(EmailVerification.created_at.desc()) .limit(1) ) def requests_last_hour(user, purpose=DEFAULT_PURPOSE) -> int: """Codes issued to this account for this purpose in the trailing hour.""" purpose = _as_purpose(purpose) since = utcnow() - timedelta(hours=1) return db.session.scalar( db.select(db.func.count(EmailVerification.id)).where( EmailVerification.user_id == user.id, EmailVerification.purpose == purpose, EmailVerification.created_at >= since, ) ) or 0 def seconds_until_resend(user, purpose=DEFAULT_PURPOSE) -> int: """Seconds left on the resend cooldown for this purpose (0 when allowed).""" purpose = _as_purpose(purpose) latest = _latest_request(user, purpose) if latest is None: return 0 cooldown = _cfg("OTP_RESEND_COOLDOWN_SECONDS", 60) elapsed = (utcnow() - latest.created_at).total_seconds() return max(0, int(cooldown - elapsed + 0.999)) def _active_code(user, purpose: VerificationPurpose) -> EmailVerification | None: """Most recent still-valid, unredeemed code for this user + purpose.""" now = utcnow() return db.session.scalar( db.select(EmailVerification) .where( EmailVerification.user_id == user.id, EmailVerification.purpose == purpose, EmailVerification.verified_at.is_(None), EmailVerification.expires_at > now, ) .order_by(EmailVerification.created_at.desc()) .limit(1) ) def has_active_code(user, purpose=DEFAULT_PURPOSE) -> bool: """True when a live, unredeemed code exists for this user + purpose.""" return _active_code(user, _as_purpose(purpose)) is not None def invalidate_codes(user, purpose=DEFAULT_PURPOSE) -> None: """Expire any live codes for this user + purpose (e.g. on resend or lockout). Rows are kept (not deleted) so they still count toward the hourly limit; pushing ``expires_at`` into the past simply makes them unusable.""" purpose = _as_purpose(purpose) now = utcnow() db.session.execute( db.update(EmailVerification) .where( EmailVerification.user_id == user.id, EmailVerification.purpose == purpose, EmailVerification.verified_at.is_(None), EmailVerification.expires_at > now, ) .values(expires_at=now) ) db.session.commit() # --- Issue a code -------------------------------------------------------------- def issue_verification_code(user, purpose=DEFAULT_PURPOSE, *, email=None, enforce_cooldown: bool = True, event: str = "issued", expiry_minutes: int | None = None) -> str: """Generate, persist (hashed) and email a fresh code for ``purpose``. Enforces the hourly cap always and the resend cooldown when ``enforce_cooldown`` is set. Issuing a new code invalidates the previous one for this purpose. ``email`` overrides the delivery address (defaults to ``user.email``) for flows that verify a new address. ``event`` labels the audit entry (``"issued"`` or ``"resent"``). ``expiry_minutes`` optionally overrides the shared ``OTP_EXPIRY_MINUTES`` lifetime for this one purpose (e.g. login codes live longer than registration codes) — the emailed "expires in N minutes" copy stays in sync. Returns the plain code for internal/test use — it is **never logged or stored in the clear**. Raises :class:`OtpRateLimited` when a code cannot be issued yet. """ purpose = _as_purpose(purpose) if requests_last_hour(user, purpose) >= _cfg("OTP_MAX_PER_HOUR", 5): raise OtpRateLimited( "You've requested too many codes. Please try again in a little while." ) if enforce_cooldown: wait = seconds_until_resend(user, purpose) if wait > 0: raise OtpRateLimited( f"Please wait {wait} second{'s' if wait != 1 else ''} before requesting a new code." ) invalidate_codes(user, purpose) minutes = expiry_minutes or _cfg("OTP_EXPIRY_MINUTES", 5) code = generate_code() now = utcnow() record = EmailVerification( user_id=user.id, purpose=purpose, verification_token=hash_code(user.id, purpose, code), expires_at=now + timedelta(minutes=minutes), ) db.session.add(record) db.session.commit() # OTP persisted BEFORE sending — never sent from a rolled-back txn delivered = _send_email(user, purpose, code, email=email, minutes=minutes) audit_event(event, user, purpose, verification=record.id, delivered=delivered) if not delivered: current_app.logger.warning( "Verification code #%s created but email delivery FAILED for user %s " "(purpose=%s) — they will not receive a code until delivery is fixed.", record.id, user.id, purpose.value, ) return code def ensure_active_code(user, purpose=DEFAULT_PURPOSE, *, expiry_minutes: int | None = None) -> bool: """Send a code only if none is currently live for this purpose (used by the register/login gate so we don't email on every page hit). Returns ``True`` if a code was sent. ``expiry_minutes`` optionally sets a per-purpose lifetime. Silently respects the hourly cap.""" purpose = _as_purpose(purpose) if _active_code(user, purpose) is not None: return False try: issue_verification_code(user, purpose, enforce_cooldown=False, expiry_minutes=expiry_minutes) return True except OtpRateLimited: return False # --- Verify a code (generic: validate + consume, no side effects) -------------- def verify_code(user, purpose, code: str) -> tuple[bool, str | None]: """Validate ``code`` against the active code for ``user`` + ``purpose``. On success the code is marked redeemed (single-use). Returns ``(ok, error_message)``. Applies **no** purpose-specific consequence — the caller does that (e.g. mark the email verified, reset the password), keeping the framework generic. Audits verified / failed / expired. """ purpose = _as_purpose(purpose) code = (code or "").strip() length = _cfg("OTP_LENGTH", 6) if not code.isdigit() or len(code) != length: return False, f"Enter the {length}-digit code from your email." record = _active_code(user, purpose) if record is None: latest = _latest_request(user, purpose) if latest is not None and latest.verified_at is None: audit_event("expired", user, purpose) return False, "This code has expired. Please request a new one." return False, "No active code. Please request a new one." if not hmac.compare_digest(record.verification_token, hash_code(user.id, purpose, code)): audit_event("failed", user, purpose, verification=record.id) return False, "Incorrect code. Please check your email and try again." record.verified_at = utcnow() db.session.commit() audit_event("verified", user, purpose, verification=record.id) return True, None # --- Email delivery (reuses Flask-Mail; shared by every purpose) ---------------- # # Deliverability notes: emails are sent as MIME multipart/alternative (plain text # + branded HTML), UTF-8, with a real Reply-To (a monitored support mailbox), and # Flask-Mail auto-adds a Date and a unique Message-ID at send time. The copy # avoids spam-trigger wording (no ALL CAPS / "FREE" / "URGENT" / "!!!"). These are # the application-level levers only; once Student Bazaar has its own sending # domain, configure **SPF, DKIM and DMARC** DNS records for that domain — those # are the decisive deliverability signals and cannot be set from the app. Nothing # here is Gmail-specific. # Per-purpose message metadata. Adding a purpose is one entry — the layout, # branding and delivery mechanics are shared. _PURPOSE_META = { VerificationPurpose.EMAIL_VERIFICATION: { "subject": "Verify your Student Bazaar account", "heading": "Verify your email address", "intro": "Welcome to Student Bazaar! Please confirm your email address to activate your account.", "template": "verification", }, VerificationPurpose.PASSWORD_RESET: { "subject": "Reset your Student Bazaar password", "heading": "Reset your password", "intro": "We received a request to reset your Student Bazaar password.", "template": "password_reset", }, VerificationPurpose.EMAIL_CHANGE: { "subject": "Confirm your new email address", "heading": "Confirm your email address", "intro": "Please confirm this email address for your Student Bazaar account.", "template": "verification", }, VerificationPurpose.LOGIN_VERIFICATION: { "subject": "Your Student Bazaar login code", "heading": "Confirm your sign-in", "intro": "Use the code below to finish signing in to Student Bazaar.", "template": "verification", }, VerificationPurpose.TWO_FACTOR: { "subject": "Your Student Bazaar security code", "heading": "Your security code", "intro": "Use the code below to complete your security check.", "template": "verification", }, } _DEFAULT_META = { "subject": "Your Student Bazaar verification code", "heading": "Your verification code", "intro": "Use the code below to continue.", "template": "verification", } def render_verification_email(purpose, code: str, minutes: int) -> tuple[str, str, str]: """Render ``(subject, text_body, html_body)`` from the shared email templates. HTML and plain text are rendered from Jinja templates under ``templates/emails/`` so branding lives in one place (never in Python strings) and the plain-text alternative always matches the HTML.""" from flask import render_template purpose = _as_purpose(purpose) meta = _PURPOSE_META.get(purpose, _DEFAULT_META) ctx = { "code": code, "minutes": minutes, "heading": meta["heading"], "intro": meta["intro"], "support_email": current_app.config.get("SUPPORT_EMAIL", "support@studentbazaar.in"), } text_body = render_template(f"emails/{meta['template']}.txt", **ctx) html_body = render_template(f"emails/{meta['template']}.html", **ctx) return meta["subject"], text_body, html_body def _build_message(subject: str, recipient: str, text_body: str, html_body: str) -> Message: """Construct a well-formed MIME multipart message (text + HTML) with a monitored Reply-To. Date and Message-ID are added by Flask-Mail on send.""" msg = Message(subject=subject, recipients=[recipient]) msg.body = text_body # plain-text alternative (accessibility + deliverability) msg.html = html_body # branded HTML msg.charset = "utf-8" reply_to = current_app.config.get("MAIL_REPLY_TO") if reply_to: msg.reply_to = reply_to return msg def send_email(recipient: str, subject: str, text_body: str, html_body: str, *, label: str = "email") -> bool: """Send one MIME multipart (text + HTML) email via Flask-Mail. **The single outbound-send path** — every account email (OTP codes, security notifications) funnels through here so the MIME/Reply-To construction and the success/failure logging live in exactly one place. Best-effort (a delivery failure never breaks the calling flow) but **never silent**: failures are logged at ERROR with the full SMTP traceback. ``label`` names the email in the logs. Returns whether it was handed to the SMTP server.""" server = current_app.config.get("MAIL_SERVER") port = current_app.config.get("MAIL_PORT") logger = current_app.logger try: logger.info("Sending %s email to %s via %s:%s", label, recipient, server, port) mail.send(_build_message(subject, recipient, text_body, html_body)) logger.info("Sent %s email to %s", label, recipient) return True except Exception: logger.error( "FAILED to send %s email to %s via %s:%s (suppress=%s). " "Check MAIL_* config / SMTP credentials.", label, recipient, server, port, current_app.config.get("MAIL_SUPPRESS_SEND"), exc_info=True, ) return False def send_verification_email(recipient: str, purpose, code: str, minutes: int | None = None) -> bool: """Render and send a verification email to ``recipient``. **Single sender** used by both the user-bound flow and pre-user registration. ``minutes`` overrides the "expires in N minutes" copy (defaults to the shared ``OTP_EXPIRY_MINUTES``) so a per-purpose lifetime is reflected accurately. Delegates delivery to :func:`send_email`.""" purpose = _as_purpose(purpose) minutes = minutes or _cfg("OTP_EXPIRY_MINUTES", 5) subject, text_body, html_body = render_verification_email(purpose, code, minutes) return send_email(recipient, subject, text_body, html_body, label=purpose.value) def _send_email(user, purpose: VerificationPurpose, code: str, *, email=None, minutes=None) -> bool: """User-bound wrapper over :func:`send_verification_email` (delivery address defaults to the user's email).""" return send_verification_email(email or user.email, purpose, code, minutes=minutes)