|
+
+ Блокировка на ТСПУ
+ {probe.verdict === "tspu_block" ? ` обнаружена после ${probe.target_hop} прыжка` : " не обнаружена"}
+
+
{#each probe.host_results as host}
- Сервер {host.host_id}
+ Сервер{host.host_id}
({host.host === "Blacklist" ? "в заблокированных" : "в доступных"} диапазонах)
@@ -176,7 +187,7 @@ const verdictStyles = {
{:else if host.probe_evidence.type === 'ClientHello'}
Блокировка после ClientHello
{:else if host.probe_evidence.type === 'DataTimeout'}
- Таймаут получения данных, получено {host.probe_evidence.bytes} байт
+ Таймаут получения данных, получено{host.probe_evidence.bytes} байт
{:else if host.probe_evidence.type === 'ConnectionError'}
Ошибка подключения
{/if}
diff --git a/frontend/src/routes/check/+page.svelte b/frontend/src/routes/check/+page.svelte
index 3a837ed..94c0b73 100644
--- a/frontend/src/routes/check/+page.svelte
+++ b/frontend/src/routes/check/+page.svelte
@@ -31,9 +31,7 @@ const checkQuery = createQuery(() => ({
}));
const queryId = $derived(checkQuery.data?.id);
-const shouldProbe = $derived(
- !!queryId && checkQuery.data?.targetType === "Домен",
-);
+const shouldProbe = $derived(!!queryId);
function createInitialProbeData(id: string): ProbeQueryData {
return {
diff --git a/frontend/vite.config.ts b/frontend/vite.config.ts
index 6b0cbbd..c160615 100644
--- a/frontend/vite.config.ts
+++ b/frontend/vite.config.ts
@@ -6,7 +6,7 @@ export default defineConfig({
server: {
proxy: {
"/api": {
- target: "http://localhost:8000",
+ target: "http://localhost:8080",
},
},
},
diff --git a/probe/Cargo.toml b/probe/Cargo.toml
index 5dc8fec..b1e8e3b 100644
--- a/probe/Cargo.toml
+++ b/probe/Cargo.toml
@@ -1,6 +1,6 @@
[package]
name = "probe"
-version = "0.1.0"
+version = "0.1.1"
edition = "2024"
license-file = "../LICENSE"
description = "Dynamic network probe daemon for Cheburcheck"
@@ -25,8 +25,11 @@ assets = [
anyhow = "1.0"
clap = { version = "4.6.0", features = ["derive", "env"] }
env_logger = "0.11.10"
+etherparse = "0.21"
futures = "0.3"
log = { workspace = true }
+polling = "3.11"
+rustix = { version = "1.1", features = ["net"] }
rumqttc = { version = "0.25", features = ["use-rustls", "websocket"] }
serde = { workspace = true }
serde_json = "1.0"
@@ -34,3 +37,5 @@ tokio = { workspace = true }
reports = { path = "../reports" }
rustls = "0.23"
tokio-rustls = "0.26"
+rand = "0.8"
+socket2 = { version = "0.6", features = ["all"] }
diff --git a/probe/Dockerfile b/probe/Dockerfile
index 53486ce..0fabcb6 100644
--- a/probe/Dockerfile
+++ b/probe/Dockerfile
@@ -15,13 +15,18 @@ RUN --mount=type=cache,id=probe-target,target=/build/target \
FROM docker.io/debian:bookworm-slim
-RUN groupadd --system app && \
+RUN apt-get update && \
+ apt-get install --yes --no-install-recommends libcap2-bin && \
+ rm -rf /var/lib/apt/lists/* && \
+ groupadd --system app && \
useradd --system --gid app --home-dir /app --shell /usr/sbin/nologin app
WORKDIR /app
COPY --from=build /build/probe-bin ./probe
+RUN setcap cap_net_raw=ep ./probe
+
USER app
ENTRYPOINT ["./probe"]
diff --git a/probe/README.md b/probe/README.md
index b83665a..745416d 100644
--- a/probe/README.md
+++ b/probe/README.md
@@ -104,6 +104,7 @@ cargo run --package probe --bin cheburprobe
```shell
docker run --rm \
+ --cap-add NET_RAW \
-e PROBE_ID=1 \
-e PROBE_TOKEN=ваш-токен \
-e MQTT_HOST=wss://cheburcheck.ru/mqtt \
@@ -121,9 +122,10 @@ docker run --rm \
| `--probe-id`, `PROBE_ID` | ID сканера. | обязательно |
| `--probe-token`, `PROBE_TOKEN` | Секретный токен сканера. | обязательно |
| `--max-concurrent-tasks`, `MAX_CONCURRENT_TASKS` | Максимальное количество одновременных заданий. | `8` |
+| `--traceroute-max-hops`, `TRACEROUTE_MAX_HOPS` | Максимальный TTL для TCP traceroute. | `5` |
| `RUST_LOG` | Уровень логирования. | `info` |
-`MAX_CONCURRENT_TASKS` должен быть больше нуля.
+`MAX_CONCURRENT_TASKS` и `TRACEROUTE_MAX_HOPS` должны быть больше нуля. Для получения ICMP-ответов traceroute процессу требуется capability `CAP_NET_RAW`; systemd unit и Docker-образ настраивают её автоматически.
## Как работает проверка
@@ -131,8 +133,8 @@ docker run --rm \
1. публикует retained-статус `online` в MQTT;
2. подписывается на конфигурацию динамического сканирования;
-3. получает задания на проверку доменов;
-4. параллельно проверяет домен на настроенных тестовых хостах;
+3. получает задания на проверку доменов и IP-адресов;
+4. параллельно запускает SNI-проверки (для доменов), TCP traceroute до цели и контрольный TCP traceroute;
5. отправляет результат обратно в Cheburcheck.
Для каждого тестового хоста сканер открывает TCP-соединение, начинает TLS-handshake с проверяемым доменом в SNI, затем отправляет простой HTTP GET-запрос.
diff --git a/probe/debian/cheburprobe.service b/probe/debian/cheburprobe.service
index 75738b1..c7efbc6 100644
--- a/probe/debian/cheburprobe.service
+++ b/probe/debian/cheburprobe.service
@@ -10,6 +10,8 @@ ExecStart=/usr/bin/cheburprobe
Restart=always
RestartSec=5s
DynamicUser=yes
+AmbientCapabilities=CAP_NET_RAW
+CapabilityBoundingSet=CAP_NET_RAW
LimitNOFILE=16384
[Install]
diff --git a/probe/src/main.rs b/probe/src/main.rs
index 1eab789..4afee48 100644
--- a/probe/src/main.rs
+++ b/probe/src/main.rs
@@ -1,24 +1,29 @@
+mod sni;
+mod traceroute;
+
use anyhow::{Context, Result, bail};
use clap::Parser;
-use futures::future::join_all;
use log::{error, info, warn};
-use reports::probe::{Host, HostProbeResult, ProbeConfig, ProbeEvidence, ProbeStatus, ProbeTask};
+use rand::seq::SliceRandom;
+use reports::probe::{ProbeConfig, ProbeResult, ProbeStatus, ProbeTask};
use rumqttc::{
AsyncClient, Event, Incoming, LastWill, MqttOptions, NetworkOptions, QoS, Transport,
};
-use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
-use rustls::pki_types::{CertificateDer, ServerName, UnixTime};
-use rustls::{ClientConfig, DigitallySignedStruct, Error as TlsError, SignatureScheme};
+use std::collections::HashSet;
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use std::sync::Arc;
use std::time::{Duration, Instant};
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::net::TcpStream;
use tokio::sync::RwLock;
-use tokio::time;
-use tokio_rustls::TlsConnector;
const CONFIG_TOPIC: &str = "probe/config/v1";
+#[derive(Clone)]
+struct LoadedProbeConfig {
+ config: ProbeConfig,
+ control_hosts_v4: Vec,
+ control_hosts_v6: Vec,
+}
+
#[derive(Parser, Debug, Clone)]
#[command(author, version, about = "Dynamic probing daemon")]
struct Args {
@@ -39,6 +44,9 @@ struct Args {
#[arg(long, env = "MAX_CONCURRENT_TASKS", default_value_t = 8)]
max_concurrent_tasks: usize,
+
+ #[arg(long, env = "TRACEROUTE_MAX_HOPS", default_value_t = 5)]
+ traceroute_max_hops: u8,
}
#[tokio::main]
@@ -48,6 +56,9 @@ async fn main() -> Result<()> {
if args.max_concurrent_tasks == 0 {
bail!("max_concurrent_tasks must be greater than zero");
}
+ if args.traceroute_max_hops == 0 {
+ bail!("traceroute_max_hops must be greater than zero");
+ }
let status_topic = format!("probe/status/v1/{}", args.probe_id);
let offline_status = serde_json::to_vec(&ProbeStatus {
@@ -155,9 +166,35 @@ fn mqtt_transport(mqtt_host: &str) -> Result {
}
}
-async fn update_config(config: &Arc>>, payload: &[u8]) -> Result<()> {
- let value = serde_json::from_slice(payload).context("decode probe config")?;
- *config.write().await = Some(value);
+async fn update_config(
+ config: &Arc>>,
+ payload: &[u8],
+) -> Result<()> {
+ let value: ProbeConfig = serde_json::from_slice(payload).context("decode probe config")?;
+ let mut control_hosts_v4 = HashSet::new();
+ let mut control_hosts_v6 = HashSet::new();
+ for domain in &value.control_hosts {
+ match tokio::net::lookup_host((domain.as_str(), 443)).await {
+ Ok(addresses) => {
+ for address in addresses {
+ match address.ip() {
+ IpAddr::V4(address) => {
+ control_hosts_v4.insert(address);
+ }
+ IpAddr::V6(address) => {
+ control_hosts_v6.insert(address);
+ }
+ }
+ }
+ }
+ Err(error) => warn!("failed to resolve control host {domain}: {error}"),
+ }
+ }
+ *config.write().await = Some(LoadedProbeConfig {
+ config: value,
+ control_hosts_v4: control_hosts_v4.into_iter().collect(),
+ control_hosts_v6: control_hosts_v6.into_iter().collect(),
+ });
info!("updated retained probe config");
Ok(())
}
@@ -201,7 +238,7 @@ async fn publish_status(
async fn handle_task(
client: &AsyncClient,
args: &Args,
- config: &Arc>>,
+ config: &Arc>>,
topic: &str,
task: ProbeTask<'_>,
received_at: Instant,
@@ -219,180 +256,52 @@ async fn handle_task(
return Ok(());
};
- let config = config.read().await.clone();
- let Some(config) = config else {
- bail!("no config");
- };
let result_topic = format!("probe/results/v1/{job_id}/{}", args.probe_id);
- let probing = join_all(config.hosts.into_iter().map(|host| {
- let target = task.target.to_string();
- async move {
- let probe_evidence = probe_host(&host, &target).await;
- HostProbeResult {
- probe_evidence,
- host_id: host.id,
- }
+ let config = config.read().await.clone();
+ let control_target = config.as_ref().and_then(|config| {
+ let mut rng = rand::thread_rng();
+ match task.ip {
+ IpAddr::V4(_) => config
+ .control_hosts_v4
+ .choose(&mut rng)
+ .copied()
+ .map(IpAddr::V4),
+ IpAddr::V6(_) => config
+ .control_hosts_v6
+ .choose(&mut rng)
+ .copied()
+ .map(IpAddr::V6),
}
- }));
- let result = match time::timeout(remaining, probing).await {
- Ok(result) => result,
- Err(_) => {
- warn!(
- "dropping expired task {job_id}: timeout {}ms",
- task.timeout_ms
- );
- return Ok(());
+ });
+ let sni_check = sni::check_sni(
+ config.as_ref().map(|config| &config.config),
+ task.domain,
+ remaining,
+ job_id,
+ task.timeout_ms,
+ );
+ let target_traceroute = traceroute::tcp_traceroute(task.ip, args.traceroute_max_hops);
+ let control_traceroute = async {
+ match control_target {
+ Some(target) => traceroute::tcp_traceroute(target, args.traceroute_max_hops).await,
+ None => None,
}
};
+ let (responses, target_traceroute, control_traceroute) =
+ tokio::join!(sni_check, target_traceroute, control_traceroute);
+ let responses = responses?;
client
.publish(
result_topic,
QoS::AtLeastOnce,
false,
- serde_json::to_vec(&result)?,
+ serde_json::to_vec(&ProbeResult {
+ responses,
+ target_traceroute,
+ control_traceroute,
+ })?,
)
.await
.context("publish probe result")
}
-
-async fn probe_host(host: &Host, target: &str) -> ProbeEvidence {
- let timeout = Duration::from_secs(host.timeout_sec as u64);
- let tcp = match time::timeout(timeout, TcpStream::connect((host.host.as_str(), 443))).await {
- Ok(Ok(tcp)) => tcp,
- Ok(Err(_)) | Err(_) => return ProbeEvidence::ConnectionError,
- };
-
- let tls_config = ClientConfig::builder()
- .dangerous()
- .with_custom_certificate_verifier(Arc::new(NoCertificateVerification))
- .with_no_client_auth();
- let connector = TlsConnector::from(Arc::new(tls_config));
-
- let server_name = match ServerName::try_from(target.to_string()) {
- Ok(server_name) => server_name,
- Err(_) => return ProbeEvidence::ClientHello,
- };
-
- let mut tls = match time::timeout(timeout, connector.connect(server_name, tcp)).await {
- Ok(Ok(tls)) => tls,
- Ok(Err(_)) | Err(_) => return ProbeEvidence::ClientHello,
- };
-
- let request = format!(
- "GET /{} HTTP/1.1\r\nHost: {}\r\nUser-Agent: cheburcheck-probe/{}\r\nRange: bytes=0-{}\r\nConnection: close\r\n\r\n",
- host.file_path.trim_start_matches('/'),
- target,
- env!("CARGO_PKG_VERSION"),
- host.min_data.saturating_sub(1)
- );
-
- if !matches!(
- time::timeout(timeout, tls.write_all(request.as_bytes())).await,
- Ok(Ok(()))
- ) {
- return ProbeEvidence::ClientHello;
- }
-
- let mut received = 0u32;
- let mut headers_done = false;
- let mut pending = Vec::new();
- let mut buffer = [0u8; 8192];
- loop {
- match time::timeout(timeout, tls.read(&mut buffer)).await {
- Ok(Ok(0)) | Err(_) => {
- return if received >= host.min_data {
- ProbeEvidence::Good
- } else {
- ProbeEvidence::DataTimeout { bytes: received }
- };
- }
- Ok(Ok(bytes)) => {
- add_response_body_bytes(
- &buffer[..bytes],
- &mut pending,
- &mut headers_done,
- &mut received,
- );
- if received >= host.min_data {
- return ProbeEvidence::Good;
- }
- }
- Ok(Err(_)) => {
- return if received >= host.min_data {
- ProbeEvidence::Good
- } else {
- ProbeEvidence::DataTimeout { bytes: received }
- };
- }
- }
- }
-}
-
-fn add_response_body_bytes(
- chunk: &[u8],
- pending: &mut Vec,
- headers_done: &mut bool,
- received: &mut u32,
-) {
- if *headers_done {
- *received = received.saturating_add(chunk.len() as u32);
- return;
- }
-
- pending.extend_from_slice(chunk);
- if let Some(body_start) = pending.windows(4).position(|window| window == b"\r\n\r\n") {
- *headers_done = true;
- let body_bytes = pending.len().saturating_sub(body_start + 4);
- *received = received.saturating_add(body_bytes as u32);
- pending.clear();
- }
-}
-
-#[derive(Debug)]
-struct NoCertificateVerification;
-
-impl ServerCertVerifier for NoCertificateVerification {
- fn verify_server_cert(
- &self,
- _end_entity: &CertificateDer<'_>,
- _intermediates: &[CertificateDer<'_>],
- _server_name: &ServerName<'_>,
- _ocsp_response: &[u8],
- _now: UnixTime,
- ) -> Result {
- Ok(ServerCertVerified::assertion())
- }
-
- fn verify_tls12_signature(
- &self,
- _message: &[u8],
- _cert: &CertificateDer<'_>,
- _dss: &DigitallySignedStruct,
- ) -> Result {
- Ok(HandshakeSignatureValid::assertion())
- }
-
- fn verify_tls13_signature(
- &self,
- _message: &[u8],
- _cert: &CertificateDer<'_>,
- _dss: &DigitallySignedStruct,
- ) -> Result {
- Ok(HandshakeSignatureValid::assertion())
- }
-
- fn supported_verify_schemes(&self) -> Vec {
- vec![
- SignatureScheme::ECDSA_NISTP256_SHA256,
- SignatureScheme::ECDSA_NISTP384_SHA384,
- SignatureScheme::ED25519,
- SignatureScheme::RSA_PSS_SHA256,
- SignatureScheme::RSA_PSS_SHA384,
- SignatureScheme::RSA_PSS_SHA512,
- SignatureScheme::RSA_PKCS1_SHA256,
- SignatureScheme::RSA_PKCS1_SHA384,
- SignatureScheme::RSA_PKCS1_SHA512,
- ]
- }
-}
diff --git a/probe/src/sni.rs b/probe/src/sni.rs
new file mode 100644
index 0000000..da8204a
--- /dev/null
+++ b/probe/src/sni.rs
@@ -0,0 +1,184 @@
+use anyhow::{Result, bail};
+use futures::future::join_all;
+use log::warn;
+use reports::probe::{Host, HostProbeResult, ProbeConfig, ProbeEvidence};
+use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
+use rustls::pki_types::{CertificateDer, ServerName, UnixTime};
+use rustls::{ClientConfig, DigitallySignedStruct, Error as TlsError, SignatureScheme};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+use tokio::time;
+use tokio_rustls::TlsConnector;
+
+pub async fn check_sni(
+ config: Option<&ProbeConfig>,
+ domain: Option<&str>,
+ timeout: Duration,
+ job_id: &str,
+ task_timeout_ms: u64,
+) -> Result>> {
+ let Some(domain) = domain else {
+ return Ok(None);
+ };
+ let Some(config) = config else {
+ bail!("no config");
+ };
+ let probing = join_all(config.hosts.iter().map(|host| async move {
+ let probe_evidence = probe_host(host, domain).await;
+ HostProbeResult {
+ probe_evidence,
+ host_id: host.id.clone(),
+ }
+ }));
+
+ match time::timeout(timeout, probing).await {
+ Ok(responses) => Ok(Some(responses)),
+ Err(_) => {
+ warn!("SNI checks expired for task {job_id}: timeout {task_timeout_ms}ms");
+ Ok(None)
+ }
+ }
+}
+
+async fn probe_host(host: &Host, target: &str) -> ProbeEvidence {
+ let timeout = Duration::from_secs(host.timeout_sec as u64);
+ let tcp = match time::timeout(timeout, TcpStream::connect((host.host.as_str(), 443))).await {
+ Ok(Ok(tcp)) => tcp,
+ Ok(Err(_)) | Err(_) => return ProbeEvidence::ConnectionError,
+ };
+
+ let tls_config = ClientConfig::builder()
+ .dangerous()
+ .with_custom_certificate_verifier(Arc::new(NoCertificateVerification))
+ .with_no_client_auth();
+ let connector = TlsConnector::from(Arc::new(tls_config));
+
+ let server_name = match ServerName::try_from(target.to_string()) {
+ Ok(server_name) => server_name,
+ Err(_) => return ProbeEvidence::ClientHello,
+ };
+
+ let mut tls = match time::timeout(timeout, connector.connect(server_name, tcp)).await {
+ Ok(Ok(tls)) => tls,
+ Ok(Err(_)) | Err(_) => return ProbeEvidence::ClientHello,
+ };
+
+ let request = format!(
+ "GET /{} HTTP/1.1\r\nHost: {}\r\nUser-Agent: cheburcheck-probe/{}\r\nRange: bytes=0-{}\r\nConnection: close\r\n\r\n",
+ host.file_path.trim_start_matches('/'),
+ target,
+ env!("CARGO_PKG_VERSION"),
+ host.min_data.saturating_sub(1)
+ );
+
+ if !matches!(
+ time::timeout(timeout, tls.write_all(request.as_bytes())).await,
+ Ok(Ok(()))
+ ) {
+ return ProbeEvidence::ClientHello;
+ }
+
+ let mut received = 0u32;
+ let mut headers_done = false;
+ let mut pending = Vec::new();
+ let mut buffer = [0u8; 8192];
+ loop {
+ match time::timeout(timeout, tls.read(&mut buffer)).await {
+ Ok(Ok(0)) | Err(_) => {
+ return if received >= host.min_data {
+ ProbeEvidence::Good
+ } else {
+ ProbeEvidence::DataTimeout { bytes: received }
+ };
+ }
+ Ok(Ok(bytes)) => {
+ add_response_body_bytes(
+ &buffer[..bytes],
+ &mut pending,
+ &mut headers_done,
+ &mut received,
+ );
+ if received >= host.min_data {
+ return ProbeEvidence::Good;
+ }
+ }
+ Ok(Err(_)) => {
+ return if received >= host.min_data {
+ ProbeEvidence::Good
+ } else {
+ ProbeEvidence::DataTimeout { bytes: received }
+ };
+ }
+ }
+ }
+}
+
+fn add_response_body_bytes(
+ chunk: &[u8],
+ pending: &mut Vec,
+ headers_done: &mut bool,
+ received: &mut u32,
+) {
+ if *headers_done {
+ *received = received.saturating_add(chunk.len() as u32);
+ return;
+ }
+
+ pending.extend_from_slice(chunk);
+ if let Some(body_start) = pending.windows(4).position(|window| window == b"\r\n\r\n") {
+ *headers_done = true;
+ let body_bytes = pending.len().saturating_sub(body_start + 4);
+ *received = received.saturating_add(body_bytes as u32);
+ pending.clear();
+ }
+}
+
+#[derive(Debug)]
+struct NoCertificateVerification;
+
+impl ServerCertVerifier for NoCertificateVerification {
+ fn verify_server_cert(
+ &self,
+ _end_entity: &CertificateDer<'_>,
+ _intermediates: &[CertificateDer<'_>],
+ _server_name: &ServerName<'_>,
+ _ocsp_response: &[u8],
+ _now: UnixTime,
+ ) -> Result {
+ Ok(ServerCertVerified::assertion())
+ }
+
+ fn verify_tls12_signature(
+ &self,
+ _message: &[u8],
+ _cert: &CertificateDer<'_>,
+ _dss: &DigitallySignedStruct,
+ ) -> Result {
+ Ok(HandshakeSignatureValid::assertion())
+ }
+
+ fn verify_tls13_signature(
+ &self,
+ _message: &[u8],
+ _cert: &CertificateDer<'_>,
+ _dss: &DigitallySignedStruct,
+ ) -> Result {
+ Ok(HandshakeSignatureValid::assertion())
+ }
+
+ fn supported_verify_schemes(&self) -> Vec {
+ vec![
+ SignatureScheme::ECDSA_NISTP256_SHA256,
+ SignatureScheme::ECDSA_NISTP384_SHA384,
+ SignatureScheme::ED25519,
+ SignatureScheme::RSA_PSS_SHA256,
+ SignatureScheme::RSA_PSS_SHA384,
+ SignatureScheme::RSA_PSS_SHA512,
+ SignatureScheme::RSA_PKCS1_SHA256,
+ SignatureScheme::RSA_PKCS1_SHA384,
+ SignatureScheme::RSA_PKCS1_SHA512,
+ ]
+ }
+}
diff --git a/probe/src/traceroute.rs b/probe/src/traceroute.rs
new file mode 100644
index 0000000..53cac39
--- /dev/null
+++ b/probe/src/traceroute.rs
@@ -0,0 +1,286 @@
+use etherparse::{
+ Icmpv4Type, Icmpv6Slice, Icmpv6Type, IpNumber, LaxNetSlice, LaxSlicedPacket, TransportSlice,
+ icmpv4, icmpv6,
+};
+use polling::{Event, Events, Poller};
+use reports::probe::{TcpTracerouteOutcome, TcpTracerouteResult};
+use socket2::{Domain, Protocol, SockAddr, Socket, Type};
+use std::io::{self, Read};
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+use std::time::{Duration, Instant};
+
+const HTTPS_PORT: u16 = 443;
+const HOP_TIMEOUT: Duration = Duration::from_secs(1);
+
+pub async fn tcp_traceroute(target: IpAddr, max_hops: u8) -> Option {
+ tokio::task::spawn_blocking(move || trace_blocking(target, max_hops))
+ .await
+ .map_err(|error| log::warn!("TCP traceroute task failed for {target}: {error}"))
+ .ok()?
+ .map_err(|error| log::warn!("TCP traceroute failed for {target}: {error}"))
+ .ok()
+}
+
+fn trace_blocking(target: IpAddr, max_hops: u8) -> io::Result {
+ let (domain, icmp_protocol) = match target {
+ IpAddr::V4(_) => (Domain::IPV4, Protocol::ICMPV4),
+ IpAddr::V6(_) => (Domain::IPV6, Protocol::ICMPV6),
+ };
+ let receiver = Socket::new(domain, Type::RAW, Some(icmp_protocol))?;
+ let mut last_icmp_hop = None;
+
+ for ttl in 1..=max_hops {
+ let tcp = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))?;
+ tcp.set_nonblocking(true)?;
+ match target {
+ IpAddr::V4(_) => tcp.set_ttl_v4(ttl as u32)?,
+ IpAddr::V6(_) => tcp.set_unicast_hops_v6(ttl as u32)?,
+ }
+ let unspecified = match target {
+ IpAddr::V4(_) => SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0),
+ IpAddr::V6(_) => SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0),
+ };
+ tcp.bind(&SockAddr::from(unspecified))?;
+ let destination = SockAddr::from(SocketAddr::new(target, HTTPS_PORT));
+ if let Err(error) = tcp.connect(&destination) {
+ if error.kind() == io::ErrorKind::ConnectionRefused {
+ return Ok(TcpTracerouteResult {
+ target,
+ result: TcpTracerouteOutcome::Rst { hop: ttl },
+ });
+ }
+ if !is_connect_in_progress(&error) {
+ return Err(error);
+ }
+ }
+ let source_port = tcp
+ .local_addr()?
+ .as_socket()
+ .map(|addr| addr.port())
+ .unwrap_or(0);
+ match wait_for_hop_response(&receiver, &tcp, target, source_port)? {
+ HopResponse::IcmpTimeExceeded => last_icmp_hop = Some(ttl),
+ HopResponse::Rst => {
+ return Ok(TcpTracerouteResult {
+ target,
+ result: TcpTracerouteOutcome::Rst { hop: ttl },
+ });
+ }
+ HopResponse::Connected => {
+ return Ok(TcpTracerouteResult {
+ target,
+ result: TcpTracerouteOutcome::Connected { hop: ttl },
+ });
+ }
+ HopResponse::Timeout => {}
+ }
+ }
+
+ Ok(TcpTracerouteResult {
+ target,
+ result: last_icmp_hop
+ .map(|hop| TcpTracerouteOutcome::IcmpTimeExceeded { hop })
+ .unwrap_or(TcpTracerouteOutcome::Timeout),
+ })
+}
+
+fn is_connect_in_progress(error: &io::Error) -> bool {
+ error.kind() == io::ErrorKind::WouldBlock
+ || error.raw_os_error() == Some(rustix::io::Errno::INPROGRESS.raw_os_error())
+}
+
+enum HopResponse {
+ IcmpTimeExceeded,
+ Rst,
+ Connected,
+ Timeout,
+}
+
+fn wait_for_hop_response(
+ receiver: &Socket,
+ tcp: &Socket,
+ target: IpAddr,
+ source_port: u16,
+) -> io::Result {
+ const ICMP_KEY: usize = 1;
+ const TCP_KEY: usize = 2;
+
+ let poller = Poller::new()?;
+ // SAFETY: both sockets remain alive and are removed from the poller before this function exits.
+ unsafe {
+ poller.add(receiver, Event::readable(ICMP_KEY))?;
+ if let Err(error) = poller.add(tcp, Event::writable(TCP_KEY)) {
+ poller.delete(receiver)?;
+ return Err(error);
+ }
+ }
+
+ let result = wait_on_poller(&poller, receiver, tcp, target, source_port);
+ let tcp_delete = poller.delete(tcp);
+ let receiver_delete = poller.delete(receiver);
+ let cleanup = tcp_delete.and(receiver_delete);
+ match result {
+ Ok(response) => cleanup.map(|()| response),
+ Err(error) => Err(error),
+ }
+}
+
+fn wait_on_poller(
+ poller: &Poller,
+ receiver: &Socket,
+ tcp: &Socket,
+ target: IpAddr,
+ source_port: u16,
+) -> io::Result {
+ const ICMP_KEY: usize = 1;
+ const TCP_KEY: usize = 2;
+
+ let deadline = Instant::now() + HOP_TIMEOUT;
+ let mut watch_tcp = true;
+ let mut buffer = [0u8; 2048];
+ let mut events = Events::new();
+
+ loop {
+ let remaining = deadline.saturating_duration_since(Instant::now());
+ if remaining.is_zero() {
+ return Ok(HopResponse::Timeout);
+ }
+ events.clear();
+ match poller.wait(&mut events, Some(remaining)) {
+ Ok(0) => return Ok(HopResponse::Timeout),
+ Ok(_) => {}
+ Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
+ Err(error) => return Err(error),
+ }
+
+ if watch_tcp && events.iter().any(|event| event.key == TCP_KEY) {
+ match tcp.take_error()? {
+ Some(error) if error.kind() == io::ErrorKind::ConnectionRefused => {
+ return Ok(HopResponse::Rst);
+ }
+ None => return Ok(HopResponse::Connected),
+ Some(_) => watch_tcp = false,
+ }
+ }
+
+ if events.iter().any(|event| event.key == ICMP_KEY) {
+ let mut raw = receiver;
+ match raw.read(&mut buffer) {
+ Ok(bytes) if is_matching_time_exceeded(&buffer[..bytes], target, source_port) => {
+ return Ok(HopResponse::IcmpTimeExceeded);
+ }
+ Ok(_) => poller.modify(receiver, Event::readable(ICMP_KEY))?,
+ Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
+ poller.modify(receiver, Event::readable(ICMP_KEY))?;
+ }
+ Err(error) => return Err(error),
+ }
+ }
+ }
+}
+fn is_matching_time_exceeded(packet: &[u8], target: IpAddr, source_port: u16) -> bool {
+ match target {
+ IpAddr::V4(target) => matching_ipv4_time_exceeded(packet, target, source_port),
+ IpAddr::V6(target) => matching_ipv6_time_exceeded(packet, target, source_port),
+ }
+}
+
+fn matching_ipv4_time_exceeded(packet: &[u8], target: Ipv4Addr, source_port: u16) -> bool {
+ let Ok(outer) = LaxSlicedPacket::from_ip(packet) else {
+ return false;
+ };
+ let Some(TransportSlice::Icmpv4(icmp)) = outer.transport else {
+ return false;
+ };
+ matches!(
+ icmp.icmp_type(),
+ Icmpv4Type::TimeExceeded(icmpv4::TimeExceededCode::TtlExceededInTransit)
+ ) && matching_quoted_tcp(icmp.payload(), IpAddr::V4(target), source_port)
+}
+
+fn matching_ipv6_time_exceeded(packet: &[u8], target: Ipv6Addr, source_port: u16) -> bool {
+ let icmp = if packet.first().map(|byte| byte >> 4) == Some(6) {
+ let Ok(outer) = LaxSlicedPacket::from_ip(packet) else {
+ return false;
+ };
+ let Some(TransportSlice::Icmpv6(icmp)) = outer.transport else {
+ return false;
+ };
+ icmp
+ } else {
+ let Ok(icmp) = Icmpv6Slice::from_slice(packet) else {
+ return false;
+ };
+ icmp
+ };
+ matches!(
+ icmp.icmp_type(),
+ Icmpv6Type::TimeExceeded(icmpv6::TimeExceededCode::HopLimitExceeded)
+ ) && matching_quoted_tcp(icmp.payload(), IpAddr::V6(target), source_port)
+}
+
+fn matching_quoted_tcp(packet: &[u8], target: IpAddr, source_port: u16) -> bool {
+ let Ok(quoted) = LaxSlicedPacket::from_ip(packet) else {
+ return false;
+ };
+ let (ip_number, payload) = match (quoted.net, target) {
+ (Some(LaxNetSlice::Ipv4(ip)), IpAddr::V4(target))
+ if ip.header().destination_addr() == target =>
+ {
+ (ip.payload().ip_number, ip.payload().payload)
+ }
+ (Some(LaxNetSlice::Ipv6(ip)), IpAddr::V6(target))
+ if ip.header().destination_addr() == target =>
+ {
+ (ip.payload().ip_number, ip.payload().payload)
+ }
+ _ => return false,
+ };
+ if ip_number != IpNumber::TCP {
+ return false;
+ }
+ let Some(ports) = payload.get(..4) else {
+ return false;
+ };
+ u16::from_be_bytes([ports[0], ports[1]]) == source_port
+ && u16::from_be_bytes([ports[2], ports[3]]) == HTTPS_PORT
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn matches_ipv4_time_exceeded_for_tcp_flow() {
+ let target = Ipv4Addr::new(203, 0, 113, 10);
+ let mut packet = vec![0; 20 + 8 + 20 + 8];
+ packet[0] = 0x45;
+ packet[2..4].copy_from_slice(&56u16.to_be_bytes());
+ packet[9] = IpNumber::ICMP.0;
+ packet[20] = 11;
+ packet[28] = 0x45;
+ packet[30..32].copy_from_slice(&28u16.to_be_bytes());
+ packet[37] = 6;
+ packet[44..48].copy_from_slice(&target.octets());
+ packet[48..50].copy_from_slice(&42_000u16.to_be_bytes());
+ packet[50..52].copy_from_slice(&HTTPS_PORT.to_be_bytes());
+
+ assert!(matching_ipv4_time_exceeded(&packet, target, 42_000));
+ assert!(!matching_ipv4_time_exceeded(&packet, target, 42_001));
+ }
+
+ #[test]
+ fn matches_ipv6_time_exceeded_for_tcp_flow() {
+ let target = "2001:db8::10".parse::().unwrap();
+ let mut packet = vec![0; 8 + 40 + 8];
+ packet[0] = 3;
+ packet[8] = 0x60;
+ packet[14] = 6;
+ packet[32..48].copy_from_slice(&target.octets());
+ packet[48..50].copy_from_slice(&42_000u16.to_be_bytes());
+ packet[50..52].copy_from_slice(&HTTPS_PORT.to_be_bytes());
+
+ assert!(matching_ipv6_time_exceeded(&packet, target, 42_000));
+ assert!(!matching_ipv6_time_exceeded(&packet, target, 42_001));
+ }
+}
diff --git a/reports/src/probe.rs b/reports/src/probe.rs
index abbb232..b749eea 100644
--- a/reports/src/probe.rs
+++ b/reports/src/probe.rs
@@ -1,4 +1,5 @@
use serde::{Deserialize, Serialize};
+use std::net::IpAddr;
#[derive(Clone, Serialize, Deserialize)]
pub struct ProbeStatus<'a> {
@@ -13,6 +14,8 @@ pub struct ProbeConfig {
pub task_timeout_ms: u64,
pub published_at: String,
pub hosts: Vec,
+ #[serde(default)]
+ pub control_hosts: Vec,
}
#[derive(Clone, Serialize, Deserialize)]
@@ -35,7 +38,8 @@ pub enum HostType {
pub struct ProbeTask<'a> {
pub id: String,
pub query_id: String,
- pub target: &'a str,
+ pub domain: Option<&'a str>,
+ pub ip: IpAddr,
pub created_at: String,
pub timeout_ms: u64,
}
@@ -45,6 +49,30 @@ pub struct ProbeResultEvent {
pub job_id: String,
pub probe_id: String,
pub host_results: Vec,
+ pub target_traceroute: Option,
+ pub control_traceroute: Option,
+}
+
+#[derive(Clone, Serialize, Deserialize)]
+pub struct ProbeResult {
+ pub responses: Option>,
+ pub target_traceroute: Option,
+ pub control_traceroute: Option,
+}
+
+#[derive(Clone, Serialize, Deserialize)]
+pub struct TcpTracerouteResult {
+ pub target: IpAddr,
+ pub result: TcpTracerouteOutcome,
+}
+
+#[derive(Clone, Serialize, Deserialize)]
+#[serde(tag = "type")]
+pub enum TcpTracerouteOutcome {
+ Rst { hop: u8 },
+ Connected { hop: u8 },
+ IcmpTimeExceeded { hop: u8 },
+ Timeout,
}
#[derive(Clone, Serialize, Deserialize)]
diff --git a/website/Cargo.toml b/website/Cargo.toml
index 2b83cac..cb25e7a 100644
--- a/website/Cargo.toml
+++ b/website/Cargo.toml
@@ -1,6 +1,6 @@
[package]
name = "website"
-version = "1.1.0"
+version = "1.1.1"
edition = "2024"
[dependencies]
diff --git a/website/migrations/20260817000000_probe_traceroutes.sql b/website/migrations/20260817000000_probe_traceroutes.sql
new file mode 100644
index 0000000..88e6006
--- /dev/null
+++ b/website/migrations/20260817000000_probe_traceroutes.sql
@@ -0,0 +1,5 @@
+ALTER TABLE probe_reports
+ ADD COLUMN target_hop_count SMALLINT,
+ ADD COLUMN target_trace_result VARCHAR(32),
+ ADD COLUMN control_hop_count SMALLINT,
+ ADD COLUMN control_trace_result VARCHAR(32);
diff --git a/website/probe-hosts.toml b/website/probe-hosts.toml
index 52fdf3e..4afcbba 100644
--- a/website/probe-hosts.toml
+++ b/website/probe-hosts.toml
@@ -1,5 +1,6 @@
timeout_sec = 3
min_data = 65536
+control_hosts = ["kinopoisk.ru", "wildberries.ru", "yandex.ru", "mail.ru"]
[[hosts]]
id = "hil-hetzner" # hil-speed.hetzner.com
diff --git a/website/src/api/probe.rs b/website/src/api/probe.rs
index cf594fc..4013548 100644
--- a/website/src/api/probe.rs
+++ b/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 = 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)> =
+ 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::().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,
) -> 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, 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![],
+ }
+ }
+}
diff --git a/website/src/mqtt.rs b/website/src/mqtt.rs
index 8a9af34..948461c 100644
--- a/website/src/mqtt.rs
+++ b/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,
hosts: Vec,
}
@@ -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 {
- 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
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, PublishError> {
+struct ParsedProbeConfig {
+ hosts: Vec,
+ control_hosts: Vec,
+}
+
+fn parse_probe_hosts(contents: &str) -> Result {
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, 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,
});
}
}
diff --git a/website/src/mqtt_auth.rs b/website/src/mqtt_auth.rs
index f653f31..3c7cd78 100644
--- a/website/src/mqtt_auth.rs
+++ b/website/src/mqtt_auth.rs
@@ -56,7 +56,7 @@ pub async fn auth(
pool: &rocket::State,
) -> Json {
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())
|