working with login and building inventory

trade-hub/backend/app/security.py

@@ -0,0 +1,140 @@
+"""Password hashing, session tokens, origin checks and login throttling."""
+import hashlib
+import re
+import secrets
+import threading
+import time
+from collections import deque
+from urllib.parse import urlsplit
+
+from argon2 import PasswordHasher
+from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError
+
+_hasher = PasswordHasher()
+# Hash of a throwaway secret, verified when a username does not exist so that
+# "unknown user" and "wrong password" take comparable time.
+_DUMMY_HASH = _hasher.hash(secrets.token_urlsafe(16))
+
+USERNAME_RE = re.compile(r"^[A-Za-z0-9_.-]{3,32}$")
+MIN_PASSWORD_LENGTH = 8
+MAX_PASSWORD_LENGTH = 128
+
+
+def normalize_username(username: str) -> str:
+    return username.strip().lower()
+
+
+def valid_username(username: str) -> bool:
+    return bool(USERNAME_RE.match(username.strip()))
+
+
+def hash_password(password: str) -> str:
+    return _hasher.hash(password)
+
+
+def verify_password(password: str, password_hash: str) -> bool:
+    try:
+        return _hasher.verify(password_hash, password)
+    except (VerifyMismatchError, VerificationError, InvalidHashError):
+        return False
+
+
+def burn_password_check(password: str) -> None:
+    """Constant-ish work for unknown users, to blunt username enumeration."""
+    try:
+        _hasher.verify(_DUMMY_HASH, password)
+    except (VerifyMismatchError, VerificationError, InvalidHashError):
+        pass
+
+
+def needs_rehash(password_hash: str) -> bool:
+    return _hasher.check_needs_rehash(password_hash)
+
+
+def new_session_token() -> str:
+    return secrets.token_urlsafe(32)
+
+
+def hash_token(token: str) -> str:
+    return hashlib.sha256(token.encode("utf-8")).hexdigest()
+
+
+def client_ip(request) -> str:
+    """Real client address as seen by the reverse proxy.
+
+    nginx overwrites X-Real-IP with the actual peer address, so it is safe to
+    trust here. X-Forwarded-For is client-appendable and is never used.
+    """
+    real_ip = request.headers.get("x-real-ip")
+    if real_ip:
+        return real_ip.strip()
+    return request.client.host if request.client else "unknown"
+
+
+def same_origin(request) -> bool:
+    """Reject cross-origin state-changing requests (CSRF defense in depth)."""
+    origin = request.headers.get("origin")
+    if not origin:
+        return True  # non-browser client; the session cookie is still required
+    host = request.headers.get("x-forwarded-host") or request.headers.get("host") or ""
+    return urlsplit(origin).netloc.lower() == host.lower()
+
+class LoginThrottle:
+    """In-process sliding-window limiter for failed logins.
+
+    Tracks failures per (username, ip) and, with a higher ceiling, per ip so a
+    single address cannot spray passwords across many usernames.
+    """
+
+    def __init__(self, max_attempts: int, window_seconds: int):
+        self.max_attempts = max_attempts
+        self.window = window_seconds
+        self._events: dict[tuple[str, str], deque[float]] = {}
+        self._lock = threading.Lock()
+
+    def _keys(self, username_key: str, ip: str) -> list[tuple[tuple[str, str], int]]:
+        return [
+            ((username_key, ip), self.max_attempts),
+            (("*", ip), self.max_attempts * 4),
+        ]
+
+    def _recent(self, key: tuple[str, str], now: float) -> deque[float] | None:
+        events = self._events.get(key)
+        if not events:
+            return None
+        cutoff = now - self.window
+        while events and events[0] < cutoff:
+            events.popleft()
+        if not events:
+            self._events.pop(key, None)
+            return None
+        return events
+
+    def retry_after(self, username_key: str, ip: str) -> int:
+        """Seconds to wait before another attempt is allowed (0 if allowed)."""
+        now = time.monotonic()
+        with self._lock:
+            waits = []
+            for key, limit in self._keys(username_key, ip):
+                events = self._recent(key, now)
+                if events and len(events) >= limit:
+                    waits.append(int(self.window - (now - events[0])) + 1)
+            return max(waits) if waits else 0
+
+    def record_failure(self, username_key: str, ip: str) -> None:
+        now = time.monotonic()
+        with self._lock:
+            if len(self._events) > 5000:  # opportunistic cleanup
+                for key in list(self._events):
+                    self._recent(key, now)
+            for key, _limit in self._keys(username_key, ip):
+                self._events.setdefault(key, deque()).append(now)
+
+    def reset(self, username_key: str, ip: str) -> None:
+        with self._lock:
+            for key, _limit in self._keys(username_key, ip):
+                self._events.pop(key, None)
+
+    def clear(self) -> None:
+        with self._lock:
+            self._events.clear()