From 8c74e360d27b9d9e83970e0ded997ece89844040 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 12:23:12 +0200 Subject: [PATCH] feat(auth): throttle password login attempts --- README.md | 18 + pyproject.toml | 3 +- src/govoplan_access/backend/api/v1/auth.py | 69 +++- .../backend/security/login_throttle.py | 358 ++++++++++++++++++ tests/test_login_throttle.py | 219 +++++++++++ tests/test_optional_tenancy_contract.py | 2 +- 6 files changed, 666 insertions(+), 3 deletions(-) create mode 100644 src/govoplan_access/backend/security/login_throttle.py create mode 100644 tests/test_login_throttle.py diff --git a/README.md b/README.md index 3113c1e..8d2da50 100644 --- a/README.md +++ b/README.md @@ -113,3 +113,21 @@ From the core checkout: cd /mnt/DATA/git/govoplan-core ./.venv/bin/python -m pip install -e ../govoplan-access ``` + +## Login Throttling + +Interactive password login is throttled by normalized global login identity and by +the directly connected client address. The deployment defaults are 10 identity +failures and 100 client failures in a 15-minute window. Counters use +`REDIS_URL` when Redis is reachable, allowing all API workers to share the same +limits. Local development and Redis outages fall back automatically to a +bounded, process-local counter; authentication remains available, but limits +then apply per API process. + +The deployment settings are `AUTH_LOGIN_THROTTLE_ENABLED`, +`AUTH_LOGIN_THROTTLE_IDENTITY_LIMIT`, `AUTH_LOGIN_THROTTLE_CLIENT_LIMIT`, +`AUTH_LOGIN_THROTTLE_WINDOW_SECONDS`, and +`AUTH_LOGIN_THROTTLE_REDIS_RETRY_SECONDS`. Client-supplied forwarding headers +are not trusted for throttling. A reverse proxy should pass the real peer +address only through the platform's separately configured trusted-proxy +boundary. diff --git a/pyproject.toml b/pyproject.toml index 6d333f0..50fe2bd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,8 @@ requires-python = ">=3.12" license = { file = "LICENSE" } authors = [{ name = "GovOPlaN" }] dependencies = [ - "govoplan-core>=0.1.8", + "govoplan-core>=0.1.9", + "redis>=5,<6", "SQLAlchemy>=2,<3", ] diff --git a/src/govoplan_access/backend/api/v1/auth.py b/src/govoplan_access/backend/api/v1/auth.py index b6de81f..6ac039a 100644 --- a/src/govoplan_access/backend/api/v1/auth.py +++ b/src/govoplan_access/backend/api/v1/auth.py @@ -1,6 +1,7 @@ from __future__ import annotations from dataclasses import dataclass +from functools import lru_cache from fastapi import APIRouter, Depends, HTTPException, Request, Response, status from sqlalchemy.orm import Session @@ -51,6 +52,11 @@ from govoplan_access.backend.semantic import collect_function_assignment_ids, co from govoplan_access.backend.auth.tenant_context import AccessTenantContextSwitcher from govoplan_access.backend.security.api_keys import authenticate_api_key from govoplan_access.backend.security.passwords import DUMMY_PASSWORD_HASH, verify_password +from govoplan_access.backend.security.login_throttle import ( + LoginThrottle, + LoginThrottleDecision, + build_login_throttle, +) from govoplan_access.backend.security.sessions import ( authenticate_session_token, collect_user_authorization_context, @@ -277,6 +283,67 @@ def _resolve_login_user(session: Session, payload: LoginRequest) -> tuple[Accoun return account, row[0], row[1] +@lru_cache(maxsize=1) +def _configured_login_throttle() -> LoginThrottle | None: + if not settings.auth_login_throttle_enabled: + return None + return build_login_throttle( + redis_url=settings.redis_url, + identity_limit=settings.auth_login_throttle_identity_limit, + client_limit=settings.auth_login_throttle_client_limit, + window_seconds=settings.auth_login_throttle_window_seconds, + redis_retry_seconds=settings.auth_login_throttle_redis_retry_seconds, + ) + + +def _raise_login_throttled(decision: LoginThrottleDecision) -> None: + retry_after = max(1, decision.retry_after_seconds) + # The detail deliberately matches every credential failure. Status and + # Retry-After communicate endpoint throttling without disclosing whether + # the supplied identity exists or has an active membership. + raise HTTPException( + status_code=status.HTTP_429_TOO_MANY_REQUESTS, + detail="Invalid login", + headers={"Retry-After": str(retry_after)}, + ) + + +def _resolve_throttled_login_user( + session: Session, + payload: LoginRequest, + request: Request, +) -> tuple[Account, User, Tenant]: + throttle = _configured_login_throttle() + if throttle is None: + return _resolve_login_user(session, payload) + + normalized_email = normalize_email(payload.email) + client_address = request.client.host if request.client else None + context = { + "normalized_email": normalized_email, + "tenant_slug": payload.tenant_slug, + "client_address": client_address, + } + decision = throttle.check(**context) + if not decision.allowed: + _raise_login_throttled(decision) + + try: + resolved = _resolve_login_user(session, payload) + except HTTPException as exc: + if exc.status_code == status.HTTP_401_UNAUTHORIZED: + decision = throttle.record_failure(**context) + if not decision.allowed: + _raise_login_throttled(decision) + raise + + throttle.record_success( + normalized_email=normalized_email, + tenant_slug=payload.tenant_slug, + ) + return resolved + + def _extract_auth_token(request: Request) -> tuple[str | None, str]: x_api_key = request.headers.get("x-api-key") if x_api_key: @@ -620,7 +687,7 @@ def _me_response( @router.post("/login", response_model=LoginResponse) def login(payload: LoginRequest, request: Request, response: Response, session: Session = Depends(get_session)): - account, user, tenant = _resolve_login_user(session, payload) + account, user, tenant = _resolve_throttled_login_user(session, payload, request) identity_directory = _identity_directory_from_request(request) authorization_context = collect_user_authorization_context( session, diff --git a/src/govoplan_access/backend/security/login_throttle.py b/src/govoplan_access/backend/security/login_throttle.py new file mode 100644 index 0000000..73f2dfb --- /dev/null +++ b/src/govoplan_access/backend/security/login_throttle.py @@ -0,0 +1,358 @@ +from __future__ import annotations + +import hashlib +import logging +import threading +import time +from dataclasses import dataclass +from typing import Protocol + +from redis import Redis +from redis.exceptions import RedisError + + +logger = logging.getLogger(__name__) + +_REDIS_INCREMENT_SCRIPT = """ +local count = redis.call('INCR', KEYS[1]) +if count == 1 then + redis.call('EXPIRE', KEYS[1], ARGV[1]) +end +local ttl = redis.call('TTL', KEYS[1]) +return {count, ttl} +""" + + +@dataclass(frozen=True, slots=True) +class AttemptBucket: + count: int = 0 + retry_after_seconds: int = 0 + + +@dataclass(frozen=True, slots=True) +class LoginThrottleDecision: + allowed: bool + retry_after_seconds: int = 0 + + +class LoginAttemptStore(Protocol): + def read(self, key: str) -> AttemptBucket: ... + + def increment(self, key: str, *, window_seconds: int) -> AttemptBucket: ... + + def delete(self, key: str) -> None: ... + + +class InMemoryLoginAttemptStore: + """Bounded process-local fallback for development and Redis outages.""" + + def __init__(self, *, max_entries: int = 10_000) -> None: + self._entries: dict[str, tuple[int, float]] = {} + self._lock = threading.Lock() + self._max_entries = max(2, max_entries) + + def read(self, key: str) -> AttemptBucket: + now = time.monotonic() + with self._lock: + entry = self._active_entry(key, now=now) + if entry is None: + return AttemptBucket() + count, expires_at = entry + return AttemptBucket( + count=count, + retry_after_seconds=max(1, int(expires_at - now + 0.999)), + ) + + def increment(self, key: str, *, window_seconds: int) -> AttemptBucket: + now = time.monotonic() + with self._lock: + entry = self._active_entry(key, now=now) + if entry is None: + self._make_room(now=now, incoming_key=key) + count = 1 + expires_at = now + window_seconds + else: + count = entry[0] + 1 + expires_at = entry[1] + self._entries[key] = (count, expires_at) + return AttemptBucket( + count=count, + retry_after_seconds=max(1, int(expires_at - now + 0.999)), + ) + + def delete(self, key: str) -> None: + with self._lock: + self._entries.pop(key, None) + + def _active_entry(self, key: str, *, now: float) -> tuple[int, float] | None: + entry = self._entries.get(key) + if entry is None: + return None + if entry[1] <= now: + self._entries.pop(key, None) + return None + return entry + + def _make_room(self, *, now: float, incoming_key: str) -> None: + if incoming_key in self._entries or len(self._entries) < self._max_entries: + return + expired = [key for key, (_, expires_at) in self._entries.items() if expires_at <= now] + for key in expired: + self._entries.pop(key, None) + while len(self._entries) >= self._max_entries: + self._entries.pop(next(iter(self._entries))) + + +class RedisLoginAttemptStore: + """Redis-backed fixed-window counters shared by all API workers.""" + + def __init__(self, redis_url: str) -> None: + self._client = Redis.from_url( + redis_url, + decode_responses=True, + socket_connect_timeout=0.25, + socket_timeout=0.25, + health_check_interval=30, + ) + + def read(self, key: str) -> AttemptBucket: + pipeline = self._client.pipeline(transaction=False) + pipeline.get(key) + pipeline.ttl(key) + raw_count, raw_ttl = pipeline.execute() + count = int(raw_count or 0) + ttl = int(raw_ttl or 0) + return AttemptBucket(count=count, retry_after_seconds=max(0, ttl)) + + def increment(self, key: str, *, window_seconds: int) -> AttemptBucket: + result = self._client.eval(_REDIS_INCREMENT_SCRIPT, 1, key, window_seconds) + if not isinstance(result, (list, tuple)) or len(result) != 2: + raise RedisError("Unexpected login throttle response from Redis") + return AttemptBucket( + count=int(result[0]), + retry_after_seconds=max(1, int(result[1])), + ) + + def delete(self, key: str) -> None: + self._client.delete(key) + + +class ResilientLoginAttemptStore: + """Prefer the distributed store and fail safely to a local bounded store.""" + + def __init__( + self, + primary: LoginAttemptStore | None, + fallback: LoginAttemptStore, + *, + retry_seconds: int = 30, + ) -> None: + self._primary = primary + self._fallback = fallback + self._retry_seconds = max(1, retry_seconds) + self._primary_unavailable_until = 0.0 + self._state_lock = threading.Lock() + + def read(self, key: str) -> AttemptBucket: + fallback_result = self._fallback.read(key) + primary = self._available_primary() + if primary is None: + return fallback_result + try: + return _stricter_bucket(primary.read(key), fallback_result) + except (RedisError, OSError, TimeoutError, ConnectionError) as exc: + self._mark_primary_unavailable(exc) + return fallback_result + + def increment(self, key: str, *, window_seconds: int) -> AttemptBucket: + primary = self._available_primary() + if primary is not None: + fallback_result = self._fallback.increment(key, window_seconds=window_seconds) + try: + # Mirror the active process's failures so a later Redis outage + # or recovery cannot restart its protection window from zero. + primary_result = primary.increment(key, window_seconds=window_seconds) + return _stricter_bucket(primary_result, fallback_result) + except (RedisError, OSError, TimeoutError, ConnectionError) as exc: + self._mark_primary_unavailable(exc) + return fallback_result + return self._fallback.increment(key, window_seconds=window_seconds) + + def delete(self, key: str) -> None: + self._fallback.delete(key) + primary = self._available_primary() + if primary is None: + return + try: + primary.delete(key) + except (RedisError, OSError, TimeoutError, ConnectionError) as exc: + self._mark_primary_unavailable(exc) + + def _available_primary(self) -> LoginAttemptStore | None: + if self._primary is None: + return None + with self._state_lock: + if time.monotonic() < self._primary_unavailable_until: + return None + return self._primary + + def _mark_primary_unavailable(self, exc: Exception) -> None: + should_log = False + with self._state_lock: + now = time.monotonic() + if now >= self._primary_unavailable_until: + should_log = True + self._primary_unavailable_until = now + self._retry_seconds + if should_log: + logger.warning( + "Redis login throttling is unavailable; using the process-local fallback (%s)", + type(exc).__name__, + ) + + +def _stricter_bucket(first: AttemptBucket, second: AttemptBucket) -> AttemptBucket: + return AttemptBucket( + count=max(first.count, second.count), + retry_after_seconds=max(first.retry_after_seconds, second.retry_after_seconds), + ) + + +class LoginThrottle: + def __init__( + self, + store: LoginAttemptStore, + *, + identity_limit: int, + client_limit: int, + window_seconds: int, + key_prefix: str = "govoplan:access:login:v1", + ) -> None: + self._store = store + self._identity_limit = max(1, identity_limit) + self._client_limit = max(1, client_limit) + self._window_seconds = max(1, window_seconds) + self._key_prefix = key_prefix.rstrip(":") + + def check( + self, + *, + normalized_email: str, + tenant_slug: str | None, + client_address: str | None, + ) -> LoginThrottleDecision: + return self._decision( + self._buckets( + normalized_email=normalized_email, + tenant_slug=tenant_slug, + client_address=client_address, + ), + increment=False, + ) + + def record_failure( + self, + *, + normalized_email: str, + tenant_slug: str | None, + client_address: str | None, + ) -> LoginThrottleDecision: + return self._decision( + self._buckets( + normalized_email=normalized_email, + tenant_slug=tenant_slug, + client_address=client_address, + ), + increment=True, + ) + + def record_success( + self, + *, + normalized_email: str, + tenant_slug: str | None, + ) -> None: + del tenant_slug + identity_key, _ = self._keys( + normalized_email=normalized_email, + client_address=None, + ) + self._store.delete(identity_key) + + def _decision( + self, + buckets: tuple[tuple[str, int], ...], + *, + increment: bool, + ) -> LoginThrottleDecision: + blocked_retry_after = 0 + for key, limit in buckets: + state = ( + self._store.increment(key, window_seconds=self._window_seconds) + if increment + else self._store.read(key) + ) + if state.count >= limit: + blocked_retry_after = max( + blocked_retry_after, + max(1, state.retry_after_seconds), + ) + return LoginThrottleDecision( + allowed=blocked_retry_after == 0, + retry_after_seconds=blocked_retry_after, + ) + + def _buckets( + self, + *, + normalized_email: str, + tenant_slug: str | None, + client_address: str | None, + ) -> tuple[tuple[str, int], ...]: + # Accounts and their passwords are global login identities. Do not put + # the caller-supplied tenant slug into the identity key: rotating fake + # slugs must not bypass the account-level limit. + del tenant_slug + identity_key, client_key = self._keys( + normalized_email=normalized_email, + client_address=client_address, + ) + buckets = [(identity_key, self._identity_limit)] + if client_key is not None: + buckets.append((client_key, self._client_limit)) + return tuple(buckets) + + def _keys( + self, + *, + normalized_email: str, + client_address: str | None, + ) -> tuple[str, str | None]: + identity = normalized_email.strip().casefold() + identity_digest = hashlib.sha256(identity.encode()).hexdigest() + identity_key = f"{self._key_prefix}:identity:{identity_digest}" + if not client_address: + return identity_key, None + client_digest = hashlib.sha256(client_address.strip().casefold().encode()).hexdigest() + return identity_key, f"{self._key_prefix}:client:{client_digest}" + + +def build_login_throttle( + *, + redis_url: str | None, + identity_limit: int, + client_limit: int, + window_seconds: int, + redis_retry_seconds: int, +) -> LoginThrottle: + redis_store = RedisLoginAttemptStore(redis_url) if redis_url and redis_url.strip() else None + resilient_store = ResilientLoginAttemptStore( + redis_store, + InMemoryLoginAttemptStore(), + retry_seconds=redis_retry_seconds, + ) + return LoginThrottle( + resilient_store, + identity_limit=identity_limit, + client_limit=client_limit, + window_seconds=window_seconds, + ) diff --git a/tests/test_login_throttle.py b/tests/test_login_throttle.py new file mode 100644 index 0000000..ac90fa4 --- /dev/null +++ b/tests/test_login_throttle.py @@ -0,0 +1,219 @@ +from __future__ import annotations + +import unittest +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from fastapi import HTTPException +from redis.exceptions import RedisError + +from govoplan_access.backend.api.v1 import auth +from govoplan_access.backend.security.login_throttle import ( + AttemptBucket, + InMemoryLoginAttemptStore, + LoginThrottle, + LoginThrottleDecision, + ResilientLoginAttemptStore, +) +from govoplan_core.api.v1.schemas import LoginRequest + + +class LoginThrottleTests(unittest.TestCase): + def test_identity_and_client_buckets_enforce_independent_limits(self) -> None: + throttle = LoginThrottle( + InMemoryLoginAttemptStore(), + identity_limit=2, + client_limit=3, + window_seconds=60, + ) + context = { + "normalized_email": "person@example.test", + "tenant_slug": "tenant-a", + "client_address": "192.0.2.4", + } + + self.assertTrue(throttle.record_failure(**context).allowed) + self.assertFalse(throttle.record_failure(**context).allowed) + + other_identity = {**context, "normalized_email": "other@example.test"} + self.assertFalse(throttle.record_failure(**other_identity).allowed) + + def test_success_clears_identity_bucket_without_erasing_client_failures(self) -> None: + throttle = LoginThrottle( + InMemoryLoginAttemptStore(), + identity_limit=2, + client_limit=3, + window_seconds=60, + ) + context = { + "normalized_email": "person@example.test", + "tenant_slug": "tenant-a", + "client_address": "192.0.2.8", + } + self.assertTrue(throttle.record_failure(**context).allowed) + + throttle.record_success( + normalized_email=context["normalized_email"], + tenant_slug=context["tenant_slug"], + ) + + self.assertTrue(throttle.check(**context).allowed) + other_identity = {**context, "normalized_email": "other@example.test"} + self.assertTrue(throttle.record_failure(**other_identity).allowed) + self.assertFalse(throttle.record_failure(**other_identity).allowed) + + def test_rotating_untrusted_tenant_slug_does_not_bypass_identity_limit(self) -> None: + throttle = LoginThrottle( + InMemoryLoginAttemptStore(), + identity_limit=2, + client_limit=100, + window_seconds=60, + ) + + self.assertTrue( + throttle.record_failure( + normalized_email="person@example.test", + tenant_slug="tenant-a", + client_address="192.0.2.1", + ).allowed + ) + self.assertFalse( + throttle.record_failure( + normalized_email="person@example.test", + tenant_slug="made-up-tenant", + client_address="198.51.100.9", + ).allowed + ) + + def test_bucket_keys_do_not_contain_identity_or_client_data(self) -> None: + store = MagicMock() + store.increment.return_value = AttemptBucket(count=1, retry_after_seconds=60) + throttle = LoginThrottle( + store, + identity_limit=10, + client_limit=100, + window_seconds=60, + ) + + throttle.record_failure( + normalized_email="private.person@example.test", + tenant_slug="private-tenant", + client_address="192.0.2.9", + ) + + keys = [call.args[0] for call in store.increment.call_args_list] + self.assertEqual(len(keys), 2) + for key in keys: + self.assertNotIn("private", key) + self.assertNotIn("example", key) + self.assertNotIn("192.0.2.9", key) + + def test_redis_failure_uses_local_store_during_retry_window(self) -> None: + primary = MagicMock() + primary.read.side_effect = RedisError("not available") + fallback = InMemoryLoginAttemptStore() + store = ResilientLoginAttemptStore(primary, fallback, retry_seconds=60) + + with self.assertLogs( + "govoplan_access.backend.security.login_throttle", + level="WARNING", + ) as captured: + self.assertEqual(store.read("bucket"), AttemptBucket()) + result = store.increment("bucket", window_seconds=60) + + self.assertEqual(result.count, 1) + self.assertEqual(primary.read.call_count, 1) + primary.increment.assert_not_called() + self.assertIn("process-local fallback", captured.output[0]) + + def test_redis_recovery_does_not_erase_failures_counted_by_the_fallback(self) -> None: + primary = MagicMock() + primary.read.side_effect = [RedisError("not available"), AttemptBucket(count=1, retry_after_seconds=30)] + primary.increment.return_value = AttemptBucket(count=2, retry_after_seconds=30) + fallback = InMemoryLoginAttemptStore() + store = ResilientLoginAttemptStore(primary, fallback, retry_seconds=60) + + with self.assertLogs("govoplan_access.backend.security.login_throttle", level="WARNING"): + store.read("bucket") + store.increment("bucket", window_seconds=60) + store.increment("bucket", window_seconds=60) + + store._primary_unavailable_until = 0 # Simulate the next Redis retry window. + recovered = store.read("bucket") + incremented = store.increment("bucket", window_seconds=60) + + self.assertEqual(recovered.count, 2) + self.assertEqual(incremented.count, 3) + + def test_in_memory_store_is_bounded(self) -> None: + store = InMemoryLoginAttemptStore(max_entries=2) + for key in ("one", "two", "three"): + store.increment(key, window_seconds=60) + + active = sum(store.read(key).count > 0 for key in ("one", "two", "three")) + self.assertEqual(active, 2) + + +class LoginThrottleRouteTests(unittest.TestCase): + def setUp(self) -> None: + self.payload = LoginRequest( + email="Person@Example.Test", + password="attempt", + tenant_slug="tenant-a", + ) + self.request = SimpleNamespace(client=SimpleNamespace(host="192.0.2.10")) + + def test_throttled_identity_gets_same_generic_failure_detail(self) -> None: + throttle = MagicMock() + throttle.check.return_value = LoginThrottleDecision(False, 42) + + with patch.object(auth, "_configured_login_throttle", return_value=throttle): + with self.assertRaises(HTTPException) as raised: + auth._resolve_throttled_login_user(MagicMock(), self.payload, self.request) # type: ignore[arg-type] + + self.assertEqual(raised.exception.status_code, 429) + self.assertEqual(raised.exception.detail, "Invalid login") + self.assertEqual(raised.exception.headers, {"Retry-After": "42"}) + + def test_failed_login_is_counted_and_threshold_response_stays_generic(self) -> None: + throttle = MagicMock() + throttle.check.return_value = LoginThrottleDecision(True) + throttle.record_failure.return_value = LoginThrottleDecision(False, 60) + generic_failure = HTTPException(status_code=401, detail="Invalid login") + + with ( + patch.object(auth, "_configured_login_throttle", return_value=throttle), + patch.object(auth, "_resolve_login_user", side_effect=generic_failure), + ): + with self.assertRaises(HTTPException) as raised: + auth._resolve_throttled_login_user(MagicMock(), self.payload, self.request) # type: ignore[arg-type] + + self.assertEqual(raised.exception.status_code, 429) + self.assertEqual(raised.exception.detail, "Invalid login") + throttle.record_failure.assert_called_once_with( + normalized_email="person@example.test", + tenant_slug="tenant-a", + client_address="192.0.2.10", + ) + + def test_success_clears_only_the_identity_bucket(self) -> None: + throttle = MagicMock() + throttle.check.return_value = LoginThrottleDecision(True) + resolved = (object(), object(), object()) + + with ( + patch.object(auth, "_configured_login_throttle", return_value=throttle), + patch.object(auth, "_resolve_login_user", return_value=resolved), + ): + result = auth._resolve_throttled_login_user(MagicMock(), self.payload, self.request) # type: ignore[arg-type] + + self.assertEqual(result, resolved) + throttle.record_failure.assert_not_called() + throttle.record_success.assert_called_once_with( + normalized_email="person@example.test", + tenant_slug="tenant-a", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_optional_tenancy_contract.py b/tests/test_optional_tenancy_contract.py index f348cc5..464143d 100644 --- a/tests/test_optional_tenancy_contract.py +++ b/tests/test_optional_tenancy_contract.py @@ -15,7 +15,7 @@ class OptionalTenancyContractTests(unittest.TestCase): dependencies = tuple(project["dependencies"]) - self.assertIn("govoplan-core>=0.1.8", dependencies) + self.assertIn("govoplan-core>=0.1.9", dependencies) self.assertNotIn("govoplan-tenancy>=0.1.8", dependencies) self.assertFalse(any(item.startswith("govoplan-tenancy") for item in dependencies))