feat: ip probe traceroute

website/src/api/probe.rs

@@ -4,6 +4,7 @@ use log::warn;
 use querying::target::Target;
 use reports::probe::{
     Host, HostProbeResult, HostType, ProbeConfig, ProbeEvidence, ProbeResultEvent,
+    TcpTracerouteOutcome, TcpTracerouteResult,
 };
 use rocket::State;
 use rocket::http::Status;
@@ -15,6 +16,7 @@ use rocket_client_addr::ClientRealAddr;
 use sqlx::postgres::PgPool;
 use sqlx::types::Uuid;
 use std::collections::{HashMap, HashSet};
+use std::net::IpAddr;
 use std::sync::Arc;
 
 #[derive(sqlx::FromRow)]
@@ -37,23 +39,33 @@ pub async fn probe_query(
     }
 
     let id = Uuid::try_parse(id).map_err(|_| Status::BadRequest)?;
-    let query: Option<String> = sqlx::query_scalar("SELECT query FROM queries WHERE id = $1")
-        .bind(id)
-        .fetch_optional(&**pool)
-        .await
-        .map_err(|_| Status::InternalServerError)?;
+    let query: Option<(String, Vec<String>)> =
+        sqlx::query_as("SELECT query, resolved_ips FROM queries WHERE id = $1")
+            .bind(id)
+            .fetch_optional(&**pool)
+            .await
+            .map_err(|_| Status::InternalServerError)?;
 
