From 16e876f607cd3dedcbacd12f6539ee94292e0dbb Mon Sep 17 00:00:00 2001 From: portable Date: Thu, 20 Aug 2026 17:36:02 +0100 Subject: [PATCH 1/2] fix(pause): make protocol pause Redis-backed and constant-time - PauseController now uses redis.asyncio for cross-worker state - Pause survives worker restarts (Redis key protocol:paused) - Fail-closed: defaults to paused on Redis outage for safety - 5s in-memory cache to avoid Redis round-trip on every request - verify_admin_token uses hmac.compare_digest for constant-time comparison - Router endpoints use APIError envelope - 23 tests covering Redis state, cache, fail-closed, exempt paths Closes #416 --- quantara/web_app/api/pausable.py | 170 ++++++++++++++---- quantara/web_app/tests/test_pausable.py | 230 ++++++++++++++++++++---- 2 files changed, 336 insertions(+), 64 deletions(-) diff --git a/quantara/web_app/api/pausable.py b/quantara/web_app/api/pausable.py index e72c3532..f0b56f5f 100644 --- a/quantara/web_app/api/pausable.py +++ b/quantara/web_app/api/pausable.py @@ -1,59 +1,147 @@ """ Admin pause controls and request guard for incident response. + +Pause state is persisted in Redis so it propagates across all worker +processes. A short in-memory cache avoids a Redis round-trip on every +request. If Redis is unreachable the controller defaults to *paused* +(fail-closed) so that an infrastructure outage automatically suspends +user-facing operations. """ +from __future__ import annotations + +import asyncio +import hmac import os -from dataclasses import dataclass -from threading import Lock +import time -from fastapi import APIRouter, Header, HTTPException, Request +import redis.asyncio as aioredis +from fastapi import APIRouter, Header, Request from fastapi.responses import JSONResponse +from web_app.api.errors import APIError + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- PROTOCOL_PAUSED_DETAIL = "Protocol paused" ADMIN_PAUSE_PREFIX = "/api/admin/pause" PAUSE_ADMIN_TOKEN_ENV = "PAUSE_ADMIN_TOKEN" +REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379") +PAUSE_KEY = "protocol:paused" -@dataclass(frozen=True) -class PauseStatus: - paused: bool - +_CACHE_TTL = 5 # seconds -class PauseController: - """Thread-safe in-process protocol pause switch.""" - def __init__(self) -> None: - self._paused = False - self._lock = Lock() +# --------------------------------------------------------------------------- +# PauseController — Redis-backed with in-memory cache +# --------------------------------------------------------------------------- - def pause(self) -> PauseStatus: - with self._lock: - self._paused = True - return PauseStatus(paused=self._paused) - def unpause(self) -> PauseStatus: - with self._lock: - self._paused = False - return PauseStatus(paused=self._paused) +class PauseController: + """Redis-backed protocol pause switch. + + The pause flag is stored under ``PAUSE_KEY`` (``"1"`` = paused, key + absent = unpaused). A tiny in-memory cache with a configurable TTL + prevents a Redis call on every single request. + + **Fail-closed**: if Redis is unavailable the controller reports + *paused* so that an incident that takes down Redis also takes down + user-facing operations (the safe default). + """ + + def __init__(self, redis_url: str = REDIS_URL) -> None: + self._redis_url = redis_url + self._redis: aioredis.Redis | None = None + + # in-memory cache + self._cache_value: bool = False + self._cache_ts: float = 0.0 + + # -- Redis lifecycle ---------------------------------------------------- + + async def _get_redis(self) -> aioredis.Redis: + """Lazy-initialise the Redis connection on first use.""" + if self._redis is None: + self._redis = aioredis.from_url( + self._redis_url, decode_responses=True + ) + return self._redis + + # -- public async API --------------------------------------------------- + + async def pause(self) -> bool: + """Set the pause flag in Redis. Returns the new paused state.""" + client = await self._get_redis() + await client.set(PAUSE_KEY, "1") + self._invalidate_cache() + return True + + async def unpause(self) -> bool: + """Clear the pause flag in Redis. Returns the new paused state.""" + client = await self._get_redis() + await client.delete(PAUSE_KEY) + self._invalidate_cache() + return False + + async def is_paused(self) -> bool: + """Return whether the protocol is paused. + + Uses a short-lived in-memory cache to avoid hitting Redis on every + request. On Redis failure, defaults to *paused* (fail-closed). + """ + now = time.monotonic() + if (now - self._cache_ts) < _CACHE_TTL: + return self._cache_value + + try: + client = await self._get_redis() + value = await client.get(PAUSE_KEY) + paused = value == "1" + except (aioredis.RedisError, OSError): + paused = True # fail-closed + + self._cache_value = paused + self._cache_ts = now + return paused + + # -- cache helpers ------------------------------------------------------ + + def _invalidate_cache(self) -> None: + self._cache_ts = 0.0 + + +# Module-level singleton used by the middleware and router. +pause_controller = PauseController() - def status(self) -> PauseStatus: - with self._lock: - return PauseStatus(paused=self._paused) +router = APIRouter(prefix=ADMIN_PAUSE_PREFIX, tags=["Admin"]) -pause_controller = PauseController() -router = APIRouter(prefix=ADMIN_PAUSE_PREFIX, tags=["Admin"]) +# --------------------------------------------------------------------------- +# Admin token helpers +# --------------------------------------------------------------------------- def _admin_token() -> str | None: return os.getenv(PAUSE_ADMIN_TOKEN_ENV) or os.getenv("ADMIN_API_KEY") -def verify_admin_token(x_admin_token: str = Header(...)) -> None: +def verify_admin_token(x_admin_token: str) -> None: + """Raise :class:`APIError` if the supplied token does not match.""" expected_token = _admin_token() - if not expected_token or x_admin_token != expected_token: - raise HTTPException(status_code=403, detail="Admin authorization required") + if not expected_token or not hmac.compare_digest(x_admin_token, expected_token): + raise APIError( + status_code=403, + code="admin_auth_required", + detail="Admin authorization required", + ) + + +# --------------------------------------------------------------------------- +# Exempt-path check +# --------------------------------------------------------------------------- def is_pause_exempt_path(path: str) -> bool: @@ -64,23 +152,37 @@ def is_pause_exempt_path(path: str) -> bool: ) +# --------------------------------------------------------------------------- +# ASGI middleware +# --------------------------------------------------------------------------- + + async def protocol_pause_middleware(request: Request, call_next): if ( request.url.path.startswith("/api/") and not is_pause_exempt_path(request.url.path) - and pause_controller.status().paused + and await pause_controller.is_paused() ): - return JSONResponse(status_code=503, content={"detail": PROTOCOL_PAUSED_DETAIL}) + return JSONResponse( + status_code=503, + content={"detail": PROTOCOL_PAUSED_DETAIL}, + ) return await call_next(request) +# --------------------------------------------------------------------------- +# Router endpoints +# --------------------------------------------------------------------------- + + @router.get("", summary="Get protocol pause status") async def get_pause_status( x_admin_token: str = Header(..., alias="X-Admin-Token"), ) -> dict: verify_admin_token(x_admin_token) - return {"paused": pause_controller.status().paused} + paused = await pause_controller.is_paused() + return {"paused": paused} @router.post("", summary="Pause protocol user-facing operations") @@ -88,7 +190,8 @@ async def pause_protocol( x_admin_token: str = Header(..., alias="X-Admin-Token"), ) -> dict: verify_admin_token(x_admin_token) - return {"paused": pause_controller.pause().paused} + paused = await pause_controller.pause() + return {"paused": paused} @router.delete("", summary="Unpause protocol user-facing operations") @@ -96,4 +199,5 @@ async def unpause_protocol( x_admin_token: str = Header(..., alias="X-Admin-Token"), ) -> dict: verify_admin_token(x_admin_token) - return {"paused": pause_controller.unpause().paused} + paused = await pause_controller.unpause() + return {"paused": paused} diff --git a/quantara/web_app/tests/test_pausable.py b/quantara/web_app/tests/test_pausable.py index dff98d77..cb521ea3 100644 --- a/quantara/web_app/tests/test_pausable.py +++ b/quantara/web_app/tests/test_pausable.py @@ -1,46 +1,214 @@ -from unittest.mock import MagicMock +"""Tests for the Redis-backed pause controller and related helpers.""" -from fastapi.testclient import TestClient +from __future__ import annotations -from web_app.api.main import app -from web_app.api.pausable import pause_controller -from web_app.db.database import get_database +import asyncio +import hmac +from unittest.mock import AsyncMock, MagicMock, patch +import pytest +import redis.asyncio as aioredis -ADMIN_HEADERS = {"X-Admin-Token": "test-admin-token"} +from web_app.api.pausable import ( + PauseController, + is_pause_exempt_path, + pause_controller, + verify_admin_token, +) +# --------------------------------------------------------------------------- +# PauseController unit tests (no real Redis needed) +# --------------------------------------------------------------------------- -def test_admin_can_pause_and_unpause_protocol(monkeypatch): - monkeypatch.setenv("PAUSE_ADMIN_TOKEN", "test-admin-token") - pause_controller.unpause() - app.dependency_overrides[get_database] = lambda: MagicMock() - client = TestClient(app) - assert client.get("/api/admin/pause", headers=ADMIN_HEADERS).json() == { - "paused": False - } +@pytest.fixture +def ctrl() -> PauseController: + """Return a fresh PauseController with a mock Redis client.""" + controller = PauseController.__new__(PauseController) + controller._redis_url = "redis://localhost:6379" + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=None) + mock_client.set = AsyncMock() + mock_client.delete = AsyncMock() + controller._redis = mock_client + controller._cache_value = False + controller._cache_ts = 0.0 + return controller - pause_response = client.post("/api/admin/pause", headers=ADMIN_HEADERS) - assert pause_response.status_code == 200 - assert pause_response.json() == {"paused": True} - blocked_response = client.get("/api/check-user?wallet_id=test-wallet") - assert blocked_response.status_code == 503 - assert blocked_response.json() == {"detail": "Protocol paused"} +@pytest.mark.asyncio +async def test_pause_sets_redis_key(ctrl: PauseController): + result = await ctrl.pause() + ctrl._redis.set.assert_awaited_once_with("protocol:paused", "1") + assert result is True - unpause_response = client.delete("/api/admin/pause", headers=ADMIN_HEADERS) - assert unpause_response.status_code == 200 - assert unpause_response.json() == {"paused": False} - app.dependency_overrides.clear() +@pytest.mark.asyncio +async def test_unpause_deletes_redis_key(ctrl: PauseController): + result = await ctrl.unpause() + ctrl._redis.delete.assert_awaited_once_with("protocol:paused") + assert result is False -def test_pause_admin_requires_token(monkeypatch): - monkeypatch.setenv("PAUSE_ADMIN_TOKEN", "test-admin-token") - pause_controller.unpause() - client = TestClient(app) +@pytest.mark.asyncio +async def test_is_paused_true_when_key_present(ctrl: PauseController): + ctrl._redis.get = AsyncMock(return_value="1") + assert await ctrl.is_paused() is True - response = client.post("/api/admin/pause", headers={"X-Admin-Token": "wrong"}) - assert response.status_code == 403 - assert response.json() == {"detail": "Admin authorization required"} +@pytest.mark.asyncio +async def test_is_paused_false_when_key_absent(ctrl: PauseController): + ctrl._redis.get = AsyncMock(return_value=None) + assert await ctrl.is_paused() is False + + +@pytest.mark.asyncio +async def test_cache_avoids_repeated_redis_calls(ctrl: PauseController): + ctrl._redis.get = AsyncMock(return_value=None) + # first call — cache miss + assert await ctrl.is_paused() is False + assert ctrl._redis.get.await_count == 1 + # second call — served from cache, no extra Redis call + assert await ctrl.is_paused() is False + assert ctrl._redis.get.await_count == 1 + + +@pytest.mark.asyncio +async def test_cache_invalidation_on_pause(ctrl: PauseController): + ctrl._redis.get = AsyncMock(return_value=None) + await ctrl.is_paused() + await ctrl.pause() + ctrl._redis.get = AsyncMock(return_value="1") + # cache should be invalidated, so this hits Redis again + assert await ctrl.is_paused() is True + + +@pytest.mark.asyncio +async def test_fail_closed_on_redis_error(ctrl: PauseController): + ctrl._redis.get = AsyncMock(side_effect=aioredis.RedisError("connection refused")) + assert await ctrl.is_paused() is True + + +@pytest.mark.asyncio +async def test_fail_closed_on_os_error(ctrl: PauseController): + ctrl._redis.get = AsyncMock(side_effect=OSError("network unreachable")) + assert await ctrl.is_paused() is True + + +@pytest.mark.asyncio +async def test_cache_refreshes_after_ttl(ctrl: PauseController): + import time + + ctrl._redis.get = AsyncMock(return_value=None) + await ctrl.is_paused() + assert ctrl._redis.get.await_count == 1 + + # simulate TTL expiry + ctrl._cache_ts = time.monotonic() - 10 + ctrl._redis.get = AsyncMock(return_value="1") + assert await ctrl.is_paused() is True + assert ctrl._redis.get.await_count == 1 + + +# --------------------------------------------------------------------------- +# verify_admin_token — constant-time comparison +# --------------------------------------------------------------------------- + + +def test_verify_admin_token_accepts_valid(monkeypatch): + monkeypatch.setenv("PAUSE_ADMIN_TOKEN", "super-secret") + verify_admin_token("super-secret") # should not raise + + +def test_verify_admin_token_rejects_wrong(monkeypatch): + monkeypatch.setenv("PAUSE_ADMIN_TOKEN", "super-secret") + with pytest.raises(Exception): + verify_admin_token("wrong-token") + + +def test_verify_admin_token_uses_constant_time(monkeypatch): + monkeypatch.setenv("PAUSE_ADMIN_TOKEN", "secret") + with patch("web_app.api.pausable.hmac.compare_digest", wraps=hmac.compare_digest) as mock_cmp: + verify_admin_token("secret") + mock_cmp.assert_called_once_with("secret", "secret") + + +def test_verify_admin_token_raises_api_error(monkeypatch): + monkeypatch.setenv("PAUSE_ADMIN_TOKEN", "tok") + from web_app.api.errors import APIError + with pytest.raises(APIError) as exc_info: + verify_admin_token("bad") + assert exc_info.value.status_code == 403 + assert exc_info.value.code == "admin_auth_required" + + +# --------------------------------------------------------------------------- +# Exempt paths +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "path", + [ + "/api/admin/pause", + "/api/admin/pause/status", + "/health", + "/metrics/something", + ], +) +def test_exempt_paths(path: str): + assert is_pause_exempt_path(path) is True + + +def test_non_exempt_path(): + assert is_pause_exempt_path("/api/check-user") is False + + +# --------------------------------------------------------------------------- +# Integration-style tests — middleware blocks when paused +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_middleware_blocks_when_paused(): + with patch.object(pause_controller, "is_paused", new_callable=lambda: AsyncMock(return_value=True)): + mock_request = MagicMock() + mock_request.url.path = "/api/check-user" + mock_call_next = AsyncMock(return_value=MagicMock(status_code=200)) + + from web_app.api.pausable import protocol_pause_middleware + + response = await protocol_pause_middleware(mock_request, mock_call_next) + assert response.status_code == 503 + mock_call_next.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_middleware_passes_when_unpaused(): + with patch.object(pause_controller, "is_paused", new_callable=lambda: AsyncMock(return_value=False)): + mock_request = MagicMock() + mock_request.url.path = "/api/check-user" + mock_call_next = AsyncMock(return_value=MagicMock(status_code=200)) + + from web_app.api.pausable import protocol_pause_middleware + + response = await protocol_pause_middleware(mock_request, mock_call_next) + assert response.status_code == 200 + mock_call_next.assert_awaited_once() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "path", + ["/health", "/metrics/foo", "/api/admin/pause"], +) +async def test_middleware_exempt_paths_always_pass(path: str): + with patch.object(pause_controller, "is_paused", new_callable=lambda: AsyncMock(return_value=True)): + mock_request = MagicMock() + mock_request.url.path = path + mock_call_next = AsyncMock(return_value=MagicMock(status_code=200)) + + from web_app.api.pausable import protocol_pause_middleware + + response = await protocol_pause_middleware(mock_request, mock_call_next) + assert response.status_code == 200 From d163249d45fb7438864dc1ea6d32a5103055079f Mon Sep 17 00:00:00 2001 From: portable Date: Fri, 21 Aug 2026 13:14:28 +0100 Subject: [PATCH 2/2] fix(tests): disable pause controller in test fixtures to prevent Redis fail-closed 503s --- quantara/web_app/tests/conftest.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/quantara/web_app/tests/conftest.py b/quantara/web_app/tests/conftest.py index af399e56..077ba1c3 100644 --- a/quantara/web_app/tests/conftest.py +++ b/quantara/web_app/tests/conftest.py @@ -11,6 +11,7 @@ from sqlalchemy.orm import scoped_session from web_app.api.main import app +from web_app.api.pausable import pause_controller from web_app.api.rate_limiter import limiter as _ORIGINAL_LIMITER from web_app.api.wallet_auth import verify_wallet_signature from web_app.db.crud import DBConnector, PositionDBConnector, UserDBConnector @@ -18,6 +19,21 @@ from web_app.db.models import ExtraDeposit +@pytest.fixture(autouse=True) +def disable_pause_controller(): + """Force the protocol pause controller to report *not paused* during tests. + + The Redis-backed PauseController defaults to *paused* (fail-closed) when + Redis is unreachable, which causes every /api/ request to return 503 in + CI environments that don't have a Redis service. We bypass this by + patching ``is_paused`` to always return ``False``. + """ + patcher = patch.object(pause_controller, "is_paused", new_callable=AsyncMock, return_value=False) + patcher.start() + yield + patcher.stop() + + @pytest.fixture(autouse=True) def disable_rate_limiting(): """Disable rate limiting in all tests to avoid Redis dependency.