script/research/dam: sleep deprevation detected

script/research/dam/damd/src/lib.rs

@@ -0,0 +1,202 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2025 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
+
+use log::{debug, error, info};
+use smol::lock::Mutex;
+
+use darkfi::{
+    net::settings::Settings,
+    rpc::{
+        jsonrpc::JsonSubscriber,
+        server::{listen_and_serve, RequestHandler},
+        settings::RpcSettings,
+    },
+    system::{ExecutorPtr, StoppableTask, StoppableTaskPtr},
+    Error, Result,
+};
+
+/// JSON-RPC server methods
+mod rpc;
+
+/// P2P net protocols
+mod proto;
+use proto::{DamP2pHandler, DamP2pHandlerPtr};
+
+/// P2P network flooder
+mod flooder;
+use flooder::{DamFlooder, DamFlooderPtr};
+
+/// Atomic pointer to the Denial-of-service Analysis Multitool node
+pub type DamNodePtr = Arc<DamNode>;
+
+/// Structure representing a Denial-of-service Analysis Multitool node
+pub struct DamNode {
+    /// P2P network protocols handler.
+    p2p_handler: DamP2pHandlerPtr,
+    /// A map of various subscribers exporting live info from the node
+    subscribers: HashMap<&'static str, JsonSubscriber>,
+    /// JSON-RPC connection tracker
+    rpc_connections: Mutex<HashSet<StoppableTaskPtr>>,
+    /// Network flooder
+    flooder: DamFlooderPtr,
+}
+
+impl DamNode {
+    pub async fn new(
+        p2p_handler: DamP2pHandlerPtr,
+        subscribers: HashMap<&'static str, JsonSubscriber>,
+        flooder: DamFlooderPtr,
+    ) -> DamNodePtr {
+        Arc::new(Self {
+            p2p_handler,
+            subscribers,
+            rpc_connections: Mutex::new(HashSet::new()),
+            flooder,
+        })
+    }
+}
+
+/// Atomic pointer to the Denial-of-service Analysis Multitool daemon
+pub type DamdPtr = Arc<Damd>;
+
+/// Structure representing a Denial-of-service Analysis Multitool daemon
+pub struct Damd {
+    /// Darkfi node instance
+    node: DamNodePtr,
+    /// `dnet` background task
+    dnet_task: StoppableTaskPtr,
+    /// JSON-RPC background task
+    rpc_task: StoppableTaskPtr,
+}
+
+impl Damd {
+    /// Initialize a Denial-of-service Analysis Multitool daemon.
+    ///
+    /// Generates a new `DamNode` for provided configuration,
+    /// along with all the corresponding background tasks.
+    pub async fn init(net_settings: &Settings, ex: &ExecutorPtr) -> Result<DamdPtr> {
+        info!(target: "damd::Damd::init", "Initializing a Denial-of-service Analysis Multitool daemon...");
+
+        // Initialize P2P network
+        let p2p_handler = DamP2pHandler::init(net_settings, ex).await?;
+
+        // Here we initialize various subscribers that can export live network data.
+        let mut subscribers = HashMap::new();
+        subscribers.insert("dnet", JsonSubscriber::new("dnet.subscribe_events"));
+        subscribers.insert("foo", JsonSubscriber::new("protocols.subscribe_foo"));
+        subscribers.insert("attack_foo", JsonSubscriber::new("protocols.subscribe_attack_foo"));
+        subscribers.insert("bar", JsonSubscriber::new("protocols.subscribe_bar"));
+        subscribers.insert("attack_bar", JsonSubscriber::new("protocols.subscribe_attack_bar"));
+
+        // Initialize flooder
+        let flooder = DamFlooder::init(&p2p_handler.p2p, ex);
+
+        // Initialize node
+        let node = DamNode::new(p2p_handler, subscribers, flooder).await;
+
+        // Generate the background tasks
+        let dnet_task = StoppableTask::new();
+        let rpc_task = StoppableTask::new();
+
+        info!(target: "damd::Damd::init", "Denial-of-service Analysis Multitool daemon initialized successfully!");
+
+        Ok(Arc::new(Self { node, dnet_task, rpc_task }))
+    }
+
+    /// Start the Denial-of-service Analysis Multitool daemon in the given executor,
+    /// using the provided JSON-RPC configuration.
+    pub async fn start(&self, executor: &ExecutorPtr, rpc_settings: &RpcSettings) -> Result<()> {
+        info!(target: "damd::Damd::start", "Starting Denial-of-service Analysis Multitool daemon...");
+
+        // Start the `dnet` task
+        info!(target: "damd::Damd::start", "Starting dnet subs task");
+        let dnet_sub_ = self.node.subscribers.get("dnet").unwrap().clone();
+        let p2p_ = self.node.p2p_handler.p2p.clone();
+        self.dnet_task.clone().start(
+            async move {
+                let dnet_sub = p2p_.dnet_subscribe().await;
+                loop {
+                    let event = dnet_sub.receive().await;
+                    debug!(target: "damd::Damd::dnet_task", "Got dnet event: {:?}", event);
+                    dnet_sub_.notify(vec![event.into()].into()).await;
+                }
+            },
+            |res| async {
+                match res {
+                    Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                    Err(e) => {
+                        error!(target: "damd::Damd::start", "Failed starting dnet subs task: {}", e)
+                    }
+                }
+            },
+            Error::DetachedTaskStopped,
+            executor.clone(),
+        );
+
+        // Start the JSON-RPC task
+        info!(target: "damd::Damd::start", "Starting JSON-RPC server");
+        let node_ = self.node.clone();
+        self.rpc_task.clone().start(
+            listen_and_serve(rpc_settings.clone(), self.node.clone(), None, executor.clone()),
+            |res| async move {
+                match res {
+                    Ok(()) | Err(Error::RpcServerStopped) => node_.stop_connections().await,
+                    Err(e) => error!(target: "damd::Damd::start", "Failed starting JSON-RPC server: {}", e),
+                }
+            },
+            Error::RpcServerStopped,
+            executor.clone(),
+        );
+
+        // Start the P2P network
+        info!(target: "damd::Damd::start", "Starting P2P network");
+        self.node.p2p_handler.clone().start(executor, &self.node.subscribers).await?;
+
+        info!(target: "damd::Damd::start", "Denial-of-service Analysis Multitool daemon started successfully!");
+        Ok(())
+    }
+
+    /// Stop the Denial-of-service Analysis Multitool daemon.
+    pub async fn stop(&self) -> Result<()> {
+        info!(target: "damd::Damd::stop", "Terminating Denial-of-service Analysis Multitool daemon...");
+
+        // Stop the flooder
+        info!(target: "damd::Damd::stop", "Stopping the flooder...");
+        self.node.flooder.stop().await;
+
+        // Stop the `dnet` node
+        info!(target: "damd::Damd::stop", "Stopping dnet subs task...");
+        self.dnet_task.stop().await;
+
+        // Stop the JSON-RPC task
+        info!(target: "damd::Damd::stop", "Stopping JSON-RPC server...");
+        self.rpc_task.stop().await;
+
+        // Stop the P2P network
+        info!(target: "damd::Damd::stop", "Stopping P2P network protocols handler...");
+        self.node.p2p_handler.stop().await;
+
+        info!(target: "damd::Damd::stop", "Denial-of-service Analysis Multitool daemon terminated successfully!");
+        Ok(())
+    }
+}

