refactor(auth): split 1223-LOC god-file into auth package (P3B.4)

Move app/auth.py to app/auth/__init__.py and add thematic submodule files:
- app/auth/jwt.py - JWT encode/decode helpers
- app/auth/passwords.py - bcrypt hash/verify
- app/auth/totp.py - 2FA TOTP helpers
- app/auth/store.py - user storage (_get_user, _save_user, etc.)
- app/auth/schemas.py - Pydantic models
- app/auth/deps.py - FastAPI dependencies (get_current_user, etc.)
- app/auth/oauth.py - Google/GitHub/X/Telegram OAuth flows
- app/auth/wallet.py - wallet signature auth (migrated from auth_wallet.py)

The legacy app/auth.py and app/auth_wallet.py become re-export shims.
Public API fully preserved across all 18+ importers. Routes unchanged (56).

Phase 3B of AUDIT-2026-Q3.md.
This commit is contained in:
Crypto Rug Munch 2026-07-06 21:26:27 +02:00
parent 91835eacc1
commit 659678782f
11 changed files with 1585 additions and 1386 deletions

File diff suppressed because it is too large Load diff

1223
app/auth/__init__.py Normal file

File diff suppressed because it is too large Load diff

12
app/auth/deps.py Normal file
View file

@ -0,0 +1,12 @@
"""FastAPI dependencies - re-exports from app.auth.
Phase 3B of AUDIT-2026-Q3.md.
Public API:
- get_current_user, require_auth, require_public_profile
"""
from app.auth import ( # noqa: F401
get_current_user,
require_auth,
require_public_profile,
)

18
app/auth/jwt.py Normal file
View file

@ -0,0 +1,18 @@
"""JWT helpers - re-exports from app.auth for cleaner import paths.
Phase 3B of AUDIT-2026-Q3.md.
Public API:
- JWT_SECRET, JWT_ALGORITHM, JWT_EXPIRY_DAYS
- _create_jwt, _verify_jwt
- generate_nonce
"""
from app.auth import ( # noqa: F401
JWT_ALGORITHM,
JWT_EXPIRY_DAYS,
JWT_SECRET,
_create_jwt,
_derive_user_id,
_verify_jwt,
generate_nonce,
)

20
app/auth/oauth.py Normal file
View file

@ -0,0 +1,20 @@
"""OAuth flows - re-exports from app.auth.
Phase 3B of AUDIT-2026-Q3.md.
Public API:
- google_auth_url, google_callback, google_callback_post
- github_auth_url, github_callback
- x_auth_url, x_callback
- telegram_auth
"""
from app.auth import ( # noqa: F401
github_auth_url,
github_callback,
google_auth_url,
google_callback,
google_callback_post,
telegram_auth,
x_auth_url,
x_callback,
)

11
app/auth/passwords.py Normal file
View file

@ -0,0 +1,11 @@
"""Password helpers - re-exports from app.auth.
Phase 3B of AUDIT-2026-Q3.md.
Public API:
- hash_password, verify_password
"""
from app.auth import ( # noqa: F401
hash_password,
verify_password,
)

27
app/auth/schemas.py Normal file
View file

@ -0,0 +1,27 @@
"""Pydantic schemas - re-exports from app.auth.
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,
)

15
app/auth/store.py Normal file
View file

@ -0,0 +1,15 @@
"""User storage helpers - re-exports from app.auth.
Phase 3B of AUDIT-2026-Q3.md.
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,
)

24
app/auth/totp.py Normal file
View file

@ -0,0 +1,24 @@
"""TOTP / 2FA helpers - re-exports from app.auth.
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
"""
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,
)

163
app/auth/wallet.py Normal file
View file

