RPC Categorization (Namespaces)

kaspad/src/args.rs

@@ -5,6 +5,7 @@ use kaspa_consensus_core::{
 };
 use kaspa_core::kaspad_env::version;
 use kaspa_notify::address::tracker::Tracker;
+use kaspa_rpc_core::api::namespaces::Namespaces;
 use kaspa_utils::networking::ContextualNetAddress;
 use kaspa_wrpc_server::address::WrpcNetAddress;
 use serde::Deserialize;
@@ -90,6 +91,8 @@ pub struct Args {
     #[serde(rename = "nogrpc")]
     pub disable_grpc: bool,
     pub ram_scale: f64,
+    #[serde(rename = "rpc-api")]
+    pub rpc_namespaces: Option<Namespaces>,
 }
 
 impl Default for Args {
@@ -140,6 +143,7 @@ impl Default for Args {
             disable_dns_seeding: false,
             disable_grpc: false,
             ram_scale: 1.0,
+            rpc_namespaces: None,
         }
     }
 }
@@ -369,7 +373,15 @@ Setting to 0 prevents the preallocation and sets the maximum to {}, leading to 0
                 .help("Apply a scale factor to memory allocation bounds. Nodes with limited RAM (~4-8GB) should set this to ~0.3-0.5 respectively. Nodes with
 a large RAM (~64GB) can set this value to ~3.0-4.0 and gain superior performance especially for syncing peers faster"),
         )
-        ;
+        .arg(
+            Arg::new("rpc-api")
+                .long("rpc-api")
+                .value_name("namespaces")
+                .require_equals(true)
+                .default_missing_value(None)
+                .value_parser(clap::value_parser!(Namespaces))
+                .help("Specify allowed RPC namespaces exposed over RPC servers."),
+        );
 
     #[cfg(feature = "devnet-prealloc")]
     let cmd = cmd
@@ -448,6 +460,7 @@ impl Args {
             disable_dns_seeding: arg_match_unwrap_or::<bool>(&m, "nodnsseed", defaults.disable_dns_seeding),
             disable_grpc: arg_match_unwrap_or::<bool>(&m, "nogrpc", defaults.disable_grpc),
             ram_scale: arg_match_unwrap_or::<f64>(&m, "ram-scale", defaults.ram_scale),
+            rpc_namespaces: m.get_one::<Namespaces>("rpc-api").cloned().or(defaults.rpc_namespaces),
 
             #[cfg(feature = "devnet-prealloc")]
             num_prealloc_utxos: m.get_one::<u64>("num-prealloc-utxos").cloned(),
@@ -560,4 +573,6 @@ fn arg_match_many_unwrap_or<T: Clone + Send + Sync + 'static>(m: &clap::ArgMatch
       --override-dag-params-file=           Overrides DAG params (allowed only on devnet)
   -s, --service=                            Service command {install, remove, start, stop}
       --nogrpc                              Don't initialize the gRPC server
+      --rpc-api=                            Set available namespaces over RPC server(s).
+                                            (By default all namespaces are enabled)
 */

kaspad/src/daemon.rs

@@ -574,6 +574,7 @@ do you confirm? (answer y/n or pass --yes to the Kaspad command line to confirm
         p2p_tower_counters.clone(),
         grpc_tower_counters.clone(),
         system_info,
+        args.rpc_namespaces.clone(),
     ));
     let grpc_service_broadcasters: usize = 3; // TODO: add a command line argument or derive from other arg/config/host-related fields
     let grpc_service = if !args.disable_grpc {

rpc/core/src/api/Extending RpcApi.md

@@ -23,7 +23,7 @@ As an illustration, let's pretend that we add a new `submit_block` method.
    Implement the first as a call to the second.
    (ie. `async fn submit_block(&self, block: RpcBlock, allow_non_daa_blocks: bool) -> RpcResult<SubmitBlockResponse>` and
    `async fn submit_block_call(&self, request: SubmitBlockRequest) -> RpcResult<SubmitBlockResponse>;`)
-6. Implement the function having a `_call` suffix into `kaspa_rpc_core::server::service::RpcCoreService`.
+6. Implement the function having a `_call` suffix into `kaspa_rpc_core::server::service::RpcCoreService` and wrap it with namespace macro with its corresponding category.
 
 ## rpc-grpc
 

rpc/core/src/api/mod.rs

@@ -4,6 +4,7 @@
 
 pub mod connection;
 pub mod ctl;
+pub mod namespaces;
 pub mod notifications;
 pub mod ops;
 pub mod rpc;

rpc/core/src/api/namespaces.rs

@@ -0,0 +1,115 @@
+use kaspa_notify::scope::Scope;
+use serde::Deserialize;
+use std::collections::HashSet;
+use std::str::FromStr;
+use thiserror::Error;
+
+/// Enum representing available namespace groups
+#[derive(Debug, Hash, Eq, PartialEq, Clone, Deserialize)]
+pub enum Namespace {
+    General,
+    Networking,
+    DAG,
+    Mining,
+    Wallet,
+    Metrics,
+    Mempool,
+}
+
+#[derive(Debug, Error)]
+pub enum NamespaceError {
+    #[error("Invalid namespace value: {0}")]
+    InvalidValue(String),
+}
+
+impl FromStr for Namespace {
+    type Err = NamespaceError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "General" => Ok(Namespace::General),
+            "Networking" => Ok(Namespace::Networking),
+            "DAG" => Ok(Namespace::DAG),
+            "Mining" => Ok(Namespace::Mining),
+            "Wallet" => Ok(Namespace::Wallet),
+            "Metrics" => Ok(Namespace::Metrics),
+            "Mempool" => Ok(Namespace::Mempool),
+            _ => Err(NamespaceError::InvalidValue(s.to_string())),
+        }
+    }
+}
+
+impl std::fmt::Display for Namespace {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Namespace::General => write!(f, "General"),
+            Namespace::Networking => write!(f, "Networking"),
+            Namespace::DAG => write!(f, "DAG"),
+            Namespace::Mining => write!(f, "Mining"),
+            Namespace::Wallet => write!(f, "Wallet"),
+            Namespace::Metrics => write!(f, "Metrics"),
+            Namespace::Mempool => write!(f, "Mempool"),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct Namespaces {
+    enabled: HashSet<Namespace>,
+}
+
+impl Namespaces {
+    /// Check if a namespace is enabled
+    pub fn is_enabled(&self, namespace: &Namespace) -> bool {
+        self.enabled.contains(namespace)
+    }
+
+    // Determine the namespace associated with a given subscription scope
+    pub fn get_scope_namespace(&self, scope: &Scope) -> Namespace {
+        match scope {
+            Scope::BlockAdded(_) => Namespace::DAG,
+            Scope::VirtualChainChanged(_) => Namespace::DAG,
+            Scope::FinalityConflict(_) => Namespace::DAG,
+            Scope::FinalityConflictResolved(_) => Namespace::DAG,
+            Scope::UtxosChanged(_) => Namespace::Wallet,
+            Scope::SinkBlueScoreChanged(_) => Namespace::DAG,
+            Scope::VirtualDaaScoreChanged(_) => Namespace::DAG,
+            Scope::PruningPointUtxoSetOverride(_) => Namespace::DAG,
+            Scope::NewBlockTemplate(_) => Namespace::Mining,
+        }
+    }
+
+    /// Return enabled namespaces as string for get_info
+    pub fn enabled_namespaces(&self) -> Vec<String> {
+        self.enabled.iter().map(|namespace| namespace.to_string()).collect::<Vec<_>>()
+    }
+}
+
+impl FromStr for Namespaces {
+    type Err = NamespaceError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let enabled = s
+            .split(',')
+            .map(str::trim) // To support case like "DAG, Metrics"
+            .map(|name| name.parse::<Namespace>())
+            .collect::<Result<HashSet<_>, _>>()?;
+        Ok(Namespaces { enabled })
+    }
+}
+
+impl Default for Namespaces {
+    fn default() -> Self {
+        Self {
+            enabled: HashSet::from([
+                Namespace::General,
+                Namespace::Networking,
+                Namespace::DAG,
+                Namespace::Mining,
+                Namespace::Wallet,
+                Namespace::Metrics,
+                Namespace::Mempool,
+            ]),
+        }
+    }
+}

