Make wildcard storage change subscriptions RPC-unsafe (#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>

substrate/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(

substrate/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(_)))));
+}