lake: gate /query and stop it reading outside the two lake views

The SELECT guard was a denylist of DDL/DML verbs. DuckDB's file readers and
setting introspection are plain SELECTs, so it passed all of these:

  SELECT * FROM read_csv_auto('/etc/passwd')     -> 200, file contents
  SELECT * FROM glob('/etc/*')                   -> 200, directory listing
  SELECT current_setting('s3_secret_access_key') -> 200, the bucket credentials

The last one is the worst: bucket mode installed the S3 credentials with
SET s3_secret_access_key, and settings are readable back out. So one
unauthenticated POST returned the ERA5 bucket key in plaintext -- precisely what
the module docstring says this service exists to avoid ("without shipping S3
credentials to every web task"). /query had no authentication of any kind, while
the comparable /internal/* router has gated its whole surface since it landed.

Three changes, outermost first:

- /query now requires the shared internal token, reusing internal_routes'
  dependency. Nothing calls it: the only in-repo caller of this service is
  era5lake.py hitting /history, and Centralis reaches the lake through its own
  DuckDB rather than this endpoint. /history is deliberately left open for now --
  gating it means threading the token through the web->lake hot path.
- Credentials are installed as a DuckDB SECRET instead of SET s3_*. Secret values
  are opaque: current_setting() returns NULL and duckdb_secrets() lists the name
  and scope but no key material.
- Bucket mode disables LocalFileSystem after the views are created (they resolve
  lazily, so ordering matters) and locks the configuration. Local mode cannot do
  this -- its views read local parquet -- so the function denylist stays as
  defence in depth for dev and tests.

The new guard tests assert the *guard* refused, not merely that the request
failed: several of these shapes (read_parquet on a text file, a bare
'/etc/passwd') error inside DuckDB anyway, so a status-only assertion would keep
passing with the guard deleted. That is the trap the original four-case test fell
into -- it tested negatives shaped like the regex that had been written.

backend/lake_app.py

@@ -31,10 +31,11 @@ import threading
 import time
 
 import polars as pl
-from fastapi import FastAPI, HTTPException
+from fastapi import Depends, FastAPI, HTTPException
 from fastapi.responses import FileResponse, Response
 from pydantic import BaseModel
 
+from api.internal_routes import _require_internal_token
 from data import era5lake
 
 CACHE_DIR = os.environ.get("THERMOGRAPH_LAKE_CACHE", "/state/lake-cache")
@@ -133,6 +134,32 @@ _FORBIDDEN = re.compile(
     r"\b(insert|update|delete|drop|create|alter|attach|copy|truncate|install|load"
     r"|export|import|pragma|set|call)\b", re.I)
 
+# The verb denylist above is not a security boundary on its own: DuckDB's file
+# readers and setting introspection are plain SELECTs, so every one of these used
+# to sail past it. `SELECT current_setting('s3_secret_access_key')` handed the
+# bucket credentials straight back to the caller -- the precise outcome the
+# module docstring says this service exists to prevent -- and read_csv_auto()/
+# glob() turned /query into an arbitrary read of the container filesystem.
+#
+# A denylist over a full query language is not winnable, so this is defence in
+# depth only. The real boundaries are, in order: the shared-secret gate on the
+# route, DuckDB SECRETs (opaque to current_setting) instead of SET s3_*, and
+# disabled_filesystems in bucket mode. See _connect().
+_DENIED_FUNCS = re.compile(
+    r"\b(read_csv|read_csv_auto|read_parquet|parquet_scan|parquet_metadata"
+    r"|parquet_schema|read_json|read_json_auto|read_json_objects|read_ndjson"
+    r"|read_ndjson_auto|read_text|read_blob|glob|iceberg_scan|iceberg_metadata"
+    r"|delta_scan|sniff_csv|current_setting|getenv|duckdb_settings"
+    r"|duckdb_secrets|duckdb_extensions|duckdb_functions|sql_auto_complete"
+    r"|which_secret|open|shell)\s*\(", re.I)
+
+# `FROM 'some/path'` is DuckDB's replacement scan: a bare string literal in table
+# position reads that path as a file. Legitimate queries name era5_daily or
+# manifest, never a literal.
+_FROM_LITERAL = re.compile(r"\b(from|join)\s*\(?\s*['\"]", re.I)
+
+_GUARD_DETAIL = "only reads of the era5_daily and manifest views are allowed"
+
 
 def _connect():
     """A fresh in-memory DuckDB with the lake views registered. Returns None
@@ -164,11 +191,14 @@ def _connect():
     con.execute("LOAD httpfs")   # both baked into the image at build time,
     con.execute("LOAD iceberg")  # as the runtime user (see Dockerfile)
     host = cfg["endpoint"].split("://", 1)[-1]
-    con.execute(f"SET s3_endpoint='{host}'")
+    # A SECRET, not `SET s3_secret_access_key`: settings are readable back out
-    con.execute("SET s3_url_style='path'")
+    # with current_setting(), so the old form let any caller of /query retrieve
-    con.execute(f"SET s3_region='{cfg['region']}'")
+    # the bucket credentials. Secret values are opaque -- duckdb_secrets() lists
-    con.execute("SET s3_access_key_id=?", [cfg["access_key"]])
+    # the name and scope but never the key material.
-    con.execute("SET s3_secret_access_key=?", [cfg["secret_key"]])
+    con.execute(
+        "CREATE OR REPLACE SECRET lake (TYPE S3, KEY_ID ?, SECRET ?, "
+        "REGION ?, ENDPOINT ?, URL_STYLE 'path')",
+        [cfg["access_key"], cfg["secret_key"], cfg["region"], host])
     # Newest metadata JSON wins (one single-page LIST of metadata/ — cheap);
     # pyiceberg names them NNNNN-uuid.metadata.json, so lexicographic max is
     # the numeric max.
@@ -180,16 +210,26 @@ def _connect():
     con.execute(f"CREATE VIEW era5_daily AS SELECT * FROM iceberg_scan('{meta}')")
     con.execute(f"""CREATE VIEW manifest AS SELECT * FROM
         read_parquet('s3://{cfg['bucket']}/{era5lake.MANIFEST_KEY}')""")
+    # Views are resolved lazily, so this must come after they are created --
+    # and only in bucket mode, where every scan goes over httpfs and nothing
+    # legitimate touches local disk. It makes read_csv_auto('/etc/passwd') and
+    # glob('/etc/*') a PermissionException in the engine rather than relying on
+    # the regex above. lock_configuration then stops a caller re-enabling it
+    # (belt and braces: `set` is already a forbidden verb).
+    con.execute("SET disabled_filesystems='LocalFileSystem'")
+    con.execute("SET lock_configuration=true")
     return con
 
 
-@app.post("/query")
+@app.post("/query", dependencies=[Depends(_require_internal_token)])
 def query(q: Query):
     sql = q.sql.strip().rstrip(";").strip()
     if not sql.lower().startswith(("select", "with")) or ";" in sql \
             or _FORBIDDEN.search(sql):
         raise HTTPException(status_code=400,
                             detail="only a single SELECT statement is allowed")
+    if _DENIED_FUNCS.search(sql) or _FROM_LITERAL.search(sql):
+        raise HTTPException(status_code=400, detail=_GUARD_DETAIL)
     con = _connect()
     if con is None:
         raise HTTPException(status_code=503, detail="lake not configured")

backend/tests/web/test_lake_app.py

@@ -7,12 +7,20 @@ import polars as pl
 import pytest
 from fastapi.testclient import TestClient
 
+# /query is gated on the same shared secret as /internal/* (it is an operator
+# surface, not a public one). The fixture provisions it so the existing query
+# tests exercise the authorised path; the auth tests below drive the closed ones.
+TOKEN = "test-internal-token"
+HDR = {"X-Thermograph-Internal-Token": TOKEN}
+
 
 @pytest.fixture
 def lake(tmp_path, monkeypatch):
     """A tiny two-point lake (points, hive partitions, manifest) on disk."""
     from data import era5lake
 
+    monkeypatch.setenv("THERMOGRAPH_INTERNAL_TOKEN", TOKEN)
+
     points = [(154, 1439), (154, 0)]           # around London
     d0 = datetime.date(1940, 1, 1)
     dates = [d0 + datetime.timedelta(days=k) for k in range(12000)]
@@ -89,7 +97,7 @@ def test_history_404_for_point_outside_lake(lake):
 
 
 def test_query_selects_with_partition_predicates(lake):
-    r = lake.post("/query", json={"sql":
+    r = lake.post("/query", headers=HDR, json={"sql":
         "SELECT year, month, count(*) AS n FROM era5_daily "
         "WHERE year = 1940 AND month = 1 GROUP BY year, month"})
     assert r.status_code == 200
@@ -99,7 +107,7 @@ def test_query_selects_with_partition_predicates(lake):
 
 
 def test_query_manifest_view(lake):
-    r = lake.post("/query", json={"sql": "SELECT count(*) FROM manifest"})
+    r = lake.post("/query", headers=HDR, json={"sql": "SELECT count(*) FROM manifest"})
     assert r.status_code == 200
     assert r.json()["rows"][0][0] == 2
 
@@ -111,4 +119,115 @@ def test_query_manifest_view(lake):
     "COPY era5_daily TO 'x'",
 ])
 def test_query_rejects_non_select(lake, sql):
-    assert lake.post("/query", json={"sql": sql}).status_code == 400
+    assert lake.post("/query", headers=HDR,
+                     json={"sql": sql}).status_code == 400
+
+
+# --- /query is an operator surface, not a public one -------------------------
+
+def test_query_requires_the_internal_token(lake):
+    """No header at all: the caller is not the operator. 401, not a result."""
+    r = lake.post("/query", json={"sql": "SELECT count(*) FROM manifest"})
+    assert r.status_code == 401
+
+
+def test_query_rejects_a_wrong_internal_token(lake):
+    r = lake.post("/query", headers={"X-Thermograph-Internal-Token": "nope"},
+                  json={"sql": "SELECT count(*) FROM manifest"})
+    assert r.status_code == 401
+
+
+def test_query_is_closed_when_no_token_is_provisioned(lake, monkeypatch):
+    """Unprovisioned means the surface does not exist (404) rather than falling
+    open to no-auth — same posture as /internal/* and /api/v2/metrics."""
+    monkeypatch.delenv("THERMOGRAPH_INTERNAL_TOKEN", raising=False)
+    monkeypatch.delenv("THERMOGRAPH_AUTH_SECRET", raising=False)
+    r = lake.post("/query", headers=HDR,
+                  json={"sql": "SELECT count(*) FROM manifest"})
+    assert r.status_code == 404
+
+
+# --- the SELECT guard must reject reads outside the two lake views -----------
+#
+# The old guard was a denylist of DDL/DML *verbs*, so it passed every one of
+# these: DuckDB's reader functions and setting introspection are plain SELECTs.
+# In bucket mode `SET s3_secret_access_key` made the bucket credentials readable
+# straight back out via current_setting() -- the exact thing this service exists
+# to avoid (see the module docstring). These are the attack shapes, not the
+# shapes the old regex happened to be written for.
+
+@pytest.mark.parametrize("sql", [
+    "SELECT * FROM read_csv_auto('/etc/passwd')",
+    "SELECT * FROM read_csv('/etc/passwd')",
+    "SELECT * FROM read_parquet('/etc/passwd')",
+    "SELECT * FROM read_json_auto('/etc/passwd')",
+    "SELECT * FROM read_text('/etc/passwd')",
+    "SELECT * FROM read_blob('/etc/passwd')",
+    "SELECT * FROM glob('/etc/*')",
+    "SELECT * FROM duckdb_settings()",
+    "SELECT * FROM duckdb_secrets()",
+    "SELECT current_setting('s3_secret_access_key')",
+    "SELECT getenv('THERMOGRAPH_LAKE_S3_SECRET')",
+    "SELECT * FROM 'file:///etc/passwd'",
+    "SELECT * FROM '/etc/passwd'",
+    # Nested one level down, so a prefix-only check doesn't pass it.
+    "SELECT * FROM (SELECT * FROM read_csv_auto('/etc/passwd'))",
+    "WITH x AS (SELECT * FROM glob('/etc/*')) SELECT * FROM x",
+])
+def test_query_rejects_reads_outside_the_lake_views(lake, sql):
+    r = lake.post("/query", headers=HDR, json={"sql": sql})
+    assert r.status_code == 400, f"guard let this through: {sql}"
+    # Assert the GUARD refused it, not that DuckDB happened to choke on the file
+    # format. Several of these (read_parquet on a text file, a bare '/etc/passwd')
+    # error in the engine anyway, so a status-only assertion would still pass with
+    # the guard deleted -- exactly the "negatives shaped like the regex" trap the
+    # original four-case test fell into.
+    detail = r.json()["detail"]
+    assert not detail.startswith("query failed"), (
+        f"rejected by the engine, not the guard: {sql} -> {detail}")
+
+
+def test_query_still_allows_a_block_comment_inside_a_real_select(lake):
+    """Defence in depth must not become a false positive: an inline comment in
+    an otherwise ordinary query is fine."""
+    r = lake.post("/query", headers=HDR,
+                  json={"sql": "SELECT count(*) /* rows */ FROM manifest"})
+    assert r.status_code == 200
+
+
+def test_bucket_mode_uses_a_secret_not_a_readable_setting(monkeypatch, tmp_path):
+    """Regression guard for the credential-disclosure shape: the S3 credentials
+    must be installed as a DuckDB SECRET (opaque to current_setting) rather than
+    via SET s3_secret_access_key, which hands them back to any caller."""
+    import lake_app
+
+    statements = []
+
+    class _FakeCon:
+        def execute(self, sql, *args):
+            statements.append(sql)
+            return self
+
+        def fetchone(self):
+            return ("s3://b/iceberg/era5_daily/metadata/1.metadata.json",)
+
+        def close(self):
+            pass
+
+    monkeypatch.setattr(lake_app, "_source", lambda: ("bucket", {
+        "endpoint": "https://example.com", "bucket": "b", "region": "r",
+        "access_key": "AK", "secret_key": "SUPER-SECRET-KEY"}))
+    monkeypatch.setitem(__import__("sys").modules, "duckdb",
+                        type("m", (), {"connect": staticmethod(lambda: _FakeCon())}))
+
+    lake_app._connect()
+    joined = "\n".join(statements)
+    assert "SECRET" in joined.upper() and "TYPE S3" in joined.upper()
+    assert "SET s3_secret_access_key" not in joined
+    assert "SET s3_access_key_id" not in joined
+    # The key material goes as a bound parameter, never interpolated into SQL.
+    assert "SUPER-SECRET-KEY" not in joined
+    # The local filesystem is unreachable in bucket mode, and the caller cannot
+    # turn it back on.
+    assert "disabled_filesystems" in joined
+    assert "lock_configuration" in joined