fix(auth): remove KDF from pre-auth lookup, offload bcrypt, throttle /auth/*

src/qwed_new/auth/routes.py

@@ -1,13 +1,15 @@
 """
 Authentication routes for QWED Enterprise Portal.
 """
-from fastapi import APIRouter, HTTPException, Depends, Header
+import asyncio
+from fastapi import APIRouter, HTTPException, Depends, Header, Request
 from typing import Optional, List
 from datetime import datetime
 from sqlmodel import Session, select
 
 from qwed_new.core.database import get_session
 from qwed_new.core.models import User, Organization, ApiKey
+from qwed_new.core.rate_limiter import check_auth_rate_limit
 from .models import (
     SignUpRequest, SignInRequest, TokenResponse,
     APIKeyCreateRequest, APIKeyResponse, APIKeyListItem
@@ -19,15 +21,35 @@
 
 router = APIRouter(prefix="/auth", tags=["authentication"])
 
+# bcrypt cost-12 verify burns ~270 ms on an unknown email and returns in the
+# same time for a known one, equalizing the email-enumeration timing oracle
+# (issue #334). Initialized lazily so module import stays cheap.
+_dummy_password_hash: Optional[str] = None
+
+
+async def _burn_one_bcrypt(password: str) -> None:
+    """Run one bcrypt verify against a throwaway hash (timing equalizer)."""
+    global _dummy_password_hash
+    if _dummy_password_hash is None:
+        _dummy_password_hash = await asyncio.to_thread(
+            hash_password, "qwed-timing-equalizer"
+        )
+    await asyncio.to_thread(verify_password, password, _dummy_password_hash)
+
 @router.post("/signup", response_model=TokenResponse)
 async def signup(
     request: SignUpRequest,
+    req: Request,
     session: Session = Depends(get_session)
 ):
     """
     Sign up a new user and create their organization.
     Returns JWT token for immediate login.
     """
+    # Anonymous route: per-IP throttle before any DB or bcrypt work
+    # (bcrypt is ~269 ms of CPU; issues #226/#334).
+    check_auth_rate_limit(req)
+
     # Check if email already exists
     statement = select(User).where(User.email == request.email)
     existing_user = session.exec(statement).first()
@@ -42,6 +64,11 @@
         # For now, let's fail
         raise HTTPException(status_code=400, detail="Organization name already taken")
 
+    # Hash in the threadpool — bcrypt must never run on the event loop
+    # (issue #334). Also BEFORE any row is written so a hash failure
+    # cannot strand an orphaned Organization row.
+    password_hash = await asyncio.to_thread(hash_password, request.password)
+
     org = Organization(
         name=request.organization_name,
         display_name=request.organization_name,
@@ -52,7 +79,6 @@
     session.refresh(org)
 
     # Create user (first user is owner)
-    password_hash = hash_password(request.password)
     try:
         print(f"DEBUG: Creating user with email={request.email}, org_id={org.id}")
         user = User(
@@ -89,13 +115,24 @@
 @router.post("/signin", response_model=TokenResponse)
 async def signin(
     request: SignInRequest,
+    req: Request,
     session: Session = Depends(get_session)
 ):
     """Sign in an existing user."""
+    # Anonymous route: per-IP throttle — unthrottled signin is a password-
+    # guessing oracle and a ~4 req/s whole-service DoS (issues #226/#334).
+    check_auth_rate_limit(req)
+
     statement = select(User).where(User.email == request.email)
     user = session.exec(statement).first()
-
-    if not user or not verify_password(request.password, user.password_hash):
+
+    # bcrypt in the threadpool (issue #334); when the email is unknown,
+    # burn one bcrypt anyway so response timing does not enumerate emails.
+    if user is None:
+        await _burn_one_bcrypt(request.password)
+        raise HTTPException(status_code=401, detail="Invalid email or password")
+
+    if not await asyncio.to_thread(verify_password, request.password, user.password_hash):
         raise HTTPException(status_code=401, detail="Invalid email or password")
 
     if not user.is_active:

src/qwed_new/auth/security.py

@@ -1,4 +1,4 @@
 """
 Security utilities for QWED authentication.
 Handles password hashing, JWT token generation, and API key management.
 """
@@ -6,6 +6,7 @@
 import jwt
 import secrets
 import hashlib
+import hmac
 import os
 from datetime import datetime, timedelta, timezone
 from typing import Optional
@@ -62,7 +63,7 @@
     Format: qwed_live_<32_random_chars>
     """
     random_part = secrets.token_urlsafe(32)
     plaintext_key = f"{prefix}_{random_part}"
 
     # Hash the key for storage
     key_hash = hash_api_key(plaintext_key)
@@ -72,29 +73,31 @@
 
 def hash_api_key(api_key: str) -> str:
     """
-    Derive a hash for an API key using PBKDF2-HMAC-SHA256.
+    Derive a deterministic lookup digest for an API key.
 
-    This is intentionally computationally expensive to make brute-force attacks
-    against stored API key hashes more difficult, while remaining deterministic
-    for lookup purposes.
+    This is a fast keyed MAC (HMAC-SHA256, microsecond cost), NOT a KDF.
+    The previous PBKDF2-HMAC-SHA256 with 100,000 iterations sat on the
+    unauthenticated request path (hash-then-lookup) and let ~15 req/s of
+    garbage x-api-key values saturate the whole service (issue #333).
+    The cost bought no brute-force resistance: API keys are 258-bit random
+    tokens, so equality lookup is unbreakable at any digest speed.
+
+    NOTE: not compatible with pre-v7.2 PBKDF2 key_hash rows. Existing keys
+    must be re-issued once via the rotation path (key_rotation.py uses this
+    same function, so newly issued/rotated keys are HMAC digests). Do NOT
+    add a PBKDF2 fallback for legacy rows — that re-introduces #333.
     """
-    # Derive a salt from SECRET_KEY; fall back to a constant development salt.
     if isinstance(SECRET_KEY, str):
         secret_bytes = SECRET_KEY.encode()
     else:
-        secret_bytes = b"default_dev_salt" # Fallback if secret is somehow bytes or None
-
-    # Namespace the salt for API key hashing to avoid cross-protocol reuse.
-    salt = secret_bytes + b":qwed_api_key"
+        secret_bytes = b"default_dev_salt"  # Fallback if secret is somehow bytes or None
 
-    # Use PBKDF2-HMAC-SHA256 with a reasonable iteration count.
-    dk = hashlib.pbkdf2_hmac(
-        "sha256",
+    # Namespace the MAC for API key hashing to avoid cross-protocol reuse.
+    return hmac.new(
+        secret_bytes + b":qwed_api_key_lookup",
         api_key.encode("utf-8"),
-        salt,
-        100_000,
-    )
-    return dk.hex()
+        hashlib.sha256,
+    ).hexdigest()
 
 def mask_api_key(api_key: str) -> str:
     """

src/qwed_new/core/key_rotation.py

@@ -1,4 +1,4 @@
 """
 API Key Rotation and Management Module.
 Handles key expiration, rotation, and lifecycle management.
 
@@ -42,11 +42,11 @@
         Returns (ApiKey object, raw_key_string).
         """
         # Generate secure random key
         raw_key = f"qwed_live_{secrets.token_urlsafe(32)}"
 
-        # Hash for storage (using PBKDF2 to match auth/security.py)
+        # Hash for storage (fast keyed-MAC lookup digest per auth/security.py)
         key_hash = hash_api_key(raw_key)
         key_preview = f"{raw_key[:10]}...{raw_key[-4:]}"
 
         expires_at = datetime.utcnow() + timedelta(days=expires_in_days)
 

src/qwed_new/core/rate_limiter.py

@@ -23,23 +23,39 @@ class RateLimiter:
     Environment Variables:
         QWED_RATE_LIMIT_PER_KEY: Requests per minute per API key (default: 100)
         QWED_RATE_LIMIT_GLOBAL: Requests per minute globally (default: 1000)
+        QWED_RATE_LIMIT_PER_IP: Requests per minute per client IP on
+            anonymous /auth/* routes (default: 10) — these routes have no
+            API key to key a bucket on, so without a per-IP bucket they are
+            an unthrottled bcrypt/DoS surface (issues #226, #334).
     """
-    
+
     def __init__(self):
         self._lock = threading.Lock()
 
         # Per-API-key request timestamps: {api_key: [timestamp1, timestamp2, ...]}
         self.api_key_requests: Dict[str, list] = defaultdict(list)
-
+
+        # Per-client-IP request timestamps for anonymous auth routes:
+        # {ip: [timestamp1, timestamp2, ...]}
+        self.ip_requests: Dict[str, list] = defaultdict(list)
+
         # Global request timestamps: [timestamp1, timestamp2, ...]
         self.global_requests: list = []
-        
+
         # Rate limit configurations - configurable via env vars
         self.PER_KEY_LIMIT = int(os.environ.get("QWED_RATE_LIMIT_PER_KEY", "100"))
         self.PER_KEY_WINDOW = 60  # seconds
-        
+
         self.GLOBAL_LIMIT = int(os.environ.get("QWED_RATE_LIMIT_GLOBAL", "1000"))
         self.GLOBAL_WINDOW = 60  # seconds
+
+        self.PER_IP_LIMIT = int(os.environ.get("QWED_RATE_LIMIT_PER_IP", "10"))
+        self.PER_IP_WINDOW = 60  # seconds
+
+        # Bound the per-IP table: floods from spoofed/varied IPs must not
+        # grow memory unboundedly. Above the cap, drop IPs whose windows
+        # have fully expired.
+        self.MAX_TRACKED_IPS = 50_000
 
     def _clean_old_requests(self, requests: list, window_seconds: int) -> list:
         """Remove timestamps older than the window."""
@@ -90,6 +106,45 @@ def check_global_limit(self) -> bool:
             self.global_requests.append(time.time())
             return True
 
+    def check_ip_limit(self, client_ip: str) -> bool:
+        """
+        Check if a client IP has exceeded the anonymous-route rate limit.
+
+        Returns:
+            True if request is allowed, False if rate limit exceeded
+        """
+        with self._lock:
+            if len(self.ip_requests) > self.MAX_TRACKED_IPS:
+                cutoff = time.time() - self.PER_IP_WINDOW
+                self.ip_requests = defaultdict(
+                    list,
+                    {
+                        ip: stamps
+                        for ip, stamps in self.ip_requests.items()
+                        if stamps and stamps[-1] > cutoff
+                    },
+                )
+
+            self.ip_requests[client_ip] = self._clean_old_requests(
+                self.ip_requests[client_ip],
+                self.PER_IP_WINDOW,
+            )
+
+            if len(self.ip_requests[client_ip]) >= self.PER_IP_LIMIT:
+                return False
+
+            self.ip_requests[client_ip].append(time.time())
+            return True
+
+    def get_ip_reset_time(self, client_ip: str) -> int:
+        """Seconds until this IP's auth-route window resets."""
+        with self._lock:
+            requests = list(self.ip_requests.get(client_ip, []))
+            if not requests:
+                return 0
+            oldest = min(requests)
+            return max(0, int(oldest + self.PER_IP_WINDOW - time.time()))
+
     def get_reset_time(self, api_key: Optional[str] = None) -> int:
         """
         Get seconds until rate limit resets.
@@ -149,3 +204,37 @@ def check_rate_limit(api_key: Optional[str] = None):
                 detail=f"API key rate limit exceeded. Try again in {reset_after} seconds.",
                 headers={"Retry-After": str(reset_after)}
             )
+
+
+def client_ip_of(request) -> str:
+    """
+    Best-effort client IP for anonymous-route rate limiting.
+
+    Prefers the first X-Forwarded-For hop (set by the ingress/replica proxy);
+    falls back to the direct peer address. X-Forwarded-For is client-spoofable
+    where no trusted proxy strips it — this is a DoS-mitigation bucket, not an
+    identity boundary, and false attribution only widens or narrows one bucket.
+    """
+    forwarded = request.headers.get("x-forwarded-for", "")
+    first_hop = forwarded.split(",")[0].strip() if forwarded else ""
+    if first_hop:
+        return first_hop
+    return request.client.host if request.client else "unknown"
+
+
+def check_auth_rate_limit(request):
+    """
+    FastAPI dependency for anonymous /auth/* routes: per-IP bucket only.
+
+    These routes have no API key, so the per-key limiter cannot apply; an
+    unthrottled bcrypt signup/signin is a ~4 req/s whole-service DoS plus a
+    password-guessing oracle (issues #226, #334).
+    """
+    ip = client_ip_of(request)
+    if not rate_limiter.check_ip_limit(ip):
+        reset_after = rate_limiter.get_ip_reset_time(ip)
+        raise HTTPException(
+            status_code=429,
+            detail=f"Too many authentication attempts. Try again in {reset_after} seconds.",
+            headers={"Retry-After": str(reset_after)},
+        )