fix(relay): protect sync cursors and ID reads

Signed-off-by: Daniel Rouleau <drouleau@squareup.com>

crates/buzz-db/src/store/artifact_sync.rs

@@ -120,14 +120,6 @@ impl Db {
         if ids.is_empty() {
             return Ok(vec![]);
         }
-        let mut tags = query.tags.clone();
-        tags.push(("d".into(), ids.iter().map(Uuid::to_string).collect()));
-        let narrowed = ArtifactQuery {
-            view: query.view,
-            tags,
-            limit: PAGE_ROWS as i64,
-            offset: 0,
-        };
         let connection = crate::observability::acquire_writer(
             &self.pool,
             crate::observability::WriterOperation::SubscriptionHistory,
@@ -141,7 +133,8 @@ impl Db {
             .execute(&mut *tx)
             .await?;
         let mut qb = QueryBuilder::new("SELECT e.id,e.pubkey,e.created_at,e.kind,e.tags,e.content,e.sig,e.received_at,e.channel_id FROM events e ");
-        super::artifact_query::predicates(&mut qb, community, reader, &narrowed);
+        super::artifact_query::predicates(&mut qb, community, reader, query);
+        qb.push(" AND a.artifact_id=ANY(").push_bind(ids).push(")");
         qb.push(" LIMIT ").push_bind(PAGE_ROWS as i64);
         let mut events = Vec::new();
         let mut bytes = 0;
@@ -204,7 +197,7 @@ WITH readable AS NOT MATERIALIZED (
       AND m.sync_modified_at <= $4 AND a.channel_id IN (SELECT id FROM readable)
       AND (cardinality($8::text[])=0 OR a.artifact_type=ANY($8))
     UNION ALL
-    SELECT r.modified_at,3,encode(r.event_id,'hex'),r.artifact_id,NULL::uuid
+    SELECT DISTINCT r.modified_at,3,r.artifact_id::text,r.artifact_id,NULL::uuid
     FROM artifact_revisions r WHERE r.community_id=$1
       AND r.source_channel_id IS NOT NULL AND r.source_channel_id IN (SELECT id FROM readable)
       AND $3::timestamptz IS NOT NULL AND r.modified_at >= $3 AND r.modified_at <= $4

crates/buzz-relay/src/api/artifact_postgres_tests/sync_access_postgres_tests.rs

@@ -384,3 +384,30 @@ async fn sync_progress_replays_queries_with_escaped_values() {
     assert!(status.is_success(), "{status}: {next}");
     assert_eq!(next["changes"], json!([]));
 }
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn sync_current_intersects_known_ids_with_query_scope() {
+    let _guard = SERIAL.lock().await;
+    let f = fixture().await;
+    let included = Uuid::new_v4();
+    let excluded = Uuid::new_v4();
+    let event = f.revision(included, "create", None);
+    accept(&f, &event).await;
+    accept(&f, &f.revision(excluded, "create", None)).await;
+    let (status, body) = f
+        .request(
+            "/query",
+            json!([{"artifact":"sync","action":"current",
+                "query":{"artifact":"current","#type":["buzz.task"],"#d":[included]},
+                "artifactIds":[included,excluded]}]),
+        )
+        .await;
+    assert!(status.is_success(), "{status}: {body}");
+    assert_eq!(body["changes"][0]["artifactId"], included.to_string());
+    assert_eq!(body["changes"][0]["event"]["id"], event.id.to_hex());
+    assert_eq!(
+        body["changes"][1],
+        json!({"type":"remove","artifactId":excluded})
+    );
+}

crates/buzz-relay/src/api/artifact_postgres_tests/sync_cursor_postgres_tests.rs

@@ -0,0 +1,102 @@
+use super::*;
+
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn sync_move_cursor_hides_private_and_redacted_revision_ids() {
+    let _guard = SERIAL.lock().await;
+    let f = fixture().await;
+    let (_, empty) = f
+        .request_as(
+            &f.peer,
+            "/query",
+            json!([{"artifact":"sync","query":query()}]),
+        )
+        .await;
+    let mut current = std::collections::HashMap::new();
+    let mut private_ids = Vec::new();
+    for _ in 0..101 {
+        let id = Uuid::new_v4();
+        let created = f.revision(id, "create", None);
+        accept(&f, &created).await;
+        let moved = f.revision_as(&f.owner, f.private, id, "move", Some(&created));
+        accept(&f, &moved).await;
+        private_ids.push(moved.id.to_hex());
+        current.insert(id, moved);
+    }
+    let (&id, previous) = current.iter().next().unwrap();
+    let back = f.revision(id, "move", Some(previous));
+    accept(&f, &back).await;
+    let moved = f.revision_as(&f.owner, f.private, id, "move", Some(&back));
+    accept(&f, &moved).await;
+    private_ids.push(moved.id.to_hex());
+    current.insert(id, moved);
+    let stamp: chrono::DateTime<chrono::Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
+        .fetch_one(&f.pool)
+        .await
+        .unwrap();
+    sqlx::query("UPDATE artifact_revisions SET modified_at=$2 WHERE community_id=$1")
+        .bind(f.community.as_uuid())
+        .bind(stamp)
+        .execute(&f.pool)
+        .await
+        .unwrap();
+    for redacted in [false, true] {
+        let (status, first) = f
+            .request_as(
+                &f.peer,
+                "/query",
+                json!([{"artifact":"sync","query":query(),"checkpoint":empty["checkpoint"]}]),
+            )
+            .await;
+        assert!(status.is_success(), "{status}: {first}");
+        let cursor = first["nextCursor"].as_str().unwrap();
+        let payload = cursor.split_once('.').unwrap().0;
+        let decoded = base64::engine::general_purpose::URL_SAFE_NO_PAD
+            .decode(payload)
+            .unwrap();
+        let decoded = String::from_utf8(decoded).unwrap();
+        for private_id in &private_ids {
+            assert!(
+                !decoded.contains(private_id),
+                "cursor exposes a private revision ID (redacted={redacted})"
+            );
+        }
+        assert_eq!(first["changes"].as_array().unwrap().len(), 100);
+        let (status, last) = f
+            .request_as(
+                &f.peer,
+                "/query",
+                json!([{"artifact":"sync","query":query(),"cursor":cursor}]),
+            )
+            .await;
+        assert!(status.is_success(), "{status}: {last}");
+        assert!(last["checkpoint"].is_string());
+        let mut ids = std::collections::HashSet::new();
+        for change in first["changes"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .chain(last["changes"].as_array().unwrap())
+        {
+            assert_eq!(change["type"], "remove");
+            assert!(ids.insert(change["artifactId"].as_str().unwrap().to_owned()));
+        }
+        assert_eq!(ids.len(), 101);
+        if !redacted {
+            let boundary = first["changes"].as_array().unwrap().last().unwrap()["artifactId"]
+                .as_str()
+                .unwrap()
+                .parse::<Uuid>()
+                .unwrap();
+            assert!(buzz_db::event::soft_delete_event_and_update_thread(
+                &f.pool,
+                f.community,
+                current[&boundary].id.as_bytes(),
+                None,
+                None
+            )
+            .await
+            .unwrap());
+        }
+    }
+}

crates/buzz-relay/src/api/artifact_postgres_tests/sync_postgres_tests.rs

@@ -2,6 +2,8 @@ use super::*;
 use axum::http::StatusCode;
 #[path = "sync_access_postgres_tests.rs"]
 mod access_postgres_tests;
+#[path = "sync_cursor_postgres_tests.rs"]
+mod cursor_postgres_tests;
 #[path = "sync_policy_postgres_tests.rs"]
 mod policy_postgres_tests;
 #[path = "sync_scale_postgres_tests.rs"]