Demote read_context to pure retrieval; RLS user_id hardening

db/pg_queries/node_memory.py

@@ -110,46 +110,83 @@ async def log_node_read(
     *,
     conversation_id: str | None = None,
     title: str | None = None,
+    user_id: str | None = None,
 ) -> None:
     """Record that the current user read node_id at level_ordinal in this conversation.
 
     Called by read_context and read_node_memory on every access.
     conversation_id may be None for admin/debug calls without conversation context.
+
+    user_id: RLS hardening — explicit caller-supplied user_id, inserted directly
+    instead of relying solely on the `app.current_user_id` session GUC. When
+    None (default), falls back to the session GUC as before, so this is
+    backward compatible for any caller that hasn't been updated to bind it.
     """
     conv_uuid = _uuid.UUID(conversation_id) if conversation_id else None
-    await conn.execute(
-        """
-        INSERT INTO node_read_log (user_id, conversation_id, node_id, level_ordinal, title)
-        VALUES (
-            current_setting('app.current_user_id', true)::uuid,
-            $1, $2::uuid, $3, $4
+    if user_id is not None:
+        await conn.execute(
+            """
+            INSERT INTO node_read_log (user_id, conversation_id, node_id, level_ordinal, title)
+            VALUES ($1::uuid, $2, $3::uuid, $4, $5)
+            """,
+            _uuid.UUID(user_id), conv_uuid, _uuid.UUID(node_id), level_ordinal, title,
+        )
+    else:
+        await conn.execute(
+            """
+            INSERT INTO node_read_log (user_id, conversation_id, node_id, level_ordinal, title)
+            VALUES (
+                current_setting('app.current_user_id', true)::uuid,
+                $1, $2::uuid, $3, $4
+            )
+            """,
+            conv_uuid, _uuid.UUID(node_id), level_ordinal, title,
         )
-        """,
-        conv_uuid, _uuid.UUID(node_id), level_ordinal, title,
-    )
 
 
 async def has_read_node_in_conversation(
     conn: asyncpg.Connection,
     node_id: str,
     conversation_id: str,
+    *,
+    user_id: str | None = None,
 ) -> bool:
     """Return True if any read of node_id was logged for this conversation.
 
     Used by write_node_memory advisory read-before-write check:
       v1: log WARNING if False but allow the write
       v2: hard block if False once Stream C callers always provide conversation_id
+
+    user_id: RLS hardening — when provided, the query binds this value
+    directly as an explicit `WHERE user_id = $N` filter instead of relying
+    on the `app.current_user_id` session GUC. This makes the scoping correct
+    even on an "unscoped" connection (no `SET LOCAL app.current_user_id`
+    applied) where the GUC-based filter would either be wrong or raise a
+    cast error on an empty string. When None (default), behavior is
+    unchanged (session-GUC only) for backward compatibility.
     """
-    count = await conn.fetchval(
-        """
-        SELECT COUNT(*)
-        FROM node_read_log
-        WHERE user_id         = current_setting('app.current_user_id', true)::uuid
-          AND conversation_id = $1::uuid
-          AND node_id         = $2::uuid
-        """,
-        _uuid.UUID(conversation_id), _uuid.UUID(node_id),
-    )
+    if user_id is not None:
+        count = await conn.fetchval(
+            """
+            SELECT COUNT(*)
+            FROM node_read_log
+            WHERE user_id         = $1::uuid
+              AND conversation_id = $2::uuid
+              AND node_id         = $3::uuid
+            """,
+            _uuid.UUID(user_id), _uuid.UUID(conversation_id), _uuid.UUID(node_id),
+        )
+    else:
+        count = await conn.fetchval(
+            """
+            SELECT COUNT(*)
+            FROM node_read_log
+            WHERE user_id         = current_setting('app.current_user_id', true)::uuid
+              AND conversation_id = $1::uuid
+              AND node_id         = $2::uuid
+            """,
+            _uuid.UUID(conversation_id), _uuid.UUID(node_id),
+        )
     return (count or 0) > 0
 
 
