diff --git a/.env.example b/.env.example index 026f94a..d2f38e8 100644 --- a/.env.example +++ b/.env.example @@ -5,6 +5,9 @@ APP_MESSAGE="Datastore API is running!" MAX_REQUEST_BODY_MB=50 LOG_LEVEL=INFO +# One JSON analytics event per datastore action / dump request. +# false leaves the analytics middleware unmounted. +ANALYTICS_ENABLED=false # Cross-origin requests: `*` allows every origin, a comma-separated list # allows only those domains (e.g. https://data.example.org,https://app.example.org), # empty disables CORS entirely. diff --git a/datastore/analytics.py b/datastore/analytics.py new file mode 100644 index 0000000..24694a1 --- /dev/null +++ b/datastore/analytics.py @@ -0,0 +1,220 @@ +"""One structured event per datastore action call or dump. + +The event shape is shared with ckanext-analytics: both services emit one +JSON line per request (here on this module's ``datastore.analytics`` logger), all +meant for one BigQuery table (see that repo's ``bigquery.sql``), told apart +by the ``service`` field. Whatever ships the logs (Loki, a GCP log sink, a +file) is what carries events downstream. What the ingress log cannot see is +what this records - the resource a POST body names, who called it, and the +dataset and organization the resource belongs to. + +Two pieces: + +``AnalyticsMiddleware`` + Pure ASGI, so handled error responses are recorded with their status and + an unhandled crash is recorded as a 500 before it propagates. Tracks + ``/api/3/action/*`` and ``/datastore/dump/*``; probes, docs and the + welcome page are excluded by definition. + +``authorization_dict`` + Called by ``RequestContext.authorize`` with the authorized data_dict. + The CKAN auth provider already fetched (and cached) the user, resource + and package on the way to its verdict, so the caller's name and the + resolved dataset / organization names cost the event nothing. Under jwt + or anonymous auth only the JWT subject (if any) and the raw reference + are recorded. +""" + +from __future__ import annotations + +import json +import logging +import uuid +from datetime import datetime, timezone +from typing import Any +from urllib.parse import parse_qs + +from starlette.datastructures import Headers +from starlette.types import ASGIApp, Receive, Scope, Send + +log = logging.getLogger(__name__) + +ACTION_PREFIX = "/api/3/action/" +DUMP_PREFIX = "/datastore/dump/" + + +def action_name(path: str) -> str | None: + """What to call the event for this path, or None if it is not tracked.""" + if path.startswith(ACTION_PREFIX): + name = path[len(ACTION_PREFIX):].strip("/").split("/", 1)[0] + return name or None + if path == "/datastore/dump/query": + return "datastore_dump_query" + if path.startswith(DUMP_PREFIX): + return "datastore_dump" + return None + + +def authorization_dict(request: Any, data_dict: dict[str, Any]) -> None: + """Keep what authorization learned, for the event this request becomes. + + Lives on ``scope["state"]``, which the middleware and the endpoint's + ``Request`` share. First answer wins per field: ``datastore_search_sql`` + authorizes several resources, and the event is attributed to the first. + """ + state: dict[str, Any] = request.scope.setdefault("state", {}) + info: dict[str, Any] = state.setdefault("analytics", {}) + resource = data_dict.get("resource") or {} + package = data_dict.get("package") or {} + organization = package.get("organization") or {} + for key, value in ( + ("user", data_dict.get("user")), + ("resource", resource.get("name") or resource.get("id")), + ("dataset", package.get("name") or package.get("id")), + ("organization", organization.get("name")), + ): + if value and info.get(key) is None: + info[key] = value + + +class AnalyticsMiddleware: + """Record every tracked request, whatever became of it.""" + + METHODS = frozenset({"GET", "POST", "PUT", "PATCH", "DELETE"}) + + #: POST bodies are read only this far - enough for any reference, and a + #: bound on what a bulk upsert costs to copy. + BODY_CAP = 64 * 1024 + + REQUEST_ID_HEADER = "x-request-id" + REAL_IP_HEADER = "x-real-ip" + FORWARDED_FOR_HEADER = "x-forwarded-for" + + def __init__(self, app: ASGIApp, service: str = "datastore-api") -> None: + self.app = app + self.service = service + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http" or scope["method"] not in self.METHODS: + return await self.app(scope, receive, send) + action = action_name(scope["path"]) + if action is None: + return await self.app(scope, receive, send) + + # Created here if authorize has not run yet, so both sides mutate the + # one dict Starlette's `Request.state` also uses. + state: dict[str, Any] = scope.setdefault("state", {}) + status = 500 # what reaches the caller if the app dies before a response + body = bytearray() + capture = scope["method"] == "POST" and scope["path"].startswith(ACTION_PREFIX) + + async def tee_receive() -> Any: + message = await receive() + if capture and message["type"] == "http.request" and len(body) < self.BODY_CAP: + body.extend(message.get("body", b"")[: self.BODY_CAP - len(body)]) + return message + + async def watch_send(message: Any) -> None: + nonlocal status + if message["type"] == "http.response.start": + status = message["status"] + await send(message) + + try: + await self.app(scope, tee_receive, watch_send) + finally: + self._record(scope, action, status, bytes(body), state.get("analytics") or {}) + + def _record( + self, + scope: Scope, + action: str, + status: int, + body: bytes, + resolved: dict[str, Any], + ) -> None: + """Build and emit the event. A listener that raises would become the + request's problem, so it may not.""" + try: + emit_event(self._event(scope, action, status, body, resolved)) + except Exception: + log.exception("analytics: could not record request") + + def _event( + self, + scope: Scope, + action: str, + status: int, + body: bytes, + resolved: dict[str, Any], + ) -> dict[str, Any]: + headers = Headers(scope=scope) + return { + "timestamp": datetime.now(timezone.utc).isoformat(), + "request_id": headers.get(self.REQUEST_ID_HEADER) or uuid.uuid4().hex, + "service": self.service, + "method": scope["method"], + "endpoint": scope["path"], + "query_string": scope.get("query_string", b"").decode("latin-1") or None, + "action_type": action, + "status_code": status, + "user_agent": headers.get("user-agent") or None, + "request_ip": self._request_ip(scope, headers), + "user": resolved.get("user"), + "dataset": resolved.get("dataset"), + "resource": resolved.get("resource") or self._resource_ref(scope, body), + "organization": resolved.get("organization"), + "group": None, + } + + def _request_ip(self, scope: Scope, headers: Headers) -> str | None: + """The caller's address, as the ingress reports it - the same rules + as ckanext-analytics, and like there, fine for analytics only.""" + real_ip = headers.get(self.REAL_IP_HEADER) + if real_ip: + return real_ip.strip() or None + + forwarded = headers.get(self.FORWARDED_FOR_HEADER) + if forwarded: + return forwarded.rsplit(",", 1)[-1].strip() or None + + client = scope.get("client") + return client[0] if client else None + + def _resource_ref(self, scope: Scope, body: bytes) -> str | None: + """The resource reference the caller sent, wherever they put it. + + The dump URL holds it as a path segment, a GET in the query string, + a POST in the JSON body. ``datastore_search_sql`` carries its + resources inside the SQL text, which only authorization can name - + those events rely on ``authorization_dict`` alone. + """ + path: str = scope["path"] + if path.startswith(DUMP_PREFIX): + ref = path[len(DUMP_PREFIX):].split("/", 1)[0] + return ref if ref and ref != "query" else None + + query: bytes = scope.get("query_string", b"") + params = parse_qs(query.decode("latin-1")) + if params.get("resource_id"): + return params["resource_id"][0] + + if body: + try: + parsed = json.loads(body) + except ValueError: + return None + if isinstance(parsed, dict): + sent = parsed.get("resource_id") or parsed.get("id") + return sent if isinstance(sent, str) else None + return None + + +def emit_event(event: dict[str, Any]) -> None: + """Where an event goes: one bare JSON line on the event logger. + + The line is the complete record - whatever ships the logs is what carries + events downstream. ckanext-analytics' ``bigquery.sql`` documents the table + they are meant to land in. + """ + log.info(json.dumps(event, separators=(",", ":"), default=str)) diff --git a/datastore/api/auth.py b/datastore/api/auth.py index 7ed6a7c..25f4bb4 100644 --- a/datastore/api/auth.py +++ b/datastore/api/auth.py @@ -38,8 +38,10 @@ async def authorize( ) -> dict[str, Any]: """Run policy checks, delegate to the provider, return endpoint data_dict. - Endpoints merge the returned dict into their `data_dict`: - `{"resource": , "package": }` + `{"user": , "resource": , "package": }` + — `user` is the acting username the provider reported; `RequestContext` + notes it for analytics and strips it before endpoints merge the rest + into their `data_dict`. """ if bool(resource_id) == bool(package_id): raise ValidationError("exactly one of resource_id or package_id required") @@ -54,13 +56,17 @@ async def authorize( "Access denied: Action requires an authenticated user" ) - decision = await provider.authorize( + result = await provider.authorize( credential=api_key, resource_id=resource_id, package_id=package_id, permission=permission, ) - return {"resource": decision.resource or {}, "package": decision.package or {}} + return { + "user": result.subject, + "resource": result.resource or {}, + "package": result.package or {}, + } def ensure_resource_writable( diff --git a/datastore/api/context.py b/datastore/api/context.py index 60e55b3..f4b1951 100644 --- a/datastore/api/context.py +++ b/datastore/api/context.py @@ -7,6 +7,7 @@ from fastapi.security import APIKeyHeader from starlette.requests import Request +from datastore.analytics import authorization_dict from datastore.api import auth as auth_fns from datastore.api.auth import Permission from datastore.auth.base import AuthProvider @@ -70,6 +71,7 @@ async def handler(payload: ..., context: Context): api_key: str | None = field(repr=False) auth_provider: AuthProvider ckan: CKANClient | None + request: Request | None = None async def authorize( self, @@ -77,16 +79,21 @@ async def authorize( package_id: str | None = None, permission: Permission | None = None, ) -> dict[str, Any]: - return await auth_fns.authorize( + data_dict = await auth_fns.authorize( api_key=self.api_key, provider=self.auth_provider, resource_id=resource_id, package_id=package_id, permission=permission, ) + if self.request is not None and self.config.ANALYTICS_ENABLED: + authorization_dict(self.request, data_dict) + data_dict.pop("user", None) + return data_dict def get_context( + request: Request, config: ConfigDep, ckan: Annotated[CKANClient | None, Depends(get_ckan_client)], provider: Annotated[AuthProvider, Depends(get_auth_provider)], @@ -98,6 +105,7 @@ def get_context( api_key=api_key, auth_provider=provider, ckan=ckan.bind(api_key) if ckan is not None else None, + request=request, ) diff --git a/datastore/auth/ckan/provider.py b/datastore/auth/ckan/provider.py index b052b7e..1c49547 100644 --- a/datastore/auth/ckan/provider.py +++ b/datastore/auth/ckan/provider.py @@ -78,11 +78,9 @@ async def authorize( package_id=package_id, permission=permission, ) - # `subject` rides through the cache (orjson-serialised). Never - # store the raw credential there — use the same hash we already - # derive for the cache key. + decision = Decision( - subject=self.key_id(credential) if credential else None, + subject=result.get("user"), resource=result.get("resource"), package=result.get("package"), ) diff --git a/datastore/core/config.py b/datastore/core/config.py index c201e5d..5f62e34 100644 --- a/datastore/core/config.py +++ b/datastore/core/config.py @@ -56,6 +56,15 @@ class Config(BaseSettings): description="Maximum request body size in MB", ) + # Analytics + ANALYTICS_ENABLED: bool = Field( + default=False, + description=( + "Emit one JSON analytics event per datastore action / dump " + "request. `false` leaves the analytics middleware unmounted." + ), + ) + # CORS CORS_ORIGINS: str = Field( default="*", diff --git a/datastore/main.py b/datastore/main.py index 516679c..415925e 100644 --- a/datastore/main.py +++ b/datastore/main.py @@ -11,6 +11,7 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.gzip import GZipMiddleware +from datastore.analytics import AnalyticsMiddleware from datastore.api.error_handlers import register_exception_handlers from datastore.api.middleware import BodySizeLimitMiddleware from datastore.api.responses import ORJSONResponse @@ -24,8 +25,12 @@ warmup_engines, ) -log = logging.getLogger("uvicorn.error") +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s %(message)s", +) +logger = logging.getLogger(__name__) OPENAPI_TAGS = [ { @@ -113,11 +118,12 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: warmup_engines(config) stack.callback(reset_engine_cache) - log.info( - "datastore ready: Engine=%r Auth=%r Cache=%s", + logger.info( + "datastore ready: Engine=%r Auth=%r Cache=%s Analytics=%s", config.DATASTORE_ENGINE, config.AUTH_TYPE, "redis" if config.REDIS_URL else "memory", + "on" if config.ANALYTICS_ENABLED else "off", ) yield @@ -153,6 +159,10 @@ def create_app() -> FastAPI: BodySizeLimitMiddleware, max_bytes=config.MAX_REQUEST_BODY_MB * 1024 * 1024, ) + # Outside the body-size guard, so a rejected oversize upload is an + # event too; inside CORS, which only decorates headers. + if config.ANALYTICS_ENABLED: + app.add_middleware(AnalyticsMiddleware, service="Datastore") # Added last = outermost, so 4xx/5xx envelopes carry CORS headers too. # `CORS_ORIGINS=*` allows every origin, a comma-separated list allows # only those domains, empty skips the middleware entirely. diff --git a/tests/auth/ckan/test_provider.py b/tests/auth/ckan/test_provider.py index 37e3853..ac64664 100644 --- a/tests/auth/ckan/test_provider.py +++ b/tests/auth/ckan/test_provider.py @@ -27,6 +27,7 @@ def __init__(self, result: dict[str, Any] | None = None) -> None: self._result = result or { "package": {"id": "pkg-1"}, "resource": {"id": "res-1", "package_id": "pkg-1"}, + "user": "jhon", } self.calls: list[dict[str, Any]] = [] self.raise_on_authorize: Exception | None = None @@ -102,9 +103,8 @@ def test_authorize_binds_credential_and_maps_response_to_decision() -> None: "permission": "read", } ] - # `subject` carries a hash of the credential (raw key never leaves - # this provider). Same shape as `key_id`. - assert decision.subject == provider.key_id("token-xyz") + # `subject` is the acting username CKAN resolved from the api key. + assert decision.subject == "jhon" assert decision.resource == {"id": "res-1", "package_id": "pkg-1"} assert decision.package == {"id": "pkg-1"} assert decision.claims is None @@ -234,9 +234,9 @@ def test_malformed_cache_entry_falls_through_to_ckan() -> None: assert len(ckan.calls) == 1 -def test_subject_in_cached_decision_is_hashed_not_raw_credential() -> None: +def test_subject_never_carries_the_raw_credential() -> None: # Security: the raw credential must never end up in the cache. - # `Decision.subject` is what gets serialised — store the hash. + # `Decision.subject` is what gets serialised — store CKAN's username. ckan = FakeCKAN() provider = _provider(ckan=ckan, cache=InMemoryCache()) @@ -245,9 +245,21 @@ def test_subject_in_cached_decision_is_hashed_not_raw_credential() -> None: resource_id="res-1", package_id=None, permission="read", )) - assert decision.subject is not None + assert decision.subject == "jhon" assert "raw-api-key-do-not-leak" not in decision.subject - assert decision.subject.startswith("h:") + + +def test_subject_is_none_when_ckan_names_no_user() -> None: + # An older CKAN whose datastore_authorize predates the `user` field + # (or an anonymous caller) still authorizes — the event just has no user. + ckan = FakeCKAN(result={"package": {"id": "pkg-1"}, "resource": {"id": "res-1"}}) + provider = _provider(ckan=ckan) + + decision = asyncio.run(provider.authorize( + credential="tok", resource_id="res-1", package_id=None, permission="read", + )) + + assert decision.subject is None # --- key derivation + name -------------------------------------------------- diff --git a/tests/auth/test_orchestration.py b/tests/auth/test_orchestration.py index fe90e15..ec6e533 100644 --- a/tests/auth/test_orchestration.py +++ b/tests/auth/test_orchestration.py @@ -50,7 +50,7 @@ def key_id(self, credential: str) -> str: # --- happy path ------------------------------------------------------------- -def test_provider_decision_is_returned_as_endpoint_data_dict_shape() -> None: +def test_provider_verdict_is_returned_as_endpoint_data_dict() -> None: provider = FakeProvider() result = asyncio.run(authorize( api_key="tok", @@ -60,7 +60,11 @@ def test_provider_decision_is_returned_as_endpoint_data_dict_shape() -> None: permission="read", )) - assert result == {"resource": {"id": "res-1"}, "package": {"id": "pkg-1"}} + assert result == { + "user": None, + "resource": {"id": "res-1"}, + "package": {"id": "pkg-1"}, + } assert provider.calls == [ { "credential": "tok", @@ -71,8 +75,8 @@ def test_provider_decision_is_returned_as_endpoint_data_dict_shape() -> None: ] -def test_decision_without_metadata_yields_empty_dicts() -> None: - # Anonymous / JWT providers return Decision() with no resource/package; +def test_verdict_without_metadata_yields_empty_dicts() -> None: + # Anonymous / JWT providers answer with no user/resource/package; # endpoint code reads from the dict so we must substitute empty dicts. result = asyncio.run(authorize( api_key="tok", @@ -81,7 +85,7 @@ def test_decision_without_metadata_yields_empty_dicts() -> None: package_id=None, permission="read", )) - assert result == {"resource": {}, "package": {}} + assert result == {"user": None, "resource": {}, "package": {}} # --- anonymous-read policy -------------------------------------------------- diff --git a/tests/conftest.py b/tests/conftest.py index 6170304..a0afdae 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -26,6 +26,10 @@ # fine because it has no .env). `os.environ[...] =` forces an override, # unlike `setdefault` above which respects a CI-supplied value. os.environ["AUTH_TYPE"] = "ckan" +# Analytics defaults to off; the analytics tests assert emitted events +# through the shared `client` fixture, so force the middleware on for the +# suite. The disable-path tests monkeypatch it back to "false" per test. +os.environ["ANALYTICS_ENABLED"] = "true" from collections.abc import Iterator # noqa: E402 from typing import Any # noqa: E402 @@ -111,6 +115,8 @@ async def datastore_authorize( if self._api_key and self._api_key in self.deny_keys: raise AuthorizationError(f"key '{self._api_key}' is not allowed") + # The real action names the acting user (resolved from the api key). + user = "jhon" if self._api_key else None if resource_id is not None: existing = self.resources.get(resource_id) if existing is None: @@ -119,13 +125,17 @@ async def datastore_authorize( package = self.packages.get(pkg_id) if package is None: raise NotFoundError(f"package '{pkg_id}' not found") - return {"package": package, "resource": existing} + return {"package": package, "resource": existing, "user": user} assert package_id is not None package = self.packages.get(package_id) if package is None: raise NotFoundError(f"package '{package_id}' not found") - return {"package": package, "resource": {"package_id": package_id}} + return { + "package": package, + "resource": {"package_id": package_id}, + "user": user, + } async def resource_create(self, *, resource: dict[str, Any]) -> dict[str, Any]: self._guard() diff --git a/tests/test_analytics.py b/tests/test_analytics.py new file mode 100644 index 0000000..bfe2ac9 --- /dev/null +++ b/tests/test_analytics.py @@ -0,0 +1,312 @@ +"""One structured analytics event per datastore action call or dump. + +The middleware is exercised through the real app - the same routes, error +handlers and auth the service runs - with the emitter captured. The event +shape is shared with ckanext-analytics: both emit JSON log lines meant for +one BigQuery table, told apart by `service`. +""" + +from __future__ import annotations + +import asyncio +import json +from typing import Any +from unittest.mock import patch + +import pytest +from datastore import analytics +from datastore.api.context import RequestContext, get_auth_provider, get_ckan_client +from datastore.auth.base import Decision +from datastore.auth.ckan import Provider as CKANAuthProvider +from datastore.core.config import get_config +from datastore.infrastructure.cache import InMemoryCache +from datastore.infrastructure.engines.bigquery import BigQueryBackend +from datastore.main import create_app +from fastapi.testclient import TestClient + +from tests.conftest import FakeCKAN + +FIELDS = { + "timestamp", + "request_id", + "service", + "method", + "endpoint", + "query_string", + "action_type", + "status_code", + "user_agent", + "request_ip", + "user", + "dataset", + "resource", + "organization", + "group", +} + +SEARCH_URL = "/api/3/action/datastore_search" +RESOURCE = "balancing_auction_results_2025" + + +@pytest.fixture +def recorded(monkeypatch: pytest.MonkeyPatch) -> list[dict]: + events: list[dict] = [] + monkeypatch.setattr(analytics, "emit_event", events.append) + return events + + +# --- what gets recorded ----------------------------------------------------- + + +def test_a_search_is_recorded_with_the_whole_field_set( + client: TestClient, recorded: list[dict] +) -> None: + response = client.get(SEARCH_URL, params={"resource_id": RESOURCE}) + + assert response.status_code == 200 + assert len(recorded) == 1 + event = recorded[0] + assert set(event) == FIELDS + assert event["service"] == "Datastore" + assert event["method"] == "GET" + assert event["endpoint"] == SEARCH_URL + assert event["query_string"] == f"resource_id={RESOURCE}" + assert event["action_type"] == "datastore_search" + assert event["status_code"] == 200 + + +def test_the_resource_is_resolved_through_the_auth_decision( + client: TestClient, recorded: list[dict] +) -> None: + """Authorize already fetched the CKAN names - the event reuses them.""" + client.get(SEARCH_URL, params={"resource_id": RESOURCE}) + + event = recorded[0] + assert event["resource"] == "balancing-auction-results-2025" + assert event["dataset"] == "balancing-2025" + + +def test_the_caller_is_recorded_by_username( + client: TestClient, recorded: list[dict] +) -> None: + """CKAN's datastore_authorize names the acting user; the event keeps it.""" + client.get(SEARCH_URL, params={"resource_id": RESOURCE}) + + assert recorded[0]["user"] == "jhon" + + +def test_a_post_carries_its_resource_in_the_body( + client: TestClient, recorded: list[dict] +) -> None: + """nginx cannot see a POST body; this is why the service records itself.""" + client.post( + "/api/3/action/datastore_upsert", + json={"resource_id": RESOURCE, "force": True, "records": [{"a": 1}]}, + ) + + event = recorded[0] + assert event["action_type"] == "datastore_upsert" + assert event["method"] == "POST" + assert event["resource"] == "balancing-auction-results-2025" + + +def test_a_dump_is_recorded_as_a_download( + client: TestClient, recorded: list[dict] +) -> None: + url = "https://storage.googleapis.com/bucket/dumps/x/abc.csv?Sig=abc" + + async def fake_dump(self: BigQueryBackend, resource_id: str, fmt: str) -> list[str]: + return [url] + + with patch.object(BigQueryBackend, "dump", fake_dump): + response = client.get(f"/datastore/dump/{RESOURCE}", follow_redirects=False) + + assert response.status_code == 302 + event = recorded[0] + assert event["action_type"] == "datastore_dump" + assert event["status_code"] == 302 + assert event["resource"] == "balancing-auction-results-2025" + + +def test_a_sql_dump_is_recorded_under_its_own_name( + client: TestClient, recorded: list[dict] +) -> None: + client.get("/datastore/dump/query", params={"sql": "SELECT 1"}) + + assert recorded[0]["action_type"] == "datastore_dump_query" + + +# --- failures are half the data ---------------------------------------------- + + +def test_a_denied_call_is_recorded_with_the_raw_reference( + client: TestClient, recorded: list[dict], fake_ckan: FakeCKAN +) -> None: + """No names were resolved, so the event keeps what the caller sent.""" + fake_ckan.deny("bad-key") + + response = client.get( + SEARCH_URL, + params={"resource_id": RESOURCE}, + headers={"Authorization": "bad-key"}, + ) + + assert response.status_code == 403 + event = recorded[0] + assert event["status_code"] == 403 + assert event["resource"] == RESOURCE + + +def test_a_missing_resource_is_recorded_with_its_status( + client: TestClient, recorded: list[dict] +) -> None: + client.get(SEARCH_URL, params={"resource_id": "no-such-resource"}) + + assert recorded[0]["status_code"] == 404 + assert recorded[0]["resource"] == "no-such-resource" + + +def test_an_unmounted_action_is_recorded_as_its_status( + client: TestClient, recorded: list[dict] +) -> None: + """This service only mounts datastore actions; attempts still count.""" + client.get("/api/3/action/package_show") + + assert recorded[0]["action_type"] == "package_show" + assert recorded[0]["status_code"] == 404 + + +# --- request metadata --------------------------------------------------------- + + +def test_request_metadata_comes_from_the_ingress_headers( + client: TestClient, recorded: list[dict] +) -> None: + client.get( + SEARCH_URL, + params={"resource_id": RESOURCE}, + headers={ + "X-Request-ID": "abc123-from-nginx", + "X-Real-IP": "203.0.113.7", + "User-Agent": "curl/8.4.0", + }, + ) + + event = recorded[0] + assert event["request_id"] == "abc123-from-nginx" + assert event["request_ip"] == "203.0.113.7" + assert event["user_agent"] == "curl/8.4.0" + + +def test_a_request_id_is_generated_when_nothing_upstream_set_one( + client: TestClient, recorded: list[dict] +) -> None: + client.get(SEARCH_URL, params={"resource_id": RESOURCE}) + + assert recorded[0]["request_id"] + + +def test_the_ip_falls_back_to_the_last_forwarded_for_entry( + client: TestClient, recorded: list[dict] +) -> None: + client.get( + SEARCH_URL, + params={"resource_id": RESOURCE}, + headers={"X-Forwarded-For": "10.0.0.9, 203.0.113.7"}, + ) + + assert recorded[0]["request_ip"] == "203.0.113.7" + + +# --- what does not get recorded, and what cannot break ------------------------ + + +def test_health_and_pages_are_not_recorded( + client: TestClient, recorded: list[dict] +) -> None: + client.get("/") + client.get("/datastore/api/health") + + assert recorded == [] + + +def test_a_broken_emitter_does_not_break_the_request( + client: TestClient, monkeypatch: pytest.MonkeyPatch +) -> None: + def explode(event: dict) -> None: + raise RuntimeError("the stream is down") + + monkeypatch.setattr(analytics, "emit_event", explode) + + response = client.get(SEARCH_URL, params={"resource_id": RESOURCE}) + + assert response.status_code == 200 + + +def test_analytics_can_be_disabled_by_env( + fake_ckan: FakeCKAN, + cache: InMemoryCache, + recorded: list[dict], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """ANALYTICS_ENABLED=false leaves the middleware unmounted — requests + work as ever, no event is emitted.""" + monkeypatch.setenv("ANALYTICS_ENABLED", "false") + get_config.cache_clear() + + app = create_app() + app.dependency_overrides[get_ckan_client] = lambda: fake_ckan + app.dependency_overrides[get_auth_provider] = lambda: CKANAuthProvider( + ckan=fake_ckan, cache=cache, cache_ttl=60, + ) + with TestClient(app) as c: + c.headers["Authorization"] = "test-token" + response = c.get(SEARCH_URL, params={"resource_id": RESOURCE}) + + assert response.status_code == 200 + assert recorded == [] + + +def test_authorize_captures_nothing_when_analytics_disabled( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """With the middleware unmounted nobody reads the attribution - + authorize should not stash it on the request either.""" + monkeypatch.setenv("ANALYTICS_ENABLED", "false") + get_config.cache_clear() + + class StubRequest: + def __init__(self) -> None: + self.scope: dict[str, Any] = {} + + class StubProvider: + name = "stub" + + async def authorize(self, **_: object) -> Decision: + return Decision(subject="jhon", resource={"id": "r"}, package={"id": "p"}) + + def key_id(self, credential: str) -> str: + return "h:stub" + + request = StubRequest() + context = RequestContext( + config=get_config(), + api_key="tok", + auth_provider=StubProvider(), + ckan=None, + request=request, # type: ignore[arg-type] + ) + asyncio.run(context.authorize(resource_id="r", permission="read")) + + assert request.scope.get("state", {}).get("analytics") is None + + +def test_the_emitted_line_is_bare_json(caplog: pytest.LogCaptureFixture) -> None: + """Nothing downstream can parse the event unless the line is only the object.""" + with caplog.at_level("INFO", logger="datastore.analytics"): + analytics.emit_event({"action_type": "datastore_search", "status_code": 200}) + + assert json.loads(caplog.records[0].getMessage()) == { + "action_type": "datastore_search", + "status_code": 200, + }