From 1a06b8a0dc2799d4fbce1a6bcfbf8f242f0d8d13 Mon Sep 17 00:00:00 2001 From: Lowder Date: Mon, 17 Aug 2026 00:45:31 +0500 Subject: [PATCH] feat: ip probe traceroute --- .github/workflows/probe-build.yml | 21 +- Cargo.lock | 38 ++- frontend/src/lib/api/probe.ts | 5 +- .../lib/components/result/ProbeTable.svelte | 23 +- frontend/src/routes/check/+page.svelte | 4 +- frontend/vite.config.ts | 2 +- probe/Cargo.toml | 7 +- probe/Dockerfile | 7 +- probe/README.md | 8 +- probe/debian/cheburprobe.service | 2 + probe/src/main.rs | 261 ++++++---------- probe/src/sni.rs | 184 +++++++++++ probe/src/traceroute.rs | 286 ++++++++++++++++++ reports/src/probe.rs | 30 +- website/Cargo.toml | 2 +- .../20260817000000_probe_traceroutes.sql | 5 + website/probe-hosts.toml | 1 + website/src/api/probe.rs | 178 ++++++++++- website/src/mqtt.rs | 39 ++- website/src/mqtt_auth.rs | 11 +- 20 files changed, 877 insertions(+), 237 deletions(-) create mode 100644 probe/src/sni.rs create mode 100644 probe/src/traceroute.rs create mode 100644 website/migrations/20260817000000_probe_traceroutes.sql diff --git a/.github/workflows/probe-build.yml b/.github/workflows/probe-build.yml index bab0a33..6fc7581 100644 --- a/.github/workflows/probe-build.yml +++ b/.github/workflows/probe-build.yml @@ -21,28 +21,33 @@ jobs: - name: Install Rust uses: dtolnay/rust-toolchain@stable + with: + targets: x86_64-unknown-linux-musl + + - name: Install musl toolchain + run: sudo apt-get update && sudo apt-get install --yes musl-tools - name: Install cargo-deb run: cargo install cargo-deb - name: Build release binary - run: cargo build --release --package probe --bin cheburprobe + run: cargo build --release --package probe --bin cheburprobe --target x86_64-unknown-linux-musl - name: Build Debian package - run: cargo deb --package probe --no-build -- --bin cheburprobe + run: cargo deb --package probe --target x86_64-unknown-linux-musl --no-strip --no-build -- --bin cheburprobe - name: Upload Linux amd64 binary uses: actions/upload-artifact@v7 with: name: cheburprobe-linux-amd64 - path: target/release/cheburprobe + path: target/x86_64-unknown-linux-musl/release/cheburprobe compression-level: 0 - name: Upload Linux amd64 Debian package uses: actions/upload-artifact@v7 with: name: cheburprobe-linux-amd64-deb - path: target/debian/*.deb + path: target/x86_64-unknown-linux-musl/debian/*.deb compression-level: 0 linux-arm64: @@ -61,25 +66,25 @@ jobs: uses: houseabsolute/actions-rust-cross@v1 with: command: build - target: "aarch64-unknown-linux-gnu" + target: "aarch64-unknown-linux-musl" args: "--release --package probe --bin cheburprobe" strip: true - name: Build Debian package - run: cargo deb --package probe --target aarch64-unknown-linux-gnu --no-strip --no-build -- --bin cheburprobe + run: cargo deb --package probe --target aarch64-unknown-linux-musl --no-strip --no-build -- --bin cheburprobe - name: Upload Linux arm64 binary uses: actions/upload-artifact@v7 with: name: cheburprobe-linux-arm64 - path: target/aarch64-unknown-linux-gnu/release/cheburprobe + path: target/aarch64-unknown-linux-musl/release/cheburprobe compression-level: 0 - name: Upload Linux arm64 Debian package uses: actions/upload-artifact@v7 with: name: cheburprobe-linux-arm64-deb - path: target/aarch64-unknown-linux-gnu/debian/*.deb + path: target/aarch64-unknown-linux-musl/debian/*.deb compression-level: 0 windows-amd64: diff --git a/Cargo.lock b/Cargo.lock index 136e11f..d92696d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -97,6 +97,12 @@ dependencies = [ "rustversion", ] +[[package]] +name = "arrayvec" +version = "0.7.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" + [[package]] name = "async-stream" version = "0.3.6" @@ -806,6 +812,15 @@ dependencies = [ "windows-sys 0.48.0", ] +[[package]] +name = "etherparse" +version = "0.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "17304d06addb3283cdc4bd528e42dd95e73c8ee2d6492ffce415e93660885449" +dependencies = [ + "arrayvec", +] + [[package]] name = "event-listener" version = "5.4.1" @@ -2531,6 +2546,20 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" +[[package]] +name = "polling" +version = "3.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218" +dependencies = [ + "cfg-if", + "concurrent-queue", + "hermit-abi", + "pin-project-lite", + "rustix", + "windows-sys 0.61.2", +] + [[package]] name = "portable-atomic" version = "1.13.1" @@ -2603,18 +2632,23 @@ dependencies = [ [[package]] name = "probe" -version = "0.1.0" +version = "0.1.1" dependencies = [ "anyhow", "clap", "env_logger", + "etherparse", "futures", "log", + "polling", + "rand 0.8.5", "reports", "rumqttc 0.25.1", + "rustix", "rustls 0.23.35", "serde", "serde_json", + "socket2 0.6.3", "tokio", "tokio-rustls 0.26.4", ] @@ -4671,7 +4705,7 @@ dependencies = [ [[package]] name = "website" -version = "1.1.0" +version = "1.1.1" dependencies = [ "dotenvy", "env_logger", diff --git a/frontend/src/lib/api/probe.ts b/frontend/src/lib/api/probe.ts index fea24db..d92819b 100644 --- a/frontend/src/lib/api/probe.ts +++ b/frontend/src/lib/api/probe.ts @@ -17,8 +17,9 @@ export type ProbeResult = { region?: string | null; provider?: string | null; asn?: string | null; - verdict: "uncertain" | "sni_block" | "whitelist" | "ok"; - host_results: ProbeHostResult[]; + verdict: "uncertain" | "sni_block" | "tspu_block" | "whitelist" | "ok"; + host_results: ProbeHostResult[] | null; + target_hop: number | null; }; export type ProbeStatus = { diff --git a/frontend/src/lib/components/result/ProbeTable.svelte b/frontend/src/lib/components/result/ProbeTable.svelte index e9a300b..92ec4bf 100644 --- a/frontend/src/lib/components/result/ProbeTable.svelte +++ b/frontend/src/lib/components/result/ProbeTable.svelte @@ -49,6 +49,13 @@ const verdictStyles = { bg: "bg-red-500/10", border: "border-red-500/20", }, + tspu_block: { + icon: CircleX, + text: "ТСПУ Блок", + class: "text-red-500", + bg: "bg-red-500/10", + border: "border-red-500/20", + }, whitelist: { icon: ShieldCheck, text: "Белый список", @@ -84,10 +91,10 @@ const verdictStyles = { 0 ? 'bg-green-500 animate-pulse' : 'bg-neutral-600'}`} > - Сканеров онлайн: {status.online_probes} + Сканеров онлайн:{status.online_probes}
- Получено ответов: {probes.length} / {status.online_probes} + Получено ответов:{probes.length} /{status.online_probes}
@@ -123,13 +130,12 @@ const verdictStyles = { {#each probes as probe (probe.probe_id)} - {@const style = verdictStyles[(isStaticBlocked && probe.verdict === "ok") ? "cdn_block" : probe.verdict]} + {@const style = verdictStyles[(isStaticBlocked && probe.host_results?.length !== 0 && probe.verdict === "ok") ? "cdn_block" : probe.verdict]} {@const isExpanded = !!expandedRows[probe.probe_id]} toggleRow(probe.probe_id)} onkeydown={(e) => e.key === 'Enter' && toggleRow(probe.probe_id)} - role="button" tabindex="0" > @@ -160,6 +166,11 @@ const verdictStyles = { {#if isExpanded} +
+ Блокировка на ТСПУ + {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())