Round-trip binary primary key values in row URLs (#2419)

path_from_row_pks encoded a bytes (BLOB) primary key with str(bit), which
isn't reversible, and the row lookup decoded every URL component to text, so a
row with a binary primary key produced a 404. Encode bytes losslessly with a
new tilde_encode_bytes, add tilde_decode_to_bytes, and decode BLOB primary key
components back to bytes when looking up the row (non-UTF-8 values shown as
hex). Adds regression tests.

datasette/app.py

@@ -2819,8 +2819,9 @@ class Datasette:
 
     async def resolve_row(self, request):
         db, table_name, _ = await self.resolve_table(request)
-        pk_values = urlsafe_components(request.url_vars["pks"])
-        sql, params, pks = await row_sql_params_pks(db, table_name, pk_values)
+        sql, params, pks, pk_values = await row_sql_params_pks(
+            db, table_name, request.url_vars["pks"]
+        )
         if len(pk_values) != len(pks):
             raise BadRequest(
                 "URL row identifier does not match the primary key for this table"

datasette/utils/__init__.py

@@ -192,7 +192,14 @@ def path_from_row_pks(row, pks, use_rowid, quote=True):
             row[pk]["value"] if isinstance(row[pk], dict) else row[pk] for pk in pks
         ]
     if quote:
-        bits = [tilde_encode(str(bit)) for bit in bits]
+        bits = [
+            (
+                tilde_encode_bytes(bit)
+                if isinstance(bit, bytes)
+                else tilde_encode(str(bit))
+            )
+            for bit in bits
+        ]
     else:
         bits = [str(bit) for bit in bits]
 
@@ -1417,14 +1424,27 @@ def tilde_encode(s: str) -> str:
     return "".join(_tilde_encoder(char) for char in s.encode("utf-8"))
 
 
+def tilde_encode_bytes(b: bytes) -> str:
+    "Tilde-encodes raw bytes, used for binary (BLOB) primary key values"
+    return "".join(_tilde_encoder(byte) for byte in b)
+
+
 @documented
 def tilde_decode(s: str) -> str:
     "Decodes a tilde-encoded string, so ``~2Ffoo~2Fbar`` -> ``/foo/bar``"
+    return tilde_decode_to_bytes(s).decode("utf-8")
+
+
+def tilde_decode_to_bytes(s: str) -> bytes:
+    """Decodes a tilde-encoded string to raw bytes, preserving non-UTF-8 data.
+
+    Used to round-trip binary (BLOB) primary key values through row URLs."""
     # Avoid accidentally decoding a %2f style sequence
     temp = secrets.token_hex(16)
     s = s.replace("%", temp)
-    decoded = urllib.parse.unquote_plus(s.replace("~", "%"))
-    return decoded.replace(temp, "%")
+    # "+" encodes a space (as with urllib.parse.unquote_plus)
+    decoded = urllib.parse.unquote_to_bytes(s.replace("~", "%").replace("+", " "))
+    return decoded.replace(temp.encode("ascii"), b"%")
 
 
 def resolve_routes(routes, path):
@@ -1445,19 +1465,40 @@ def truncate_url(url, length):
     return url[: length - 1] + "…"
 
 
-async def row_sql_params_pks(db, table, pk_values):
+async def row_sql_params_pks(db, table, pks_string):
     pks = await db.primary_keys(table)
     use_rowid = not pks
     select = "*"
     if use_rowid:
         select = "rowid, *"
         pks = ["rowid"]
+    # BLOB primary keys need their query parameter as bytes so binary values
+    # match; the returned pk_values stay as text for display / identity.
+    blob_pks = set()
+    if not use_rowid:
+        blob_pks = {
+            column.name
+            for column in await db.table_column_details(table)
+            if column.name in pks and (column.type or "").upper() == "BLOB"
+        }
+    pk_values = []
+    param_values = []
+    for i, component in enumerate(pks_string.split(",")):
+        if i < len(pks) and pks[i] in blob_pks:
+            raw = tilde_decode_to_bytes(component)
+            param_values.append(raw)
+            try:
+                pk_values.append(raw.decode("utf-8"))
+            except UnicodeDecodeError:
+                pk_values.append(raw.hex())
+        else:
+            value = tilde_decode(component)
+            param_values.append(value)
+            pk_values.append(value)
     wheres = [f'"{pk}"=:p{i}' for i, pk in enumerate(pks)]
     sql = f"select {select} from {escape_sqlite(table)} where {' AND '.join(wheres)}"
-    params = {}
-    for i, pk_value in enumerate(pk_values):
-        params[f"p{i}"] = pk_value
-    return sql, params, pks
+    params = {f"p{i}": value for i, value in enumerate(param_values)}
+    return sql, params, pks, pk_values
 
 
 def _handle_pair(key: str, value: str) -> dict:

tests/test_api_write.py

@@ -1,6 +1,6 @@
 from datasette.app import Datasette
 from datasette.events import RenameTableEvent
-from datasette.utils import error_body, escape_sqlite, sqlite3
+from datasette.utils import error_body, escape_sqlite, path_from_row_pks, sqlite3
 from .utils import last_event
 import pytest
 import time
@@ -2717,3 +2717,27 @@ async def test_create_using_alter_against_existing_table(
         insert_rows_event = ds_write._tracked_events[1]
         assert insert_rows_event.name == "insert-rows"
         assert insert_rows_event.num_rows == 1
+
+
+@pytest.mark.asyncio
+async def test_row_page_with_binary_primary_key(ds_write):
+    # Regression test for #2419 - a row with a binary (BLOB) primary key value
+    # should be reachable via its generated URL instead of returning a 404.
+    db = ds_write.get_database("data")
+    await db.execute_write(
+        "create table binary_pk (id integer, term blob, primary key (id, term))"
+    )
+    terms = (b"0thei'", b"\xff\x00abc")
+    for term in terms:
+        await db.execute_write(
+            "insert into binary_pk (id, term) values (?, ?)", [3, term]
+        )
+    for term in terms:
+        path = path_from_row_pks(
+            {"id": 3, "term": term}, ["id", "term"], use_rowid=False
+        )
+        html = await ds_write.client.get(f"/data/binary_pk/{path}")
+        assert html.status_code == 200, html.text
+        js = await ds_write.client.get(f"/data/binary_pk/{path}.json?_shape=array")
+        assert js.status_code == 200, js.text
+        assert len(js.json()) == 1

tests/test_utils.py

@@ -803,6 +803,17 @@ def test_tilde_encoding(original, expected):
     assert original == utils.tilde_decode(actual)
 
 
+@pytest.mark.parametrize(
+    "value",
+    (b"0thei'", b"\xff\x00abc", b"hello", b"", b"a,b~c+d"),
+)
+def test_binary_pk_roundtrips_through_path(value):
+    # A binary (BLOB) primary key value survives path_from_row_pks and can be
+    # decoded back to the exact bytes - regression test for #2419
+    encoded = utils.path_from_row_pks({"pk": value}, ["pk"], False)
+    assert utils.tilde_decode_to_bytes(encoded) == value
+
+
 @pytest.mark.parametrize(
     "url,length,expected",
     (