Spaces:
Runtime error
Runtime error
File size: 4,508 Bytes
00a4363 1e1198f 00a4363 1e1198f 00a4363 1e1198f 00a4363 1e1198f 00a4363 1e1198f 00a4363 | 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 117 118 119 120 121 122 123 124 125 126 127 128 129 | """6 digit codes: hashed, 10 minute TTL, 5 attempts, 60s resend cooldown, purged after 1 day."""
import hashlib
import hmac
import secrets
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from sqlalchemy import delete, select
from sqlalchemy.orm import Session
from app.config import settings
from app.models import EmailVerificationCode, VerificationPurpose
CODE_TTL = timedelta(minutes=10)
MAX_ATTEMPTS = 5
RESEND_COOLDOWN = timedelta(seconds=60)
PURGE_AFTER = timedelta(days=1)
class VerificationError(Exception):
def __init__(self, code: str, message: str):
super().__init__(message)
self.code = code
self.message = message
@dataclass
class IssuedCode:
code: str
record: EmailVerificationCode
def _hash(email: str, code: str) -> str:
return hmac.new(settings.jwt_secret.encode(), f"{email}:{code}".encode(), hashlib.sha256).hexdigest()
def _aware(dt: datetime) -> datetime:
return dt if dt.tzinfo else dt.replace(tzinfo=UTC)
def _latest(db: Session, email: str, purpose: VerificationPurpose) -> EmailVerificationCode | None:
return db.scalar(
select(EmailVerificationCode)
.where(
EmailVerificationCode.email == email,
EmailVerificationCode.purpose == purpose,
EmailVerificationCode.consumed_at.is_(None),
)
.order_by(EmailVerificationCode.created_at.desc())
.limit(1)
)
def _alive(record: EmailVerificationCode, now: datetime) -> bool:
"""A code that can still be entered. Dead codes (expired, out of tries) never hold up a resend."""
return now <= _aware(record.expires_at) and record.attempts < MAX_ATTEMPTS
def issue_code(
db: Session,
email: str,
purpose: VerificationPurpose,
pending_name: str | None = None,
pending_password_hash: str | None = None,
) -> IssuedCode:
now = datetime.now(UTC)
latest = _latest(db, email, purpose)
if latest and _alive(latest, now) and now - _aware(latest.created_at) < RESEND_COOLDOWN:
raise VerificationError("cooldown", "Please wait a minute before requesting another code.")
clear_codes(db, email, purpose)
code = f"{secrets.randbelow(1_000_000):06d}"
record = EmailVerificationCode(
email=email,
purpose=purpose,
code_hash=_hash(email, code),
pending_name=pending_name,
pending_password_hash=pending_password_hash,
expires_at=now + CODE_TTL,
)
db.add(record)
db.flush()
return IssuedCode(code=code, record=record)
def check_code(db: Session, email: str, purpose: VerificationPurpose, code: str) -> EmailVerificationCode:
"""Validate without consuming (used by the password reset 'check' step)."""
record = _latest(db, email, purpose)
if not record:
raise VerificationError("not_found", "No active code. Request a new one.")
if record.attempts >= MAX_ATTEMPTS:
raise VerificationError("too_many_attempts", "Too many wrong tries. This code has expired. Request a new one.")
if datetime.now(UTC) > _aware(record.expires_at):
raise VerificationError("expired", "That code expired. Request a new one.")
if not hmac.compare_digest(record.code_hash, _hash(email, code.strip())):
record.attempts += 1
left = MAX_ATTEMPTS - record.attempts
if left <= 0:
# Expire rather than consume: sign up resend still needs the pending name and password.
record.expires_at = datetime.now(UTC)
db.commit()
raise VerificationError("too_many_attempts", "Too many wrong tries. This code has expired. Request a new one.")
db.commit()
raise VerificationError("invalid", f"That code is not right. {left} {'try' if left == 1 else 'tries'} left.")
return record
def consume_code(db: Session, email: str, purpose: VerificationPurpose, code: str) -> EmailVerificationCode:
record = check_code(db, email, purpose, code)
record.consumed_at = datetime.now(UTC)
return record
def clear_codes(db: Session, email: str, purpose: VerificationPurpose) -> None:
db.execute(
delete(EmailVerificationCode).where(
EmailVerificationCode.email == email, EmailVerificationCode.purpose == purpose
)
)
def purge_old_codes(db: Session) -> None:
db.execute(
delete(EmailVerificationCode).where(
EmailVerificationCode.created_at < datetime.now(UTC) - PURGE_AFTER
)
)
db.commit()
|