-    let query = query.ok_or(Status::NotFound)?;
-    let Target::Domain(domain) = Target::from(query.trim()) else {
-        return Err(Status::BadRequest);
+    let (query, resolved_ips) = query.ok_or(Status::NotFound)?;
+    let target = Target::from(query.trim());
+    let domain = match &target {
+        Target::Domain(domain) => Some(domain.as_str()),
+        Target::Ipv4(_) | Target::Ipv6(_) => None,
+        Target::Ipv4Subnet(_) | Target::Ipv6Subnet(_) | Target::Asn(_) => {
+            return Err(Status::BadRequest);
+        }
     };
+    let ip = resolved_ips
+        .first()
+        .and_then(|ip| ip.parse::<IpAddr>().ok())
+        .ok_or(Status::BadRequest)?;
 
     let mut results = mqtt.subscribe_probe_results(id).await.map_err(|error| {
         warn!("api: failed to subscribe to probe results for {id}: {error}");
         publish_error_status(error)
     })?;
 
-    mqtt.publish_probe_task(id, &domain)
+    mqtt.publish_probe_task(id, domain, ip)
         .await
         .map_err(|error| {
             warn!("api: failed to publish probe task for {id}: {error}");
@@ -73,7 +85,7 @@ pub async fn probe_query(
 
         yield Event::data(json!({
             "id": id,
-            "target": domain,
+            "target": query,
             "online_probes": online_probes,
         }).to_string()).event("started");
 
@@ -88,6 +100,8 @@ pub async fn probe_query(
                     match result {
                         Ok(result) => {
                             responded_probes.insert(result.probe_id.clone());
+                            let target_traceroute = result.target_traceroute.clone();
+                            let control_traceroute = result.control_traceroute.clone();
                             let reporter_info = match fetch_probe_reporter_info(&result.probe_id, &pool).await {
                                 Ok(info) => info,
                                 Err(error) => {
@@ -99,7 +113,13 @@ pub async fn probe_query(
                                 }
                             };
                             let response = build_probe_response(result, &probe_config, reporter_info);
-                            if let Err(error) = insert_probe_report(query_id, &response, &pool).await {
+                            if let Err(error) = insert_probe_report(
+                                query_id,
+                                &response,
+                                target_traceroute.as_ref(),
+                                control_traceroute.as_ref(),
+                                &pool,
+                            ).await {
                                 warn!("api: failed to save probe report for query {id}: {error}");
                             }
                             yield Event::data(response.to_string()).event("result");
@@ -127,7 +147,21 @@ pub fn build_probe_response(
     reporter_info: Option<ProbeReporterInfo>,
 ) -> Value {
     let hosts: HashMap<&String, &Host> = config.hosts.iter().map(|h| (&h.id, h)).collect();
-    let verdict = build_probe_verdict(&raw.host_results, config);
+    let verdict = build_probe_verdict(
+        &raw.host_results,
+        config,
+        raw.target_traceroute.as_ref(),
+        raw.control_traceroute.as_ref(),
+    );
+    let target_hop =
+        raw.target_traceroute
+            .as_ref()
+            .and_then(|traceroute| match &traceroute.result {
+                TcpTracerouteOutcome::Rst { hop }
+                | TcpTracerouteOutcome::Connected { hop }
+                | TcpTracerouteOutcome::IcmpTimeExceeded { hop } => Some(*hop),
+                TcpTracerouteOutcome::Timeout => None,
+            });
     let region = reporter_info.as_ref().and_then(|info| info.region.as_ref());
     let provider = reporter_info
         .as_ref()
@@ -155,12 +189,15 @@ pub fn build_probe_response(
         "asn": asn,
         "verdict": verdict,
         "host_results": host_results,
+        "target_hop": target_hop,
     })
 }
 
 async fn insert_probe_report(
     query_id: Uuid,
     response: &Value,
+    target_traceroute: Option<&TcpTracerouteResult>,
+    control_traceroute: Option<&TcpTracerouteResult>,
     pool: &PgPool,
 ) -> Result<(), sqlx::Error> {
     let probe_id = response
@@ -171,6 +208,8 @@ async fn insert_probe_report(
         .get("verdict")
         .and_then(Value::as_str)
         .unwrap_or("uncertain");
+    let (target_hop_count, target_trace_result) = traceroute_columns(target_traceroute);
+    let (control_hop_count, control_trace_result) = traceroute_columns(control_traceroute);
 
     let Some(probe_id) = probe_id else {
         warn!("api: ignoring probe report with non-numeric probe_id");
@@ -179,25 +218,57 @@ async fn insert_probe_report(
 
     sqlx::query(
         r#"
-        INSERT INTO probe_reports (query_id, probe_id, verdict, result)
-        VALUES ($1, $2, $3, $4)
+        INSERT INTO probe_reports (
+            query_id,
+            probe_id,
+            verdict,
+            result,
+            target_hop_count,
+            target_trace_result,
+            control_hop_count,
+            control_trace_result
+        )
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
         ON CONFLICT (query_id, probe_id)
         DO UPDATE SET
             date = NOW(),
             verdict = EXCLUDED.verdict,
-            result = EXCLUDED.result
+            result = EXCLUDED.result,
+            target_hop_count = EXCLUDED.target_hop_count,
+            target_trace_result = EXCLUDED.target_trace_result,
+            control_hop_count = EXCLUDED.control_hop_count,
+            control_trace_result = EXCLUDED.control_trace_result
         "#,
     )
     .bind(query_id)
     .bind(probe_id)
     .bind(verdict)
     .bind(response)
+    .bind(target_hop_count)
+    .bind(target_trace_result)
+    .bind(control_hop_count)
+    .bind(control_trace_result)
     .execute(pool)
     .await?;
 
     Ok(())
 }
 
+fn traceroute_columns(
+    traceroute: Option<&TcpTracerouteResult>,
+) -> (Option<i16>, Option<&'static str>) {
+    let Some(traceroute) = traceroute else {
+        return (None, None);
+    };
+    let (hop, result) = match &traceroute.result {
+        TcpTracerouteOutcome::Rst { hop } => (Some(*hop), "Rst"),
+        TcpTracerouteOutcome::Connected { hop } => (Some(*hop), "Connected"),
+        TcpTracerouteOutcome::IcmpTimeExceeded { hop } => (Some(*hop), "IcmpTimeExceeded"),
+        TcpTracerouteOutcome::Timeout => (None, "Timeout"),
+    };
+    (hop.map(i16::from), Some(result))
+}
+
 async fn fetch_probe_reporter_info(
     probe_id: &str,
     pool: &PgPool,
@@ -210,7 +281,31 @@ async fn fetch_probe_reporter_info(
     .await
 }
 
-fn build_probe_verdict(results: &[HostProbeResult], config: &ProbeConfig) -> &'static str {
+fn build_probe_verdict(
+    results: &[HostProbeResult],
+    config: &ProbeConfig,
+    target_traceroute: Option<&TcpTracerouteResult>,
+    control_traceroute: Option<&TcpTracerouteResult>,
+) -> &'static str {
+    if let (
+        Some(TcpTracerouteResult {
+            result: TcpTracerouteOutcome::IcmpTimeExceeded { hop: target_hop },
+            ..
+        }),
+        Some(TcpTracerouteResult {
+            result: TcpTracerouteOutcome::IcmpTimeExceeded { hop: control_hop },
+            ..
+        }),
+    ) = (target_traceroute, control_traceroute)
+        && target_hop < control_hop
+    {
+        return "tspu_block";
+    }
+
+    if results.is_empty() {
+        return "ok";
+    }
+
     let matched = results
         .iter()
         .filter_map(|result| {
@@ -306,3 +401,54 @@ fn done_event(id: &str, response_count: usize, online_probes: usize) -> Event {
 fn is_strict_majority(total: usize, count: usize) -> bool {
     count > total / 2
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn icmp_trace(hop: u8) -> TcpTracerouteResult {
+        TcpTracerouteResult {
+            target: "192.0.2.1".parse().unwrap(),
+            result: TcpTracerouteOutcome::IcmpTimeExceeded { hop },
+        }
+    }
+
+    #[test]
+    fn tspu_block_requires_an_earlier_target_icmp_hop() {
+        let target = icmp_trace(3);
+        let control = icmp_trace(5);
+        assert_eq!(
+            build_probe_verdict(&[], &empty_config(), Some(&target), Some(&control)),
+            "tspu_block"
+        );
+
+        let target = icmp_trace(5);
+        assert_eq!(
+            build_probe_verdict(&[], &empty_config(), Some(&target), Some(&control)),
+            "uncertain"
+        );
+    }
+
+    #[test]
+    fn tspu_block_requires_two_icmp_outcomes() {
+        let target = TcpTracerouteResult {
+            target: "192.0.2.1".parse().unwrap(),
+            result: TcpTracerouteOutcome::Connected { hop: 2 },
+        };
+        let control = icmp_trace(5);
+        assert_eq!(
+            build_probe_verdict(&[], &empty_config(), Some(&target), Some(&control)),
+            "uncertain"
+        );
+    }
+
+    fn empty_config() -> ProbeConfig {
+        ProbeConfig {
+            version: String::new(),
+            task_timeout_ms: 0,
+            published_at: String::new(),
+            hosts: vec![],
+            control_hosts: vec![],
+        }
+    }
+}

website/src/mqtt.rs

@@ -1,6 +1,6 @@
 use log::{info, warn};
 use reports::probe::HostType;
-use reports::probe::{Host, ProbeConfig, ProbeResultEvent, ProbeStatus, ProbeTask};
+use reports::probe::{Host, ProbeConfig, ProbeResult, ProbeResultEvent, ProbeStatus, ProbeTask};
 use rocket::serde::json::serde_json;
 use rumqttc::{AsyncClient, Event as MqttEvent, Incoming, MqttOptions, QoS};
 use serde::Deserialize;
@@ -8,6 +8,7 @@ use sqlx::types::Uuid;
 use sqlx::types::chrono::Utc;
 use std::collections::{HashMap, HashSet};
 use std::fmt;
+use std::net::IpAddr;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -60,6 +61,8 @@ pub type ProbeResultReceiver = rocket::tokio::sync::broadcast::Receiver<ProbeRes
 struct ProbeHostsFile {
     timeout_sec: u32,
     min_data: u32,
+    #[serde(default)]
+    control_hosts: Vec<String>,
     hosts: Vec<ProbeHostEntry>,
 }
 
@@ -85,6 +88,7 @@ impl MqttPublisher {
                 task_timeout_ms,
                 published_at: Utc::now().to_rfc3339(),
                 hosts: Vec::new(),
+                control_hosts: Vec::new(),
             }
         }));
         let admin_token = match std::env::var("MQTT_ADMIN_TOKEN") {
@@ -216,14 +220,16 @@ impl MqttPublisher {
     pub async fn publish_probe_task(
         &self,
         query_id: Uuid,
-        domain: &str,
+        domain: Option<&str>,
+        ip: IpAddr,
     ) -> Result<(), PublishError> {
         let client = self.client.as_ref().ok_or(PublishError::NotConfigured)?;
         let query_id = query_id.to_string();
         let task = ProbeTask {
             id: query_id.clone(),
             query_id: query_id.clone(),
-            target: domain,
+            domain,
+            ip,
             created_at: Utc::now().to_rfc3339(),
             timeout_ms: self.task_timeout_ms,
         };
@@ -250,7 +256,7 @@ async fn publish_probe_config(
 }
 
 fn load_probe_config(task_timeout_ms: u64) -> Result<ProbeConfig, PublishError> {
-    let hosts = if let Some(path) = std::env::var_os("PROBE_CONFIG_PATH") {
+    let config = if let Some(path) = std::env::var_os("PROBE_CONFIG_PATH") {
         let contents = std::fs::read_to_string(path).map_err(PublishError::Config)?;
         parse_probe_hosts(&contents)?
     } else {
@@ -260,14 +266,20 @@ fn load_probe_config(task_timeout_ms: u64) -> Result<ProbeConfig, PublishError>
         version: env!("CARGO_PKG_VERSION").to_string(),
         task_timeout_ms,
         published_at: Utc::now().to_rfc3339(),
-        hosts,
+        hosts: config.hosts,
+        control_hosts: config.control_hosts,
     })
 }
 
-fn parse_probe_hosts(contents: &str) -> Result<Vec<Host>, PublishError> {
+struct ParsedProbeConfig {
+    hosts: Vec<Host>,
+    control_hosts: Vec<String>,
+}
+
+fn parse_probe_hosts(contents: &str) -> Result<ParsedProbeConfig, PublishError> {
     let config: ProbeHostsFile = toml::from_str(&contents).map_err(PublishError::ConfigParse)?;
 
-    Ok(config
+    let hosts = config
         .hosts
         .into_iter()
         .map(|host| Host {
@@ -278,7 +290,12 @@ fn parse_probe_hosts(contents: &str) -> Result<Vec<Host>, PublishError> {
             timeout_sec: host.timeout_sec.unwrap_or(config.timeout_sec),
             min_data: host.min_data.unwrap_or(config.min_data),
         })
-        .collect())
+        .collect();
+
+    Ok(ParsedProbeConfig {
+        hosts,
+        control_hosts: config.control_hosts,
+    })
 }
 
 async fn dispatch_probe_status(
@@ -320,7 +337,7 @@ async fn dispatch_probe_result(
         return;
     };
 
-    let result = match serde_json::from_slice(payload) {
+    let result: ProbeResult = match serde_json::from_slice(payload) {
         Ok(result) => result,
         Err(error) => {
             warn!("ignoring invalid probe result JSON on {topic}: {error}");
@@ -333,7 +350,9 @@ async fn dispatch_probe_result(
         let _ = sender.send(ProbeResultEvent {
             job_id: job_id.to_string(),
             probe_id: probe_id.to_string(),
-            host_results: result,
+            host_results: result.responses.unwrap_or_default(),
+            target_traceroute: result.target_traceroute,
+            control_traceroute: result.control_traceroute,
         });
     }
 }

website/src/mqtt_auth.rs

@@ -56,7 +56,7 @@ pub async fn auth(
     pool: &rocket::State<PgPool>,
 ) -> Json<MqttAuthResponse> {
     let request = request.into_inner();
-    let _ = (request.clientid, request.protocol);
+    let _ = request.protocol;
 
     if request.username == "admin"
         && std::env::var("MQTT_ADMIN_TOKEN")
@@ -66,20 +66,19 @@ pub async fn auth(
         return Json(MqttAuthResponse::allow_superuser());
     }
 
-    if request.username != "probe" || request.password.is_empty() {
+    if request.username != "probe" || request.clientid.is_empty() || request.password.is_empty() {
         return Json(MqttAuthResponse::deny());
     }
 
-    let token_exists =
+    let reporter_id =
         sqlx::query_scalar::<_, i32>("SELECT id FROM reporters WHERE token = $1 LIMIT 1")
             .bind(request.password)
             .fetch_optional(&**pool)
             .await
             .ok()
-            .flatten()
-            .is_some();
+            .flatten();
 
-    if token_exists {
+    if reporter_id.is_some_and(|id| id.to_string() == request.clientid) {
         Json(MqttAuthResponse::allow())
     } else {
         Json(MqttAuthResponse::deny())