cryptorugmunch
0a8c73d99b
- Make app/domains/auth/ and app/core/redis.py mypy-clean under strict. - Add mypy-gate.ini and Makefile mypy-gate target; promote typecheck-gate in CI. - Consolidate domains into app/domains/: bulletin, admin, intelligence, markets. - Extract referral domain incl. DeFi partner DEX ref links; keep Telegram bot wired. - Move app/mcp/ package and app/api/v1/mcp/router into app/domains/mcp/. - Archive dead app/mcp_router.py.
92 lines
2.6 KiB
Python
92 lines
2.6 KiB
Python
"""TOTP / 2FA helpers.
|
|
|
|
Phase 3B of AUDIT-2026-Q3.md.
|
|
|
|
Public API:
|
|
- PYOTP_AVAILABLE
|
|
- TOTP_ISSUER, TOTP_DIGITS, TOTP_INTERVAL, TOTP_WINDOW
|
|
- _generate_totp_secret, _build_totp, _verify_totp
|
|
- _get_totp_uri, _generate_qr_base64
|
|
- _generate_backup_codes, _hash_backup_code, _verify_backup_code
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import io
|
|
import logging
|
|
import os
|
|
import secrets
|
|
from typing import Any
|
|
|
|
from app.domains.auth.passwords import hash_password, verify_password
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
try:
|
|
import pyotp
|
|
import qrcode
|
|
|
|
PYOTP_AVAILABLE = True
|
|
except ImportError:
|
|
PYOTP_AVAILABLE = False
|
|
logger.warning("pyotp not installed - 2FA endpoints will return 503")
|
|
|
|
TOTP_ISSUER = os.getenv("TOTP_ISSUER", "RugMunch Intelligence")
|
|
TOTP_DIGITS = 6
|
|
TOTP_INTERVAL = 30
|
|
TOTP_WINDOW = 1 # Allow ±1 interval clock drift
|
|
|
|
|
|
def _generate_totp_secret() -> str:
|
|
"""Generate a new base32 TOTP secret."""
|
|
if not PYOTP_AVAILABLE:
|
|
raise RuntimeError("pyotp not installed")
|
|
return str(pyotp.random_base32())
|
|
|
|
|
|
def _build_totp(secret: str) -> Any:
|
|
"""Build a TOTP object from secret."""
|
|
return pyotp.TOTP(secret, digits=TOTP_DIGITS, interval=TOTP_INTERVAL)
|
|
|
|
|
|
def _verify_totp(secret: str, code: str) -> bool:
|
|
"""Verify a TOTP code with clock-drift tolerance."""
|
|
if not PYOTP_AVAILABLE or not secret or not code:
|
|
return False
|
|
totp = _build_totp(secret)
|
|
return bool(totp.verify(code, valid_window=TOTP_WINDOW))
|
|
|
|
|
|
def _get_totp_uri(secret: str, email: str) -> str:
|
|
"""Get otpauth:// URI for QR code generation."""
|
|
totp = _build_totp(secret)
|
|
return str(totp.provisioning_uri(name=email, issuer_name=TOTP_ISSUER))
|
|
|
|
|
|
def _generate_qr_base64(uri: str) -> str:
|
|
"""Generate QR code as base64 PNG data URI."""
|
|
img = qrcode.make(uri)
|
|
buf = io.BytesIO()
|
|
img.save(buf, format="PNG")
|
|
return "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode()
|
|
|
|
|
|
def _generate_backup_codes(count: int = 8) -> list[str]:
|
|
"""Generate single-use backup codes."""
|
|
codes = []
|
|
for _ in range(count):
|
|
# 8-digit numeric codes, hyphenated for readability
|
|
raw = secrets.randbelow(10_000_000_000)
|
|
code = f"{raw:010d}"[:8]
|
|
codes.append(f"{code[:4]}-{code[4:]}")
|
|
return codes
|
|
|
|
|
|
def _hash_backup_code(code: str) -> str:
|
|
"""Hash a backup code for storage (same as password hashing)."""
|
|
return hash_password(code)
|
|
|
|
|
|
def _verify_backup_code(code: str, hashed: str) -> bool:
|
|
"""Verify a backup code against its hash."""
|
|
return verify_password(code, hashed)
|