mirror of
https://github.com/escalante29/WealthySmart.git
synced 2026-07-17 16:28:47 +02:00
b4123703ef
Both login endpoints now share verify_admin_credentials (hmac compare_digest, SEC-02) and a 5/min/IP sliding-window rate limit that returns 429 with Retry-After (SEC-03). CORS is pinned to the prod domain and localhost dev origins instead of '*' (SEC-04). The session cookie's Secure flag is driven by COOKIE_SECURE instead of hardcoded false (SEC-08). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
133 lines
4.8 KiB
Python
133 lines
4.8 KiB
Python
import hashlib
|
|
import hmac
|
|
import re
|
|
import time
|
|
from collections import defaultdict, deque
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
|
|
from fastapi import Cookie, Depends, Header, HTTPException, Request, status
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from jose import JWTError, jwt
|
|
from sqlmodel import Session, select
|
|
|
|
from app.config import settings
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
|
|
|
|
ALGORITHM = "HS256"
|
|
|
|
# ── Login hardening ──────────────────────────────────────────────────────────
|
|
# Single-process app: an in-memory sliding window per client IP is sufficient.
|
|
|
|
LOGIN_RATE_LIMIT = 5
|
|
LOGIN_RATE_WINDOW_SECONDS = 60
|
|
|
|
_login_attempts: dict[str, deque[float]] = defaultdict(deque)
|
|
|
|
|
|
def _client_ip(request: Request) -> str:
|
|
# nginx-proxy sets X-Forwarded-For; the Hono proxy forwards it through.
|
|
forwarded = request.headers.get("x-forwarded-for", "")
|
|
if forwarded:
|
|
return forwarded.split(",")[0].strip()
|
|
return request.client.host if request.client else "unknown"
|
|
|
|
|
|
def check_login_rate_limit(request: Request) -> None:
|
|
"""Raise 429 if this client has attempted login too often. Call BEFORE
|
|
verifying credentials so attempts are counted regardless of outcome."""
|
|
now = time.monotonic()
|
|
attempts = _login_attempts[_client_ip(request)]
|
|
while attempts and now - attempts[0] > LOGIN_RATE_WINDOW_SECONDS:
|
|
attempts.popleft()
|
|
if len(attempts) >= LOGIN_RATE_LIMIT:
|
|
retry_after = int(LOGIN_RATE_WINDOW_SECONDS - (now - attempts[0])) + 1
|
|
raise HTTPException(
|
|
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
detail="Too many login attempts, try again later",
|
|
headers={"Retry-After": str(retry_after)},
|
|
)
|
|
attempts.append(now)
|
|
# Bound memory against spoofed X-Forwarded-For values.
|
|
if len(_login_attempts) > 1000:
|
|
for key in [k for k, v in _login_attempts.items() if not v]:
|
|
del _login_attempts[key]
|
|
|
|
|
|
def verify_admin_credentials(username: str, password: str) -> None:
|
|
"""Constant-time credential check shared by both login endpoints."""
|
|
username_ok = hmac.compare_digest(
|
|
username.encode(), settings.ADMIN_USERNAME.encode()
|
|
)
|
|
password_ok = hmac.compare_digest(
|
|
password.encode(), settings.ADMIN_PASSWORD.encode()
|
|
)
|
|
if not (username_ok and password_ok):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials"
|
|
)
|
|
|
|
|
|
def create_access_token(subject: str) -> str:
|
|
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
return jwt.encode({"sub": subject, "exp": expire}, settings.SECRET_KEY, algorithm=ALGORITHM)
|
|
|
|
|
|
def hash_token(token: str) -> str:
|
|
return hashlib.sha256(token.encode()).hexdigest()
|
|
|
|
|
|
def _validate_token(token: str) -> str:
|
|
"""Validate JWT and return subject, or raise 401."""
|
|
try:
|
|
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
|
|
username: str = payload.get("sub")
|
|
if username is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
|
|
return username
|
|
except JWTError:
|
|
pass
|
|
|
|
# Fallback: check API token
|
|
from app.db import get_session
|
|
from app.models.models import APIToken
|
|
|
|
token_hash = hash_token(token)
|
|
with next(get_session()) as session:
|
|
api_token = session.exec(
|
|
select(APIToken).where(
|
|
APIToken.token_hash == token_hash,
|
|
APIToken.is_active == True,
|
|
)
|
|
).first()
|
|
if api_token:
|
|
if api_token.expires_at and api_token.expires_at < datetime.utcnow():
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token expired")
|
|
return f"api:{api_token.name}"
|
|
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
|
|
|
|
|
|
def get_current_user_cookie_or_bearer(
|
|
authorization: Optional[str] = Header(default=None),
|
|
ws_token: Optional[str] = Cookie(default=None),
|
|
) -> str:
|
|
"""Accepts httpOnly cookie (SPA) or Bearer token (API clients / n8n)."""
|
|
token: Optional[str] = None
|
|
if authorization and authorization.lower().startswith("bearer "):
|
|
token = authorization.split(" ", 1)[1].strip()
|
|
elif ws_token:
|
|
token = ws_token
|
|
if not token:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
|
|
return _validate_token(token)
|
|
|
|
|
|
def get_current_user(
|
|
authorization: Optional[str] = Header(default=None),
|
|
ws_token: Optional[str] = Cookie(default=None),
|
|
) -> str:
|
|
"""SPA cookie or Bearer token. Single dependency for all v1 endpoints."""
|
|
return get_current_user_cookie_or_bearer(authorization, ws_token)
|