From dbaf09f0d9ee67cbab4afc92f887df2699b6245e Mon Sep 17 00:00:00 2001 From: Sagar Ghimire Date: Mon, 17 Aug 2026 13:43:09 +0545 Subject: [PATCH 1/2] feat: emit one analytics event per datastore action / dump request Adds datastore/analytics.py: a pure-ASGI AnalyticsMiddleware that records every /api/3/action/* and /datastore/dump/* request as one JSON log line on the datastore.analytics logger - the same event shape ckanext-analytics emits, told apart by the service field. Handled errors keep their status; an unhandled crash is recorded as a 500 before it propagates. Attribution rides on the authorize step: the provider's decision now carries the acting username, and RequestContext stashes user / resource / dataset / organization on the request scope for the middleware to pick up (the CKAN provider already fetched them on the way to its verdict). The resource reference falls back to whatever the caller sent - dump path segment, query string, or capped POST body. Gated by ANALYTICS_ENABLED (default false, middleware unmounted). --- .env.example | 3 + datastore/analytics.py | 218 ++++++++++++++++++++++ datastore/api/auth.py | 14 +- datastore/api/context.py | 10 +- datastore/auth/ckan/provider.py | 6 +- datastore/core/config.py | 9 + datastore/main.py | 16 +- tests/auth/ckan/test_provider.py | 26 ++- tests/auth/test_orchestration.py | 14 +- tests/conftest.py | 14 +- tests/test_analytics.py | 308 +++++++++++++++++++++++++++++++ 11 files changed, 612 insertions(+), 26 deletions(-) create mode 100644 datastore/analytics.py create mode 100644 tests/test_analytics.py 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..dfda61b --- /dev/null +++ b/datastore/analytics.py @@ -0,0 +1,218 @@ +"""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"], + "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..6d4885b --- /dev/null +++ b/tests/test_analytics.py @@ -0,0 +1,308 @@ +"""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", + "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["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, + } From a4fc9bfccced53836b41963af7f128f705b72c7b Mon Sep 17 00:00:00 2001 From: Sagar Ghimire Date: Mon, 17 Aug 2026 14:03:31 +0545 Subject: [PATCH 2/2] feat: include endpoint and query_string in analytics events --- datastore/analytics.py | 2 ++ tests/test_analytics.py | 4 ++++ 2 files changed, 6 insertions(+) diff --git a/datastore/analytics.py b/datastore/analytics.py index dfda61b..24694a1 100644 --- a/datastore/analytics.py +++ b/datastore/analytics.py @@ -154,6 +154,8 @@ def _event( "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, diff --git a/tests/test_analytics.py b/tests/test_analytics.py index 6d4885b..bfe2ac9 100644 --- a/tests/test_analytics.py +++ b/tests/test_analytics.py @@ -31,6 +31,8 @@ "request_id", "service", "method", + "endpoint", + "query_string", "action_type", "status_code", "user_agent", @@ -67,6 +69,8 @@ def test_a_search_is_recorded_with_the_whole_field_set( 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