rmi-backend/app/domains/auth/jwt.py
cryptorugmunch 0a8c73d99b feat(domains): consolidate bulletin, intelligence, markets, admin, referral, mcp + mypy gate
- 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.
2026-07-07 16:43:49 +07:00

66 lines
1.5 KiB
Python

"""JWT helpers.
Phase 3B of AUDIT-2026-Q3.md.
Public API:
- JWT_SECRET, JWT_ALGORITHM, JWT_EXPIRY_DAYS
- _create_jwt, _verify_jwt
- generate_nonce, _derive_user_id
"""
from __future__ import annotations
import hashlib
import os
import secrets
from datetime import datetime, timedelta
from typing import Any
from jose import JWTError, jwt
JWT_SECRET = os.getenv("JWT_SECRET", "rmi-jwt-secret-change-me")
JWT_ALGORITHM = "HS256"
JWT_EXPIRY_DAYS = 7
def _create_jwt(
user_id: str,
email: str,
tier: str = "FREE",
role: str = "USER",
wallet: str | None = None,
) -> str:
now = datetime.utcnow()
payload = {
"sub": user_id,
"email": email,
"tier": tier,
"role": role,
"iat": now,
"exp": now + timedelta(days=JWT_EXPIRY_DAYS),
}
if wallet:
payload["wallet"] = wallet
return str(jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM))
def _verify_jwt(token: str) -> dict[str, Any] | None:
try:
payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
return {
"id": payload["sub"],
"email": payload["email"],
"tier": payload.get("tier", "FREE"),
"role": payload.get("role", "USER"),
"wallet": payload.get("wallet"),
}
except JWTError:
return None
def _derive_user_id(email: str) -> str:
return hashlib.sha256(email.lower().encode()).hexdigest()[:32]
def generate_nonce() -> str:
return secrets.token_urlsafe(16)