script/research/dam/damd/src/proto/mod.rs

@@ -0,0 +1,123 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2025 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::{collections::HashMap, sync::Arc};
+
+use darkfi::{
+    net::{P2p, P2pPtr, Settings},
+    rpc::jsonrpc::JsonSubscriber,
+    system::ExecutorPtr,
+    Result,
+};
+use log::info;
+
+/// `Foo` messages broadcast protocol
+pub mod protocol_foo;
+pub use protocol_foo::{ProtocolFooHandler, ProtocolFooHandlerPtr};
+
+/// `Bar` messages broadcast protocol
+pub mod protocol_bar;
+pub use protocol_bar::{ProtocolBarHandler, ProtocolBarHandlerPtr};
+
+/// Atomic pointer to the Denial-of-service Analysis Multitool P2P protocols handler.
+pub type DamP2pHandlerPtr = Arc<DamP2pHandler>;
+
+/// Denial-of-service Analysis Multitool P2P protocols handler.
+pub struct DamP2pHandler {
+    /// P2P network pointer
+    pub p2p: P2pPtr,
+    /// `ProtocolFoo` messages handler
+    foo_handler: ProtocolFooHandlerPtr,
+    /// `ProtocolBar` messages handler
+    bar_handler: ProtocolBarHandlerPtr,
+}
+
+impl DamP2pHandler {
+    /// Initialize a Denial-of-service Analysis Multitool P2P protocols handler.
+    ///
+    /// A new P2P instance is generated using provided settings and all
+    /// corresponding protocols are registered.
+    pub async fn init(settings: &Settings, executor: &ExecutorPtr) -> Result<DamP2pHandlerPtr> {
+        info!(
+            target: "damd::proto::mod::DamP2pHandler::init",
+            "Initializing a new Denial-of-service Analysis Multitool P2P handler..."
+        );
+
+        // Generate a new P2P instance
+        let p2p = P2p::new(settings.clone(), executor.clone()).await?;
+
+        // Generate a new `ProtocolFoo` messages handler
+        let foo_handler = ProtocolFooHandler::init(&p2p).await;
+
+        // Generate a new `ProtocolBar` messages handler
+        let bar_handler = ProtocolBarHandler::init(&p2p).await;
+
+        info!(
+            target: "damd::proto::mod::DamP2pHandler::init",
+            "Denial-of-service Analysis Multitool P2P handler generated successfully!"
+        );
+
+        Ok(Arc::new(Self { p2p, foo_handler, bar_handler }))
+    }
+
+    /// Start the Denial-of-service Analysis Multitool P2P protocols handler.
+    pub async fn start(
+        &self,
+        executor: &ExecutorPtr,
+        subscribers: &HashMap<&'static str, JsonSubscriber>,
+    ) -> Result<()> {
+        info!(
+            target: "damd::proto::mod::DamP2pHandler::start",
+            "Starting the Denial-of-service Analysis Multitool P2P handler..."
+        );
+
+        // Start the `ProtocolFoo` messages handler
+        let subscriber = subscribers.get("foo").unwrap().clone();
+        self.foo_handler.start(executor, subscriber).await?;
+
+        // Start the `ProtocolBar` messages handler
+        let subscriber = subscribers.get("bar").unwrap().clone();
+        self.bar_handler.start(executor, subscriber).await?;
+
+        // Start the P2P instance
+        self.p2p.clone().start().await?;
+
+        info!(
+            target: "damd::proto::mod::DamP2pHandler::start",
+            "Denial-of-service Analysis Multitool P2P handler started successfully!"
+        );
+
+        Ok(())
+    }
+
+    /// Stop the Denial-of-service Analysis P2P protocols handler.
+    pub async fn stop(&self) {
+        info!(target: "damd::proto::mod::DamP2pHandler::stop", "Terminating Denial-of-service Analysis Multitool P2P handler...");
+
+        // Stop the P2P instance
+        self.p2p.stop().await;
+
+        // Start the `ProtocolFoo` messages handler
+        self.foo_handler.stop().await;
+
+        // Start the `ProtocolBar` messages handler
+        self.bar_handler.stop().await;
+
+        info!(target: "damd::proto::mod::DamP2pHandler::stop", "Denial-of-service Analysis Multitool P2P handler terminated successfully!");
+    }
+}