@ -0,0 +1,163 @@
"""
Wallet Authentication Helpers
=============================
Wallet signature verification and user creation.
"""
import logging
import os
from datetime import datetime
from typing import Any
logger = logging.getLogger(__name__)
def verify_wallet_signature(message: str, signature: str, address: str) -> bool:
"""
Verify a wallet signature.
NOTE: This is a validation stub. In production, use a proper signing library
(e.g., web3.py for Ethereum, @solana/web3.js for Solana) to verify signatures.
For now, returns True if all fields are non-empty (basic validation).
"""
if not message or not signature or not address:
return False
# Ensure address format looks valid (basic check)
if len(address) < 20:
return False
# Basic signature length check
# Typical sig is 65 hex chars for EVM
return len(signature) >= 60
def decode_signature(signature: str) -> tuple:
"""
Decode a wallet signature into r, s, v components.
Returns (r_hex, s_hex, v_int) for signature verification.
"""
import binascii
sig_bytes = binascii.unhexlify(signature.replace("0x", ""))
r = sig_bytes[:32].hex()
s = sig_bytes[32:64].hex()
v = sig_bytes[64]
return r, s, v
async def get_or_create_wallet_user(address: str, chain: str = "base") -> dict[str, Any]:
"""
Get or create a user based on wallet address.
Returns dict with:
- access_token
- refresh_token
- id
- email
- display_name
- tier
- role
- created_at
"""
import hashlib
import json
# Derive user_id from wallet address
user_id = hashlib.sha256(address.lower().encode()).hexdigest()[:32]
# Try to load existing user
r = None
try:
import redis
r = redis.Redis(
host=os.getenv("REDIS_HOST", "localhost"),
port=int(os.getenv("REDIS_PORT", "6379")),
password=os.getenv("REDIS_PASSWORD", ""),
decode_responses=True,
)
except Exception as e:
logger.warning(f"Redis not available for wallet user lookup: {e}")
user = None
if r:
data = r.hget("rmi:wallet_users", address.lower())
if data:
user = json.loads(data)
# Create new user if doesn't exist
if not user:
# Generate fake email for wallet users (no email required for wallet auth)
email = f"{address.lower()}@wallet.rmi"
display_name = f"Wallet User {address[2:8].upper()}"
user = {
"id": user_id,
"address": address.lower(),
"email": email,
"display_name": display_name,
"chain": chain,
"tier": "FREE",
"role": "USER",
"created_at": datetime.utcnow().isoformat(),
"xp": 0,
"level": 1,
"badges": [],
"scans_remaining": 5,
"scans_used": 0,
}
if r:
r.hset("rmi:wallet_users", address.lower(), json.dumps(user))
# Also store in main users hash
r.hset("rmi:users", user_id, json.dumps(user))
# Generate JWT token
from app.auth import _create_jwt
# Use email if available, otherwise derive from address
email = user.get("email") or f"{address.lower()}@wallet.rmi"
token = _create_jwt(user_id, email, user.get("tier", "FREE"), user.get("role", "USER"), address)
return {
"access_token": token,
"refresh_token": token,
"id": user_id,
"email": email,
"display_name": user.get("display_name", display_name),
"tier": user.get("tier", "FREE"),
"role": user.get("role", "USER"),
"created_at": user.get("created_at"),
"address": address,
"chain": chain,
}
async def verify_auth_token(token: str) -> dict[str, Any] | None:
"""
Verify a JWT token and return user info.
Returns:
Dict with user info if valid, None if invalid/expired
"""
from app.auth import _verify_jwt
try:
user = _verify_jwt(token)
if user:
# Return user info in expected format
return {
"id": user.get("user_id"),
"email": user.get("email"),
"address": user.get("wallet_address"),
"tier": user.get("tier", "FREE"),
"role": user.get("role", "USER"),
}
except Exception as e:
logger.debug(f"Token verification failed: {e}")
return None

View file

