refactor(auth): split store, schemas, deps, totp out of auth __init__.py (P4.3 cont)
This commit is contained in:
parent
3eb22495b1
commit
b31b564f36
5 changed files with 353 additions and 325 deletions
|
|
@ -2,20 +2,19 @@
|
|||
Auth Router - Complete authentication system (email, wallet, OAuth, Telegram)
|
||||
"""
|
||||
|
||||
import base64
|
||||
import io
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import secrets
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Request
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.core.redis import get_redis
|
||||
from app.domains.auth.deps import (
|
||||
get_current_user,
|
||||
require_auth as require_auth,
|
||||
require_public_profile as require_public_profile,
|
||||
)
|
||||
from app.domains.auth.jwt import (
|
||||
JWT_ALGORITHM,
|
||||
JWT_EXPIRY_DAYS as JWT_EXPIRY_DAYS,
|
||||
|
|
@ -26,227 +25,53 @@ from app.domains.auth.jwt import (
|
|||
generate_nonce,
|
||||
)
|
||||
from app.domains.auth.passwords import hash_password, verify_password
|
||||
from app.domains.auth.schemas import (
|
||||
EmailLoginRequest,
|
||||
EmailRegisterRequest,
|
||||
GoogleAuthResponse,
|
||||
NonceResponse,
|
||||
TelegramAuthRequest,
|
||||
TwoFAEnableRequest,
|
||||
TwoFALoginRequest,
|
||||
TwoFASetupResponse as TwoFASetupResponse,
|
||||
TwoFAStatusResponse,
|
||||
TwoFAVerifyRequest,
|
||||
UserResponse,
|
||||
WalletAuthResponse,
|
||||
WalletNonceRequest,
|
||||
WalletVerifyRequest,
|
||||
)
|
||||
from app.domains.auth.store import (
|
||||
_delete_user as _delete_user,
|
||||
_get_user,
|
||||
_get_user_by_email,
|
||||
_is_valid_email,
|
||||
_save_user,
|
||||
)
|
||||
from app.domains.auth.totp import (
|
||||
PYOTP_AVAILABLE,
|
||||
TOTP_DIGITS as TOTP_DIGITS,
|
||||
TOTP_INTERVAL as TOTP_INTERVAL,
|
||||
TOTP_ISSUER as TOTP_ISSUER,
|
||||
TOTP_WINDOW as TOTP_WINDOW,
|
||||
_build_totp as _build_totp,
|
||||
_generate_backup_codes,
|
||||
_generate_qr_base64,
|
||||
_generate_totp_secret,
|
||||
_get_totp_uri,
|
||||
_hash_backup_code,
|
||||
_verify_backup_code,
|
||||
_verify_totp,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(tags=["auth"])
|
||||
|
||||
# ── 2FA / TOTP ──
|
||||
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 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 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 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:
|
||||
"""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)
|
||||
|
||||
|
||||
# ── Config ──
|
||||
|
||||
|
||||
# ── Storage (Redis-backed user store) ──
|
||||
def _get_user(user_id: str) -> dict[str, Any] | None:
|
||||
"""Fetch user record by id from Redis hash rmi:users."""
|
||||
r = get_redis()
|
||||
if not r:
|
||||
return None
|
||||
raw = r.hget("rmi:users", user_id)
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
return json.loads(raw)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
def _get_user_by_email(email: str) -> dict[str, Any] | None:
|
||||
"""Fetch user record by email via rmi:users:email index → rmi:users hash."""
|
||||
r = get_redis()
|
||||
if not r:
|
||||
return None
|
||||
user_id = r.hget("rmi:users:email", email.lower())
|
||||
if not user_id:
|
||||
return None
|
||||
if isinstance(user_id, bytes):
|
||||
user_id = user_id.decode("utf-8")
|
||||
return _get_user(user_id)
|
||||
|
||||
|
||||
def _save_user(user: dict[str, Any]) -> None:
|
||||
"""Persist user record to Redis hash rmi:users, refresh email index if present."""
|
||||
r = get_redis()
|
||||
if not r:
|
||||
return
|
||||
user_id = user["id"]
|
||||
r.hset("rmi:users", user_id, json.dumps(user))
|
||||
if user.get("email"):
|
||||
r.hset("rmi:users:email", user["email"].lower(), user_id)
|
||||
|
||||
|
||||
# ── Helpers ──
|
||||
def _is_valid_email(email: str) -> bool:
|
||||
"""Basic email validation."""
|
||||
pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
|
||||
return re.match(pattern, email) is not None
|
||||
|
||||
|
||||
# ── Storage (Redis-backed user store) ──
|
||||
class EmailLoginRequest(BaseModel):
|
||||
email: str
|
||||
password: str
|
||||
|
||||
|
||||
class EmailRegisterRequest(BaseModel):
|
||||
email: str
|
||||
password: str
|
||||
display_name: str
|
||||
wallet_address: str | None = None
|
||||
wallet_chain: str | None = None
|
||||
|
||||
|
||||
class WalletNonceRequest(BaseModel):
|
||||
address: str
|
||||
chain: str
|
||||
|
||||
|
||||
class WalletVerifyRequest(BaseModel):
|
||||
address: str
|
||||
chain: str
|
||||
nonce: str
|
||||
signature: str
|
||||
message: str
|
||||
|
||||
|
||||
class NonceResponse(BaseModel):
|
||||
nonce: str
|
||||
timestamp: str
|
||||
message: str | None = None
|
||||
|
||||
|
||||
class WalletAuthResponse(BaseModel):
|
||||
access_token: str
|
||||
refresh_token: str
|
||||
user: dict[str, Any]
|
||||
|
||||
|
||||
class UserResponse(BaseModel):
|
||||
id: str
|
||||
email: str
|
||||
display_name: str
|
||||
wallet_address: str | None
|
||||
wallet_chain: str | None
|
||||
tier: str
|
||||
role: str
|
||||
created_at: str
|
||||
xp: int = 0
|
||||
level: int = 1
|
||||
badges: list = []
|
||||
|
||||
|
||||
class GoogleAuthResponse(BaseModel):
|
||||
url: str
|
||||
|
||||
|
||||
class TelegramAuthRequest(BaseModel):
|
||||
id: int
|
||||
first_name: str | None = None
|
||||
last_name: str | None = None
|
||||
username: str | None = None
|
||||
photo_url: str | None = None
|
||||
auth_date: int
|
||||
hash: str
|
||||
|
||||
|
||||
# ── 2FA Models ──
|
||||
class TwoFASetupResponse(BaseModel):
|
||||
secret: str
|
||||
qr_code: str # base64 data URI
|
||||
uri: str # otpauth:// URI
|
||||
backup_codes: list # plaintext - shown once
|
||||
|
||||
|
||||
class TwoFAEnableRequest(BaseModel):
|
||||
code: str = Field(..., min_length=6, max_length=6, pattern=r"^\d{6}$")
|
||||
|
||||
|
||||
class TwoFAVerifyRequest(BaseModel):
|
||||
code: str = Field(..., min_length=6, max_length=6, pattern=r"^\d{6}$")
|
||||
|
||||
|
||||
class TwoFALoginRequest(BaseModel):
|
||||
email: str
|
||||
password: str
|
||||
code: str = Field(..., min_length=6, max_length=10) # 6 digits or backup code XXXX-XXXX
|
||||
|
||||
|
||||
class TwoFAStatusResponse(BaseModel):
|
||||
enabled: bool
|
||||
method: str = "totp" # future: u2f, webauthn
|
||||
created_at: str | None = None
|
||||
last_used: str | None = None
|
||||
|
||||
|
||||
# ── Email Auth ──
|
||||
@router.post("/register", response_model=WalletAuthResponse)
|
||||
def register_email(req: EmailRegisterRequest):
|
||||
|
|
@ -416,7 +241,7 @@ async def wallet_verify(req: WalletVerifyRequest):
|
|||
|
||||
# ── User Profile ──
|
||||
@router.get("/user/me", response_model=UserResponse)
|
||||
async def get_current_user(request: Request):
|
||||
async def user_me(request: Request):
|
||||
"""Get current authenticated user profile."""
|
||||
auth_header = request.headers.get("Authorization", "")
|
||||
if not auth_header.startswith("Bearer "):
|
||||
|
|
@ -947,30 +772,6 @@ async def x_callback(request: Request):
|
|||
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?token={jwt_token}&provider=x")
|
||||
|
||||
|
||||
# ── FastAPI Dependency Functions (for @router/@app endpoints) ──
|
||||
async def get_current_user(request: Request) -> dict[str, Any] | None: # noqa: F811
|
||||
"""Verify JWT from Authorization header (wallet or email)."""
|
||||
auth_header = request.headers.get("Authorization", "")
|
||||
if not auth_header.startswith("Bearer "):
|
||||
return None
|
||||
token = auth_header[7:]
|
||||
payload = _verify_jwt(token)
|
||||
if not payload:
|
||||
return None
|
||||
user = _get_user(payload["id"])
|
||||
if not user:
|
||||
return None
|
||||
return user
|
||||
|
||||
|
||||
async def require_auth(request: Request) -> dict[str, Any]:
|
||||
"""Require valid JWT. Raises 401 if missing/invalid."""
|
||||
user = await get_current_user(request)
|
||||
if not user:
|
||||
raise HTTPException(status_code=401, detail="Authentication required")
|
||||
return user
|
||||
|
||||
|
||||
# ── 2FA / TOTP Endpoints ──
|
||||
|
||||
|
||||
|
|
@ -1095,30 +896,6 @@ async def twofa_status(request: Request):
|
|||
}
|
||||
|
||||
|
||||
# ── Privacy Enforcement Dependency ──
|
||||
|
||||
|
||||
async def require_public_profile(request: Request):
|
||||
"""
|
||||
Dependency to ensure the user has a public profile.
|
||||
Required for actions that interact with the community (posting, commenting).
|
||||
"""
|
||||
user = await get_current_user(request)
|
||||
if not user:
|
||||
raise HTTPException(status_code=401, detail="Authentication required")
|
||||
|
||||
privacy_settings = user.get("privacy_settings", {})
|
||||
visibility = privacy_settings.get("profile_visibility", "public")
|
||||
|
||||
if visibility != "public":
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail="A public profile is required to post or comment. Please update your privacy settings in your profile.",
|
||||
)
|
||||
|
||||
return user
|
||||
|
||||
|
||||
@router.post("/2fa/verify")
|
||||
async def twofa_verify(req: TwoFAVerifyRequest, request: Request):
|
||||
"""Verify a TOTP code (for re-auth during sensitive operations)."""
|
||||
|
|
|
|||
|
|
@ -1,12 +1,61 @@
|
|||
"""FastAPI dependencies - re-exports from app.domains.auth.
|
||||
"""FastAPI dependencies for auth domain.
|
||||
|
||||
Phase 3B of AUDIT-2026-Q3.md.
|
||||
|
||||
Public API:
|
||||
- get_current_user, require_auth, require_public_profile
|
||||
- get_current_user
|
||||
- require_auth
|
||||
- require_public_profile
|
||||
"""
|
||||
from app.auth import ( # noqa: F401
|
||||
get_current_user,
|
||||
require_auth,
|
||||
require_public_profile,
|
||||
)
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from fastapi import HTTPException, Request
|
||||
|
||||
from app.domains.auth.jwt import _verify_jwt
|
||||
from app.domains.auth.store import _get_user
|
||||
|
||||
|
||||
async def get_current_user(request: Request) -> dict[str, Any] | None:
|
||||
"""Verify JWT from Authorization header (wallet or email)."""
|
||||
auth_header = request.headers.get("Authorization", "")
|
||||
if not auth_header.startswith("Bearer "):
|
||||
return None
|
||||
token = auth_header[7:]
|
||||
payload = _verify_jwt(token)
|
||||
if not payload:
|
||||
return None
|
||||
user = _get_user(payload["id"])
|
||||
if not user:
|
||||
return None
|
||||
return user
|
||||
|
||||
|
||||
async def require_auth(request: Request) -> dict[str, Any]:
|
||||
"""Require valid JWT. Raises 401 if missing/invalid."""
|
||||
user = await get_current_user(request)
|
||||
if not user:
|
||||
raise HTTPException(status_code=401, detail="Authentication required")
|
||||
return user
|
||||
|
||||
|
||||
async def require_public_profile(request: Request) -> dict[str, Any]:
|
||||
"""
|
||||
Dependency to ensure the user has a public profile.
|
||||
Required for actions that interact with the community (posting, commenting).
|
||||
"""
|
||||
user = await get_current_user(request)
|
||||
if not user:
|
||||
raise HTTPException(status_code=401, detail="Authentication required")
|
||||
|
||||
privacy_settings = user.get("privacy_settings", {})
|
||||
visibility = privacy_settings.get("profile_visibility", "public")
|
||||
|
||||
if visibility != "public":
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail="A public profile is required to post or comment. Please update your privacy settings in your profile.",
|
||||
)
|
||||
|
||||
return user
|
||||
|
|
|
|||
|
|
@ -1,27 +1,106 @@
|
|||
"""Pydantic schemas - re-exports from app.domains.auth.
|
||||
"""Pydantic schemas for auth domain.
|
||||
|
||||
Phase 3B of AUDIT-2026-Q3.md.
|
||||
|
||||
Public API:
|
||||
- EmailLoginRequest, EmailRegisterRequest
|
||||
- WalletNonceRequest, WalletVerifyRequest, WalletAuthResponse
|
||||
- NonceResponse, UserResponse, GoogleAuthResponse
|
||||
- TelegramAuthRequest, TwoFAEnableRequest, TwoFALoginRequest,
|
||||
TwoFASetupResponse, TwoFAStatusResponse, TwoFAVerifyRequest
|
||||
"""
|
||||
from app.auth import ( # noqa: F401
|
||||
EmailLoginRequest,
|
||||
EmailRegisterRequest,
|
||||
GoogleAuthResponse,
|
||||
NonceResponse,
|
||||
TelegramAuthRequest,
|
||||
TwoFAEnableRequest,
|
||||
TwoFALoginRequest,
|
||||
TwoFASetupResponse,
|
||||
TwoFAStatusResponse,
|
||||
TwoFAVerifyRequest,
|
||||
UserResponse,
|
||||
WalletAuthResponse,
|
||||
WalletNonceRequest,
|
||||
WalletVerifyRequest,
|
||||
)
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class EmailLoginRequest(BaseModel):
|
||||
email: str
|
||||
password: str
|
||||
|
||||
|
||||
class EmailRegisterRequest(BaseModel):
|
||||
email: str
|
||||
password: str
|
||||
display_name: str
|
||||
wallet_address: str | None = None
|
||||
wallet_chain: str | None = None
|
||||
|
||||
|
||||
class WalletNonceRequest(BaseModel):
|
||||
address: str
|
||||
chain: str
|
||||
|
||||
|
||||
class WalletVerifyRequest(BaseModel):
|
||||
address: str
|
||||
chain: str
|
||||
nonce: str
|
||||
signature: str
|
||||
message: str
|
||||
|
||||
|
||||
class NonceResponse(BaseModel):
|
||||
nonce: str
|
||||
timestamp: str
|
||||
message: str | None = None
|
||||
|
||||
|
||||
class WalletAuthResponse(BaseModel):
|
||||
access_token: str
|
||||
refresh_token: str
|
||||
user: dict[str, Any]
|
||||
|
||||
|
||||
class UserResponse(BaseModel):
|
||||
id: str
|
||||
email: str
|
||||
display_name: str
|
||||
wallet_address: str | None
|
||||
wallet_chain: str | None
|
||||
tier: str
|
||||
role: str
|
||||
created_at: str
|
||||
xp: int = 0
|
||||
level: int = 1
|
||||
badges: list = []
|
||||
|
||||
|
||||
class GoogleAuthResponse(BaseModel):
|
||||
url: str
|
||||
|
||||
|
||||
class TelegramAuthRequest(BaseModel):
|
||||
id: int
|
||||
first_name: str | None = None
|
||||
last_name: str | None = None
|
||||
username: str | None = None
|
||||
photo_url: str | None = None
|
||||
auth_date: int
|
||||
hash: str
|
||||
|
||||
|
||||
# ── 2FA Models ──
|
||||
class TwoFASetupResponse(BaseModel):
|
||||
secret: str
|
||||
qr_code: str # base64 data URI
|
||||
uri: str # otpauth:// URI
|
||||
backup_codes: list # plaintext - shown once
|
||||
|
||||
|
||||
class TwoFAEnableRequest(BaseModel):
|
||||
code: str = Field(..., min_length=6, max_length=6, pattern=r"^\d{6}$")
|
||||
|
||||
|
||||
class TwoFAVerifyRequest(BaseModel):
|
||||
code: str = Field(..., min_length=6, max_length=6, pattern=r"^\d{6}$")
|
||||
|
||||
|
||||
class TwoFALoginRequest(BaseModel):
|
||||
email: str
|
||||
password: str
|
||||
code: str = Field(..., min_length=6, max_length=10) # 6 digits or backup code XXXX-XXXX
|
||||
|
||||
|
||||
class TwoFAStatusResponse(BaseModel):
|
||||
enabled: bool
|
||||
method: str = "totp" # future: u2f, webauthn
|
||||
created_at: str | None = None
|
||||
last_used: str | None = None
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
"""User storage helpers - re-exports from app.domains.auth.
|
||||
"""User storage helpers.
|
||||
|
||||
Phase 3B of AUDIT-2026-Q3.md.
|
||||
|
||||
|
|
@ -6,10 +6,65 @@ Public API:
|
|||
- _get_user, _get_user_by_email, _save_user, _delete_user
|
||||
- _is_valid_email
|
||||
"""
|
||||
from app.auth import ( # noqa: F401
|
||||
_delete_user,
|
||||
_get_user,
|
||||
_get_user_by_email,
|
||||
_is_valid_email,
|
||||
_save_user,
|
||||
)
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from app.core.redis import get_redis
|
||||
|
||||
|
||||
def _get_user(user_id: str) -> dict[str, Any] | None:
|
||||
"""Fetch user record by id from Redis hash rmi:users."""
|
||||
r = get_redis()
|
||||
if not r:
|
||||
return None
|
||||
raw = r.hget("rmi:users", user_id)
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
return json.loads(raw)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
def _get_user_by_email(email: str) -> dict[str, Any] | None:
|
||||
"""Fetch user record by email via rmi:users:email index -> rmi:users hash."""
|
||||
r = get_redis()
|
||||
if not r:
|
||||
return None
|
||||
user_id = r.hget("rmi:users:email", email.lower())
|
||||
if not user_id:
|
||||
return None
|
||||
if isinstance(user_id, bytes):
|
||||
user_id = user_id.decode("utf-8")
|
||||
return _get_user(user_id)
|
||||
|
||||
|
||||
def _save_user(user: dict[str, Any]) -> None:
|
||||
"""Persist user record to Redis hash rmi:users, refresh email index if present."""
|
||||
r = get_redis()
|
||||
if not r:
|
||||
return
|
||||
user_id = user["id"]
|
||||
r.hset("rmi:users", user_id, json.dumps(user))
|
||||
if user.get("email"):
|
||||
r.hset("rmi:users:email", user["email"].lower(), user_id)
|
||||
|
||||
|
||||
def _delete_user(user_id: str) -> None:
|
||||
"""Remove user record from Redis hash rmi:users and email index."""
|
||||
r = get_redis()
|
||||
if not r:
|
||||
return
|
||||
user = _get_user(user_id)
|
||||
if user and user.get("email"):
|
||||
r.hdel("rmi:users:email", user["email"].lower())
|
||||
r.hdel("rmi:users", user_id)
|
||||
|
||||
|
||||
def _is_valid_email(email: str) -> bool:
|
||||
"""Basic email validation."""
|
||||
pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
|
||||
return re.match(pattern, email) is not None
|
||||
|
|
|
|||
|
|
@ -1,24 +1,92 @@
|
|||
"""TOTP / 2FA helpers - re-exports from app.domains.auth.
|
||||
"""TOTP / 2FA helpers.
|
||||
|
||||
Phase 3B of AUDIT-2026-Q3.md.
|
||||
|
||||
Public API:
|
||||
- _generate_totp_secret, _build_totp, _verify_totp, _get_totp_uri
|
||||
- _generate_qr_base64, _generate_backup_codes, _hash_backup_code, _verify_backup_code
|
||||
- TOTP_ISSUER, PYOTP_AVAILABLE
|
||||
- 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 app.auth import ( # noqa: F401
|
||||
PYOTP_AVAILABLE,
|
||||
TOTP_DIGITS,
|
||||
TOTP_INTERVAL,
|
||||
TOTP_ISSUER,
|
||||
TOTP_WINDOW,
|
||||
_build_totp,
|
||||
_generate_backup_codes,
|
||||
_generate_qr_base64,
|
||||
_generate_totp_secret,
|
||||
_get_totp_uri,
|
||||
_hash_backup_code,
|
||||
_verify_backup_code,
|
||||
_verify_totp,
|
||||
)
|
||||
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 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 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 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:
|
||||
"""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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue