Make wildcard storage change subscriptions RPC-unsafe (paritytech#11259)

* When an RPC is rejected because it's RPC-unsafe say why in the error message

* Make wildcard storage change subscriptions RPC-unsafe

* Apply suggestions from code review

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Fix typo

* Fix tests

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

client/consensus/babe/rpc/src/lib.rs

@@ -290,6 +290,6 @@ mod tests {
 		let mut response: serde_json::Value = serde_json::from_str(&response).unwrap();
 		let error: RpcError = serde_json::from_value(response["error"].take()).unwrap();
 
-		assert_eq!(error, RpcError::method_not_found())
+		assert_eq!(error, sc_rpc_api::UnsafeRpcError.into())
 	}
 }

client/rpc-api/src/policy.rs

@@ -56,7 +56,7 @@ impl std::fmt::Display for UnsafeRpcError {
 impl std::error::Error for UnsafeRpcError {}
 
 impl From<UnsafeRpcError> for rpc::Error {
-	fn from(_: UnsafeRpcError) -> rpc::Error {
-		rpc::Error::method_not_found()
+	fn from(error: UnsafeRpcError) -> rpc::Error {
+		rpc::Error { code: rpc::ErrorCode::MethodNotFound, message: error.to_string(), data: None }
 	}
 }

client/rpc-api/src/state/mod.rs

@@ -144,7 +144,18 @@ pub trait StateApi<Hash> {
 		id: SubscriptionId,
 	) -> RpcResult<bool>;
 
-	/// New storage subscription
+	/// Subscribe to the changes in the storage.
+	///
+	/// This RPC endpoint has two modes of operation:
+	///   1) When `keys` is not `None` you'll only be informed about the changes
+	///      done to the specified keys; this is RPC-safe.
+	///   2) When `keys` is `None` you'll be informed of *all* of the changes;
+	///      **this is RPC-unsafe**.
+	///
+	/// When subscribed to all of the changes this API will emit every storage
+	/// change for every block that is imported. These changes will only be sent
+	/// after a block is imported. If you require a consistent view across all changes
+	/// of every block, you need to take this into account.
 	#[pubsub(subscription = "state_storage", subscribe, name = "state_subscribeStorage")]
 	fn subscribe_storage(
 		&self,

client/rpc/src/state/mod.rs

@@ -332,7 +332,15 @@ where
 		subscriber: Subscriber<StorageChangeSet<Block::Hash>>,
 		keys: Option<Vec<StorageKey>>,
 	) {
-		self.backend.subscribe_storage(meta, subscriber, keys);
+		if keys.is_none() {
+			if let Err(err) = self.deny_unsafe.check_if_safe() {
+				subscriber.reject(err.into())
+				    .expect("subscription rejection can only fail if it's been already rejected, and we're rejecting it for the first time; qed");
+				return
+			}
+		}
+
+		self.backend.subscribe_storage(meta, subscriber, keys)
 	}
 
 	fn unsubscribe_storage(

client/rpc/src/state/tests.rs

@@ -571,3 +571,39 @@ fn should_deserialize_storage_key() {
 
 	assert_eq!(k.0.len(), 32);
 }
+
+#[test]
+fn wildcard_storage_subscriptions_are_rpc_unsafe() {
+	let (subscriber, id, _) = Subscriber::new_test("test");
+
+	let client = Arc::new(substrate_test_runtime_client::new());
+	let (api, _child) = new_full(
+		client.clone(),
+		SubscriptionManager::new(Arc::new(TaskExecutor)),
+		DenyUnsafe::Yes,
+		None,
+	);
+
+	api.subscribe_storage(Default::default(), subscriber, None.into());
+
+	let error = executor::block_on(id).unwrap().unwrap_err();
+	assert_eq!(error.to_string(), "Method not found: RPC call is unsafe to be called externally");
+}
+
+#[test]
+fn concrete_storage_subscriptions_are_rpc_safe() {
+	let (subscriber, id, _) = Subscriber::new_test("test");
+
+	let client = Arc::new(substrate_test_runtime_client::new());
+	let (api, _child) = new_full(
+		client.clone(),
+		SubscriptionManager::new(Arc::new(TaskExecutor)),
+		DenyUnsafe::Yes,
+		None,
+	);
+
+	let key = StorageKey(STORAGE_KEY.to_vec());
+	api.subscribe_storage(Default::default(), subscriber, Some(vec![key]));
+
+	assert!(matches!(executor::block_on(id), Ok(Ok(SubscriptionId::String(_)))));
+}

utils/frame/rpc/system/src/lib.rs

@@ -279,7 +279,7 @@ mod tests {
 		let res = accounts.dry_run(vec![].into(), None);
 
 		// then
-		assert_eq!(block_on(res), Err(RpcError::method_not_found()));
+		assert_eq!(block_on(res), Err(sc_rpc_api::UnsafeRpcError.into()));
 	}
 
 	#[test]