diff --git a/Cargo.lock b/Cargo.lock index 069eefb..61dd16c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2632,7 +2632,7 @@ dependencies = [ [[package]] name = "probe" -version = "0.3.0" +version = "0.4.0" dependencies = [ "anyhow", "clap", @@ -4706,7 +4706,7 @@ dependencies = [ [[package]] name = "website" -version = "1.2.2" +version = "1.2.3" dependencies = [ "dotenvy", "env_logger", diff --git a/frontend/src/lib/api/probe.ts b/frontend/src/lib/api/probe.ts index 4f22c3a..b4f74b7 100644 --- a/frontend/src/lib/api/probe.ts +++ b/frontend/src/lib/api/probe.ts @@ -113,10 +113,14 @@ export type ProbeStatus = { export function startProbeSSE( id: string, + token: string, onResult: (result: ProbeResult) => void, onStatus: (status: Partial) => void, ) { - const eventSource = new EventSource(`/api/v1/probe/${id}`); + const params = new URLSearchParams(); + if (token) params.set("token", token); + const query = params.size > 0 ? `?${params.toString()}` : ""; + const eventSource = new EventSource(`/api/v1/probe/${id}${query}`); eventSource.addEventListener("started", (event) => { const data = JSON.parse(event.data); diff --git a/frontend/src/lib/components/ResultPanel.svelte b/frontend/src/lib/components/ResultPanel.svelte index 62ec845..e664b2b 100644 --- a/frontend/src/lib/components/ResultPanel.svelte +++ b/frontend/src/lib/components/ResultPanel.svelte @@ -27,11 +27,19 @@ type ResultVerdict = ResolvedProbeVerdict | "blocked"; let { result, probeVerdict = null, + token = "", }: { result: CheckResult; probeVerdict?: ResolvedProbeVerdict | null; + token?: string; } = $props(); +function checkHref(target: string): string { + const params = new URLSearchParams({ target }); + if (token) params.set("token", token); + return `/check?${params.toString()}`; +} + const valueClass = "text-right text-sm font-medium text-neutral-200"; const alertValueClass = `${valueClass} text-red-500`; const successValueClass = `${valueClass} text-green-500`; @@ -114,7 +122,7 @@ const providerCidrs = (provider: Provider) => {#each result.reverseLookup as ptr} {ptr} @@ -135,7 +143,7 @@ const providerCidrs = (provider: Provider) => {#if result.geo.asn} {result.geo.asn} diff --git a/frontend/src/lib/components/SearchForm.svelte b/frontend/src/lib/components/SearchForm.svelte index 8bf5ea1..2c3dc18 100644 --- a/frontend/src/lib/components/SearchForm.svelte +++ b/frontend/src/lib/components/SearchForm.svelte @@ -1,11 +1,16 @@
+ {#if token} + + {/if} +import { page } from "$app/state"; import SearchForm from "$lib/components/SearchForm.svelte"; import StatCard from "$lib/components/StatCard.svelte"; import { getStatusContext } from "$lib/context/status"; const statusQuery = getStatusContext(); const status = $derived(statusQuery.data); +const token = $derived(page.url.searchParams.get("token")?.trim() ?? ""); @@ -21,7 +23,7 @@ const status = $derived(statusQuery.data);

- +
({ queryKey: ["check", target], @@ -48,7 +49,7 @@ function createInitialProbeData(id: string): ProbeQueryData { } const probeQuery = createQuery(() => ({ - queryKey: ["probes", queryId], + queryKey: ["probes", queryId, token], queryFn: () => createInitialProbeData(queryId ?? ""), enabled: shouldProbe, staleTime: Infinity, @@ -59,43 +60,50 @@ $effect(() => { if (!queryId || !shouldProbe) return; queryClient.setQueryData( - ["probes", queryId], + ["probes", queryId, token], createInitialProbeData(queryId), ); const cleanup = startProbeSSE( queryId, + token, (result) => { - queryClient.setQueryData(["probes", queryId], (old) => { - const current = old ?? createInitialProbeData(queryId); - const probes = current.probes.some( - (probe) => probe.probe_id === result.probe_id, - ) - ? current.probes.map((probe) => - probe.probe_id === result.probe_id ? result : probe, - ) - : [...current.probes, result]; + queryClient.setQueryData( + ["probes", queryId, token], + (old) => { + const current = old ?? createInitialProbeData(queryId); + const probes = current.probes.some( + (probe) => probe.probe_id === result.probe_id, + ) + ? current.probes.map((probe) => + probe.probe_id === result.probe_id ? result : probe, + ) + : [...current.probes, result]; - return { - ...current, - probes, - status: { - ...current.status, - status: "progress", - response_count: probes.length, - }, - }; - }); + return { + ...current, + probes, + status: { + ...current.status, + status: "progress", + response_count: probes.length, + }, + }; + }, + ); }, (statusUpdate) => { - queryClient.setQueryData(["probes", queryId], (old) => { - const current = old ?? createInitialProbeData(queryId); + queryClient.setQueryData( + ["probes", queryId, token], + (old) => { + const current = old ?? createInitialProbeData(queryId); - return { - ...current, - status: { ...current.status, ...statusUpdate }, - }; - }); + return { + ...current, + status: { ...current.status, ...statusUpdate }, + }; + }, + ); }, ); @@ -115,7 +123,7 @@ const liveVerdict = $derived( ); - + {#if target.length === 0}
@@ -130,7 +138,7 @@ const liveVerdict = $derived(
{:else if checkQuery.data} - + {#if shouldProbe && probeQuery.data && probeQuery.data.status.online_probes > 0} Result<()> { client .subscribe("probe/tasks/v1/+", QoS::AtLeastOnce) .await?; + client + .subscribe( + format!("probe/tasks/v1/{}/+", args.probe_id), + QoS::AtLeastOnce, + ) + .await?; info!( "probe {} connected over WebSocket to {}", @@ -176,6 +182,12 @@ async fn main() -> Result<()> { client .subscribe("probe/tasks/v1/+", QoS::AtLeastOnce) .await?; + client + .subscribe( + format!("probe/tasks/v1/{}/+", args.probe_id), + QoS::AtLeastOnce, + ) + .await?; tokio::time::sleep(Duration::from_secs(2)).await; } } @@ -349,10 +361,7 @@ async fn handle_task( task: ProbeTask<'_>, received_at: Instant, ) -> Result<()> { - let job_id = topic - .strip_prefix("probe/tasks/v1/") - .filter(|id| !id.is_empty()) - .unwrap_or(&task.id); + let job_id = probe_task_job_id(topic).unwrap_or(&task.id); let timeout = Duration::from_millis(task.timeout_ms); let Some(remaining) = timeout.checked_sub(received_at.elapsed()) else { warn!( @@ -427,10 +436,27 @@ async fn handle_task( .context("publish probe result") } +fn probe_task_job_id(topic: &str) -> Option<&str> { + let parts = topic.split('/').collect::>(); + match parts.as_slice() { + ["probe", "tasks", "v1", job_id] if !job_id.is_empty() => Some(job_id), + ["probe", "tasks", "v1", _recipient, job_id] if !job_id.is_empty() => Some(job_id), + _ => None, + } +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn extracts_job_id_from_legacy_global_and_individual_topics() { + assert_eq!(probe_task_job_id("probe/tasks/v1/job-1"), Some("job-1")); + assert_eq!(probe_task_job_id("probe/tasks/v1/42/job-2"), Some("job-2")); + assert_eq!(probe_task_job_id("probe/tasks/v1"), None); + assert_eq!(probe_task_job_id("probe/tasks/v1/42/job-2/extra"), None); + } + #[test] fn decodes_separate_dpi_targets() { let config: ProbeConfig = serde_json::from_value(serde_json::json!({ diff --git a/website/Cargo.toml b/website/Cargo.toml index 9e60d7f..036769e 100644 --- a/website/Cargo.toml +++ b/website/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "website" -version = "1.2.2" +version = "1.2.3" edition = "2024" [dependencies] diff --git a/website/migrations/20260825000000_hidden_reporters.sql b/website/migrations/20260825000000_hidden_reporters.sql new file mode 100644 index 0000000..25df0c3 --- /dev/null +++ b/website/migrations/20260825000000_hidden_reporters.sql @@ -0,0 +1,2 @@ +ALTER TABLE reporters + ADD COLUMN IF NOT EXISTS hidden BOOLEAN NOT NULL DEFAULT FALSE; diff --git a/website/src/api/probe.rs b/website/src/api/probe.rs index c0e6c9b..f108b09 100644 --- a/website/src/api/probe.rs +++ b/website/src/api/probe.rs @@ -26,9 +26,10 @@ pub struct ProbeReporterInfo { pub asn: Option, } -#[get("/probe/")] +#[get("/probe/?")] pub async fn probe_query( id: &str, + token: Option<&str>, addr: &ClientRealAddr, pool: &State, mqtt: &State, @@ -67,12 +68,17 @@ pub async fn probe_query( return Err(Status::Forbidden); } + let (target_probe, eligible_probes) = load_probe_targets(pool, token).await?; + let expected_probes = mqtt.online_probe_ids(&eligible_probes).await; + let online_probes = expected_probes.len(); + let expected_probes = expected_probes.into_iter().collect::>(); + 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, ip) + mqtt.publish_probe_task(id, domain, ip, target_probe.as_deref()) .await .map_err(|error| { warn!("api: failed to publish probe task for {id}: {error}"); @@ -80,7 +86,6 @@ pub async fn probe_query( })?; let timeout = mqtt.task_timeout(); - let online_probes = mqtt.online_probe_count().await; let pool = pool.inner().clone(); let query_id = id; let id = id.to_string(); @@ -105,6 +110,9 @@ pub async fn probe_query( result = results.recv() => { match result { Ok(result) => { + if !expected_probes.contains(&result.probe_id) { + continue; + } responded_probes.insert(result.probe_id.clone()); let target_traceroute = result.target_traceroute.clone(); let reporter_info = match fetch_probe_reporter_info(&result.probe_id, &pool).await { @@ -145,6 +153,44 @@ pub async fn probe_query( }) } +async fn load_probe_targets( + pool: &PgPool, + token: Option<&str>, +) -> Result<(Option, Vec), Status> { + if token.is_some_and(str::is_empty) { + return Err(Status::BadRequest); + } + if let Some(token) = token { + let probe_id = + sqlx::query_scalar::<_, i32>("SELECT id FROM reporters WHERE token = $1 LIMIT 1") + .bind(token) + .fetch_optional(pool) + .await + .map_err(|error| { + warn!("api: failed to resolve targeted probe: {error}"); + Status::InternalServerError + })? + .ok_or(Status::NotFound)? + .to_string(); + + return Ok((Some(probe_id.clone()), vec![probe_id])); + } + + let probe_ids = + sqlx::query_scalar::<_, i32>("SELECT id FROM reporters WHERE hidden = FALSE ORDER BY id") + .fetch_all(pool) + .await + .map_err(|error| { + warn!("api: failed to load global probe recipients: {error}"); + Status::InternalServerError + })? + .into_iter() + .map(|id| id.to_string()) + .collect(); + + Ok((None, probe_ids)) +} + pub fn build_probe_response( raw: ProbeResultEvent, config: &ProbeConfig, diff --git a/website/src/mqtt.rs b/website/src/mqtt.rs index 421bb47..e381166 100644 --- a/website/src/mqtt.rs +++ b/website/src/mqtt.rs @@ -197,13 +197,17 @@ impl MqttPublisher { Duration::from_millis(self.task_timeout_ms) } - pub async fn online_probe_count(&self) -> usize { - self.probe_statuses - .read() - .await - .values() - .filter(|status| status.online) - .count() + pub async fn online_probe_ids(&self, probe_ids: &[String]) -> Vec { + let statuses = self.probe_statuses.read().await; + probe_ids + .iter() + .filter(|probe_id| { + statuses + .get(probe_id.as_str()) + .is_some_and(|status| status.online) + }) + .cloned() + .collect() } pub async fn probe_statuses(&self) -> HashMap { @@ -253,6 +257,7 @@ impl MqttPublisher { query_id: Uuid, domain: Option<&str>, ip: IpAddr, + probe_id: Option<&str>, ) -> Result<(), PublishError> { let client = self.client.as_ref().ok_or(PublishError::NotConfigured)?; let query_id = query_id.to_string(); @@ -265,7 +270,10 @@ impl MqttPublisher { timeout_ms: self.task_timeout_ms, }; let payload = serde_json::to_vec(&task).map_err(PublishError::Serialize)?; - let topic = format!("probe/tasks/v1/{query_id}"); + let topic = match probe_id { + Some(probe_id) => format!("probe/tasks/v1/{probe_id}/{query_id}"), + None => format!("probe/tasks/v1/{query_id}"), + }; client .publish(topic, QoS::AtLeastOnce, false, payload) diff --git a/website/src/mqtt_auth.rs b/website/src/mqtt_auth.rs index 7e8e975..263def8 100644 --- a/website/src/mqtt_auth.rs +++ b/website/src/mqtt_auth.rs @@ -95,7 +95,10 @@ pub async fn auth( } #[post("/acl", data = "")] -pub async fn acl(request: Form>) -> Json { +pub async fn acl( + request: Form>, + pool: &rocket::State, +) -> Json { let request = request.into_inner(); let _ = request.protocol; @@ -108,17 +111,40 @@ pub async fn acl(request: Form>) -> Json { } match request.access { - 1 if can_probe_subscribe(request.topic) => Json(MqttAuthResponse::allow()), + 1 if can_probe_subscribe(request.clientid, request.topic, pool).await => { + Json(MqttAuthResponse::allow()) + } 2 if can_probe_publish(request.clientid, request.topic) => Json(MqttAuthResponse::allow()), _ => Json(MqttAuthResponse::deny()), } } -fn can_probe_subscribe(topic: &str) -> bool { - matches!( - topic, - "probe/config/v1" | "probe/tasks/v1/+" | "probe/tasks/v1/#" - ) +async fn can_probe_subscribe(client_id: &str, topic: &str, pool: &PgPool) -> bool { + if topic == "probe/config/v1" || is_own_task_subscription(client_id, topic) { + return true; + } + if !is_global_task_subscription(topic) { + return false; + } + + let Ok(reporter_id) = client_id.parse::() else { + return false; + }; + sqlx::query_scalar::<_, bool>("SELECT NOT hidden FROM reporters WHERE id = $1") + .bind(reporter_id) + .fetch_optional(pool) + .await + .ok() + .flatten() + .unwrap_or(false) +} + +fn is_own_task_subscription(client_id: &str, topic: &str) -> bool { + topic == format!("probe/tasks/v1/{client_id}/+") +} + +fn is_global_task_subscription(topic: &str) -> bool { + topic == "probe/tasks/v1/+" } fn can_probe_publish(client_id: &str, topic: &str) -> bool { @@ -141,3 +167,24 @@ fn can_probe_publish(client_id: &str, topic: &str) -> bool { if probe_id == client_id ) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn individual_task_subscriptions_are_node_scoped() { + assert!(is_global_task_subscription("probe/tasks/v1/+")); + assert!(!is_global_task_subscription("probe/tasks/v1/#")); + assert!(is_own_task_subscription("42", "probe/tasks/v1/42/+")); + assert!(!is_own_task_subscription("42", "probe/tasks/v1/7/+")); + assert!(!is_own_task_subscription("42", "probe/tasks/v1/+/+")); + assert!(!is_own_task_subscription("42", "probe/tasks/v1/#")); + } + + #[test] + fn results_can_only_be_published_as_the_authenticated_node() { + assert!(can_probe_publish("42", "probe/results/v1/job/42")); + assert!(!can_probe_publish("42", "probe/results/v1/job/7")); + } +}