@@ -207,19 +244,38 @@ async def get_node_tree_distance(
 async def get_conversation_reads(
     conn: asyncpg.Connection,
     conversation_id: str,
+    *,
+    user_id: str | None = None,
 ) -> list[dict]:
     """Return all node read credits for a conversation (newest first).
 
     Used for diagnostics and advisory enforcement reports.
+
+    user_id: RLS hardening — when provided, binds this value directly as an
+    explicit filter instead of relying on the session GUC (see
+    has_read_node_in_conversation for the full rationale). When None
+    (default), behavior is unchanged (session-GUC only).
     """
-    rows = await conn.fetch(
-        """
-        SELECT node_id::text, level_ordinal, title, read_at
-        FROM node_read_log
-        WHERE user_id         = current_setting('app.current_user_id', true)::uuid
-          AND conversation_id = $1::uuid
-        ORDER BY read_at DESC
-        """,
-        _uuid.UUID(conversation_id),
-    )
+    if user_id is not None:
+        rows = await conn.fetch(
+            """
+            SELECT node_id::text, level_ordinal, title, read_at
+            FROM node_read_log
+            WHERE user_id         = $1::uuid
+              AND conversation_id = $2::uuid
+            ORDER BY read_at DESC
+            """,
+            _uuid.UUID(user_id), _uuid.UUID(conversation_id),
+        )
+    else:
+        rows = await conn.fetch(
+            """
+            SELECT node_id::text, level_ordinal, title, read_at
+            FROM node_read_log
+            WHERE user_id         = current_setting('app.current_user_id', true)::uuid
+              AND conversation_id = $1::uuid
+            ORDER BY read_at DESC
+            """,
+            _uuid.UUID(conversation_id),
+        )
     return [dict(r) for r in rows]

db/pg_queries/nodes.py

@@ -47,10 +47,28 @@ async def create_node(
     return _node(row)
 
 
-async def get_node(conn: asyncpg.Connection, node_id: str) -> dict | None:
-    row = await conn.fetchrow(
-        "SELECT * FROM context_nodes WHERE id = $1", _uuid.UUID(node_id)
-    )
+async def get_node(
+    conn: asyncpg.Connection,
+    node_id: str,
+    *,
+    user_id: str | None = None,
+) -> dict | None:
+    """Fetch a single context node by id.
+
+    user_id: RLS hardening — when provided, binds this value directly as an
+    explicit `AND user_id = $N` filter instead of relying solely on
+    RLS/the session GUC. When None (default), behavior is unchanged
+    (RLS-only), backward compatible.
+    """
+    if user_id is not None:
+        row = await conn.fetchrow(
+            "SELECT * FROM context_nodes WHERE id = $1 AND user_id = $2::uuid",
+            _uuid.UUID(node_id), _uuid.UUID(user_id),
+        )
+    else:
+        row = await conn.fetchrow(
+            "SELECT * FROM context_nodes WHERE id = $1", _uuid.UUID(node_id)
+        )
     if not row:
         return None
     d = _node(row)
@@ -65,21 +83,47 @@ async def get_node(conn: asyncpg.Connection, node_id: str) -> dict | None:
     return d
 
 
-async def get_node_by_path(conn: asyncpg.Connection, path: str) -> dict | None:
+async def get_node_by_path(
+    conn: asyncpg.Connection,
+    path: str,
+    *,
+    user_id: str | None = None,
+) -> dict | None:
+    """Resolve a slash-separated path (e.g. "Projects/Tether") to a node.
+
+    user_id: RLS hardening — when provided, each step's lookup binds this
+    value directly as an explicit `AND user_id = $N` filter instead of
+    relying solely on RLS/the session GUC. This makes path resolution
+    correct even on an unscoped connection. When None (default), behavior
+    is unchanged (RLS-only).
+    """
     parts = [p for p in path.split("/") if p]
     if not parts:
         return None
-    row = await conn.fetchrow(
-        "SELECT id FROM context_nodes WHERE parent_id IS NULL AND name = $1", parts[0]
-    )
+    uid = _uuid.UUID(user_id) if user_id is not None else None
+    if uid is not None:
+        row = await conn.fetchrow(
+            "SELECT id FROM context_nodes WHERE parent_id IS NULL AND name = $1 AND user_id = $2::uuid",
+            parts[0], uid,
+        )
+    else:
+        row = await conn.fetchrow(
+            "SELECT id FROM context_nodes WHERE parent_id IS NULL AND name = $1", parts[0]
+        )
     if not row:
         return None
     resolved_id = row["id"]
     for segment in parts[1:]:
-        row = await conn.fetchrow(
-            "SELECT id FROM context_nodes WHERE parent_id = $1 AND name = $2",
-            resolved_id, segment,
-        )
+        if uid is not None:
+            row = await conn.fetchrow(
+                "SELECT id FROM context_nodes WHERE parent_id = $1 AND name = $2 AND user_id = $3::uuid",
+                resolved_id, segment, uid,
+            )
+        else:
+            row = await conn.fetchrow(
+                "SELECT id FROM context_nodes WHERE parent_id = $1 AND name = $2",
+                resolved_id, segment,
+            )
         if not row:
             return None
         resolved_id = row["id"]

tests/db/test_pg_nodes.py

@@ -1,7 +1,7 @@
 """Tests for db/pg_queries/nodes.py — tree traversal, move, cycle detection."""
 import pytest
 
-from tests.db.pg_conftest import conn, TEST_USER_ID  # noqa: F401
+from tests.db.pg_conftest import conn, auth_conn, TEST_USER_ID  # noqa: F401
 from db.pg_queries.nodes import (
     create_node, get_node, get_node_by_path, get_children,
     ensure_node_path, get_all_node_paths, get_subtree,
@@ -111,3 +111,105 @@ async def test_get_all_node_paths(conn):
     await ensure_node_path(conn, "AllPaths/Sub/Leaf")
     paths = await get_all_node_paths(conn)
     assert any("AllPaths" in p for p in paths)
+
+
+# ---------------------------------------------------------------------------
+# RLS hardening (0e addendum / brief-1a) — real unscoped-connection tests.
+#
+# `auth_conn` (tests/db/pg_conftest.py) never sets `app.current_user_id`, so
+# these exercise the "defense in depth on an unscoped connection" case the
+# hardening addendum targets — unlike the `conn` fixture, which always has
+# the GUC set and so can never distinguish "explicit bind" from "GUC
+# happened to match".
+#
+# NOTE on local sandbox limits: the local dev Postgres role connects with
+# BYPASSRLS (confirmed: `rolbypassrls=True` for the `tether` role used by
+# DATABASE_URL here — see the pre-existing, out-of-scope failure of
+# tests/db/test_rls.py::test_app_db_role_is_not_superuser, which asserts
+# prod's app role is NOT superuser/bypassrls). That means RLS itself cannot
+# be exercised as a *deny* mechanism in this sandbox — a query with no
+# explicit user_id and no GUC will still see the row here, whereas in prod
+# (non-bypassing app role) RLS would deny it. What CAN be verified here,
+# independent of RLS/role privileges, is the actual code path this
+# hardening adds: an explicit user_id bind is a real SQL filter, so a
+# mismatched user_id must never resolve someone else's node, and a matching
+# explicit user_id must resolve it. That's what these tests assert.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_get_node_by_path_explicit_user_id_resolves_own_node(auth_conn):
+    await auth_conn.execute(
+        "INSERT INTO context_nodes (id, user_id, parent_id, name) "
+        "VALUES (gen_random_uuid(), $1::uuid, NULL, $2)",
+        uuid.UUID(TEST_USER_ID), "ExplicitBindRoot",
+    )
+    found = await get_node_by_path(auth_conn, "ExplicitBindRoot", user_id=TEST_USER_ID)
+    assert found is not None
+    assert found["name"] == "ExplicitBindRoot"
+
+
+@pytest.mark.asyncio
+async def test_get_node_by_path_mismatched_user_id_returns_none(auth_conn):
+    """Explicit user_id binding must actually filter — a mismatched user_id
+    must not resolve a node that belongs to someone else."""
+    other_user_id = str(uuid.uuid4())
+    await auth_conn.execute(
+        "INSERT INTO users (id, username, email, password_hash, is_admin) "
+        "VALUES ($1::uuid, 'nodes_other', 'nodes_other@example.com', 'x', false) "
+        "ON CONFLICT DO NOTHING",
+        other_user_id,
+    )
+    await auth_conn.execute(
+        "INSERT INTO context_nodes (id, user_id, parent_id, name) "
+        "VALUES (gen_random_uuid(), $1::uuid, NULL, $2)",
+        uuid.UUID(TEST_USER_ID), "MismatchRoot",
+    )
+    result = await get_node_by_path(auth_conn, "MismatchRoot", user_id=other_user_id)
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_get_node_explicit_user_id_resolves_own_node(auth_conn):
+    """get_node gains the same optional user_id explicit-bind hardening as
+    get_node_by_path — brief-1a review finding: id-based lookups (used by
+    read_context's node_ids branch and the children section_types/
+    children_count backfill) previously had no user_id param at all."""
+    node_id = str(uuid.uuid4())
+    await auth_conn.execute(
+        "INSERT INTO context_nodes (id, user_id, parent_id, name) "
+        "VALUES ($1::uuid, $2::uuid, NULL, $3)",
+        uuid.UUID(node_id), uuid.UUID(TEST_USER_ID), "ExplicitBindGetNode",
+    )
+    found = await get_node(auth_conn, node_id, user_id=TEST_USER_ID)
+    assert found is not None
+    assert found["name"] == "ExplicitBindGetNode"
+
+
+@pytest.mark.asyncio
+async def test_get_node_mismatched_user_id_returns_none(auth_conn):
+    node_id = str(uuid.uuid4())
+    other_user_id = str(uuid.uuid4())
+    await auth_conn.execute(
+        "INSERT INTO users (id, username, email, password_hash, is_admin) "
+        "VALUES ($1::uuid, 'nodes_other2', 'nodes_other2@example.com', 'x', false) "
+        "ON CONFLICT DO NOTHING",
+        other_user_id,
+    )
+    await auth_conn.execute(
+        "INSERT INTO context_nodes (id, user_id, parent_id, name) "
+        "VALUES ($1::uuid, $2::uuid, NULL, $3)",
+        uuid.UUID(node_id), uuid.UUID(TEST_USER_ID), "MismatchGetNode",
+    )
+    result = await get_node(auth_conn, node_id, user_id=other_user_id)
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_get_node_user_id_is_optional_backward_compatible(conn):
+    """user_id=None (default) preserves old RLS-only behavior on a
+    GUC-scoped connection — no regression for existing callers."""
+    node = await create_node(conn, name="BackCompatGetNode", node_type="context")
+    found = await get_node(conn, node["id"])
+    assert found is not None
+    assert found["id"] == node["id"]