60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
 
|
"""TOTP two-factor authentication (SPEC §6.1).
|
||
|
|
|
||
|
|
The base32 secret is stored envelope-encrypted in ``users.totp_secret``. A pending
|
||
|
|
(not-yet-enabled) secret lives in Redis until the user confirms one valid code.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import uuid
|
||
|
|
|
||
|
|
import pyotp
|
||
|
|
|
||
|
|
from app.config import get_settings
|
||
|
|
from app.redis_client import get_redis
|
||
|
|
from app.security.crypto import decrypt_secret, encrypt_secret
|
||
|
|
|
||
|
|
_PENDING_PREFIX = "totp_pending:"
|
||
|
|
_PENDING_TTL = 600 # 10 minutes to complete enrolment
|
||
|
|
_VALID_WINDOW = 1 # ±1 step tolerance (SPEC §6.1)
|
||
|
|
|
||
|
|
|
||
|
|
def generate_secret() -> str:
|
||
|
|
"""Return a fresh base32 TOTP secret."""
|
||
|
|
return pyotp.random_base32()
|
||
|
|
|
||
|
|
|
||
|
|
def provisioning_uri(secret: str, handle: str) -> str:
|
||
|
|
"""Return the otpauth:// URI for QR provisioning."""
|
||
|
|
return pyotp.TOTP(secret).provisioning_uri(name=handle, issuer_name=get_settings().app_domain)
|
||
|
|
|
||
|
|
|
||
|
|
def verify(secret: str, code: str) -> bool:
|
||
|
|
"""Verify a TOTP ``code`` against ``secret`` (±1 window)."""
|
||
|
|
return bool(pyotp.TOTP(secret).verify(code.strip(), valid_window=_VALID_WINDOW))
|
||
|
|
|
||
|
|
|
||
|
|
async def start_setup(user_id: uuid.UUID) -> str:
|
||
|
|
"""Generate a pending secret (stored in Redis) and return it."""
|
||
|
|
secret = generate_secret()
|
||
|
|
await get_redis().set(_PENDING_PREFIX + str(user_id), secret, ex=_PENDING_TTL)
|
||
|
|
return secret
|
||
|
|
|
||
|
|
|
||
|
|
async def get_pending(user_id: uuid.UUID) -> str | None:
|
||
|
|
"""Return the pending (un-enabled) secret for a user, if any."""
|
||
|
|
secret: str | None = await get_redis().get(_PENDING_PREFIX + str(user_id))
|
||
|
|
return secret
|
||
|
|
|
||
|
|
|
||
|
|
async def clear_pending(user_id: uuid.UUID) -> None:
|
||
|
|
await get_redis().delete(_PENDING_PREFIX + str(user_id))
|
||
|
|
|
||
|
|
|
||
|
|
def encrypt_totp_secret(secret: str) -> bytes:
|
||
|
|
return encrypt_secret(secret.encode("utf-8"))
|
||
|
|
|
||
|
|
|
||
|
|
def decrypt_totp_secret(blob: bytes) -> str:
|
||
|
|
return decrypt_secret(blob).decode("utf-8")
|