feat(rpc): implement duplicate key detection for json-rpc requests (#6455)

Co-authored-by: hanabi1224 <[email protected]>
Co-authored-by: Shashank <[email protected]>

Author: 3 people

.config/forest.dic

@@ -39,6 +39,7 @@ daemonize
 Datacap
 devnet
 DB/S
+deduplicates
 deserialize/D
 destructuring
 DNS
@@ -70,6 +71,8 @@ IPFS
 ip
 IPLD
 JSON
+jsonrpc
+jsonrpsee
 JWT
 Kademlia
 Kubernetes
@@ -110,6 +113,7 @@ SECP256k1
 seekable
 serializable
 serializer/SM
+serde_json
 skippable
 statediff
 stateful

Cargo.lock

@@ -115,6 +115,7 @@ is-terminal = "0.4"
 itertools = "0.14"
 jsonrpsee = { version = "0.26", features = ["server", "ws-client", "http-client", "macros"] }
 jsonwebtoken = { version = "10", features = ["aws_lc_rs"] }
+justjson = "0.3"
 keccak-hash = "0.12"
 kubert-prometheus-process = "0.2"
 lazy-regex = "3"

Cargo.toml

@@ -54,6 +54,7 @@ process.
 | `FOREST_JWT_DISABLE_EXP_VALIDATION`                       | 1 or true                        | empty                                          | 1                                                             | Whether or not to disable JWT expiration validation                                                                   |
 | `FOREST_ETH_BLOCK_CACHE_SIZE`                             | positive integer                 | 500                                            | 1                                                             | The size of Eth block cache                                                                                           |
 | `FOREST_RPC_BACKFILL_FULL_TIPSET_FROM_NETWORK`            | 1 or true                        | false                                          | 1                                                             | Whether or not to backfill full tipsets from the p2p network                                                          |
+| `FOREST_STRICT_JSON`                                      | 1 or true                        | false                                          | 1                                                             | Enable strict JSON validation to detect duplicate keys in RPC requests                                                |
 
 ### `FOREST_F3_SIDECAR_FFI_BUILD_OPT_OUT`
 

docs/docs/users/reference/env_variables.md

@@ -0,0 +1,158 @@
+// Copyright 2019-2026 ChainSafe Systems
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+//! JSON validation utilities for detecting duplicate keys before serde_json processing.
+//!
+//! serde_json automatically deduplicates keys at parse time using a "last-write-wins" strategy
+//! This means JSON like `{"/":"cid1", "/":"cid2"}` will keep only the last value, which can lead to unexpected behavior in RPC calls.
+
+use ahash::HashSet;
+use justjson::Value;
+
+pub const STRICT_JSON_ENV: &str = "FOREST_STRICT_JSON";
+
+#[cfg(not(test))]
+use std::sync::LazyLock;
+
+#[cfg(not(test))]
+static STRICT_MODE: LazyLock<bool> =
+    LazyLock::new(|| crate::utils::misc::env::is_env_truthy(STRICT_JSON_ENV));
+
+#[inline]
+pub fn is_strict_mode() -> bool {
+    #[cfg(test)]
+    {
+        crate::utils::misc::env::is_env_truthy(STRICT_JSON_ENV)
+    }
+    #[cfg(not(test))]
+    {
+        *STRICT_MODE
+    }
+}
+
+/// validates JSON for duplicate keys by parsing at the token level.
+pub fn validate_json_for_duplicates(json_str: &str) -> Result<(), String> {
+    if !is_strict_mode() {
+        return Ok(());
+    }
+
+    fn check_value(value: &Value) -> Result<(), String> {
+        match value {
+            Value::Object(obj) => {
+                let mut seen = HashSet::default();
+                for entry in obj.iter() {
+                    let key = entry
+                        .key
+                        .as_str()
+                        .ok_or_else(|| "Invalid JSON key".to_string())?;
+
+                    if !seen.insert(key) {
+                        return Err(format!(
+                            "duplicate key '{}' in JSON object - this likely indicates malformed input. \
+                            Set {}=0 to disable this check",
+                            key, STRICT_JSON_ENV
+                        ));
+                    }
+                    check_value(&entry.value)?;
+                }
+                Ok(())
+            }
+            Value::Array(arr) => {
+                for item in arr.iter() {
+                    check_value(item)?;
+                }
+                Ok(())
+            }
+            _ => Ok(()),
+        }
+    }
+    // defer to serde_json for invalid JSON
+    let value = match Value::from_json(json_str) {
+        Ok(v) => v,
+        Err(_) => return Ok(()),
+    };
+    check_value(&value)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serial_test::serial;
+
+    fn with_strict_mode<F>(enabled: bool, f: F)
+    where
+        F: FnOnce(),
+    {
+        let original = std::env::var(STRICT_JSON_ENV).ok();
+
+        if enabled {
+            unsafe {
+                std::env::set_var(STRICT_JSON_ENV, "1");
+            }
+        } else {
+            unsafe {
+                std::env::remove_var(STRICT_JSON_ENV);
+            }
+        }
+
+        f();
+
+        unsafe {
+            match original {
+                Some(val) => std::env::set_var(STRICT_JSON_ENV, val),
+                None => std::env::remove_var(STRICT_JSON_ENV),
+            }
+        }
+    }
+
+    #[test]
+    #[serial]
+    fn test_no_duplicates() {
+        with_strict_mode(true, || {
+            let json = r#"{"a": 1, "b": 2, "c": 3}"#;
+            assert!(validate_json_for_duplicates(json).is_ok());
+        });
+    }
+
+    #[test]
+    #[serial]
+    fn test_duplicate_keys_detected() {
+        with_strict_mode(true, || {
+            let json = r#"{"/":"cid1", "/":"cid2"}"#;
+            let result = validate_json_for_duplicates(json);
+            assert!(result.is_err(), "Should have detected duplicate key");
+            assert!(result.unwrap_err().contains("duplicate key"));
+        });
+    }
+
+    #[test]
+    #[serial]
+    fn test_strict_mode_disabled() {
+        with_strict_mode(false, || {
+            // should pass with strict mode disabled
+            let json = r#"{"/":"cid1", "/":"cid2"}"#;
+            assert!(validate_json_for_duplicates(json).is_ok());
+        });
+    }
+
+    #[test]
+    #[serial]
+    fn test_duplicate_cid_keys() {
+        with_strict_mode(true, || {
+            let json = r#"{
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "Filecoin.ChainGetMessagesInTipset",
+                "params": [[{
+                    "/":"bafy2bzacea43254b5x6c4l22ynpjfoct5qvabbbk2abcfspfcjkiltivrlyqi",
+                    "/":"bafy2bzacea4viqyaozpfk57lnemwufryb76llxzmebxc7it2rnssqz2ljdl6a",
+                    "/":"bafy2bzaceav6j67epppz5ib55v5ty26dhkq4jinbsizq2olb3azbzxvfmc73o"
+                }]]
+            }"#;
+
+            let result = validate_json_for_duplicates(json);
+            assert!(result.is_err());
+            assert!(result.unwrap_err().contains("duplicate key '/'"));
+        });
+    }
+}

