141 lines
4.6 KiB
Python
141 lines
4.6 KiB
Python
"""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()
|