Allow to expose a subset of unsafe RPCs (#5233)

* sc-cli: Use type-safe constructors for RPC/Prometheus interfaces

* service: Simplify rpc handler creation

Could probably be further simplifies once [this][commit] lands.

[commit]: https://github.com/paritytech/jsonrpc/commit/20485387ed06a48f1a70bf4d609a7cde6cf0accf

* service: Streamline some HTTP & WS server start logic

* client: Introduce a simple RPC policy mechanism

* rpc/system: Check unsafe RPCs

* rpc/offchain: Check unsafe RPCs

* rpc/author: Check unsafe RPCs

substrate/client/rpc/src/author/mod.rs

@@ -30,7 +30,7 @@ use rpc::futures::{
 };
 use futures::{StreamExt as _, compat::Compat};
 use futures::future::{ready, FutureExt, TryFutureExt};
-use sc_rpc_api::Subscriptions;
+use sc_rpc_api::{DenyUnsafe, Subscriptions};
 use jsonrpc_pubsub::{typed::Subscriber, SubscriptionId};
 use codec::{Encode, Decode};
 use sp_core::{Bytes, traits::BareCryptoStorePtr};
@@ -56,6 +56,8 @@ pub struct Author<P, Client> {
 	subscriptions: Subscriptions,
 	/// The key store.
 	keystore: BareCryptoStorePtr,
+	/// Whether to deny unsafe calls
+	deny_unsafe: DenyUnsafe,
 }
 
 impl<P, Client> Author<P, Client> {
@@ -65,12 +67,14 @@ impl<P, Client> Author<P, Client> {
 		pool: Arc<P>,
 		subscriptions: Subscriptions,
 		keystore: BareCryptoStorePtr,
+		deny_unsafe: DenyUnsafe,
 	) -> Self {
 		Author {
 			client,
 			pool,
 			subscriptions,
 			keystore,
+			deny_unsafe,
 		}
 	}
 }
@@ -97,6 +101,8 @@ impl<P, Client> AuthorApi<TxHash<P>, BlockHash<P>> for Author<P, Client>
 		suri: String,
 		public: Bytes,
 	) -> Result<()> {
+		self.deny_unsafe.check_if_safe()?;
+
 		let key_type = key_type.as_str().try_into().map_err(|_| Error::BadKeyType)?;
 		let mut keystore = self.keystore.write();
 		keystore.insert_unknown(key_type, &suri, &public[..])
@@ -105,6 +111,8 @@ impl<P, Client> AuthorApi<TxHash<P>, BlockHash<P>> for Author<P, Client>
 	}
 
 	fn rotate_keys(&self) -> Result<Bytes> {
+		self.deny_unsafe.check_if_safe()?;
+
 		let best_block_hash = self.client.info().best_hash;
 		self.client.runtime_api().generate_session_keys(
 			&generic::BlockId::Hash(best_block_hash),
@@ -113,6 +121,8 @@ impl<P, Client> AuthorApi<TxHash<P>, BlockHash<P>> for Author<P, Client>
 	}
 
 	fn has_session_keys(&self, session_keys: Bytes) -> Result<bool> {
+		self.deny_unsafe.check_if_safe()?;
+
 		let best_block_hash = self.client.info().best_hash;
 		let keys = self.client.runtime_api().decode_session_keys(
 			&generic::BlockId::Hash(best_block_hash),
@@ -124,6 +134,8 @@ impl<P, Client> AuthorApi<TxHash<P>, BlockHash<P>> for Author<P, Client>
 	}
 
 	fn has_key(&self, public_key: Bytes, key_type: String) -> Result<bool> {
+		self.deny_unsafe.check_if_safe()?;
+
 		let key_type = key_type.as_str().try_into().map_err(|_| Error::BadKeyType)?;
 		Ok(self.keystore.read().has_keys(&[(public_key.to_vec(), key_type)]))
 	}
@@ -151,6 +163,8 @@ impl<P, Client> AuthorApi<TxHash<P>, BlockHash<P>> for Author<P, Client>
 		&self,
 		bytes_or_hash: Vec<hash::ExtrinsicOrHash<TxHash<P>>>,
 	) -> Result<Vec<TxHash<P>>> {
+		self.deny_unsafe.check_if_safe()?;
+
 		let hashes = bytes_or_hash.into_iter()
 			.map(|x| match x {
 				hash::ExtrinsicOrHash::Hash(h) => Ok(h),

substrate/client/rpc/src/author/tests.rs

@@ -83,6 +83,7 @@ impl TestSetup {
 			pool: self.pool.clone(),
 			subscriptions: Subscriptions::new(Arc::new(self.runtime.executor())),
 			keystore: self.keystore.clone(),
+			deny_unsafe: DenyUnsafe::No,
 		}
 	}
 }

substrate/client/rpc/src/lib.rs

@@ -22,7 +22,7 @@
 
 mod metadata;
 
-pub use sc_rpc_api::Subscriptions;
+pub use sc_rpc_api::{DenyUnsafe, Subscriptions};
 pub use self::metadata::Metadata;
 pub use rpc::IoHandlerExtension as RpcExtension;
 

substrate/client/rpc/src/offchain/mod.rs

@@ -21,6 +21,7 @@ mod tests;
 
 /// Re-export the API for backward compatibility.
 pub use sc_rpc_api::offchain::*;
+use sc_rpc_api::DenyUnsafe;
 use self::error::{Error, Result};
 use sp_core::{
 	Bytes,
@@ -34,13 +35,15 @@ use std::sync::Arc;
 pub struct Offchain<T: OffchainStorage> {
 	/// Offchain storage
 	storage: Arc<RwLock<T>>,
+	deny_unsafe: DenyUnsafe,
 }
 
 impl<T: OffchainStorage> Offchain<T> {
 	/// Create new instance of Offchain API.
-	pub fn new(storage: T) -> Self {
+	pub fn new(storage: T, deny_unsafe: DenyUnsafe) -> Self {
 		Offchain {
 			storage: Arc::new(RwLock::new(storage)),
+			deny_unsafe,
 		}
 	}
 }
@@ -48,6 +51,8 @@ impl<T: OffchainStorage> Offchain<T> {
 impl<T: OffchainStorage + 'static> OffchainApi for Offchain<T> {
 	/// Set offchain local storage under given key and prefix.
 	fn set_local_storage(&self, kind: StorageKind, key: Bytes, value: Bytes) -> Result<()> {
+		self.deny_unsafe.check_if_safe()?;
+
 		let prefix = match kind {
 			StorageKind::PERSISTENT => sp_offchain::STORAGE_PREFIX,
 			StorageKind::LOCAL => return Err(Error::UnavailableStorageKind),
@@ -58,6 +63,8 @@ impl<T: OffchainStorage + 'static> OffchainApi for Offchain<T> {
 
 	/// Get offchain local storage under given key and prefix.
 	fn get_local_storage(&self, kind: StorageKind, key: Bytes) -> Result<Option<Bytes>> {
+		self.deny_unsafe.check_if_safe()?;
+
 		let prefix = match kind {
 			StorageKind::PERSISTENT => sp_offchain::STORAGE_PREFIX,
 			StorageKind::LOCAL => return Err(Error::UnavailableStorageKind),

substrate/client/rpc/src/offchain/tests.rs

@@ -21,7 +21,7 @@ use sp_core::{Bytes, offchain::storage::InMemOffchainStorage};
 #[test]
 fn local_storage_should_work() {
 	let storage = InMemOffchainStorage::default();
-	let offchain = Offchain::new(storage);
+	let offchain = Offchain::new(storage, DenyUnsafe::No);
 	let key = Bytes(b"offchain_storage".to_vec());
 	let value = Bytes(b"offchain_value".to_vec());
 
@@ -34,3 +34,20 @@ fn local_storage_should_work() {
 		Ok(Some(ref v)) if *v == value
 	);
 }
+
+#[test]
+fn offchain_calls_considered_unsafe() {
+	let storage = InMemOffchainStorage::default();
+	let offchain = Offchain::new(storage, DenyUnsafe::Yes);
+	let key = Bytes(b"offchain_storage".to_vec());
+	let value = Bytes(b"offchain_value".to_vec());
+
+	assert_matches!(
+		offchain.set_local_storage(StorageKind::PERSISTENT, key.clone(), value.clone()),
+		Err(Error::UnsafeRpcCalled(_))
+	);
+	assert_matches!(
+		offchain.get_local_storage(StorageKind::PERSISTENT, key),
+		Err(Error::UnsafeRpcCalled(_))
+	);
+}

substrate/client/rpc/src/system/mod.rs

@@ -21,7 +21,7 @@ mod tests;
 
 use futures::{future::BoxFuture, FutureExt, TryFutureExt};
 use futures::{channel::oneshot, compat::Compat};
-use sc_rpc_api::Receiver;
+use sc_rpc_api::{DenyUnsafe, Receiver};
 use sp_utils::mpsc::TracingUnboundedSender;
 use sp_runtime::traits::{self, Header as HeaderT};
 
@@ -31,10 +31,19 @@ pub use sc_rpc_api::system::*;
 pub use self::helpers::{SystemInfo, Health, PeerInfo, NodeRole};
 pub use self::gen_client::Client as SystemClient;
 
+macro_rules! bail_if_unsafe {
+	($value: expr) => {
+		if let Err(err) = $value.check_if_safe() {
+			return async move { Err(err.into()) }.boxed().compat();
+		}
+	};
+}
+
 /// System API implementation
 pub struct System<B: traits::Block> {
 	info: SystemInfo,
 	send_back: TracingUnboundedSender<Request<B>>,
+	deny_unsafe: DenyUnsafe,
 }
 
 /// Request to be processed.
@@ -66,10 +75,12 @@ impl<B: traits::Block> System<B> {
 	pub fn new(
 		info: SystemInfo,
 		send_back: TracingUnboundedSender<Request<B>>,
+		deny_unsafe: DenyUnsafe,
 	) -> Self {
 		System {
 			info,
 			send_back,
+			deny_unsafe,
 		}
 	}
 }
@@ -113,21 +124,37 @@ impl<B: traits::Block> SystemApi<B::Hash, <B::Header as HeaderT>::Number> for Sy
 		Receiver(Compat::new(rx))
 	}
 
-	fn system_peers(&self) -> Receiver<Vec<PeerInfo<B::Hash, <B::Header as HeaderT>::Number>>> {
+	fn system_peers(&self)
+		-> Compat<BoxFuture<'static, rpc::Result<Vec<PeerInfo<B::Hash, <B::Header as HeaderT>::Number>>>>>
+	{
+		bail_if_unsafe!(self.deny_unsafe);
+
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::Peers(tx));
-		Receiver(Compat::new(rx))
+
+		async move {
+			rx.await.map_err(|_| rpc::Error::internal_error())
+		}.boxed().compat()
 	}
 
-	fn system_network_state(&self) -> Receiver<rpc::Value> {
+	fn system_network_state(&self)
+		-> Compat<BoxFuture<'static, rpc::Result<rpc::Value>>>
+	{
+		bail_if_unsafe!(self.deny_unsafe);
+
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::NetworkState(tx));
-		Receiver(Compat::new(rx))
+
+		async move {
+			rx.await.map_err(|_| rpc::Error::internal_error())
+		}.boxed().compat()
 	}
 
 	fn system_add_reserved_peer(&self, peer: String)
 		-> Compat<BoxFuture<'static, std::result::Result<(), rpc::Error>>>
 	{
+		bail_if_unsafe!(self.deny_unsafe);
+
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::NetworkAddReservedPeer(peer, tx));
 		async move {
@@ -142,6 +169,8 @@ impl<B: traits::Block> SystemApi<B::Hash, <B::Header as HeaderT>::Number> for Sy
 	fn system_remove_reserved_peer(&self, peer: String)
 		-> Compat<BoxFuture<'static, std::result::Result<(), rpc::Error>>>
 	{
+		bail_if_unsafe!(self.deny_unsafe);
+
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::NetworkRemoveReservedPeer(peer, tx));
 		async move {

substrate/client/rpc/src/system/tests.rs

@@ -109,13 +109,17 @@ fn api<T: Into<Option<Status>>>(sync: T) -> System<Block> {
 			future::ready(())
 		}))
 	});
-	System::new(SystemInfo {
-		impl_name: "testclient".into(),
-		impl_version: "0.2.0".into(),
-		chain_name: "testchain".into(),
-		properties: Default::default(),
-		chain_type: Default::default(),
-	}, tx)
+	System::new(
+		SystemInfo {
+			impl_name: "testclient".into(),
+			impl_version: "0.2.0".into(),
+			chain_name: "testchain".into(),
+			properties: Default::default(),
+			chain_type: Default::default(),
+		},
+		tx,
+		sc_rpc_api::DenyUnsafe::No
+	)
 }
 
 fn wait_receiver<T>(rx: Receiver<T>) -> T {
@@ -238,14 +242,19 @@ fn system_local_listen_addresses_works() {
 
 #[test]
 fn system_peers() {
+	let mut runtime = tokio::runtime::current_thread::Runtime::new().unwrap();
+
 	let peer_id = PeerId::random();
+	let req = api(Status {
+		peer_id: peer_id.clone(),
+		peers: 1,
+		is_syncing: false,
+		is_dev: true,
+	}).system_peers();
+	let res = runtime.block_on(req).unwrap();
+
 	assert_eq!(
-		wait_receiver(api(Status {
-			peer_id: peer_id.clone(),
-			peers: 1,
-			is_syncing: false,
-			is_dev: true,
-		}).system_peers()),
+		res,
 		vec![PeerInfo {
 			peer_id: peer_id.to_base58(),
 			roles: "FULL".into(),
@@ -258,7 +267,10 @@ fn system_peers() {
 
 #[test]
 fn system_network_state() {
-	let res = wait_receiver(api(None).system_network_state());
+	let mut runtime = tokio::runtime::current_thread::Runtime::new().unwrap();
+	let req = api(None).system_network_state();
+	let res = runtime.block_on(req).unwrap();
+
 	assert_eq!(
 		serde_json::from_value::<sc_network::network_state::NetworkState>(res).unwrap(),
 		sc_network::network_state::NetworkState {