mirror of
https://github.com/escalante29/WealthySmart.git
synced 2026-07-17 12:08:47 +02:00
Rate-limit logins, constant-time creds, pin CORS, secure cookie
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>
This commit is contained in:
@@ -1,9 +1,12 @@
|
||||
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, status
|
||||
from fastapi import Cookie, Depends, Header, HTTPException, Request, status
|
||||
from fastapi.security import OAuth2PasswordBearer
|
||||
from jose import JWTError, jwt
|
||||
from sqlmodel import Session, select
|
||||
@@ -14,6 +17,57 @@ 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)
|
||||
|
||||
Reference in New Issue
Block a user