From 3f74894786beb1a1d4954684b7e5426d58cf380f Mon Sep 17 00:00:00 2001 From: LowderPlay Date: Tue, 18 Aug 2026 00:54:52 +0500 Subject: [PATCH] feat: better probing (#81) * feat: ip probe traceroute * feat: block bogon and rate limit * fix: block badge * fix: control sampling * docs: update probing info * feat: disable traceroute * feat: dns spoofing * docs: dns spoofing * chore: bump version * fix: default rate limit --- .github/workflows/probe-build.yml | 21 +- Cargo.lock | 39 ++- biome.json | 2 +- docker-compose.yaml | 2 + frontend/src/lib/api/probe.ts | 33 +- .../src/lib/components/ResultPanel.svelte | 22 +- .../src/lib/components/kb/KbArticle.svelte | 1 + .../lib/components/result/ProbeTable.svelte | 187 +++++++++- frontend/src/routes/check/+page.svelte | 4 +- frontend/src/routes/kb/probing/+page.svelte | 108 +++++- frontend/vite.config.ts | 2 +- probe/Cargo.toml | 8 +- probe/Dockerfile | 7 +- probe/README.md | 10 +- probe/debian/cheburprobe.service | 2 + probe/src/dns.rs | 330 ++++++++++++++++++ probe/src/main.rs | 320 ++++++++--------- probe/src/sni.rs | 184 ++++++++++ probe/src/traceroute.rs | 306 ++++++++++++++++ querying/src/target.rs | 61 ++++ reports/src/probe.rs | 93 ++++- website/Cargo.toml | 2 +- .../20260817000000_probe_traceroutes.sql | 5 + website/probe-hosts.toml | 3 + website/src/api.rs | 2 +- website/src/api/check.rs | 2 +- website/src/api/probe.rs | 239 +++++++++++-- website/src/api/rate_limit.rs | 26 +- website/src/main.rs | 6 + website/src/mqtt.rs | 57 ++- website/src/mqtt_auth.rs | 11 +- website/src/whitelist.rs | 2 +- 32 files changed, 1817 insertions(+), 280 deletions(-) create mode 100644 probe/src/dns.rs 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..fd96753 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,24 @@ dependencies = [ [[package]] name = "probe" -version = "0.1.0" +version = "0.2.0" dependencies = [ "anyhow", "clap", "env_logger", + "etherparse", "futures", + "hickory-resolver", "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 +4706,7 @@ dependencies = [ [[package]] name = "website" -version = "1.1.0" +version = "1.2.0" dependencies = [ "dotenvy", "env_logger", diff --git a/biome.json b/biome.json index f85db2b..eba3124 100644 --- a/biome.json +++ b/biome.json @@ -11,7 +11,7 @@ "enabled": true }, "formatter": { - "whitespaceSensitivity": "ignore", + "whitespaceSensitivity": "css", "enabled": true } }, diff --git a/docker-compose.yaml b/docker-compose.yaml index cbc8d56..1d3eb61 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -34,6 +34,8 @@ services: MQTT_ADMIN_TOKEN: "${MQTT_ADMIN_TOKEN}" MQTT_HOST: rmqtt MQTT_PORT: 11883 + PROBE_TRACEROUTE_ENABLED: "${PROBE_TRACEROUTE_ENABLED:-false}" + PROBE_RATE_LIMIT_RPM: "${PROBE_RATE_LIMIT_RPM:-5}" volumes: - database-cache:/var/cache/cheburcheck/databases expose: diff --git a/frontend/src/lib/api/probe.ts b/frontend/src/lib/api/probe.ts index fea24db..e70e573 100644 --- a/frontend/src/lib/api/probe.ts +++ b/frontend/src/lib/api/probe.ts @@ -11,14 +11,43 @@ export type ProbeHostResult = { }; }; +export type DnsObservation = { + provider: string; + protocol: "Udp" | "Tcp" | "Doh" | "Dot"; + suspected_spoofing: boolean; + metadata: { + response_codes: string[]; + ipv4_count: number; + ipv6_count: number; + }; + outcome: + | { type: "Answer"; addresses: string[] } + | { type: "NoRecords" } + | { type: "Error"; message: string }; +}; + export type ProbeResult = { job_id: string; probe_id: string; region?: string | null; provider?: string | null; asn?: string | null; - verdict: "uncertain" | "sni_block" | "whitelist" | "ok"; - host_results: ProbeHostResult[]; + verdict: + | "uncertain" + | "dns_spoofing" + | "sni_block" + | "tspu_block" + | "whitelist" + | "ok"; + host_results: ProbeHostResult[] | null; + target_hop: number | null; + dns: { + spoofing_detected: boolean; + suspicious_provider_count: number; + verdict_threshold: number; + samples_per_protocol: number; + observations: DnsObservation[]; + } | null; }; export type ProbeStatus = { diff --git a/frontend/src/lib/components/ResultPanel.svelte b/frontend/src/lib/components/ResultPanel.svelte index 0002e50..9406865 100644 --- a/frontend/src/lib/components/ResultPanel.svelte +++ b/frontend/src/lib/components/ResultPanel.svelte @@ -98,16 +98,18 @@ const providerCidrs = (provider: Provider) => {#if result.reverseLookup.length > 0} - {#each result.reverseLookup as ptr} - - - {ptr} - - - {/each} +
+ {#each result.reverseLookup as ptr} + + + {ptr} + + + {/each} +
{/if} diff --git a/frontend/src/lib/components/kb/KbArticle.svelte b/frontend/src/lib/components/kb/KbArticle.svelte index 16f625a..f4692aa 100644 --- a/frontend/src/lib/components/kb/KbArticle.svelte +++ b/frontend/src/lib/components/kb/KbArticle.svelte @@ -14,6 +14,7 @@ kb?.reset(); [&_code]:bg-neutral-900/70 [&_code]:px-1.5 [&_code]:py-0.5 [&_code]:rounded [&_code]:text-sm [&_code]:text-neutral-200 [&_h1]:mb-16 [&_h1]:text-5xl [&_h1]:font-black [&_h1]:tracking-tight [&_h1]:text-neutral-100 [&_h1]:uppercase [&_li]:my-3 [&_li]:text-neutral-300 + [&_ol]:list-decimal [&_ol]:pl-6 [&_ol]:my-6 [&_p]:my-6 [&_p]:text-lg [&_p]:leading-relaxed [&_p]:text-neutral-300 [&_ul]:list-disc [&_ul]:pl-6 [&_ul]:my-6 " diff --git a/frontend/src/lib/components/result/ProbeTable.svelte b/frontend/src/lib/components/result/ProbeTable.svelte index e9a300b..c32a84e 100644 --- a/frontend/src/lib/components/result/ProbeTable.svelte +++ b/frontend/src/lib/components/result/ProbeTable.svelte @@ -9,7 +9,7 @@ import { LoaderCircle, ShieldCheck, } from "@lucide/svelte"; -import type { ProbeResult, ProbeStatus } from "$lib/api/probe"; +import type { DnsObservation, ProbeResult, ProbeStatus } from "$lib/api/probe"; let { probes, @@ -27,6 +27,38 @@ function toggleRow(id: string) { expandedRows[id] = !expandedRows[id]; } +function dnsObservationStatus( + observation: DnsObservation, +): "ok" | "spoofed" | "error" { + if (observation.outcome.type === "Error") return "error"; + return observation.suspected_spoofing ? "spoofed" : "ok"; +} + +const dnsProtocolLabels: Record = { + Udp: "UDP", + Tcp: "TCP", + Doh: "DoH", + Dot: "DoT", +}; + +const dnsProtocols = ["Udp", "Tcp", "Doh", "Dot"] as const; + +function groupDnsObservations(observations: DnsObservation[]) { + const providers = new Map< + string, + Partial> + >(); + for (const observation of observations) { + const protocols = providers.get(observation.provider) ?? {}; + protocols[observation.protocol] = observation; + providers.set(observation.provider, protocols); + } + return [...providers].map(([provider, protocols]) => ({ + provider, + protocols, + })); +} + const verdictStyles = { ok: { icon: CircleCheck, @@ -49,9 +81,23 @@ 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", + }, + dns_spoofing: { + icon: CircleX, + text: "Подмена DNS", + class: "text-red-500", + bg: "bg-red-500/10", + border: "border-red-500/20", + }, whitelist: { icon: ShieldCheck, - text: "Белый список", + text: "Исключение для CDN", class: "text-amber-500", bg: "bg-amber-500/10", border: "border-amber-500/20", @@ -123,13 +169,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" > @@ -142,12 +187,23 @@ const verdictStyles = {
{probe.asn || ""}
-
- - {style.text} -
+ {#if probe.verdict === "whitelist"} + event.stopPropagation()} + class={`inline-flex items-center gap-1.5 px-2 py-1 rounded border ${style.bg} ${style.border} ${style.class} text-xs font-bold transition-colors hover:bg-amber-500/20 hover:border-amber-500/50 hover:text-amber-400`} + > + + {style.text} + + {:else} +
+ + {style.text} +
+ {/if} {#if isExpanded} @@ -160,6 +216,110 @@ const verdictStyles = { {#if isExpanded} + + {#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} + + {/each} + + + + {#each groupDnsObservations(probe.dns.observations) as provider} + + + {#each dnsProtocols as protocol} + {@const observation = provider.protocols[protocol]} + + {/each} + + {/each} + +
Провайдер + {dnsProtocolLabels[protocol]} +
+ {provider.provider} + + {#if observation} + {@const dnsStatus = dnsObservationStatus(observation)} +
+
+ {dnsStatus === "spoofed" ? "Подозрительно" : dnsStatus === "error" ? "Ошибка" : "Норма"} +
+
+ {observation.metadata.response_codes.join(", ") || "—"} + · + {observation.metadata.ipv4_count}/{observation.metadata.ipv6_count} +
+
+ {:else} + — + {/if} +
+
+

+ Сравниваются код ответа и количество уникальных + IPv4/IPv6. Выполнено по + {probe.dns.samples_per_protocol || 1} + запроса на протокол. Вердикт требует подтверждения от + {probe.dns.verdict_threshold || 2} + DNS-провайдеров. +

+
+ {/if} +
+

+ 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, }