script/research/dam/damd/src/proto/protocol_bar.rs

@@ -0,0 +1,135 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2025 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use log::{debug, error, info};
+use tinyjson::JsonValue;
+
+use darkfi::{
+    impl_p2p_message,
+    net::{
+        protocol::protocol_generic::{
+            ProtocolGenericAction, ProtocolGenericHandler, ProtocolGenericHandlerPtr,
+        },
+        session::SESSION_DEFAULT,
+        Message, P2pPtr,
+    },
+    rpc::jsonrpc::JsonSubscriber,
+    system::ExecutorPtr,
+    Error, Result,
+};
+use darkfi_serial::{SerialDecodable, SerialEncodable};
+
+/// Structure represening a bar message
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
+pub struct Bar {
+    /// Bar message
+    pub message: String,
+}
+
+impl_p2p_message!(Bar, "bar");
+
+/// Atomic pointer to the `ProtocolBar` handler.
+pub type ProtocolBarHandlerPtr = Arc<ProtocolBarHandler>;
+
+/// Handler managing `ProtocolBar` messages, over a generic P2P protocol.
+pub struct ProtocolBarHandler {
+    /// The generic handler for `ProtocolBar` messages.
+    handler: ProtocolGenericHandlerPtr<Bar, Bar>,
+}
+
+impl ProtocolBarHandler {
+    /// Initialize a generic prototocol handler for `ProtocolBar` messages
+    /// and registers it to the provided P2P network, using the default session flag.
+    pub async fn init(p2p: &P2pPtr) -> ProtocolBarHandlerPtr {
+        debug!(
+            target: "damd::proto::protocol_bar::init",
+            "Adding ProtocolBar to the protocol registry"
+        );
+
+        let handler = ProtocolGenericHandler::new(p2p, "ProtocolBar", SESSION_DEFAULT).await;
+
+        Arc::new(Self { handler })
+    }
+
+    /// Start the `ProtocolBar` background task.
+    pub async fn start(&self, executor: &ExecutorPtr, subscriber: JsonSubscriber) -> Result<()> {
+        debug!(
+            target: "damd::proto::protocol_bar::start",
+            "Starting ProtocolBar handler task..."
+        );
+
+        self.handler.task.clone().start(
+            handle_receive_bar(self.handler.clone(), subscriber),
+            |res| async move {
+                match res {
+                    Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                    Err(e) => error!(target: "damd::proto::protocol_bar::start", "Failed starting ProtocolBar handler task: {e}"),
+                }
+            },
+            Error::DetachedTaskStopped,
+            executor.clone(),
+        );
+
+        debug!(
+            target: "damd::proto::protocol_bar::start",
+            "ProtocolBar handler task started!"
+        );
+
+        Ok(())
+    }
+
+    /// Stop the `ProtocolBar` background task.
+    pub async fn stop(&self) {
+        debug!(target: "damd::proto::protocol_bar::stop", "Terminating ProtocolBar handler task...");
+        self.handler.task.stop().await;
+        debug!(target: "damd::proto::protocol_bar::stop", "ProtocolBar handler task terminated!");
+    }
+}
+
+/// Background handler function for ProtocolBar.
+async fn handle_receive_bar(
+    handler: ProtocolGenericHandlerPtr<Bar, Bar>,
+    subscriber: JsonSubscriber,
+) -> Result<()> {
+    debug!(target: "damd::proto::protocol_bar::handle_receive_bar", "START");
+    loop {
+        // Wait for a new bar message
+        let (channel, bar) = match handler.receiver.recv().await {
+            Ok(r) => r,
+            Err(e) => {
+                debug!(
+                    target: "damd::proto::protocol_bar::handle_receive_bar",
+                    "recv fail: {e}"
+                );
+                continue
+            }
+        };
+
+        let notification = format!("Received bar message from {channel}: {}", bar.message);
+        info!(target: "damd::proto::protocol_bar::handle_receive_bar", "{notification}");
+
+        // Notify subscriber
+        subscriber.notify(vec![JsonValue::String(notification)].into()).await;
+
+        // Signal handler to broadcast the message to rest nodes
+        handler.send_action(channel, ProtocolGenericAction::Broadcast).await;
+    }
+}

script/research/dam/damd/src/proto/protocol_foo.rs

@@ -0,0 +1,146 @@
+/* This file is part of DarkFi (https://dark.fi)
+ *
+ * Copyright (C) 2020-2025 Dyne.org foundation
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use log::{debug, error, info};
+use tinyjson::JsonValue;
+
+use darkfi::{
+    impl_p2p_message,
+    net::{
+        protocol::protocol_generic::{
+            ProtocolGenericAction, ProtocolGenericHandler, ProtocolGenericHandlerPtr,
+        },
+        session::SESSION_DEFAULT,
+        Message, P2pPtr,
+    },
+    rpc::jsonrpc::JsonSubscriber,
+    system::ExecutorPtr,
+    Error, Result,
+};
+use darkfi_serial::{SerialDecodable, SerialEncodable};
+
+/// Structure represening a foo request.
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
+pub struct FooRequest {
+    /// Request message
+    pub message: String,
+}
+
+impl_p2p_message!(FooRequest, "foorequest");
+
+/// Structure representing the response to `FooRequest`.
+#[derive(Clone, Debug, SerialEncodable, SerialDecodable)]
+pub struct FooResponse {
+    /// Response code
+    pub code: u8,
+}
+
+impl_p2p_message!(FooResponse, "fooresponse");
+
+/// Atomic pointer to the `ProtocolFoo` handler.
+pub type ProtocolFooHandlerPtr = Arc<ProtocolFooHandler>;
+
+/// Handler managing all `ProtocolFoo` messages, over generic P2P protocols.
+pub struct ProtocolFooHandler {
+    /// The generic handler for `FooRequest` messages.
+    handler: ProtocolGenericHandlerPtr<FooRequest, FooResponse>,
+}
+
+impl ProtocolFooHandler {
+    /// Initialize the generic prototocol handlers for all `ProtocolFoo` messages
+    /// and register them to the provided P2P network, using the default session flag.
+    pub async fn init(p2p: &P2pPtr) -> ProtocolFooHandlerPtr {
+        debug!(
+            target: "damd::proto::protocol_foo::init",
+            "Adding all foo protocols to the protocol registry"
+        );
+
+        let handler = ProtocolGenericHandler::new(p2p, "ProtocolFoo", SESSION_DEFAULT).await;
+
+        Arc::new(Self { handler })
+    }
+
+    /// Start all `ProtocolFoo` background tasks.
+    pub async fn start(&self, executor: &ExecutorPtr, subscriber: JsonSubscriber) -> Result<()> {
+        debug!(
+            target: "damd::proto::protocol_foo::start",
+            "Starting foo protocols handlers tasks..."
+        );
+
+        self.handler.task.clone().start(
+            handle_receive_foo_request(self.handler.clone(), subscriber),
+            |res| async move {
+                match res {
+                    Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                    Err(e) => error!(target: "damd::proto::protocol_foo::start", "Failed starting ProtocolFoo handler task: {e}"),
+                }
+            },
+            Error::DetachedTaskStopped,
+            executor.clone(),
+        );
+
+        debug!(
+            target: "damd::proto::protocol_foo::start",
+            "Foo protocols handlers tasks started!"
+        );
+
+        Ok(())
+    }
+
+    /// Stop all `ProtocolSync` background tasks.
+    pub async fn stop(&self) {
+        debug!(target: "damd::proto::protocol_foo::stop", "Terminating foo protocols handlers tasks...");
+        self.handler.task.stop().await;
+        debug!(target: "damd::proto::protocol_foo::stop", "Foo protocols handlers tasks terminated!");
+    }
+}
+
+/// Background handler function for ProtocolFoo.
+async fn handle_receive_foo_request(
+    handler: ProtocolGenericHandlerPtr<FooRequest, FooResponse>,
+    subscriber: JsonSubscriber,
+) -> Result<()> {
+    debug!(target: "damd::proto::protocol_foo::handle_receive_foo_request", "START");
+    loop {
+        // Wait for a new foo request message
+        let (channel, request) = match handler.receiver.recv().await {
+            Ok(r) => r,
+            Err(e) => {
+                debug!(
+                    target: "damd::proto::protocol_foo::handle_receive_foo_request",
+                    "recv fail: {e}"
+                );
+                continue
+            }
+        };
+
+        let notification = format!("Received foo request from {channel}: {}", request.message);
+        info!(target: "damd::proto::protocol_foo::handle_receive_foo_request", "{notification}");
+
+        // Notify subscriber
+        subscriber.notify(vec![JsonValue::String(notification)].into()).await;
+
+        // Send response
+        handler
+            .send_action(channel, ProtocolGenericAction::Response(FooResponse { code: 42 }))
+            .await;
+    }
+}