fix(drive): bound identifier text in subscription filters before decoding it

Addresses the third review of #5285:
- Blocking: base58 decoding is quadratic in its input, and filter operands
  come from unauthenticated subscription requests. Identifier operands given
  as text longer than any 32-byte identifier (44 characters) are refused
  before decoding, for system identifier fields and identifier properties;
  long string-property operands are unaffected.
- An empty byte-array operand no longer becomes null through the index-key
  round trip (its index form is empty), so `== []` and `IN [..., []]` work.
- SDK: a contract stays queued for its proved refresh until it is rebound,
  so a cancelled or failed `next()` retries it; a checkpoint at the maximum
  height is rejected instead of overflowing the resume height.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: PastaPastaPasta

packages/rs-drive/src/query/filter.rs

@@ -39,7 +39,7 @@ use std::collections::BTreeMap;
 use dpp::data_contract::accessors::v0::DataContractV0Getters;
 use dpp::data_contract::document_type::methods::{DocumentTypeBasicMethods, DocumentTypeV0Methods};
 use dpp::data_contract::document_type::accessors::DocumentTypeV0Getters;
-use dpp::data_contract::document_type::DocumentTypeRef;
+use dpp::data_contract::document_type::{DocumentPropertyType, DocumentTypeRef};
 use dpp::data_contract::DataContract;
 use dpp::platform_value::Value;
 use dpp::state_transition::batch_transition::batched_transition::document_transition::DocumentTransitionV0Methods;
@@ -814,15 +814,32 @@ fn canonical_value_for_key(
     value: &Value,
     platform_version: &PlatformVersion,
 ) -> Option<Value> {
-    if let Some(property) = document_type.flattened_properties().get(key) {
-        let encoded = property
-            .property_type
-            .encode_value_for_tree_keys(value)
-            .ok()?;
-        return property
-            .property_type
-            .decode_value_for_tree_keys(&encoded)
-            .ok();
+    let property_type = document_type
+        .flattened_properties()
+        .get(key)
+        .map(|property| &property.property_type);
+    // Base58 decoding takes time quadratic in its input, and a filter's operands come from
+    // unauthenticated requests: refuse identifier text longer than any 32-byte identifier.
+    let identifier = matches!(key, "$id" | "$ownerId" | "$creatorId" | "$moderatedBy")
+        || matches!(
+            property_type,
+            Some(
+                DocumentPropertyType::Identifier | DocumentPropertyType::IdentifierWithReference(_)
+            )
+        );
+    if identifier && matches!(value, Value::Text(text) if text.len() > MAX_BASE58_IDENTIFIER_LEN) {
+        return None;
+    }
+    if let Some(property_type) = property_type {
+        let encoded = property_type.encode_value_for_tree_keys(value).ok()?;
+        // The index-key form of an empty byte array is empty, which decodes as null.
+        if encoded.is_empty() && !value.is_null() {
+            return match property_type {
+                DocumentPropertyType::ByteArray(_) => Some(Value::Bytes(vec![])),
+                _ => None,
+            };
+        }
+        return property_type.decode_value_for_tree_keys(&encoded).ok();
     }
     let serialized = document_type
         .serialize_value_for_key(key, value, platform_version)
@@ -832,6 +849,10 @@ fn canonical_value_for_key(
         .ok()
 }
 
+/// The longest base58 text of a 32-byte identifier.
+#[cfg(any(feature = "server", feature = "verify"))]
+const MAX_BASE58_IDENTIFIER_LEN: usize = 44;
+
 /// Canonicalize a clause operand with `canonical`: element-wise for the list operands of
 /// `IN` and `BETWEEN*`, directly otherwise.
 #[cfg(any(feature = "server", feature = "verify"))]
@@ -3341,5 +3362,95 @@ mod tests {
                 TransitionCheckResult::Pass
             );
         }
+
+        #[test]
+        fn should_refuse_identifier_text_longer_than_any_identifier() {
+            let contract = contract();
+            for field in ["identifierField", "$id"] {
+                let clauses = if field == "$id" {
+                    InternalClauses {
+                        primary_key_equal_clause: Some(WhereClause {
+                            field: field.to_string(),
+                            operator: WhereOperator::Equal,
+                            value: Value::Text("z".repeat(120_000)),
+                        }),
+                        ..Default::default()
+                    }
+                } else {
+                    equal(field, Value::Text("z".repeat(120_000)))
+                };
+                let mut filter = DriveDocumentQueryFilter {
+                    contract: &contract,
+                    document_type_name: "withByteArrays".to_string(),
+                    action_clauses: DocumentActionMatchClauses::Delete {
+                        original_document_clauses: clauses,
+                    },
+                };
+                assert!(
+                    filter
+                        .canonicalize_clause_values(PlatformVersion::latest())
+                        .is_err(),
+                    "{field}"
+                );
+            }
+        }
+
+        #[test]
+        fn should_keep_empty_byte_array_operands() {
+            let contract = contract();
+            let platform_version = PlatformVersion::latest();
+            for (operator, value) in [
+                (WhereOperator::Equal, Value::Bytes(vec![])),
+                (
+                    WhereOperator::In,
+                    Value::Array(vec![Value::Bytes(vec![]), Value::Bytes(vec![1])]),
+                ),
+            ] {
+                let mut filter = DriveDocumentQueryFilter {
+                    contract: &contract,
+                    document_type_name: "withByteArrays".to_string(),
+                    action_clauses: DocumentActionMatchClauses::Create {
+                        new_document_clauses: InternalClauses {
+                            equal_clauses: if operator == WhereOperator::Equal {
+                                BTreeMap::from([(
+                                    "byteArrayField".to_string(),
+                                    WhereClause {
+                                        field: "byteArrayField".to_string(),
+                                        operator,
+                                        value: value.clone(),
+                                    },
+                                )])
+                            } else {
+                                BTreeMap::new()
+                            },
+                            in_clauses: if operator == WhereOperator::In {
+                                vec![WhereClause {
+                                    field: "byteArrayField".to_string(),
+                                    operator,
+                                    value: value.clone(),
+                                }]
+                            } else {
+                                vec![]
+                            },
+                            ..Default::default()
+                        },
+                    },
+                };
+                filter
+                    .canonicalize_clause_values(platform_version)
+                    .expect("an empty byte array is a byte array");
+                assert!(filter.validate().is_valid(), "{operator:?}");
+                let transition = create(
+                    &contract,
+                    "withByteArrays",
+                    BTreeMap::from([("byteArrayField".to_string(), Value::Bytes(vec![]))]),
+                );
+                assert_eq!(
+                    filter.matches_document_transition(&transition, None, platform_version),
+                    TransitionCheckResult::Pass,
+                    "{operator:?}"
+                );
+            }
+        }
     }
 }

