refactor(auth): move password helpers to app.domains.auth.passwords (P4.3 cont)
This commit is contained in:
parent
59d6f2f0f6
commit
3eb22495b1
3 changed files with 70 additions and 35 deletions
|
|
@ -12,20 +12,20 @@ import secrets
|
|||
from datetime import datetime, timedelta
|
||||
from typing import Any
|
||||
|
||||
import bcrypt
|
||||
from fastapi import APIRouter, HTTPException, Request
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.core.redis import get_redis
|
||||
from app.domains.auth.jwt import (
|
||||
JWT_ALGORITHM,
|
||||
JWT_EXPIRY_DAYS,
|
||||
JWT_EXPIRY_DAYS as JWT_EXPIRY_DAYS,
|
||||
JWT_SECRET,
|
||||
_create_jwt,
|
||||
_derive_user_id,
|
||||
_verify_jwt,
|
||||
generate_nonce,
|
||||
)
|
||||
from app.domains.auth.passwords import hash_password, verify_password
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -105,19 +105,6 @@ def _verify_backup_code(code: str, hashed: str) -> bool:
|
|||
# ── Config ──
|
||||
|
||||
|
||||
def hash_password(password: str) -> str:
|
||||
"""Hash password using bcrypt directly."""
|
||||
return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
|
||||
|
||||
|
||||
def verify_password(password: str, hashed: str) -> bool:
|
||||
"""Verify password against hash."""
|
||||
try:
|
||||
return bcrypt.checkpw(password.encode("utf-8"), hashed.encode("utf-8"))
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
# ── 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."""
|
||||
|
|
@ -464,7 +451,9 @@ async def get_current_user(request: Request):
|
|||
async def google_auth_url():
|
||||
"""Get Google OAuth URL."""
|
||||
client_id = os.getenv("GOOGLE_CLIENT_ID")
|
||||
redirect_uri = os.getenv("GOOGLE_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/google/callback")
|
||||
redirect_uri = os.getenv(
|
||||
"GOOGLE_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/google/callback"
|
||||
)
|
||||
|
||||
if not client_id:
|
||||
return {"url": "/auth/callback?error=google_not_configured"}
|
||||
|
|
@ -504,7 +493,9 @@ async def google_callback(request: Request):
|
|||
|
||||
client_id = os.getenv("GOOGLE_CLIENT_ID")
|
||||
client_secret = os.getenv("GOOGLE_CLIENT_SECRET")
|
||||
redirect_uri = os.getenv("GOOGLE_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/google/callback")
|
||||
redirect_uri = os.getenv(
|
||||
"GOOGLE_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/google/callback"
|
||||
)
|
||||
|
||||
if not client_id or not client_secret:
|
||||
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=not_configured")
|
||||
|
|
@ -556,7 +547,9 @@ async def google_callback(request: Request):
|
|||
r.hset("rmi:users", user_id, json.dumps(user))
|
||||
r.hset("rmi:users:email", email.lower(), user_id)
|
||||
|
||||
jwt_token = _create_jwt(user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER"))
|
||||
jwt_token = _create_jwt(
|
||||
user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER")
|
||||
)
|
||||
|
||||
# Redirect to frontend with token
|
||||
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?token={jwt_token}&provider=google")
|
||||
|
|
@ -575,7 +568,9 @@ async def google_callback_post(request: Request):
|
|||
|
||||
client_id = os.getenv("GOOGLE_CLIENT_ID")
|
||||
client_secret = os.getenv("GOOGLE_CLIENT_SECRET")
|
||||
redirect_uri = os.getenv("GOOGLE_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/google/callback")
|
||||
redirect_uri = os.getenv(
|
||||
"GOOGLE_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/google/callback"
|
||||
)
|
||||
|
||||
if not client_id or not client_secret:
|
||||
raise HTTPException(status_code=500, detail="Google OAuth not configured")
|
||||
|
|
@ -627,7 +622,9 @@ async def google_callback_post(request: Request):
|
|||
r.hset("rmi:users", user_id, json.dumps(user))
|
||||
r.hset("rmi:users:email", email.lower(), user_id)
|
||||
|
||||
jwt_token = _create_jwt(user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER"))
|
||||
jwt_token = _create_jwt(
|
||||
user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER")
|
||||
)
|
||||
|
||||
return {
|
||||
"access_token": jwt_token,
|
||||
|
|
@ -680,7 +677,11 @@ async def telegram_auth(req: TelegramAuthRequest):
|
|||
user = _get_user(telegram_user_id)
|
||||
|
||||
if not user:
|
||||
display_name = f"{req.first_name or ''} {req.last_name or ''}".strip() or req.username or f"User{req.id}"
|
||||
display_name = (
|
||||
f"{req.first_name or ''} {req.last_name or ''}".strip()
|
||||
or req.username
|
||||
or f"User{req.id}"
|
||||
)
|
||||
email = f"{req.id}@telegram.rmi"
|
||||
|
||||
user = {
|
||||
|
|
@ -702,7 +703,9 @@ async def telegram_auth(req: TelegramAuthRequest):
|
|||
r.hset("rmi:users", telegram_user_id, json.dumps(user))
|
||||
r.hset("rmi:users:telegram", str(req.id), telegram_user_id)
|
||||
|
||||
jwt_token = _create_jwt(user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER"))
|
||||
jwt_token = _create_jwt(
|
||||
user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER")
|
||||
)
|
||||
|
||||
return {
|
||||
"access_token": jwt_token,
|
||||
|
|
@ -725,7 +728,9 @@ async def telegram_auth(req: TelegramAuthRequest):
|
|||
async def github_auth_url():
|
||||
"""Get GitHub OAuth URL."""
|
||||
client_id = os.getenv("GITHUB_CLIENT_ID")
|
||||
redirect_uri = os.getenv("GITHUB_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/github/callback")
|
||||
redirect_uri = os.getenv(
|
||||
"GITHUB_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/github/callback"
|
||||
)
|
||||
|
||||
if not client_id:
|
||||
return {"url": "/auth/callback?error=github_not_configured"}
|
||||
|
|
@ -759,7 +764,9 @@ async def github_callback(request: Request):
|
|||
|
||||
client_id = os.getenv("GITHUB_CLIENT_ID")
|
||||
client_secret = os.getenv("GITHUB_CLIENT_SECRET")
|
||||
redirect_uri = os.getenv("GITHUB_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/github/callback")
|
||||
redirect_uri = os.getenv(
|
||||
"GITHUB_REDIRECT_URI", "https://rugmunch.io/api/v1/auth/github/callback"
|
||||
)
|
||||
|
||||
if not client_id or not client_secret:
|
||||
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?error=not_configured")
|
||||
|
|
@ -802,7 +809,11 @@ async def github_callback(request: Request):
|
|||
if resp.status_code == 200:
|
||||
emails = resp.json()
|
||||
primary = next((e for e in emails if e.get("primary")), None)
|
||||
email = primary.get("email") if primary else (emails[0].get("email") if emails else None)
|
||||
email = (
|
||||
primary.get("email")
|
||||
if primary
|
||||
else (emails[0].get("email") if emails else None)
|
||||
)
|
||||
|
||||
if not email:
|
||||
email = f"{github_user.get('login', 'github_user')}@github.rmi"
|
||||
|
|
@ -827,7 +838,9 @@ async def github_callback(request: Request):
|
|||
r.hset("rmi:users", user_id, json.dumps(user))
|
||||
r.hset("rmi:users:email", email.lower(), user_id)
|
||||
|
||||
jwt_token = _create_jwt(user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER"))
|
||||
jwt_token = _create_jwt(
|
||||
user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER")
|
||||
)
|
||||
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?token={jwt_token}&provider=github")
|
||||
|
||||
|
||||
|
|
@ -928,7 +941,9 @@ async def x_callback(request: Request):
|
|||
r.hset("rmi:users", user_id, json.dumps(user))
|
||||
r.hset("rmi:users:email", email.lower(), user_id)
|
||||
|
||||
jwt_token = _create_jwt(user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER"))
|
||||
jwt_token = _create_jwt(
|
||||
user["id"], user["email"], user.get("tier", "FREE"), user.get("role", "USER")
|
||||
)
|
||||
return RedirectResponse(f"{FRONTEND_URL}/auth/callback?token={jwt_token}&provider=x")
|
||||
|
||||
|
||||
|
|
@ -971,7 +986,9 @@ async def twofa_setup(request: Request):
|
|||
|
||||
# Check if already enabled
|
||||
if user.get("totp_secret"):
|
||||
raise HTTPException(status_code=400, detail="2FA already enabled. Disable first to reconfigure.")
|
||||
raise HTTPException(
|
||||
status_code=400, detail="2FA already enabled. Disable first to reconfigure."
|
||||
)
|
||||
|
||||
secret = _generate_totp_secret()
|
||||
uri = _get_totp_uri(secret, user["email"])
|
||||
|
|
@ -1015,7 +1032,9 @@ async def twofa_enable(req: TwoFAEnableRequest, request: Request):
|
|||
|
||||
# Verify the code
|
||||
if not _verify_totp(secret, req.code):
|
||||
raise HTTPException(status_code=400, detail="Invalid TOTP code. Check your authenticator app time sync.")
|
||||
raise HTTPException(
|
||||
status_code=400, detail="Invalid TOTP code. Check your authenticator app time sync."
|
||||
)
|
||||
|
||||
# Enable 2FA on user record
|
||||
user["totp_secret"] = secret
|
||||
|
|
@ -1138,7 +1157,9 @@ async def twofa_login(req: TwoFALoginRequest):
|
|||
|
||||
# Check if 2FA is enabled
|
||||
if not user.get("totp_enabled") or not user.get("totp_secret"):
|
||||
raise HTTPException(status_code=400, detail="2FA not enabled for this account. Use /login instead.")
|
||||
raise HTTPException(
|
||||
status_code=400, detail="2FA not enabled for this account. Use /login instead."
|
||||
)
|
||||
|
||||
code = req.code.strip().replace("-", "")
|
||||
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ Public API:
|
|||
- _create_jwt, _verify_jwt
|
||||
- generate_nonce, _derive_user_id
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
|
|
|
|||
|
|
@ -1,11 +1,24 @@
|
|||
"""Password helpers - re-exports from app.domains.auth.
|
||||
"""Password helpers.
|
||||
|
||||
Phase 3B of AUDIT-2026-Q3.md.
|
||||
|
||||
Public API:
|
||||
- hash_password, verify_password
|
||||
"""
|
||||
from app.auth import ( # noqa: F401
|
||||
hash_password,
|
||||
verify_password,
|
||||
)
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import bcrypt
|
||||
|
||||
|
||||
def hash_password(password: str) -> str:
|
||||
"""Hash password using bcrypt directly."""
|
||||
return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
|
||||
|
||||
|
||||
def verify_password(password: str, hashed: str) -> bool:
|
||||
"""Verify password against hash."""
|
||||
try:
|
||||
return bcrypt.checkpw(password.encode("utf-8"), hashed.encode("utf-8"))
|
||||
except Exception:
|
||||
return False
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue