diff --git a/CHANGELOG.md b/CHANGELOG.md index a68e93a..3660903 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed (Phase D review batch) +- **Runtime prefetch registration is executable** - `SchedulerClient.register_volatile_fetch` (used by consolidation's `_register_prefetch`) registered tasks that were dead on arrival three ways: the JSON body sat under the ignored `body` key (`rest_api_executor` only reads `config["payload"]`), there was no `auth` block (the scheduled POST would 401 against library-desk's `verify_api_key`), and `user` was in the body while every `/volatile/fetch` endpoint requires it as a QUERY parameter (would 422). The config now puts `user` in the URL query string (URL-encoded), an empty `payload`, and `auth: {type: bearer, token: "${LIBRARY_API_KEY}"}` (substituted from the Scheduler's environment; never stored raw). `SchedulerClient` itself also sent no Authorization to the Scheduler API, so registration failed silently at consolidation time — it now sends `Authorization: Bearer ` (new `scheduler_api_key` setting; a client without a key logs a warning). - **Registrar authenticates to the Scheduler API** - `scripts/register_scheduler_tasks.py --execute` sent no `Authorization` header while the Scheduler's task-management endpoints are Bearer-guarded (`verify_api_key`: 401 on missing key), so the existence probes 401'd (misread as "task absent") and every registration failed; only the public `/health` gate passed. `--execute` now requires `SCHEDULER_API_KEY` in the environment (refuses to run without it, key is never stored) and sends `Authorization: Bearer $SCHEDULER_API_KEY` on all of its own HTTP calls. Deploy note updated alongside the existing `LIBRARY_API_KEY` requirement. ### Fixed (hazards batch) diff --git a/src/clients/scheduler_client.py b/src/clients/scheduler_client.py index 21a36c2..b75c14a 100644 --- a/src/clients/scheduler_client.py +++ b/src/clients/scheduler_client.py @@ -8,10 +8,17 @@ Registers and manages scheduled tasks for prefetch operations import httpx import logging from typing import Optional, Any +from urllib.parse import quote from pydantic import BaseModel, Field logger = logging.getLogger(__name__) +#: Literal placeholder stored in task auth.token; the Scheduler's +#: rest_api_executor substitutes ${ENV_VAR} from ITS OWN environment at +#: execution time, so the raw library-desk key is never stored in the +#: scheduled_tasks.config column. +LIBRARY_API_KEY_PLACEHOLDER = "${LIBRARY_API_KEY}" + class SchedulerTask(BaseModel): """Task definition for scheduler registration.""" @@ -39,24 +46,38 @@ class SchedulerTask(BaseModel): class SchedulerClient: """Client for external scheduler service.""" - def __init__(self, base_url: str, timeout: float = 30.0): + def __init__(self, base_url: str, api_key: str = "", timeout: float = 30.0): """ Initialize scheduler client. Args: base_url: Scheduler API base URL (e.g., "http://scheduler:8090") + api_key: Bearer key for the Scheduler's task-management API. + The /tasks endpoints are guarded by verify_api_key (401 when + the Authorization header is missing), so without this key + every registration call fails. timeout: HTTP request timeout in seconds """ self.base_url = base_url.rstrip("/") + self.api_key = api_key self.timeout = timeout self._client: Optional[httpx.AsyncClient] = None + if not api_key: + logger.warning( + "SchedulerClient created without an API key; task-management " + "calls will be rejected by the Scheduler (401)" + ) async def _get_client(self) -> httpx.AsyncClient: - """Get or create HTTP client.""" + """Get or create HTTP client (with Scheduler API Bearer auth).""" if self._client is None or self._client.is_closed: + headers = ( + {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {} + ) self._client = httpx.AsyncClient( base_url=self.base_url, timeout=self.timeout, + headers=headers, ) return self._client @@ -301,13 +322,25 @@ class SchedulerClient: day_of_week=schedule.get("day_of_week", -1), config={ "method": "POST", - "url": f"http://library-desk:8089/volatile/fetch/{namespace}/{key}", + # /volatile/fetch endpoints take user as a REQUIRED QUERY + # parameter (RequiredUserQuery) - a body user would 422. + "url": ( + f"http://library-desk:8089/volatile/fetch/" + f"{namespace}/{key}?user={quote(user, safe='')}" + ), "headers": { "Content-Type": "application/json" }, - "body": { - "user": user - } + # rest_api_executor only reads config["payload"] as the JSON + # body (a "body" key is silently ignored). + "payload": {}, + # Substituted from the Scheduler's environment at execution + # time; without it the scheduled POST 401s against + # library-desk's verify_api_key. + "auth": { + "type": "bearer", + "token": LIBRARY_API_KEY_PLACEHOLDER, + }, } ) diff --git a/src/config.py b/src/config.py index 1df6e61..a6351ea 100644 --- a/src/config.py +++ b/src/config.py @@ -145,6 +145,13 @@ class Settings(BaseSettings): # Scheduler Service scheduler_url: str = Field(default="http://scheduler:8090", description="Scheduler service URL") + scheduler_api_key: str = Field( + default="", + description=( + "Bearer key for the Scheduler's auth-guarded task-management API; " + "required for runtime prefetch task registration" + ), + ) @property def qdrant_url(self) -> str: diff --git a/src/core/dependencies.py b/src/core/dependencies.py index 8fc1cf0..7245e54 100644 --- a/src/core/dependencies.py +++ b/src/core/dependencies.py @@ -259,7 +259,10 @@ def get_scheduler_client() -> SchedulerClient: Note: Used for registering prefetch tasks discovered during HybridRAG searches """ settings = get_settings() - client = SchedulerClient(base_url=settings.scheduler_url) + client = SchedulerClient( + base_url=settings.scheduler_url, + api_key=settings.scheduler_api_key, + ) logger.debug(f"Created Scheduler client: {settings.scheduler_url}") return client diff --git a/tests/test_scheduler_client.py b/tests/test_scheduler_client.py new file mode 100644 index 0000000..e2c8a47 --- /dev/null +++ b/tests/test_scheduler_client.py @@ -0,0 +1,122 @@ +""" +Offline unit tests for SchedulerClient (Phase D review batch). + +Pins the runtime prefetch-registration fixes: +- SchedulerClient sends Authorization: Bearer on its + own calls (the Scheduler's /tasks endpoints are auth-guarded; a bare + client 401s and registration fails silently at consolidation time) +- register_volatile_fetch task config is actually executable by the + Scheduler's rest_api_executor: + * JSON body under config["payload"] (a "body" key is silently ignored) + * user as a QUERY parameter (the /volatile/fetch endpoints use + RequiredUserQuery; a body user would 422) + * an auth block with the ${LIBRARY_API_KEY} placeholder so the + scheduled POST passes library-desk's verify_api_key +""" + +import httpx +import pytest + +from src.clients.scheduler_client import ( + LIBRARY_API_KEY_PLACEHOLDER, + SchedulerClient, +) + + +def _mock_transport(seen): + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + path = request.url.path + if request.method == "GET" and path.startswith("/tasks/"): + return httpx.Response(404) # task does not exist yet + if request.method == "POST" and path == "/tasks": + return httpx.Response(200, json={"ok": True}) + return httpx.Response(200, json={}) + + return httpx.MockTransport(handler) + + +@pytest.fixture +def seen_requests(): + return [] + + +@pytest.fixture +def client(seen_requests, monkeypatch): + """SchedulerClient whose real _get_client builds against a MockTransport.""" + import src.clients.scheduler_client as mod + + real_async_client = httpx.AsyncClient + + def client_factory(**kwargs): + kwargs["transport"] = _mock_transport(seen_requests) + return real_async_client(**kwargs) + + monkeypatch.setattr(mod.httpx, "AsyncClient", client_factory) + return SchedulerClient(base_url="http://scheduler.test:8090", api_key="sched-key") + + +@pytest.mark.unit +class TestSchedulerApiAuth: + @pytest.mark.asyncio + async def test_bearer_auth_sent_on_all_calls(self, client, seen_requests): + ok = await client.register_volatile_fetch( + namespace="weather", + key="rotterdam", + user="llm_tester", + schedule={"minute": 0}, + ) + + assert ok is True + assert seen_requests, "no HTTP calls were made" + for request in seen_requests: + assert request.headers.get("Authorization") == "Bearer sched-key" + + def test_missing_api_key_is_flagged(self, caplog): + with caplog.at_level("WARNING"): + SchedulerClient(base_url="http://scheduler.test:8090") + assert any("without an API key" in r.message for r in caplog.records) + + +@pytest.mark.unit +class TestRegisterVolatileFetchConfig: + @pytest.mark.asyncio + async def _registered_config(self, client, seen_requests, user="llm_tester"): + ok = await client.register_volatile_fetch( + namespace="weather", + key="rotterdam", + user=user, + schedule={"minute": 0, "hour": -1}, + ) + assert ok is True + create = next( + r for r in seen_requests + if r.method == "POST" and r.url.path == "/tasks" + ) + import json + return json.loads(create.content)["config"] + + @pytest.mark.asyncio + async def test_body_key_not_used_payload_is(self, client, seen_requests): + config = await self._registered_config(client, seen_requests) + # rest_api_executor only reads config["payload"]; "body" is ignored + assert "body" not in config + assert config["payload"] == {} + + @pytest.mark.asyncio + async def test_user_is_query_parameter(self, client, seen_requests): + config = await self._registered_config(client, seen_requests) + assert config["url"] == ( + "http://library-desk:8089/volatile/fetch/weather/rotterdam" + "?user=llm_tester" + ) + + @pytest.mark.asyncio + async def test_auth_block_uses_placeholder(self, client, seen_requests): + config = await self._registered_config(client, seen_requests) + assert config["auth"] == { + "type": "bearer", + "token": LIBRARY_API_KEY_PLACEHOLDER, + } + # never the raw key, never in plain headers (not substituted there) + assert "Authorization" not in config.get("headers", {})