src/rpc/json_validator.rs

@@ -7,11 +7,13 @@ mod channel;
 mod client;
 mod filter_layer;
 mod filter_list;
+pub mod json_validator;
 mod log_layer;
 mod metrics_layer;
 mod request;
 mod segregation_layer;
 mod set_extension_layer;
+mod validation_layer;
 
 use crate::rpc::eth::types::RandomHexStringIdProvider;
 use crate::shim::clock::ChainEpoch;
@@ -623,6 +625,7 @@ where
                     .layer(SetExtensionLayer { path })
                     .layer(SegregationLayer)
                     .layer(FilterLayer::new(filter_list.clone()))
+                    .layer(validation_layer::JsonValidationLayer)
                     .layer(AuthLayer {
                         headers,
                         keystore: keystore.clone(),

src/rpc/mod.rs

@@ -0,0 +1,136 @@
+// Copyright 2019-2026 ChainSafe Systems
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+use futures::future::Either;
+use jsonrpsee::MethodResponse;
+use jsonrpsee::core::middleware::{Batch, BatchEntry, BatchEntryErr, Notification};
+use jsonrpsee::server::middleware::rpc::RpcServiceT;
+use jsonrpsee::types::ErrorObject;
+use tower::Layer;
+
+use super::json_validator;
+
+/// JSON-RPC error code for invalid request
+const INVALID_REQUEST: i32 = -32600;
+
+/// stateless jsonrpsee layer for validating JSON-RPC requests
+#[derive(Clone, Default)]
+pub(super) struct JsonValidationLayer;
+
+impl<S> Layer<S> for JsonValidationLayer {
+    type Service = Validation<S>;
+
+    fn layer(&self, service: S) -> Self::Service {
+        Validation { service }
+    }
+}
+
+#[derive(Clone)]
+pub(super) struct Validation<S> {
+    service: S,
+}
+
+impl<S> Validation<S> {
+    fn validation_enabled() -> bool {
+        json_validator::is_strict_mode()
+    }
+
+    fn validate_params(params_str: &str) -> Result<(), String> {
+        json_validator::validate_json_for_duplicates(params_str)
+    }
+}
+
+impl<S> RpcServiceT for Validation<S>
+where
+    S: RpcServiceT<MethodResponse = MethodResponse, NotificationResponse = MethodResponse>
+        + Send
+        + Sync
+        + Clone
+        + 'static,
+{
+    type MethodResponse = S::MethodResponse;
+    type NotificationResponse = S::NotificationResponse;
+    type BatchResponse = S::BatchResponse;
+
+    fn call<'a>(
+        &self,
+        req: jsonrpsee::types::Request<'a>,
+    ) -> impl Future<Output = Self::MethodResponse> + Send + 'a {
+        if !Self::validation_enabled() {
+            return Either::Left(self.service.call(req));
+        }
+
+        if let Err(e) =
+            json_validator::validate_json_for_duplicates(req.params().as_str().unwrap_or("[]"))
+        {
+            let err = ErrorObject::owned(INVALID_REQUEST, e, None::<()>);
+            return Either::Right(async move { MethodResponse::error(req.id(), err) });
+        }
+
+        Either::Left(self.service.call(req))
+    }
+
+    fn batch<'a>(
+        &self,
+        mut batch: Batch<'a>,
+    ) -> impl Future<Output = Self::BatchResponse> + Send + 'a {
+        let service = self.service.clone();
+
+        async move {
+            if !Self::validation_enabled() {
+                return service.batch(batch).await;
+            }
+
+            for entry in batch.iter_mut() {
+                if let Ok(batch_entry) = entry {
+                    let params_str = match batch_entry.params() {
+                        Some(p) => p.as_ref().get(),
+                        None => continue,
+                    };
+
+                    if let Err(e) = Self::validate_params(params_str) {
+                        match batch_entry {
+                            BatchEntry::Call(req) => {
+                                let err = ErrorObject::owned(INVALID_REQUEST, e, None::<()>);
+                                let err_entry = BatchEntryErr::new(req.id().clone(), err);
+                                *entry = Err(err_entry);
+                            }
+                            BatchEntry::Notification(_notif) => {
+                                let err = ErrorObject::owned(INVALID_REQUEST, e, None::<()>);
+                                let err_entry = BatchEntryErr::new(jsonrpsee::types::Id::Null, err);
+                                *entry = Err(err_entry);
+                            }
+                        }
+                    }
+                }
+            }
+
+            service.batch(batch).await
+        }
+    }
+
+    fn notification<'a>(
+        &self,
+        n: Notification<'a>,
+    ) -> impl Future<Output = Self::NotificationResponse> + Send + 'a {
+        let service = self.service.clone();
+
+        async move {
+            if !Self::validation_enabled() {
+                return service.notification(n).await;
+            }
+
+            let params_str = match n.params() {
+                Some(p) => p.as_ref().get(),
+                None => return service.notification(n).await,
+            };
+
+            if let Err(e) = Self::validate_params(params_str) {
+                let err = ErrorObject::owned(INVALID_REQUEST, e, None::<()>);
+                return MethodResponse::error(jsonrpsee::types::Id::Null, err);
+            }
+
+            service.notification(n).await
+        }
+    }
+}