rpc/core/src/error.rs

@@ -60,6 +60,9 @@ pub enum RpcError {
     #[error("Transaction {0} not found")]
     TransactionNotFound(TransactionId),
 
+    #[error("Method unavailable. {0} namespace is not available.")]
+    UnauthorizedMethod(String),
+
     #[error("Method unavailable. Run the node with the --utxoindex argument.")]
     NoUtxoIndex,
 

rpc/core/src/model/message.rs

@@ -269,6 +269,7 @@ pub struct GetInfoResponse {
     pub server_version: String,
     pub is_utxo_indexed: bool,
     pub is_synced: bool,
+    pub namespaces: Vec<String>,
     pub has_notify_command: bool,
     pub has_message_id: bool,
 }
@@ -283,6 +284,7 @@ impl Serializer for GetInfoResponse {
         store!(bool, &self.is_synced, writer)?;
         store!(bool, &self.has_notify_command, writer)?;
         store!(bool, &self.has_message_id, writer)?;
+        store!(Vec<String>, &self.namespaces, writer)?;
 
         Ok(())
     }
@@ -298,8 +300,9 @@ impl Deserializer for GetInfoResponse {
         let is_synced = load!(bool, reader)?;
         let has_notify_command = load!(bool, reader)?;
         let has_message_id = load!(bool, reader)?;
+        let namespaces = load!(Vec<String>, reader)?;
 
-        Ok(Self { p2p_id, mempool_size, server_version, is_utxo_indexed, is_synced, has_notify_command, has_message_id })
+        Ok(Self { p2p_id, mempool_size, server_version, is_utxo_indexed, is_synced, namespaces, has_notify_command, has_message_id })
     }
 }
 

rpc/core/src/model/tests.rs

