fix(token-db): rewrite TokenDB.execute to speak libsql Hrana v2 protocol (#61)

bk86a · claude · bk86a · commit 68b3f494950f · 2026-04-29T18:37:20.000+02:00
The v0.17.0 TokenDB assumed a generic POST /query with {sql, params} body —
that's not the wire protocol the project's actual managed-database provider
uses. Replace with libsql/Hrana v2:

- POST {base}/v2/pipeline (with libsql:// rewritten to https://)
- Wrap statements in {requests: [{type: "execute", stmt: {sql, args}}]}
- Encode params as typed value objects {type: "integer"|"text"|..., value}
- Decode rows from arrays of typed value objects back to {col: pyvalue} dicts
- Bearer token in Authorization header (new PC2NUTS_TOKEN_DB_AUTH_TOKEN env var)
- New --auth-token CLI flag

Verified end-to-end against the production database: SELECT round-trips
typed values correctly, init_schema is idempotent, add/revoke/list_active/
list_all all work, RETURNING id is honoured. The unit tests now mock the
Hrana wire shape and stay fast (172 tests pass in 1.4s).

CLI tests' lambda for _make_db extended to accept the new auth_token kwarg.
Bandit B107 false positive on `auth_token: str = ""` suppressed with a
clearly-explained nosec comment.

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/app/config.py b/app/config.py
@@ -20,6 +20,7 @@ class Settings(BaseSettings):
     extra_sources: str = ""
     trusted_tokens_raw: str = Field(default="", validation_alias="PC2NUTS_TRUSTED_TOKENS")
     token_db_url: str = ""
+    token_db_auth_token: str = ""
     token_refresh_seconds: int = Field(default=60, ge=1)
     rate_limit: str = _defaults.get("rate_limit", "60/minute")
     rate_limit_headers: bool = _defaults.get("rate_limit_headers", True)
diff --git a/app/main.py b/app/main.py
@@ -108,7 +108,10 @@ async def lifespan(app: FastAPI):
         from app import auth as auth_mod
         from app.token_db import TokenDB
 
-        token_db = TokenDB(_config.settings.token_db_url)
+        token_db = TokenDB(
+            _config.settings.token_db_url,
+            auth_token=_config.settings.token_db_auth_token,
+        )
 
         async def _refresh_loop():
             interval = max(1, _config.settings.token_refresh_seconds)
diff --git a/app/token_db.py b/app/token_db.py
@@ -1,10 +1,17 @@
-"""Thin HTTP client for the configured SQLite-compatible managed database.
+"""Thin HTTP client for a libsql / SQLite-over-HTTP managed database.
 
 See docs/superpowers/specs/2026-04-29-db-backed-trusted-tokens-design.md.
 
-The wire shape (POST /query with {sql, params}, response {rows, rowsAffected,
-lastInsertRowId}) is the working assumption; adjust `execute` if the
-configured provider differs.
+Wire protocol: Hrana v2 (libsql HTTP). Endpoint is ``POST {base}/v2/pipeline``
+where {base} is the database URL with ``libsql://`` rewritten to ``https://``.
+The single-statement request is wrapped in ``{requests: [{type: "execute",
+stmt: {sql, args}}]}``; rows are returned as arrays of typed value objects
+``{type, value}`` with one row per query result, alongside a ``cols`` list
+that gives column names. This adapter converts back and forth between
+Python-native types and the typed-value envelope so callers get plain dicts.
+
+Authentication: a Bearer token in the Authorization header. Both URL and
+token come from environment configuration; no values are committed.
 """
 
 from __future__ import annotations
@@ -18,31 +25,91 @@ class TokenDBError(Exception):
     """Raised when the token database returns an error or is unreachable."""
 
 
+def _to_libsql_arg(v: Any) -> dict:
+    """Encode a Python value as a libsql typed value object."""
+    if v is None:
+        return {"type": "null"}
+    if isinstance(v, bool):
+        # libsql has no native bool; SQLite stores 0/1
+        return {"type": "integer", "value": "1" if v else "0"}
+    if isinstance(v, int):
+        return {"type": "integer", "value": str(v)}
+    if isinstance(v, float):
+        return {"type": "float", "value": v}
+    if isinstance(v, (bytes, bytearray)):
+        import base64
+
+        return {"type": "blob", "base64": base64.b64encode(bytes(v)).decode("ascii")}
+    return {"type": "text", "value": str(v)}
+
+
+def _from_libsql_value(cell: dict) -> Any:
+    """Decode a libsql typed value object back to a Python value."""
+    t = cell.get("type")
+    if t == "null":
+        return None
+    if t == "integer":
+        return int(cell["value"])
+    if t == "float":
+        return float(cell["value"])
+    if t == "blob":
+        import base64
+
+        return base64.b64decode(cell["base64"])
+    # text and unknown types fall through as strings
+    return cell.get("value")
+
+
 class TokenDB:
-    """Minimal HTTP client over a SQLite-compatible managed database.
+    """Minimal HTTP client over a libsql managed database.
 
     All methods are blocking. Callers running inside an asyncio loop should
     wrap calls in asyncio.to_thread().
     """
 
-    def __init__(self, url: str) -> None:
-        self.url = url.rstrip("/")
+    def __init__(self, url: str, auth_token: str = "") -> None:  # nosec B107 — empty default means "no auth", not a hardcoded credential
+        # libsql://host/ → https://host (drop trailing slash, swap scheme)
+        u = url.strip()
+        if u.startswith("libsql://"):
+            u = "https://" + u[len("libsql://") :]
+        self.url = u.rstrip("/")
+        self.auth_token = auth_token
 
     def execute(self, sql: str, params: list[Any] | None = None) -> list[dict]:
-        """Send a SQL statement. Returns the `rows` list from the response.
+        """Send a SQL statement. Returns the result rows as a list of dicts.
 
-        Writes (INSERT/UPDATE/DELETE/CREATE) typically return an empty list
-        unless the SQL uses RETURNING.
+        Each dict maps column name → Python-native value. Writes (INSERT/UPDATE/
+        DELETE/CREATE) typically return an empty list unless the SQL uses RETURNING.
+
+        Raises TokenDBError on transport failure or libsql-reported error.
         """
-        payload = {"sql": sql, "params": list(params) if params else []}
+        stmt: dict = {"sql": sql}
+        if params:
+            stmt["args"] = [_to_libsql_arg(p) for p in params]
+        payload = {"requests": [{"type": "execute", "stmt": stmt}]}
+        headers = {}
+        if self.auth_token:
+            headers["Authorization"] = f"Bearer {self.auth_token}"
+
         try:
             with httpx.Client(timeout=10) as client:
-                resp = client.post(f"{self.url}/query", json=payload, headers={})
+                resp = client.post(f"{self.url}/v2/pipeline", json=payload, headers=headers)
                 resp.raise_for_status()
                 body = resp.json()
         except (httpx.HTTPError, ValueError) as exc:
             raise TokenDBError(f"DB request failed: {exc}") from exc
-        return body.get("rows") or []
+
+        results = body.get("results") or []
+        if not results:
+            return []
+        first = results[0]
+        if first.get("type") != "ok":
+            err = (first.get("error") or {}).get("message", "unknown error")
+            raise TokenDBError(f"DB statement failed: {err}")
+        result = (first.get("response") or {}).get("result") or {}
+        cols = [c.get("name") for c in (result.get("cols") or [])]
+        rows = result.get("rows") or []
+        return [{cols[i]: _from_libsql_value(cell) for i, cell in enumerate(row)} for row in rows]
 
     # ── Schema ──────────────────────────────────────────────────────────────
 
diff --git a/scripts/tokens.py b/scripts/tokens.py
@@ -18,9 +18,9 @@
 from app.token_db import TokenDB, TokenDBError
 
 
-def _make_db(url: str) -> TokenDB:
+def _make_db(url: str, auth_token: str = "") -> TokenDB:
     """Indirection seam for tests."""
-    return TokenDB(url)
+    return TokenDB(url, auth_token=auth_token)
 
 
 def _token_id(token: str) -> str:
@@ -35,6 +35,12 @@ def _resolve_db_url(args_url: str | None) -> str | None:
     return env or None
 
 
+def _resolve_auth_token(args_token: str | None) -> str:
+    if args_token:
+        return args_token
+    return os.environ.get("PC2NUTS_TOKEN_DB_AUTH_TOKEN", "").strip()
+
+
 def _cmd_init(db: TokenDB) -> int:
     db.init_schema()
     print("Schema initialised (idempotent).")
@@ -95,6 +101,10 @@ def _cmd_revoke(db: TokenDB, token_id_arg: int) -> int:
 def main(argv: Sequence[str] | None = None) -> int:
     parser = argparse.ArgumentParser(prog="scripts.tokens", description=__doc__)
     parser.add_argument("--db-url", help="Override PC2NUTS_TOKEN_DB_URL")
+    parser.add_argument(
+        "--auth-token",
+        help="Override PC2NUTS_TOKEN_DB_AUTH_TOKEN (Bearer token for the DB).",
+    )
     sub = parser.add_subparsers(dest="cmd", required=True)
 
     sub.add_parser("init", help="Create the trusted_tokens table (idempotent)")
@@ -123,7 +133,8 @@ def main(argv: Sequence[str] | None = None) -> int:
         )
         return 2
 
-    db = _make_db(url)
+    auth_token = _resolve_auth_token(args.auth_token)
+    db = _make_db(url, auth_token=auth_token)
 
     if args.cmd == "init":
         return _cmd_init(db)
diff --git a/tests/test_token_db.py b/tests/test_token_db.py
@@ -55,6 +55,24 @@ def test_init_strips_trailing_slash(self):
         db = TokenDB("https://db.example/v1/")
         assert db.url == "https://db.example/v1"
 
+    def test_init_rewrites_libsql_scheme(self):
+        from app.token_db import TokenDB
+
+        db = TokenDB("libsql://db.example/")
+        assert db.url == "https://db.example"
+
+    def test_init_stores_auth_token(self):
+        from app.token_db import TokenDB
+
+        db = TokenDB("https://db.example", auth_token="jwt-xyz")
+        assert db.auth_token == "jwt-xyz"
+
+    def test_init_default_auth_token_empty(self):
+        from app.token_db import TokenDB
+
+        db = TokenDB("https://db.example")
+        assert db.auth_token == ""
+
 
 class TestTokenDBExecute:
     def _mock_response(self, monkeypatch, *, json_body: dict, status_code: int = 200):
@@ -87,33 +105,90 @@ def _post(self, url, json=None, headers=None, timeout=None):
         monkeypatch.setattr(httpx.Client, "post", _post)
         return captured
 
-    def test_execute_sends_post_with_sql_and_params(self, monkeypatch):
+    def _ok(self, rows: list[list[dict]] | None = None, cols: list[str] | None = None) -> dict:
+        """Build a Hrana v2 success-response envelope."""
+        return {
+            "results": [
+                {
+                    "type": "ok",
+                    "response": {
+                        "type": "execute",
+                        "result": {
+                            "cols": [{"name": c} for c in (cols or [])],
+                            "rows": rows or [],
+                        },
+                    },
+                }
+            ]
+        }
+
+    def test_execute_posts_to_v2_pipeline(self, monkeypatch):
         from app.token_db import TokenDB
 
-        captured = self._mock_response(monkeypatch, json_body={"rows": [], "rowsAffected": 0})
+        captured = self._mock_response(monkeypatch, json_body=self._ok())
+        db = TokenDB("https://db.example/v1")
+        db.execute("SELECT 1")
+        assert captured["url"] == "https://db.example/v1/v2/pipeline"
+
+    def test_execute_wraps_sql_and_params_in_hrana(self, monkeypatch):
+        from app.token_db import TokenDB
+
+        captured = self._mock_response(monkeypatch, json_body=self._ok())
         db = TokenDB("https://db.example/v1")
         db.execute("SELECT * FROM t WHERE id = ?", [42])
 
-        assert captured["json"] == {"sql": "SELECT * FROM t WHERE id = ?", "params": [42]}
+        assert captured["json"] == {
+            "requests": [
+                {
+                    "type": "execute",
+                    "stmt": {
+                        "sql": "SELECT * FROM t WHERE id = ?",
+                        "args": [{"type": "integer", "value": "42"}],
+                    },
+                }
+            ]
+        }
 
-    def test_execute_returns_rows(self, monkeypatch):
+    def test_execute_passes_auth_token_in_header(self, monkeypatch):
         from app.token_db import TokenDB
 
-        self._mock_response(
-            monkeypatch,
-            json_body={"rows": [{"id": 1, "label": "a"}], "rowsAffected": 0},
+        captured = self._mock_response(monkeypatch, json_body=self._ok())
+        db = TokenDB("https://db.example/v1", auth_token="jwt-xyz")
+        db.execute("SELECT 1")
+        assert captured["headers"].get("Authorization") == "Bearer jwt-xyz"
+
+    def test_execute_no_auth_header_when_token_empty(self, monkeypatch):
+        from app.token_db import TokenDB
+
+        captured = self._mock_response(monkeypatch, json_body=self._ok())
+        db = TokenDB("https://db.example/v1")
+        db.execute("SELECT 1")
+        assert "Authorization" not in (captured["headers"] or {})
+
+    def test_execute_unwraps_hrana_rows_to_dicts(self, monkeypatch):
+        from app.token_db import TokenDB
+
+        body = self._ok(
+            cols=["id", "label"],
+            rows=[
+                [{"type": "integer", "value": "1"}, {"type": "text", "value": "a"}],
+                [{"type": "integer", "value": "2"}, {"type": "null"}],
+            ],
         )
+        self._mock_response(monkeypatch, json_body=body)
         db = TokenDB("https://db.example/v1")
         rows = db.execute("SELECT id, label FROM t")
-        assert rows == [{"id": 1, "label": "a"}]
+        assert rows == [{"id": 1, "label": "a"}, {"id": 2, "label": None}]
 
-    def test_execute_no_params_sends_empty_list(self, monkeypatch):
+    def test_execute_no_params_omits_args(self, monkeypatch):
         from app.token_db import TokenDB
 
-        captured = self._mock_response(monkeypatch, json_body={"rows": [], "rowsAffected": 0})
+        captured = self._mock_response(monkeypatch, json_body=self._ok())
         db = TokenDB("https://db.example/v1")
         db.execute("CREATE TABLE x (id INTEGER)")
-        assert captured["json"]["params"] == []
+        # When no params, the stmt should not have an "args" key (or have an empty list).
+        stmt = captured["json"]["requests"][0]["stmt"]
+        assert stmt.get("args", []) == []
 
     def test_execute_http_error_raises_token_db_error(self, monkeypatch):
         from app.token_db import TokenDB, TokenDBError
@@ -123,6 +198,18 @@ def test_execute_http_error_raises_token_db_error(self, monkeypatch):
         with pytest.raises(TokenDBError):
             db.execute("SELECT 1")
 
+    def test_execute_libsql_error_raises_token_db_error(self, monkeypatch):
+        from app.token_db import TokenDB, TokenDBError
+
+        body = {
+            "results": [{"type": "error", "error": {"message": "UNIQUE constraint failed", "code": "..."}}]
+        }
+        self._mock_response(monkeypatch, json_body=body)
+        db = TokenDB("https://db.example/v1")
+        with pytest.raises(TokenDBError) as exc_info:
+            db.execute("INSERT ...")
+        assert "UNIQUE" in str(exc_info.value)
+
 
 # ── TokenDB query methods ────────────────────────────────────────────────────
 
diff --git a/tests/test_tokens_cli.py b/tests/test_tokens_cli.py
@@ -46,7 +46,7 @@ def fake_db(monkeypatch):
     from scripts import tokens
 
     fake = FakeTokenDB()
-    monkeypatch.setattr(tokens, "_make_db", lambda url: fake)
+    monkeypatch.setattr(tokens, "_make_db", lambda url, auth_token="": fake)
     monkeypatch.setenv("PC2NUTS_TOKEN_DB_URL", "fake://")
     return fake
 
@@ -210,7 +210,7 @@ def test_db_url_arg_overrides_env(monkeypatch, capsys):
     monkeypatch.setattr(
         tokens,
         "_make_db",
-        lambda url: (captured_urls.append(url), FakeTokenDB(url))[1],
+        lambda url, auth_token="": (captured_urls.append(url), FakeTokenDB(url))[1],
     )
     rc = main(["--db-url", "https://override.example", "init"])
     assert rc == 0
