diff --git a/AGENTS.md b/AGENTS.md index dde615769..29887f07c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -433,6 +433,8 @@ TruffleHog, detect-secrets, large file check (max 1000KB), merge conflict marker - Internal service key for fl-server-to-hub auth (separate from trust keys). - Trust-internal service key for trust-api / imaging-api / fl-client → imaging-api / data-access-api auth (per-trust, never leaves trust env). See **Trust-internal Service Authentication** below. - FL clients intentionally have no Central Hub credentials. +- Cohort-query validation is three-layer and **deliberately asymmetric — do not "sync" the layers**. Only the trust-side `data_access_api.services.cohort.validate_query` is authoritative (single parse-validate-emit; length, single-statement, SELECT-only, no `INSERT`/`UPDATE`/`DELETE`/`MERGE` anywhere in the tree — a writable CTE parses as a top-level `Select`, so the shape check alone misses it — `omop`-schema pin, literal `LIMIT`/`OFFSET`; re-emits from the checked AST; backed by the read-only `data_analyst_reader` role — pass its return value to the engine, never the caller's raw string). The hub-side `flip_api.cohort_services.submit_cohort_query.validate_query` is a *fast-feedback validity pre-check only, not a security control*: it exists so a malformed query fails in-hand instead of after an async fan-out to every trust, and enforces only what every trust would reject anyway. The flip-ui cohort form validates required-field only. A trust must stay safe regardless of what the hub checked, so hub drift is safe by construction. **No layer uses a keyword denylist** — the removed one blocked legitimate `SUBSTRING()` while stopping nothing; blind extraction is defeated by the literal-`LIMIT` rule and DDL/DML by the read-only role. See [`trust/data-access-api/README.md`](trust/data-access-api/README.md#cohort-query-validation). +- Row-level cohort egress is gated on `COHORT_QUERY_THRESHOLD` at **both** row-level routes — `/cohort/dataframe` (FL training data) and `/cohort/accession-ids` (the accession list that decides whose imaging is pulled into XNAT) — sharing one fixed refusal string so a below-threshold cohort is indistinguishable from an empty one. The threshold is the trust's own disclosure floor (default 10, set per trust in its kit file), enforced trust-side rather than relying on the hub's staging guard. Both gates evaluate the **live** cohort on every call: FLIP stores the cohort only as a SQL string and re-runs it against OMOP at every stage, so a project can import cleanly and later start refusing (FLIP#857). - Do not hardcode env values in Dockerfiles or compose files. - 72-hour supply-chain cooldown on Python/npm package installs — enforced by uv `exclude-newer` (`[tool.uv]` in every `pyproject.toml`) and npm `min-release-age` (`flip-ui/.npmrc`, requires npm >= 11.10 which Node 24 LTS ships), backstopped by a `uv lock --check` CI gate in `secret-scanning.yml`. See CONTRIBUTING.md ("Dependency cooldown"). diff --git a/CLAUDE.md b/CLAUDE.md index c0e3f16f4..79a03c9b2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -433,6 +433,8 @@ TruffleHog, detect-secrets, large file check (max 1000KB), merge conflict marker - Internal service key for fl-server-to-hub auth (separate from trust keys). - Trust-internal service key for trust-api / imaging-api / fl-client → imaging-api / data-access-api auth (per-trust, never leaves trust env). See **Trust-internal Service Authentication** below. - FL clients intentionally have no Central Hub credentials. +- Cohort-query validation is three-layer and **deliberately asymmetric — do not "sync" the layers**. Only the trust-side `data_access_api.services.cohort.validate_query` is authoritative (single parse-validate-emit; length, single-statement, SELECT-only, no `INSERT`/`UPDATE`/`DELETE`/`MERGE` anywhere in the tree — a writable CTE parses as a top-level `Select`, so the shape check alone misses it — `omop`-schema pin, literal `LIMIT`/`OFFSET`; re-emits from the checked AST; backed by the read-only `data_analyst_reader` role — pass its return value to the engine, never the caller's raw string). The hub-side `flip_api.cohort_services.submit_cohort_query.validate_query` is a *fast-feedback validity pre-check only, not a security control*: it exists so a malformed query fails in-hand instead of after an async fan-out to every trust, and enforces only what every trust would reject anyway. The flip-ui cohort form validates required-field only. A trust must stay safe regardless of what the hub checked, so hub drift is safe by construction. **No layer uses a keyword denylist** — the removed one blocked legitimate `SUBSTRING()` while stopping nothing; blind extraction is defeated by the literal-`LIMIT` rule and DDL/DML by the read-only role. See [`trust/data-access-api/README.md`](trust/data-access-api/README.md#cohort-query-validation). +- Row-level cohort egress is gated on `COHORT_QUERY_THRESHOLD` at **both** row-level routes — `/cohort/dataframe` (FL training data) and `/cohort/accession-ids` (the accession list that decides whose imaging is pulled into XNAT) — sharing one fixed refusal string so a below-threshold cohort is indistinguishable from an empty one. The threshold is the trust's own disclosure floor (default 10, set per trust in its kit file), enforced trust-side rather than relying on the hub's staging guard. Both gates evaluate the **live** cohort on every call: FLIP stores the cohort only as a SQL string and re-runs it against OMOP at every stage, so a project can import cleanly and later start refusing (FLIP#857). - Do not hardcode env values in Dockerfiles or compose files. - 72-hour supply-chain cooldown on Python/npm package installs — enforced by uv `exclude-newer` (`[tool.uv]` in every `pyproject.toml`) and npm `min-release-age` (`flip-ui/.npmrc`, requires npm >= 11.10 which Node 24 LTS ships), backstopped by a `uv lock --check` CI gate in `secret-scanning.yml`. See CONTRIBUTING.md ("Dependency cooldown"). diff --git a/flip-api/pyproject.toml b/flip-api/pyproject.toml index 3561e55e0..8eebe9b45 100644 --- a/flip-api/pyproject.toml +++ b/flip-api/pyproject.toml @@ -36,7 +36,7 @@ dependencies = [ "cryptography>=46.0.7", "python-dotenv>=1.2.2", "psycopg2-binary>=2.9.10", - "sqlparse>=0.5.4", + "sqlglot>=26.33.0", "pydantic-settings>=2.9.1", "apscheduler>=3.11.0", "slowapi>=0.1.9", diff --git a/flip-api/src/flip_api/cohort_services/submit_cohort_query.py b/flip-api/src/flip_api/cohort_services/submit_cohort_query.py index 300a6fc08..dce129da9 100644 --- a/flip-api/src/flip_api/cohort_services/submit_cohort_query.py +++ b/flip-api/src/flip_api/cohort_services/submit_cohort_query.py @@ -11,13 +11,12 @@ # import json -import re from uuid import UUID -# SQL parser library - would need Python equivalent -# For this example using sqlparse, but may need a more robust solution -import sqlparse # type: ignore[import] +import sqlglot +import sqlglot.expressions from fastapi import APIRouter, Body, Depends, HTTPException, Request +from sqlglot.errors import SqlglotError from sqlmodel import Session, select from flip_api.auth.access_manager import can_modify_project @@ -37,58 +36,83 @@ router = APIRouter(prefix="/cohort", tags=["cohort_services"]) -FORBIDDEN_COMMANDS = [ - "alter user", - "alter table", - "alter database", - "drop table", - "drop user", - "drop role", - "drop database", - "create table", - "substring", -] +# Reject pathologically large queries before the parser allocates an AST. Kept +# equal to the trust-side MAX_QUERY_LENGTH so the hub does not accept a query +# that every trust would then refuse. +MAX_QUERY_LENGTH = 10_240 # 10 KiB -# Match any occurrence of any forbidden commands -REGEX = re.compile(f"({'|'.join(FORBIDDEN_COMMANDS)})", re.IGNORECASE) - - -def contains_forbidden_commands(query: str) -> bool: - """ - Check if the query contains any forbidden commands - - Args: - query: SQL query string - - Returns: - bool: True if the query contains forbidden commands, False otherwise - """ - return bool(REGEX.search(query)) +# Top-level statement shapes that count as SELECT-like for the cohort API. +_SELECT_LIKE_STATEMENTS: tuple[type[sqlglot.expressions.Expression], ...] = ( + sqlglot.expressions.Select, + sqlglot.expressions.Union, + sqlglot.expressions.Intersect, + sqlglot.expressions.Except, +) def validate_query(query: str) -> None: """ - Validate the SQL query syntax + Fast-feedback validity pre-check for a cohort query. **Not a security control.** + + Why this exists at all + ---------------------- + The trust-side ``data-access-api`` is the *authority* on what a cohort query + may do (see ``data_access_api.services.cohort.validate_query``), and it is + deliberately self-sufficient: a trust holds patient data and must stay safe + regardless of what the hub did or did not check, because the hub is a + separate administrative domain that the trust does not trust. Anything this + function enforced for safety would therefore have to be enforced trust-side + anyway, so duplicating the trust's rules here would buy no security — only + two copies of one policy to keep in sync. + + What it buys instead is **fast feedback**. Submitting a cohort query fans it + out to every registered trust as an encrypted task, each of which runs its + own validation asynchronously and reports back. Without a pre-check, a + researcher who typos their SQL waits for that whole round-trip, across N + trusts, to be told. Catching "this cannot possibly succeed anywhere" while + the request is still in-hand turns a multi-minute fan-out into a 400. + + The asymmetry is intentional and safe by construction + ----------------------------------------------------- + This check is deliberately *weaker* than the trust's, and only ever rejects + queries that every trust would reject too. It intentionally does **not** + enforce trust-local policy — the ``omop`` schema pin, literal-``LIMIT`` + rule, read-only role, or minimum-cohort threshold — because those depend on + facts the hub has no authority over and which may legitimately differ per + trust. + + That makes drift between the two harmless in the direction that matters: a + hub lagging behind the trust merely wastes a fan-out on a query the trust + then refuses, never a bypass. Do not "fix" the asymmetry by copying the + trust's rules in here. Args: - query: SQL query string + query (str): The SQL query string submitted by the researcher. Raises: - ValueError: If the query is not valid SQL + ValueError: If the query is over length, unparseable, not exactly one + statement, or not SELECT-shaped. """ + if len(query) > MAX_QUERY_LENGTH: + raise ValueError(f"Query exceeds maximum length of {MAX_QUERY_LENGTH} characters.") + try: - # Use sqlparse to verify SQL validity - # This is a simplified version - you might need a more robust parser - # TODO: Replace with a more robust SQL parser if needed. - # TODO: Create tests to veryfy if sql injection can be done - parsed = sqlparse.parse(query) - if not parsed: - raise ValueError("Empty or invalid SQL") - - logger.info("Query is valid SQL") - except Exception as e: - logger.error({"message": "Query is not valid SQL.", "error": str(e)}) - raise ValueError("Invalid SQL Query") + statements = sqlglot.parse(query, read="postgres") + except SqlglotError as e: + # SqlglotError covers both ParseError and TokenError (e.g. an + # unterminated string literal), so a tokenizer failure cannot bubble up + # as an unhandled 500. + logger.info({"message": "Cohort query rejected: could not be parsed as SQL.", "error": str(e)}) + raise ValueError("Query could not be parsed as SQL.") from e + + # sqlglot yields None for empty input and for stray semicolons (``SELECT 1; ;`` + # parses to ``[Select, None]``), so filtering None out would let a caller + # smuggle an extra statement past the count check. + if len(statements) != 1 or statements[0] is None: + raise ValueError("Exactly one SQL statement is allowed per request.") + + if not isinstance(statements[0], _SELECT_LIKE_STATEMENTS): + raise ValueError("Only SELECT statements are allowed.") # TODO [#114] This endpoint was not defined in the old repo. The old repo defined a step function that ran the @@ -123,11 +147,9 @@ def submit_cohort_query( detail=f"User with ID: {user_id} is not allowed to modify this project", ) - # Additional validation - if contains_forbidden_commands(cohort_query.query): - raise HTTPException(status_code=400, detail="Invalid query: Contains forbidden SQL commands") - - # Validate SQL syntax + # Fast-feedback validity pre-check only — the trust is the authority on + # query safety. See validate_query's docstring for why this is + # deliberately weaker than the trust-side check. try: validate_query(cohort_query.query) except ValueError as e: diff --git a/flip-api/tests/unit/cohort_services/test_submit_cohort_query.py b/flip-api/tests/unit/cohort_services/test_submit_cohort_query.py index 9e97f803e..7153203d3 100644 --- a/flip-api/tests/unit/cohort_services/test_submit_cohort_query.py +++ b/flip-api/tests/unit/cohort_services/test_submit_cohort_query.py @@ -16,7 +16,11 @@ import pytest from fastapi import HTTPException, Request -from flip_api.cohort_services.submit_cohort_query import submit_cohort_query +from flip_api.cohort_services.submit_cohort_query import ( + MAX_QUERY_LENGTH, + submit_cohort_query, + validate_query, +) from flip_api.db.models.main_models import TrustTask from flip_api.domain.schemas.cohort import SubmitCohortQuery from flip_api.domain.schemas.status import TaskType @@ -155,22 +159,80 @@ def test_submit_cohort_query_warns_when_query_row_missing( assert "queried_trust_ids was not persisted" in caplog.text -@patch("flip_api.cohort_services.submit_cohort_query.can_modify_project", return_value=True) -def test_submit_cohort_query_forbidden_sql(mock_can_modify, mock_auth_request): - """Queries with forbidden SQL commands should be rejected.""" - query = SubmitCohortQuery( - name="Hack", - query="DROP TABLE patients;", +def _query(sql: str) -> SubmitCohortQuery: + """Build a SubmitCohortQuery carrying ``sql``, for the pre-check tests below.""" + return SubmitCohortQuery( + name="Test Query", + query=sql, project_id=project_id, query_id=query_id, - authenticationToken=mock_auth_request.headers.get("Authorization", ""), + authenticationToken="Bearer test-token", ) + +@patch("flip_api.cohort_services.submit_cohort_query.can_modify_project", return_value=True) +def test_submit_cohort_query_rejects_non_select(mock_can_modify, mock_auth_request): + """Statements that are not SELECT-shaped are rejected by the hub pre-check.""" + with pytest.raises(HTTPException) as exc_info: + submit_cohort_query(mock_auth_request, _query("DROP TABLE patients;"), MagicMock(), user_id) + + assert exc_info.value.status_code == 400 + assert "SELECT" in str(exc_info.value.detail) + + +@patch("flip_api.cohort_services.submit_cohort_query.can_modify_project", return_value=True) +def test_submit_cohort_query_rejects_unparseable_sql(mock_can_modify, mock_auth_request): + """Input that is not SQL at all is rejected. + + The previous sqlparse-based check accepted this: ``sqlparse.parse`` is a + non-validating tokenizer and returns a truthy result for arbitrary text. + """ + with pytest.raises(HTTPException) as exc_info: + submit_cohort_query(mock_auth_request, _query("$$$$ not sql at all !!!"), MagicMock(), user_id) + + assert exc_info.value.status_code == 400 + + +@patch("flip_api.cohort_services.submit_cohort_query.can_modify_project", return_value=True) +def test_submit_cohort_query_rejects_stacked_statements(mock_can_modify, mock_auth_request): + """Query stacking is rejected — only one statement may be submitted.""" + with pytest.raises(HTTPException) as exc_info: + submit_cohort_query( + mock_auth_request, _query("SELECT 1; DROP TABLE patients"), MagicMock(), user_id + ) + + assert exc_info.value.status_code == 400 + + +@patch("flip_api.cohort_services.submit_cohort_query.can_modify_project", return_value=True) +def test_submit_cohort_query_rejects_oversized_query(mock_can_modify, mock_auth_request): + """Pathologically large queries are rejected before the parser allocates an AST.""" + oversized = f"SELECT {'a' * (MAX_QUERY_LENGTH + 1)} FROM omop.person" + with pytest.raises(HTTPException) as exc_info: - submit_cohort_query(mock_auth_request, query, MagicMock(), user_id) + submit_cohort_query(mock_auth_request, _query(oversized), MagicMock(), user_id) assert exc_info.value.status_code == 400 - assert "forbidden SQL commands" in str(exc_info.value.detail) + assert "length" in str(exc_info.value.detail).lower() + + +def test_validate_query_accepts_substring_function(): + """``SUBSTRING()`` is legitimate SQL and must not be rejected. + + The removed denylist contained the bare token "substring", so every query + using the standard function was refused — the flexibility cost of a keyword + denylist. Blind data extraction via ``SUBSTRING`` is defeated trust-side by + the literal-LIMIT rule, not by banning the function name. + """ + validate_query("SELECT SUBSTRING(gender_source_value, 1, 1) FROM omop.person") + + +def test_validate_query_accepts_cte_with_set_operation(): + """Real cohort queries use CTEs and set operations; both are SELECT-shaped.""" + validate_query( + "WITH a AS (SELECT person_id FROM omop.person) " + "SELECT person_id FROM a UNION SELECT person_id FROM omop.observation" + ) @patch("flip_api.cohort_services.submit_cohort_query.can_modify_project", return_value=True) diff --git a/flip-api/uv.lock b/flip-api/uv.lock index 92e791518..994f448a2 100644 --- a/flip-api/uv.lock +++ b/flip-api/uv.lock @@ -9,7 +9,7 @@ resolution-markers = [ ] [options] -exclude-newer = "2026-07-26T16:49:01.419455055Z" +exclude-newer = "2026-07-31T12:01:34.75169372Z" exclude-newer-span = "P3D" [[package]] @@ -818,8 +818,8 @@ dependencies = [ { name = "python-multipart" }, { name = "slowapi" }, { name = "sqlalchemy" }, + { name = "sqlglot" }, { name = "sqlmodel" }, - { name = "sqlparse" }, { name = "starlette" }, { name = "tornado" }, { name = "urllib3" }, @@ -867,8 +867,8 @@ requires-dist = [ { name = "python-multipart", specifier = ">=0.0.26" }, { name = "slowapi", specifier = ">=0.1.9" }, { name = "sqlalchemy", specifier = ">=2.0.27" }, + { name = "sqlglot", specifier = ">=26.33.0" }, { name = "sqlmodel", specifier = ">=0.0.24" }, - { name = "sqlparse", specifier = ">=0.5.4" }, { name = "starlette", specifier = ">=0.49.1" }, { name = "tornado", specifier = ">=6.5.2" }, { name = "urllib3", specifier = ">=2.7.0" }, @@ -2383,6 +2383,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d0/10/f7220e9b784d295d241c86ed99aeb537f92afcd469a64861f2717e9bb077/sqlalchemy-2.0.50-py3-none-any.whl", hash = "sha256:92064363517a3ff8212b5a93b8c62876579d8dfd1ca5b561335f30152d884fa9", size = 1943861, upload-time = "2026-05-24T19:59:01.119Z" }, ] +[[package]] +name = "sqlglot" +version = "30.14.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/3a/cd/39a94f0f98076ee8e7c7c38fd4bba8d7845b0c629ff967057c64ef2c0989/sqlglot-30.14.0.tar.gz", hash = "sha256:df2ef5d2b8ca814313781f4ff35bf63e58f821ef517eeddbd523c19a61fa9bb9", size = 5944410, upload-time = "2026-07-27T11:23:30.698Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/33/ec/a729883ceda22dcd9117ce182f64d884bf494e72c4dfce00c2ad0a5978e1/sqlglot-30.14.0-py3-none-any.whl", hash = "sha256:fc768e24889d63a5e1237dea7ad305e5ffb4356a98b0bed828f89591ebcd3636", size = 719007, upload-time = "2026-07-27T11:23:28.637Z" }, +] + [[package]] name = "sqlmodel" version = "0.0.38" @@ -2397,15 +2406,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/72/c7/10c60af0607ab6fa136264f7f39d205932218516226d38585324ffda705d/sqlmodel-0.0.38-py3-none-any.whl", hash = "sha256:84e3fa990a77395461ded72a6c73173438ce8449d5c1c4d97fbff1b1df692649", size = 27294, upload-time = "2026-04-02T21:03:56.406Z" }, ] -[[package]] -name = "sqlparse" -version = "0.5.5" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/90/76/437d71068094df0726366574cf3432a4ed754217b436eb7429415cf2d480/sqlparse-0.5.5.tar.gz", hash = "sha256:e20d4a9b0b8585fdf63b10d30066c7c94c5d7a7ec47c889a2d83a3caa93ff28e", size = 120815, upload-time = "2025-12-19T07:17:45.073Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/49/4b/359f28a903c13438ef59ebeee215fb25da53066db67b305c125f1c6d2a25/sqlparse-0.5.5-py3-none-any.whl", hash = "sha256:12a08b3bf3eec877c519589833aed092e2444e68240a3577e8e26148acc7b1ba", size = 46138, upload-time = "2025-12-19T07:17:46.573Z" }, -] - [[package]] name = "stack-data" version = "0.6.3" diff --git a/flip-ui/src/partials/cohort-query/CohortQuery.vue b/flip-ui/src/partials/cohort-query/CohortQuery.vue index 0c0c2d6c5..8e43388a5 100644 --- a/flip-ui/src/partials/cohort-query/CohortQuery.vue +++ b/flip-ui/src/partials/cohort-query/CohortQuery.vue @@ -79,7 +79,6 @@ import router from "@/router"; import { ICohortQueryCreate, sendQuery } from "@/services/cohort-query-service"; import { IProject } from "@/services/project-service"; import { useProjectStore } from "@/store/project"; -import { containsForbiddenCommands } from "@/utils/cohort/query"; import { Snackbar } from "@/utils/snackbar"; import CohortAggregateCard from "./CohortAggregateCard.vue"; @@ -104,20 +103,15 @@ onBeforeMount(() => { watch(projectStore, () => project.value = projectStore.project); +// Required-field validation only. SQL validity is checked by the hub, which +// answers synchronously before fanning the query out to any trust, and the +// trust's data-access-api is the authority on what may run. A client-side +// keyword denylist here blocked legitimate SQL (e.g. SUBSTRING()) without +// adding security — see trust/data-access-api/README.md#cohort-query-validation. const schema = object().shape({ query: string() .trim() .required("A query is required and can't be left blank") - .test( - "valid-query", - "Please enter a valid query", - function() { - try { - return !containsForbiddenCommands(this.parent.query); - } catch { - return false; - } - }) }); const runCohortQuery = async (v: unknown) => { diff --git a/flip-ui/src/utils/cohort/__tests__/query.spec.ts b/flip-ui/src/utils/cohort/__tests__/query.spec.ts index 8bc5bb7af..72150014f 100644 --- a/flip-ui/src/utils/cohort/__tests__/query.spec.ts +++ b/flip-ui/src/utils/cohort/__tests__/query.spec.ts @@ -14,69 +14,9 @@ -import { containsForbiddenCommands, filterByQueriedTrustIds } from "@/utils/cohort/query"; +import { filterByQueriedTrustIds } from "@/utils/cohort/query"; describe("Query", () => { - describe("containsForbiddenCommands", () => { - it("returns true for each forbidden command", () => { - const forbiddenCommands = [ - "alter user", - "alter table", - "alter database", - "drop table", - "drop user", - "drop role", - "drop database", - "create table", - "substring" - ]; - - forbiddenCommands.forEach((command) => { - const result = containsForbiddenCommands(command); - - expect(result).toBeTruthy(); - }); - }); - - it("returns true for each forbidden command ignoring case", () => { - const forbiddenCommands = [ - "ALTER USER", - "ALTER table", - "alter DATABASE", - "DrOp TaBlE", - "DROP USER", - "drop ROLE", - "DROP database", - "CREATE table", - "SUBSTRING" - ]; - - forbiddenCommands.forEach((command) => { - const result = containsForbiddenCommands(command); - - expect(result).toBeTruthy(); - }); - }); - - it("returns false for valid command", () => { - const command = "SELECT * FROM AValidTable;"; - - const result = containsForbiddenCommands(command); - - expect(result).toBeFalsy(); - }); - - it("returns the same response if the command is ran twice", () => { - const command = "ALTER USER serviceaccount WITH PASSWORD 'new_password';"; - - const result1 = containsForbiddenCommands(command); - const result2 = containsForbiddenCommands(command); - - expect(result1).toBeTruthy(); - expect(result2).toBeTruthy(); - }); - }); - describe("filterByQueriedTrustIds", () => { const trusts = [{ id: "a" }, { id: "b" }, { id: "c" }]; diff --git a/flip-ui/src/utils/cohort/query.ts b/flip-ui/src/utils/cohort/query.ts index 23a10250c..99c21a486 100644 --- a/flip-ui/src/utils/cohort/query.ts +++ b/flip-ui/src/utils/cohort/query.ts @@ -14,26 +14,20 @@ -const forbiddenCommands = [ - "alter user", - "alter table", - "alter database", - "drop table", - "drop user", - "drop role", - "drop database", - "create table", - "substring" -]; - -/** - * Match any occurrence of any forbidden commands. +/* + * There is deliberately no client-side SQL denylist here. + * + * This module used to export `containsForbiddenCommands`, a keyword regex that + * duplicated one the hub also carried. It blocked every legitimate use of the + * standard `SUBSTRING()` function while stopping nothing an attacker would do, + * and being a third copy of one policy it was guaranteed to drift. + * + * The layering is now: this form validates only that a query was entered; the + * hub runs a parser-based validity pre-check and answers synchronously before + * fanning the query out, so malformed SQL still fails fast; and the trust's + * data-access-api is the authority that decides what may actually run. See + * `trust/data-access-api/README.md#cohort-query-validation`. */ -const regex = new RegExp(`(${forbiddenCommands.join("|")})`, "i"); - -export const containsForbiddenCommands = (query: string): boolean => { - return regex.test(query); -}; /** * Restrict a list of trusts to those that participated in a cohort query. diff --git a/trust/.env.GSTT.development.example b/trust/.env.GSTT.development.example index f68be3cec..533417004 100644 --- a/trust/.env.GSTT.development.example +++ b/trust/.env.GSTT.development.example @@ -43,6 +43,12 @@ OMOP_DB_TAG=latest DATA_ACCESS_POSTGRES_USER=data_analyst_reader DATA_ACCESS_POSTGRES_PASSWORD=change_me_analyst_secure_password +# Minimum cohort size this trust will release anything about — statistics are suppressed +# below it and both row-level routes refuse. The trust's own disclosure floor; raise to +# release less. Must be a positive integer — 0 would disable all three checks, so it is +# rejected and the service refuses to start. +COHORT_QUERY_THRESHOLD=10 + XNAT_ADMIN_USER=admin XNAT_ADMIN_INITIAL_PASSWORD=admin XNAT_ADMIN_PASSWORD=admin diff --git a/trust/.env.KCH.development.example b/trust/.env.KCH.development.example index 46604132d..023e73c1c 100644 --- a/trust/.env.KCH.development.example +++ b/trust/.env.KCH.development.example @@ -43,6 +43,12 @@ OMOP_DB_TAG=latest DATA_ACCESS_POSTGRES_USER=data_analyst_reader DATA_ACCESS_POSTGRES_PASSWORD=change_me_analyst_secure_password +# Minimum cohort size this trust will release anything about — statistics are suppressed +# below it and both row-level routes refuse. The trust's own disclosure floor; raise to +# release less. Must be a positive integer — 0 would disable all three checks, so it is +# rejected and the service refuses to start. +COHORT_QUERY_THRESHOLD=10 + XNAT_ADMIN_USER=admin XNAT_ADMIN_INITIAL_PASSWORD=admin XNAT_ADMIN_PASSWORD=admin diff --git a/trust/.env.example b/trust/.env.example index 2f4848a94..d5108cd15 100644 --- a/trust/.env.example +++ b/trust/.env.example @@ -66,6 +66,13 @@ OMOP_DB_TAG=latest DATA_ACCESS_POSTGRES_USER=data_analyst_reader DATA_ACCESS_POSTGRES_PASSWORD=change_me_analyst_secure_password +# Minimum cohort size this trust will release anything about. Cohort statistics below it +# are privacy-suppressed, and both row-level routes (/cohort/dataframe for FL training data, +# /cohort/accession-ids for the imaging pull) refuse outright. Raise it to release less; +# this is the trust's own disclosure floor, not a hub setting. Must be a positive integer — +# 0 would disable all three checks, so it is rejected and the service refuses to start. +COHORT_QUERY_THRESHOLD=10 + # XNAT admin + service account XNAT_ADMIN_USER=admin XNAT_ADMIN_INITIAL_PASSWORD=admin diff --git a/trust/README.md b/trust/README.md index ac9407f24..ff22648d6 100644 --- a/trust/README.md +++ b/trust/README.md @@ -70,7 +70,15 @@ FL_KIT_SLOT_NUMBER= EXPECTED_TRUST_ID= ``` -plus the trust's identity (`TRUST_NAME` / `TRUST_CODE` / `TRUST_REGION`, read by `register-trust`) and its host-local ports and data directories. The optional `EXPECTED_TRUST_ID` lets trust-api self-check the hub-resolved id at startup. The same schema serves dev trusts (GSTT/KCH against a local hub), on-prem trusts (against a prod hub), and laptop-against-prod testing — operator picks the kit code (`trust/.env..`) and `make -C trust up-trust KIT= PROD=` handles the rest. +plus the trust's identity (`TRUST_NAME` / `TRUST_CODE` / `TRUST_REGION`, read by `register-trust`) and its host-local ports and data directories. The optional `EXPECTED_TRUST_ID` lets trust-api self-check the hub-resolved id at startup. + +The kit also carries the trust's **disclosure floor**: + +```sh +COHORT_QUERY_THRESHOLD=10 +``` + +This is the minimum cohort size the trust will release anything about. Cohort statistics below it are privacy-suppressed (a genuine zero and a small count are indistinguishable), and both row-level routes refuse outright — `/cohort/dataframe`, which supplies FL training data, and `/cohort/accession-ids`, which decides whose imaging is pulled into XNAT. Raise it to release less. It is the operator's setting, not the hub's: trusts need not agree on a value, and the hub cannot lower it. See [`data-access-api/README.md`](data-access-api/README.md#row-level-data-and-the-disclosure-threshold). The same schema serves dev trusts (GSTT/KCH against a local hub), on-prem trusts (against a prod hub), and laptop-against-prod testing — operator picks the kit code (`trust/.env..`) and `make -C trust up-trust KIT= PROD=` handles the rest. ### 3. Start the trust against the hub diff --git a/trust/data-access-api/AGENTS.md b/trust/data-access-api/AGENTS.md index 77c6312ad..c56f2f57d 100644 --- a/trust/data-access-api/AGENTS.md +++ b/trust/data-access-api/AGENTS.md @@ -10,6 +10,10 @@ FastAPI service for querying the OMOP Common Data Model database. Receives cohor - Receives internal requests from trust-api (all `/cohort` endpoints) and from imaging-api (`/cohort/accession-ids`); not directly exposed - Every internal caller authenticates with the per-trust `TRUST_INTERNAL_SERVICE_KEY` header (see the root `AGENTS.md` "Trust-internal Service Authentication" section). `/health` stays unauthenticated - OMOP CDM query translation layer +- `services/cohort.py::validate_query` is the **authority** on cohort-query safety: one parse-validate-emit pass that returns the query re-emitted from the AST it checked. Pass that return value to the engine, never the caller's raw string. The hub runs its own pre-check, but it is fast-feedback only and deliberately weaker — never relax a rule here on the assumption the hub filtered first, and do not mirror trust-local rules onto the hub. See [`README.md`](README.md#cohort-query-validation) +- Both row-level routes — `/cohort/dataframe` (FL training data) and `/cohort/accession-ids` (the accession list that decides whose imaging is pulled into XNAT) — are gated on `COHORT_QUERY_THRESHOLD` and share one fixed refusal string, so a below-threshold cohort is indistinguishable from an empty one and the refusal cannot act as a row-count oracle. The trust enforces this itself; the hub's staging guard is not relied on +- The gates evaluate the **live** cohort on every call, not the cohort as approved. The cohort query is re-run against OMOP at every stage (the imaging status poll alone re-runs it roughly every 10s while a project page is open), so a project can import cleanly and later start refusing. There is no frozen approved-cohort artefact anywhere in FLIP — see FLIP#857 +- `COHORT_QUERY_THRESHOLD` is the trust's own disclosure floor (default 10), set per trust in its kit file. It is a `PositiveInt`: `0` would disable both row-level gates (`len(df) < 0` is never true) *and* the statistics suppression at once, so a non-positive value is rejected at import and the service refuses to start rather than run with no floor. Its default lives in one place — the module-level `DEFAULT_COHORT_QUERY_THRESHOLD`, read by both the field default and the empty-string coercion validator; do not re-inline the literal. Requiring the value to be at least the shipped 10 rather than merely positive is FLIP#870. Any new non-`str` setting needs its own empty-string coercion validator: the service Makefile's `export $(shell sed 's/=.*//' $(KIT_ENV_FILE))` strips values from commented lines too, so a commented-out entry arrives as `""` and pydantic rejects it at import ## Commands diff --git a/trust/data-access-api/CLAUDE.md b/trust/data-access-api/CLAUDE.md index 879509c7f..196fd2cb2 100644 --- a/trust/data-access-api/CLAUDE.md +++ b/trust/data-access-api/CLAUDE.md @@ -10,6 +10,10 @@ FastAPI service for querying the OMOP Common Data Model database. Receives cohor - Receives internal requests from trust-api (all `/cohort` endpoints) and from imaging-api (`/cohort/accession-ids`); not directly exposed - Every internal caller authenticates with the per-trust `TRUST_INTERNAL_SERVICE_KEY` header (see the root `CLAUDE.md` "Trust-internal Service Authentication" section). `/health` stays unauthenticated - OMOP CDM query translation layer +- `services/cohort.py::validate_query` is the **authority** on cohort-query safety: one parse-validate-emit pass that returns the query re-emitted from the AST it checked. Pass that return value to the engine, never the caller's raw string. The hub runs its own pre-check, but it is fast-feedback only and deliberately weaker — never relax a rule here on the assumption the hub filtered first, and do not mirror trust-local rules onto the hub. See [`README.md`](README.md#cohort-query-validation) +- Both row-level routes — `/cohort/dataframe` (FL training data) and `/cohort/accession-ids` (the accession list that decides whose imaging is pulled into XNAT) — are gated on `COHORT_QUERY_THRESHOLD` and share one fixed refusal string, so a below-threshold cohort is indistinguishable from an empty one and the refusal cannot act as a row-count oracle. The trust enforces this itself; the hub's staging guard is not relied on +- The gates evaluate the **live** cohort on every call, not the cohort as approved. The cohort query is re-run against OMOP at every stage (the imaging status poll alone re-runs it roughly every 10s while a project page is open), so a project can import cleanly and later start refusing. There is no frozen approved-cohort artefact anywhere in FLIP — see FLIP#857 +- `COHORT_QUERY_THRESHOLD` is the trust's own disclosure floor (default 10), set per trust in its kit file. It is a `PositiveInt`: `0` would disable both row-level gates (`len(df) < 0` is never true) *and* the statistics suppression at once, so a non-positive value is rejected at import and the service refuses to start rather than run with no floor. Its default lives in one place — the module-level `DEFAULT_COHORT_QUERY_THRESHOLD`, read by both the field default and the empty-string coercion validator; do not re-inline the literal. Requiring the value to be at least the shipped 10 rather than merely positive is FLIP#870. Any new non-`str` setting needs its own empty-string coercion validator: the service Makefile's `export $(shell sed 's/=.*//' $(KIT_ENV_FILE))` strips values from commented lines too, so a commented-out entry arrives as `""` and pydantic rejects it at import ## Commands diff --git a/trust/data-access-api/README.md b/trust/data-access-api/README.md index 0addf697c..7cfbc23ea 100644 --- a/trust/data-access-api/README.md +++ b/trust/data-access-api/README.md @@ -91,6 +91,149 @@ Each trust has a distinct key. A trust's `TRUST_INTERNAL_SERVICE_KEY` is minted For the threat model, see the **Trust-internal Service Authentication** section in [`CLAUDE.md`](../../CLAUDE.md). +## Cohort query validation + +A cohort query passes three layers before it runs, and they do **different jobs**. They are not +copies of each other and must not be made into copies: + +| Layer | Job | Authority? | +| --- | --- | --- | +| flip-ui cohort form | Required-field validation only — "you typed something" | No | +| flip-api `submit_cohort_query.validate_query` | Fast-feedback validity pre-check before fan-out | No | +| data-access-api `services/cohort.validate_query` | Decides what may actually run against OMOP | **Yes** | + +### This service is the authority + +`validate_query` in [`services/cohort.py`](data_access_api/services/cohort.py) is the security +boundary, and it is deliberately self-sufficient. A trust holds patient data and the central hub +is a separate administrative domain; the trust must stay safe regardless of what the hub did or +did not check, because a compromised, misconfigured, or simply out-of-date hub must not be able to +widen what a trust will execute. Nothing here may be relaxed on the assumption that the hub +filtered first. + +It performs a single parse-validate-emit pass and enforces: + +1. A maximum query length (cheap DoS guard, before the parser allocates an AST). +2. Exactly one non-empty statement (defeats query stacking and stray semicolons). +3. A SELECT-shaped top-level statement (`SELECT`/`UNION`/`INTERSECT`/`EXCEPT`). +4. No `INSERT`/`UPDATE`/`DELETE`/`MERGE` node anywhere in the tree. Rule 3 only inspects the + top-level node, and Postgres allows a writable CTE — `WITH x AS (DELETE FROM t RETURNING *) + SELECT * FROM x` parses as a `Select` and would otherwise pass. The read-only role rejects + the write regardless, so this is defence in depth rather than the only barrier. +5. Schema-qualified tables limited to `omop` (blocks `information_schema` / `pg_catalog` + enumeration, which Postgres exposes to role `public` by default). +6. Literal-integer `LIMIT`/`OFFSET` (defeats blind extraction that makes the row count a function + of a character value and reads it back through the cohort-size response). + +It then returns the query **re-emitted from the AST it just checked**. Callers pass that string to +the engine, never the caller's original — so what reaches Postgres is generated from a validated +tree. Underneath all of this the service connects as `data_analyst_reader`, a role with `SELECT` +only and `INSERT`/`UPDATE`/`DELETE`/`TRUNCATE`/`CREATE` revoked, so DDL and DML are refused by +Postgres itself. That is why `validate_query` does not keyword-filter for `DROP` and friends. + +Emitting from `validate_query` rather than from a second helper is deliberate: it keeps one parse +and one policy, so there is no second copy of the single-statement and SELECT-shape rules to drift +out of step. + +### The hub check is fast feedback, not security + +`validate_query` in `flip_api/cohort_services/submit_cohort_query.py` runs before the hub fans a +query out. Because this service is self-sufficient, anything the hub enforced for safety would +have to be enforced here anyway — so duplicating these rules on the hub would buy no security, +only two copies of one policy to keep in sync. + +What it buys instead is **fast feedback**. Submitting a cohort query fans it out to every +registered trust as an encrypted task, each validating asynchronously and reporting back. Without +a pre-check, a researcher who typos their SQL waits for that entire round-trip, across N trusts, +to find out. Catching "this cannot succeed anywhere" while the request is still in hand turns a +multi-minute fan-out into an immediate 400. + +So the hub check is deliberately *weaker*, and only rejects what every trust would reject too: +unparseable input, multiple statements, non-SELECT statements, and over-length queries. It +intentionally does **not** enforce trust-local policy — the `omop` schema pin, the literal-`LIMIT` +rule, the writable-CTE rejection, the read-only role, or the cohort-size threshold — because those depend on facts the hub has +no authority over and which may legitimately differ per trust. + +That asymmetry makes drift harmless in the direction that matters. A hub lagging behind a trust +merely wastes a fan-out on a query the trust then refuses; it is never a bypass. A hub that got +*ahead* would reject something a trust would have allowed, which surfaces immediately as a usability +regression. Neither is a security failure — which is a far cheaper invariant to maintain than +"these two must match". + +### No keyword denylists anywhere + +The hub check does not use a keyword denylist, and neither does the UI. Both previously carried +the same regex banning substrings including `substring` — which blocked every legitimate use of +the standard `SUBSTRING()` function while stopping nothing an attacker would actually do. The +blind-extraction technique it was aimed at is defeated properly here by rule 5 above, and DDL/DML +is refused by the read-only database role. + +Two copies of one denylist across two services was also exactly the drift hazard this layering +exists to avoid: a rule that is not authoritative anywhere still had to be maintained in both +places. The UI now validates only that a query was entered, and reports SQL problems from the +hub's response. + +### Row-level data and the disclosure threshold + +`/cohort` returns aggregate statistics and suppresses any count below `COHORT_QUERY_THRESHOLD`, +including a genuine zero, so the response cannot reveal that at least one patient matched. + +`/cohort/dataframe` is the training-data path — user FL code reaches it through +`flip.get_dataframe(...)` — so it necessarily returns row-level records; a model trains on rows. +The data stays inside the trust and only model updates leave it. It applies the same +`COHORT_QUERY_THRESHOLD`: a cohort below the threshold is refused with a 403 whose message is +identical for zero rows and for threshold-minus-one, so the refusal cannot be used as an oracle. +Such a cohort has no training value anyway. + +There is deliberately no column allowlist on `/cohort/dataframe`. `accession_id` is load-bearing — +it is how returned rows join to the imaging studies pulled into XNAT — and shipped tutorials +legitimately select `*`, so a column filter would break every FL app while a caller could trivially +alias around it. Column-level minimisation belongs in the cohort query a project submits and in +project approval, not at this layer. + +`/cohort/accession-ids` is the minimal-disclosure endpoint: it wraps the caller's validated query +so only the `accession_id` column can cross the boundary. It applies the same threshold and the +same fixed refusal text, because accession IDs are still row-level identifiers — and they are the +pointer set into the imaging data, deciding whose studies get pulled into XNAT where project +members view them. Releasing them for a cohort of three is the disclosure the threshold exists to +prevent, whatever the column count. + +The trust applies that check itself rather than relying on the hub. The hub does have a guard — +`stage_project` refuses to stage a trust whose cohort came back empty or suppressed — but relying +on it would be exactly the "assume the hub filtered first" this layering rejects, and the hub's own +`start_project_imaging_creation` endpoint does not re-check staging. + +**Both row-level gates evaluate the cohort as it is now, not as it was at approval.** FLIP has no +frozen approved-cohort artefact: the cohort query is a SQL string that is re-run against live OMOP +at every stage — including by the imaging status poll roughly every 10 seconds while a user has the +project page open. A project can therefore import cleanly and later start refusing if its cohort +shrinks below the threshold. That is the correct behaviour for a disclosure control, but it means +the gate is not a one-time approval-time check; see FLIP#857 for the underlying design gap. + +When `/cohort/accession-ids` refuses, imaging-api translates the 403 into a +`CohortBelowThresholdError` rather than a generic transport failure, so the initial pull logs a +clear reason and queues nothing, and the status path reports a readable message to the hub instead +of a raw HTTP error. + +### Configuring the threshold + +`COHORT_QUERY_THRESHOLD` defaults to `10` and is the trust's own disclosure floor — set it in the +trust's kit file (`trust/.env..`) to raise it. It is not a hub setting and trusts need +not agree on a value. + +It must be a **positive integer**; `0` or a negative value is rejected at startup rather than +accepted. A threshold of `0` would disable every check that reads it in one stroke — both row-level +gates (`len(df) < 0` is never true, so `/cohort/dataframe` and `/cohort/accession-ids` would release +a cohort of any size) and the statistics suppression on `/cohort`. Settings are built at import, so +a bad value stops the service starting instead of leaving it running with no floor. Requiring the +value to be at least the shipped `10`, rather than merely positive, is tracked in FLIP#870. + +Note for anyone adding settings here: the service Makefile exports kit-file names with +`sed 's/=.*//'`, which strips the value from *every* line including commented ones, so a +commented-out entry reaches the process as an empty string. Any non-`str` setting therefore needs +an empty-string coercion validator (see `coerce_empty_cohort_query_threshold` in `config.py`), or +the service fails at import. + ## Testing Tests are split into `tests/` (unit-level, no real backing services — `tests/routers/`, `tests/services/`, `tests/db/`, etc.) and `tests/integration/` (real OMOP database via the shared `trust/deploy/compose.test.yml` stack). See [Where does my test go?](../../CONTRIBUTING.md#where-does-my-test-go) in `CONTRIBUTING.md` for the placement rule, and [`trust/README.md`](../README.md#integration-tests-cohort-query-end-to-end) for how the cohort-query end-to-end suite is wired. diff --git a/trust/data-access-api/data_access_api/config.py b/trust/data-access-api/data_access_api/config.py index b8d857c76..8c3a4e687 100644 --- a/trust/data-access-api/data_access_api/config.py +++ b/trust/data-access-api/data_access_api/config.py @@ -16,6 +16,10 @@ from pydantic import PositiveInt, SecretStr, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict +# The shipped disclosure floor. Referenced by both the field default and the empty-string +# coercion validator below, so the two cannot drift apart. +DEFAULT_COHORT_QUERY_THRESHOLD = 10 + class Settings(BaseSettings): """Common settings shared across all environments (development and production).""" @@ -45,7 +49,31 @@ def coerce_empty_env(cls, v: str) -> str: LOG_LEVEL: str = "INFO" # - COHORT_QUERY_THRESHOLD: int = 10 # Minimum number of records required to return statistics + # Minimum cohort size below which no row-level data or statistics are released. The + # trust's own disclosure floor: operators may raise it in their kit file, and both + # row-level routes (/cohort/dataframe, /cohort/accession-ids) enforce it alongside the + # statistics suppression on /cohort. ``PositiveInt`` because a threshold of 0 or less + # silently disables every one of those checks (``len(df) < 0`` is never true) — a + # disclosure control that can be turned off by a typo is worse than one that refuses to + # boot. Flooring it at the shipped 10 rather than 1 is FLIP#870. + COHORT_QUERY_THRESHOLD: PositiveInt = DEFAULT_COHORT_QUERY_THRESHOLD + + @field_validator("COHORT_QUERY_THRESHOLD", mode="before") + @classmethod + def coerce_empty_cohort_query_threshold(cls, v: object) -> object: + """Treat an empty-string threshold as the default. + + The service Makefile exports kit-file names with ``sed 's/=.*//'``, which strips the + value from every line — including commented ones — so an entry the operator has + commented out arrives as ``COHORT_QUERY_THRESHOLD=""``. pydantic treats a present but + empty env var as a real override and rejects it against ``int``, which would fail at + import and take the whole service (and its test collection) down. Same shape as + ``coerce_empty_env``. + """ + if v is None or v == "": + return DEFAULT_COHORT_QUERY_THRESHOLD + return v + CACHE_TTL_DAYS: int = 60 # Number of days before cached query results expire CACHE_MAX_RESULT_ROWS: PositiveInt = 50_000 # Max rows per cached result; larger results skip caching CACHE_MAX_ENTRIES: PositiveInt = 64 # Max number of cached query results diff --git a/trust/data-access-api/data_access_api/routers/cohort.py b/trust/data-access-api/data_access_api/routers/cohort.py index ca6104a0e..8df5cb4db 100644 --- a/trust/data-access-api/data_access_api/routers/cohort.py +++ b/trust/data-access-api/data_access_api/routers/cohort.py @@ -12,9 +12,6 @@ from typing import Any -import sqlglot -import sqlglot.errors -import sqlglot.expressions from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.exc import SQLAlchemyError @@ -30,52 +27,12 @@ from data_access_api.utils.internal_auth import authenticate_internal_service from data_access_api.utils.logger import logger -_READ_ONLY_STATEMENT_TYPES = ( - sqlglot.expressions.Select, - sqlglot.expressions.Union, - sqlglot.expressions.Intersect, - sqlglot.expressions.Except, -) - - -def _parse_and_emit(query: str) -> str: - """Parse SQL with sqlglot and re-emit it to break the injection taint chain. - - Using sqlglot as a parse-then-emit step ensures the string reaching the - database engine is generated from a validated AST, not directly from the - HTTP request body. The output is semantically equivalent to the input for - all valid SELECT queries while also normalising trailing semicolons and - whitespace. - - Only read-only SELECT-shaped statements (SELECT, UNION, INTERSECT, EXCEPT) - are permitted. DML (INSERT, UPDATE, DELETE) and DDL (DROP, CREATE, ALTER, - TRUNCATE) are rejected with HTTP 400. - - Args: - query: Raw SQL string from the caller. - - Returns: - Re-emitted SQL string. - - Raises: - HTTPException: 400 if the query cannot be parsed, is empty, contains - multiple statements, or is not a read-only SELECT statement. - """ - try: - transpiled = sqlglot.transpile(query, read="postgres", write="postgres") - except sqlglot.errors.SqlglotError as exc: - raise HTTPException(status_code=400, detail=f"Invalid SQL: {exc}") from exc - if not transpiled or not transpiled[0].strip(): - raise HTTPException(status_code=400, detail="SQL query is empty or could not be parsed") - if len(transpiled) > 1: - raise HTTPException(status_code=400, detail="Multiple SQL statements are not allowed") - try: - ast = sqlglot.parse_one(transpiled[0], dialect="postgres") - except sqlglot.errors.SqlglotError as exc: - raise HTTPException(status_code=400, detail=f"Invalid SQL: {exc}") from exc - if not isinstance(ast, _READ_ONLY_STATEMENT_TYPES): - raise HTTPException(status_code=400, detail="Only SELECT statements are allowed") - return transpiled[0] +# Returned instead of row-level data when a cohort is smaller than +# COHORT_QUERY_THRESHOLD, by both row-level routes (/cohort/dataframe and +# /cohort/accession-ids). Deliberately fixed text: it must be identical for a +# cohort of zero and a cohort of threshold-minus-one, or the refusal itself +# becomes a one-row oracle for probing the database. +_BELOW_THRESHOLD_DETAIL = "Cohort is too small for row-level data to be released." # Create Router @@ -111,11 +68,9 @@ def receive_cohort_query(query_input: CohortQueryInput) -> StatisticsResponse: minimum_cohort_size = get_settings().COHORT_QUERY_THRESHOLD logger.info(f"Minimum cohort size needed to return statistics: {minimum_cohort_size}") - validate_query(query_input.query) - # On the original implementation get_records was invoked within get_statistics. However, to better handle # exceptions and log the query execution, we separate the two calls here. - safe_query = _parse_and_emit(query_input.query) + safe_query = validate_query(query_input.query) try: logger.info("Executing cohort query") @@ -130,8 +85,11 @@ def receive_cohort_query(query_input: CohortQueryInput) -> StatisticsResponse: try: results = get_statistics(df, query_input=query_input, threshold=minimum_cohort_size) - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) + except Exception: + # Detail is category-only: the trust forwards it to the hub, which shows it to every + # project member, and raw exception text from the aggregation can carry row values. + logger.exception("Cohort statistics aggregation failed") + raise HTTPException(status_code=500, detail="Statistics aggregation failed.") logger.info("Cohort query returned results") return results @@ -142,10 +100,24 @@ def get_dataframe(query_input: DataframeQuery) -> dict[str, list[Any]]: """ Retrieves query results in a DataFrame-like structure (column-oriented dictionary). - TODO Do not return certain columns? e.g. "accession_id", "referring_physician", etc. - - 1. Decrypt the central hub project ID. - 2. Send the query using get_dataframe(project_id, query). + This is the training-data path: user-supplied FL code inside the trust's + fl-client reaches it through ``flip.get_dataframe(...)``, so it necessarily + returns row-level records — a model trains on rows. The data stays inside + the trust; only model updates leave it. + + Because it is row-level, the cohort must clear ``COHORT_QUERY_THRESHOLD`` + before anything is released, mirroring the suppression the ``/cohort`` + statistics route already applies. A cohort below the threshold is refused + outright rather than returned truncated: there is no training value in it, + and releasing a handful of identifiable rows is exactly the disclosure the + threshold exists to prevent. + + Note there is deliberately no column allowlist here. ``accession_id`` is + load-bearing — it is how the returned rows join to the imaging studies + pulled into XNAT — and shipped tutorials legitimately select ``*``, so a + column filter would break every FL app on the platform while a caller could + trivially alias around it. Column-level minimisation belongs in the cohort + query the project submits and in project approval, not here. Args: query_input (DataframeQuery): The input data for the DataFrame query. @@ -154,22 +126,39 @@ def get_dataframe(query_input: DataframeQuery) -> dict[str, list[Any]]: dict[str, list[Any]]: The query results in a DataFrame-like structure. Raises: - HTTPException: If there is an error during the execution of the query or if the query returns too few records. + HTTPException: 400 if the query is invalid, 403 if the cohort is below + the disclosure threshold, 500 if the query fails to execute. """ project_id = decrypt(query_input.encrypted_project_id) logger.info(f"Received DataFrame query for project {project_id}") - validate_query(query_input.query) - - safe_query = _parse_and_emit(query_input.query) + safe_query = validate_query(query_input.query) try: df = get_records(safe_query) - except SQLAlchemyError as e: - raise HTTPException(status_code=500, detail=str(e)) - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) + except HTTPException: + # get_records already converts driver errors into category-only + # HTTPExceptions; re-wrapping them below would discard that work and + # turn every categorised 400 into an opaque 500. + raise + except SQLAlchemyError: + logger.exception("DataFrame query failed with a database error") + raise HTTPException(status_code=500, detail="Query execution failed.") + except Exception: + # Detail is category-only: the trust forwards it to the hub, which shows + # it to every project member, and raw exception text can carry row + # values and connection internals. + logger.exception("DataFrame query failed unexpectedly") + raise HTTPException(status_code=500, detail="Query execution failed.") + + minimum_cohort_size = get_settings().COHORT_QUERY_THRESHOLD + if len(df) < minimum_cohort_size: + logger.warning( + f"Withholding row-level data for project {project_id}: " + f"cohort below the minimum size of {minimum_cohort_size}" + ) + raise HTTPException(status_code=403, detail=_BELOW_THRESHOLD_DETAIL) return df.to_dict(orient="list") @@ -184,6 +173,25 @@ def get_accession_ids(query_input: DataframeQuery) -> AccessionIdsResponse: endpoint used by imaging-api to fetch the accession numbers it needs to import studies from PACS — it does not expose row-level patient attributes. + Accession IDs are still row-level identifiers, and they are the pointer set into + the imaging data: they decide whose studies get pulled into XNAT, where project + members view them. So the cohort must clear ``COHORT_QUERY_THRESHOLD`` here just + as it must on ``/cohort/dataframe``, and the refusal reuses that route's fixed + text so a zero-row cohort and a below-threshold one are indistinguishable. + + The trust applies this itself rather than relying on the hub's staging guard + (``flip_api.project_services.stage_project``, which refuses to stage a trust whose + cohort came back empty or suppressed). The hub is a separate administrative domain; + a trust must stay safe regardless of what the hub checked, and the hub's own + ``start_project_imaging_creation`` endpoint does not re-check staging. + + **This is evaluated against the live cohort on every call, not once at approval.** + The endpoint is re-invoked by the imaging status poll roughly every 10 s while a + user has the project page open, and again on reimport — each time re-running the + cohort SQL against OMOP, which changes underneath. A project can therefore pull + cleanly at approval and later start refusing if its cohort shrinks below the + threshold. FLIP has no frozen approved-cohort artefact; see FLIP#857. + Args: query_input (DataframeQuery): The cohort query. @@ -191,36 +199,50 @@ def get_accession_ids(query_input: DataframeQuery) -> AccessionIdsResponse: AccessionIdsResponse: The accession IDs returned by the cohort query. Raises: - HTTPException: If the query is invalid, does not select an ``accession_id`` - column, or fails during execution. + HTTPException: 400 if the query is invalid or does not select an + ``accession_id`` column, 403 if the cohort is below the disclosure + threshold, 500 if the query fails to execute. """ project_id = decrypt(query_input.encrypted_project_id) logger.info(f"Received accession-ids query for project {project_id}") - validate_query(query_input.query) - - # Parse and re-emit the caller's SQL via sqlglot to break any injection - # taint chain. sqlglot also strips trailing semicolons so the inner query - # composes cleanly inside the outer SELECT subquery. - safe_inner = _parse_and_emit(query_input.query) + # validate_query returns the caller's SQL re-emitted from its parsed AST, + # which breaks any injection taint chain and strips trailing semicolons so + # the inner query composes cleanly inside the outer SELECT subquery. + safe_inner = validate_query(query_input.query) wrapped_query = f"SELECT accession_id FROM ({safe_inner}) AS cohort_subquery" try: df = get_records(wrapped_query) except HTTPException: + # get_records already converts driver errors into category-only + # HTTPExceptions; re-wrapping them below would discard that work and + # turn every categorised 400 into an opaque 500. raise - except SQLAlchemyError as e: - raise HTTPException(status_code=500, detail=str(e)) - except Exception as e: - raise HTTPException(status_code=500, detail=str(e)) + except SQLAlchemyError: + logger.exception("Accession-ids query failed with a database error") + raise HTTPException(status_code=500, detail="Query execution failed.") + except Exception: + # Detail is category-only: the trust forwards it to the hub, which shows + # it to every project member, and raw exception text can carry row + # values and connection internals. + logger.exception("Accession-ids query failed unexpectedly") + raise HTTPException(status_code=500, detail="Query execution failed.") - if "accession_id" not in df.columns: - raise HTTPException( - status_code=400, - detail="Cohort query did not return an 'accession_id' column.", + minimum_cohort_size = get_settings().COHORT_QUERY_THRESHOLD + if len(df) < minimum_cohort_size: + logger.warning( + f"Withholding accession IDs for project {project_id}: " + f"cohort below the minimum size of {minimum_cohort_size}" ) + raise HTTPException(status_code=403, detail=_BELOW_THRESHOLD_DETAIL) + # No "did the DataFrame come back with accession_id?" guard here: it could never fire. + # The wrapper above selects the column explicitly, so a cohort that does not project it + # fails inside get_records with UndefinedColumn — surfacing as a category 400 through the + # `except HTTPException: raise` branch, before any DataFrame exists. Pinned by + # tests/integration/test_cohort_endpoint.py::test_accession_ids_missing_column_surfaces_get_records_400. accession_ids = [str(value) for value in df["accession_id"].tolist()] logger.info(f"accession-ids query for project {project_id} returned {len(accession_ids)} ids") return AccessionIdsResponse(accession_ids=accession_ids) diff --git a/trust/data-access-api/data_access_api/services/cohort.py b/trust/data-access-api/data_access_api/services/cohort.py index eb577e277..c404dd820 100644 --- a/trust/data-access-api/data_access_api/services/cohort.py +++ b/trust/data-access-api/data_access_api/services/cohort.py @@ -32,8 +32,6 @@ from data_access_api.utils.logger import logger from data_access_api.utils.sql_parsers import extract_missing_identifier -COHORT_QUERY_THRESHOLD = get_settings().COHORT_QUERY_THRESHOLD - # OMOP schema is the only schema callers may reference. Any qualified # reference to a different schema is rejected by validate_query. ALLOWED_SCHEMA = "omop" @@ -43,7 +41,12 @@ # defence in depth and stops the parser allocating an arbitrarily large AST. MAX_QUERY_LENGTH = 10_240 # 10 KiB -# Top-level statement shapes that count as SELECT-like for the cohort API. +# Top-level statement shapes that count as SELECT-like for the cohort API. This is an +# allowlist and must stay one — never add ``exp.Command``, sqlglot's catch-all for syntax it +# does not model (``EXPLAIN`` lands there, as does anything a future sqlglot stops +# understanding). A Command node round-trips the raw text verbatim and exposes no children, +# so the DML, schema and LIMIT/OFFSET walks below would all traverse nothing and pass it +# through unchecked. _ALLOWED_QUERY_TYPES: tuple[type[exp.Expression], ...] = ( exp.Select, exp.Union, @@ -51,13 +54,24 @@ exp.Except, ) +# Data-modifying nodes rejected anywhere in the tree, not just at the top level. +# Postgres allows a writable CTE — ``WITH x AS (DELETE ... RETURNING *) SELECT * FROM x`` +# — which sqlglot parses with a top-level ``exp.Select``, so the SELECT-shape check +# alone passes it through. +_DATA_MODIFYING_TYPES: tuple[type[exp.Expression], ...] = ( + exp.Insert, + exp.Update, + exp.Delete, + exp.Merge, +) + def _invalid_query(detail: str) -> HTTPException: logger.warning(f"Query validation failed: {detail}") return HTTPException(status_code=400, detail=detail) -def validate_query(query: str) -> bool: +def validate_query(query: str) -> str: """ Validates that an inbound SQL query is structurally safe to run against OMOP. @@ -70,6 +84,9 @@ def validate_query(query: str) -> bool: ``CREATE`` explicitly REVOKEd. Any DDL or DML is therefore rejected by Postgres itself, so this function does NOT keyword-filter for ``DROP`` / ``INSERT`` / ``UPDATE`` / etc. — those are already covered at the DB layer. + Rules 3 and 4 below still reject writes *structurally*, from the parsed tree + rather than from a keyword scan, so a write fails in-hand with a clear 400 + instead of as an opaque permission error from the engine. What this function enforces --------------------------- @@ -79,22 +96,35 @@ def validate_query(query: str) -> bool: 2. The query parses as exactly one non-empty statement (defeats query stacking, stray semicolons that bypass the count check, and empty inputs). 3. The top-level statement is SELECT-shaped (rejects ``COPY``, ``EXPLAIN``, - and any DDL/DML the DB would also reject — fail fast at the API). - 4. Any schema-qualified table reference targets only the ``omop`` schema + and top-level DDL/DML — fail fast at the API rather than at the DB). + 4. No ``INSERT`` / ``UPDATE`` / ``DELETE`` / ``MERGE`` node appears *anywhere* + in the tree. Rule 3 inspects only the top-level node, and Postgres allows a + writable CTE — ``WITH x AS (DELETE FROM t RETURNING *) SELECT * FROM x`` + parses as a ``Select`` and would otherwise pass. The read-only role rejects + the write regardless, so this is defence in depth, not the only barrier. + 5. Any schema-qualified table reference targets only the ``omop`` schema (blocks enumeration of ``information_schema``, ``pg_catalog``, ``pg_class`` etc., which Postgres makes readable to role ``public`` by default). - 5. Every ``LIMIT`` and ``OFFSET`` is a literal integer (defeats the blind + 6. Every ``LIMIT`` and ``OFFSET`` is a literal integer (defeats the blind data-extraction technique that abuses ``LIMIT CASE WHEN THEN n ELSE m END`` to make the row count a function of a single character value, then reads it back via the cohort-size error message). + This function is the **authority** on cohort-query safety. The central hub + runs its own pre-check before fanning a query out + (``flip_api.cohort_services.submit_cohort_query.validate_query``), but that + one exists purely for fast feedback and is deliberately weaker: the hub is a + separate administrative domain, so nothing here may be relaxed on the + assumption that the hub filtered first. + Args: query: The SQL query string from the caller. Returns: - ``True`` when the query is structurally safe. + The validated query re-emitted from its parsed AST — pass *this* to the + database, never the caller's original string. Raises: HTTPException(400): When any of the rules above is violated. @@ -122,6 +152,12 @@ def validate_query(query: str) -> bool: if not isinstance(stmt, _ALLOWED_QUERY_TYPES): raise _invalid_query("Only SELECT statements are allowed.") + # The check above only inspects the top-level node, and Postgres lets a write + # hide inside a CTE body while the outer statement still parses as a SELECT. + # Walk the whole tree for data-modifying nodes. + if any(stmt.find_all(*_DATA_MODIFYING_TYPES)): + raise _invalid_query("Data-modifying statements are not allowed.") + # Walk the whole AST so subqueries, CTEs, and set-operation arms are checked. for table in stmt.find_all(exp.Table): schema_node = table.args.get("db") @@ -144,7 +180,14 @@ def validate_query(query: str) -> bool: if not isinstance(value, exp.Literal) or not value.is_int: raise _invalid_query(f"{label} must be a literal integer.") - return True + # Re-emit from the AST we just validated rather than handing the caller's + # original string to the engine. The string that reaches the database is + # therefore generated by sqlglot from a checked tree, which breaks the + # injection taint chain and incidentally normalises trailing semicolons and + # whitespace. Emitting here — instead of in a second helper that re-parses — + # keeps one parse and one policy: there is no second copy of the + # single-statement and SELECT-shape rules to drift out of step with these. + return stmt.sql(dialect="postgres") def get_records( @@ -373,7 +416,7 @@ def verify_cardinality(df: pd.DataFrame, threshold: float = 0.05) -> bool: percentage_unique = unique_count / len(df) if len(df) > 0 else 0 logger.info(f"Column '{col}' has {unique_count} unique values ({percentage_unique:.2%} of total)") if all([ - unique_count < COHORT_QUERY_THRESHOLD, # Absolute threshold + unique_count < get_settings().COHORT_QUERY_THRESHOLD, # Absolute threshold percentage_unique < threshold, # Relative threshold ]): logger.info(f"Column '{col}' has insufficient unique values ({threshold=}, {unique_count=})") @@ -381,17 +424,22 @@ def verify_cardinality(df: pd.DataFrame, threshold: float = 0.05) -> bool: return True -def make_other_category(results: list[dict], min_count: int = COHORT_QUERY_THRESHOLD) -> list[dict]: +def make_other_category(results: list[dict], min_count: int | None = None) -> list[dict]: """ Groups entries in the results list with counts less than min_count into an "Other" category. Args: results (list of dict): List of dictionaries with 'value' and 'count' keys. - min_count (int): Minimum count threshold to avoid grouping into "Other". + min_count (int | None): Minimum count threshold to avoid grouping into "Other". + Defaults to ``COHORT_QUERY_THRESHOLD``, resolved at call time — a default + argument would bind the setting at import and ignore a per-trust override. Returns: list of dict: Updated list with low-count entries grouped into "Other". """ + if min_count is None: + min_count = get_settings().COHORT_QUERY_THRESHOLD + other_count = sum(item["count"] for item in results if item["count"] < min_count) filtered_results = [item for item in results if item["count"] >= min_count] @@ -421,21 +469,27 @@ def get_statistics(df: pd.DataFrame, query_input: CohortQueryInput, threshold: i Args: df (pd.DataFrame): Query results dataframe. query_input (data_access_api.routers.schema.CohortQueryInput): Input object containing the query and metadata. - threshold (int): Minimum number of records required to return results. + threshold (int): Minimum number of records the caller requires. ``COHORT_QUERY_THRESHOLD`` + is applied as a floor underneath it, so a caller can raise the bar but never + lower it below the trust's configured disclosure threshold. Returns: StatisticsResponse: Contains the aggregated statistics, or a 0-count empty response - when below ``COHORT_QUERY_THRESHOLD``. + when below the effective threshold. """ record_count = len(df) + # The configured threshold is a floor, not a default: a caller passing a smaller value + # must not be able to weaken suppression. Read live rather than at import so a per-trust + # override actually applies. + threshold = max(threshold, get_settings().COHORT_QUERY_THRESHOLD) - if record_count < COHORT_QUERY_THRESHOLD: + if record_count < threshold: # Privacy-suppress every below-threshold count, INCLUDING a genuine zero: a true # zero and a small (1..threshold-1) count return identically (record_count=0, # suppressed=True) so the response can't reveal that >=1 patient matched. # Distinguishing them would leak membership/existence (issue #519, security review). logger.info( - f"Query returned {record_count} records (< {COHORT_QUERY_THRESHOLD});" + f"Query returned {record_count} records (< {threshold});" " returning privacy-suppressed 0-count response" ) return StatisticsResponse( @@ -459,10 +513,10 @@ def get_statistics(df: pd.DataFrame, query_input: CohortQueryInput, threshold: i if "person_id" in df.columns: logger.info("person_id column found in the query results; including age and sex distribution calculations.") age = get_age_distribution(df) - age["results"] = make_other_category(age["results"], min_count=COHORT_QUERY_THRESHOLD) + age["results"] = make_other_category(age["results"], min_count=threshold) sex = get_sex_distribution(df) - sex["results"] = make_other_category(sex["results"], min_count=COHORT_QUERY_THRESHOLD) + sex["results"] = make_other_category(sex["results"], min_count=threshold) stats.data += [age, sex] return stats diff --git a/trust/data-access-api/tests/integration/test_cohort_endpoint.py b/trust/data-access-api/tests/integration/test_cohort_endpoint.py index cbf72b21c..69d938dfd 100644 --- a/trust/data-access-api/tests/integration/test_cohort_endpoint.py +++ b/trust/data-access-api/tests/integration/test_cohort_endpoint.py @@ -41,6 +41,19 @@ def _cohort_payload(query: str) -> dict: } +def _dataframe_payload(query: str) -> dict: + """Payload for the two row-level routes, which take ``DataframeQuery`` (two fields only). + + ``encrypted_project_id`` is encrypted with the same AES key the container is configured + with, so the decrypt path stays real rather than mocked. The import is function-local for + the same reason it is in ``test_dataframe_endpoint_returns_seeded_columns``: the conftest + pins ``AES_KEY_BASE64`` before any ``data_access_api`` module builds its Settings singleton. + """ + from data_access_api.utils.encryption import encrypt + + return {"encrypted_project_id": encrypt("integration-project-1"), "query": query} + + def test_cohort_endpoint_returns_aggregates_for_image_occurrences(http_client): """All 24 image_occurrence rows clear the threshold and come back with the full aggregate set.""" response = http_client.post( @@ -148,3 +161,83 @@ def test_dataframe_endpoint_returns_seeded_columns(http_client): assert body["modality"].count("XR") == 4 # accession_id is unique per row in the seed. assert len(set(body["accession_id"])) == 24 + + +def test_accession_ids_returns_seeded_ids(http_client): + """``/cohort/accession-ids`` projects the id column out of the caller's cohort. + + The route wraps the caller's (re-emitted) SQL as ``SELECT accession_id FROM (...) AS + cohort_subquery``, so only that one column ever crosses the trust boundary regardless of + what the inner query selected. + """ + response = http_client.post( + "/cohort/accession-ids", + json=_dataframe_payload("SELECT person_id, accession_id FROM omop.image_occurrence"), + ) + assert response.status_code == 200, response.text + + accession_ids = response.json()["accession_ids"] + assert len(accession_ids) == 24 + # accession_id is unique per row in the seed (ACC-1001 … ACC-1024). + assert len(set(accession_ids)) == 24 + assert "ACC-1001" in accession_ids + + +def test_accession_ids_missing_column_surfaces_get_records_400(http_client): + """A cohort that does not project ``accession_id`` fails inside ``get_records``, not after it. + + Pins which 400 actually reaches the caller. Because the route wraps the inner query in + ``SELECT accession_id FROM (...)``, Postgres raises ``UndefinedColumn`` while the query is + executing — so ``get_records`` converts it to a category 400 and the router's + ``except HTTPException: raise`` branch re-raises it before any DataFrame is returned. A + router-level "did the DataFrame come back with the column?" guard could never fire, which + is why there isn't one; this test is the standing proof of that. The cohort here is 16 + rows — comfortably over the stack's threshold — so the refusal is about the column, not + the cohort size. + """ + response = http_client.post( + "/cohort/accession-ids", + json=_dataframe_payload("SELECT person_id FROM omop.person"), + ) + assert response.status_code == 400, response.text + assert response.json()["detail"] == "The column 'accession_id' does not exist." + + +def test_accession_ids_below_threshold_is_refused(http_client): + """A below-threshold cohort is refused outright — no partial list, no count. + + modality_concept_id 4013632 = 'XR'; the seed has 4 such rows, under the stack's + COHORT_QUERY_THRESHOLD of 5. + """ + response = http_client.post( + "/cohort/accession-ids", + json=_dataframe_payload( + "SELECT accession_id FROM omop.image_occurrence WHERE modality_concept_id = 4013632" + ), + ) + assert response.status_code == 403, response.text + assert response.json()["detail"] == "Cohort is too small for row-level data to be released." + + +def test_accession_ids_zero_rows_indistinguishable_from_below_threshold(http_client): + """Privacy regression, against real Postgres rather than a mocked DataFrame. + + A cohort matching nothing and a cohort of 1-4 rows must produce byte-identical refusals, + or the response itself becomes a row-count oracle: a caller could binary-search a + predicate and learn whether *any* patient matches it, one bit at a time. + """ + below_threshold = http_client.post( + "/cohort/accession-ids", + json=_dataframe_payload( + "SELECT accession_id FROM omop.image_occurrence WHERE modality_concept_id = 4013632" + ), + ) + genuine_zero = http_client.post( + "/cohort/accession-ids", + json=_dataframe_payload( + "SELECT accession_id FROM omop.image_occurrence WHERE accession_id = 'NONEXISTENT'" + ), + ) + + assert below_threshold.status_code == genuine_zero.status_code == 403 + assert below_threshold.text == genuine_zero.text diff --git a/trust/data-access-api/tests/routers/test_cohort.py b/trust/data-access-api/tests/routers/test_cohort.py index 067456397..dd215f160 100644 --- a/trust/data-access-api/tests/routers/test_cohort.py +++ b/trust/data-access-api/tests/routers/test_cohort.py @@ -10,7 +10,7 @@ # limitations under the License. # -from unittest.mock import MagicMock, patch +from unittest.mock import patch import pandas as pd import pytest @@ -71,7 +71,9 @@ def test_receive_cohort_query_success(mock_get_statistics, mock_validate_query, assert response.status_code == 200 assert response.json() == sample_statistics_response mock_validate_query.assert_called_once_with(sample_query_input["query"]) - mock_get_records.assert_called_once_with(sample_query_input["query"]) + # The engine receives what validate_query emitted from the checked AST, + # never the caller's raw string. + mock_get_records.assert_called_once_with(mock_validate_query.return_value) mock_get_statistics.assert_called_once() @@ -94,8 +96,10 @@ def test_receive_cohort_query_invalid_validation(mock_validate_query, mock_get_s def test_receive_cohort_query_statistics_error( mock_get_statistics, mock_validate_query, mock_get_settings, mock_get_records ): + """Aggregation failures return a category only — the hub relays this detail to every + project member, and raw exception text can carry row values (FLIP-PT-016).""" mock_get_settings.return_value.COHORT_QUERY_THRESHOLD = 5 - mock_get_statistics.side_effect = RuntimeError("Statistics computation failed") + mock_get_statistics.side_effect = RuntimeError("failed on patient 12345 birth_datetime") # Mock DataFrame mock_df = pd.DataFrame({"col1": range(10)}) @@ -104,7 +108,8 @@ def test_receive_cohort_query_statistics_error( response = client.post("/cohort", json=sample_query_input, headers=AUTH_HEADERS) assert response.status_code == 500 - assert response.json()["detail"] == "Statistics computation failed" + assert response.json()["detail"] == "Statistics aggregation failed." + assert "12345" not in response.text @patch("data_access_api.routers.cohort.get_records") @@ -189,20 +194,24 @@ def test_receive_cohort_query_execution_error(mock_validate_query, mock_get_sett } +@patch("data_access_api.routers.cohort.get_settings") @patch("data_access_api.routers.cohort.decrypt") @patch("data_access_api.routers.cohort.get_records") -def test_get_dataframe_success(mock_get_records, mock_decrypt): +@patch("data_access_api.routers.cohort.validate_query") +def test_get_dataframe_success(mock_validate_query, mock_get_records, mock_decrypt, mock_get_settings): + mock_get_settings.return_value.COHORT_QUERY_THRESHOLD = 2 mock_decrypt.return_value = "decrypted-id" - mock_df = MagicMock() - mock_df.to_dict.return_value = sample_df_dict - mock_get_records.return_value = mock_df + mock_get_records.return_value = pd.DataFrame(sample_df_dict) response = client.post("/cohort/dataframe", json=sample_dataframe_query, headers=AUTH_HEADERS) assert response.status_code == 200 assert response.json() == sample_df_dict mock_decrypt.assert_called_once_with("encrypted-id") - mock_get_records.assert_called_once_with(sample_dataframe_query["query"]) + mock_validate_query.assert_called_once_with(sample_dataframe_query["query"]) + # The engine receives what validate_query emitted from the checked AST, + # never the caller's raw string. + mock_get_records.assert_called_once_with(mock_validate_query.return_value) @patch("data_access_api.routers.cohort.decrypt") @@ -220,26 +229,95 @@ def test_get_dataframe_invalid_query(mock_validate_query, mock_decrypt): @patch("data_access_api.routers.cohort.decrypt") @patch("data_access_api.routers.cohort.get_records") def test_get_dataframe_sqlalchemy_error(mock_get_records, mock_decrypt): + """Driver text must not reach the caller — the hub relays this detail to the UI.""" mock_decrypt.return_value = "decrypted-id" - mock_get_records.side_effect = SQLAlchemyError("SQLAlchemy error") + mock_get_records.side_effect = SQLAlchemyError("relation omop.person has 42 rows for patient Bob") response = client.post("/cohort/dataframe", json=sample_dataframe_query, headers=AUTH_HEADERS) - # SQLAlchemyError is caught as a general Exception if not explicitly imported and matched assert response.status_code == 500 - assert response.json()["detail"] == "SQLAlchemy error" + assert "Bob" not in response.json()["detail"] @patch("data_access_api.routers.cohort.decrypt") @patch("data_access_api.routers.cohort.get_records") def test_get_dataframe_generic_error(mock_get_records, mock_decrypt): + """Unexpected exception text must not reach the caller either.""" mock_decrypt.return_value = "decrypted-id" - mock_get_records.side_effect = RuntimeError("Unexpected failure") + mock_get_records.side_effect = RuntimeError("connection string postgres://user:hunter2@omop-db") response = client.post("/cohort/dataframe", json=sample_dataframe_query, headers=AUTH_HEADERS) assert response.status_code == 500 - assert response.json()["detail"] == "Unexpected failure" + assert "hunter2" not in response.json()["detail"] + + +@patch("data_access_api.routers.cohort.decrypt") +@patch("data_access_api.routers.cohort.get_records") +def test_get_dataframe_preserves_http_exception_from_get_records(mock_get_records, mock_decrypt): + """A categorised 400 from get_records must not be re-wrapped as an opaque 500. + + ``get_records`` deliberately converts driver errors into category-only + HTTPExceptions. ``except Exception`` also catches HTTPException, so the + route used to swallow that work and re-raise every one of them as a 500 + whose detail was the *repr of the HTTPException*. + """ + mock_decrypt.return_value = "decrypted-id" + mock_get_records.side_effect = HTTPException(status_code=400, detail="Column 'nope' does not exist") + + response = client.post("/cohort/dataframe", json=sample_dataframe_query, headers=AUTH_HEADERS) + + assert response.status_code == 400 + assert response.json()["detail"] == "Column 'nope' does not exist" + + +@patch("data_access_api.routers.cohort.get_settings") +@patch("data_access_api.routers.cohort.decrypt") +@patch("data_access_api.routers.cohort.get_records") +def test_get_dataframe_rejects_cohort_below_threshold(mock_get_records, mock_decrypt, mock_get_settings): + """Row-level data is withheld for cohorts smaller than COHORT_QUERY_THRESHOLD.""" + mock_get_settings.return_value.COHORT_QUERY_THRESHOLD = 10 + mock_decrypt.return_value = "decrypted-id" + mock_get_records.return_value = pd.DataFrame({"age": range(9)}) + + response = client.post("/cohort/dataframe", json=sample_dataframe_query, headers=AUTH_HEADERS) + + assert response.status_code == 403 + + +@patch("data_access_api.routers.cohort.get_settings") +@patch("data_access_api.routers.cohort.decrypt") +@patch("data_access_api.routers.cohort.get_records") +def test_get_dataframe_below_threshold_does_not_disclose_row_count( + mock_get_records, mock_decrypt, mock_get_settings +): + """The refusal must not reveal how many rows matched — 0 and 9 look identical.""" + mock_get_settings.return_value.COHORT_QUERY_THRESHOLD = 10 + mock_decrypt.return_value = "decrypted-id" + + details = [] + for row_count in (0, 9): + mock_get_records.return_value = pd.DataFrame({"age": range(row_count)}) + response = client.post("/cohort/dataframe", json=sample_dataframe_query, headers=AUTH_HEADERS) + details.append(response.json()["detail"]) + + assert details[0] == details[1] + assert "9" not in details[1] + + +@patch("data_access_api.routers.cohort.get_settings") +@patch("data_access_api.routers.cohort.decrypt") +@patch("data_access_api.routers.cohort.get_records") +def test_get_dataframe_allows_cohort_at_threshold(mock_get_records, mock_decrypt, mock_get_settings): + """A cohort exactly at the threshold is released — the gate is not off-by-one.""" + mock_get_settings.return_value.COHORT_QUERY_THRESHOLD = 10 + mock_decrypt.return_value = "decrypted-id" + mock_get_records.return_value = pd.DataFrame({"age": range(10)}) + + response = client.post("/cohort/dataframe", json=sample_dataframe_query, headers=AUTH_HEADERS) + + assert response.status_code == 200 + assert len(response.json()["age"]) == 10 # --------------------------------------------------------------------------- @@ -247,9 +325,11 @@ def test_get_dataframe_generic_error(mock_get_records, mock_decrypt): # --------------------------------------------------------------------------- +@patch("data_access_api.routers.cohort.get_settings") @patch("data_access_api.routers.cohort.decrypt") @patch("data_access_api.routers.cohort.get_records") -def test_get_accession_ids_success(mock_get_records, mock_decrypt): +def test_get_accession_ids_success(mock_get_records, mock_decrypt, mock_get_settings): + mock_get_settings.return_value.COHORT_QUERY_THRESHOLD = 2 mock_decrypt.return_value = "decrypted-id" mock_get_records.return_value = pd.DataFrame({"accession_id": ["ACC1", "ACC2", "ACC3"]}) @@ -264,9 +344,11 @@ def test_get_accession_ids_success(mock_get_records, mock_decrypt): assert sample_dataframe_query["query"] in called_query +@patch("data_access_api.routers.cohort.get_settings") @patch("data_access_api.routers.cohort.decrypt") @patch("data_access_api.routers.cohort.get_records") -def test_get_accession_ids_strips_trailing_semicolon(mock_get_records, mock_decrypt): +def test_get_accession_ids_strips_trailing_semicolon(mock_get_records, mock_decrypt, mock_get_settings): + mock_get_settings.return_value.COHORT_QUERY_THRESHOLD = 1 mock_decrypt.return_value = "decrypted-id" mock_get_records.return_value = pd.DataFrame({"accession_id": ["ACC1"]}) @@ -297,14 +379,25 @@ def test_get_accession_ids_invalid_query(mock_validate_query, mock_decrypt): @patch("data_access_api.routers.cohort.decrypt") @patch("data_access_api.routers.cohort.get_records") -def test_get_accession_ids_missing_column(mock_get_records, mock_decrypt): +def test_get_accession_ids_missing_column_propagates_400(mock_get_records, mock_decrypt): + """A cohort that does not project ``accession_id`` is refused by ``get_records``, not the router. + + The route wraps the inner query as ``SELECT accession_id FROM (...)``, so Postgres raises + ``UndefinedColumn`` during execution and ``get_records`` turns it into a category 400 — + there is no DataFrame to inspect afterwards, which is why the router carries no + column-presence guard. This mocks the conversion ``get_records`` performs; the real + Postgres behaviour it stands in for is pinned by + ``tests/integration/test_cohort_endpoint.py::test_accession_ids_missing_column_surfaces_get_records_400``. + """ mock_decrypt.return_value = "decrypted-id" - mock_get_records.return_value = pd.DataFrame({"some_other_column": [1, 2]}) + mock_get_records.side_effect = HTTPException( + status_code=400, detail="The column 'accession_id' does not exist." + ) response = client.post("/cohort/accession-ids", json=sample_dataframe_query, headers=AUTH_HEADERS) assert response.status_code == 400 - assert "accession_id" in response.json()["detail"] + assert response.json()["detail"] == "The column 'accession_id' does not exist." @patch("data_access_api.routers.cohort.decrypt") @@ -325,26 +418,84 @@ def test_get_accession_ids_propagates_http_exception(mock_get_records, mock_decr @patch("data_access_api.routers.cohort.decrypt") @patch("data_access_api.routers.cohort.get_records") -def test_get_accession_ids_sqlalchemy_error(mock_get_records, mock_decrypt): +def test_get_accession_ids_sqlalchemy_error_does_not_leak(mock_get_records, mock_decrypt): + """Driver text must not reach the caller: the trust forwards this detail to the hub, + which shows it to every project member (FLIP-PT-016).""" mock_decrypt.return_value = "decrypted-id" - mock_get_records.side_effect = SQLAlchemyError("SQLAlchemy error") + mock_get_records.side_effect = SQLAlchemyError("relation omop.secret_table row 42") response = client.post("/cohort/accession-ids", json=sample_dataframe_query, headers=AUTH_HEADERS) assert response.status_code == 500 - assert response.json()["detail"] == "SQLAlchemy error" + assert response.json()["detail"] == "Query execution failed." + assert "secret_table" not in response.text @patch("data_access_api.routers.cohort.decrypt") @patch("data_access_api.routers.cohort.get_records") -def test_get_accession_ids_generic_error(mock_get_records, mock_decrypt): +def test_get_accession_ids_generic_error_does_not_leak(mock_get_records, mock_decrypt): mock_decrypt.return_value = "decrypted-id" - mock_get_records.side_effect = RuntimeError("Unexpected failure") + mock_get_records.side_effect = RuntimeError("connection to 10.0.0.5 failed for user svc_omop") response = client.post("/cohort/accession-ids", json=sample_dataframe_query, headers=AUTH_HEADERS) assert response.status_code == 500 - assert response.json()["detail"] == "Unexpected failure" + assert response.json()["detail"] == "Query execution failed." + assert "svc_omop" not in response.text + + +@patch("data_access_api.routers.cohort.get_settings") +@patch("data_access_api.routers.cohort.decrypt") +@patch("data_access_api.routers.cohort.get_records") +def test_get_accession_ids_rejects_cohort_below_threshold(mock_get_records, mock_decrypt, mock_get_settings): + """Accession IDs are row-level identifiers and decide whose imaging is pulled into XNAT, + so a below-threshold cohort is refused just as it is on /cohort/dataframe.""" + mock_get_settings.return_value.COHORT_QUERY_THRESHOLD = 10 + mock_decrypt.return_value = "decrypted-id" + mock_get_records.return_value = pd.DataFrame({"accession_id": [f"ACC{i}" for i in range(9)]}) + + response = client.post("/cohort/accession-ids", json=sample_dataframe_query, headers=AUTH_HEADERS) + + assert response.status_code == 403 + assert response.json()["detail"] == "Cohort is too small for row-level data to be released." + # No identifier may appear in the refusal. + assert "ACC" not in response.text + + +@patch("data_access_api.routers.cohort.get_settings") +@patch("data_access_api.routers.cohort.decrypt") +@patch("data_access_api.routers.cohort.get_records") +def test_get_accession_ids_below_threshold_is_indistinguishable_from_zero( + mock_get_records, mock_decrypt, mock_get_settings +): + """A zero-row cohort and a below-threshold one must return byte-identical responses, + or the refusal itself becomes a row-count oracle.""" + mock_get_settings.return_value.COHORT_QUERY_THRESHOLD = 10 + mock_decrypt.return_value = "decrypted-id" + + mock_get_records.return_value = pd.DataFrame({"accession_id": []}) + zero_response = client.post("/cohort/accession-ids", json=sample_dataframe_query, headers=AUTH_HEADERS) + + mock_get_records.return_value = pd.DataFrame({"accession_id": [f"ACC{i}" for i in range(9)]}) + below_response = client.post("/cohort/accession-ids", json=sample_dataframe_query, headers=AUTH_HEADERS) + + assert zero_response.status_code == below_response.status_code == 403 + assert zero_response.text == below_response.text + + +@patch("data_access_api.routers.cohort.get_settings") +@patch("data_access_api.routers.cohort.decrypt") +@patch("data_access_api.routers.cohort.get_records") +def test_get_accession_ids_allows_cohort_at_threshold(mock_get_records, mock_decrypt, mock_get_settings): + """Exactly at the threshold is allowed — the gate is `<`, not `<=`.""" + mock_get_settings.return_value.COHORT_QUERY_THRESHOLD = 10 + mock_decrypt.return_value = "decrypted-id" + mock_get_records.return_value = pd.DataFrame({"accession_id": [f"ACC{i}" for i in range(10)]}) + + response = client.post("/cohort/accession-ids", json=sample_dataframe_query, headers=AUTH_HEADERS) + + assert response.status_code == 200 + assert len(response.json()["accession_ids"]) == 10 # --------------------------------------------------------------------------- @@ -391,92 +542,86 @@ def test_health_does_not_require_auth(): # --------------------------------------------------------------------------- -# _parse_and_emit unit tests +# Parse-then-emit tests +# +# validate_query is the single parse-validate-emit step: it returns the caller's +# SQL re-emitted from the AST it just checked. These cases used to target a +# separate _parse_and_emit helper in this router, which re-parsed the query and +# kept its own copy of the single-statement and SELECT-shape rules. That second +# copy is gone; the behaviour it guarded is asserted here against the one +# remaining implementation. # --------------------------------------------------------------------------- -from data_access_api.routers.cohort import _parse_and_emit # noqa: E402 +from data_access_api.services.cohort import validate_query # noqa: E402 -def test_parse_and_emit_empty_string(): +def test_validate_query_rejects_empty_string(): with pytest.raises(HTTPException) as exc_info: - _parse_and_emit("") + validate_query("") assert exc_info.value.status_code == 400 - assert "empty" in exc_info.value.detail.lower() -def test_parse_and_emit_whitespace_only(): +def test_validate_query_rejects_whitespace_only(): with pytest.raises(HTTPException) as exc_info: - _parse_and_emit(" \n\t ") + validate_query(" \n\t ") assert exc_info.value.status_code == 400 -def test_parse_and_emit_multi_statement(): +def test_validate_query_rejects_multi_statement(): with pytest.raises(HTTPException) as exc_info: - _parse_and_emit("SELECT 1; SELECT 2") + validate_query("SELECT 1; SELECT 2") assert exc_info.value.status_code == 400 - assert "Multiple SQL statements" in exc_info.value.detail + assert "one SQL statement" in exc_info.value.detail -def test_parse_and_emit_multi_statement_with_drop(): +def test_validate_query_rejects_multi_statement_with_drop(): """A semicolon-separated DML statement is rejected before any execution.""" with pytest.raises(HTTPException) as exc_info: - _parse_and_emit("SELECT id FROM omop.person; DROP TABLE omop.person") + validate_query("SELECT id FROM omop.person; DROP TABLE omop.person") assert exc_info.value.status_code == 400 - assert "Multiple SQL statements" in exc_info.value.detail + assert "one SQL statement" in exc_info.value.detail -def test_parse_and_emit_strips_trailing_semicolon(): - result = _parse_and_emit("SELECT 1;") +def test_validate_query_strips_trailing_semicolon(): + result = validate_query("SELECT 1;") assert ";" not in result assert "1" in result -def test_parse_and_emit_valid_select(): - result = _parse_and_emit("SELECT * FROM omop.radiology_occurrence") +def test_validate_query_emits_valid_select(): + result = validate_query("SELECT * FROM omop.radiology_occurrence") assert "omop.radiology_occurrence" in result -def test_parse_and_emit_cte_roundtrip(): +def test_validate_query_cte_roundtrip(): query = "WITH cte AS (SELECT id FROM omop.person) SELECT * FROM cte" - result = _parse_and_emit(query) + result = validate_query(query) assert "cte" in result.lower() assert result.upper().startswith("WITH") -def test_parse_and_emit_complex_pg_syntax(): - """PG-specific constructs (window functions, FILTER, LATERAL) survive the round-trip.""" +def test_validate_query_complex_pg_syntax_roundtrip(): + """PG-specific constructs (aggregate FILTER clauses) survive the round-trip.""" query = ( "SELECT person_id, COUNT(*) FILTER (WHERE age > 18) AS adult_count " "FROM omop.person GROUP BY person_id" ) - result = _parse_and_emit(query) + result = validate_query(query) assert "person_id" in result - assert "adult_count" in result.lower() or "ADULT_COUNT" in result - - -def test_parse_and_emit_rejects_insert(): - with pytest.raises(HTTPException) as exc_info: - _parse_and_emit("INSERT INTO omop.person (person_id) VALUES (1)") - assert exc_info.value.status_code == 400 - assert "SELECT" in exc_info.value.detail - - -def test_parse_and_emit_rejects_drop(): - with pytest.raises(HTTPException) as exc_info: - _parse_and_emit("DROP TABLE omop.person") - assert exc_info.value.status_code == 400 - assert "SELECT" in exc_info.value.detail - - -def test_parse_and_emit_rejects_update(): - with pytest.raises(HTTPException) as exc_info: - _parse_and_emit("UPDATE omop.person SET gender_concept_id = 0 WHERE person_id = 1") - assert exc_info.value.status_code == 400 - assert "SELECT" in exc_info.value.detail + assert "adult_count" in result.lower() -def test_parse_and_emit_rejects_delete(): +@pytest.mark.parametrize( + "query", + [ + "INSERT INTO omop.person (person_id) VALUES (1)", + "DROP TABLE omop.person", + "UPDATE omop.person SET gender_concept_id = 0 WHERE person_id = 1", + "DELETE FROM omop.person WHERE person_id = 1", + ], +) +def test_validate_query_rejects_dml_and_ddl(query: str): with pytest.raises(HTTPException) as exc_info: - _parse_and_emit("DELETE FROM omop.person WHERE person_id = 1") + validate_query(query) assert exc_info.value.status_code == 400 assert "SELECT" in exc_info.value.detail diff --git a/trust/data-access-api/tests/services/test_cohort.py b/trust/data-access-api/tests/services/test_cohort.py index 729984622..e0cc03544 100644 --- a/trust/data-access-api/tests/services/test_cohort.py +++ b/trust/data-access-api/tests/services/test_cohort.py @@ -14,10 +14,12 @@ import pandas as pd import pytest +import sqlglot from fastapi import HTTPException from pandas.errors import DatabaseError as PandasDatabaseError from psycopg2 import errors as pg_errors from sqlalchemy.exc import DBAPIError, SQLAlchemyError +from sqlglot import exp from data_access_api.routers.schema import CohortQueryInput from data_access_api.services.cohort import ( @@ -113,10 +115,13 @@ def test_get_statistics_below_threshold(mock_read_sql, mock_df_below_threshold): assert stats.suppressed is True -@patch("data_access_api.services.cohort.COHORT_QUERY_THRESHOLD", 10) @patch("pandas.read_sql") def test_get_statistics_fails_global_threshold(mock_read_sql): - """Record count above caller threshold but below the global threshold also returns 0.""" + """Record count above caller threshold but below the configured threshold also returns 0. + + The configured ``COHORT_QUERY_THRESHOLD`` (10 by default) is a floor: a caller passing a + lower value cannot weaken suppression. + """ # Create a dataframe with 8 records (between 5 and 10) mock_df_medium = pd.DataFrame({ "modality": ["CT"] * 8, @@ -134,8 +139,8 @@ def test_get_statistics_fails_global_threshold(mock_read_sql): trust_id="mock_trust", ) - # Pass a low threshold (5) so the first check (record_count < threshold) passes (8 < 5 is False) - # but the global check (len(df) < COHORT_QUERY_THRESHOLD) fails (8 < 10 is True). + # Pass a low threshold (5): on its own 8 would clear it, but the configured floor of 10 + # is applied underneath, so 8 < 10 suppresses. stats = get_statistics(mock_df_medium, query_input, threshold=5) assert stats.record_count == 0 assert stats.data == [] @@ -413,8 +418,23 @@ def test_get_records_pandas_error_without_dbapi_cause(mock_read_sql): ], ) def test_validate_query_accepts_well_formed_select(query: str): - """Plain SELECT (with or without omop qualification, CTE, or UNION) is accepted.""" - assert validate_query(query) is True + """Plain SELECT (with or without omop qualification, CTE, or UNION) is accepted. + + Asserts the emit contract, not just truthiness: the return value is the query + re-emitted from the validated AST, and it is what callers hand to the engine — so + it must be a non-empty string that still parses back to a SELECT-shaped statement. + """ + emitted = validate_query(query) + + assert isinstance(emitted, str) + assert emitted.strip() + # A trailing semicolon in the input must not survive into the emitted SQL, which is + # composed into a subquery by /cohort/accession-ids. + assert ";" not in emitted + + reparsed = sqlglot.parse(emitted, read="postgres") + assert len(reparsed) == 1 + assert isinstance(reparsed[0], (exp.Select, exp.Union, exp.Intersect, exp.Except)) @pytest.mark.parametrize( @@ -443,6 +463,33 @@ def test_validate_query_rejects_garbage_that_parses_as_non_select(): validate_query("INVALID SQL") +@pytest.mark.parametrize( + "query", + [ + "WITH x AS (DELETE FROM omop.person RETURNING *) SELECT * FROM x", + "WITH x AS (INSERT INTO omop.person (person_id) VALUES (1) RETURNING *) SELECT * FROM x", + "WITH x AS (UPDATE omop.person SET person_id = 1 RETURNING *) SELECT * FROM x", + # Nested one level deeper — the walk must reach it, not just the top-level CTE list. + "WITH x AS (WITH y AS (DELETE FROM omop.person RETURNING *) SELECT * FROM y) SELECT * FROM x", + ], +) +def test_validate_query_rejects_data_modifying_cte(query: str): + """Postgres allows a writable CTE, and sqlglot parses the whole thing as a top-level + ``Select`` — so the SELECT-shape check alone lets it through. The walk must catch it. + + Defence in depth rather than the only barrier: ``data_analyst_reader`` has no write + grant, so Postgres would reject these too. But the validator claims to reject writes, + and it should actually do so. + """ + with pytest.raises(HTTPException, match="Data-modifying statements are not allowed"): + validate_query(query) + + +def test_validate_query_still_accepts_read_only_cte(): + """The DML walk must not catch an ordinary read-only CTE.""" + assert validate_query("WITH p AS (SELECT * FROM omop.person) SELECT * FROM p") + + @pytest.mark.parametrize( "query", [ @@ -490,12 +537,23 @@ def test_validate_query_rejects_oversized_query(): "CREATE TABLE foo (id int)", "ALTER TABLE omop.person ADD COLUMN col int", "TRUNCATE omop.person", + # COPY and EXPLAIN are named by rule 3 of the validate_query docstring but were + # previously untested. sqlglot parses them successfully — to exp.Copy and to the + # catch-all exp.Command respectively — so neither is caught by the parse step; both + # are rejected purely because the allowlist admits only SELECT-shaped top-level nodes. + "COPY (SELECT 1) TO STDOUT", + "COPY omop.person FROM PROGRAM 'curl attacker.example'", + "EXPLAIN SELECT * FROM omop.person", ], ) def test_validate_query_rejects_non_select_statements(query: str): """ Non-SELECT statements are rejected at the API layer even though Postgres rejects them too — data_analyst_reader has no DDL/DML privileges. + + The COPY and EXPLAIN cases are characterisation tests: they lock in behaviour the + allowlist already provides, so a future widening of ``_ALLOWED_QUERY_TYPES`` that let + either through fails here rather than silently contradicting the docstring. """ with pytest.raises(HTTPException, match="Only SELECT statements are allowed"): validate_query(query) @@ -1032,6 +1090,25 @@ def test_make_other_category_custom_min_count(): assert result == expected +@patch("data_access_api.services.cohort.get_settings") +def test_make_other_category_defaults_to_configured_threshold(mock_get_settings): + """Omitting min_count resolves COHORT_QUERY_THRESHOLD at call time. + + Resolving in the body rather than as a default argument is what lets a per-trust override + apply: a default argument would bind the value once at import. + """ + mock_get_settings.return_value.COHORT_QUERY_THRESHOLD = 20 + results = [ + {"value": "A", "count": 25}, + {"value": "B", "count": 15}, # Below the configured 20, so grouped + ] + + assert make_other_category(results) == [ + {"value": "A", "count": 25}, + {"value": "Other", "count": 15}, + ] + + # Additional integration tests for get_statistics with the new helper functions diff --git a/trust/data-access-api/tests/test_config.py b/trust/data-access-api/tests/test_config.py new file mode 100644 index 000000000..dfe26d597 --- /dev/null +++ b/trust/data-access-api/tests/test_config.py @@ -0,0 +1,67 @@ +# Copyright (c) 2026 Guy's and St Thomas' NHS Foundation Trust & King's College London +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import pytest +from pydantic import ValidationError + +from data_access_api.config import Settings + +# Deliberately a literal rather than an import of ``config.DEFAULT_COHORT_QUERY_THRESHOLD``: +# these tests pin the shipped value independently of its source, so importing the constant +# would make ``test_cohort_query_threshold_defaults_to_ten`` tautological. +DEFAULT_COHORT_QUERY_THRESHOLD = 10 + + +def test_cohort_query_threshold_defaults_to_ten(): + """The shipped disclosure floor. Kept explicit so a change to it is a deliberate edit.""" + assert Settings().COHORT_QUERY_THRESHOLD == DEFAULT_COHORT_QUERY_THRESHOLD + + +@pytest.mark.parametrize("value", ["", None]) +def test_cohort_query_threshold_coerces_empty_to_default(value): + """An empty value must fall back to the default rather than fail validation. + + The service Makefile exports kit-file names with ``sed 's/=.*//'``, stripping the value + from every line including commented ones — so a commented-out entry reaches the process + as an empty string. Without coercion pydantic rejects it against ``int`` at import, + which takes down the service and every test that imports it. + """ + assert Settings(COHORT_QUERY_THRESHOLD=value).COHORT_QUERY_THRESHOLD == DEFAULT_COHORT_QUERY_THRESHOLD + + +def test_cohort_query_threshold_empty_coercion_tracks_field_default(): + """The empty-string coercion must yield whatever the field default is, not a copy of it. + + Regression test for the drift the validator used to carry: it returned a hard-coded ``10`` + alongside a field default of ``10``, so changing one silently left the other behind. Both + now read ``DEFAULT_COHORT_QUERY_THRESHOLD``, and re-introducing a literal fails this. + """ + assert Settings(COHORT_QUERY_THRESHOLD="").COHORT_QUERY_THRESHOLD == Settings().COHORT_QUERY_THRESHOLD + + +def test_cohort_query_threshold_accepts_operator_override(): + """A trust may raise its own floor; the value still parses from a string as env vars do.""" + assert Settings(COHORT_QUERY_THRESHOLD="25").COHORT_QUERY_THRESHOLD == 25 + + +@pytest.mark.parametrize("value", [0, -1, "0", "-5"]) +def test_cohort_query_threshold_rejects_non_positive(value): + """A non-positive threshold must fail loudly rather than disable the control. + + ``COHORT_QUERY_THRESHOLD=0`` would turn off every check that reads it at once: both + row-level gates (``len(df) < 0`` is never true, so ``/cohort/dataframe`` and + ``/cohort/accession-ids`` would release any cohort, including a single patient) and the + statistics suppression on ``/cohort``. Settings are built at import, so rejecting here + means the service refuses to start instead of running with the floor silently removed. + """ + with pytest.raises(ValidationError): + Settings(COHORT_QUERY_THRESHOLD=value) diff --git a/trust/deploy/compose_trust.development.yml b/trust/deploy/compose_trust.development.yml index a21435cd9..17f766f8f 100644 --- a/trust/deploy/compose_trust.development.yml +++ b/trust/deploy/compose_trust.development.yml @@ -159,6 +159,10 @@ services: ENV: development LOG_LEVEL: ${TRUST_LOG_LEVEL} TRUST_INTERNAL_SERVICE_KEY: ${TRUST_INTERNAL_SERVICE_KEY} + # The trust's own minimum-cohort disclosure floor. Operator-owned: raise it in the + # kit file to release less. The `:-10` keeps an unset variable from arriving as an + # empty string, which pydantic would reject against `int`. + COHORT_QUERY_THRESHOLD: ${COHORT_QUERY_THRESHOLD:-10} # Dev-on-pull (see `user:` above): import via PYTHONPATH, skip the editable # install (UV_NO_SYNC), keep uv's cache + HOME writable. HOME: /tmp diff --git a/trust/deploy/compose_trust.production.yml b/trust/deploy/compose_trust.production.yml index 3bea63092..11b1162cf 100644 --- a/trust/deploy/compose_trust.production.yml +++ b/trust/deploy/compose_trust.production.yml @@ -116,6 +116,10 @@ services: ENV: production LOG_LEVEL: ${TRUST_LOG_LEVEL} TRUST_INTERNAL_SERVICE_KEY: ${TRUST_INTERNAL_SERVICE_KEY} + # The trust's own minimum-cohort disclosure floor. Operator-owned: raise it in the + # kit file to release less. The `:-10` keeps an unset variable from arriving as an + # empty string, which pydantic would reject against `int`. + COHORT_QUERY_THRESHOLD: ${COHORT_QUERY_THRESHOLD:-10} trust-api: image: ghcr.io/londonaicentre/trust-api:${DOCKER_TAG} diff --git a/trust/imaging-api/imaging_api/services/retrieval.py b/trust/imaging-api/imaging_api/services/retrieval.py index 6857ded34..e791bb0c6 100644 --- a/trust/imaging-api/imaging_api/services/retrieval.py +++ b/trust/imaging-api/imaging_api/services/retrieval.py @@ -31,7 +31,7 @@ from imaging_api.services_external.data_access import get_accession_ids from imaging_api.utils.auth import get_xnat_auth_headers from imaging_api.utils.encryption import encrypt -from imaging_api.utils.exceptions import NotFoundError +from imaging_api.utils.exceptions import CohortBelowThresholdError, NotFoundError from imaging_api.utils.logger import logger XNATAuthHeaders = Annotated[dict[str, str], Depends(get_xnat_auth_headers)] @@ -64,7 +64,9 @@ async def retrieve_images_for_project(project_id: str, query: str, headers: XNAT headers (XNATAuthHeaders): The headers containing XNAT authentication details. Returns: - bool: True if all studies were successfully queued, False otherwise. + bool: True if all studies were successfully queued, False otherwise — including when + the cohort is below the trust's ``COHORT_QUERY_THRESHOLD``, in which case the Data + Access API withholds the accession IDs and nothing is imported. Raises: HTTPException: If the request cannot be processed. @@ -81,7 +83,18 @@ async def retrieve_images_for_project(project_id: str, query: str, headers: XNAT # query to the accession_id column server-side, so no other columns are # transmitted across the trust boundary. encrypted_project_id = encrypt(project_id) - accession_ids: list[str] = await get_accession_ids(encrypted_project_id, query) + try: + accession_ids: list[str] = await get_accession_ids(encrypted_project_id, query) + except CohortBelowThresholdError: + # A settled outcome, not a failure to retry: the trust will not release identifiers + # for a cohort this small, so there is nothing to import. Logged rather than raised + # because this runs as a background task — the create-project response has already + # been sent, so an exception here would only reach the server log as a traceback. + logger.warning( + f"Not importing images for project {project_id}: the cohort is below the trust's " + "minimum size, so the Data Access API withheld its accession IDs." + ) + return False studies_list: list[ImportStudy] = [] @@ -166,14 +179,30 @@ async def get_import_status(project_id: str, query: str, headers: XNATAuthHeader ImportStatus: An object containing the status of study imports. Raises: - HTTPException: If the request cannot be processed. + HTTPException: 403 if the cohort has fallen below the trust's ``COHORT_QUERY_THRESHOLD`` + so its accession IDs cannot be released — note the cohort query is re-run against + live OMOP on every status check, so a project that imported cleanly can start + refusing later if its cohort shrinks (see FLIP#857). Otherwise, if the request + cannot be processed. """ # Encrypt project ID to send to the data access API encrypted_project_id = encrypt(project_id) # Get accession IDs from data access API (server-side projection — no other # cohort columns leave the trust). - accession_ids: list[str] = await get_accession_ids(encrypted_project_id, query) + try: + accession_ids: list[str] = await get_accession_ids(encrypted_project_id, query) + except CohortBelowThresholdError as e: + # Unlike the import path, this one has a caller waiting on a response, so surface the + # reason: trust-api relays this detail to the hub, and without it the per-trust status + # shows only a raw transport error for what is actually a policy decision. + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=( + "Cohort is below the trust's minimum size, so its accession IDs cannot be " + "released and import status cannot be reported." + ), + ) from e # Fetches a list of XNAT experiments associated with a given XNAT project. experiments = get_experiments(project_id, headers) diff --git a/trust/imaging-api/imaging_api/services_external/data_access.py b/trust/imaging-api/imaging_api/services_external/data_access.py index 53cdeb373..1607070dd 100644 --- a/trust/imaging-api/imaging_api/services_external/data_access.py +++ b/trust/imaging-api/imaging_api/services_external/data_access.py @@ -16,6 +16,7 @@ from pydantic import BaseModel, Field from imaging_api.config import get_settings +from imaging_api.utils.exceptions import CohortBelowThresholdError from imaging_api.utils.logger import logger DATA_ACCESS_API_URL = get_settings().DATA_ACCESS_API_URL @@ -45,8 +46,10 @@ async def get_accession_ids(encrypted_project_id: str, query: str) -> list[str]: list[str]: The accession IDs returned by the cohort query, in query order. Raises: - RuntimeError: If the HTTP call to the Data Access API fails (network error or non-2xx - response). + CohortBelowThresholdError: If the cohort is smaller than the trust's + ``COHORT_QUERY_THRESHOLD`` and data-access-api refuses to release identifiers. + RuntimeError: If the HTTP call to the Data Access API fails for any other reason + (network error or other non-2xx response). """ request = AccessionIdsRequest(encrypted_project_id=encrypted_project_id, query=query) @@ -67,6 +70,21 @@ async def get_accession_ids(encrypted_project_id: str, query: str) -> list[str]: response.raise_for_status() return list(response.json().get("accession_ids", [])) + except httpx.HTTPStatusError as exc: + if exc.response.status_code == httpx.codes.FORBIDDEN: + # A deliberate refusal, not a failure: the cohort is below the trust's disclosure + # threshold. Typed separately so callers can report it as a settled outcome rather + # than as a transport error they might retry. + message = ( + "get_accession_ids: the Data Access API refused to release accession IDs — " + "the cohort is below the trust's minimum size." + ) + logger.warning(message) + raise CohortBelowThresholdError(message) from exc + error_message = f"get_accession_ids: HTTP error occurred while calling the Data Access API: {exc}" + logger.error(error_message) + raise RuntimeError(error_message) from exc + except httpx.HTTPError as exc: error_message = f"get_accession_ids: HTTP error occurred while calling the Data Access API: {exc}" logger.error(error_message) diff --git a/trust/imaging-api/imaging_api/utils/exceptions.py b/trust/imaging-api/imaging_api/utils/exceptions.py index 57cbfda89..6ca73fdd5 100644 --- a/trust/imaging-api/imaging_api/utils/exceptions.py +++ b/trust/imaging-api/imaging_api/utils/exceptions.py @@ -39,6 +39,22 @@ class XnatFetchError(Exception): pass +class CohortBelowThresholdError(Exception): + """Exception raised when data-access-api refuses a cohort as below the disclosure threshold. + + A typed alternative to the generic ``RuntimeError`` that ``get_accession_ids`` raises for + transport failures, so callers can tell a deliberate policy refusal (HTTP 403) apart from + "the call broke". The distinction matters because the two want opposite handling: a broken + call should be retried and surfaced as an error, whereas a below-threshold cohort is a + settled answer — nothing to import, and retrying cannot change it. + + A plain ``Exception`` subclass (no custom ``__init__``) so ``str(err)`` renders the raised + message verbatim, matching ``XnatFetchError`` above. + """ + + pass + + class InternalServerError(Exception): """Exception raised for internal server errors.""" diff --git a/trust/imaging-api/tests/services/test_retrieval.py b/trust/imaging-api/tests/services/test_retrieval.py index 29638d0df..75c8dc9f4 100644 --- a/trust/imaging-api/tests/services/test_retrieval.py +++ b/trust/imaging-api/tests/services/test_retrieval.py @@ -23,7 +23,7 @@ retrieve_images_for_project, retry_retrieve_images_for_project, ) -from imaging_api.utils.exceptions import NotFoundError +from imaging_api.utils.exceptions import CohortBelowThresholdError, NotFoundError @pytest.fixture @@ -107,6 +107,48 @@ async def test_retrieve_images_project_generic_error(mock_get_project, headers): assert exc_info.value.status_code == 500 +@pytest.mark.asyncio +@patch("imaging_api.services.retrieval.queue_image_import_request") +@patch("imaging_api.services.retrieval.query_by_accession_number") +@patch("imaging_api.services.retrieval.get_accession_ids", new_callable=AsyncMock) +@patch("imaging_api.services.retrieval.encrypt") +@patch("imaging_api.services.retrieval.get_project") +async def test_retrieve_images_below_threshold_queues_nothing( + mock_get_project, mock_encrypt, mock_get_accession_ids, mock_query, mock_queue, headers, +): + """A below-threshold cohort is a settled outcome, not a crash. + + This runs as a background task after the create-project response has been sent, so letting + the exception escape would surface only as a traceback in the server log while the hub + still recorded the imaging project as created. Return False and queue nothing instead. + """ + mock_get_project.return_value = MagicMock() + mock_encrypt.return_value = "encrypted_id" + mock_get_accession_ids.side_effect = CohortBelowThresholdError("cohort below minimum size") + + result = await retrieve_images_for_project("proj1", "SELECT *", headers) + + assert result is False + mock_query.assert_not_called() + mock_queue.assert_not_called() + + +@pytest.mark.asyncio +@patch("imaging_api.services.retrieval.get_accession_ids", new_callable=AsyncMock) +@patch("imaging_api.services.retrieval.encrypt") +async def test_get_import_status_below_threshold_raises_403(mock_encrypt, mock_get_accession_ids, headers): + """The status path has a caller waiting, so the refusal must reach it as a 403 with a + readable reason — trust-api relays this detail to the hub for the per-trust status.""" + mock_encrypt.return_value = "encrypted_id" + mock_get_accession_ids.side_effect = CohortBelowThresholdError("cohort below minimum size") + + with pytest.raises(HTTPException) as exc_info: + await get_import_status("proj1", "SELECT *", headers) + + assert exc_info.value.status_code == 403 + assert "below the trust's minimum size" in exc_info.value.detail + + @pytest.mark.asyncio @patch("imaging_api.services.retrieval.query_by_accession_number") @patch("imaging_api.services.retrieval.get_accession_ids", new_callable=AsyncMock) diff --git a/trust/imaging-api/tests/services_external/test_data_access.py b/trust/imaging-api/tests/services_external/test_data_access.py index cedd48369..26737899c 100644 --- a/trust/imaging-api/tests/services_external/test_data_access.py +++ b/trust/imaging-api/tests/services_external/test_data_access.py @@ -17,6 +17,7 @@ from imaging_api.config import get_settings from imaging_api.services_external.data_access import get_accession_ids +from imaging_api.utils.exceptions import CohortBelowThresholdError class TestGetAccessionIds: @@ -77,3 +78,43 @@ async def test_http_error_raises_runtime_error(self, mock_client_cls): with pytest.raises(RuntimeError, match="HTTP error occurred"): await get_accession_ids("encrypted-proj-id", "SELECT * FROM cohort") + + @staticmethod + def _client_returning_status(mock_client_cls, status_code: int): + """Wires the mocked client so ``raise_for_status`` raises for ``status_code``.""" + request = httpx.Request("POST", "http://data-access-api/cohort/accession-ids") + response = httpx.Response(status_code, request=request) + + mock_response = MagicMock() + mock_response.raise_for_status = MagicMock( + side_effect=httpx.HTTPStatusError("error", request=request, response=response) + ) + + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + @pytest.mark.asyncio + @patch("imaging_api.services_external.data_access.httpx.AsyncClient") + async def test_403_raises_cohort_below_threshold(self, mock_client_cls): + """A 403 is the trust refusing to release identifiers for a too-small cohort. + + It must be typed distinctly from a transport failure: callers report it as a settled + outcome rather than an error to retry, and retrying cannot change the answer. + """ + self._client_returning_status(mock_client_cls, 403) + + with pytest.raises(CohortBelowThresholdError, match="below the trust's minimum size"): + await get_accession_ids("encrypted-proj-id", "SELECT * FROM cohort") + + @pytest.mark.asyncio + @pytest.mark.parametrize("status_code", [400, 401, 500, 503]) + @patch("imaging_api.services_external.data_access.httpx.AsyncClient") + async def test_other_status_errors_still_raise_runtime_error(self, mock_client_cls, status_code): + """Only 403 is special-cased; every other non-2xx stays a RuntimeError.""" + self._client_returning_status(mock_client_cls, status_code) + + with pytest.raises(RuntimeError, match="HTTP error occurred"): + await get_accession_ids("encrypted-proj-id", "SELECT * FROM cohort")