@@ -1,7 +1,7 @@
 #[cfg(test)]
 mod mockery {
 
-    use crate::{model::*, RpcScriptClass};
+    use crate::{api::namespaces::Namespaces, model::*, RpcScriptClass};
     use kaspa_addresses::{Prefix, Version};
     use kaspa_consensus_core::api::BlockCount;
     use kaspa_consensus_core::network::NetworkType;
@@ -486,6 +486,7 @@ mod mockery {
                 server_version: "0.4.2".to_string(),
                 is_utxo_indexed: true,
                 is_synced: false,
+                namespaces: Namespaces::default().enabled_namespaces(),
                 has_notify_command: true,
                 has_message_id: false,
             }

rpc/core/src/wasm/message.rs

@@ -243,6 +243,7 @@ declare! {
         serverVersion : string;
         isUtxoIndexed : boolean;
         isSynced : boolean;
+        namespaces : string[];
         /** GRPC ONLY */
         hasNotifyCommand : boolean;
         /** GRPC ONLY */

rpc/grpc/core/proto/rpc.proto

@@ -702,6 +702,7 @@ message GetInfoResponseMessage{
   string serverVersion = 3;
   bool isUtxoIndexed = 4;
   bool isSynced = 5;
+  repeated string namespaces = 6;
   bool hasNotifyCommand = 11;
   bool hasMessageId = 12;
   RPCError error = 1000;

rpc/grpc/core/src/convert/message.rs

@@ -183,6 +183,7 @@ from!(item: RpcResult<&kaspa_rpc_core::GetInfoResponse>, protowire::GetInfoRespo
         server_version: item.server_version.clone(),
         is_utxo_indexed: item.is_utxo_indexed,
         is_synced: item.is_synced,
+        namespaces: item.namespaces.clone(),
         has_notify_command: item.has_notify_command,
         has_message_id: item.has_message_id,
         error: None,
@@ -648,6 +649,7 @@ try_from!(item: &protowire::GetInfoResponseMessage, RpcResult<kaspa_rpc_core::Ge
         server_version: item.server_version.clone(),
         is_utxo_indexed: item.is_utxo_indexed,
         is_synced: item.is_synced,
+        namespaces: item.namespaces.clone(),
         has_notify_command: item.has_notify_command,
         has_message_id: item.has_message_id,
     }

rpc/grpc/server/src/tests/rpc_core_mock.rs

@@ -1,3 +1,4 @@
+use api::namespaces::Namespaces;
 use async_channel::{unbounded, Receiver};
 use async_trait::async_trait;
 use kaspa_notify::events::EVENT_TYPE_ARRAY;
@@ -73,6 +74,7 @@ impl RpcApi for RpcCoreMock {
             server_version: "mock".to_string(),
             is_utxo_indexed: false,
             is_synced: false,
+            namespaces: Namespaces::default().enabled_namespaces(),
             has_notify_command: true,
             has_message_id: true,
         })

rpc/macros/src/core/mod.rs

@@ -0,0 +1 @@
+pub mod service;

rpc/macros/src/core/service.rs

@@ -0,0 +1,19 @@
+use proc_macro::TokenStream;
+use quote::quote;
+use syn::{parse_macro_input, ItemFn, Path};
+
+pub fn namespace(attr: TokenStream, item: TokenStream) -> TokenStream {
+    let api_namespace = parse_macro_input!(attr as Path);
+    let mut func = parse_macro_input!(item as ItemFn);
+
+    let check = syn::parse2(quote! {
+        if !self.namespaces.is_enabled(&#api_namespace) {
+            // As macro processing happens after async_trait processing its wrapped with async_trait return type
+            return std::boxed::Box::pin(std::future::ready(Err(RpcError::UnauthorizedMethod(#api_namespace.to_string()))));
+        }
+    })
+    .unwrap();
+
+    func.block.stmts.insert(0, check);
+    quote!(#func).into()
+}

rpc/macros/src/lib.rs

@@ -1,5 +1,6 @@
 use proc_macro::TokenStream;
 use proc_macro_error::proc_macro_error;
+mod core;
 mod grpc;
 mod handler;
 mod wrpc;
@@ -45,3 +46,8 @@ pub fn build_grpc_server_interface(input: TokenStream) -> TokenStream {
 pub fn test_wrpc_serializer(input: TokenStream) -> TokenStream {
     wrpc::test::build_test(input)
 }
+
+#[proc_macro_attribute]
+pub fn namespace(attr: TokenStream, item: TokenStream) -> TokenStream {
+    core::service::namespace(attr, item)
+}

rpc/service/Cargo.toml

@@ -24,6 +24,7 @@ kaspa-p2p-flows.workspace = true
 kaspa-p2p-lib.workspace = true
 kaspa-perf-monitor.workspace = true
 kaspa-rpc-core.workspace = true
+kaspa-rpc-macros.workspace = true
 kaspa-txscript.workspace = true
 kaspa-utils.workspace = true
 kaspa-utils-tower.workspace = true
@@ -33,4 +34,6 @@ async-trait.workspace = true
 log.workspace = true
 tokio.workspace = true
 triggered.workspace = true
-workflow-rpc.workspace = true
+workflow-rpc.workspace = true
+thiserror.workspace = true
+serde.workspace = true