[CDN] Implement hooks for DDoS protection

Cargo.toml

@@ -57,7 +57,7 @@ espresso-systems-common = { git = "https://github.com/espressosystems/espresso-s
 ethereum-types = { version = "0.14", default-features = false, features = [
   "serialize",
 ] }
-derive_more = { version = "1.0", features = [ "from" ]  }
+derive_more = { version = "1.0", features = ["from"] }
 futures = { version = "0.3", default-features = false }
 # TODO generic-array should not be a direct dependency
 # https://github.com/EspressoSystems/HotShot/issues/1850
@@ -121,10 +121,10 @@ anyhow = "1"
 
 
 # Push CDN imports
-cdn-client = { git = "https://github.com/EspressoSystems/Push-CDN", tag = "0.4.7" }
-cdn-broker = { git = "https://github.com/EspressoSystems/Push-CDN", tag = "0.4.7" }
-cdn-marshal = { git = "https://github.com/EspressoSystems/Push-CDN", tag = "0.4.7" }
-cdn-proto = { git = "https://github.com/EspressoSystems/Push-CDN", tag = "0.4.7" }
+cdn-client = { git = "https://github.com/EspressoSystems/Push-CDN", tag = "0.4.9" }
+cdn-broker = { git = "https://github.com/EspressoSystems/Push-CDN", tag = "0.4.9" }
+cdn-marshal = { git = "https://github.com/EspressoSystems/Push-CDN", tag = "0.4.9" }
+cdn-proto = { git = "https://github.com/EspressoSystems/Push-CDN", tag = "0.4.9" }
 
 ### Profiles
 ###

crates/examples/combined/all.rs

@@ -14,10 +14,10 @@ use async_compatibility_layer::{
     art::async_spawn,
     logging::{setup_backtrace, setup_logging},
 };
-use cdn_broker::Broker;
+use cdn_broker::{reexports::def::hook::NoMessageHook, Broker};
 use cdn_marshal::Marshal;
 use hotshot::{
-    traits::implementations::{KeyPair, TestingDef, WrappedSignatureKey},
+    traits::implementations::{HotShotMessageHook, KeyPair, TestingDef, WrappedSignatureKey},
     types::SignatureKey,
 };
 use hotshot_example_types::{node_types::TestVersions, state_types::TestTypes};
@@ -72,28 +72,34 @@ async fn main() {
         let private_address = format!("127.0.0.1:{private_port}");
         let public_address = format!("127.0.0.1:{public_port}");
 
-        let config: cdn_broker::Config<TestingDef<<TestTypes as NodeType>::SignatureKey>> =
-            cdn_broker::Config {
-                discovery_endpoint: discovery_endpoint.clone(),
-                public_advertise_endpoint: public_address.clone(),
-                public_bind_endpoint: public_address,
-                private_advertise_endpoint: private_address.clone(),
-                private_bind_endpoint: private_address,
-
-                keypair: KeyPair {
-                    public_key: WrappedSignatureKey(broker_public_key),
-                    private_key: broker_private_key.clone(),
-                },
+        // Create the message hooks
+        let broker_message_hook = NoMessageHook;
+        let user_message_hook = HotShotMessageHook::default();
+
+        let config: cdn_broker::Config<TestingDef<TestTypes>> = cdn_broker::Config {
+            discovery_endpoint: discovery_endpoint.clone(),
+            public_advertise_endpoint: public_address.clone(),
+            public_bind_endpoint: public_address,
+            private_advertise_endpoint: private_address.clone(),
+            private_bind_endpoint: private_address,
+
+            keypair: KeyPair {
+                public_key: WrappedSignatureKey(broker_public_key),
+                private_key: broker_private_key.clone(),
+            },
+
+            metrics_bind_endpoint: None,
+            ca_cert_path: None,
+            ca_key_path: None,
+            global_memory_pool_size: Some(1024 * 1024 * 1024),
 
-                metrics_bind_endpoint: None,
-                ca_cert_path: None,
-                ca_key_path: None,
-                global_memory_pool_size: Some(1024 * 1024 * 1024),
-            };
+            user_message_hook,
+            broker_message_hook,
+        };
 
         // Create and spawn the broker
         async_spawn(async move {
-            let broker: Broker<TestingDef<<TestTypes as NodeType>::SignatureKey>> =
+            let broker: Broker<TestingDef<TestTypes>> =
                 Broker::new(config).await.expect("broker failed to start");
 
             // Error if we stopped unexpectedly
@@ -121,10 +127,9 @@ async fn main() {
 
     // Spawn the marshal
     async_spawn(async move {
-        let marshal: Marshal<TestingDef<<TestTypes as NodeType>::SignatureKey>> =
-            Marshal::new(marshal_config)
-                .await
-                .expect("failed to spawn marshal");
+        let marshal: Marshal<TestingDef<TestTypes>> = Marshal::new(marshal_config)
+            .await
+            .expect("failed to spawn marshal");
 
         // Error if we stopped unexpectedly
         if let Err(err) = marshal.start().await {

crates/examples/push-cdn/all.rs

@@ -11,10 +11,13 @@ pub mod types;
 use std::path::Path;
 
 use async_compatibility_layer::art::async_spawn;
-use cdn_broker::{reexports::crypto::signature::KeyPair, Broker};
+use cdn_broker::{
+    reexports::{crypto::signature::KeyPair, def::hook::NoMessageHook},
+    Broker,
+};
 use cdn_marshal::Marshal;
 use hotshot::{
-    traits::implementations::{TestingDef, WrappedSignatureKey},
+    traits::implementations::{HotShotMessageHook, TestingDef, WrappedSignatureKey},
     types::SignatureKey,
 };
 use hotshot_example_types::{node_types::TestVersions, state_types::TestTypes};
@@ -78,28 +81,34 @@ async fn main() {
         let private_address = format!("127.0.0.1:{private_port}");
         let public_address = format!("127.0.0.1:{public_port}");
 
-        let config: cdn_broker::Config<TestingDef<<TestTypes as NodeType>::SignatureKey>> =
-            cdn_broker::Config {
-                discovery_endpoint: discovery_endpoint.clone(),
-                public_advertise_endpoint: public_address.clone(),
-                public_bind_endpoint: public_address,
-                private_advertise_endpoint: private_address.clone(),
-                private_bind_endpoint: private_address,
-
-                keypair: KeyPair {
-                    public_key: WrappedSignatureKey(broker_public_key),
-                    private_key: broker_private_key.clone(),
-                },
+        // Create the message hooks
+        let broker_message_hook = NoMessageHook;
+        let user_message_hook = HotShotMessageHook::default();
+
+        let config: cdn_broker::Config<TestingDef<TestTypes>> = cdn_broker::Config {
+            discovery_endpoint: discovery_endpoint.clone(),
+            public_advertise_endpoint: public_address.clone(),
+            public_bind_endpoint: public_address,
+            private_advertise_endpoint: private_address.clone(),
+            private_bind_endpoint: private_address,
+
+            keypair: KeyPair {
+                public_key: WrappedSignatureKey(broker_public_key),
+                private_key: broker_private_key.clone(),
+            },
+
+            metrics_bind_endpoint: None,
+            ca_cert_path: None,
+            ca_key_path: None,
+            global_memory_pool_size: Some(1024 * 1024 * 1024),
 
-                metrics_bind_endpoint: None,
-                ca_cert_path: None,
-                ca_key_path: None,
-                global_memory_pool_size: Some(1024 * 1024 * 1024),
-            };
+            user_message_hook,
+            broker_message_hook,
+        };
 
         // Create and spawn the broker
         async_spawn(async move {
-            let broker: Broker<TestingDef<<TestTypes as NodeType>::SignatureKey>> =
+            let broker: Broker<TestingDef<TestTypes>> =
                 Broker::new(config).await.expect("broker failed to start");
 
             // Error if we stopped unexpectedly
@@ -125,10 +134,9 @@ async fn main() {
 
     // Spawn the marshal
     async_spawn(async move {
-        let marshal: Marshal<TestingDef<<TestTypes as NodeType>::SignatureKey>> =
-            Marshal::new(marshal_config)
-                .await
-                .expect("failed to spawn marshal");
+        let marshal: Marshal<TestingDef<TestTypes>> = Marshal::new(marshal_config)
+            .await
+            .expect("failed to spawn marshal");
 
         // Error if we stopped unexpectedly
         if let Err(err) = marshal.start().await {

crates/examples/push-cdn/broker.rs

@@ -7,9 +7,11 @@
 //! The following is the main `Broker` binary, which just instantiates and runs
 //! a `Broker` object.
 use anyhow::Result;
-use cdn_broker::{Broker, Config};
+use cdn_broker::{reexports::def::hook::NoMessageHook, Broker, Config};
 use clap::Parser;
-use hotshot::traits::implementations::{KeyPair, ProductionDef, WrappedSignatureKey};
+use hotshot::traits::implementations::{
+    HotShotMessageHook, KeyPair, ProductionDef, WrappedSignatureKey,
+};
 use hotshot_example_types::node_types::TestTypes;
 use hotshot_types::traits::{node_implementation::NodeType, signature_key::SignatureKey};
 use sha2::Digest;
@@ -91,8 +93,12 @@ async fn main() -> Result<()> {
     let (public_key, private_key) =
         <TestTypes as NodeType>::SignatureKey::generated_from_seed_indexed(key_hash.into(), 1337);
 
+    // Create the message hooks
+    let broker_message_hook = NoMessageHook;
+    let user_message_hook = HotShotMessageHook::default();
+
     // Create config
-    let broker_config: Config<ProductionDef<<TestTypes as NodeType>::SignatureKey>> = Config {
+    let broker_config: Config<ProductionDef<TestTypes>> = Config {
         ca_cert_path: args.ca_cert_path,
         ca_key_path: args.ca_key_path,
 
@@ -108,6 +114,9 @@ async fn main() -> Result<()> {
         private_bind_endpoint: args.private_bind_endpoint,
         private_advertise_endpoint: args.private_advertise_endpoint,
         global_memory_pool_size: Some(args.global_memory_pool_size),
+
+        user_message_hook,
+        broker_message_hook,
     };
 
     // Create new `Broker`

crates/examples/push-cdn/marshal.rs

@@ -12,7 +12,6 @@ use cdn_marshal::{Config, Marshal};
 use clap::Parser;
 use hotshot::traits::implementations::ProductionDef;
 use hotshot_example_types::node_types::TestTypes;
-use hotshot_types::traits::node_implementation::NodeType;
 use tracing_subscriber::EnvFilter;
 
 // TODO: forall, add logging where we need it
@@ -81,8 +80,7 @@ async fn main() -> Result<()> {
     };
 
     // Create new `Marshal` from the config
-    let marshal =
-        Marshal::<ProductionDef<<TestTypes as NodeType>::SignatureKey>>::new(config).await?;
+    let marshal = Marshal::<ProductionDef<TestTypes>>::new(config).await?;
 
     // Start the main loop, consuming it
     marshal.start().await?;

crates/hotshot/Cargo.toml

@@ -61,7 +61,9 @@ blake3.workspace = true
 sha2 = { workspace = true }
 url = { workspace = true }
 num_enum = "0.7"
+twox-hash = { version = "1", default-features = false }
 parking_lot = "0.12"
+simple_moving_average = "1"
 
 [target.'cfg(all(async_executor_impl = "tokio"))'.dependencies]
 tokio = { workspace = true }

crates/hotshot/src/traits.rs

@@ -24,8 +24,12 @@ pub mod implementations {
         },
         memory_network::{MasterMap, MemoryNetwork},
         push_cdn_network::{
-            CdnMetricsValue, KeyPair, ProductionDef, PushCdnNetwork, TestingDef, Topic as CdnTopic,
-            WrappedSignatureKey,
+            definition::{
+                message_hook::HotShotMessageHook, signature_key::WrappedSignatureKey,
+                ProductionDef, TestingDef, Topic as CdnTopic,
+            },
+            metrics::CdnMetricsValue,
+            KeyPair, PushCdnNetwork,
         },
     };
 }