packages/rs-sdk/src/platform/subscriptions.rs

@@ -265,7 +265,8 @@ impl StateTransitionSubscription {
     /// Rebind the filters on contracts a delivered update changed to their current version,
     /// read with a proof. A contract that can no longer be read ends the subscription.
     async fn refresh_contracts(&mut self) -> Result<(), Error> {
-        while let Some(data_contract_id) = self.contracts_to_refresh.pop_first() {
+        // An id leaves the queue only once rebound, so a cancelled or failed refresh is retried.
+        while let Some(data_contract_id) = self.contracts_to_refresh.first().copied() {
             let contract = DataContract::fetch(&self.sdk, data_contract_id)
                 .await?
                 .ok_or_else(|| {
@@ -276,6 +277,7 @@ impl StateTransitionSubscription {
                 })?;
             self.resolved
                 .rebind_data_contract(Arc::new(contract), self.sdk.version());
+            self.contracts_to_refresh.remove(&data_contract_id);
         }
         Ok(())
     }
@@ -303,7 +305,13 @@ impl StateTransitionSubscription {
         };
         Ok(match responses {
             Responses::Checkpoint(checkpoint) => {
-                self.resume_from = Some(checkpoint.block_height + 1);
+                let resume_from = checkpoint.block_height.checked_add(1).ok_or_else(|| {
+                    Error::Generic(format!(
+                        "the node sent a checkpoint at height {}, after which no stream can resume",
+                        checkpoint.block_height
+                    ))
+                })?;
+                self.resume_from = Some(resume_from);
                 Some(SubscriptionEvent::Checkpoint {
                     block_height: checkpoint.block_height,
                 })
@@ -623,4 +631,33 @@ mod tests {
         // ...and the contract is queued to be read again with a proof.
         assert!(subscription.contracts_to_refresh.contains(&proved.id()));
     }
+
+    #[test]
+    fn should_reject_a_checkpoint_no_stream_can_resume_after() {
+        let mut subscription = subscription();
+        subscription
+            .accept(response(Responses::Checkpoint(Checkpoint {
+                block_height: 41,
+            })))
+            .expect("accepted");
+        assert!(subscription
+            .accept(response(Responses::Checkpoint(Checkpoint {
+                block_height: u64::MAX,
+            })))
+            .is_err());
+        assert_eq!(subscription.resume_height(), Some(42));
+    }
+
+    #[tokio::test]
+    async fn should_keep_a_contract_queued_until_its_refresh_completes() {
+        let mut subscription = subscription();
+        let data_contract_id = Identifier::from([5u8; 32]);
+        subscription.contracts_to_refresh.insert(data_contract_id);
+        // The mock SDK has no expectation for this fetch, so the refresh fails part-way, as it
+        // would if the future were dropped while fetching.
+        assert!(subscription.refresh_contracts().await.is_err());
+        assert!(subscription
+            .contracts_to_refresh
+            .contains(&data_contract_id));
+    }
 }