handle prefixes instead of addresses

Author: matthieugouel

Cargo.lock

@@ -12,7 +12,9 @@ clap-verbosity-flag = "3.0.0"
 csv = "1.3.1"
 env_logger = "0.11.5"
 hyperloglog = "1.0.2"
+ipnet = "2.10.1"
 log = "0.4.22"
 pcap = "2.2.0"
+rand = "0.8.5"
 rdkafka = { version = "0.37.0" }
 tokio = { version = "1.42.0", features = ["full"] }

Cargo.toml

@@ -1,10 +1,9 @@
 use anyhow::Result;
 use caracat::models::Probe;
 use caracat::rate_limiter::RateLimitingMethod;
-// use log::{info, warn};
-// use rdkafka::message::BorrowedMessage;
-// use rdkafka::message::{Headers, Message};
-use std::net::IpAddr;
+use ipnet::IpNet;
+use rand::seq::SliceRandom;
+use rand::thread_rng;
 use std::net::Ipv4Addr;
 use std::net::Ipv6Addr;
 use std::time::Duration;
@@ -66,68 +65,99 @@ fn create_config() -> Config {
 }
 
 struct Payload {
-    dst_addr: IpAddr,
+    prefix: IpNet,
     min_ttl: u8,
     max_ttl: u8,
+    n_flows: u64,
 }
 
 fn decode_payload(payload: &str) -> Result<Payload> {
     let parts: Vec<&str> = payload.split(',').collect();
     Ok(Payload {
-        dst_addr: parts[0].parse()?,
+        prefix: parts[0].parse()?,
         min_ttl: parts[1].parse()?,
         max_ttl: parts[2].parse()?,
+        n_flows: parts[3].parse()?,
     })
 }
 
-// pub async fn handle(m: &BorrowedMessage<'_>) -> Result<()> {
+fn generate_probes(payload: &Payload) -> Result<Vec<Probe>> {
+    // TODO: We should pass an iterator instead of a vector.
+    let mut probes = vec![];
+
+    // First start by dividing the prefix into /24s or /64s, if necessary.
+    let subnets = match payload.prefix {
+        IpNet::V4(_) => {
+            let prefix_len = payload.prefix.prefix_len();
+            let target_len = if prefix_len > 24 { prefix_len } else { 24 };
+            payload.prefix.subnets(target_len)
+        }
+        IpNet::V6(_) => {
+            let prefix_len = payload.prefix.prefix_len();
+            let target_len = if prefix_len > 64 { prefix_len } else { 64 };
+            payload.prefix.subnets(target_len)
+        }
+    }?;
+
+    // Iterate over the subnets and generate the probes.
+    for subnet in subnets {
+        // Right now the probe generation is simplistic, we just iterate over the hosts.
+        // If we need more flows than hosts, we will we explicitely fail.
+        // TODO: implement mapper-like generator such as the ones in diamond-miner.
+        // https://github.com/dioptra-io/diamond-miner/blob/main/diamond_miner/mappers.py
+        let mut prefix_hosts = subnet.hosts();
+        if payload.n_flows > prefix_hosts.count().try_into()? {
+            return Err(anyhow::anyhow!("Not enough hosts in the prefix"));
+        }
+
+        for _ in 0..payload.n_flows {
+            let dst_addr = prefix_hosts.next().unwrap();
+
+            // Randomize the probes order within a flow.
+            // In YARRP we randomize the probes over the entire probing space.
+            // This is of course a very big simplication, but it's not that silly.
+            // The rational is to avoid results errors due to path changes.
+            // So, for now, probes belonging to the same traceroute flow will be sent close in time.
+            // TODO: is this shuffle fast?
+            let mut ttls: Vec<u8> = (payload.min_ttl..payload.max_ttl).collect();
+            ttls.shuffle(&mut thread_rng());
+
+            for i in ttls {
+                probes.push(Probe {
+                    dst_addr,
+                    src_port: 24000,
+                    dst_port: 33434,
+                    ttl: i,
+                    protocol: caracat::models::L4::ICMPv6,
+                });
+            }
+        }
+    }
+
+    Ok(probes)
+}
+
 pub async fn handle(
     brokers: &str,
     _in_group_id: &str,
     _in_topics: &[&str],
     out_topic: &str,
 ) -> Result<()> {
-    // let payload = match m.payload_view::<str>() {
-    //     None => "",
-    //     Some(Ok(s)) => s,
-    //     Some(Err(e)) => {
-    //         warn!("Error while deserializing message payload: {:?}", e);
-    //         ""
-    //     }
-    // };
-
-    // info!(
-    //     "key: '{:?}', payload: '{}', topic: {}, partition: {}, offset: {}, timestamp: {:?}",
-    //     m.key(),
-    //     payload,
-    //     m.topic(),
-    //     m.partition(),
-    //     m.offset(),
-    //     m.timestamp()
-    // );
-
-    // if let Some(headers) = m.headers() {
-    //     for header in headers.iter() {
-    //         info!("  Header {:#?}: {:?}", header.key, header.value);
-    //     }
-    // }
-
-    let payload = "2606:4700:4700::1111,1,32";
+    let payload = "2606:4700:4700::1111/128,1,32,1";
 
     // Probing
     let config = create_config();
     let payload = decode_payload(payload)?;
 
-    let mut probes_to_send = vec![];
-    for i in payload.min_ttl..=payload.max_ttl {
-        probes_to_send.push(Probe {
-            dst_addr: payload.dst_addr.clone(),
-            src_port: 24000,
-            dst_port: 33434,
-            ttl: i,
-            protocol: caracat::models::L4::ICMPv6,
-        });
-    }
+    let probes_to_send = generate_probes(&payload);
+    let probes_to_send = match probes_to_send {
+        Ok(probes) => probes,
+        Err(e) => {
+            eprintln!("Error: {}", e);
+            return Ok(());
+        }
+    };
+
     let result = task::spawn_blocking(move || probe(config, probes_to_send.into_iter())).await?;
 
     let (_, _, results) = result?;

src/handler.rs

@@ -2,7 +2,7 @@
 
 Docker Compose setup to facilitate the tests of Osiris.
 
-The testbed consists in a Redpanda instance.
+The testbed consists in a Redpanda and ClickHouse instance. Required ClickHouse [tables](config/clickhouse/docker-entrypoint-initdb.d/init.sql) are created on startup. The `osiris.results_broker` is using the ClickHouse [Kafka engine](https://clickhouse.com/docs/en/engines/table-engines/integrations/kafka) to fetch the results from Redpanda. The `osiris.results` table is used to store the results.
 
 
 ## Usage
@@ -12,3 +12,15 @@ The testbed consists in a Redpanda instance.
 ```sh
 docker compose up -d --force-recreate --renew-anon-volumes
 ```
+
+* Run Osiris (from the root of the repository)
+
+```sh
+cargo run
+```
+
+* Stop the testbed
+
+```sh
+docker compose down
+```