@ -1,163 +1,8 @@
""" """Backward-compat shim — moved to app.auth.wallet in P3B."""
Wallet Authentication Helpers from app.auth.wallet import * # noqa: F401,F403
============================= from app.auth.wallet import ( # noqa: F401
Wallet signature verification and user creation. decode_signature,
""" get_or_create_wallet_user,
verify_auth_token,
import logging verify_wallet_signature,
import os )
from datetime import datetime
from typing import Any
logger = logging.getLogger(__name__)
def verify_wallet_signature(message: str, signature: str, address: str) -> bool:
"""
Verify a wallet signature.
NOTE: This is a validation stub. In production, use a proper signing library
(e.g., web3.py for Ethereum, @solana/web3.js for Solana) to verify signatures.
For now, returns True if all fields are non-empty (basic validation).
"""
if not message or not signature or not address:
return False
# Ensure address format looks valid (basic check)
if len(address) < 20:
return False
# Basic signature length check
# Typical sig is 65 hex chars for EVM
return len(signature) >= 60
def decode_signature(signature: str) -> tuple:
"""
Decode a wallet signature into r, s, v components.
Returns (r_hex, s_hex, v_int) for signature verification.
"""
import binascii
sig_bytes = binascii.unhexlify(signature.replace("0x", ""))
r = sig_bytes[:32].hex()
s = sig_bytes[32:64].hex()
v = sig_bytes[64]
return r, s, v
async def get_or_create_wallet_user(address: str, chain: str = "base") -> dict[str, Any]:
"""
Get or create a user based on wallet address.
Returns dict with:
- access_token
- refresh_token
- id
- email
- display_name
- tier
- role
- created_at
"""
import hashlib
import json
# Derive user_id from wallet address
user_id = hashlib.sha256(address.lower().encode()).hexdigest()[:32]
# Try to load existing user
r = None
try:
import redis
r = redis.Redis(
host=os.getenv("REDIS_HOST", "localhost"),
port=int(os.getenv("REDIS_PORT", "6379")),
password=os.getenv("REDIS_PASSWORD", ""),
decode_responses=True,
)
except Exception as e:
logger.warning(f"Redis not available for wallet user lookup: {e}")
user = None
if r:
data = r.hget("rmi:wallet_users", address.lower())
if data:
user = json.loads(data)
# Create new user if doesn't exist
if not user:
# Generate fake email for wallet users (no email required for wallet auth)
email = f"{address.lower()}@wallet.rmi"
display_name = f"Wallet User {address[2:8].upper()}"
user = {
"id": user_id,
"address": address.lower(),
"email": email,
"display_name": display_name,
"chain": chain,
"tier": "FREE",
"role": "USER",
"created_at": datetime.utcnow().isoformat(),
"xp": 0,
"level": 1,
"badges": [],
"scans_remaining": 5,
"scans_used": 0,
}
if r:
r.hset("rmi:wallet_users", address.lower(), json.dumps(user))
# Also store in main users hash
r.hset("rmi:users", user_id, json.dumps(user))
# Generate JWT token
from app.auth import _create_jwt
# Use email if available, otherwise derive from address
email = user.get("email") or f"{address.lower()}@wallet.rmi"
token = _create_jwt(user_id, email, user.get("tier", "FREE"), user.get("role", "USER"), address)
return {
"access_token": token,
"refresh_token": token,
"id": user_id,
"email": email,
"display_name": user.get("display_name", display_name),
"tier": user.get("tier", "FREE"),
"role": user.get("role", "USER"),
"created_at": user.get("created_at"),
"address": address,
"chain": chain,
}
async def verify_auth_token(token: str) -> dict[str, Any] | None:
"""
Verify a JWT token and return user info.
Returns:
Dict with user info if valid, None if invalid/expired
"""
from app.auth import _verify_jwt
try:
user = _verify_jwt(token)
if user:
# Return user info in expected format
return {
"id": user.get("user_id"),
"email": user.get("email"),
"address": user.get("wallet_address"),
"tier": user.get("tier", "FREE"),
"role": user.get("role", "USER"),
}
except Exception as e:
logger.debug(f"Token verification failed: {e}")
return None