Compare commits

..

9 commits

Author SHA1 Message Date
Lowder
d7e355c0c7 fix: remove positive hop limit
Some checks are pending
Build Reporter / validate (push) Waiting to run
Build Reporter / Linux amd64 (.deb) (push) Blocked by required conditions
Build Reporter / Linux arm64 (.deb via cross) (push) Blocked by required conditions
Build Reporter / Windows amd64 (push) Blocked by required conditions
Build Reporter / release (push) Blocked by required conditions
CodeQL Advanced / Analyze (rust) (push) Waiting to run
Build and Publish Docker Images / validate (push) Waiting to run
Build and Publish Docker Images / Build website image (push) Blocked by required conditions
Build and Publish Docker Images / Build frontend image (push) Blocked by required conditions
Build and Publish Docker Images / Deploy (push) Blocked by required conditions
Build and Publish Docker Images / release (push) Blocked by required conditions
Build Probe / validate (push) Waiting to run
Build Probe / Linux amd64 binary and .deb (push) Blocked by required conditions
Build Probe / Linux arm64 binary, .deb, and OpenWrt packages (push) Blocked by required conditions
Build Probe / Windows amd64 binary (push) Blocked by required conditions
Build Probe / Docker image amd64/arm64/armv7 (push) Blocked by required conditions
Build Probe / release (push) Blocked by required conditions
2026-10-19 17:51:08 +05:00
Lowder
df7e3f494f fix: silent drop detection and fluctuating cdn_blocks
Some checks failed
Build and Publish Docker Images / validate (push) Has been cancelled
Build Reporter / validate (push) Has been cancelled
Build Probe / validate (push) Has been cancelled
CodeQL Advanced / Analyze (rust) (push) Has been cancelled
Build Probe / Linux amd64 binary and .deb (push) Has been cancelled
Build Reporter / Linux amd64 (.deb) (push) Has been cancelled
Build and Publish Docker Images / Build website image (push) Has been cancelled
Build and Publish Docker Images / Build frontend image (push) Has been cancelled
Build and Publish Docker Images / Deploy (push) Has been cancelled
Build and Publish Docker Images / release (push) Has been cancelled
Build Reporter / Linux arm64 (.deb via cross) (push) Has been cancelled
Build Reporter / Windows amd64 (push) Has been cancelled
Build Reporter / release (push) Has been cancelled
Build Probe / Linux arm64 binary, .deb, and OpenWrt packages (push) Has been cancelled
Build Probe / Windows amd64 binary (push) Has been cancelled
Build Probe / Docker image amd64/arm64/armv7 (push) Has been cancelled
Build Probe / release (push) Has been cancelled
2026-10-18 04:21:35 +05:00
Lowder
04c11a9d4a feat: rst dpi detection 2026-10-18 02:21:19 +05:00
LowderPlay
3a45460ff0
feat: clickhouse storage (#106)
Some checks failed
Build Reporter / validate (push) Has been cancelled
Build and Publish Docker Images / validate (push) Has been cancelled
CodeQL Advanced / Analyze (rust) (push) Has been cancelled
Build Probe / validate (push) Has been cancelled
Build Reporter / Linux amd64 (.deb) (push) Has been cancelled
Build Reporter / Linux arm64 (.deb via cross) (push) Has been cancelled
Build Reporter / Windows amd64 (push) Has been cancelled
Build Reporter / release (push) Has been cancelled
Build and Publish Docker Images / Build website image (push) Has been cancelled
Build and Publish Docker Images / Build frontend image (push) Has been cancelled
Build and Publish Docker Images / Deploy (push) Has been cancelled
Build and Publish Docker Images / release (push) Has been cancelled
Build Probe / Linux amd64 binary and .deb (push) Has been cancelled
Build Probe / Linux arm64 binary, .deb, and OpenWrt packages (push) Has been cancelled
Build Probe / Windows amd64 binary (push) Has been cancelled
Build Probe / Docker image amd64/arm64/armv7 (push) Has been cancelled
Build Probe / release (push) Has been cancelled
* feat: clickhouse storage

* feat: use deadpool for clickhouse

* chore: bump version
2026-10-13 03:48:33 +05:00
Lowder
2c42cf434f
docs: fix curl typo
Some checks failed
Build Probe / validate (push) Has been cancelled
Build Reporter / validate (push) Has been cancelled
CodeQL Advanced / Analyze (rust) (push) Has been cancelled
Build and Publish Docker Images / validate (push) Has been cancelled
Build Reporter / Linux amd64 (.deb) (push) Has been cancelled
Build Reporter / Linux arm64 (.deb via cross) (push) Has been cancelled
Build Reporter / release (push) Has been cancelled
Build and Publish Docker Images / Build website image (push) Has been cancelled
Build and Publish Docker Images / Build frontend image (push) Has been cancelled
Build and Publish Docker Images / Deploy (push) Has been cancelled
Build and Publish Docker Images / release (push) Has been cancelled
Build Reporter / Windows amd64 (push) Has been cancelled
Build Probe / Linux amd64 binary and .deb (push) Has been cancelled
Build Probe / Linux arm64 binary, .deb, and OpenWrt packages (push) Has been cancelled
Build Probe / Windows amd64 binary (push) Has been cancelled
Build Probe / Docker image amd64/arm64/armv7 (push) Has been cancelled
Build Probe / release (push) Has been cancelled
2026-10-08 21:15:18 +05:00
Lowder
9db493bb0a
feat: traceroutes packets capture 2026-10-08 21:08:25 +05:00
Lowder
c023d3dfb5
chore: bump version
Some checks are pending
Build Reporter / validate (push) Waiting to run
Build Reporter / Linux amd64 (.deb) (push) Blocked by required conditions
Build Reporter / Linux arm64 (.deb via cross) (push) Blocked by required conditions
Build Reporter / Windows amd64 (push) Blocked by required conditions
Build Reporter / release (push) Blocked by required conditions
CodeQL Advanced / Analyze (rust) (push) Waiting to run
Build and Publish Docker Images / validate (push) Waiting to run
Build and Publish Docker Images / Build website image (push) Blocked by required conditions
Build and Publish Docker Images / Build frontend image (push) Blocked by required conditions
Build and Publish Docker Images / Deploy (push) Blocked by required conditions
Build and Publish Docker Images / release (push) Blocked by required conditions
Build Probe / validate (push) Waiting to run
Build Probe / Linux amd64 binary and .deb (push) Blocked by required conditions
Build Probe / Linux arm64 binary, .deb, and OpenWrt packages (push) Blocked by required conditions
Build Probe / Windows amd64 binary (push) Blocked by required conditions
Build Probe / Docker image amd64/arm64/armv7 (push) Blocked by required conditions
Build Probe / release (push) Blocked by required conditions
2026-10-08 00:59:02 +05:00
Lowder
12c674ea96
fix: Курская область 2026-10-08 00:59:02 +05:00
Ilya Pavlushin
71402c4ae7
feat: update Docker build for armv7 and refactor architecture detection (#103)
Some checks failed
Build and Publish Docker Images / Build frontend image (push) Has been cancelled
Build and Publish Docker Images / Deploy (push) Has been cancelled
Build and Publish Docker Images / release (push) Has been cancelled
Build Probe / Linux amd64 binary and .deb (push) Has been cancelled
Build Probe / Linux arm64 binary, .deb, and OpenWrt packages (push) Has been cancelled
Build Probe / Windows amd64 binary (push) Has been cancelled
Build Probe / Docker image amd64/arm64/armv7 (push) Has been cancelled
Build Probe / release (push) Has been cancelled
Build Reporter / validate (push) Has been cancelled
CodeQL Advanced / Analyze (rust) (push) Has been cancelled
Build and Publish Docker Images / validate (push) Has been cancelled
Build Probe / validate (push) Has been cancelled
Build Reporter / Linux amd64 (.deb) (push) Has been cancelled
Build Reporter / Linux arm64 (.deb via cross) (push) Has been cancelled
Build Reporter / Windows amd64 (push) Has been cancelled
Build Reporter / release (push) Has been cancelled
Build and Publish Docker Images / Build website image (push) Has been cancelled
2026-10-06 20:10:14 +05:00
37 changed files with 3382 additions and 255 deletions

View file

@ -169,7 +169,7 @@ jobs:
docker:
needs: validate
name: Docker image amd64/arm64
name: Docker image amd64/arm64/armv7
runs-on: ubuntu-latest
permissions:
contents: read
@ -242,7 +242,7 @@ jobs:
with:
context: .
file: ./probe/Dockerfile
platforms: linux/amd64,linux/arm64
platforms: linux/amd64,linux/arm64,linux/arm/v7
push: ${{ github.event_name != 'pull_request' }}
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}

198
Cargo.lock generated
View file

@ -329,7 +329,17 @@ checksum = "93698b29de5e97ad0ae26447b344c482a7284c737d9ddc5f9e52b74a336671bb"
dependencies = [
"chrono",
"chrono-tz-build",
"phf",
"phf 0.11.3",
]
[[package]]
name = "chrono-tz"
version = "0.10.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a6139a8597ed92cf816dfb33f5dd6cf0bb93a6adc938f11039f371bc5bcd26c3"
dependencies = [
"chrono",
"phf 0.12.1",
]
[[package]]
@ -339,10 +349,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0c088aee841df9c3041febbb73934cfc39708749bf96dc827e3359cd39ef11b1"
dependencies = [
"parse-zoneinfo",
"phf",
"phf 0.11.3",
"phf_codegen",
]
[[package]]
name = "cityhash-rs"
version = "1.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "93a719913643003b84bd13022b4b7e703c09342cd03b679c4641c7d2e50dc34d"
[[package]]
name = "clap"
version = "4.6.6"
@ -399,6 +415,29 @@ dependencies = [
"memchr",
]
[[package]]
name = "compiler-tools"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cc80bd305611ea6275a49dfa4b7525f5dbf628274e4ad92382e89231c2074e91"
dependencies = [
"regex",
"serde",
]
[[package]]
name = "compiler-tools-derive"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0748adeee3cbac61b70d26e701515182d22d227baabe596d190f2b4154895842"
dependencies = [
"indexmap",
"proc-macro2",
"quote",
"regex",
"syn 2.0.119",
]
[[package]]
name = "concurrent-queue"
version = "2.5.0"
@ -601,6 +640,22 @@ version = "2.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06"
[[package]]
name = "deadpool"
version = "0.13.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3e98a7e119cd347f4201e1159b19831029e203e2d8b790547708e8157b4acf1e"
dependencies = [
"deadpool-runtime",
"tokio",
]
[[package]]
name = "deadpool-runtime"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2657f61fb1dd8bf37a8d51093cc7cee4e77125b22f7753f49b289f831bec2bae"
[[package]]
name = "defmt"
version = "1.1.1"
@ -1821,6 +1876,44 @@ dependencies = [
"wasm-bindgen",
]
[[package]]
name = "klickhouse"
version = "0.15.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "43df4e6356e65204d50730a3699b49a0ad345a947b725a0aaf7544f691c8e98a"
dependencies = [
"async-trait",
"chrono",
"chrono-tz 0.10.4",
"cityhash-rs",
"compiler-tools",
"compiler-tools-derive",
"futures-util",
"indexmap",
"klickhouse_derive",
"log",
"lz4",
"paste",
"refinery-core",
"rustc_version",
"thiserror 2.0.20",
"time",
"tokio",
"tokio-stream",
"uuid",
]
[[package]]
name = "klickhouse_derive"
version = "0.15.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2f1dd88efce7b8a0b0ac272f544db8b0b7d36846cb00485518689517e4b7093c"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.119",
]
[[package]]
name = "kqueue"
version = "1.2.1"
@ -1932,6 +2025,25 @@ version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154"
[[package]]
name = "lz4"
version = "1.28.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a20b523e860d03443e98350ceaac5e71c6ba89aea7d960769ec3ce37f4de5af4"
dependencies = [
"lz4-sys",
]
[[package]]
name = "lz4-sys"
version = "1.11.1+lz4-1.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6bd8c0d6c6ed0cd30b3652886bb8711dc4bb01d637a68105a3d5158039b418e6"
dependencies = [
"cc",
"libc",
]
[[package]]
name = "matchers"
version = "0.2.0"
@ -2245,6 +2357,12 @@ dependencies = [
"regex",
]
[[package]]
name = "paste"
version = "1.0.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a"
[[package]]
name = "pear"
version = "0.2.9"
@ -2341,7 +2459,16 @@ version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fd6780a80ae0c52cc120a26a1a42c1ae51b247a253e4e06113d23d2c2edd078"
dependencies = [
"phf_shared",
"phf_shared 0.11.3",
]
[[package]]
name = "phf"
version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "913273894cec178f401a31ec4b656318d95473527be05c0752cc41cdc32be8b7"
dependencies = [
"phf_shared 0.12.1",
]
[[package]]
@ -2351,7 +2478,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "aef8048c789fa5e851558d709946d6d79a8ff88c0440c587967f8e94bfb1216a"
dependencies = [
"phf_generator",
"phf_shared",
"phf_shared 0.11.3",
]
[[package]]
@ -2360,7 +2487,7 @@ version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3c80231409c20246a13fddb31776fb942c38553c51e871f8cbd687a4cfb5843d"
dependencies = [
"phf_shared",
"phf_shared 0.11.3",
"rand 0.8.8",
]
@ -2373,6 +2500,15 @@ dependencies = [
"siphasher",
]
[[package]]
name = "phf_shared"
version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06005508882fb681fd97892ecff4b7fd0fee13ef1aa569f8695dae7ab9099981"
dependencies = [
"siphasher",
]
[[package]]
name = "pin-project-lite"
version = "0.2.17"
@ -2488,7 +2624,7 @@ dependencies = [
[[package]]
name = "probe"
version = "0.7.0"
version = "0.8.2"
dependencies = [
"anyhow",
"clap",
@ -2783,6 +2919,46 @@ dependencies = [
"syn 3.0.5",
]
[[package]]
name = "refinery"
version = "0.9.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6e2a344cdb48871e27addeafbbaffab8828cc12cec2b9041119e9bea0c0f551a"
dependencies = [
"refinery-core",
"refinery-macros",
]
[[package]]
name = "refinery-core"
version = "0.9.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "24eeafd893124f29183dd6afa9137a27e7bef59250223b8b660005279c60aea4"
dependencies = [
"async-trait",
"cfg-if",
"log",
"regex",
"siphasher",
"thiserror 2.0.20",
"time",
"url",
"walkdir",
]
[[package]]
name = "refinery-macros"
version = "0.9.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a90cea6d11a9a4e8a85a884b6305461004101b28ca65dd35ec028e009a898e16"
dependencies = [
"proc-macro2",
"quote",
"refinery-core",
"regex",
"syn 2.0.119",
]
[[package]]
name = "regex"
version = "1.13.1"
@ -2837,6 +3013,7 @@ name = "reports"
version = "0.1.0"
dependencies = [
"serde",
"serde_json",
]
[[package]]
@ -3857,7 +4034,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e8004bca281f2d32df3bacd59bc67b312cb4c70cea46cbd79dbe8ac5ed206722"
dependencies = [
"chrono",
"chrono-tz",
"chrono-tz 0.9.0",
"globwalk",
"humansize",
"lazy_static",
@ -4509,15 +4686,19 @@ dependencies = [
[[package]]
name = "website"
version = "1.4.4"
version = "1.5.2"
dependencies = [
"anyhow",
"deadpool",
"dotenvy",
"env_logger",
"governor",
"klickhouse",
"log",
"percent-encoding",
"querying",
"rand 0.10.3",
"refinery",
"reports",
"reqwest",
"rocket",
@ -4529,6 +4710,7 @@ dependencies = [
"serde",
"sqlx",
"toml",
"url",
"uuid",
]

View file

@ -29,6 +29,10 @@ services:
ROCKET_ADDRESS: "::"
ROCKET_PORT: 8000
DATABASE_URL: "${DATABASE_URL}"
CLICKHOUSE_URL: "${CLICKHOUSE_URL}"
CLICKHOUSE_POOL_SIZE: "${CLICKHOUSE_POOL_SIZE:-4}"
CLICKHOUSE_ACQUIRE_TIMEOUT_SECONDS: "${CLICKHOUSE_ACQUIRE_TIMEOUT_SECONDS:-5}"
CLICKHOUSE_QUERY_TIMEOUT_SECONDS: "${CLICKHOUSE_QUERY_TIMEOUT_SECONDS:-10}"
DATABASE_CACHE_DIR: /var/cache/cheburcheck/databases
DATABASE_RETRY_INTERVAL_SECONDS: "${DATABASE_RETRY_INTERVAL_SECONDS:-300}"
MQTT_ADMIN_TOKEN: "${MQTT_ADMIN_TOKEN}"

View file

@ -27,6 +27,7 @@ export type DnsObservation = {
};
export type ProbeVerdict =
| "cdn_block"
| "uncertain"
| "dns_spoofing"
| "sni_block"
@ -62,12 +63,16 @@ export function displayProbeVerdicts(
isStaticCdn: boolean,
): DisplayProbeVerdict[] {
return probe.verdicts.map((verdict) =>
isStaticCdn &&
!probe.cdn_unblocked &&
probe.host_results?.length !== 0 &&
verdict === "ok"
? "cdn_block"
: verdict,
verdict === "cdn_block"
? isStaticCdn
? "cdn_block"
: "ok"
: isStaticCdn &&
!probe.cdn_unblocked &&
probe.host_results?.length !== 0 &&
verdict === "ok"
? "cdn_block"
: verdict,
);
}

View file

@ -39,7 +39,7 @@ const names: Record<string, string> = {
"RU-AMU": "Амурская область",
"RU-KB": "Кабардино-Балкарская Республика",
"RU-KDA": "Краснодарский край",
"RU-KRS": "Курская область",
"RU-KUR": "Курская область",
"RU-LEN": "Ленинградская область",
"RU-ME": "Республика Марий Эл",
"RU-MOW": "Москва",

View file

@ -63,7 +63,7 @@ export const russiaTilePositions: Partial<
"RU-TY": [13, 5, "ТЫВ"],
"RU-ZAB": [14, 5, "ЗАБ"],
"RU-PRI": [16, 5, "ПРИ"],
"RU-KRS": [3, 6, "КРС"],
"RU-KUR": [3, 6, "КУР"],
"RU-LIP": [4, 6, "ЛИП"],
"RU-TAM": [5, 6, "ТАМ"],
"RU-PNZ": [6, 6, "ПЕН"],

View file

@ -39,8 +39,58 @@ type Hop = {
address: string | null;
reverse_names: string[];
outcome: string;
tcp_diagnostics?: TcpDiagnostics;
};
type TcpPacket = {
observed_ms: number;
source: string;
destination: string;
ttl: number | null;
ip_id: number | null;
sequence: number;
acknowledgment: number;
window: number;
flags: string[];
timestamp: number | null;
timestamp_echo: number | null;
options_hex: string;
payload_bytes: number;
};
type TcpDiagnostics = {
packets: TcpPacket[];
capture_error: string | null;
connect_error: string | null;
send_error: string | null;
truncated: boolean;
client_hello_sent_ms: number | null;
junk_sent_ms: number | null;
payload_sent?: boolean;
peer_closed_before_payload?: boolean;
};
type CalibrationDebug = {
method: string;
dpi_hop: number | null;
fallback_reason?: string;
failure_reason?: string;
control_sni?: string;
last_clear_ttl?: number;
first_reset_ttl?: number;
control_witness_ttl?: number;
attempts?: {
ttl: number;
control: boolean;
outcome: string;
router: string | null;
error?: string;
tcp_diagnostics?: TcpDiagnostics;
}[];
};
type DpiHop = {
ttl: number;
src: string | null;
outcome: string;
tcp_diagnostics?: TcpDiagnostics;
};
type DpiHop = { ttl: number; src: string | null; outcome: string };
type CommandType =
| "resubscribe_tasks"
| "traceroute"
@ -55,6 +105,8 @@ type CommandResult =
dpi_hop_v6: number | null;
dpi_hops_v4: DpiHop[];
dpi_hops_v6: DpiHop[];
calibration_v4?: CalibrationDebug;
calibration_v6?: CalibrationDebug;
}
| {
type: "sni_traceroute";
@ -63,6 +115,7 @@ type CommandResult =
target: string;
dpi_hop: number | null;
hops: DpiHop[];
calibration?: CalibrationDebug;
}
| { type: "error"; message: string };
type Form = Pick<
@ -393,6 +446,147 @@ const outcomeLabel: Record<string, string> = {
};
</script>
{#snippet tcpDiagnostics(diagnostics: TcpDiagnostics | undefined)}
{#if diagnostics}
<details
class="mt-2 rounded-lg border border-neutral-700 p-3 text-xs"
open={diagnostics.packets.some((packet) => packet.flags.includes("RST") || packet.flags.includes("FIN"))}
>
<summary class="cursor-pointer text-neutral-300">
TCP: {diagnostics.packets.length} входящих пакетов{diagnostics.truncated ? " (список сокращён)" : ""}
</summary>
{#if diagnostics.capture_error}
<p class="mt-2 text-amber-300">
Ошибка захвата: {diagnostics.capture_error}
</p>
{/if}
{#if diagnostics.connect_error}
<p class="mt-2 text-amber-300">
Ошибка соединения: {diagnostics.connect_error}
</p>
{/if}
{#if diagnostics.send_error}
<p class="mt-2 text-amber-300">
Ошибка отправки: {diagnostics.send_error}
</p>
{/if}
{#if diagnostics.payload_sent === false}
<p class="mt-2 text-amber-300">
TTL-данные не записаны полностью{diagnostics.peer_closed_before_payload ? ": соединение закрыто до попытки отправки" : ""}.
</p>
{/if}
<p class="mt-2 text-neutral-400">
Время приблизительное, от начала захвата.
{#if diagnostics.client_hello_sent_ms !== null}
Попытка ClientHello: {diagnostics.client_hello_sent_ms} мс; попытка
записи данных:
{diagnostics.junk_sent_ms ?? "—"}
мс.
{/if}
Δ TTL сравнивает пакет с SYN-ACK того же соединения; это не расстояние
до DPI.
</p>
<div class="mt-3 overflow-x-auto">
<table class="w-full whitespace-nowrap text-left font-mono">
<thead class="text-neutral-500">
<tr>
<th class="pr-4">мс</th>
<th class="pr-4">Источник → цель</th>
<th class="pr-4">Флаги</th>
<th class="pr-4">TTL / Δ</th>
<th class="pr-4">SEQ / ACK</th>
<th class="pr-4">Окно / IP ID</th>
<th class="pr-4">TS / echo</th>
<th>Байты / опции</th>
</tr>
</thead>
<tbody>
{#each diagnostics.packets as packet}
{@const baseline = diagnostics.packets.find((p) => p.destination === packet.destination && p.flags.includes("SYN") && p.flags.includes("ACK"))}
<tr
class="border-t border-neutral-800"
class:text-amber-300={packet.flags.includes("RST") || packet.flags.includes("FIN")}
>
<td class="py-2 pr-4">{packet.observed_ms}</td>
<td class="pr-4">{packet.source} → {packet.destination}</td>
<td class="pr-4">{packet.flags.join(" ") || "—"}</td>
<td class="pr-4">
{packet.ttl ?? "—"}
/
{packet.ttl !== null && baseline?.ttl != null ? packet.ttl - baseline.ttl : "—"}
</td>
<td class="pr-4">{packet.sequence}/ {packet.acknowledgment}</td>
<td class="pr-4">{packet.window} / {packet.ip_id ?? "—"}</td>
<td class="pr-4">
{packet.timestamp ?? "—"}
/ {packet.timestamp_echo ?? "—"}
</td>
<td>{packet.payload_bytes} / {packet.options_hex || "—"}</td>
</tr>
{/each}
</tbody>
</table>
</div>
{#if diagnostics.packets.some((p) => p.ttl === null)}
<p class="mt-2 text-neutral-500">
Для части пакетов IP-заголовок недоступен; TTL не измерен.
</p>
{/if}
</details>
{/if}
{/snippet}
{#snippet calibrationDebug(calibration: CalibrationDebug | undefined)}
{#if calibration}
<details class="my-3 rounded-lg border border-neutral-700 p-3 text-xs">
<summary class="cursor-pointer text-neutral-300">
Калибровка: {calibration.method}; DPI hop: {calibration.dpi_hop ?? "—"}
</summary>
{#if calibration.fallback_reason}
<p class="mt-2">Причина fallback: {calibration.fallback_reason}</p>
{/if}
{#if calibration.failure_reason}
<p class="mt-2 text-amber-300">
Не подтверждено: {calibration.failure_reason}
</p>
{/if}
{#if calibration.control_sni}
<p class="mt-2">Контрольный SNI: {calibration.control_sni}</p>
{/if}
{#if calibration.first_reset_ttl !== undefined}
<p class="mt-2">
Последний подтверждённый TTL до фильтра:
{calibration.last_clear_ttl ?? "—"}; первый TTL с RST:
{calibration.first_reset_ttl}. TCP-проверки начинаются после верхней
границы.
</p>
{/if}
{#if calibration.control_witness_ttl !== undefined}
<p class="mt-2">
Контрольный ICMP за границей: TTL {calibration.control_witness_ttl}.
</p>
{/if}
<ol class="mt-3 space-y-2">
{#each calibration.attempts ?? [] as attempt}
<li class="rounded border border-neutral-800 p-2">
<span class="font-mono"
>TTL {attempt.ttl} ·
{attempt.control ? "контроль" : "проверяемый SNI"}
·
{attempt.outcome}
· {attempt.router ?? "—"}</span
>
{#if attempt.error}
<p class="text-amber-300">{attempt.error}</p>
{/if}
{@render tcpDiagnostics(attempt.tcp_diagnostics)}
</li>
{/each}
</ol>
</details>
{/if}
{/snippet}
{#snippet dpiHops(hops: DpiHop[])}
<ol class="space-y-1">
{#each hops as hop}
@ -404,6 +598,11 @@ const outcomeLabel: Record<string, string> = {
<span class="text-xs text-neutral-500"
>{outcomeLabel[hop.outcome] ?? hop.outcome}</span
>
{#if hop.tcp_diagnostics}
<div class="col-span-3 min-w-0">
{@render tcpDiagnostics(hop.tcp_diagnostics)}
</div>
{/if}
</li>
{/each}
</ol>
@ -782,9 +981,10 @@ PROBE_TOKEN={created.token}</pre>
></label
>
<p class="mb-4 text-sm text-neutral-400">
TCP-соединение к хосту на порту 443, ClientHello с указанным SNI,
затем пакеты с возрастающим TTL. Хост разрешается сканером;
используется первый IP адрес.
Для каждого TTL создаётся новое TCP-соединение к хосту на порту 443.
ClientHello с указанным SNI отправляется с обычным TTL; через 100 мс
отправляются случайные данные с проверяемым TTL. Хост разрешается
сканером; используется первый IP адрес.
</p>
{/if}
<div class="grid gap-3 sm:grid-cols-[1fr_2fr_100px_auto] sm:items-end">
@ -823,6 +1023,15 @@ PROBE_TOKEN={created.token}</pre>
{#if busy.endsWith(":command")}
<p class="mt-5 text-sm text-cyan-300">Ожидание ответа сканера…</p>
{/if}
{#if result && result.type !== "error"}
<button
type="button"
class="btn-secondary mt-4"
onclick={() => result && void copy(JSON.stringify(result, null, 2))}
>
Копировать JSON результата
</button>
{/if}
{#if result?.type === "error"}
<p class="mt-5 text-sm text-red-300">{result.message}</p>
{:else if result?.type === "resubscribe_tasks"}
@ -835,9 +1044,10 @@ PROBE_TOKEN={created.token}</pre>
{result.dpi_hop_v6 ?? "—"}.
</p>
<div class="mt-4 grid gap-4 sm:grid-cols-2">
{#each [{ label: "IPv4", hops: result.dpi_hops_v4 }, { label: "IPv6", hops: result.dpi_hops_v6 }] as trace}
{#each [{ label: "IPv4", hops: result.dpi_hops_v4, calibration: result.calibration_v4 }, { label: "IPv6", hops: result.dpi_hops_v6, calibration: result.calibration_v6 }] as trace}
<div>
<h3 class="mb-2 font-semibold">{trace.label}</h3>
{@render calibrationDebug(trace.calibration)}
{@render dpiHops(trace.hops)}
</div>
{/each}
@ -851,6 +1061,7 @@ PROBE_TOKEN={created.token}</pre>
SNI: {result.sni}; DPI hop: {result.dpi_hop ?? "—"}
</p>
{@render dpiHops(result.hops)}
{@render calibrationDebug(result.calibration)}
</div>
{:else if result?.type === "traceroute"}
<div class="mt-6">
@ -875,6 +1086,11 @@ PROBE_TOKEN={created.token}</pre>
<span class="text-xs text-neutral-500"
>{outcomeLabel[hop.outcome] ?? hop.outcome}</span
>
{#if hop.tcp_diagnostics}
<div class="col-span-3 min-w-0">
{@render tcpDiagnostics(hop.tcp_diagnostics)}
</div>
{/if}
</li>
{/each}
</ol>

View file

@ -95,7 +95,7 @@ const chartProps = {
</p>
<!-- biome-ignore format: code block -->
<KbCode>
$ curl -k https://cheburcheck.ru/100MB.bin -o/dev/null -r 0-65536 --resolve cheburcheck.ru:443:5.78.7.195 --max-time 5
$ curl -k https://cheburcheck.ru/100MB.bin -o/dev/null -r 0-65536 --connect-to ::5.78.7.195:443 --max-time 5
% Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
24 65537 24 16101 0 0 3220 0 0:00:20 0:00:04 0:00:16 0
@ -114,7 +114,7 @@ curl: (28) Operation timed out after 5000 milliseconds with 16101 out of 65537 b
</p>
<!-- biome-ignore format: code block -->
<KbCode>
$ curl -k https://ok.ru/100MB.bin -o/dev/null -r 0-65536 --resolve ok.ru:443:5.78.7.195
$ curl -k https://ok.ru/100MB.bin -o/dev/null -r 0-65536 --connect-to ::5.78.7.195:443
% Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
100 65537 100 65537 0 0 55226 0 0:00:01 0:00:01 --:--:-- 55258
@ -125,7 +125,7 @@ $ curl -k https://ok.ru/100MB.bin -o/dev/null -r 0-65536 --resolve ok.ru:443:5.7
</p>
<!-- biome-ignore format: code block -->
<KbCode>
$ curl -k https://ok.ru/100MB.bin -o/dev/null -r 0-65536 --resolve ok.ru:443:5.78.7.195
$ curl -k https://cheburcheck.ok.ru/100MB.bin -o/dev/null -r 0-65536 --connect-to ::5.78.7.195:443
% Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
100 65537 100 65537 0 0 55226 0 0:00:01 0:00:01 --:--:-- 55258

View file

@ -1,6 +1,6 @@
[package]
name = "probe"
version = "0.7.0"
version = "0.8.2"
edition = "2024"
license-file = "../LICENSE"
description = "Dynamic network probe daemon for Cheburcheck"

View file

@ -1,3 +1,4 @@
use crate::packet_capture::{TcpCapture, connect_with_port};
use etherparse::{
Icmpv4Type, Icmpv6Slice, Icmpv6Type, IpNumber, LaxNetSlice, LaxSlicedPacket, TransportSlice,
icmpv4, icmpv6,
@ -15,6 +16,8 @@ use std::time::{Duration, Instant};
const PROBE_BYTES: usize = 256;
const DPI_CLASSIFICATION_DELAY: Duration = Duration::from_millis(100);
mod rst_calibration;
#[derive(Debug, Clone)]
pub struct DpiHopProbeConfig {
pub target: SocketAddr,
@ -24,6 +27,7 @@ pub struct DpiHopProbeConfig {
pub hop_timeout: Duration,
pub accept_rst_fin_as_timeout: bool,
pub accept_ack_as_timeout: bool,
pub collect_tcp_metadata: bool,
}
#[derive(Debug, Clone)]
@ -33,6 +37,7 @@ pub struct DpiHopProbeResult {
pub client_hello_bytes: usize,
pub max_icmp_time_exceeded_ttl: Option<u8>,
pub hops: Vec<DpiHopProbeHop>,
pub calibration: reports::probe::DpiCalibrationDebug,
}
pub use reports::probe::{
@ -46,6 +51,18 @@ pub async fn detect_dpi_hop(config: DpiHopProbeConfig) -> io::Result<DpiHopProbe
}
pub fn detect_dpi_hop_blocking(config: DpiHopProbeConfig) -> io::Result<DpiHopProbeResult> {
let mut result = detect_post_client_hello(&config)?;
if let Some(reason) = result.calibration.failure_reason.as_deref()
&& matches!(reason, "early_close" | "no_icmp")
{
// A reset can close the classified stream before a TTL-limited packet
// ever leaves it. Locate the trigger with separate, paired ClientHellos instead
result.calibration = rst_calibration::calibrate(&config, reason.to_owned());
}
Ok(result)
}
fn detect_post_client_hello(config: &DpiHopProbeConfig) -> io::Result<DpiHopProbeResult> {
if config.max_ttl == 0 {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
@ -65,6 +82,7 @@ pub fn detect_dpi_hop_blocking(config: DpiHopProbeConfig) -> io::Result<DpiHopPr
let mut max_icmp_time_exceeded_ttl = None;
let mut result_local_addr: Option<SocketAddr> = None;
let mut client_hello_bytes = None;
let mut early_close = false;
for ttl in 1..=config.max_ttl {
// TCP is a byte stream: once a low-TTL segment is lost, later writes on
@ -72,7 +90,38 @@ pub fn detect_dpi_hop_blocking(config: DpiHopProbeConfig) -> io::Result<DpiHopPr
// subsequently changed socket TTL. Use an independent connection for
// every TTL so each hop corresponds to an actual packet and cannot
// affect later hops.
let mut tcp = TcpStream::connect_timeout(&config.target, config.connect_timeout)?;
let capture = config
.collect_tcp_metadata
.then(|| TcpCapture::start(config.target));
let (port, connection) = if capture.is_some() {
connect_with_port(config.target, config.connect_timeout)
} else {
(
None,
TcpStream::connect_timeout(&config.target, config.connect_timeout),
)
};
let mut tcp = match connection {
Ok(tcp) => tcp,
Err(error) => {
// Preserve partial progress for automatic calibration too.
let mut diagnostics = capture
.map(|capture| capture.finish(&port.into_iter().collect::<Vec<_>>()))
.unwrap_or_default();
diagnostics.connect_error = Some(error.to_string());
hops.push(DpiHopProbeHop {
ttl,
router: None,
outcome: if error.kind() == io::ErrorKind::ConnectionRefused {
DpiHopProbeHopOutcome::TcpClosed
} else {
DpiHopProbeHopOutcome::Timeout
},
tcp_diagnostics: Some(diagnostics),
});
break;
}
};
tcp.set_nodelay(true)?;
tcp.set_write_timeout(Some(config.connect_timeout))?;
let local_addr = tcp.local_addr()?;
@ -97,28 +146,70 @@ pub fn detect_dpi_hop_blocking(config: DpiHopProbeConfig) -> io::Result<DpiHopPr
result_local_addr = Some(local_addr);
client_hello_bytes = Some(client_hello.len());
}
tcp.write_all(&client_hello)?;
// A blocking DPI may silently discard the ClientHello, so waiting for
// a TCP acknowledgement would deadlock the measurement. Give the DPI
// a short interval to classify this new flow before sending the
// TTL-limited payload instead. This is needed for every attempt because
// each TTL deliberately uses a fresh TCP connection.
std::thread::sleep(DPI_CLASSIFICATION_DELAY);
drain_socket(&icmp)?;
let mut payload = [0u8; PROBE_BYTES];
rand::thread_rng().fill_bytes(&mut payload);
if !send_with_ttl(&mut tcp, config.target, &payload, ttl)? {
// Classify the flow with a normal-TTL ClientHello. Limiting this first
// packet triggers low-TTL filtering even for control SNI on some paths.
let hello_sent_ms = capture.as_ref().map(TcpCapture::elapsed_ms);
if let Err(error) = tcp.write_all(&client_hello) {
// Continue observing after a fast reset to catch competing server
// packets as well. Automatic measurements retain their old behavior.
std::thread::sleep(config.hop_timeout);
let mut diagnostics = finish_capture(capture, local_addr, hello_sent_ms, None);
if let Some(diagnostics) = &mut diagnostics {
diagnostics.send_error = Some(error.to_string());
}
early_close = matches!(
error.kind(),
io::ErrorKind::ConnectionReset | io::ErrorKind::BrokenPipe
);
hops.push(DpiHopProbeHop {
ttl,
router: None,
outcome: classify_hop(None, true, false, &config),
outcome: if matches!(
error.kind(),
io::ErrorKind::ConnectionReset | io::ErrorKind::BrokenPipe
) {
DpiHopProbeHopOutcome::TcpClosed
} else {
DpiHopProbeHopOutcome::Timeout
},
tcp_diagnostics: diagnostics,
});
if config.accept_rst_fin_as_timeout {
continue;
break;
}
std::thread::sleep(DPI_CLASSIFICATION_DELAY);
drain_socket(&icmp)?;
if peer_closed(&tcp)? {
early_close = true;
let mut diagnostics = finish_capture(capture, local_addr, hello_sent_ms, None);
if let Some(diagnostics) = &mut diagnostics {
diagnostics.payload_sent = Some(false);
diagnostics.peer_closed_before_payload = Some(true);
}
hops.push(DpiHopProbeHop {
ttl,
router: None,
outcome: DpiHopProbeHopOutcome::TcpClosed,
tcp_diagnostics: diagnostics,
});
break;
}
let mut payload = [0u8; PROBE_BYTES];
rand::thread_rng().fill_bytes(&mut payload);
let junk_sent_ms = capture.as_ref().map(TcpCapture::elapsed_ms);
let payload_sent = send_with_ttl(&mut tcp, config.target, &payload, ttl)?;
if !payload_sent {
early_close = true;
let mut diagnostics = finish_capture(capture, local_addr, hello_sent_ms, junk_sent_ms);
if let Some(diagnostics) = &mut diagnostics {
diagnostics.payload_sent = Some(false);
diagnostics.peer_closed_before_payload = Some(false);
}
hops.push(DpiHopProbeHop {
ttl,
router: None,
outcome: DpiHopProbeHopOutcome::TcpClosed,
tcp_diagnostics: diagnostics,
});
break;
}
@ -126,28 +217,100 @@ pub fn detect_dpi_hop_blocking(config: DpiHopProbeConfig) -> io::Result<DpiHopPr
listen_for_time_exceeded(&icmp, local_addr, config.target, config.hop_timeout)?;
let closed = peer_closed(&tcp)?;
let acknowledged = !closed && tcp_payload_acknowledged(&tcp)?;
let outcome = classify_hop(router, closed, acknowledged, &config);
let outcome = classify_hop(router, closed, acknowledged, config);
if outcome == DpiHopProbeHopOutcome::IcmpTimeExceeded {
max_icmp_time_exceeded_ttl = Some(ttl);
}
let mut diagnostics = finish_capture(capture, local_addr, hello_sent_ms, junk_sent_ms);
if let Some(diagnostics) = &mut diagnostics {
diagnostics.payload_sent = Some(true);
diagnostics.peer_closed_before_payload = Some(false);
}
hops.push(DpiHopProbeHop {
ttl,
router,
outcome,
tcp_diagnostics: diagnostics,
});
if outcome.invalidates_measurement() {
break;
}
}
let calibration = legacy_calibration(&hops, max_icmp_time_exceeded_ttl, early_close);
Ok(DpiHopProbeResult {
target: config.target,
local_addr: result_local_addr
.ok_or_else(|| io::Error::other("DPI hop probe made no TTL attempts"))?,
client_hello_bytes: client_hello_bytes
.ok_or_else(|| io::Error::other("DPI hop probe produced no ClientHello"))?,
local_addr: result_local_addr.unwrap_or_else(|| {
SocketAddr::new(
if config.target.is_ipv4() {
IpAddr::V4(Ipv4Addr::UNSPECIFIED)
} else {
IpAddr::V6(std::net::Ipv6Addr::UNSPECIFIED)
},
0,
)
}),
client_hello_bytes: client_hello_bytes.unwrap_or(0),
max_icmp_time_exceeded_ttl,
hops,
calibration,
})
}
fn legacy_calibration(
hops: &[DpiHopProbeHop],
max_icmp_time_exceeded_ttl: Option<u8>,
early_close: bool,
) -> reports::probe::DpiCalibrationDebug {
let failure_reason = if hops.iter().any(|hop| {
hop.tcp_diagnostics
.as_ref()
.is_some_and(|d| d.connect_error.is_some())
}) {
Some("connect_error")
} else if early_close {
Some("early_close")
} else if hops.iter().any(|hop| {
hop.outcome.invalidates_measurement()
|| hop
.tcp_diagnostics
.as_ref()
.is_some_and(|d| d.send_error.is_some())
}) {
Some("invalid_tcp_state")
} else if max_icmp_time_exceeded_ttl.is_none() {
Some("no_icmp")
} else {
None
};
reports::probe::DpiCalibrationDebug {
method: "post_client_hello".into(),
dpi_hop: failure_reason
.is_none()
.then_some(max_icmp_time_exceeded_ttl)
.flatten(),
failure_reason: failure_reason.map(str::to_owned),
..Default::default()
}
}
fn finish_capture(
capture: Option<TcpCapture>,
local: SocketAddr,
hello_ms: Option<u64>,
junk_ms: Option<u64>,
) -> Option<reports::probe::TcpDiagnostics> {
capture.map(|capture| {
let mut diagnostics = capture.finish(&[local.port()]);
for packet in &mut diagnostics.packets {
// Raw IPv6 sockets may omit the destination IP header.
if packet.destination.ip().is_unspecified() {
packet.destination.set_ip(local.ip());
}
}
diagnostics.client_hello_sent_ms = hello_ms;
diagnostics.junk_sent_ms = junk_ms;
diagnostics
})
}
@ -470,6 +633,156 @@ fn matching_quoted_tcp_tuple(packet: &[u8], local_addr: SocketAddr, target: Sock
mod tests {
use super::*;
#[test]
fn legacy_calibration_preserves_working_boundaries_but_never_invents_zero() {
let mut hops = vec![DpiHopProbeHop {
ttl: 2,
router: Some("192.0.2.1".parse().unwrap()),
outcome: DpiHopProbeHopOutcome::IcmpTimeExceeded,
tcp_diagnostics: None,
}];
assert_eq!(legacy_calibration(&hops, Some(2), false).dpi_hop, Some(2));
assert_eq!(legacy_calibration(&[], None, false).dpi_hop, None);
assert_eq!(legacy_calibration(&hops, Some(2), true).dpi_hop, None);
hops.push(DpiHopProbeHop {
ttl: 3,
router: None,
outcome: DpiHopProbeHopOutcome::TcpAcknowledged,
tcp_diagnostics: None,
});
assert_eq!(legacy_calibration(&hops, Some(2), false).dpi_hop, None);
hops.last_mut().unwrap().tcp_diagnostics = Some(reports::probe::TcpDiagnostics {
connect_error: Some("connection timed out".into()),
..Default::default()
});
assert_eq!(
legacy_calibration(&hops, Some(2), false)
.failure_reason
.as_deref(),
Some("connect_error")
);
}
#[test]
#[ignore = "requires CAP_NET_RAW and local socket access"]
fn live_reset_before_payload_triggers_fallback_without_trusting_control_resets() {
use std::io::Read;
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let target = listener.local_addr().unwrap();
let server = std::thread::spawn(move || {
// One legacy attempt, then the fallback's clean/blocked pair.
for _ in 0..3 {
let (mut stream, _) = listener.accept().unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
let mut header = [0; 5];
stream.read_exact(&mut header).unwrap();
let mut hello = vec![0; u16::from_be_bytes([header[3], header[4]]) as usize];
stream.read_exact(&mut hello).unwrap();
SockRef::from(&stream)
.set_linger(Some(Duration::ZERO))
.unwrap();
}
});
let trace = detect_dpi_hop_blocking(DpiHopProbeConfig {
target,
max_ttl: 2,
collect_tcp_metadata: true,
accept_rst_fin_as_timeout: true,
..probe_config()
})
.unwrap();
server.join().unwrap();
assert_eq!(trace.hops.len(), 1);
assert_eq!(trace.calibration.dpi_hop, None);
assert_eq!(trace.calibration.method, "rst_client_hello");
assert_eq!(
trace.calibration.fallback_reason.as_deref(),
Some("early_close")
);
assert_eq!(
trace.calibration.failure_reason.as_deref(),
Some("control_not_clear_at_reset")
);
let diagnostics = trace.hops[0].tcp_diagnostics.as_ref().unwrap();
assert_eq!(diagnostics.payload_sent, Some(false));
assert_eq!(diagnostics.peer_closed_before_payload, Some(true));
assert_eq!(diagnostics.junk_sent_ms, None);
assert!(
diagnostics
.packets
.iter()
.any(|p| p.flags.iter().any(|f| f == "RST"))
);
}
#[test]
#[ignore = "requires CAP_NET_RAW and local socket access"]
fn live_clienthello_keeps_default_ttl_and_junk_uses_tested_ttl() {
use std::io::Read;
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let target = listener.local_addr().unwrap();
let hello_len = make_client_hello("example.com").unwrap().len();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
let mut hello = vec![0; hello_len];
stream.read_exact(&mut hello).unwrap();
assert_eq!(hello[0], 22); // TLS handshake record
let mut junk = [0; PROBE_BYTES];
stream.read_exact(&mut junk).unwrap();
stream.write_all(b"response").unwrap();
std::thread::sleep(Duration::from_millis(250));
});
let receiver = Socket::new(Domain::IPV4, Type::RAW, Some(Protocol::TCP)).unwrap();
receiver
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
let observer = std::thread::spawn(move || {
let mut packets = Vec::new();
let mut buffer = [0; 65535];
while packets.len() < 2 {
let (bytes, _) = recv_socket(&receiver, &mut buffer).unwrap();
let packet = LaxSlicedPacket::from_ip(&buffer[..bytes]).unwrap();
if let (Some(LaxNetSlice::Ipv4(ip)), Some(TransportSlice::Tcp(tcp))) =
(packet.net, packet.transport)
{
if tcp.destination_port() == target.port() && !tcp.payload().is_empty() {
packets.push((ip.header().ttl(), tcp.payload().len()));
}
}
}
packets
});
let default_ttl = Socket::new(Domain::IPV4, Type::STREAM, Some(Protocol::TCP))
.unwrap()
.ttl_v4()
.unwrap() as u8;
let trace = detect_dpi_hop_blocking(DpiHopProbeConfig {
target,
max_ttl: 1,
hop_timeout: Duration::from_millis(100),
collect_tcp_metadata: true,
..probe_config()
})
.unwrap();
let packets = observer.join().unwrap();
server.join().unwrap();
assert_eq!(packets, [(default_ttl, hello_len), (1, PROBE_BYTES)]);
let metadata = trace.hops[0].tcp_diagnostics.as_ref().unwrap();
assert_eq!(metadata.capture_error, None);
assert!(
metadata
.packets
.iter()
.any(|p| p.flags.contains(&"SYN".into()))
);
assert!(metadata.junk_sent_ms.unwrap() >= metadata.client_hello_sent_ms.unwrap() + 100);
}
fn probe_config() -> DpiHopProbeConfig {
DpiHopProbeConfig {
target: "192.0.2.1:443".parse().unwrap(),
@ -479,6 +792,7 @@ mod tests {
hop_timeout: Duration::from_secs(1),
accept_rst_fin_as_timeout: false,
accept_ack_as_timeout: false,
collect_tcp_metadata: false,
}
}

View file

@ -0,0 +1,703 @@
//! Reset- and silent-drop-aware calibration stays on the calibration destination. Ordinary
//! target tasks still send only the existing TTL-limited TCP probes.
use super::*;
use reports::probe::{DpiCalibrationAttempt, DpiCalibrationDebug};
const BUDGET: Duration = Duration::from_secs(45);
const REPEATS: usize = 3;
pub(super) fn calibrate(config: &DpiHopProbeConfig, reason: String) -> DpiCalibrationDebug {
// Keep the negative control different even for a manual trace of example.com.
let control_sni = if config
.control_sni
.trim_end_matches('.')
.eq_ignore_ascii_case("example.com")
{
"example.net"
} else {
"example.com"
};
let deadline = Instant::now() + BUDGET;
let mut debug = scan(config.max_ttl, |ttl, control| {
attempt(
config,
if control {
control_sni
} else {
&config.control_sni
},
ttl,
control,
deadline,
)
});
debug.control_sni = Some(control_sni.into());
debug.fallback_reason = Some(reason);
debug
}
fn scan(
max_ttl: u8,
mut observe: impl FnMut(u8, bool) -> io::Result<DpiCalibrationAttempt>,
) -> DpiCalibrationDebug {
let mut debug = DpiCalibrationDebug {
method: "rst_client_hello".into(),
..Default::default()
};
let result = (|| -> Result<u8, String> {
let mut last_clear = None;
for ttl in 1..=max_ttl {
let observation = pair(&mut debug, &mut observe, ttl)?;
if observation.clear() {
last_clear = Some(ttl);
}
// Require repeated positive controls at/above the drop. A clear
// lower hop is optional: filtering may already act at TTL 1.
if observation.differential_drop() {
if let Some(lower) = last_clear {
for _ in 0..REPEATS {
if !pair(&mut debug, &mut observe, lower)?.clear() {
return Err("unstable_lower_boundary".into());
}
}
}
for _ in 1..REPEATS {
if !pair(&mut debug, &mut observe, ttl)?.differential_drop() {
return Err("unstable_drop_boundary".into());
}
}
let higher = ttl
.checked_add(1)
.filter(|higher| *higher <= max_ttl)
.ok_or("no_control_expiry_beyond_drop")?;
for _ in 0..REPEATS {
if !pair(&mut debug, &mut observe, higher)?.differential_drop() {
return Err("non_monotonic_drop_boundary".into());
}
}
debug.method = "silent_drop_client_hello".into();
debug.last_clear_ttl = last_clear;
debug.first_drop_ttl = Some(ttl);
debug.control_witness_ttl = Some(higher);
// First-hop filtering can be tested with TTL 1, so represent
// that case as hop 0 (ordinary checks start at dpi_hop + 1).
// Keep conservative upper bounds for later/gapped boundaries.
return Ok(if ttl == 1 { 0 } else { ttl });
}
if observation.differential_reset() {
let mut witness = observation.control_expired().then_some(ttl);
// Routers can suppress ICMP. Recheck the last positive lower
// observation, which need not be adjacent to the reset TTL.
if let Some(lower) = last_clear {
for _ in 0..REPEATS {
if !pair(&mut debug, &mut observe, lower)?.clear() {
return Err("unstable_lower_boundary".into());
}
}
}
for _ in 1..REPEATS {
let confirmation = pair(&mut debug, &mut observe, ttl)?;
if !confirmation.differential_reset() {
return Err("unstable_reset_boundary".into());
}
witness = witness.or(confirmation.control_expired().then_some(ttl));
}
// A positive control expiry at/above the reset is required:
// silence alone cannot exclude a reset from the destination.
// Also check that the reset persists above the chosen bound.
for higher in ttl.saturating_add(1)..=max_ttl {
if higher == ttl {
break;
}
let confirmation = pair(&mut debug, &mut observe, higher)?;
if !confirmation.differential_reset() {
return Err("non_monotonic_reset_boundary".into());
}
witness = witness.or(confirmation.control_expired().then_some(higher));
if witness.is_some() {
break;
}
}
if witness.is_none() {
return Err("no_control_expiry_beyond_reset".into());
}
debug.last_clear_ttl = last_clear;
debug.first_reset_ttl = Some(ttl);
debug.control_witness_ttl = witness;
// This is an upper bound on the trigger, not the last ICMP hop.
// The existing task path starts at dpi_hop + 1, beyond that bound.
return Ok(ttl);
}
if observation.test == "tcp_reset" {
return Err("control_not_clear_at_reset".into());
}
}
Err("no_repeatable_reset_boundary".into())
})();
match result {
Ok(ttl) => debug.dpi_hop = Some(ttl),
Err(reason) => debug.failure_reason = Some(reason),
}
debug
}
struct PairObservation {
control: String,
test: String,
}
impl PairObservation {
fn clear(&self) -> bool {
self.control_expired() && self.test == "icmp_time_exceeded"
}
fn control_expired(&self) -> bool {
self.control == "icmp_time_exceeded"
}
fn differential_drop(&self) -> bool {
self.control_expired() && self.test == "timeout"
}
fn differential_reset(&self) -> bool {
matches!(self.control.as_str(), "icmp_time_exceeded" | "timeout")
&& self.test == "tcp_reset"
}
}
fn pair(
debug: &mut DpiCalibrationDebug,
observe: &mut impl FnMut(u8, bool) -> io::Result<DpiCalibrationAttempt>,
ttl: u8,
) -> Result<PairObservation, String> {
let mut outcomes = Vec::with_capacity(2);
for control in [true, false] {
let attempt = observe(ttl, control).map_err(|error| error.to_string())?;
let error = attempt.error.clone();
outcomes.push(attempt.outcome.clone());
debug.attempts.push(attempt);
if let Some(error) = error {
return Err(error);
}
}
Ok(PairObservation {
control: outcomes.remove(0),
test: outcomes.remove(0),
})
}
fn remaining(deadline: Instant, limit: Duration) -> io::Result<Duration> {
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
Err(io::Error::new(
io::ErrorKind::TimedOut,
"rst_calibration_budget_exhausted",
))
} else {
Ok(remaining.min(limit))
}
}
fn attempt(
config: &DpiHopProbeConfig,
sni: &str,
ttl: u8,
control: bool,
deadline: Instant,
) -> io::Result<DpiCalibrationAttempt> {
let timeout = remaining(deadline, config.connect_timeout)?;
let (domain, protocol) = if config.target.is_ipv4() {
(Domain::IPV4, Protocol::ICMPV4)
} else {
(Domain::IPV6, Protocol::ICMPV6)
};
let icmp = Socket::new(domain, Type::RAW, Some(protocol))?;
// Capture is necessary even in automatic mode: a closed socket alone cannot
// distinguish an injected RST from FIN, a TLS alert, or a local socket error.
let capture = TcpCapture::start(config.target);
let (port, connection) = connect_with_port(config.target, timeout);
let mut router = None;
let mut hello_ms = None;
let mut local = None;
let mut stream = None;
let measurement = (|| -> io::Result<()> {
stream = Some(connection?);
let tcp = stream.as_mut().unwrap();
tcp.set_nodelay(true)?;
tcp.set_write_timeout(Some(remaining(deadline, config.connect_timeout)?))?;
let address = tcp.local_addr()?;
local = Some(address);
let mut bind = address;
bind.set_port(0);
icmp.bind(&bind.into())?;
// The handshake used normal TTL; only the ClientHello is TTL-limited.
// Leave that TTL installed until drop, including any retransmissions.
set_ttl(tcp, config.target, u32::from(ttl))?;
let hello = make_client_hello(sni)?;
hello_ms = Some(capture.elapsed_ms());
tcp.write_all(&hello)?;
router = listen_for_time_exceeded(
&icmp,
address,
config.target,
remaining(deadline, config.hop_timeout)?,
)?;
Ok(())
})();
let mut diagnostics = capture.finish(&port.into_iter().collect::<Vec<_>>());
// Normalize raw IPv6 capture addresses in the same way as the legacy trace.
if let Some(local) = local {
for packet in &mut diagnostics.packets {
if packet.destination.ip().is_unspecified() {
packet.destination.set_ip(local.ip());
}
}
}
diagnostics.client_hello_sent_ms = hello_ms;
let reset = diagnostics.packets.iter().any(|packet| {
hello_ms.is_some_and(|sent| packet.observed_ms >= sent)
&& packet.flags.iter().any(|flag| flag == "RST")
});
let mut error = None;
if let Err(failure) = measurement {
let message = failure.to_string();
if hello_ms.is_none() {
diagnostics.connect_error = Some(message.clone());
} else {
diagnostics.send_error = Some(message.clone());
}
// An observed reset is useful even if write_all also reports the reset.
if !reset {
error = Some(message);
}
}
if let Some(capture_error) = &diagnostics.capture_error {
error = Some(format!("rst_capture_unavailable: {capture_error}"));
}
let outcome = if reset {
"tcp_reset"
} else if router.is_some() {
"icmp_time_exceeded"
} else if error.is_some() {
"error"
} else if diagnostics
.packets
.iter()
.any(|packet| packet.flags.iter().any(|flag| flag == "FIN"))
{
"tcp_closed"
} else if diagnostics
.packets
.iter()
.any(|packet| packet.payload_bytes > 0)
{
"tcp_response"
} else {
"timeout"
};
// Keep command replies bounded even when the peer sends a large TLS flight.
if diagnostics.packets.len() > 8 {
diagnostics.packets.truncate(8);
diagnostics.truncated = true;
}
Ok(DpiCalibrationAttempt {
ttl,
control,
outcome: outcome.into(),
router,
error,
tcp_diagnostics: config.collect_tcp_metadata.then_some(diagnostics),
})
}
#[cfg(test)]
mod tests {
use super::*;
fn sample(ttl: u8, control: bool, outcome: &str) -> DpiCalibrationAttempt {
DpiCalibrationAttempt {
ttl,
control,
outcome: outcome.into(),
router: (outcome == "icmp_time_exceeded").then(|| "192.0.2.1".parse().unwrap()),
error: None,
tcp_diagnostics: None,
}
}
#[test]
fn brackets_reset_injection_and_uses_upper_bound_for_target_probes() {
let result = scan(8, |ttl, control| {
Ok(sample(
ttl,
control,
if !control && ttl >= 4 {
"tcp_reset"
} else {
"icmp_time_exceeded"
},
))
});
assert_eq!(result.dpi_hop, Some(4));
assert_eq!(result.last_clear_ttl, Some(3));
assert_eq!(result.first_reset_ttl, Some(4));
assert_eq!(
result
.attempts
.iter()
.filter(|a| a.ttl == 4 && !a.control)
.count(),
REPEATS
);
assert!(result.attempts.iter().any(|a| a.ttl == 5));
}
#[test]
fn silent_drop_recovers_node_40_boundary() {
let result = scan(15, |ttl, control| {
Ok(sample(
ttl,
control,
if ttl < 8 || control {
"icmp_time_exceeded"
} else {
"timeout"
},
))
});
assert_eq!(result.dpi_hop, Some(8));
assert_eq!(result.last_clear_ttl, Some(7));
assert_eq!(result.first_drop_ttl, Some(8));
assert_eq!(result.first_reset_ttl, None);
assert_eq!(result.control_witness_ttl, Some(9));
assert_eq!(result.method, "silent_drop_client_hello");
}
#[test]
fn transient_drop_and_changing_network_are_rejected() {
for changing_lower in [false, true] {
let mut counts = std::collections::HashMap::new();
let result = scan(6, |ttl, control| {
let count = counts.entry((ttl, control)).or_insert(0);
*count += 1;
let dropped = !control
&& ((ttl >= 3 && (changing_lower || *count == 1))
|| (changing_lower && ttl == 2 && *count > 1));
Ok(sample(
ttl,
control,
if dropped {
"timeout"
} else {
"icmp_time_exceeded"
},
))
});
assert_eq!(result.dpi_hop, None);
assert_eq!(
result.failure_reason.as_deref(),
Some(if changing_lower {
"unstable_lower_boundary"
} else {
"unstable_drop_boundary"
})
);
}
}
#[test]
fn silent_drop_must_persist_at_a_higher_ttl() {
let result = scan(6, |ttl, control| {
Ok(sample(
ttl,
control,
if ttl == 3 && !control {
"timeout"
} else {
"icmp_time_exceeded"
},
))
});
assert_eq!(result.dpi_hop, None);
assert_eq!(
result.failure_reason.as_deref(),
Some("non_monotonic_drop_boundary")
);
}
#[test]
fn first_hop_silent_drop_is_hop_zero_without_a_lower_observation() {
let result = scan(6, |ttl, control| {
Ok(sample(
ttl,
control,
if control {
"icmp_time_exceeded"
} else {
"timeout"
},
))
});
assert_eq!(result.dpi_hop, Some(0));
assert_eq!(result.last_clear_ttl, None);
assert_eq!(result.first_drop_ttl, Some(1));
assert_eq!(result.control_witness_ttl, Some(2));
assert_eq!(result.failure_reason, None);
for ttl in [1, 2] {
assert_eq!(
result
.attempts
.iter()
.filter(|a| a.ttl == ttl && !a.control)
.count(),
REPEATS
);
}
}
#[test]
fn missing_lower_observation_after_icmp_gaps_keeps_upper_bound() {
let result = scan(6, |ttl, control| {
Ok(sample(
ttl,
control,
if ttl >= 3 && control {
"icmp_time_exceeded"
} else {
"timeout"
},
))
});
assert_eq!(result.dpi_hop, Some(3));
assert_eq!(result.last_clear_ttl, None);
assert_eq!(result.first_drop_ttl, Some(3));
}
#[test]
fn first_hop_transient_silence_is_rejected() {
let mut dropped = false;
let result = scan(6, |ttl, control| {
let timeout = ttl == 1 && !control && !dropped;
dropped |= timeout;
Ok(sample(
ttl,
control,
if timeout {
"timeout"
} else {
"icmp_time_exceeded"
},
))
});
assert_eq!(result.dpi_hop, None);
assert_eq!(
result.failure_reason.as_deref(),
Some("unstable_drop_boundary")
);
}
#[test]
fn silent_drop_still_requires_a_higher_positive_control() {
let result = scan(4, |ttl, control| {
Ok(sample(
ttl,
control,
if ttl < 4 || control {
"icmp_time_exceeded"
} else {
"timeout"
},
))
});
assert_eq!(result.dpi_hop, None);
assert_eq!(
result.failure_reason.as_deref(),
Some("no_control_expiry_beyond_drop")
);
}
#[test]
fn identical_low_ttl_resets_are_not_a_boundary() {
let result = scan(8, |ttl, control| Ok(sample(ttl, control, "tcp_reset")));
assert_eq!(result.dpi_hop, None);
assert_eq!(
result.failure_reason.as_deref(),
Some("control_not_clear_at_reset")
);
}
#[test]
fn missing_icmp_is_unknown_not_hop_zero() {
let result = scan(4, |ttl, control| Ok(sample(ttl, control, "timeout")));
assert_eq!(result.dpi_hop, None);
}
#[test]
fn icmp_gaps_keep_a_range_with_a_positive_control_witness() {
let result = scan(8, |ttl, control| {
let outcome = if !control && ttl >= 4 {
"tcp_reset"
} else if ttl == 3 || ttl == 4 || ttl == 5 {
"timeout"
} else {
"icmp_time_exceeded"
};
Ok(sample(ttl, control, outcome))
});
assert_eq!(result.dpi_hop, Some(4));
assert_eq!(result.last_clear_ttl, Some(2));
assert_eq!(result.control_witness_ttl, Some(6));
}
#[test]
fn reset_without_any_control_expiry_is_not_trusted() {
let result = scan(4, |ttl, control| {
Ok(sample(
ttl,
control,
if control { "timeout" } else { "tcp_reset" },
))
});
assert_eq!(result.dpi_hop, None);
assert_eq!(
result.failure_reason.as_deref(),
Some("no_control_expiry_beyond_reset")
);
}
#[test]
fn transient_reset_does_not_survive_confirmation() {
let mut injected = false;
let result = scan(6, |ttl, control| {
let reset = ttl == 3 && !control && !injected;
injected |= reset;
Ok(sample(
ttl,
control,
if reset {
"tcp_reset"
} else {
"icmp_time_exceeded"
},
))
});
assert_eq!(result.dpi_hop, None);
assert_eq!(
result.failure_reason.as_deref(),
Some("unstable_reset_boundary")
);
}
#[test]
fn non_monotonic_threshold_is_rejected() {
let result = scan(6, |ttl, control| {
Ok(sample(
ttl,
control,
if ttl == 3 && !control {
"tcp_reset"
} else {
"icmp_time_exceeded"
},
))
});
assert_eq!(result.dpi_hop, None);
assert_eq!(
result.failure_reason.as_deref(),
Some("non_monotonic_reset_boundary")
);
}
#[test]
fn partial_attempts_survive_a_connection_failure() {
let result = scan(6, |ttl, control| {
let mut attempt = sample(ttl, control, "icmp_time_exceeded");
if ttl == 3 {
attempt.error = Some("connection timed out".into());
}
Ok(attempt)
});
assert_eq!(result.dpi_hop, None);
assert_eq!(result.attempts.len(), 5);
assert_eq!(
result.failure_reason.as_deref(),
Some("connection timed out")
);
}
#[test]
fn expired_budget_prevents_another_connection() {
assert!(remaining(Instant::now(), Duration::from_secs(5)).is_err());
}
#[test]
#[ignore = "requires CAP_NET_RAW and local socket access"]
fn live_clienthello_ttl_is_limited_and_fin_is_not_a_reset() {
use std::io::Read;
for reset in [false, true] {
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let target = listener.local_addr().unwrap();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
let mut header = [0; 5];
stream.read_exact(&mut header).unwrap();
let mut hello = vec![0; u16::from_be_bytes([header[3], header[4]]) as usize];
stream.read_exact(&mut hello).unwrap();
if reset {
SockRef::from(&stream)
.set_linger(Some(Duration::ZERO))
.unwrap();
}
});
let receiver = Socket::new(Domain::IPV4, Type::RAW, Some(Protocol::TCP)).unwrap();
receiver
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
let observer = std::thread::spawn(move || {
let mut buffer = [0; 65535];
loop {
let (bytes, _) = recv_socket(&receiver, &mut buffer).unwrap();
let packet = LaxSlicedPacket::from_ip(&buffer[..bytes]).unwrap();
if let (Some(LaxNetSlice::Ipv4(ip)), Some(TransportSlice::Tcp(tcp))) =
(packet.net, packet.transport)
&& tcp.destination_port() == target.port()
&& !tcp.payload().is_empty()
{
return ip.header().ttl();
}
}
});
let config = DpiHopProbeConfig {
target,
control_sni: "instagram.com".into(),
max_ttl: 4,
connect_timeout: Duration::from_secs(1),
hop_timeout: Duration::from_millis(100),
accept_rst_fin_as_timeout: true,
accept_ack_as_timeout: false,
collect_tcp_metadata: true,
};
let result = attempt(
&config,
"instagram.com",
3,
false,
Instant::now() + Duration::from_secs(3),
)
.unwrap();
assert_eq!(observer.join().unwrap(), 3);
server.join().unwrap();
assert_eq!(
result.outcome,
if reset { "tcp_reset" } else { "tcp_closed" }
);
assert!(result.error.is_none());
let diagnostics = result.tcp_diagnostics.unwrap();
assert!(diagnostics.client_hello_sent_ms.is_some());
assert_eq!(diagnostics.junk_sent_ms, None);
}
}
}

View file

@ -1,5 +1,6 @@
mod dns;
mod dpi_hop;
mod packet_capture;
mod sni;
mod traceroute;
mod update;
@ -41,6 +42,8 @@ struct DpiHops {
v6: Option<u8>,
hops_v4: Vec<reports::probe::DpiProbeHop>,
hops_v6: Vec<reports::probe::DpiProbeHop>,
calibration_v4: Option<reports::probe::DpiCalibrationDebug>,
calibration_v6: Option<reports::probe::DpiCalibrationDebug>,
}
#[derive(Parser, Debug, Clone)]
@ -344,6 +347,7 @@ async fn main() -> Result<()> {
v6: config.dpi_hop_v6,
hops_v4: config.dpi_hops_v4.clone(),
hops_v6: config.dpi_hops_v6.clone(),
..Default::default()
});
publish_status(&client, &status_topic, &args, true, dpi_hops).await?;
client.subscribe(CONFIG_TOPIC, QoS::AtLeastOnce).await?;
@ -461,7 +465,7 @@ async fn update_config(
bail!("dns_spoofing_provider_threshold must be between 1 and 4");
}
validate_dpi_probe_config(value.dpi_probe.as_ref())?;
let dpi_hops = measure_dpi_hops(value.dpi_probe.as_ref()).await;
let dpi_hops = measure_dpi_hops(value.dpi_probe.as_ref(), false).await;
let loaded = LoadedProbeConfig {
config: value,
dpi_hop_v4: dpi_hops.v4,
@ -494,7 +498,15 @@ async fn run_dpi_command(
.dpi_probe
.as_ref()
.context("DPI hop measurement is not configured")?;
let hops = measure_dpi_hops(Some(dpi)).await;
let hops = measure_dpi_hops(Some(dpi), true).await;
let mut status_hops = hops.clone();
for hop in status_hops
.hops_v4
.iter_mut()
.chain(&mut status_hops.hops_v6)
{
hop.tcp_diagnostics = None;
}
let mut guard = config.write().await;
let loaded = guard.as_mut().context("probe config is not loaded")?;
if loaded.config.version != snapshot.config.version
@ -506,16 +518,18 @@ async fn run_dpi_command(
}
loaded.dpi_hop_v4 = hops.v4;
loaded.dpi_hop_v6 = hops.v6;
loaded.dpi_hops_v4 = hops.hops_v4.clone();
loaded.dpi_hops_v6 = hops.hops_v6.clone();
loaded.dpi_hops_v4 = status_hops.hops_v4.clone();
loaded.dpi_hops_v6 = status_hops.hops_v6.clone();
Ok((
ProbeCommandResult::RemeasureDpiHop {
dpi_hop_v4: hops.v4,
dpi_hop_v6: hops.v6,
dpi_hops_v4: hops.hops_v4.clone(),
dpi_hops_v6: hops.hops_v6.clone(),
calibration_v4: hops.calibration_v4.clone(),
calibration_v6: hops.calibration_v6.clone(),
},
Some(hops),
Some(status_hops),
))
}
ProbeCommand::SniTraceroute {
@ -553,6 +567,7 @@ async fn run_dpi_command(
accept_ack_as_timeout: dpi_config
.as_ref()
.is_some_and(|dpi| dpi.accept_ack_as_timeout),
collect_tcp_metadata: true,
})
.await?;
Ok((
@ -562,6 +577,7 @@ async fn run_dpi_command(
target,
dpi_hop: dpi_hop_from_result(&trace),
hops: trace.hops,
calibration: Some(trace.calibration),
},
None,
))
@ -598,7 +614,7 @@ fn validate_dpi_probe_config(config: Option<&DpiProbeConfig>) -> Result<()> {
Ok(())
}
async fn measure_dpi_hops(config: Option<&DpiProbeConfig>) -> DpiHops {
async fn measure_dpi_hops(config: Option<&DpiProbeConfig>, collect_tcp_metadata: bool) -> DpiHops {
let Some(config) = config else {
debug!("DPI hop measurement is not configured");
return DpiHops::default();
@ -611,6 +627,7 @@ async fn measure_dpi_hops(config: Option<&DpiProbeConfig>) -> DpiHops {
hop_timeout: Duration::from_millis(config.hop_timeout_ms),
accept_rst_fin_as_timeout: config.accept_rst_fin_as_timeout,
accept_ack_as_timeout: config.accept_ack_as_timeout,
collect_tcp_metadata,
};
let (v4, v6) = tokio::join!(
measure_dpi_hop(common(config.target_v4.into())),
@ -621,26 +638,47 @@ async fn measure_dpi_hops(config: Option<&DpiProbeConfig>) -> DpiHops {
v6: v6.0,
hops_v4: v4.1,
hops_v6: v6.1,
calibration_v4: v4.2,
calibration_v6: v6.2,
}
}
async fn measure_dpi_hop(
config: dpi_hop::DpiHopProbeConfig,
) -> (Option<u8>, Vec<reports::probe::DpiProbeHop>) {
) -> (
Option<u8>,
Vec<reports::probe::DpiProbeHop>,
Option<reports::probe::DpiCalibrationDebug>,
) {
let target = config.target;
match dpi_hop::detect_dpi_hop(config).await {
Ok(result) => (dpi_hop_from_result(&result), result.hops),
Ok(result) => (
dpi_hop_from_result(&result),
result.hops,
Some(result.calibration),
),
Err(error) => {
warn!("failed to measure DPI hop for {target}: {error}");
(None, Vec::new())
(
None,
Vec::new(),
Some(reports::probe::DpiCalibrationDebug {
method: "post_client_hello".into(),
failure_reason: Some(error.to_string()),
..Default::default()
}),
)
}
}
}
fn dpi_hop_from_result(result: &dpi_hop::DpiHopProbeResult) -> Option<u8> {
debug!(
"DPI probe completed: target={}, local={}, ClientHello={} bytes",
result.target, result.local_addr, result.client_hello_bytes
"DPI probe completed: target={}, local={}, ClientHello={} bytes, legacy ICMP hop={:?}",
result.target,
result.local_addr,
result.client_hello_bytes,
result.max_icmp_time_exceeded_ttl
);
for hop in &result.hops {
debug!(
@ -648,18 +686,29 @@ fn dpi_hop_from_result(result: &dpi_hop::DpiHopProbeResult) -> Option<u8> {
result.target, hop.ttl, hop.router, hop.outcome
);
}
if let Some(invalid_hop) = result
.hops
.iter()
.find(|hop| hop.outcome.invalidates_measurement())
{
debug!(
"DPI calibration: method={}, hop={:?}, failure={:?}",
result.calibration.method, result.calibration.dpi_hop, result.calibration.failure_reason
);
if let Some(reason) = &result.calibration.failure_reason {
warn!(
"DPI hop measurement for {} is invalid: {:?} at TTL {}",
result.target, invalid_hop.outcome, invalid_hop.ttl
"DPI calibration for {} is unknown: method={}, reason={reason}",
result.target, result.calibration.method
);
} else if matches!(
result.calibration.method.as_str(),
"rst_client_hello" | "silent_drop_client_hello"
) {
info!(
"DPI boundary for {}: method={}, lower={:?}, upper={:?}, control witness={:?}",
result.target,
result.calibration.method,
result.calibration.last_clear_ttl,
result.calibration.dpi_hop,
result.calibration.control_witness_ttl
);
return None;
}
Some(result.max_icmp_time_exceeded_ttl.unwrap_or(0))
result.calibration.dpi_hop
}
async fn wait_for_connection(eventloop: &mut rumqttc::EventLoop) {
@ -927,17 +976,20 @@ mod tests {
ttl: 1,
router: Some("192.0.2.1".parse().unwrap()),
outcome: reports::probe::DpiProbeHopOutcome::IcmpTimeExceeded,
tcp_diagnostics: None,
}],
dpi_hops_v6: vec![
reports::probe::DpiProbeHop {
ttl: 1,
router: Some("2001:db8::1".parse().unwrap()),
outcome: reports::probe::DpiProbeHopOutcome::IcmpTimeExceeded,
tcp_diagnostics: None,
},
reports::probe::DpiProbeHop {
ttl: 2,
router: None,
outcome: reports::probe::DpiProbeHopOutcome::TcpClosed,
tcp_diagnostics: None,
},
],
};
@ -962,41 +1014,21 @@ mod tests {
}
#[test]
fn direct_tcp_delivery_invalidates_only_its_measurement() {
for outcome in [
dpi_hop::DpiHopProbeHopOutcome::TcpClosed,
dpi_hop::DpiHopProbeHopOutcome::TcpAcknowledged,
] {
let result = dpi_hop::DpiHopProbeResult {
target: "[2001:db8::10]:443".parse().unwrap(),
local_addr: "[2001:db8::1]:45000".parse().unwrap(),
client_hello_bytes: 256,
max_icmp_time_exceeded_ttl: Some(4),
hops: vec![dpi_hop::DpiHopProbeHop {
ttl: 5,
router: None,
outcome,
}],
};
assert_eq!(dpi_hop_from_result(&result), None);
}
}
#[test]
fn missing_icmp_time_exceeded_defaults_dpi_hop_to_zero() {
fn fallback_boundary_is_used_instead_of_the_failed_legacy_trace() {
let result = dpi_hop::DpiHopProbeResult {
target: "203.0.113.10:443".parse().unwrap(),
local_addr: "192.0.2.10:45000".parse().unwrap(),
client_hello_bytes: 256,
max_icmp_time_exceeded_ttl: None,
hops: vec![dpi_hop::DpiHopProbeHop {
ttl: 1,
router: None,
outcome: dpi_hop::DpiHopProbeHopOutcome::Timeout,
}],
hops: vec![],
calibration: reports::probe::DpiCalibrationDebug {
method: "rst_client_hello".into(),
dpi_hop: Some(4),
last_clear_ttl: Some(3),
first_reset_ttl: Some(4),
..Default::default()
},
};
assert_eq!(dpi_hop_from_result(&result), Some(0));
assert_eq!(dpi_hop_from_result(&result), Some(4));
}
}

395
probe/src/packet_capture.rs Normal file
View file

@ -0,0 +1,395 @@
use etherparse::{LaxNetSlice, LaxSlicedPacket, TcpOptionElement, TcpSlice, TransportSlice};
use reports::probe::{TcpDiagnostics, TcpPacketMetadata};
use socket2::{Domain, Protocol, Socket, Type};
use std::io;
use std::mem::MaybeUninit;
use std::net::{IpAddr, SocketAddr};
use std::sync::{
Arc,
atomic::{AtomicBool, Ordering},
};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};
const MAX_PACKETS: usize = 32;
/// A separate reader observes TCP headers without consuming the stream's data.
/// It runs before connect so a SYN-ACK or an immediate reset is not missed.
pub struct TcpCapture {
started: Instant,
stop: Arc<AtomicBool>,
reader: Option<JoinHandle<TcpDiagnostics>>,
error: Option<String>,
}
impl TcpCapture {
pub fn start(target: SocketAddr) -> Self {
let started = Instant::now();
let stop = Arc::new(AtomicBool::new(false));
let socket = (|| {
let domain = if target.is_ipv4() {
Domain::IPV4
} else {
Domain::IPV6
};
let socket = Socket::new(domain, Type::RAW, Some(Protocol::TCP))?;
socket.set_read_timeout(Some(Duration::from_millis(20)))?;
Ok::<_, io::Error>(socket)
})();
match socket {
Ok(socket) => {
let reader_stop = stop.clone();
let reader = std::thread::spawn(move || {
let mut result = TcpDiagnostics::default();
let mut buffer = vec![MaybeUninit::<u8>::uninit(); 65535];
let mut shutdown_reads = 0;
loop {
if reader_stop.load(Ordering::Relaxed) {
shutdown_reads += 1;
if shutdown_reads > 256 {
result.truncated = true;
break;
}
if let Err(e) = socket.set_nonblocking(true) {
result.capture_error = Some(e.to_string());
break;
}
}
match socket.recv_from(&mut buffer) {
Ok((len, source)) => {
// SAFETY: recv_from initialized the returned prefix.
let bytes = unsafe {
std::slice::from_raw_parts(buffer.as_ptr().cast::<u8>(), len)
};
if let Some(packet) = parse_packet(
bytes,
source.as_socket(),
target,
started.elapsed().as_millis() as u64,
) {
// Filter to the exact local port(s) at finish. Bound the
// temporary queue too, in case other traces run concurrently.
if result.packets.len() < 256 {
result.packets.push(packet);
} else {
result.truncated = true;
}
}
}
Err(e)
if matches!(
e.kind(),
io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
) =>
{
if reader_stop.load(Ordering::Relaxed) {
break;
}
}
Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
Err(e) => {
result.capture_error = Some(e.to_string());
break;
}
}
}
result
});
Self {
started,
stop,
reader: Some(reader),
error: None,
}
}
Err(e) => Self {
started,
stop,
reader: None,
error: Some(e.to_string()),
},
}
}
pub fn elapsed_ms(&self) -> u64 {
self.started.elapsed().as_millis() as u64
}
pub fn finish(mut self, ports: &[u16]) -> TcpDiagnostics {
self.stop.store(true, Ordering::Relaxed);
let mut result = match self.reader.take() {
Some(reader) => reader.join().unwrap_or_else(|_| TcpDiagnostics {
capture_error: Some("TCP capture reader panicked".into()),
..Default::default()
}),
None => TcpDiagnostics {
capture_error: self.error.take(),
..Default::default()
},
};
result
.packets
.retain(|p| ports.contains(&p.destination.port()));
if result.packets.len() > MAX_PACKETS {
result.packets.truncate(MAX_PACKETS);
result.truncated = true;
}
result
}
}
/// Binding before connecting preserves the local port even when connect fails.
pub fn connect_with_port(
target: SocketAddr,
timeout: Duration,
) -> (Option<u16>, io::Result<std::net::TcpStream>) {
let mut port = None;
let result = (|| {
let domain = if target.is_ipv4() {
Domain::IPV4
} else {
Domain::IPV6
};
let socket = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))?;
let unspecified = if target.is_ipv4() {
IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED)
} else {
IpAddr::V6(std::net::Ipv6Addr::UNSPECIFIED)
};
socket.bind(&SocketAddr::new(unspecified, 0).into())?;
port = socket.local_addr()?.as_socket().map(|a| a.port());
socket.connect_timeout(&target.into(), timeout)?;
Ok(socket.into())
})();
(port, result)
}
impl Drop for TcpCapture {
fn drop(&mut self) {
self.stop.store(true, Ordering::Relaxed);
if let Some(reader) = self.reader.take() {
let _ = reader.join();
}
}
}
fn parse_packet(
bytes: &[u8],
source: Option<SocketAddr>,
target: SocketAddr,
observed_ms: u64,
) -> Option<TcpPacketMetadata> {
// Linux raw IPv6 sockets deliver the TCP segment without an IPv6 header.
// recv_from supplies its source; the destination IP is unavailable there.
let (src, dst, ttl, ip_id, tcp) =
if target.is_ipv6() && source.is_some_and(|s| s.ip() == target.ip()) {
match LaxSlicedPacket::from_ip(bytes) {
Ok(packet) if matches!(packet.net, Some(LaxNetSlice::Ipv6(_))) => ip_tcp(packet)?,
_ => (
target.ip(),
IpAddr::V6(std::net::Ipv6Addr::UNSPECIFIED),
None,
None,
TcpSlice::from_slice(bytes).ok()?,
),
}
} else {
ip_tcp(LaxSlicedPacket::from_ip(bytes).ok()?)?
};
if src != target.ip() || tcp.source_port() != target.port() {
return None;
}
let mut flags = Vec::new();
for (set, name) in [
(tcp.syn(), "SYN"),
(tcp.ack(), "ACK"),
(tcp.rst(), "RST"),
(tcp.fin(), "FIN"),
(tcp.psh(), "PSH"),
(tcp.urg(), "URG"),
(tcp.ece(), "ECE"),
(tcp.cwr(), "CWR"),
] {
if set {
flags.push(name.to_owned());
}
}
let mut timestamp = None;
let mut timestamp_echo = None;
for option in tcp.options_iterator().flatten() {
if let TcpOptionElement::Timestamp(value, echo) = option {
timestamp = Some(value);
timestamp_echo = Some(echo);
}
}
Some(TcpPacketMetadata {
observed_ms,
source: SocketAddr::new(src, tcp.source_port()),
destination: SocketAddr::new(dst, tcp.destination_port()),
ttl,
ip_id,
sequence: tcp.sequence_number(),
acknowledgment: tcp.acknowledgment_number(),
window: tcp.window_size(),
flags,
timestamp,
timestamp_echo,
options_hex: tcp.options().iter().map(|b| format!("{b:02x}")).collect(),
payload_bytes: tcp.payload().len(),
})
}
#[cfg(test)]
mod tests {
use super::*;
use etherparse::PacketBuilder;
#[test]
fn captures_reset_fingerprint_without_payload() {
let mut bytes = Vec::new();
PacketBuilder::ipv4([192, 0, 2, 1], [192, 0, 2, 2], 117)
.tcp(443, 45000, 1234, 2048)
.rst()
.ack(5678)
.options(&[TcpOptionElement::Timestamp(100, 200)])
.unwrap()
.write(&mut bytes, b"private payload")
.unwrap();
bytes[4..6].copy_from_slice(&42u16.to_be_bytes());
let packet = parse_packet(&bytes, None, "192.0.2.1:443".parse().unwrap(), 12).unwrap();
assert_eq!(packet.flags, ["ACK", "RST"]);
assert_eq!(packet.ttl, Some(117));
assert_eq!(packet.ip_id, Some(42));
assert_eq!(
(packet.sequence, packet.acknowledgment, packet.window),
(1234, 5678, 2048)
);
assert_eq!(
(packet.timestamp, packet.timestamp_echo),
(Some(100), Some(200))
);
assert_eq!(packet.payload_bytes, 15);
let json = serde_json::to_string(&packet).unwrap();
assert!(!json.contains("private payload"));
assert!(parse_packet(&bytes, None, "192.0.2.3:443".parse().unwrap(), 0).is_none());
assert!(parse_packet(&bytes, None, "192.0.2.1:444".parse().unwrap(), 0).is_none());
assert!(parse_packet(&bytes[..22], None, "192.0.2.1:443".parse().unwrap(), 0).is_none());
}
#[test]
fn captures_ipv6_fin_with_or_without_ip_header() {
let target: SocketAddr = "[2001:db8::1]:443".parse().unwrap();
let dst: std::net::Ipv6Addr = "2001:db8::2".parse().unwrap();
let IpAddr::V6(src) = target.ip() else {
unreachable!()
};
let mut bytes = Vec::new();
PacketBuilder::ipv6(src.octets(), dst.octets(), 55)
.tcp(443, 45000, 9, 123)
.fin()
.ack(10)
.write(&mut bytes, &[])
.unwrap();
let full = parse_packet(&bytes, Some(target), target, 20).unwrap();
assert_eq!(full.ttl, Some(55));
assert_eq!(full.destination.ip(), IpAddr::V6(dst));
assert_eq!(full.flags, ["ACK", "FIN"]);
let bare = parse_packet(&bytes[40..], Some(target), target, 21).unwrap();
assert_eq!(bare.ttl, None);
assert_eq!(bare.sequence, full.sequence);
assert!(bare.destination.ip().is_unspecified());
}
#[test]
fn legacy_hops_remain_compatible() {
let json = r#"{"ttl":1,"src":null,"outcome":"tcp_closed"}"#;
let hop: reports::probe::DpiProbeHop = serde_json::from_str(json).unwrap();
assert!(hop.tcp_diagnostics.is_none());
assert_eq!(serde_json::to_string(&hop).unwrap(), json);
let diagnostics = TcpDiagnostics {
packets: vec![],
capture_error: Some("permission denied".into()),
..Default::default()
};
let hop = reports::probe::DpiProbeHop {
tcp_diagnostics: Some(diagnostics),
..hop
};
let restored: reports::probe::DpiProbeHop =
serde_json::from_str(&serde_json::to_string(&hop).unwrap()).unwrap();
assert_eq!(restored.tcp_diagnostics, hop.tcp_diagnostics);
}
#[test]
#[ignore = "requires CAP_NET_RAW and local socket access"]
fn live_capture_records_syn_ack_and_reset() {
use std::io::{Read, Write};
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let target = listener.local_addr().unwrap();
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
stream.write_all(b"x").unwrap();
std::thread::sleep(Duration::from_millis(50));
socket2::SockRef::from(&stream)
.set_linger(Some(Duration::ZERO))
.unwrap();
});
let capture = TcpCapture::start(target);
let (port, connection) = connect_with_port(target, Duration::from_secs(1));
let mut stream = connection.unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(1)))
.unwrap();
let mut byte = [0];
stream.read_exact(&mut byte).unwrap();
assert_eq!(
stream.read(&mut byte).unwrap_err().kind(),
io::ErrorKind::ConnectionReset
);
server.join().unwrap();
let result = capture.finish(&[port.unwrap()]);
assert_eq!(result.capture_error, None);
assert!(
result
.packets
.iter()
.any(|p| p.flags.contains(&"SYN".into()) && p.flags.contains(&"ACK".into()))
);
assert!(
result
.packets
.iter()
.any(|p| p.flags.contains(&"RST".into()))
);
assert!(
result
.packets
.iter()
.all(|p| p.ttl.is_some() && p.source == target)
);
}
}
fn ip_tcp(
packet: LaxSlicedPacket<'_>,
) -> Option<(IpAddr, IpAddr, Option<u8>, Option<u16>, TcpSlice<'_>)> {
let (src, dst, ttl, id) = match packet.net? {
LaxNetSlice::Ipv4(ip) => (
IpAddr::V4(ip.header().source_addr()),
IpAddr::V4(ip.header().destination_addr()),
Some(ip.header().ttl()),
Some(ip.header().identification()),
),
LaxNetSlice::Ipv6(ip) => (
IpAddr::V6(ip.header().source_addr()),
IpAddr::V6(ip.header().destination_addr()),
Some(ip.header().hop_limit()),
None,
),
_ => return None,
};
let Some(TransportSlice::Tcp(tcp)) = packet.transport else {
return None;
};
Some((src, dst, ttl, id, tcp))
}

View file

@ -1,3 +1,4 @@
use crate::packet_capture::TcpCapture;
use etherparse::{
Icmpv4Type, Icmpv6Slice, Icmpv6Type, IpNumber, LaxNetSlice, LaxSlicedPacket, TransportSlice,
icmpv4, icmpv6,
@ -74,6 +75,7 @@ fn trace_all_blocking(
let receiver = Socket::new(domain, Type::RAW, Some(protocol))?;
let mut hops = Vec::new();
for ttl in 1..=max_hops {
let capture = TcpCapture::start(SocketAddr::new(target, HTTPS_PORT));
let attempts = connect_attempts(target, ttl, retries)?;
let response = if attempts.is_empty() {
HopResponse::Timeout
@ -95,6 +97,9 @@ fn trace_all_blocking(
address,
reverse_names: Vec::new(),
outcome,
tcp_diagnostics: Some(
capture.finish(&attempts.iter().map(|(_, port)| *port).collect::<Vec<_>>()),
),
});
if complete {
break;

View file

@ -227,11 +227,7 @@ fn detect_platform() -> Result<(PackageKind, String, bool)> {
.any(|line| line.split_whitespace().next() == Some("luci-app-cheburprobe"));
Ok((PackageKind::Opkg, architecture.to_owned(), luci_installed))
} else {
let architecture = match env::consts::ARCH {
"x86_64" => "amd64",
"aarch64" => "arm64",
architecture => bail!("unsupported standalone Linux architecture: {architecture}"),
};
let architecture = standalone_linux_architecture(env::consts::ARCH)?;
Ok((PackageKind::Linux, architecture.to_owned(), false))
}
@ -239,6 +235,16 @@ fn detect_platform() -> Result<(PackageKind, String, bool)> {
bail!("updates are not supported on this operating system")
}
#[cfg(target_os = "linux")]
fn standalone_linux_architecture(architecture: &str) -> Result<&'static str> {
match architecture {
"x86_64" => Ok("amd64"),
"aarch64" => Ok("arm64"),
"arm" => Ok("armv7"),
architecture => bail!("unsupported standalone Linux architecture: {architecture}"),
}
}
#[cfg(target_os = "linux")]
fn openwrt_apk_arch() -> Result<String> {
if let Ok(architecture) = std::fs::read_to_string("/etc/apk/arch") {

View file

@ -5,3 +5,6 @@ edition = "2024"
[dependencies]
serde = { workspace = true }
[dev-dependencies]
serde_json = "1.0"

View file

@ -26,6 +26,80 @@ pub struct DpiProbeHop {
#[serde(rename = "src")]
pub router: Option<IpAddr>,
pub outcome: DpiProbeHopOutcome,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tcp_diagnostics: Option<TcpDiagnostics>,
}
/// Incoming TCP headers observed during one manual hop attempt. No payload is retained.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
#[serde(default)]
pub struct TcpDiagnostics {
pub packets: Vec<TcpPacketMetadata>,
pub capture_error: Option<String>,
pub truncated: bool,
pub connect_error: Option<String>,
pub send_error: Option<String>,
pub client_hello_sent_ms: Option<u64>,
pub junk_sent_ms: Option<u64>,
/// Whether the complete low-TTL payload was accepted by the socket
#[serde(skip_serializing_if = "Option::is_none")]
pub payload_sent: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub peer_closed_before_payload: Option<bool>,
}
/// Optional command diagnostics.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct DpiCalibrationDebug {
pub method: String,
pub dpi_hop: Option<u8>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub fallback_reason: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub failure_reason: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub control_sni: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_clear_ttl: Option<u8>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub first_reset_ttl: Option<u8>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub first_drop_ttl: Option<u8>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub control_witness_ttl: Option<u8>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub attempts: Vec<DpiCalibrationAttempt>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DpiCalibrationAttempt {
pub ttl: u8,
pub control: bool,
pub outcome: String,
pub router: Option<IpAddr>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tcp_diagnostics: Option<TcpDiagnostics>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct TcpPacketMetadata {
/// Approximate observation time relative to the start of this hop's capture.
pub observed_ms: u64,
pub source: std::net::SocketAddr,
pub destination: std::net::SocketAddr,
/// Absent when the platform's raw IPv6 socket omits the IP header.
pub ttl: Option<u8>,
pub ip_id: Option<u16>,
pub sequence: u32,
pub acknowledgment: u32,
pub window: u16,
pub flags: Vec<String>,
pub timestamp: Option<u32>,
pub timestamp_echo: Option<u32>,
pub options_hex: String,
pub payload_bytes: usize,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
@ -224,6 +298,10 @@ pub enum ProbeCommandResult {
dpi_hop_v6: Option<u8>,
dpi_hops_v4: Vec<DpiProbeHop>,
dpi_hops_v6: Vec<DpiProbeHop>,
#[serde(default, skip_serializing_if = "Option::is_none")]
calibration_v4: Option<DpiCalibrationDebug>,
#[serde(default, skip_serializing_if = "Option::is_none")]
calibration_v6: Option<DpiCalibrationDebug>,
},
SniTraceroute {
host: String,
@ -231,6 +309,8 @@ pub enum ProbeCommandResult {
target: std::net::SocketAddr,
dpi_hop: Option<u8>,
hops: Vec<DpiProbeHop>,
#[serde(default, skip_serializing_if = "Option::is_none")]
calibration: Option<DpiCalibrationDebug>,
},
Error {
message: String,
@ -243,6 +323,8 @@ pub struct ManualTracerouteHop {
pub address: Option<IpAddr>,
pub reverse_names: Vec<String>,
pub outcome: ManualTracerouteOutcome,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tcp_diagnostics: Option<TcpDiagnostics>,
}
#[derive(Clone, Copy, Serialize, Deserialize)]
@ -310,6 +392,74 @@ pub fn validate_sni_traceroute(host: &str, sni: &str) -> Result<(), &'static str
mod command_tests {
use super::*;
#[test]
fn old_probe_results_do_not_require_calibration_debug_fields() {
for wire in [
r#"{"type":"sni_traceroute","host":"example.com","sni":"instagram.com","target":"192.0.2.1:443","dpi_hop":2,"hops":[]}"#,
r#"{"type":"remeasure_dpi_hop","dpi_hop_v4":2,"dpi_hop_v6":null,"dpi_hops_v4":[],"dpi_hops_v6":[]}"#,
] {
let result: ProbeCommandResult = serde_json::from_str(wire).unwrap();
// Adding optional diagnostics must not change a legacy response.
assert_eq!(
serde_json::to_value(result).unwrap(),
serde_json::from_str::<serde_json::Value>(wire).unwrap()
);
}
}
#[test]
fn new_results_keep_the_legacy_envelope_and_commands() {
#[derive(Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum LegacyResult {
SniTraceroute {
host: String,
sni: String,
target: std::net::SocketAddr,
dpi_hop: Option<u8>,
hops: Vec<DpiProbeHop>,
},
}
let wire = serde_json::to_value(ProbeCommandResult::SniTraceroute {
host: "example.com".into(),
sni: "instagram.com".into(),
target: "192.0.2.1:443".parse().unwrap(),
dpi_hop: Some(4),
hops: vec![],
calibration: Some(DpiCalibrationDebug {
method: "rst_client_hello".into(),
dpi_hop: Some(4),
..Default::default()
}),
})
.unwrap();
let LegacyResult::SniTraceroute {
host,
sni,
target,
dpi_hop,
hops,
} = serde_json::from_value(wire).unwrap();
assert_eq!(host, "example.com");
assert_eq!(sni, "instagram.com");
assert_eq!(target.port(), 443);
assert_eq!(dpi_hop, Some(4));
assert!(hops.is_empty());
assert_eq!(
serde_json::to_value(ProbeCommand::RemeasureDpiHop).unwrap(),
serde_json::json!({"type":"remeasure_dpi_hop"})
);
assert_eq!(
serde_json::to_value(ProbeCommand::SniTraceroute {
host: "example.com".into(),
sni: "instagram.com".into(),
max_hops: 15
})
.unwrap(),
serde_json::json!({"type":"sni_traceroute","host":"example.com","sni":"instagram.com","max_hops":15})
);
}
#[test]
fn validates_separate_host_and_sni() {
for host in ["example.com", "192.0.2.1", "2001:db8::1"] {

View file

@ -1,6 +1,6 @@
[package]
name = "website"
version = "1.4.4"
version = "1.5.2"
edition = "2024"
[dependencies]
@ -9,6 +9,9 @@ semver = "1.0"
rocket = { version = "0.5.1", features = ["msgpack", "json"] }
rocket_dyn_templates = { version = "0.2.0", features = ["tera"] }
sqlx = { version = "0.8.6", features = ["runtime-tokio", "postgres", "migrate", "chrono", "uuid"] }
klickhouse = { version = "0.15.3", default-features = false, features = ["compression", "derive", "refinery"] }
deadpool = { version = "0.13.1", default-features = false, features = ["managed"] }
refinery = { version = "0.9.2", default-features = false }
rocket-client-addr = "0.5.4"
serde = { workspace = true }
reports = { path = "../reports" }
@ -23,3 +26,5 @@ toml = "0.8"
uuid = { version = "1", features = ["v4"] }
reqwest = { workspace = true }
rand = "0.10.3"
url = "2"
percent-encoding = "2"

48
website/README.md Normal file
View file

@ -0,0 +1,48 @@
# Website API
The API uses PostgreSQL for operational data and ClickHouse for query, probe,
and feedback records. `CLICKHOUSE_URL` must identify an existing database over
native TCP, for example `clickhouse://user:password@host:9000/analytics`.
Percent-encode special characters in credentials. Add
`?stats_timezone=Asia%2FYekaterinburg` to change the complaint statistics timezone
from UTC.
## ClickHouse connection pool
All API requests share a Deadpool managed pool of independent native connections.
A small adapter creates and checks `klickhouse` connections; Deadpool handles pool
capacity, waiters, and connection reuse. Connections are opened on demand and
checked out exclusively for each operation.
| Environment variable | Default | Purpose |
| --- | --- | --- |
| `CLICKHOUSE_POOL_SIZE` | `4` | Maximum simultaneous connections per API process. |
| `CLICKHOUSE_ACQUIRE_TIMEOUT_SECONDS` | `5` | Deadline for waiting for a slot and establishing its connection. |
| `CLICKHOUSE_QUERY_TIMEOUT_SECONDS` | `10` | Deadline for each API read or write. |
Values must be positive integers. Docker Compose forwards these settings to the
website service. Connection setup also has a ten-second upper limit, bounded by
the acquisition deadline. Startup migrations have a separate 120-second deadline.
Cancelled, timed-out, and broken connections are closed and replaced on demand.
Reads retry once after a connection failure, releasing the failed connection
before acquiring another slot. Query timeouts are returned immediately. Writes
are never automatically retried because a failed response may follow a committed
insert.
## Tests
Run the regular website tests with the checked-in SQLx metadata:
```sh
SQLX_OFFLINE=true cargo test --locked -p website
```
The ClickHouse integration tests require a disposable native server. They create
and delete a temporary database, and the migration test needs database creation
permissions:
```sh
SQLX_OFFLINE=true CLICKHOUSE_TEST_URL=clickhouse://localhost:9000/default \
cargo test --locked -p website analytics::tests -- --include-ignored
```

View file

@ -1,5 +1,7 @@
fn main() -> Result<(), Box<dyn std::error::Error>> {
println!("cargo:rerun-if-changed=migrations");
println!("cargo:rerun-if-changed=clickhouse/migrations");
println!("cargo:rerun-if-changed=clickhouse/queries");
println!("cargo:rerun-if-changed=build.rs");
Ok(())

View file

@ -0,0 +1,4 @@
-- Removes only the ClickHouse mappings; the PostgreSQL tables are untouched.
DROP TABLE IF EXISTS backfill_pg_human_reports;
DROP TABLE IF EXISTS backfill_pg_probe_reports;
DROP TABLE IF EXISTS backfill_pg_queries;

View file

@ -0,0 +1,11 @@
INSERT INTO human_reports
(query_id, query, source_ip, date, works)
SELECT
h.id, q.query, h.source_ip,
if(isNull(h.date), NULL,
toTimeZone(parseDateTime64BestEffort(ifNull(h.date, '1970-01-01 00:00:00'),
6, {source_timezone:String}), 'UTC')),
h.works
FROM backfill_pg_human_reports AS h
INNER JOIN queries AS q ON q.id = h.id
SETTINGS external_table_functions_use_nulls = 1;

View file

@ -0,0 +1,12 @@
INSERT INTO probe_reports
(postgres_id, query_id, query, probe_id, date, verdicts, result,
target_hop_count, target_trace_result, duration_ms)
SELECT
p.id, p.query_id, q.query, p.probe_id,
if(isNull(p.date), NULL,
toTimeZone(parseDateTime64BestEffort(ifNull(p.date, '1970-01-01 00:00:00'),
6, {source_timezone:String}), 'UTC')),
p.verdicts, p.result, p.target_hop_count, p.target_trace_result, p.duration_ms
FROM backfill_pg_probe_reports AS p
INNER JOIN queries AS q ON q.id = p.query_id
SETTINGS external_table_functions_use_nulls = 1;

View file

@ -0,0 +1,13 @@
INSERT INTO queries
(id, query, source_ip, source_country_code, source_city_geo_name_id,
target_country_code, target_asn, target_provider, resolved_ips,
cdn_networks, cdn_providers, rkn_domain, date)
SELECT
id, query, source_ip, source_country_code, source_city_geo_name_id,
target_country_code, target_asn, target_provider, resolved_ips,
cdn_networks, cdn_providers, rkn_domain,
if(isNull(date), NULL,
toTimeZone(parseDateTime64BestEffort(ifNull(date, '1970-01-01 00:00:00'),
6, {source_timezone:String}), 'UTC'))
FROM backfill_pg_queries
SETTINGS external_table_functions_use_nulls = 1;

View file

@ -0,0 +1,44 @@
-- Configure the backfill_postgres named collection on the ClickHouse server first.
-- These are remote table mappings, not copies. Never INSERT into them.
-- Read timestamps as text so PostgreSQL TIMESTAMP wall-clock values can be
-- interpreted explicitly in source_timezone without losing microseconds.
CREATE TABLE backfill_pg_queries
(
id UUID,
query String,
source_ip String,
source_country_code Nullable(String),
source_city_geo_name_id Nullable(Int32),
target_country_code Nullable(String),
target_asn Nullable(String),
target_provider Nullable(String),
resolved_ips Array(Nullable(String)),
cdn_networks Array(Nullable(String)),
cdn_providers Array(Nullable(String)),
rkn_domain Nullable(String),
date Nullable(String)
)
ENGINE = PostgreSQL(backfill_postgres, table = 'queries');
CREATE TABLE backfill_pg_probe_reports
(
id Int64,
query_id UUID,
probe_id Int32,
date Nullable(String),
verdicts Array(Nullable(String)),
result String,
target_hop_count Nullable(Int16),
target_trace_result Nullable(String),
duration_ms Nullable(Int64)
)
ENGINE = PostgreSQL(backfill_postgres, table = 'probe_reports');
CREATE TABLE backfill_pg_human_reports
(
id UUID,
source_ip String,
date Nullable(String),
works Nullable(Bool)
)
ENGINE = PostgreSQL(backfill_postgres, table = 'human_reports');

View file

@ -0,0 +1,63 @@
CREATE TABLE IF NOT EXISTS queries
(
id UUID DEFAULT generateUUIDv4(),
query String,
query_lower String MATERIALIZED lowerUTF8(query),
source_ip String,
source_country_code Nullable(String),
source_city_geo_name_id Nullable(Int32),
target_country_code Nullable(String),
target_asn Nullable(String),
target_provider Nullable(String),
resolved_ips Array(Nullable(String)),
cdn_networks Array(Nullable(String)),
cdn_providers Array(Nullable(String)),
rkn_domain Nullable(String),
date Nullable(DateTime64(6, 'UTC')) DEFAULT now64(6),
PROJECTION by_id (SELECT * ORDER BY id),
PROJECTION by_date (SELECT * ORDER BY (ifNull(date, toDateTime64(0, 6, 'UTC')), id))
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(ifNull(date, toDateTime64(0, 6, 'UTC')))
ORDER BY (query_lower, ifNull(date, toDateTime64(0, 6, 'UTC')), id);
CREATE TABLE IF NOT EXISTS probe_reports
(
id UUID DEFAULT generateUUIDv4(),
postgres_id Nullable(Int64),
query_id UUID,
query String,
query_lower String MATERIALIZED lowerUTF8(query),
probe_id Int32,
date Nullable(DateTime64(6, 'UTC')) DEFAULT now64(6),
verdicts Array(Nullable(String)),
result String CODEC(ZSTD(3)),
target_hop_count Nullable(Int16),
target_trace_result Nullable(String),
duration_ms Nullable(Int64),
INDEX query_id_idx query_id TYPE bloom_filter(0.01) GRANULARITY 1,
PROJECTION by_probe
(
SELECT probe_id, date, query_id, query_lower, verdicts,
target_hop_count, target_trace_result, duration_ms
ORDER BY (probe_id, ifNull(date, toDateTime64(0, 6, 'UTC')), query_id)
)
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(ifNull(date, toDateTime64(0, 6, 'UTC')))
ORDER BY (query_lower, ifNull(date, toDateTime64(0, 6, 'UTC')), probe_id, query_id);
CREATE TABLE IF NOT EXISTS human_reports
(
id UUID DEFAULT generateUUIDv4(),
query_id UUID,
query String,
query_lower String MATERIALIZED lowerUTF8(query),
source_ip String,
date Nullable(DateTime64(6, 'UTC')) DEFAULT now64(6),
works Nullable(Bool),
INDEX query_id_idx query_id TYPE bloom_filter(0.01) GRANULARITY 1
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(ifNull(date, toDateTime64(0, 6, 'UTC')))
ORDER BY (query_lower, ifNull(date, toDateTime64(0, 6, 'UTC')), query_id, id);

View file

@ -0,0 +1,16 @@
WITH toDate(now(), $2) AS today_day
SELECT toString(days.day) AS date, toInt64(ifNull(reports.count, 0)) AS count
FROM (
SELECT today_day - 13 + toInt32(number) AS day
FROM numbers(14)
) AS days
LEFT JOIN (
SELECT toDate(date, $2) AS day, uniqExact(source_ip) AS count
FROM human_reports
WHERE date >= toDateTime(today_day - 13, $2)
AND date < toDateTime(today_day + 1, $2)
AND works = false
AND query_lower = lowerUTF8($1)
GROUP BY day
) AS reports ON reports.day = days.day
ORDER BY days.day

View file

@ -0,0 +1,4 @@
INSERT INTO probe_reports (
query_id, query, probe_id, verdicts, result,
target_hop_count, target_trace_result, duration_ms
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)

View file

@ -0,0 +1,5 @@
INSERT INTO queries (
id, query, source_ip, source_country_code, source_city_geo_name_id,
target_country_code, target_asn, target_provider, resolved_ips,
cdn_networks, cdn_providers, rkn_domain
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)

View file

@ -0,0 +1,4 @@
SELECT query, resolved_ips
FROM queries
WHERE id = $1
LIMIT 1

860
website/src/analytics.rs Normal file
View file

@ -0,0 +1,860 @@
use anyhow::{Context, Result};
use deadpool::managed::{Manager, Metrics, Object, Pool, RecycleError, RecycleResult};
use klickhouse::{Client, ClientOptions, KlickhouseError, QueryBuilder, Row};
use rocket::tokio::sync::Semaphore;
use rocket::tokio::time::timeout;
use std::sync::Arc;
use std::time::Duration;
use uuid::Uuid;
mod embedded {
refinery::embed_migrations!("clickhouse/migrations");
}
const COMPLAINTS_SQL: &str = include_str!("../clickhouse/queries/complaints.sql");
const QUERY_SQL: &str = include_str!("../clickhouse/queries/query_by_id.sql");
pub const INSERT_QUERY_SQL: &str = include_str!("../clickhouse/queries/insert_query.sql");
pub const INSERT_PROBE_REPORT_SQL: &str =
include_str!("../clickhouse/queries/insert_probe_report.sql");
struct ConnectionConfig {
host: String,
port: u16,
options: ClientOptions,
stats_timezone: String,
}
impl ConnectionConfig {
fn parse(uri: &str) -> Result<Self> {
let uri = url::Url::parse(uri).context("invalid ClickHouse URI")?;
anyhow::ensure!(
uri.scheme() == "clickhouse",
"ClickHouse URI must use clickhouse:// for native TCP"
);
anyhow::ensure!(
uri.fragment().is_none(),
"ClickHouse URI must not contain a fragment"
);
let host = match uri.host().context("ClickHouse URI must include a host")? {
url::Host::Domain(host) => host.to_owned(),
url::Host::Ipv4(host) => host.to_string(),
url::Host::Ipv6(host) => host.to_string(),
};
let database = uri.path().strip_prefix('/').unwrap_or(uri.path());
anyhow::ensure!(
!database.is_empty() && !database.contains('/'),
"ClickHouse URI must include one database path segment"
);
let decode = |value: &str| -> Result<String> {
Ok(percent_encoding::percent_decode_str(value)
.decode_utf8()
.context("ClickHouse URI contains invalid UTF-8")?
.into_owned())
};
let mut stats_timezone = None;
for (name, value) in uri.query_pairs() {
anyhow::ensure!(
name == "stats_timezone",
"unsupported ClickHouse URI query parameter"
);
anyhow::ensure!(
stats_timezone.is_none() && !value.is_empty(),
"specify stats_timezone once with a nonempty value"
);
stats_timezone = Some(value.into_owned());
}
Ok(Self {
host,
port: uri.port().unwrap_or(9000),
options: ClientOptions {
username: if uri.username().is_empty() {
"default".into()
} else {
decode(uri.username())?
},
password: decode(uri.password().unwrap_or_default())?,
default_database: decode(database)?,
..Default::default()
},
stats_timezone: stats_timezone.unwrap_or_else(|| "UTC".into()),
})
}
async fn connect(&self) -> Result<(Connection, String)> {
let connection = timeout(Duration::from_secs(10), async {
let stream =
rocket::tokio::net::TcpStream::connect((self.host.as_str(), self.port)).await?;
stream.set_nodelay(self.options.tcp_nodelay)?;
let stream = stream.into_std()?;
// Retain a shutdown handle before starting the driver's background task.
// Its initial handshake does not observe dropped Client handles.
let socket = SocketGuard(stream.try_clone()?);
let stream = rocket::tokio::net::TcpStream::from_std(stream)?;
let (reader, writer) = stream.into_split();
let client = Client::connect_stream(reader, writer, self.options.clone()).await?;
anyhow::ensure!(
!client.is_closed(),
"ClickHouse connection closed during setup"
);
Ok::<_, anyhow::Error>(Connection {
client,
_socket: socket,
})
})
.await
.context("ClickHouse connection timed out")?
.context("failed to connect to ClickHouse")?;
Ok((connection, self.stats_timezone.clone()))
}
}
struct SocketGuard(std::net::TcpStream);
impl Drop for SocketGuard {
fn drop(&mut self) {
let _ = self.0.shutdown(std::net::Shutdown::Both);
}
}
struct Connection {
client: Client,
_socket: SocketGuard,
}
impl std::ops::Deref for Connection {
type Target = Client;
fn deref(&self) -> &Client {
&self.client
}
}
#[derive(Clone, Copy)]
struct PoolSettings {
size: usize,
acquire_timeout: Duration,
query_timeout: Duration,
}
impl Default for PoolSettings {
fn default() -> Self {
Self {
size: 4,
acquire_timeout: Duration::from_secs(5),
query_timeout: Duration::from_secs(10),
}
}
}
impl PoolSettings {
fn from_env() -> Result<Self> {
fn positive(name: &str, default: usize) -> Result<usize> {
match std::env::var(name) {
Ok(value) => {
let value = value
.parse::<usize>()
.with_context(|| format!("invalid {name}"))?;
anyhow::ensure!(value > 0, "{name} must be positive");
Ok(value)
}
Err(std::env::VarError::NotPresent) => Ok(default),
Err(error) => Err(error).with_context(|| format!("invalid {name}")),
}
}
let defaults = Self::default();
let size = positive("CLICKHOUSE_POOL_SIZE", defaults.size)?;
anyhow::ensure!(
size <= Semaphore::MAX_PERMITS,
"CLICKHOUSE_POOL_SIZE is too large"
);
Ok(Self {
size,
acquire_timeout: Duration::from_secs(
positive("CLICKHOUSE_ACQUIRE_TIMEOUT_SECONDS", 5)? as u64,
),
query_timeout: Duration::from_secs(
positive("CLICKHOUSE_QUERY_TIMEOUT_SECONDS", 10)? as u64
),
})
}
}
struct ConnectionManager {
config: ConnectionConfig,
}
impl Manager for ConnectionManager {
type Type = Connection;
type Error = KlickhouseError;
async fn create(&self) -> klickhouse::Result<Connection> {
self.config
.connect()
.await
.map(|(connection, _)| connection)
.map_err(|error| {
KlickhouseError::ProtocolError(format!(
"failed to connect to ClickHouse: {error:#}"
))
})
}
async fn recycle(
&self,
connection: &mut Connection,
_: &Metrics,
) -> RecycleResult<KlickhouseError> {
if connection.is_closed() {
Err(RecycleError::message("ClickHouse connection closed"))
} else {
Ok(())
}
}
}
struct ConnectionPool {
inner: Pool<ConnectionManager>,
settings: PoolSettings,
}
// A lease starts non-reusable: cancellation or a timeout must close the connection,
// since the driver's query may still be running. Only a completed operation can
// return it to the pool. The socket guard explicitly closes discarded connections.
struct ConnectionLease {
client: Option<Object<ConnectionManager>>,
reusable: bool,
}
impl ConnectionLease {
fn client(&self) -> &Client {
self.client.as_ref().expect("connection lease has a client")
}
fn finish<T>(&mut self, result: klickhouse::Result<T>) -> klickhouse::Result<T> {
// A decoding error can leave unread blocks in flight. Only success or a
// server exception establishes that the operation has finished safely.
self.reusable = !self.client().is_closed()
&& matches!(
&result,
Ok(_) | Err(KlickhouseError::ServerException { .. })
);
result.and_then(|value| {
if self.client().is_closed() {
Err(KlickhouseError::ProtocolError(
"ClickHouse connection closed while executing query".into(),
))
} else {
Ok(value)
}
})
}
}
impl Drop for ConnectionLease {
fn drop(&mut self) {
if !self.reusable {
if let Some(connection) = self.client.take() {
// Remove it permanently: returning a timed-out operation to the
// pool would queue later requests behind the unfinished query.
drop(Object::take(connection));
}
}
}
}
impl ConnectionPool {
async fn acquire(&self) -> klickhouse::Result<ConnectionLease> {
let client = timeout(self.settings.acquire_timeout, self.inner.get())
.await
.map_err(|_| {
KlickhouseError::ProtocolError("ClickHouse pool acquisition timed out".into())
})?
.map_err(|error| KlickhouseError::ProtocolError(format!("ClickHouse pool: {error}")))?;
Ok(ConnectionLease {
client: Some(client),
reusable: false,
})
}
}
#[derive(Clone)]
pub struct Analytics {
pool: Arc<ConnectionPool>,
stats_timezone: String,
}
#[derive(klickhouse::Row)]
pub struct StoredQuery {
pub query: String,
pub resolved_ips: Vec<Option<String>>,
}
#[derive(klickhouse::Row, serde::Serialize, Debug)]
pub struct ComplaintDay {
pub date: String,
pub count: i64,
}
impl Analytics {
pub async fn connect_from_env() -> Result<Self> {
let uri = std::env::var("CLICKHOUSE_URL")
.context("CLICKHOUSE_URL must specify clickhouse://user:password@host:9000/database")?;
let config = ConnectionConfig::parse(&uri)?;
let settings = PoolSettings::from_env()?;
let analytics = Self::new(config, settings)?;
analytics
.execute(
QueryBuilder::new("SELECT toDate(now(), $1)").arg(analytics.stats_timezone.clone()),
)
.await
.context("failed to validate ClickHouse stats_timezone")?;
Ok(analytics)
}
fn new(config: ConnectionConfig, settings: PoolSettings) -> Result<Self> {
let stats_timezone = config.stats_timezone.clone();
let inner = Pool::builder(ConnectionManager { config })
.max_size(settings.size)
.build()
.context("failed to build ClickHouse pool")?;
Ok(Self {
pool: Arc::new(ConnectionPool { inner, settings }),
stats_timezone,
})
}
pub async fn migrate(&self) -> Result<()> {
let mut connection = self.pool.acquire().await?;
// Migrations may contain multiple DDL statements and need a longer deadline.
let result = timeout(
Duration::from_secs(120),
embedded::migrations::runner()
.run_async(&mut connection.client.as_mut().unwrap().client),
)
.await
.context("ClickHouse migrations timed out")?;
connection.reusable = result.is_ok() && !connection.client().is_closed();
result.context("failed to apply embedded ClickHouse migrations")?;
anyhow::ensure!(
connection.reusable,
"ClickHouse connection closed during migrations"
);
Ok(())
}
async fn read<T: Row>(&self, query: QueryBuilder<'_>) -> klickhouse::Result<Vec<T>> {
let query = query.finalize()?.to_string();
for attempt in 0..2 {
let mut connection = self.pool.acquire().await?;
let result = timeout(
self.pool.settings.query_timeout,
connection.client().query_collect(query.clone()),
)
.await
.map_err(|_| KlickhouseError::ProtocolError("ClickHouse query timed out".into()))?;
let result = connection.finish(result);
if attempt == 0
&& matches!(
&result,
Err(KlickhouseError::Io(_) | KlickhouseError::ProtocolError(_))
)
{
// Release the failed lease before acquiring a replacement (also for size 1).
drop(connection);
continue;
}
return result;
}
unreachable!("read makes at most two attempts")
}
// A failed insert can already have committed. Discard its connection without
// automatically replaying an ambiguous write.
pub async fn execute(&self, query: QueryBuilder<'_>) -> klickhouse::Result<()> {
let mut connection = self.pool.acquire().await?;
let result = timeout(
self.pool.settings.query_timeout,
connection.client().execute(query),
)
.await
.map_err(|_| KlickhouseError::ProtocolError("ClickHouse query timed out".into()))?;
connection.finish(result)
}
pub async fn query_by_id(&self, id: Uuid) -> klickhouse::Result<Option<StoredQuery>> {
Ok(self.read(QueryBuilder::new(QUERY_SQL).arg(id)).await?.pop())
}
pub async fn complaints(&self, target: &str) -> klickhouse::Result<Vec<ComplaintDay>> {
self.read(
QueryBuilder::new(COMPLAINTS_SQL)
.arg(target)
.arg(self.stats_timezone.clone()),
)
.await
}
pub async fn feedback(
&self,
id: Uuid,
query: &str,
source_ip: &str,
works: bool,
) -> klickhouse::Result<()> {
self.execute(
QueryBuilder::new(
"INSERT INTO human_reports (query_id, query, source_ip, works) VALUES ($1, $2, $3, $4)",
)
.arg(id)
.arg(query)
.arg(source_ip)
.arg(works),
)
.await
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_clickhouse_uri_defaults() {
let config = ConnectionConfig::parse("clickhouse://localhost/analytics").unwrap();
assert_eq!(config.host, "localhost");
assert_eq!(config.port, 9000);
assert_eq!(config.options.username, "default");
assert_eq!(config.options.password, "");
assert_eq!(config.options.default_database, "analytics");
assert_eq!(config.stats_timezone, "UTC");
}
#[test]
fn parses_encoded_credentials_ipv6_and_timezone() {
let config = ConnectionConfig::parse(
"clickhouse://user%40team:p%40ss%3A%2F%23%25@[::1]:19000/analytics%20db?stats_timezone=Asia%2FYekaterinburg",
).unwrap();
assert_eq!(config.host, "::1");
assert_eq!(config.port, 19000);
assert_eq!(config.options.username, "user@team");
assert_eq!(config.options.password, "p@ss:/#%");
assert_eq!(config.options.default_database, "analytics db");
assert_eq!(config.stats_timezone, "Asia/Yekaterinburg");
}
#[test]
fn rejects_invalid_clickhouse_uris_without_disclosing_credentials() {
for uri in [
"clickhouse://user:secret@localhost",
"clickhouse://user:secret@localhost/a/b",
"http://user:secret@localhost/analytics",
"clickhouses://user:secret@localhost/analytics",
"clickhouse://user:secret@localhost:bad/analytics",
"clickhouse://user:secret@localhost/analytics#fragment",
"clickhouse://user:secret@localhost/analytics?unknown=value",
"clickhouse://user:secret@localhost/analytics?stats_timezone=",
"clickhouse://user:secret@localhost/analytics?stats_timezone=UTC&stats_timezone=UTC",
"clickhouse://user:secret@localhost/%FF",
] {
let error = ConnectionConfig::parse(uri)
.err()
.expect("URI must be rejected");
assert!(!format!("{error:#}").contains("secret"));
}
}
#[derive(klickhouse::Row, Debug)]
struct Count {
count: u64,
}
#[derive(klickhouse::Row)]
struct StoredProbe {
query_id: Uuid,
result: String,
verdicts: Vec<Option<String>>,
duration_ms: Option<i64>,
}
async fn test_analytics(settings: PoolSettings) -> Result<Analytics> {
let config = ConnectionConfig::parse(&std::env::var("CLICKHOUSE_TEST_URL")?)?;
let analytics = Analytics::new(config, settings)?;
analytics.execute(QueryBuilder::new("SELECT 1")).await?;
Ok(analytics)
}
#[rocket::async_test]
#[ignore = "requires CLICKHOUSE_TEST_URL pointing to a disposable native ClickHouse server"]
async fn pool_is_bounded_and_shared_between_clones() -> Result<()> {
let analytics = test_analytics(PoolSettings {
size: 2,
acquire_timeout: Duration::from_millis(100),
..PoolSettings::default()
})
.await?;
let clone = analytics.clone();
let mut first = analytics.pool.acquire().await?;
let mut second = clone.pool.acquire().await?;
assert_eq!(analytics.pool.inner.status().available, 0);
assert!(analytics.pool.acquire().await.is_err());
assert_eq!(analytics.pool.inner.status().available, 0);
// Separate native connections have distinct TCP source addresses.
let first_address = first.client.as_ref().unwrap()._socket.0.local_addr()?;
let second_address = second.client.as_ref().unwrap()._socket.0.local_addr()?;
assert_ne!(first_address, second_address);
let result = first.client().execute("SELECT 1").await;
first.finish(result)?;
let result = second.client().execute("SELECT 1").await;
second.finish(result)?;
drop(first);
let mut reused = clone.pool.acquire().await?;
let reused_address = reused.client.as_ref().unwrap()._socket.0.local_addr()?;
assert_eq!(first_address, reused_address);
let result = reused.client().execute("SELECT 1").await;
reused.finish(result)?;
drop(reused);
drop(second);
assert_eq!(analytics.pool.inner.status().available, 2);
assert_eq!(analytics.pool.inner.status().size, 2);
Ok(())
}
#[rocket::async_test]
#[ignore = "requires CLICKHOUSE_TEST_URL pointing to a disposable native ClickHouse server"]
async fn slow_query_does_not_block_other_connections() -> Result<()> {
let analytics = test_analytics(PoolSettings {
size: 2,
..PoolSettings::default()
})
.await?;
let mut slow = analytics.pool.acquire().await?;
let (slow_result, fast_result) = rocket::tokio::join!(
slow.client().execute("SELECT sleep(1)"),
timeout(
Duration::from_millis(500),
analytics.read::<Count>(QueryBuilder::new("SELECT toUInt64(1) AS count"))
),
);
slow.finish(slow_result)?;
assert_eq!(fast_result??[0].count, 1);
Ok(())
}
#[rocket::async_test]
#[ignore = "requires CLICKHOUSE_TEST_URL pointing to a disposable native ClickHouse server"]
async fn query_timeouts_discard_connections_and_release_capacity() -> Result<()> {
let analytics = test_analytics(PoolSettings {
size: 1,
query_timeout: Duration::from_millis(100),
..PoolSettings::default()
})
.await?;
assert!(
analytics
.execute(QueryBuilder::new("SELECT sleep(1)"))
.await
.unwrap_err()
.to_string()
.contains("query timed out")
);
assert_eq!(analytics.pool.inner.status().available, 0);
assert_eq!(analytics.pool.inner.status().size, 0);
// Recovery must use a new connection rather than queue behind the slow query.
let rows: Vec<Count> = analytics
.read(QueryBuilder::new("SELECT toUInt64(1) AS count"))
.await?;
assert_eq!(rows[0].count, 1);
assert!(
analytics
.read::<Count>(QueryBuilder::new(
"SELECT toUInt64(1) AS count WHERE sleep(1) = 0"
))
.await
.unwrap_err()
.to_string()
.contains("query timed out")
);
assert_eq!(analytics.pool.inner.status().available, 0);
assert_eq!(analytics.pool.inner.status().size, 0);
analytics.execute(QueryBuilder::new("SELECT 1")).await?;
Ok(())
}
#[rocket::async_test]
#[ignore = "requires CLICKHOUSE_TEST_URL pointing to a disposable native ClickHouse server"]
async fn cancellation_discards_connection_and_releases_capacity() -> Result<()> {
let analytics = test_analytics(PoolSettings {
size: 1,
..PoolSettings::default()
})
.await?;
// Cancelling a future with a lease must not return an in-flight query to the pool.
assert!(
timeout(
Duration::from_millis(100),
analytics.execute(QueryBuilder::new("SELECT sleep(1)")),
)
.await
.is_err()
);
assert_eq!(analytics.pool.inner.status().available, 0);
assert_eq!(analytics.pool.inner.status().size, 0);
let rows: Vec<Count> = analytics
.read(QueryBuilder::new("SELECT toUInt64(1) AS count"))
.await?;
assert_eq!(rows[0].count, 1);
Ok(())
}
#[rocket::async_test]
#[ignore = "requires CLICKHOUSE_TEST_URL pointing to a disposable native ClickHouse server"]
async fn stalled_handshake_times_out_and_closes_socket() -> Result<()> {
use rocket::tokio::{io::AsyncReadExt, net::TcpListener};
let mut config = ConnectionConfig::parse(&std::env::var("CLICKHOUSE_TEST_URL")?)?;
let listener = TcpListener::bind("127.0.0.1:0").await?;
let address = listener.local_addr()?;
config.host = address.ip().to_string();
config.port = address.port();
let analytics = Analytics::new(
config,
PoolSettings {
size: 1,
acquire_timeout: Duration::from_millis(100),
..PoolSettings::default()
},
)?;
let peer = rocket::tokio::spawn(async move {
let (mut socket, _) = listener.accept().await?;
// Accept TCP but never send the ClickHouse hello. Cancellation must
// close the socket even though the driver's handshake is still waiting.
let mut received = Vec::new();
socket.read_to_end(&mut received).await?;
Ok::<_, std::io::Error>(received)
});
let error = analytics
.pool
.acquire()
.await
.err()
.expect("handshake must time out");
assert!(error.to_string().contains("acquisition timed out"));
assert_eq!(analytics.pool.inner.status().size, 0);
assert!(!timeout(Duration::from_secs(2), peer).await???.is_empty());
Ok(())
}
// A single-connection TCP proxy lets the test simulate a server outage
// without stopping or modifying the disposable ClickHouse server itself.
fn proxy(
listener: rocket::tokio::net::TcpListener,
upstream: (String, u16),
drop_read_once: bool,
) -> rocket::tokio::task::JoinHandle<()> {
rocket::tokio::spawn(async move {
if drop_read_once {
use rocket::tokio::io::{AsyncReadExt, AsyncWriteExt};
let (mut downstream, _) = listener.accept().await.unwrap();
let mut upstream_socket = rocket::tokio::net::TcpStream::connect(upstream.clone())
.await
.unwrap();
let mut received = Vec::new();
let mut down_buffer = [0; 4096];
let mut up_buffer = [0; 4096];
loop {
rocket::tokio::select! {
bytes = downstream.read(&mut down_buffer) => {
let bytes = bytes.unwrap();
if bytes == 0 { break; }
received.extend_from_slice(&down_buffer[..bytes]);
if received.windows(b"disconnect_test_marker".len())
.any(|part| part == b"disconnect_test_marker") {
break;
}
upstream_socket.write_all(&down_buffer[..bytes]).await.unwrap();
}
bytes = upstream_socket.read(&mut up_buffer) => {
let bytes = bytes.unwrap();
if bytes == 0 { break; }
downstream.write_all(&up_buffer[..bytes]).await.unwrap();
}
}
}
// Drop only the first connection; keep listening for the read retry.
}
let (mut downstream, _) = listener.accept().await.unwrap();
let mut upstream = rocket::tokio::net::TcpStream::connect(upstream)
.await
.unwrap();
let _ = rocket::tokio::io::copy_bidirectional(&mut downstream, &mut upstream).await;
})
}
#[rocket::async_test]
#[ignore = "requires CLICKHOUSE_TEST_URL pointing to a disposable native ClickHouse server"]
async fn reconnects_after_outage() -> Result<()> {
use rocket::tokio::{
net::TcpListener,
time::{Duration, timeout},
};
let uri = std::env::var("CLICKHOUSE_TEST_URL")?;
let mut config = ConnectionConfig::parse(&uri)?;
let upstream = (config.host.clone(), config.port);
let listener = TcpListener::bind("127.0.0.1:0").await?;
let address = listener.local_addr()?;
let forwarding = proxy(listener, upstream.clone(), false);
config.host = address.ip().to_string();
config.port = address.port();
let analytics = Analytics::new(
config,
PoolSettings {
size: 1,
..PoolSettings::default()
},
)?;
let clone = analytics.clone();
let rows: Vec<Count> = analytics
.read(QueryBuilder::new("SELECT toUInt64(1) AS count"))
.await?;
assert_eq!(rows[0].count, 1);
let mut lease = analytics.pool.acquire().await?;
let initial = lease.client().clone();
lease.finish(Ok(()))?;
drop(lease);
forwarding.abort();
let _ = forwarding.await;
timeout(Duration::from_secs(2), async {
while !initial.is_closed() {
rocket::tokio::task::yield_now().await;
}
})
.await?;
// While unavailable, requests fail rather than retaining a broken client forever.
assert!(
clone
.read::<Count>(QueryBuilder::new("SELECT toUInt64(1) AS count"))
.await
.is_err()
);
let listener = TcpListener::bind(address).await?;
let forwarding = proxy(listener, upstream, true);
let rows: Vec<Count> = analytics
.read(QueryBuilder::new(
"SELECT toUInt64(2) AS count /* disconnect_test_marker */",
))
.await?;
assert_eq!(rows[0].count, 2);
clone.execute(QueryBuilder::new("SELECT 1")).await?;
forwarding.abort();
let _ = forwarding.await;
Ok(())
}
#[rocket::async_test]
#[ignore = "requires CLICKHOUSE_TEST_URL pointing to a disposable native ClickHouse server"]
async fn migrations_and_analytics_queries() -> Result<()> {
let uri = std::env::var("CLICKHOUSE_TEST_URL")?;
let mut config = ConnectionConfig::parse(&uri)?;
let (client, _) = config.connect().await?;
let database = format!("blocklist_test_{}", Uuid::new_v4().simple());
client
.execute(format!("CREATE DATABASE {database}"))
.await?;
client.execute(format!("USE {database}")).await?;
config.options.default_database = database.clone();
let analytics = Analytics::new(config, PoolSettings::default())?;
let result: Result<()> = async {
analytics.migrate().await?;
analytics.migrate().await?;
let applied = client
.query_one::<Count>("SELECT count() AS count FROM refinery_schema_history")
.await?;
assert_eq!(applied.count, 1);
let missing = Uuid::new_v4();
assert!(analytics.query_by_id(missing).await?.is_none());
let id = Uuid::new_v4();
// Quotes, backslashes, Unicode and dollar placeholders must remain data.
let target = "Example.org'\\пример$1";
analytics
.execute(
QueryBuilder::new(INSERT_QUERY_SQL)
.arg(id)
.arg(target)
.arg("2001:db8::1")
.arg(None::<String>)
.arg(None::<i32>)
.arg(None::<String>)
.arg(None::<String>)
.arg(None::<String>)
.arg(vec![Some("1.2.3.4".to_owned()), None])
.arg(Vec::<String>::new())
.arg(Vec::<String>::new())
.arg(None::<String>),
)
.await?;
let stored = analytics.query_by_id(id).await?.unwrap();
assert_eq!(stored.query, target);
assert_eq!(stored.resolved_ips, vec![Some("1.2.3.4".into()), None]);
let empty = analytics.complaints(target).await?;
assert_eq!(empty.len(), 14);
assert!(empty.iter().all(|day| day.count == 0));
analytics.feedback(id, target, "1.2.3.4", false).await?;
analytics.feedback(id, target, "1.2.3.4", false).await?;
analytics.feedback(id, target, "1.2.3.5", false).await?;
analytics.feedback(id, target, "1.2.3.6", true).await?;
analytics
.feedback(id, "other.org", "1.2.3.7", false)
.await?;
let total = client
.query_one::<Count>("SELECT count() AS count FROM human_reports")
.await?;
assert_eq!(total.count, 5);
let days = analytics.complaints(&target.to_lowercase()).await?;
assert_eq!(days.len(), 14);
assert_eq!(days.last().unwrap().count, 2);
assert_eq!(days.iter().map(|day| day.count).sum::<i64>(), 2);
assert!(days.windows(2).all(|pair| pair[0].date < pair[1].date));
let json = r#"{"message":"quoted \"value\"\nпример"}"#;
analytics
.execute(
QueryBuilder::new(INSERT_PROBE_REPORT_SQL)
.arg(id)
.arg(target)
.arg(1_i32)
.arg(vec!["blocked".to_owned()])
.arg(json)
.arg(None::<i16>)
.arg(None::<String>)
.arg(1234_i64),
)
.await?;
let probe = client
.query_one::<StoredProbe>(
"SELECT query_id, result, verdicts, duration_ms FROM probe_reports",
)
.await?;
assert_eq!(probe.query_id, id);
assert_eq!(probe.result, json);
assert_eq!(probe.verdicts, vec![Some("blocked".into())]);
assert_eq!(probe.duration_ms, Some(1234));
let available_before = analytics.pool.inner.status().available;
assert!(
analytics
.execute(QueryBuilder::new("SELECT unknown_column"))
.await
.is_err()
);
assert_eq!(available_before, analytics.pool.inner.status().available);
Ok(())
}
.await;
client.execute("USE default").await?;
client
.execute(format!("DROP DATABASE {database} SYNC"))
.await?;
result
}
}

View file

@ -1,4 +1,5 @@
use super::rate_limit::ApiRateLimiter;
use crate::analytics::{Analytics, ComplaintDay};
use crate::db::{WhitelistedEntry, check_whitelist, save_query};
use log::warn;
use querying::asn::AsnInfo;
@ -38,50 +39,13 @@ pub struct ApiCheckResponse {
pub complaints: Vec<ComplaintDay>,
}
#[derive(Serialize)]
pub struct ComplaintDay {
pub date: String,
pub count: i64,
}
async fn load_complaints(
db: &mut sqlx::PgConnection,
target: &Target,
) -> Result<Vec<ComplaintDay>, sqlx::Error> {
let rows = sqlx::query!(
r#"SELECT days.day::date::text AS "date!", COALESCE(reports.count, 0)::bigint AS "count!"
FROM generate_series(current_date - 13, current_date, interval '1 day') AS days(day)
LEFT JOIN (
SELECT h.date::date AS day, COUNT(DISTINCT h.source_ip)::bigint AS count
FROM human_reports h
JOIN queries q ON q.id = h.id
WHERE h.date >= current_date - 13
AND h.date < current_date + 1
AND h.works = false
AND LOWER(q.query) = LOWER($1)
GROUP BY h.date::date
) reports ON reports.day = days.day::date
ORDER BY days.day"#,
target.to_query()
)
.fetch_all(db)
.await?;
Ok(rows
.into_iter()
.map(|row| ComplaintDay {
date: row.date,
count: row.count,
})
.collect())
}
#[get("/check?<target>")]
pub async fn check(
target: &str,
checker: &State<Arc<RwLock<Checker>>>,
addr: &ClientRealAddr,
pool: &State<PgPool>,
analytics: &State<Analytics>,
limiter: &State<Arc<ApiRateLimiter>>,
) -> Result<Json<ApiCheckResponse>, Status> {
if !limiter.check(&addr.ip) {
@ -91,13 +55,8 @@ pub async fn check(
let target = Target::from(target.trim());
let check = checker.read().await.check(target.clone()).await;
let mut db = pool
.acquire()
.await
.map_err(|_| Status::InternalServerError)?;
let id: Option<String> = if let Ok(check) = &check {
match save_query(&mut db, &target, check, addr, checker.read().await).await {
match save_query(analytics, &target, check, addr, checker.read().await).await {
Ok(id) => Some(id.to_string()),
Err(e) => {
warn!("api: failed to save check: {:?}", e);
@ -109,6 +68,10 @@ pub async fn check(
};
let whitelist: Option<WhitelistedEntry> = if let Target::Domain(domain) = &target {
let mut db = pool
.acquire()
.await
.map_err(|_| Status::InternalServerError)?;
check_whitelist(domain, &mut db)
.await
.map_err(|_| Status::InternalServerError)?
@ -117,7 +80,7 @@ pub async fn check(
};
let complaints = match &check {
Ok(_) => match load_complaints(&mut db, &target).await {
Ok(_) => match analytics.complaints(&target.to_query()).await {
Ok(days) => days,
Err(e) => {
warn!("api: failed to load complaints: {:?}", e);

View file

@ -1,25 +1,31 @@
use crate::analytics::Analytics;
use rocket::State;
use rocket::http::Status;
use rocket_client_addr::ClientRealAddr;
use sqlx::postgres::PgPool;
use sqlx::types::Uuid;
use uuid::Uuid;
#[post("/feedback/<uuid>/<works>")]
pub async fn feedback(
uuid: &str,
works: bool,
pool: &State<PgPool>,
analytics: &State<Analytics>,
addr: &ClientRealAddr,
) -> Result<(), Status> {
sqlx::query!(
"INSERT INTO human_reports (id, source_ip, works) VALUES ($1, $2, $3)",
Uuid::try_parse(uuid).map_err(|_| Status::BadRequest)?,
addr.ip.to_string(),
works
)
.execute(&**pool)
.await
.map_err(|_| Status::InternalServerError)?;
let id = Uuid::try_parse(uuid).map_err(|_| Status::BadRequest)?;
let query = analytics
.query_by_id(id)
.await
.map_err(|error| {
log::warn!("api: failed to load feedback query {id}: {error}");
Status::InternalServerError
})?
.ok_or(Status::NotFound)?;
analytics
.feedback(id, &query.query, &addr.ip.to_string(), works)
.await
.map_err(|error| {
log::warn!("api: failed to save feedback for query {id}: {error}");
Status::InternalServerError
})?;
Ok(())
}

View file

@ -143,6 +143,7 @@ mod tests {
ttl: 1,
router: Some("192.0.2.1".parse().unwrap()),
outcome: reports::probe::DpiProbeHopOutcome::IcmpTimeExceeded,
tcp_diagnostics: None,
}];
let probes = vec![
ProbeMetadata {

View file

@ -1,5 +1,7 @@
use super::rate_limit::ProbeRateLimiter;
use crate::analytics::Analytics;
use crate::mqtt::{MqttPublisher, PublishError};
use klickhouse::QueryBuilder;
use log::warn;
use querying::target::Target;
use reports::probe::{
@ -114,6 +116,7 @@ pub async fn probe_query(
token: Option<&str>,
addr: &ClientRealAddr,
pool: &State<PgPool>,
analytics: &State<Analytics>,
mqtt: &State<MqttPublisher>,
limiter: &State<Arc<ProbeRateLimiter>>,
cache: &State<Arc<ProbeResponseCache>>,
@ -123,14 +126,16 @@ pub async fn probe_query(
}
let id = Uuid::try_parse(id).map_err(|_| Status::BadRequest)?;
let query: Option<(String, Vec<String>)> =
sqlx::query_as("SELECT query, resolved_ips FROM queries WHERE id = $1")
.bind(id)
.fetch_optional(&**pool)
.await
.map_err(|_| Status::InternalServerError)?;
let (query, resolved_ips) = query.ok_or(Status::NotFound)?;
let stored_query = analytics
.query_by_id(id)
.await
.map_err(|error| {
warn!("api: failed to load probe query {id}: {error}");
Status::InternalServerError
})?
.ok_or(Status::NotFound)?;
let query = stored_query.query;
let resolved_ips = stored_query.resolved_ips;
let target = Target::from(query.trim());
let domain = match &target {
Target::Domain(domain) => Some(domain.as_str()),
@ -146,6 +151,7 @@ pub async fn probe_query(
}
let ip = resolved_ips
.first()
.and_then(|ip| ip.as_ref())
.and_then(|ip| ip.parse::<IpAddr>().ok())
.ok_or(Status::BadRequest)?;
if Target::is_bogon(ip) {
@ -197,6 +203,7 @@ pub async fn probe_query(
let timeout = mqtt.task_timeout();
let pool = pool.inner().clone();
let analytics = analytics.inner().clone();
let cache = cache.inner().clone();
let query_id = id;
let id = id.to_string();
@ -215,10 +222,11 @@ pub async fn probe_query(
entry.response["job_id"] = json!(id);
if let Err(error) = insert_probe_report(
query_id,
&query,
&entry.response,
entry.target_traceroute.as_ref(),
entry.duration_ms,
&pool,
&analytics,
).await {
warn!("api: failed to save cached probe report for query {id}: {error}");
}
@ -267,10 +275,11 @@ pub async fn probe_query(
);
if let Err(error) = insert_probe_report(
query_id,
&query,
&response,
target_traceroute.as_ref(),
duration_ms,
&pool,
&analytics,
).await {
warn!("api: failed to save probe report for query {id}: {error}");
}
@ -412,11 +421,12 @@ pub fn build_probe_response(
async fn insert_probe_report(
query_id: Uuid,
query: &str,
response: &Value,
target_traceroute: Option<&TcpTracerouteResult>,
duration_ms: u64,
pool: &PgPool,
) -> Result<(), sqlx::Error> {
analytics: &Analytics,
) -> Result<(), klickhouse::KlickhouseError> {
let probe_id = response
.get("probe_id")
.and_then(Value::as_str)
@ -428,9 +438,10 @@ async fn insert_probe_report(
verdicts
.iter()
.filter_map(Value::as_str)
.map(str::to_owned)
.collect::<Vec<_>>()
})
.unwrap_or_else(|| vec!["uncertain"]);
.unwrap_or_else(|| vec!["uncertain".to_owned()]);
let (target_hop_count, target_trace_result) = traceroute_columns(target_traceroute);
let duration_ms = i64::try_from(duration_ms).unwrap_or(i64::MAX);
@ -439,37 +450,19 @@ async fn insert_probe_report(
return Ok(());
};
sqlx::query(
r#"
INSERT INTO probe_reports (
query_id,
probe_id,
verdicts,
result,
target_hop_count,
target_trace_result,
duration_ms
analytics
.execute(
QueryBuilder::new(crate::analytics::INSERT_PROBE_REPORT_SQL)
.arg(query_id)
.arg(query)
.arg(probe_id)
.arg(verdicts)
.arg(response.to_string())
.arg(target_hop_count)
.arg(target_trace_result)
.arg(duration_ms),
)
VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (query_id, probe_id)
DO UPDATE SET
date = NOW(),
verdicts = EXCLUDED.verdicts,
result = EXCLUDED.result,
target_hop_count = EXCLUDED.target_hop_count,
target_trace_result = EXCLUDED.target_trace_result,
duration_ms = EXCLUDED.duration_ms
"#,
)
.bind(query_id)
.bind(probe_id)
.bind(verdicts)
.bind(response)
.bind(target_hop_count)
.bind(target_trace_result)
.bind(duration_ms)
.execute(pool)
.await?;
.await?;
Ok(())
}
@ -609,6 +602,15 @@ fn build_host_verdict(
.filter(|(host, _)| matches!(host.host_type, HostType::Whitelist))
.collect::<Vec<_>>();
let most_blacklist_timeouts = !blacklist.is_empty()
&& is_strict_majority(
blacklist.len(),
blacklist
.iter()
.filter(|(_, evidence)| matches!(evidence, ProbeEvidence::DataTimeout { .. }))
.count(),
);
let most_blacklist_matches_baseline = !blacklist.is_empty()
&& is_strict_majority(
blacklist.len(),
@ -632,7 +634,9 @@ fn build_host_verdict(
.count(),
);
if most_blacklist_matches_baseline && most_whitelist_good {
if most_blacklist_timeouts && most_whitelist_good {
"cdn_block"
} else if most_blacklist_matches_baseline && most_whitelist_good {
"ok"
} else {
"uncertain"
@ -952,6 +956,50 @@ mod tests {
assert_eq!(response["verdicts"], json!(["ok"]));
}
#[test]
fn cdn_timeouts_are_classified_even_when_network_baseline_changes() {
let mut config = empty_config();
for (id, host_type) in [
("cdn", HostType::Blacklist),
("control", HostType::Whitelist),
] {
config.hosts.push(Host {
id: id.into(),
host: "192.0.2.1".into(),
host_type,
file_path: String::new(),
timeout_sec: 1,
min_data: 1,
});
}
let mut results = vec![
HostProbeResult {
host_id: "cdn".into(),
probe_evidence: ProbeEvidence::DataTimeout { bytes: 16384 },
},
HostProbeResult {
host_id: "control".into(),
probe_evidence: ProbeEvidence::Good,
},
];
for unblocked in [true, false] {
assert_eq!(
build_probe_verdicts(&results, &config, None, None, None, false, unblocked),
vec!["cdn_block"]
);
}
results[1].probe_evidence = ProbeEvidence::ClientHello;
assert_eq!(
build_probe_verdicts(&results, &config, None, None, None, false, true),
vec!["sni_block"]
);
results[1].probe_evidence = ProbeEvidence::ConnectionError;
assert_eq!(
build_probe_verdicts(&results, &config, None, None, None, false, true),
vec!["uncertain"]
);
}
#[test]
fn unblocked_reporter_still_detects_sni_block() {
let mut config = empty_config();

View file

@ -1,4 +1,6 @@
use crate::agency::Agency;
use crate::analytics::Analytics;
use klickhouse::QueryBuilder;
use querying::target::Target;
use querying::{Check, CheckVerdict, Checker};
use rocket::http::Status;
@ -12,12 +14,12 @@ use sqlx::types::Uuid;
use sqlx::types::chrono::NaiveDateTime;
pub async fn save_query(
db: &mut sqlx::PgConnection,
analytics: &Analytics,
target: &Target,
check: &Check,
addr: &ClientRealAddr,
checker: RwLockReadGuard<'_, Checker>,
) -> Result<Uuid, sqlx::Error> {
) -> Result<Uuid, klickhouse::KlickhouseError> {
let (cdn_networks, cdn_providers, rkn_domain): (Vec<_>, Vec<_>, Option<_>) =
if let CheckVerdict::Blocked {
cdn_provider_subnets,
@ -50,41 +52,30 @@ pub async fn save_query(
),
};
let id = sqlx::query_scalar(
"INSERT INTO queries (
query,
source_ip,
source_country_code,
source_city_geo_name_id,
target_country_code,
target_asn,
target_provider,
resolved_ips,
cdn_networks,
cdn_providers,
rkn_domain
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) RETURNING id",
)
.bind(target.to_query())
.bind(addr.ip.to_string())
.bind(
checker
.geo_ip(addr.ip)
.await
.map(|i| i.country_code)
.ok()
.flatten(),
)
.bind(check.geo.city_geo_name_id.map(|id| id as i32))
.bind(check.geo.country_code.clone())
.bind(check.geo.asn.clone())
.bind(check.geo.organisation.clone())
.bind(resolved_ips)
.bind(cdn_networks)
.bind(cdn_providers)
.bind(rkn_domain)
.fetch_one(db)
.await?;
let id = Uuid::new_v4();
let source_country_code = checker
.geo_ip(addr.ip)
.await
.ok()
.and_then(|info| info.country_code);
drop(checker);
analytics
.execute(
QueryBuilder::new(crate::analytics::INSERT_QUERY_SQL)
.arg(id)
.arg(target.to_query())
.arg(addr.ip.to_string())
.arg(source_country_code)
.arg(check.geo.city_geo_name_id.map(|id| id as i32))
.arg(check.geo.country_code.clone())
.arg(check.geo.asn.clone())
.arg(check.geo.organisation.clone())
.arg(resolved_ips)
.arg(cdn_networks)
.arg(cdn_providers)
.arg(rkn_domain),
)
.await?;
Ok(id)
}

View file

@ -2,6 +2,7 @@
extern crate rocket;
mod admin;
mod agency;
mod analytics;
mod api;
mod database_refresh;
mod db;
@ -36,20 +37,26 @@ fn api_error(status: Status, _: &Request) -> Json<JsonError> {
}
async fn run_migrations(rocket: Rocket<Build>) -> fairing::Result {
match rocket.state::<PgPool>() {
Some(db) => match sqlx::migrate!("./migrations").run(db).await {
Ok(_) => Ok(rocket),
Err(e) => {
error!("Failed to run database migrations: {}", e);
Err(rocket)
}
},
None => Err(rocket),
let Some(db) = rocket.state::<PgPool>() else {
return Err(rocket);
};
if let Err(e) = sqlx::migrate!("./migrations").run(db).await {
error!("Failed to run PostgreSQL migrations: {e}");
return Err(rocket);
}
let Some(analytics) = rocket.state::<analytics::Analytics>() else {
return Err(rocket);
};
if let Err(e) = analytics.migrate().await {
error!("Failed to run ClickHouse migrations: {e}");
return Err(rocket);
}
Ok(rocket)
}
#[launch]
async fn rocket() -> _ {
dotenvy::dotenv().ok();
env_logger::Builder::from_env(Env::default().default_filter_or("warn"))
.filter_module("website", LevelFilter::Info)
.filter_module("querying", LevelFilter::Info)
@ -99,16 +106,21 @@ async fn rocket() -> _ {
.await
.expect("Failed to create database pool");
let analytics = analytics::Analytics::connect_from_env()
.await
.expect("Failed to configure ClickHouse analytics");
rocket::build()
.manage(checker)
.manage(pool)
.manage(analytics)
.manage(api_limiter)
.manage(probe_limiter)
.manage(probe_update_download_limiter)
.manage(mqtt_publisher)
.manage(probe_response_cache)
.manage(probe_update_proxy)
.attach(AdHoc::try_on_ignite("SQLx Migrations", run_migrations))
.attach(AdHoc::try_on_ignite("Database Migrations", run_migrations))
.mount(
"/api/v1",
routes![