Fix/security scan findings

CLAUDE.md

@@ -465,7 +465,7 @@ dumps/<rid>/<fmt>/<rev>/          ← cache key, shared by all requests
 4. **Compose** — csv/gzip/ndjson shards are stitched into ONE object server-side with GCS `compose` (≤32 sources per call, chained beyond that). csv and gzip compose a synthesized header member in first — plain bytes for csv, a gzip member for gzip — so the result carries exactly one header. (Verified against the real bucket: BigQuery's gzip objects carry **no** `Content-Encoding`, so compose byte-concats them and multi-member gzip decompresses as one file.) Parquet is never composable (footer + magic bytes) and stays as shards. Compose sources are the explicit shard list from **this attempt only** — never a fresh listing — so nothing foreign can be swept into the output.
 5. **Publish** — write the zero-byte `<prefix><attempt>/_SUCCESS`. This is the commit point: the attempt is unreadable before it, readable after, so no caller ever sees a half-written export. Written *after* the compose, so it can never become file content.
 6. **Cleanup — in the background** (`_cleanup_in_background`; never on the response path): the compose source shards are deleted, and superseded attempts + revisions under `dumps/<rid>/<fmt>/` are swept (`_delete_old_cache` — this request's own attempt is never touched, and the sweep is **always age-gated by the signed-URL expiry** so a sibling attempt that may still be exporting, or whose URLs are live, survives).
-7. **Sign URLs** — V4 with `response-content-disposition: attachment; filename="<rid>.<ext>"` (one file) or `<rid>_NN.<ext>` (multi-file parquet, 1-indexed). Signing is offloaded to a thread (IAM round-trip under workload identity).
+7. **Sign URLs** — V4 with `response-content-disposition: attachment; filename="<rid>.<ext>"` (one file) or `<rid>_NN.<ext>` (multi-file parquet, 1-indexed). Signing is offloaded to a thread (IAM round-trip under workload identity) and runs as the **ro** identity: the signer's SA email is public in the signed URL, so the account named there should be one that can only read.
 8. **Return** (`download_response` in [api/endpoints/dump.py](datastore/api/endpoints/dump.py)):
    - 1 URL → `RedirectResponse(302)`. Bytes flow GCS → client; the server is **out of the byte path** for every format, and downloads are resumable.
    - N URLs (sharded parquet only) → `200` + **a streamed zip** of the parts (`zip_archive_writer` in [services/streaming.py](datastore/services/streaming.py)). The API fetches each signed URL over `app.state.http` and frames it into the archive a chunk at a time, so one export is always one file at one URL. Entries are `ZIP_STORED` (parquet is already compressed; deflating costs CPU for no size win) and `force_zip64=True` (member sizes aren't known up front). This is the **only** path where the server carries the bytes — no `Content-Length`, no range support, and a fetch failure mid-archive truncates a response that already returned 200.
@@ -486,12 +486,12 @@ Required IAM. Dump follows a strict **ro for reading, rw for writing/updating**
 | `compose` (csv/ndjson) | RW GCS | Server-side concat into one object: reads each source's metadata (`storage.objects.get` — **`list` alone is not enough**) and creates the composite. |
 | `upload_from_string` (csv header member) | RW GCS | The one-row header composed in front of the header-less csv shards. |
 | `delete` (GC) | RW GCS | Writing/deleting objects. |
-| `generate_signed_url` | RW GCS | Under workload identity this calls IAM `signBlob`, which typically only the rw SA holds via `iam.serviceAccountTokenCreator`. |
+| `generate_signed_url` | **RO GCS** | A V4 URL carries the signer's SA email in `X-Goog-Credential`, where it is public, and the grant it issues is bounded by what that identity can do. Signing a GET with the rw SA advertises the write identity and issues a grant from an account that could also overwrite the object — so `_signed_urls` re-binds each blob to the ro bucket first. Requires `iam.serviceAccountTokenCreator` **on the ro SA** (IAM `signBlob` under workload identity) plus `storage.objects.get`. |
 
 Concrete perm sets:
 
-- **RO SA** (`BIGQUERY_CREDENTIALS_RO`) — `bigquery.tables.get` + `storage.objects.list`.
-- **RW SA** (`BIGQUERY_CREDENTIALS`) — `bigquery.jobs.create` + `bigquery.tables.export` + `bigquery.tables.getData` + `storage.objects.{create,get,list,delete}` + `iam.serviceAccountTokenCreator`. **`get` is required by `compose`** (it reads each source object); a role with only create/list/delete 403s on the compose step.
+- **RO SA** (`BIGQUERY_CREDENTIALS_RO`) — `bigquery.tables.get` + `storage.objects.{list,get}` + `iam.serviceAccountTokenCreator` (on itself, for signing).
+- **RW SA** (`BIGQUERY_CREDENTIALS`) — `bigquery.jobs.create` + `bigquery.tables.export` + `bigquery.tables.getData` + `storage.objects.{create,get,list,delete}`. **`get` is required by `compose`** (it reads each source object); a role with only create/list/delete 403s on the compose step.
 
 A single SA works if both perm sets land on the same identity — `BIGQUERY_CREDENTIALS_RO` empty falls through to ADC; same env var can drive both. `_build_bq_client` and `_build_storage_client` on the backend are deliberately small + stub-friendly so tests inject mocks without monkey-patching `google.cloud.*` globally.
 

datastore/api/endpoints/dump.py

@@ -21,6 +21,7 @@
 from datastore.core.exceptions import ServerError
 from datastore.infrastructure.engines import get_datastore_engine
 from datastore.schemas.request import DatastoreDumpSQLRequest
+from datastore.schemas.validators import ResourceId
 from datastore.services.read import dump_sql_datastore
 from datastore.services.streaming import zip_archive_writer
 
@@ -134,7 +135,7 @@ async def dump_sql(
 async def dump(
     request: Request,
     context: Context,
-    resource_id: str,
+    resource_id: ResourceId,
     fmt: Annotated[DumpFormat, Query(alias="format")] = "csv",
 ):
     """Download an entire resource; pick the format with `?format=`."""

datastore/api/error_handlers.py

@@ -6,6 +6,7 @@
 
 from fastapi import FastAPI, HTTPException
 from fastapi.exceptions import RequestValidationError
+from starlette.exceptions import HTTPException as StarletteHTTPException
 from starlette.requests import Request
 
 from datastore.api.responses import ORJSONResponse, _error_response
@@ -42,13 +43,21 @@ def _group_errors(errors: list[dict[str, Any]]) -> dict[str, list[str]]:
 def register_exception_handlers(app: FastAPI) -> None:
     @app.exception_handler(APIError)
     async def _api_error(request: Request, exc: APIError) -> ORJSONResponse:
-        log.debug(
-            "APIError: %s -> %d (%s) at %s %s",
+        # 5xx means we are broken: log at ERROR with the operator-only
+        # `detail` and the original traceback. 4xx is the caller's doing
+        # and stays at DEBUG. `exc.message` is the only text that is ever
+        # serialised into the envelope.
+        is_server_fault = exc.status_code >= 500
+        log.log(
+            logging.ERROR if is_server_fault else logging.DEBUG,
+            "APIError: %s -> %d (%s) at %s %s: %s",
             type(exc).__name__,
             exc.status_code,
             exc.type_label,
             request.method,
             request.url.path,
+            exc.detail or exc.message,
+            exc_info=exc.__cause__ if is_server_fault else None,
         )
         return _error_response(
             request,
@@ -79,6 +88,12 @@ async def _validation(request: Request, exc: RequestValidationError) -> ORJSONRe
             fields=fields,
         )
 
+    # Starlette's router raises *its* HTTPException for an unmatched route
+    # (404) or a wrong method (405). `fastapi.HTTPException` subclasses it,
+    # so registering the subclass alone left those two falling through to
+    # the default `{"detail": ...}` body - not the envelope every other
+    # error uses. Registering the base class covers both.
+    @app.exception_handler(StarletteHTTPException)
     @app.exception_handler(HTTPException)
     async def _http(request: Request, exc: HTTPException) -> ORJSONResponse:
         label = HTTP_STATUS_TO_TYPE_LABEL.get(exc.status_code, "Internal Error")

datastore/core/exceptions.py

@@ -12,10 +12,17 @@ def __init__(
         message: str,
         *,
         fields: dict[str, list[str]] | None = None,
+        detail: str | None = None,
     ) -> None:
-        super().__init__(message)
+        # `str(exc)` carries the detail so tracebacks stay useful; the
+        # envelope is built from `.message` alone.
+        super().__init__(f"{message}: {detail}" if detail else message)
+        #: Sent to the caller. Keep it generic on 5xx - upstream text
+        #: names the engine, the cloud region and internal job ids.
         self.message = message
         self.fields = fields
+        #: Operator-only context. Logged, never serialised.
+        self.detail = detail
 
 
 class ValidationError(APIError):

datastore/infrastructure/ckan_client.py

@@ -27,6 +27,15 @@
 }
 
 
+def _ckan_transport_error(exc: Exception) -> ServerError:
+    """Wrap a CKAN transport failure.
+
+    The httpx message embeds the upstream URL, which is an internal
+    cluster address; keep it in `detail` for the log.
+    """
+    return ServerError("upstream request failed", detail=f"CKAN request failed: {exc}")
+
+
 class CKANClient:
     """Thin async wrapper around the CKAN action API.
 
@@ -88,7 +97,7 @@ async def _post_action(self, action: str, body: dict[str, Any]) -> dict[str, Any
         try:
             response = await self._http.post(url, json=body, headers=headers)
         except httpx.HTTPError as exc:
-            raise ServerError(f"CKAN request failed: {exc}") from exc
+            raise _ckan_transport_error(exc) from exc
 
         self._raise_for_status(action, response)
         return self._unwrap(action, response)

datastore/infrastructure/engines/bigquery/backend.py

@@ -30,6 +30,7 @@
 
 from datastore.core.config import Config
 from datastore.core.exceptions import (
+    APIError,
     NotFoundError,
     ServerError,
     ValidationError,
@@ -66,6 +67,45 @@
 
 log = logging.getLogger(__name__)
 
+
+# BigQuery reasons that mean "the caller sent something bad" rather than
+# "we are broken": an unknown column, a filter value that will not parse
+# as the column's type. Those are 400s.
+_CLIENT_ERROR_REASONS = frozenset({"invalidQuery", "invalid", "notFound"})
+
+_CLIENT_ERROR_MESSAGE = (
+    "the query could not be executed; check column names, filter values "
+    "and types against the resource schema"
+)
+
+
+def _bq_error_reason(e: Exception) -> str:
+    """Pull BigQuery's machine-readable `reason`.
+
+    `google.api_core` exposes it inside the `errors` list, *not* as an
+    attribute - `BadRequest.reason` is `None`. Reading the attribute
+    alone silently classifies every error as a server fault.
+    """
+    for err in getattr(e, "errors", None) or ():
+        if isinstance(err, dict) and err.get("reason"):
+            return str(err["reason"])
+    return ""
+
+
+def _classify_bq_error(e: Exception, op: str, resource_id: str | None = None) -> APIError:
+    """Map a BigQuery exception to the right side of the 4xx/5xx line.
+
+    Caller mistakes become `ValidationError`; everything else stays
+    `ServerError`. Either way the upstream text - engine name, region,
+    job id - rides in `detail`, which is logged and never serialised.
+    """
+    where = f" for resource {resource_id!r}" if resource_id else ""
+    detail = f"BigQuery {op} failed{where}: {e}"
+    if _bq_error_reason(e) in _CLIENT_ERROR_REASONS:
+        return ValidationError(_CLIENT_ERROR_MESSAGE, detail=detail)
+    return ServerError("internal error", detail=detail)
+
+
 _HEALTHCHECK_TTL_SECONDS = 15.0
 _HEALTHCHECK_TIMEOUT_SECONDS = 5.0
 
@@ -134,7 +174,8 @@ def _read_schema(self, resource_id: str) -> dict | None:
             return None
         except Exception as e:
             raise ServerError(
-                f"BigQuery tables.get failed for resource {resource_id!r}: {e}"
+                "internal error",
+                detail=f"BigQuery tables.get failed for resource {resource_id!r}: {e}",
             ) from e
         return table_to_schema(table)
 
@@ -194,7 +235,10 @@ def _run_query(
             job.result()
             return job
         except Exception as e:
-            raise ServerError(f"BigQuery {op} failed for resource {resource_id!r}: {e}") from e
+            raise ServerError(
+                "internal error",
+                detail=f"BigQuery {op} failed for resource {resource_id!r}: {e}",
+            ) from e
 
     # ----- create helpers (DDL + records + branch orchestration) --------
     def _create_table_sql(self, resource_id: str, schema: dict) -> str | None:
@@ -611,7 +655,7 @@ def search(
             search_job = self.client.query(sql, job_config=job_config)
             row_iter = search_job.result()
         except Exception as e:
-            raise ServerError(f"BigQuery search failed for resource {resource_id!r}: {e}") from e
+            raise _classify_bq_error(e, "search", resource_id) from e
 
         total: int | None = None
         if include_total:
@@ -622,9 +666,7 @@ def search(
                 try:
                     rows = list(count_job.result())
                 except Exception as e:
-                    raise ServerError(
-                        f"BigQuery search COUNT failed for resource {resource_id!r}: {e}"
-                    ) from e
+                    raise _classify_bq_error(e, "search COUNT", resource_id) from e
                 total = int(rows[0]["n"]) if rows else 0
 
         return SearchResult(
@@ -674,7 +716,10 @@ def search_sql(self, sql: str, limit: int) -> SearchResult:
                 dataset=self.config.BIGQUERY_DATASET,
             )
         except Exception as e:
-            raise ServerError(f"failed to qualify table references in SQL: {e}") from e
+            raise ServerError(
+                "the SQL could not be prepared for execution",
+                detail=f"failed to qualify table references in SQL: {e}",
+            ) from e
 
         count_sql, count_params = self._search_sql_count_query(qualified_sql)
 
@@ -699,7 +744,7 @@ def search_sql(self, sql: str, limit: int) -> SearchResult:
             )
             row_iter = data_job.result()
         except Exception as e:
-            raise ServerError(f"BigQuery search_sql failed: {e}") from e
+            raise _classify_bq_error(e, "search_sql") from e
 
         total: int | None = None
         if count_job is not None: