script/research/dam: Denial-of-service Analysis Multitool added

script/research/dam/README.md

@@ -0,0 +1,25 @@
+dam
+=======
+
+Denial-of-service Analysis Multitool.<br>
+This is a suite of tools to simulate flooding attacks on a
+P2P network, to verify and fine tune protection mechanisms
+against them.<br>
+A daemon, a command-line client and a localnet script are
+provided.
+
+## damd
+
+Dummy daemon implementing some P2P communication protocols,
+along with JSON-RPC endpoints to simulate flooding attacks
+over the network.
+
+## dam-cli
+
+Command-line client for `damd`, to trigger flooding attacks
+and monitor responses.
+
+## dam-localnet
+
+Localnet folder with script and configuration to deploy
+instances to test with.

script/research/dam/dam-cli/.gitignore

@@ -0,0 +1,4 @@
+/target
+Cargo.lock
+rustfmt.toml
+dam-cli

script/research/dam/dam-cli/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "dam-cli"
+version = "0.4.1"
+description = "CLI-utility to control a Denial-of-service Analysis Multitool daemon."
+authors = ["Dyne.org foundation <foundation@dyne.org>"]
+repository = "https://codeberg.org/darkrenaissance/darkfi"
+license = "AGPL-3.0-only"
+edition = "2021"
+
+[workspace]
+
+[dependencies]
+# Darkfi
+darkfi = {path = "../../../../", features = ["async-sdk", "rpc"]}
+darkfi-serial = "0.4.2"
+
+# Misc
+clap = {version = "4.4.11", features = ["derive"]}
+smol = "2.0.2"
+url = "2.5.4"

script/research/dam/dam-cli/Makefile

@@ -0,0 +1,37 @@
+.POSIX:
+
+# Install prefix
+PREFIX = $(HOME)/.cargo
+
+# Cargo binary
+CARGO = cargo +nightly
+
+# Compile target
+RUST_TARGET = $(shell rustc -Vv | grep '^host: ' | cut -d' ' -f2)
+# Uncomment when doing musl static builds
+#RUSTFLAGS = -C target-feature=+crt-static -C link-self-contained=yes
+
+SRC = \
+	Cargo.toml \
+	$(shell find src -type f -name '*.rs') \
+
+BIN = $(shell grep '^name = ' Cargo.toml | cut -d' ' -f3 | tr -d '"')
+
+all: $(BIN)
+
+$(BIN): $(SRC)
+	RUSTFLAGS="$(RUSTFLAGS)" $(CARGO) build --target=$(RUST_TARGET) --release --package $@
+	cp -f target/$(RUST_TARGET)/release/$@ $@
+
+fmt:
+	$(CARGO) fmt --all
+
+clippy:
+	RUSTFLAGS="$(RUSTFLAGS)" $(CARGO) clippy --target=$(RUST_TARGET) \
+		--release --all-features --workspace --tests
+
+clean:
+	RUSTFLAGS="$(RUSTFLAGS)" $(CARGO) clean --target=$(RUST_TARGET) --release --package $(BIN)
+	rm -f $(BIN)
+
+.PHONY: all fmt clippy clean

script/research/dam/dam-cli/src/lib.rs

@@ -0,0 +1,39 @@
+/* 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 darkfi::{rpc::client::RpcClient, system::ExecutorPtr, Result};
+use url::Url;
+
+/// damd JSON-RPC related methods
+pub mod rpc;
+
+/// CLI-util structure
+pub struct DamCli {
+    /// JSON-RPC client to execute requests to damd daemon
+    pub rpc_client: RpcClient,
+}
+
+impl DamCli {
+    pub async fn new(endpoint: &str, ex: &ExecutorPtr) -> Result<Self> {
+        // Initialize rpc client
+        let endpoint = Url::parse(endpoint)?;
+        let rpc_client = RpcClient::new(endpoint, ex.clone()).await?;
+
+        Ok(Self { rpc_client })
+    }
+}

script/research/dam/dam-cli/src/main.rs

@@ -0,0 +1,98 @@
+/* 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 clap::{Parser, Subcommand};
+use darkfi::{cli_desc, rpc::util::JsonValue, Result};
+use smol::Executor;
+
+use dam_cli::DamCli;
+
+#[derive(Parser)]
+#[command(about = cli_desc!())]
+struct Args {
+    #[arg(short, long, default_value = "tcp://127.0.0.1:34780")]
+    /// damd JSON-RPC endpoint
+    endpoint: String,
+
+    #[command(subcommand)]
+    /// Sub command to execute
+    command: Subcmd,
+}
+
+#[derive(Subcommand)]
+enum Subcmd {
+    /// Send a ping request to the damd RPC endpoint
+    Ping,
+
+    /// This subscription will listen for incoming notifications from damd
+    Subscribe {
+        /// The method to subscribe to
+        method: String,
+    },
+
+    /// Signal damd to execute a flooding attack against the network
+    Flood,
+
+    /// Signal damd to stop an ongoing flooding attack
+    StopFlood,
+}
+
+fn main() -> Result<()> {
+    // Initialize an executor
+    let executor = Arc::new(Executor::new());
+    let ex = executor.clone();
+    smol::block_on(executor.run(async {
+        // Parse arguments
+        let args = Args::parse();
+
+        // Execute a subcommand
+        let dam_cli = DamCli::new(&args.endpoint, &ex).await?;
+        match args.command {
+            Subcmd::Ping => {
+                dam_cli.ping().await?;
+            }
+
+            Subcmd::Subscribe { method } => {
+                dam_cli.subscribe(&args.endpoint, &method, &ex).await?;
+            }
+
+            Subcmd::Flood => {
+                dam_cli
+                    .damd_daemon_request(
+                        "flood.switch",
+                        &JsonValue::Array(vec![JsonValue::Boolean(true)]),
+                    )
+                    .await?;
+            }
+
+            Subcmd::StopFlood => {
+                dam_cli
+                    .damd_daemon_request(
+                        "flood.switch",
+                        &JsonValue::Array(vec![JsonValue::Boolean(false)]),
+                    )
+                    .await?;
+            }
+        }
+        dam_cli.rpc_client.stop().await;
+
+        Ok(())
+    }))
+}

script/research/dam/dam-cli/src/rpc.rs

@@ -0,0 +1,136 @@
+/* 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::time::Instant;
+
+use darkfi::{
+    rpc::{
+        client::RpcClient,
+        jsonrpc::{ErrorCode, JsonError, JsonRequest, JsonResult},
+        util::JsonValue,
+    },
+    system::{ExecutorPtr, Publisher, StoppableTask},
+    Error, Result,
+};
+use url::Url;
+
+use crate::DamCli;
+
+impl DamCli {
+    /// Auxiliary function to ping configured damd daemon for liveness.
+    pub async fn ping(&self) -> Result<()> {
+        println!("Executing ping request to damd...");
+        let latency = Instant::now();
+        let rep = self.damd_daemon_request("ping", &JsonValue::Array(vec![])).await?;
+        let latency = latency.elapsed();
+        println!("Got reply: {rep:?}");
+        println!("Latency: {latency:?}");
+        Ok(())
+    }
+
+    /// Auxiliary function to execute a request towards the configured damd daemon JSON-RPC endpoint.
+    pub async fn damd_daemon_request(&self, method: &str, params: &JsonValue) -> Result<JsonValue> {
+        let req = JsonRequest::new(method, params.clone());
+        let rep = self.rpc_client.request(req).await?;
+        Ok(rep)
+    }
+
+    /// Subscribes to damd's JSON-RPC notification endpoints.
+    pub async fn subscribe(&self, endpoint: &str, method: &str, ex: &ExecutorPtr) -> Result<()> {
+        println!("Subscribing to receive notifications for: {method}");
+        let endpoint = Url::parse(endpoint)?;
+        let _method = String::from(method);
+        let publisher = Publisher::new();
+        let subscription = publisher.clone().subscribe().await;
+        let _publisher = publisher.clone();
+        let _ex = ex.clone();
+        StoppableTask::new().start(
+            // Weird hack to prevent lifetimes hell
+            async move {
+                let rpc_client = RpcClient::new(endpoint, _ex).await?;
+                let req = JsonRequest::new(&_method, JsonValue::Array(vec![]));
+                rpc_client.subscribe(req, _publisher).await
+            },
+            |res| async move {
+                match res {
+                    Ok(()) => { /* Do nothing */ }
+                    Err(e) => {
+                        eprintln!("[subscribe] JSON-RPC server error: {e:?}");
+                        publisher
+                            .notify(JsonResult::Error(JsonError::new(
+                                ErrorCode::InternalError,
+                                None,
+                                0,
+                            )))
+                            .await;
+                    }
+                }
+            },
+            Error::RpcServerStopped,
+            ex.clone(),
+        );
+        println!("Detached subscription to background");
+        println!("All is good. Waiting for new notifications...");
+
+        let e = loop {
+            match subscription.receive().await {
+                JsonResult::Notification(n) => {
+                    println!("Got notification from subscription");
+                    if n.method != method {
+                        break Error::UnexpectedJsonRpc(format!(
+                            "Got foreign notification from damd: {}",
+                            n.method
+                        ))
+                    }
+
+                    // Verify parameters
+                    if !n.params.is_array() {
+                        break Error::UnexpectedJsonRpc(
+                            "Received notification params are not an array".to_string(),
+                        )
+                    }
+                    let params = n.params.get::<Vec<JsonValue>>().unwrap();
+                    if params.is_empty() {
+                        break Error::UnexpectedJsonRpc(
+                            "Notification parameters are empty".to_string(),
+                        )
+                    }
+
+                    for param in params {
+                        let param = param.get::<String>().unwrap();
+                        println!("Notification: {param}");
+                    }
+                }
+
+                JsonResult::Error(e) => {
+                    // Some error happened in the transmission
+                    break Error::UnexpectedJsonRpc(format!("Got error from JSON-RPC: {e:?}"))
+                }
+
+                x => {
+                    // And this is weird
+                    break Error::UnexpectedJsonRpc(format!(
+                        "Got unexpected data from JSON-RPC: {x:?}"
+                    ))
+                }
+            }
+        };
+
+        Err(e)
+    }
+}

script/research/dam/dam-localnet/.gitignore

@@ -0,0 +1,2 @@
+damd0
+damd1

script/research/dam/dam-localnet/README.md

@@ -0,0 +1,9 @@
+dam localnet
+================
+
+This will start two `damd` node instances in localnet mode.
+The first node is considered the defender, and we will listen
+to its incoming messages, while the second one is the attacker,
+so we will listen to its outgoing messages.
+Second node can be queried to start attacking the other one,
+using `dam-cli`.

script/research/dam/dam-localnet/clean.sh

@@ -0,0 +1,2 @@
+#!/bin/sh
+rm -rf damd0 damd1

script/research/dam/dam-localnet/damd0.toml

@@ -0,0 +1,35 @@
+## damd configuration file
+##
+## Please make sure you go through all the settings so you can configure
+## your daemon properly.
+##
+## The default values are left commented. They can be overridden either by
+## uncommenting, or by using the command-line.
+
+# JSON-RPC settings
+[rpc]
+# JSON-RPC listen URL
+rpc_listen = "tcp://127.0.0.1:44780"
+
+# Disabled RPC methods
+#rpc_disabled_methods = ["p2p.get_info"]
+
+# P2P network settings
+[net]
+# Path to the P2P datastore
+p2p_datastore = "damd0"
+
+# Path to a configured hostlist for saving known peers
+hostlist = "damd0/p2p_hostlist.tsv"
+
+# P2P accept addresses the instance listens on for inbound connections
+inbound = ["tcp+tls://0.0.0.0:44781"]
+
+# Peer nodes to manually connect to
+peers = ["tcp+tls://0.0.0.0:44881"]
+
+# Whitelisted network transports for outbound connections
+allowed_transports = ["tcp+tls"]
+
+# Allow localnet hosts
+localnet = true

script/research/dam/dam-localnet/damd1.toml

@@ -0,0 +1,35 @@
+## damd configuration file
+##
+## Please make sure you go through all the settings so you can configure
+## your daemon properly.
+##
+## The default values are left commented. They can be overridden either by
+## uncommenting, or by using the command-line.
+
+# JSON-RPC settings
+[rpc]
+# JSON-RPC listen URL
+rpc_listen = "tcp://127.0.0.1:44880"
+
+# Disabled RPC methods
+#rpc_disabled_methods = ["p2p.get_info"]
+
+# P2P network settings
+[net]
+# Path to the P2P datastore
+p2p_datastore = "damd1"
+
+# Path to a configured hostlist for saving known peers
+hostlist = "damd1/p2p_hostlist.tsv"
+
+# P2P accept addresses the instance listens on for inbound connections
+inbound = ["tcp+tls://0.0.0.0:44881"]
+
+# Peer nodes to manually connect to
+peers = ["tcp+tls://0.0.0.0:44781"]
+
+# Whitelisted network transports for outbound connections
+allowed_transports = ["tcp+tls"]
+
+# Allow localnet hosts
+localnet = true

script/research/dam/dam-localnet/tmux_sessions.sh

@@ -0,0 +1,42 @@
+#!/bin/sh
+set -e
+
+# Start a tmux session with two damd nodes.
+# Additionally, start the corresponding subscribers
+# for each node and prepare a pane to start an attack.
+
+# Path to used binaries
+DAMD="../damd/damd"
+DAMD_CLI="../dam-cli/dam-cli"
+DAMD_CLI0="$DAMD_CLI -e tcp://127.0.0.1:44780"
+DAMD_CLI1="$DAMD_CLI -e tcp://127.0.0.1:44880"
+
+session=damd-localnet
+
+if [ "$1" = "-vv" ]; then
+	verbose="-vv"
+	shift
+else
+	verbose=""
+fi
+
+tmux new-session -d -s $session -n "node0"
+tmux send-keys -t $session "$DAMD $verbose -c damd0.toml" Enter
+tmux new-window -t $session -n "node1"
+tmux send-keys -t $session "$DAMD $verbose -c damd1.toml" Enter
+sleep 1
+tmux new-window -t $session -n "flood"
+tmux send-keys -t $session "$DAMD_CLI0 subscribe protocols.subscribe_foo" Enter
+tmux split-window -t $session -v -l 20%
+tmux send-keys -t $session "$DAMD_CLI1 flood"
+tmux select-pane -t 0
+tmux split-window -t $session -h
+tmux send-keys -t $session "$DAMD_CLI1 subscribe protocols.subscribe_attack_foo" Enter
+tmux select-pane -t 0
+tmux split-window -t $session -v
+tmux send-keys -t $session "$DAMD_CLI0 subscribe protocols.subscribe_bar" Enter
+tmux select-pane -t 2
+tmux split-window -t $session -v
+tmux send-keys -t $session "$DAMD_CLI1 subscribe protocols.subscribe_attack_bar" Enter
+tmux select-pane -t 4
+tmux attach -t $session

script/research/dam/damd/.gitignore

@@ -0,0 +1,4 @@
+/target
+Cargo.lock
+rustfmt.toml
+damd

script/research/dam/damd/Cargo.toml

@@ -0,0 +1,36 @@
+[package]
+name = "damd"
+version = "0.4.1"
+description = "Denial-of-service Analysis Multitool daemon."
+authors = ["Dyne.org foundation <foundation@dyne.org>"]
+repository = "https://codeberg.org/darkrenaissance/darkfi"
+license = "AGPL-3.0-only"
+edition = "2021"
+
+[workspace]
+
+[dependencies]
+# Darkfi
+darkfi = {path = "../../../../", features = ["async-daemonize", "rpc"]}
+darkfi-serial = "0.4.2"
+
+# Misc
+log = "0.4.25"
+
+# JSON-RPC
+async-trait = "0.1.86"
+tinyjson = "2.5.1"
+url = "2.5.4"
+
+# Daemon
+async-std = {version = "1.13.0", features = ["attributes"]}
+easy-parallel = "3.3.1"
+signal-hook-async-std = "0.2.2"
+signal-hook = "0.3.17"
+simplelog = "0.12.2"
+smol = "2.0.2"
+
+# Argument parsing
+serde = {version = "1.0.217", features = ["derive"]}
+structopt = "0.3.26"
+structopt-toml = "0.5.1"

script/research/dam/damd/Makefile

@@ -0,0 +1,37 @@
+.POSIX:
+
+# Install prefix
+PREFIX = $(HOME)/.cargo
+
+# Cargo binary
+CARGO = cargo +nightly
+
+# Compile target
+RUST_TARGET = $(shell rustc -Vv | grep '^host: ' | cut -d' ' -f2)
+# Uncomment when doing musl static builds
+#RUSTFLAGS = -C target-feature=+crt-static -C link-self-contained=yes
+
+SRC = \
+	Cargo.toml \
+	$(shell find src -type f -name '*.rs') \
+
+BIN = $(shell grep '^name = ' Cargo.toml | cut -d' ' -f3 | tr -d '"')
+
+all: $(BIN)
+
+$(BIN): $(SRC)
+	RUSTFLAGS="$(RUSTFLAGS)" $(CARGO) build --target=$(RUST_TARGET) --release --package $@
+	cp -f target/$(RUST_TARGET)/release/$@ $@
+
+fmt:
+	$(CARGO) fmt --all
+
+clippy:
+	RUSTFLAGS="$(RUSTFLAGS)" $(CARGO) clippy --target=$(RUST_TARGET) \
+		--release --all-features --workspace --tests
+
+clean:
+	RUSTFLAGS="$(RUSTFLAGS)" $(CARGO) clean --target=$(RUST_TARGET) --release --package $(BIN)
+	rm -f $(BIN)
+
+.PHONY: all fmt clippy clean

script/research/dam/damd/damd_config.toml

@@ -0,0 +1,78 @@
+## damd configuration file
+##
+## Please make sure you go through all the settings so you can configure
+## your daemon properly.
+##
+## The default values are left commented. They can be overridden either by
+## uncommenting, or by using the command-line.
+
+# JSON-RPC settings
+[rpc]
+# JSON-RPC listen URL
+rpc_listen = "tcp://127.0.0.1:34780"
+
+# Disabled RPC methods
+rpc_disabled_methods = ["p2p.get_info"]
+
+# P2P network settings
+[net]
+# Path to the P2P datastore
+p2p_datastore = "~/.local/share/darkfi/damd"
+
+# Path to a configured hostlist for saving known peers
+hostlist = "~/.local/share/darkfi/damd/p2p_hostlist.tsv"
+
+# P2P accept addresses the instance listens on for inbound connections
+#inbound = ["tcp+tls://0.0.0.0:34781"]
+
+# P2P external addresses the instance advertises so other peers can
+# reach us and connect to us, as long as inbound addrs are configured.
+#external_addrs = []
+
+# Peer nodes to manually connect to
+#peers = []
+
+# Seed nodes to connect to for peer discovery and/or adversising our
+# own external addresses
+#seeds = []
+
+# Whitelisted network transports for outbound connections
+#allowed_transports = ["tcp+tls"]
+
+# Allow transport mixing (e.g. Tor would be allowed to connect to `tcp://`)
+#transport_mixing = true
+
+# Outbound connection slots number, this many connections will be
+# attempted. (This does not include manual connections)
+#outbound_connections = 8
+
+# Inbound connections slots number, this many active inbound connections
+# will be allowed. (This does not include manual or outbound connections)
+#inbound_connections = 8
+
+## White connection percent
+# gold_connect_count = 2
+
+## White connection percent
+# white_connect_percent = 70
+
+# Manual connections retry limit, 0 for forever looping
+#manual_attempt_limit = 0
+
+# Outbound connection timeout (in seconds)
+#outbound_connect_timeout = 10
+
+# Exchange versions (handshake) timeout (in seconds)
+#channel_handshake_timeout = 4
+
+# Ping-pong exchange execution interval (in seconds)
+#channel_heartbeat_interval = 10
+
+# Allow localnet hosts
+localnet = false
+
+# Cooling off time for peer discovery when unsuccessful
+#outbound_peer_discovery_cooloff_time = 30
+
+# Time between peer discovery attempts
+#outbound_peer_discovery_attempt_time = 5

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

@@ -0,0 +1,209 @@
+/* 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 darkfi::{
+    net::{channel::ChannelPtr, P2pPtr},
+    rpc::jsonrpc::JsonSubscriber,
+    system::{ExecutorPtr, StoppableTask, StoppableTaskPtr},
+    Error, Result,
+};
+use log::{debug, error, info};
+use smol::lock::Mutex;
+use tinyjson::JsonValue;
+
+use crate::proto::{
+    protocol_bar::Bar,
+    protocol_foo::{FooRequest, FooResponse},
+};
+
+/// Atomic pointer to the Denial-of-service Analysis Multitool flooder.
+pub type DamFlooderPtr = Arc<DamFlooder>;
+
+/// Denial-of-service Analysis Multitool flooder.
+pub struct DamFlooder {
+    /// P2P network pointer
+    p2p: P2pPtr,
+    /// Executor to spawn flooding tasks
+    executor: ExecutorPtr,
+    /// Set to keep track of all the spawned tasks
+    tasks: Arc<Mutex<HashSet<StoppableTaskPtr>>>,
+}
+
+impl DamFlooder {
+    /// Initialize a Denial-of-service Analysis Multitool flooder.
+    pub fn init(p2p: &P2pPtr, ex: &ExecutorPtr) -> DamFlooderPtr {
+        Arc::new(Self {
+            p2p: p2p.clone(),
+            executor: ex.clone(),
+            tasks: Arc::new(Mutex::new(HashSet::new())),
+        })
+    }
+
+    /// Start the Denial-of-service Analysis Multitool flooder.
+    pub async fn start(&self, subscribers: &HashMap<&'static str, JsonSubscriber>) {
+        info!(
+            target: "damd::flooder::DamFlooder::start",
+            "Starting the Denial-of-service Analysis Multitool flooder..."
+        );
+
+        // Check if tasks already exist
+        let mut lock = self.tasks.lock().await;
+        if !lock.is_empty() {
+            info!(
+                target: "damd::flooder::DamFlooder::start",
+                "Denial-of-service Analysis Multitool flooder already started!"
+            );
+            return
+        }
+
+        // Spawn a task for each connected peer for `Foo` messages, since we expect responses
+        for peer in self.p2p.hosts().channels() {
+            let task = StoppableTask::new();
+            task.clone().start(
+                flood_foo(self.p2p.settings().read().await.outbound_connect_timeout, peer, subscribers.get("attack_foo").unwrap().clone()),
+                |res| async move {
+                    match res {
+                        Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                        Err(e) => error!(target: "damd::Damd::start", "Failed starting flood foo task: {e}")
+                    }
+                },
+                Error::DetachedTaskStopped,
+                self.executor.clone(),
+            );
+            lock.insert(task);
+        }
+
+        // Spawn a task for `Bar` messages to broadcast to everyone
+        let task = StoppableTask::new();
+        task.clone().start(
+            flood_bar(self.p2p.clone(), subscribers.get("attack_bar").unwrap().clone()),
+            |res| async move {
+                match res {
+                    Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
+                    Err(e) => {
+                        error!(target: "damd::Damd::start", "Failed starting flood bar task: {e}")
+                    }
+                }
+            },
+            Error::DetachedTaskStopped,
+            self.executor.clone(),
+        );
+        lock.insert(task);
+
+        info!(
+            target: "damd::flooder::DamFlooder::start",
+            "Denial-of-service Analysis Multitool flooder started successfully!"
+        );
+    }
+
+    /// Stop the Denial-of-service Analysis flooder.
+    pub async fn stop(&self) {
+        info!(target: "damd::flooder::DamFlooder::stop", "Terminating Denial-of-service Analysis Multitool flooder...");
+
+        // Check if tasks already terminated
+        let mut lock = self.tasks.lock().await;
+        if lock.is_empty() {
+            info!(
+                target: "damd::flooder::DamFlooder::start",
+                "Denial-of-service Analysis Multitool flooder already terminated!"
+            );
+            return
+        }
+
+        // Terminate the tasks
+        for task in lock.iter() {
+            task.stop().await;
+        }
+
+        // Clean the set
+        *lock = HashSet::new();
+        info!(target: "damd::flooder::DamFlooder::stop", "Denial-of-service Analysis Multitool flooder terminated successfully!");
+    }
+}
+
+/// Background flooder function for `ProtocolFoo`.
+async fn flood_foo(comms_timeout: u64, peer: ChannelPtr, subscriber: JsonSubscriber) -> Result<()> {
+    debug!(target: "damd::flooder::flood_foo", "START");
+    // Communication setup
+    let Ok(response_sub) = peer.subscribe_msg::<FooResponse>().await else {
+        let notification =
+            format!("Failure during `FooResponse` communication setup with peer: {peer:?}");
+        error!(target: "damd::flooder::flood_foo", "{notification}");
+        subscriber.notify(vec![JsonValue::String(notification)].into()).await;
+        return Ok(())
+    };
+
+    // Flood the peer
+    let mut message_index = 0;
+    loop {
+        // Node creates a `FooRequest` and sends it
+        let message = format!("Flood message {message_index}");
+        let notification = format!("Sending foo request to {peer:?}: {message}");
+        info!(target: "damd::flooder::flood_foo", "{notification}");
+        subscriber.notify(vec![JsonValue::String(notification)].into()).await;
+        if let Err(e) = peer.send(&FooRequest { message }).await {
+            let notification = format!("Failure during `FooRequest` send to peer {peer:?}: {e}");
+            error!(target: "damd::flooder::flood_foo", "{notification}");
+            subscriber.notify(vec![JsonValue::String(notification)].into()).await;
+            return Ok(())
+        };
+
+        // Node waits for response
+        let Ok(response) = response_sub.receive_with_timeout(comms_timeout).await else {
+            let notification =
+                format!("Timeout while waiting for `FooResponse` from peer: {peer:?}");
+            error!(target: "damd::flooder::flood_foo", "{notification}");
+            subscriber.notify(vec![JsonValue::String(notification)].into()).await;
+            return Ok(())
+        };
+
+        // Notify subscriber
+        let notification = format!("Retrieved foo response from {peer:?}: {}", response.code);
+        info!(target: "damd::flooder::flood_foo", "{notification}");
+        subscriber.notify(vec![JsonValue::String(notification)].into()).await;
+        message_index += 1;
+    }
+}
+
+/// Background flooder function for `ProtocolBar`.
+async fn flood_bar(p2p: P2pPtr, subscriber: JsonSubscriber) -> Result<()> {
+    debug!(target: "damd::flooder::flood_bar", "START");
+
+    // Flood the network, if we are connected to peers
+    let mut message_index = 0;
+    while p2p.is_connected() {
+        // Node creates a `Bar` message and broadcasts it
+        let message = format!("Flood message {message_index}");
+        let notification = format!("Broadcasting bar message: {message}");
+        info!(target: "damd::flooder::flood_bar", "{notification}");
+        subscriber.notify(vec![JsonValue::String(notification)].into()).await;
+        p2p.broadcast(&Bar { message }).await;
+        message_index += 1;
+    }
+
+    debug!(target: "damd::flooder::flood_bar", "STOP");
+    subscriber
+        .notify(vec![JsonValue::String(String::from("We are not connected to any peers"))].into())
+        .await;
+    Ok(())
+}

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

@@ -0,0 +1,75 @@
+/* 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_std::prelude::StreamExt;
+use log::info;
+use smol::Executor;
+use structopt_toml::{serde::Deserialize, structopt::StructOpt, StructOptToml};
+
+use darkfi::{
+    async_daemonize, cli_desc, net::settings::SettingsOpt, rpc::settings::RpcSettingsOpt, Result,
+};
+
+use damd::Damd;
+
+const CONFIG_FILE: &str = "damd_config.toml";
+const CONFIG_FILE_CONTENTS: &str = include_str!("../damd_config.toml");
+
+#[derive(Clone, Debug, Deserialize, StructOpt, StructOptToml)]
+#[serde(default)]
+#[structopt(name = "damd", about = cli_desc!())]
+struct Args {
+    #[structopt(short, long)]
+    /// Configuration file to use
+    config: Option<String>,
+
+    #[structopt(flatten)]
+    /// JSON-RPC settings
+    rpc: RpcSettingsOpt,
+
+    #[structopt(flatten)]
+    /// P2P network settings
+    net: SettingsOpt,
+
+    #[structopt(short, long)]
+    /// Set log file to ouput into
+    log: Option<String>,
+
+    #[structopt(short, parse(from_occurrences))]
+    /// Increase verbosity (-vvv supported)
+    verbose: u8,
+}
+
+async_daemonize!(realmain);
+async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
+    info!(target: "damd", "Starting Denial-of-service Analysis Multitool daemon...");
+    let daemon = Damd::init(&args.net.into(), &ex).await?;
+    daemon.start(&ex, &args.rpc.into()).await?;
+
+    // Signal handling for graceful termination.
+    let (signals_handler, signals_task) = SignalHandler::new(ex)?;
+    signals_handler.wait_termination(signals_task).await?;
+    info!(target: "damd", "Caught termination signal, cleaning up and exiting");
+
+    daemon.stop().await?;
+
+    info!(target: "damd", "Shut down successfully");
+    Ok(())
+}

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

@@ -0,0 +1,207 @@
+/* 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::HashSet;
+
+use async_trait::async_trait;
+use log::debug;
+use smol::lock::MutexGuard;
+
+use darkfi::{
+    net::P2pPtr,
+    rpc::{
+        jsonrpc::{
+            ErrorCode::{InvalidParams, MethodNotFound},
+            JsonError, JsonRequest, JsonResponse, JsonResult,
+        },
+        p2p_method::HandlerP2p,
+        server::RequestHandler,
+        util::JsonValue,
+    },
+    system::StoppableTaskPtr,
+};
+
+use crate::DamNode;
+
+#[async_trait]
+impl RequestHandler<()> for DamNode {
+    async fn handle_request(&self, req: JsonRequest) -> JsonResult {
+        debug!(target: "damd::rpc", "--> {}", req.stringify().unwrap());
+
+        match req.method.as_str() {
+            // =====================
+            // Miscellaneous methods
+            // =====================
+            "ping" => self.pong(req.id, req.params).await,
+            "dnet.switch" => self.dnet_switch(req.id, req.params).await,
+            "dnet.subscribe_events" => self.dnet_subscribe_events(req.id, req.params).await,
+            "p2p.get_info" => self.p2p_get_info(req.id, req.params).await,
+
+            // =================
+            // Protocols methods
+            // =================
+            "protocols.subscribe_foo" => self.protocols_subscribe_foo(req.id, req.params).await,
+            "protocols.subscribe_attack_foo" => {
+                self.protocols_subscribe_attack_foo(req.id, req.params).await
+            }
+            "protocols.subscribe_bar" => self.protocols_subscribe_bar(req.id, req.params).await,
+            "protocols.subscribe_attack_bar" => {
+                self.protocols_subscribe_attack_bar(req.id, req.params).await
+            }
+
+            // =============
+            // Flood control
+            // =============
+            "flood.switch" => self.flood_switch(req.id, req.params).await,
+
+            // ==============
+            // Invalid method
+            // ==============
+            _ => JsonError::new(MethodNotFound, None, req.id).into(),
+        }
+    }
+
+    async fn connections_mut(&self) -> MutexGuard<'life0, HashSet<StoppableTaskPtr>> {
+        self.rpc_connections.lock().await
+    }
+}
+
+impl DamNode {
+    // RPCAPI:
+    // Activate or deactivate dnet in the P2P stack.
+    // By sending `true`, dnet will be activated, and by sending `false` dnet
+    // will be deactivated. Returns `true` on success.
+    //
+    // --> {"jsonrpc": "2.0", "method": "dnet_switch", "params": [true], "id": 42}
+    // <-- {"jsonrpc": "2.0", "result": true, "id": 42}
+    async fn dnet_switch(&self, id: u16, params: JsonValue) -> JsonResult {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if params.len() != 1 || !params[0].is_bool() {
+            return JsonError::new(InvalidParams, None, id).into()
+        }
+
+        let switch = params[0].get::<bool>().unwrap();
+
+        if *switch {
+            self.p2p_handler.p2p.dnet_enable();
+        } else {
+            self.p2p_handler.p2p.dnet_disable();
+        }
+
+        JsonResponse::new(JsonValue::Boolean(true), id).into()
+    }
+
+    // RPCAPI:
+    // Initializes a subscription to p2p dnet events.
+    // Once a subscription is established, `damd` will send JSON-RPC notifications of
+    // new network events to the subscriber.
+    //
+    // --> {"jsonrpc": "2.0", "method": "protocols.subscribe_foo", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "method": "protocols.subscribe_foo", "params": [`event`]}
+    pub async fn dnet_subscribe_events(&self, id: u16, params: JsonValue) -> JsonResult {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if !params.is_empty() {
+            return JsonError::new(InvalidParams, None, id).into()
+        }
+
+        self.subscribers.get("dnet").unwrap().clone().into()
+    }
+
+    // RPCAPI:
+    // Initializes a subscription to new incoming `Foo` messages.
+    // Once a subscription is established, `damd` will send JSON-RPC notifications of
+    // new incoming `Foo` messages to the subscriber.
+    //
+    // --> {"jsonrpc": "2.0", "method": "protocols.subscribe_foo", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "method": "protocols.subscribe_foo", "params": [`message`]}
+    pub async fn protocols_subscribe_foo(&self, id: u16, params: JsonValue) -> JsonResult {
+        self.get_subscriber(id, params, "foo").await
+    }
+
+    // RPCAPI:
+    // Initializes a subscription to new outgoing attack `Foo` messages.
+    // Once a subscription is established, `damd` will send JSON-RPC notifications of
+    // new outgoing attack `Foo` messages to the subscriber.
+    //
+    // --> {"jsonrpc": "2.0", "method": "protocols.subscribe_attack_foo", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "method": "protocols.subscribe_attack_foo", "params": [`message`]}
+    pub async fn protocols_subscribe_attack_foo(&self, id: u16, params: JsonValue) -> JsonResult {
+        self.get_subscriber(id, params, "attack_foo").await
+    }
+
+    // RPCAPI:
+    // Initializes a subscription to new incoming `Bar` messages.
+    // Once a subscription is established, `damd` will send JSON-RPC notifications of
+    // new incoming `Bar` messages to the subscriber.
+    //
+    // --> {"jsonrpc": "2.0", "method": "protocols.subscribe_bar", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "method": "protocols.subscribe_bar", "params": [`message`]}
+    pub async fn protocols_subscribe_bar(&self, id: u16, params: JsonValue) -> JsonResult {
+        self.get_subscriber(id, params, "bar").await
+    }
+
+    // RPCAPI:
+    // Initializes a subscription to new outgoing attack `Bar` messages.
+    // Once a subscription is established, `damd` will send JSON-RPC notifications of
+    // new outgoing attack `Bar` messages to the subscriber.
+    //
+    // --> {"jsonrpc": "2.0", "method": "protocols.subscribe_attack_bar", "params": [], "id": 1}
+    // <-- {"jsonrpc": "2.0", "method": "protocols.subscribe_attack_bar", "params": [`message`]}
+    pub async fn protocols_subscribe_attack_bar(&self, id: u16, params: JsonValue) -> JsonResult {
+        self.get_subscriber(id, params, "attack_bar").await
+    }
+
+    async fn get_subscriber(&self, id: u16, params: JsonValue, sub: &str) -> JsonResult {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if !params.is_empty() {
+            return JsonError::new(InvalidParams, None, id).into()
+        }
+
+        self.subscribers.get(sub).unwrap().clone().into()
+    }
+
+    // RPCAPI:
+    // Activate or deactivate damd flooder.
+    // By sending `true`, flooder will be activated, and by sending `false` flooder
+    // will be deactivated. Returns `true` on success.
+    //
+    // --> {"jsonrpc": "2.0", "method": "flood", "params": [true], "id": 42}
+    // <-- {"jsonrpc": "2.0", "result": true, "id": 42}
+    async fn flood_switch(&self, id: u16, params: JsonValue) -> JsonResult {
+        let params = params.get::<Vec<JsonValue>>().unwrap();
+        if params.len() != 1 || !params[0].is_bool() {
+            return JsonError::new(InvalidParams, None, id).into()
+        }
+
+        let switch = params[0].get::<bool>().unwrap();
+
+        if *switch {
+            self.flooder.start(&self.subscribers).await;
+        } else {
+            self.flooder.stop().await;
+        }
+
+        JsonResponse::new(JsonValue::Boolean(true), id).into()
+    }
+}
+
+impl HandlerP2p for DamNode {
+    fn p2p(&self) -> P2pPtr {
+        self.p2p_handler.p2p.clone()
+    }
+}