+
+ {#if probe.dns}
+
+
+
+ DNS-проверка
+
+ 0 ? 'text-amber-400' : 'text-green-400'}`}
+ >
+ {probe.dns.spoofing_detected
+ ? `Возможна подмена (${probe.dns.suspicious_provider_count}/${probe.dns.verdict_threshold})`
+ : probe.dns.suspicious_provider_count > 0
+ ? `Недостаточно подтверждений (${probe.dns.suspicious_provider_count}/${probe.dns.verdict_threshold})`
+ : "Подмена не выявлена"}
+
+
+
+
+
+
+ | Провайдер |
+ {#each dnsProtocols as protocol}
+
+ {dnsProtocolLabels[protocol]}
+ |
+ {/each}
+
+
+
+ {#each groupDnsObservations(probe.dns.observations) as provider}
+
+ |
+ {provider.provider}
+ |
+ {#each dnsProtocols as protocol}
+ {@const observation = provider.protocols[protocol]}
+
+ {#if observation}
+ {@const dnsStatus = dnsObservationStatus(observation)}
+
+
+ {dnsStatus === "spoofed" ? "Подозрительно" : dnsStatus === "error" ? "Ошибка" : "Норма"}
+
+
+ {observation.metadata.response_codes.join(", ") || "—"}
+ ·
+ {observation.metadata.ipv4_count}/{observation.metadata.ipv6_count}
+
+ |
+ {/each}
+
+ {/each}
+
+
+
+
+ Сравниваются код ответа и количество уникальных
+ IPv4/IPv6. Выполнено по
+ {probe.dns.samples_per_protocol || 1}
+ запроса на протокол. Вердикт требует подтверждения от
+ {probe.dns.verdict_threshold || 2}
+ DNS-провайдеров.
+
+
+
+ CDN-проверка
+
+
{#each probe.host_results as host}
Сервер {host.host_id}
- ({host.host === "Blacklist" ? "в заблокированных" : "в доступных"} диапазонах)
+ ({host.host === "Blacklist" ? "в заблокированных" : "в доступных"}
+ диапазонах)
{#if host.probe_evidence.type === 'Good'}
@@ -176,7 +337,9 @@ 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/src/routes/kb/probing/+page.svelte b/frontend/src/routes/kb/probing/+page.svelte
index 6d9c758..a9faca5 100644
--- a/frontend/src/routes/kb/probing/+page.svelte
+++ b/frontend/src/routes/kb/probing/+page.svelte
@@ -1,6 +1,5 @@
@@ -48,14 +47,63 @@ import KbNote from "$lib/components/kb/KbNote.svelte";
сразу, не дожидаясь завершения всех проверок.
- Сканер не пытается открыть сайт целиком или подключиться напрямую к серверу.
- Вместо этого он делает небольшую контрольную загрузку с тестовых хостов.
- Часть этих хостов используется как условно заблокированные направления,
- часть — как контрольные доступные направления. Это помогает отличить разные
- типы поведения: обрыв на этапе начала HTTPS-соединения, недогрузку данных
- или нормальную передачу.
+ Сканер не загружает целевой сайт целиком. Для домена он делает небольшие
+ контрольные HTTPS-загрузки с тестовых хостов, подставляя проверяемое имя в
+ SNI, а также сравнивает ответы нескольких публичных DNS-сервисов.
+
+ Методика сканирования
+
+
+ Сканер подключается к настроенным тестовым хостам, но передаёт проверяемый
+ домен в TLS ClientHello и HTTP Host. Затем он оценивает, удалось ли
+ завершить TLS-обмен и получить минимальный объём данных. Для проверки
+ IP-адреса без домена этот этап пропускается.
+
+
+
+
+ Проверка DNS
+
+ Для доменных имён сканер обращается к Google DNS, Cloudflare, Quad9 и Яндекс
+ DNS. Запросы к каждому сервису выполняются параллельно по UDP, TCP, DNS over
+ HTTPS (DoH) и DNS over TLS (DoT). Каждый вариант запускается несколько раз,
+ чтобы обычная ротация адресов CDN и балансировщиков меньше влияла на
+ результат.
+
+
+ Сами IP-адреса могут законно отличаться между запросами, поэтому сканер не
+ требует их точного совпадения. Для сравнения используется форма ответа: код
+ DNS и количество уникальных IPv4- и IPv6-адресов. Ответы DoH и DoT одного
+ провайдера служат зашифрованным ориентиром. UDP или TCP считается
+ подозрительным, когда его форма отличается от совпавших между собой
+ зашифрованных ответов.
+
+
+ Одиночное расхождение ещё не означает подмену. Итоговый вердикт появляется,
+ только когда подозрительный результат независимо подтверждают несколько
+ DNS-провайдеров — обычно не менее двух. Несколько протоколов одного
+ провайдера считаются одним подтверждением.
+
+
Что показывает результат
@@ -66,30 +114,47 @@ import KbNote from "$lib/components/kb/KbNote.svelte";
-
- Доступен – сайт не найден в списках CDN, не блокируется на
- контрольных хостах, но может блокироваться на зарубежных CDN.
+ Доступен
+ – сайт не найден в списках CDN, не блокируется на контрольных хостах, но
+ может блокироваться на зарубежных CDN.
-
- CDN Блок (16-20) – сайт найден в списках CDN, не блокируется на
- контрольных хостах, но блокируется на зарубежных CDN.
+ CDN Блок (16-20)
+ – сайт найден в списках CDN, не блокируется на контрольных хостах, но
+ блокируется на зарубежных CDN.
-
- SNI Блок – похоже на блокировку по домену сайта в начале
- HTTPS-соединения. На практике это похоже на ситуацию, когда оборудование
- оператора (ТСПУ) видит имя домена и разрывает соединение до передачи
- полезных данных.
+ SNI Блок
+ – похоже на блокировку по домену сайта в начале HTTPS-соединения. На
+ практике это похоже на ситуацию, когда оборудование оператора (ТСПУ) видит
+ имя домена и разрывает соединение до передачи полезных данных.
+ -
+ Подмена DNS
+ – открытые DNS-запросы по UDP или TCP у нескольких независимых провайдеров
+ отличаются от их ответов по защищённым протоколам DoH и DoT. Если
+ одновременно обнаружена SNI-блокировка, в общем вердикте приоритет
+ получает
+ SNI Блок.
+
+
-
Белый список – похоже на блокировку по домену сайта в начале
- HTTPS-соединения. Вероятнее всего, указывает на блокировку сайта на ТСПУ.
+ >Исключение для CDN – данные передаются в полном объеме при
+ подстановке домена в SNI при подключении к заблокированному серверу.
-
- Неясно – данных недостаточно или ответы противоречат друг другу.
+ Неясно
+ – данных недостаточно или ответы противоречат друг другу.
@@ -120,6 +185,13 @@ import KbNote from "$lib/components/kb/KbNote.svelte";
большинство ответов и на то, какие именно хосты дали сбой: контрольные,
проверочные или все сразу.
+
+ В раскрытой строке результата DNS-проверка показана отдельно от
+ CDN-проверки. Для каждого DNS-провайдера видны UDP, TCP, DoH и DoT, код
+ ответа, счётчик IPv4/IPv6 и оценка расхождения. Ошибка запроса означает лишь
+ то, что конкретное измерение не удалось, и сама по себе не считается
+ подменой.
+
= ServerGroup {
+ ips: &[IpAddr::V4(Ipv4Addr::new(77, 88, 8, 8))],
+ server_name: "common.dot.dns.yandex.net",
+ path: "/dns-query",
+};
+
+pub async fn check_dns(
+ domain: Option<&str>,
+ remaining: Duration,
+ samples_per_protocol: u8,
+ spoofing_provider_threshold: u8,
+) -> Option {
+ let domain = domain?;
+ let timeout = remaining.min(LOOKUP_TIMEOUT);
+ if timeout.is_zero() {
+ return None;
+ }
+
+ let providers = [
+ ("google", &GOOGLE),
+ ("cloudflare", &CLOUDFLARE),
+ ("quad9", &QUAD9),
+ ("yandex", &YANDEX),
+ ];
+ let mut observations = join_all(providers.into_iter().flat_map(|(provider, servers)| {
+ [
+ (DnsProtocol::Udp, resolver_config(servers, DnsProtocol::Udp)),
+ (DnsProtocol::Tcp, resolver_config(servers, DnsProtocol::Tcp)),
+ (DnsProtocol::Doh, ResolverConfig::https(servers)),
+ (DnsProtocol::Dot, ResolverConfig::tls(servers)),
+ ]
+ .map(|(protocol, config)| {
+ lookup(
+ domain,
+ provider,
+ protocol,
+ config,
+ timeout,
+ samples_per_protocol,
+ )
+ })
+ }))
+ .await;
+
+ mark_suspected_spoofing(&mut observations);
+ let suspicious_provider_count = suspicious_provider_count(&observations);
+ let spoofing_detected = suspicious_provider_count >= spoofing_provider_threshold;
+
+ Some(DnsProbeResult {
+ spoofing_detected,
+ suspicious_provider_count,
+ verdict_threshold: spoofing_provider_threshold,
+ samples_per_protocol,
+ observations,
+ })
+}
+
+fn suspicious_provider_count(observations: &[DnsObservation]) -> u8 {
+ ["google", "cloudflare", "quad9", "yandex"]
+ .into_iter()
+ .filter(|provider| {
+ observations.iter().any(|observation| {
+ observation.provider == *provider && observation.suspected_spoofing
+ })
+ })
+ .count() as u8
+}
+
+fn mark_suspected_spoofing(observations: &mut [DnsObservation]) {
+ for provider in ["google", "cloudflare", "quad9", "yandex"] {
+ let doh = observations
+ .iter()
+ .find(|item| item.provider == provider && matches!(item.protocol, DnsProtocol::Doh))
+ .filter(|item| successful_outcome(item).is_some());
+ let dot = observations
+ .iter()
+ .find(|item| item.provider == provider && matches!(item.protocol, DnsProtocol::Dot))
+ .filter(|item| successful_outcome(item).is_some());
+ let Some(reference) = doh
+ .zip(dot)
+ .and_then(|(doh, dot)| (doh.metadata == dot.metadata).then(|| doh.metadata.clone()))
+ else {
+ continue;
+ };
+
+ for observation in observations.iter_mut().filter(|item| {
+ item.provider == provider
+ && matches!(item.protocol, DnsProtocol::Udp | DnsProtocol::Tcp)
+ }) {
+ observation.suspected_spoofing =
+ successful_outcome(observation).is_some() && observation.metadata != reference;
+ }
+ }
+}
+
+fn successful_outcome(observation: &DnsObservation) -> Option<&DnsOutcome> {
+ (!matches!(observation.outcome, DnsOutcome::Error { .. })).then_some(&observation.outcome)
+}
+
+fn resolver_config(servers: &ServerGroup<'_>, protocol: DnsProtocol) -> ResolverConfig {
+ let name_servers = match protocol {
+ DnsProtocol::Udp => servers.udp().collect(),
+ DnsProtocol::Tcp => servers.tcp().collect(),
+ DnsProtocol::Doh | DnsProtocol::Dot => unreachable!(),
+ };
+ ResolverConfig::from_parts(None, Vec::new(), name_servers)
+}
+
+async fn lookup(
+ domain: &str,
+ provider: &str,
+ protocol: DnsProtocol,
+ config: ResolverConfig,
+ timeout: Duration,
+ samples_per_protocol: u8,
+) -> DnsObservation {
+ let mut options = ResolverOpts::default();
+ options.ip_strategy = LookupIpStrategy::Ipv4AndIpv6;
+ options.attempts = 1;
+ options.timeout = timeout;
+ options.cache_size = 0;
+ let (outcome, metadata) = match hickory_resolver::Resolver::builder_with_config(
+ config,
+ TokioRuntimeProvider::default(),
+ )
+ .with_options(options)
+ .build()
+ {
+ Ok(resolver) => {
+ let samples = join_all(
+ (0..samples_per_protocol)
+ .map(|_| tokio::time::timeout(timeout, resolver.lookup_ip(domain))),
+ )
+ .await;
+ let mut addresses = Vec::new();
+ let mut no_records = false;
+ let mut errors = Vec::new();
+ let mut response_codes = Vec::new();
+ for sample in samples {
+ match sample {
+ Ok(Ok(lookup)) => {
+ addresses.extend(lookup.iter());
+ response_codes.push("NoError".to_string());
+ }
+ Ok(Err(NetError::Dns(DnsError::NoRecordsFound(no_records_error)))) => {
+ no_records = true;
+ response_codes.push(no_records_error.response_code.to_string());
+ }
+ Ok(Err(NetError::Dns(DnsError::ResponseCode(code)))) => {
+ response_codes.push(code.to_string());
+ errors.push(format!("error response: {code}"));
+ }
+ Ok(Err(error)) => errors.push(error.to_string()),
+ Err(_) => errors.push("lookup timed out".to_string()),
+ }
+ }
+ response_codes.sort();
+ response_codes.dedup();
+ let outcome = if !addresses.is_empty() {
+ addresses.sort_unstable();
+ addresses.dedup();
+ DnsOutcome::Answer {
+ addresses: addresses.clone(),
+ }
+ } else if no_records {
+ DnsOutcome::NoRecords
+ } else {
+ errors.sort();
+ errors.dedup();
+ DnsOutcome::Error {
+ message: errors.join("; "),
+ }
+ };
+ let metadata = DnsResponseMetadata {
+ response_codes,
+ ipv4_count: addresses.iter().filter(|address| address.is_ipv4()).count() as u16,
+ ipv6_count: addresses.iter().filter(|address| address.is_ipv6()).count() as u16,
+ };
+ (outcome, metadata)
+ }
+ Err(error) => (
+ DnsOutcome::Error {
+ message: error.to_string(),
+ },
+ DnsResponseMetadata::default(),
+ ),
+ };
+
+ DnsObservation {
+ provider: provider.to_string(),
+ protocol,
+ outcome,
+ suspected_spoofing: false,
+ metadata,
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::{mark_suspected_spoofing, suspicious_provider_count};
+ use reports::probe::{DnsObservation, DnsOutcome, DnsProtocol, DnsResponseMetadata};
+ use std::net::{IpAddr, Ipv4Addr};
+
+ #[test]
+ fn normalized_answers_compare_independent_of_order() {
+ let mut left = vec![
+ IpAddr::V4(Ipv4Addr::new(192, 0, 2, 2)),
+ IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)),
+ ];
+ let right = vec![
+ IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)),
+ IpAddr::V4(Ipv4Addr::new(192, 0, 2, 2)),
+ ];
+ left.sort_unstable();
+ assert_eq!(left, right);
+ }
+
+ #[test]
+ fn protocol_rotation_is_not_mistaken_for_spoofing() {
+ let answer = |last| DnsOutcome::Answer {
+ addresses: vec![IpAddr::V4(Ipv4Addr::new(192, 0, 2, last))],
+ };
+ let observation = |protocol, outcome| DnsObservation {
+ provider: "quad9".to_string(),
+ protocol,
+ outcome,
+ suspected_spoofing: false,
+ metadata: DnsResponseMetadata {
+ response_codes: vec!["NoError".to_string()],
+ ipv4_count: 1,
+ ipv6_count: 0,
+ },
+ };
+ let mut observations = vec![
+ observation(DnsProtocol::Udp, answer(1)),
+ observation(DnsProtocol::Tcp, answer(2)),
+ observation(DnsProtocol::Doh, answer(1)),
+ observation(DnsProtocol::Dot, answer(2)),
+ ];
+ mark_suspected_spoofing(&mut observations);
+ assert!(observations.iter().all(|item| !item.suspected_spoofing));
+ }
+
+ #[test]
+ fn different_addresses_with_the_same_metadata_are_not_suspicious() {
+ let answer = |last| DnsOutcome::Answer {
+ addresses: vec![IpAddr::V4(Ipv4Addr::new(192, 0, 2, last))],
+ };
+ let observation = |protocol, outcome| DnsObservation {
+ provider: "quad9".to_string(),
+ protocol,
+ outcome,
+ suspected_spoofing: false,
+ metadata: DnsResponseMetadata {
+ response_codes: vec!["NoError".to_string()],
+ ipv4_count: 1,
+ ipv6_count: 0,
+ },
+ };
+ let mut observations = vec![
+ observation(DnsProtocol::Udp, answer(99)),
+ observation(DnsProtocol::Tcp, answer(1)),
+ observation(DnsProtocol::Doh, answer(1)),
+ observation(DnsProtocol::Dot, answer(1)),
+ ];
+ mark_suspected_spoofing(&mut observations);
+ assert!(observations.iter().all(|item| !item.suspected_spoofing));
+ }
+
+ #[test]
+ fn plaintext_metadata_difference_is_suspicious() {
+ let observation = |protocol, ipv4_count| DnsObservation {
+ provider: "quad9".to_string(),
+ protocol,
+ outcome: DnsOutcome::Answer {
+ addresses: vec![IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1))],
+ },
+ suspected_spoofing: false,
+ metadata: DnsResponseMetadata {
+ response_codes: vec!["NoError".to_string()],
+ ipv4_count,
+ ipv6_count: 0,
+ },
+ };
+ let mut observations = vec![
+ observation(DnsProtocol::Udp, 2),
+ observation(DnsProtocol::Tcp, 1),
+ observation(DnsProtocol::Doh, 1),
+ observation(DnsProtocol::Dot, 1),
+ ];
+ mark_suspected_spoofing(&mut observations);
+ assert!(observations[0].suspected_spoofing);
+ assert!(
+ observations[1..]
+ .iter()
+ .all(|item| !item.suspected_spoofing)
+ );
+ }
+
+ #[test]
+ fn multiple_protocols_from_one_provider_count_as_one_vote() {
+ let observation = |provider: &str, protocol| DnsObservation {
+ provider: provider.to_string(),
+ protocol,
+ outcome: DnsOutcome::NoRecords,
+ suspected_spoofing: true,
+ metadata: DnsResponseMetadata::default(),
+ };
+ let observations = vec![
+ observation("quad9", DnsProtocol::Udp),
+ observation("quad9", DnsProtocol::Tcp),
+ observation("yandex", DnsProtocol::Udp),
+ ];
+ assert_eq!(suspicious_provider_count(&observations[..2]), 1);
+ assert_eq!(suspicious_provider_count(&observations), 2);
+ }
+}
diff --git a/probe/src/main.rs b/probe/src/main.rs
index 1eab789..9c9f9c9 100644
--- a/probe/src/main.rs
+++ b/probe/src/main.rs
@@ -1,24 +1,31 @@
+mod dns;
+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, TcpTracerouteOutcome};
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 +46,15 @@ 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,
+
+ #[arg(long, env = "TRACEROUTE_RETRIES", default_value_t = 3)]
+ traceroute_retries: u8,
+
+ #[arg(long, env = "TRACEROUTE_CONTROL_HOSTS", default_value_t = 3)]
+ traceroute_control_hosts: usize,
}
#[tokio::main]
@@ -48,6 +64,15 @@ 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");
+ }
+ if args.traceroute_retries == 0 {
+ bail!("traceroute_retries must be greater than zero");
+ }
+ if args.traceroute_control_hosts == 0 {
+ bail!("traceroute_control_hosts 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 +180,43 @@ 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")?;
+ if value.dns_samples_per_protocol == 0 {
+ bail!("dns_samples_per_protocol must be greater than zero");
+ }
+ if !(1..=4).contains(&value.dns_spoofing_provider_threshold) {
+ bail!("dns_spoofing_provider_threshold must be between 1 and 4");
+ }
+ let mut control_hosts_v4 = HashSet::new();
+ let mut control_hosts_v6 = HashSet::new();
+ if value.traceroute_enabled {
+ 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 +260,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 +278,91 @@ 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 traceroute_enabled = config
+ .as_ref()
+ .is_some_and(|config| config.config.traceroute_enabled);
+ let control_targets = config.as_ref().map_or_else(Vec::new, |config| {
+ if !traceroute_enabled {
+ return Vec::new();
}
- }));
- 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 mut rng = rand::thread_rng();
+ match task.ip {
+ IpAddr::V4(_) => config
+ .control_hosts_v4
+ .choose_multiple(&mut rng, args.traceroute_control_hosts)
+ .copied()
+ .map(IpAddr::V4)
+ .collect(),
+ IpAddr::V6(_) => config
+ .control_hosts_v6
+ .choose_multiple(&mut rng, args.traceroute_control_hosts)
+ .copied()
+ .map(IpAddr::V6)
+ .collect(),
+ }
+ });
+ let sni_check = sni::check_sni(
+ config.as_ref().map(|config| &config.config),
+ task.domain,
+ remaining,
+ job_id,
+ task.timeout_ms,
+ );
+ let target_traceroute = async {
+ if traceroute_enabled {
+ traceroute::tcp_traceroute(task.ip, args.traceroute_max_hops, args.traceroute_retries)
+ .await
+ } else {
+ None
}
};
+ let dns_samples_per_protocol = config
+ .as_ref()
+ .map_or_else(reports::probe::default_dns_samples_per_protocol, |config| {
+ config.config.dns_samples_per_protocol
+ });
+ let dns_spoofing_provider_threshold = config.as_ref().map_or_else(
+ reports::probe::default_dns_spoofing_provider_threshold,
+ |config| config.config.dns_spoofing_provider_threshold,
+ );
+ let dns_check = dns::check_dns(
+ task.domain,
+ remaining,
+ dns_samples_per_protocol,
+ dns_spoofing_provider_threshold,
+ );
+ let control_traceroute = async {
+ join_all(control_targets.into_iter().map(|target| {
+ traceroute::tcp_traceroute(target, args.traceroute_max_hops, args.traceroute_retries)
+ }))
+ .await
+ .into_iter()
+ .flatten()
+ .min_by_key(|trace| match trace.result {
+ TcpTracerouteOutcome::Rst { hop }
+ | TcpTracerouteOutcome::Connected { hop }
+ | TcpTracerouteOutcome::IcmpTimeExceeded { hop } => hop,
+ TcpTracerouteOutcome::Timeout => u8::MAX,
+ })
+ };
+ let (responses, target_traceroute, control_traceroute, dns) =
+ tokio::join!(sni_check, target_traceroute, control_traceroute, dns_check);
+ 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,
+ dns,
+ })?,
)
.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..d84174e
--- /dev/null
+++ b/probe/src/traceroute.rs
@@ -0,0 +1,306 @@
+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,
+ retries: u8,
+) -> Option {
+ tokio::task::spawn_blocking(move || trace_blocking(target, max_hops, retries))
+ .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, retries: 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 destination = SockAddr::from(SocketAddr::new(target, HTTPS_PORT));
+ let mut tcp_attempts = Vec::with_capacity(retries as usize);
+ let mut last_error = None;
+ for _ in 0..retries {
+ 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))?;
+ 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) {
+ last_error = Some(error);
+ continue;
+ }
+ }
+ let source_port = tcp
+ .local_addr()?
+ .as_socket()
+ .map(|addr| addr.port())
+ .unwrap_or(0);
+ tcp_attempts.push((tcp, source_port));
+ }
+ if tcp_attempts.is_empty() {
+ return Err(last_error.unwrap_or_else(|| io::Error::other("no traceroute attempts")));
+ }
+ match wait_for_hop_response(&receiver, &tcp_attempts, target)? {
+ 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_attempts: &[(Socket, u16)],
+ target: IpAddr,
+) -> io::Result {
+ const ICMP_KEY: usize = 1;
+
+ let poller = Poller::new()?;
+ // SAFETY: all sockets remain alive until polling and cleanup finish.
+ unsafe {
+ poller.add(receiver, Event::readable(ICMP_KEY))?;
+ for (index, (tcp, _)) in tcp_attempts.iter().enumerate() {
+ if let Err(error) = poller.add(tcp, Event::writable(index + 2)) {
+ let _ = poller.delete(receiver);
+ for (added, _) in tcp_attempts.iter().take(index) {
+ let _ = poller.delete(added);
+ }
+ return Err(error);
+ }
+ }
+ }
+
+ let result = wait_on_poller(&poller, receiver, tcp_attempts, target);
+ for (tcp, _) in tcp_attempts {
+ let _ = poller.delete(tcp);
+ }
+ let _ = poller.delete(receiver);
+ result
+}
+
+fn wait_on_poller(
+ poller: &Poller,
+ receiver: &Socket,
+ tcp_attempts: &[(Socket, u16)],
+ target: IpAddr,
+) -> io::Result {
+ const ICMP_KEY: usize = 1;
+
+ let deadline = Instant::now() + HOP_TIMEOUT;
+ let mut watching_tcp = vec![true; tcp_attempts.len()];
+ 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),
+ }
+
+ for event in events.iter().filter(|event| event.key >= 2) {
+ let index = event.key - 2;
+ if !watching_tcp.get(index).copied().unwrap_or(false) {
+ continue;
+ }
+ match tcp_attempts[index].0.take_error()? {
+ Some(error) if error.kind() == io::ErrorKind::ConnectionRefused => {
+ return Ok(HopResponse::Rst);
+ }
+ None => return Ok(HopResponse::Connected),
+ Some(_) => watching_tcp[index] = false,
+ }
+ }
+
+ if events.iter().any(|event| event.key == ICMP_KEY) {
+ let mut raw = receiver;
+ match raw.read(&mut buffer) {
+ Ok(bytes)
+ if tcp_attempts.iter().any(|(_, source_port)| {
+ 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/querying/src/target.rs b/querying/src/target.rs
index 933d5e8..a869d71 100644
--- a/querying/src/target.rs
+++ b/querying/src/target.rs
@@ -57,6 +57,36 @@ impl From<&str> for Target {
}
impl Target {
+ pub fn is_bogon(ip: IpAddr) -> bool {
+ match ip {
+ IpAddr::V4(ip) => {
+ let [a, b, c, d] = ip.octets();
+ a == 0
+ || a == 10
+ || a == 127
+ || (a == 100 && (64..=127).contains(&b))
+ || (a == 169 && b == 254)
+ || (a == 172 && (16..=31).contains(&b))
+ || (a == 192 && b == 0 && c == 0 && d != 9 && d != 10)
+ || (a == 192 && b == 0 && c == 2)
+ || (a == 192 && b == 88 && c == 99)
+ || (a == 192 && b == 168)
+ || (a == 198 && (b == 18 || b == 19))
+ || (a == 198 && b == 51 && c == 100)
+ || (a == 203 && b == 0 && c == 113)
+ || a >= 224
+ }
+ IpAddr::V6(ip) => {
+ let segments = ip.segments();
+ let first = segments[0];
+ first & 0xe000 != 0x2000
+ || (first == 0x2001 && segments[1] < 0x0200)
+ || (first == 0x2001 && segments[1] == 0x0db8)
+ || (first == 0x3fff && segments[1] < 0x1000)
+ }
+ }
+ }
+
pub fn readable_type(&self) -> &'static str {
match self {
Target::Domain(_) => "Домен",
@@ -155,3 +185,34 @@ fn format_large_number(n: u128) -> String {
format!("{:.1}Z", n as f64 / 1_000_000_000_000_000_000_000.0)
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn identifies_bogon_addresses() {
+ for address in [
+ "10.0.0.1",
+ "100.64.0.1",
+ "127.0.0.1",
+ "169.254.1.1",
+ "192.0.2.1",
+ "198.18.0.1",
+ "224.0.0.1",
+ "::1",
+ "2001:db8::1",
+ "fc00::1",
+ "fe80::1",
+ ] {
+ assert!(Target::is_bogon(address.parse().unwrap()), "{address}");
+ }
+ }
+
+ #[test]
+ fn accepts_public_addresses() {
+ for address in ["1.1.1.1", "8.8.8.8", "2001:4860:4860::8888"] {
+ assert!(!Target::is_bogon(address.parse().unwrap()), "{address}");
+ }
+ }
+}
diff --git a/reports/src/probe.rs b/reports/src/probe.rs
index abbb232..28e977b 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,22 @@ pub struct ProbeConfig {
pub task_timeout_ms: u64,
pub published_at: String,
pub hosts: Vec,
+ #[serde(default)]
+ pub traceroute_enabled: bool,
+ #[serde(default)]
+ pub control_hosts: Vec,
+ #[serde(default = "default_dns_samples_per_protocol")]
+ pub dns_samples_per_protocol: u8,
+ #[serde(default = "default_dns_spoofing_provider_threshold")]
+ pub dns_spoofing_provider_threshold: u8,
+}
+
+pub const fn default_dns_samples_per_protocol() -> u8 {
+ 3
+}
+
+pub const fn default_dns_spoofing_provider_threshold() -> u8 {
+ 2
}
#[derive(Clone, Serialize, Deserialize)]
@@ -35,7 +52,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 +63,79 @@ pub struct ProbeResultEvent {
pub job_id: String,
pub probe_id: String,
pub host_results: Vec,
+ pub target_traceroute: Option,
+ pub control_traceroute: Option,
+ pub dns: Option,
+}
+
+#[derive(Clone, Serialize, Deserialize)]
+pub struct ProbeResult {
+ pub responses: Option>,
+ pub target_traceroute: Option,
+ pub control_traceroute: Option,
+ #[serde(default)]
+ pub dns: Option,
+}
+
+#[derive(Clone, Serialize, Deserialize)]
+pub struct DnsProbeResult {
+ pub spoofing_detected: bool,
+ #[serde(default)]
+ pub suspicious_provider_count: u8,
+ #[serde(default)]
+ pub verdict_threshold: u8,
+ #[serde(default)]
+ pub samples_per_protocol: u8,
+ pub observations: Vec,
+}
+
+#[derive(Clone, Serialize, Deserialize)]
+pub struct DnsObservation {
+ pub provider: String,
+ pub protocol: DnsProtocol,
+ pub outcome: DnsOutcome,
+ #[serde(default)]
+ pub suspected_spoofing: bool,
+ #[serde(default)]
+ pub metadata: DnsResponseMetadata,
+}
+
+#[derive(Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct DnsResponseMetadata {
+ pub response_codes: Vec,
+ pub ipv4_count: u16,
+ pub ipv6_count: u16,
+}
+
+#[derive(Clone, Copy, Serialize, Deserialize)]
+pub enum DnsProtocol {
+ Udp,
+ Tcp,
+ Doh,
+ Dot,
+}
+
+#[derive(Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "type")]
+pub enum DnsOutcome {
+ Answer { addresses: Vec },
+ NoRecords,
+ Error { message: String },
+}
+
+#[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..a273eca 100644
--- a/website/Cargo.toml
+++ b/website/Cargo.toml
@@ -1,6 +1,6 @@
[package]
name = "website"
-version = "1.1.0"
+version = "1.2.0"
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..6233842 100644
--- a/website/probe-hosts.toml
+++ b/website/probe-hosts.toml
@@ -1,5 +1,8 @@
timeout_sec = 3
min_data = 65536
+dns_samples_per_protocol = 3
+dns_spoofing_provider_threshold = 2
+control_hosts = ["kinopoisk.ru", "wildberries.ru", "yandex.ru", "mail.ru"]
[[hosts]]
id = "hil-hetzner" # hil-speed.hetzner.com
diff --git a/website/src/api.rs b/website/src/api.rs
index 2a8126f..4ae66ef 100644
--- a/website/src/api.rs
+++ b/website/src/api.rs
@@ -9,5 +9,5 @@ mod status;
pub use check_endpoint::check;
pub use feedback_endpoint::feedback;
pub use probe::probe_query;
-pub use rate_limit::build_rate_limiter;
+pub use rate_limit::{build_probe_rate_limiter, build_rate_limiter};
pub use status::{get_system_status, healthcheck};
diff --git a/website/src/api/check.rs b/website/src/api/check.rs
index 105ac3d..505cafe 100644
--- a/website/src/api/check.rs
+++ b/website/src/api/check.rs
@@ -45,7 +45,7 @@ pub async fn check(
pool: &State,
limiter: &State>,
) -> Result, Status> {
- if limiter.check_key(&addr.ip).is_err() {
+ if !limiter.check(&addr.ip) {
return Err(Status::TooManyRequests);
}
diff --git a/website/src/api/probe.rs b/website/src/api/probe.rs
index cf594fc..923aba7 100644
--- a/website/src/api/probe.rs
+++ b/website/src/api/probe.rs
@@ -1,9 +1,10 @@
-use super::rate_limit::ApiRateLimiter;
+use super::rate_limit::ProbeRateLimiter;
use crate::mqtt::{MqttPublisher, PublishError};
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)]
@@ -30,30 +32,47 @@ pub async fn probe_query(
addr: &ClientRealAddr,
pool: &State,
mqtt: &State,
- limiter: &State>,
+ limiter: &State>,
) -> Result {
- if limiter.check_key(&addr.ip).is_err() {
+ if !limiter.check(&addr.ip) {
return Err(Status::TooManyRequests);
}
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 probe_config = mqtt.probe_config();
+ if domain.is_none() && !probe_config.traceroute_enabled {
+ return Err(Status::BadRequest);
+ }
+ let ip = resolved_ips
+ .first()
+ .and_then(|ip| ip.parse::().ok())
+ .ok_or(Status::BadRequest)?;
+ if Target::is_bogon(ip) {
+ return Err(Status::Forbidden);
+ }
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}");
@@ -62,7 +81,6 @@ pub async fn probe_query(
let timeout = mqtt.task_timeout();
let online_probes = mqtt.online_probe_count().await;
- let probe_config = mqtt.probe_config();
let pool = pool.inner().clone();
let query_id = id;
let id = id.to_string();
@@ -73,7 +91,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 +106,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 +119,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 +153,22 @@ 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(),
+ raw.dns.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 +196,16 @@ pub fn build_probe_response(
"asn": asn,
"verdict": verdict,
"host_results": host_results,
+ "target_hop": target_hop,
+ "dns": raw.dns,
})
}
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 +216,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 +226,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 +289,33 @@ 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>,
+ dns: Option<&reports::probe::DnsProbeResult>,
+) -> &'static str {
+ let dns_spoofing = dns.is_some_and(|result| result.spoofing_detected);
+ 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 if dns_spoofing { "dns_spoofing" } else { "ok" };
+ }
+
let matched = results
.iter()
.filter_map(|result| {
@@ -223,7 +328,11 @@ fn build_probe_verdict(results: &[HostProbeResult], config: &ProbeConfig) -> &'s
.collect::>();
if matched.is_empty() {
- return "uncertain";
+ return if dns_spoofing {
+ "dns_spoofing"
+ } else {
+ "uncertain"
+ };
}
if is_strict_majority(
@@ -236,6 +345,10 @@ fn build_probe_verdict(results: &[HostProbeResult], config: &ProbeConfig) -> &'s
return "sni_block";
}
+ if dns_spoofing {
+ return "dns_spoofing";
+ }
+
if is_strict_majority(
matched.len(),
matched
@@ -306,3 +419,87 @@ 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), None),
+ "tspu_block"
+ );
+
+ let target = icmp_trace(5);
+ assert_eq!(
+ build_probe_verdict(&[], &empty_config(), Some(&target), Some(&control), None),
+ "ok"
+ );
+ }
+
+ #[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), None),
+ "ok"
+ );
+ }
+
+ #[test]
+ fn sni_block_takes_priority_over_dns_spoofing() {
+ let mut config = empty_config();
+ config.hosts.push(Host {
+ id: "test".to_string(),
+ host: "192.0.2.1".to_string(),
+ host_type: HostType::Blacklist,
+ file_path: String::new(),
+ timeout_sec: 1,
+ min_data: 1,
+ });
+ let results = vec![HostProbeResult {
+ host_id: "test".to_string(),
+ probe_evidence: ProbeEvidence::ClientHello,
+ }];
+ let dns = reports::probe::DnsProbeResult {
+ spoofing_detected: true,
+ suspicious_provider_count: 2,
+ verdict_threshold: 2,
+ samples_per_protocol: 3,
+ observations: vec![],
+ };
+
+ assert_eq!(
+ build_probe_verdict(&results, &config, None, None, Some(&dns)),
+ "sni_block"
+ );
+ }
+
+ fn empty_config() -> ProbeConfig {
+ ProbeConfig {
+ version: String::new(),
+ task_timeout_ms: 0,
+ published_at: String::new(),
+ hosts: vec![],
+ traceroute_enabled: true,
+ control_hosts: vec![],
+ dns_samples_per_protocol: reports::probe::default_dns_samples_per_protocol(),
+ dns_spoofing_provider_threshold:
+ reports::probe::default_dns_spoofing_provider_threshold(),
+ }
+ }
+}
diff --git a/website/src/api/rate_limit.rs b/website/src/api/rate_limit.rs
index 42a9be3..322a0d6 100644
--- a/website/src/api/rate_limit.rs
+++ b/website/src/api/rate_limit.rs
@@ -4,9 +4,33 @@ use governor::{Quota, RateLimiter};
use std::net::IpAddr;
use std::num::NonZeroU32;
-pub type ApiRateLimiter = RateLimiter, DefaultClock>;
+type KeyedRateLimiter = RateLimiter, DefaultClock>;
+
+pub struct ApiRateLimiter(KeyedRateLimiter);
+
+pub struct ProbeRateLimiter(KeyedRateLimiter);
+
+impl ApiRateLimiter {
+ pub fn check(&self, ip: &IpAddr) -> bool {
+ self.0.check_key(ip).is_ok()
+ }
+}
+
+impl ProbeRateLimiter {
+ pub fn check(&self, ip: &IpAddr) -> bool {
+ self.0.check_key(ip).is_ok()
+ }
+}
pub fn build_rate_limiter(per_minute: u32) -> ApiRateLimiter {
+ ApiRateLimiter(build_limiter(per_minute))
+}
+
+pub fn build_probe_rate_limiter(per_minute: u32) -> ProbeRateLimiter {
+ ProbeRateLimiter(build_limiter(per_minute))
+}
+
+fn build_limiter(per_minute: u32) -> KeyedRateLimiter {
RateLimiter::keyed(Quota::per_minute(
NonZeroU32::new(per_minute).expect("rate limit must be > 0"),
))
diff --git a/website/src/main.rs b/website/src/main.rs
index d3b6175..796c6e8 100644
--- a/website/src/main.rs
+++ b/website/src/main.rs
@@ -58,7 +58,12 @@ async fn rocket() -> _ {
.unwrap_or("30".to_string())
.parse()
.unwrap_or(30);
+ let probe_rate_limit_rpm: u32 = std::env::var("PROBE_RATE_LIMIT_RPM")
+ .unwrap_or("5".to_string())
+ .parse()
+ .unwrap_or(5);
let api_limiter = std::sync::Arc::new(api::build_rate_limiter(rate_limit_rpm));
+ let probe_limiter = std::sync::Arc::new(api::build_probe_rate_limiter(probe_rate_limit_rpm));
let mqtt_publisher = mqtt::MqttPublisher::start_from_env();
let pool = sqlx::postgres::PgPoolOptions::new()
@@ -84,6 +89,7 @@ async fn rocket() -> _ {
.manage(checker)
.manage(pool)
.manage(api_limiter)
+ .manage(probe_limiter)
.manage(mqtt_publisher)
.attach(AdHoc::try_on_ignite("SQLx Migrations", run_migrations))
.mount(
diff --git a/website/src/mqtt.rs b/website/src/mqtt.rs
index 8a9af34..b17c53b 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,12 @@ pub type ProbeResultReceiver = rocket::tokio::sync::broadcast::Receiver,
hosts: Vec,
}
@@ -85,6 +92,11 @@ impl MqttPublisher {
task_timeout_ms,
published_at: Utc::now().to_rfc3339(),
hosts: Vec::new(),
+ traceroute_enabled: false,
+ control_hosts: Vec::new(),
+ dns_samples_per_protocol: reports::probe::default_dns_samples_per_protocol(),
+ dns_spoofing_provider_threshold:
+ reports::probe::default_dns_spoofing_provider_threshold(),
}
}));
let admin_token = match std::env::var("MQTT_ADMIN_TOKEN") {
@@ -216,14 +228,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 +264,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 +274,27 @@ 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,
+ traceroute_enabled: std::env::var("PROBE_TRACEROUTE_ENABLED")
+ .ok()
+ .is_some_and(|value| value == "1" || value.eq_ignore_ascii_case("true")),
+ control_hosts: config.control_hosts,
+ dns_samples_per_protocol: config.dns_samples_per_protocol,
+ dns_spoofing_provider_threshold: config.dns_spoofing_provider_threshold,
})
}
-fn parse_probe_hosts(contents: &str) -> Result, PublishError> {
+struct ParsedProbeConfig {
+ hosts: Vec,
+ control_hosts: Vec,
+ dns_samples_per_protocol: u8,
+ dns_spoofing_provider_threshold: u8,
+}
+
+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 +305,14 @@ 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,
+ dns_samples_per_protocol: config.dns_samples_per_protocol,
+ dns_spoofing_provider_threshold: config.dns_spoofing_provider_threshold,
+ })
}
async fn dispatch_probe_status(
@@ -320,7 +354,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 +367,10 @@ 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,
+ dns: result.dns,
});
}
}
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())
diff --git a/website/src/whitelist.rs b/website/src/whitelist.rs
index 8bb4ec5..1f9a211 100644
--- a/website/src/whitelist.rs
+++ b/website/src/whitelist.rs
@@ -8,7 +8,7 @@ use rocket_cache_response::CacheResponse;
use sqlx::postgres::PgPool;
use std::io;
-enum ExportType {
+pub(crate) enum ExportType {
Full,
Domains,
}
|