diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8d8ef15..9887e7e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -234,3 +234,6 @@ jobs: - name: Worker contract and failure-path tests run: npm run test:worker + + - name: Offline protocol probe parser checks + run: node scripts/probe_proxy.mjs --self-test diff --git a/README.md b/README.md index 837c0a5..6ce545b 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,10 @@ TGLock — это **локальный прокси** на твоём компьютере: принимает и MTProto, и SOCKS5. Он перехватывает соединения Telegram, заворачивает их в WebSocket и отправляет на веб-инфраструктуру Telegram — по нескольким маршрутам сразу, переключаясь на следующий, если текущий перестал отвечать. Провайдер видит обычный HTTPS. +Для CDN DC203 есть отдельный маршрут MTProto TCP к его закреплённому адресу: +этот CDN может не принимать WebSocket. Если его IP блокируется, остаётся +настроенный пользователем Worker. Диагностика показывает выбранный транспорт. + **Кому подойдёт:** - 📱 Telegram открывается через раз, сообщения уходят с задержкой, фото и видео не грузятся diff --git a/docs/ARCHITECTURE_V2.md b/docs/ARCHITECTURE_V2.md index dd6496a..46aadb6 100644 --- a/docs/ARCHITECTURE_V2.md +++ b/docs/ARCHITECTURE_V2.md @@ -205,8 +205,10 @@ Ping приходит в читающую половину, а отвечает 4. дополнительные Telegram IP и варианты хоста; 5. системный DNS (кроме CDN DC203). -Поддерживаются DC1–5 и media/CDN DC203. DC203 использует WebSocket-host DC2, -но подключается к собственному IP. DNS fallback на DC2 для DC203 запрещён: +Поддерживаются DC1–5 и CDN DC203. Для DC203 первым используется обычный MTProto +TCP к закреплённому `91.105.192.100:443`; этот сервер может принимать MTProto, +не принимая TLS/WebSocket. Резерв через Worker соединяется с тем же IP. +WebSocket-попытки также сохраняют закреплённый адрес. DNS fallback на DC2 для DC203 запрещён: успешный WebSocket handshake с другим DC не доставляет CDN-запрос в нужный датацентр. Worker также получает именно IP DC203. @@ -228,6 +230,10 @@ Ping приходит в читающую половину, а отвечает подхватываются. Тесты проверяют доверенный сертификат, SNI, неверное имя хоста и недоверенного издателя. +Это относится к TLS-маршрутам. Прямой CDN TCP — отдельный транспорт MTProto, +а не TLS с отключённой проверкой. Он доступен только для точного назначения +DC203. Для произвольных адресов такой обход политики не добавляется. + Cloudflare Worker принимается только как пользовательская настройка. TGLock не загружает и не скрывает публичные списки чужих доменов. diff --git a/docs/ISSUES_2026-09-19.md b/docs/ISSUES_2026-09-19.md index eaf6407..9342535 100644 --- a/docs/ISSUES_2026-09-19.md +++ b/docs/ISSUES_2026-09-19.md @@ -12,11 +12,12 @@ | #58 — Worker, туннели 0/1 | Worker отвечал 101 до открытия TCP; ошибки после handshake терялись | Worker 502/504 и диагностические WebSocket close-коды, Rust-тесты раннего закрытия и reset. HTTP 426 проверяет только публикацию Worker | | #50 — ни один маршрут не работает | Последовательные TCP/TLS таймауты задерживали резерв; кратковременные туннели ошибочно сохраняли предпочтение маршрута | Ограниченный параллелизм, cooldown после upstream failure, ошибки Worker. Нельзя обещать обход, когда недоступны и Telegram, и собственный Worker | | #42 — Android через LAN | Прежние duplex-исправления уже в main; потеря ошибок upstream скрывала дальнейший отказ | Независимый Android-compatible криптографический вектор, фрагментированные init и одновременная передача 128/256 КиБ, диагностика разрывов. Нужна проверка Play Market-клиента в сети репортёра | -| #32 — медиа | DC203 fallback через DNS попадал в DC2, хотя CDN имеет свой DC и ключи | DC203 сохраняет IP CDN во всех маршрутах; тест Worker dst, двунаправленная передача. Реальные фотографии аккаунта репортёра не проверены | +| #32 — медиа | DC203 fallback через DNS попадал в DC2, хотя CDN имеет свой DC и ключи; на закреплённом CDN IP WebSocket не отвечает, но обычный MTProto TCP работает | DC203 использует точный CDN IP по TCP с преобразованием transport-шифрования; Worker остаётся резервом. Проверка живым req_pq/resPQ и тесты обеих сторон. Реальные фотографии аккаунта репортёра не проверены | | #53 — статическая aarch64 | Системный OpenSSL мешал самостоятельной musl-сборке | rustls + ring, ARM64 CI с проверкой ELF и запуском CLI, артефакт и OpenWrt-инструкция. Физический NanoPi R4S не проверен | | #9 — Android | Старый PR отстал от main, сборка требовала ручной настройки; жизненный цикл сервиса мог останавливать прокси или оставлять ложное уведомление | Перенос актуального ядра, foreground service по состоянию прокси, APK CI, постоянный секрет в каталоге приложения. HyperOS и длительная фоновая работа требуют устройства | -Дополнительно: Unicode-секрет больше не вызывает panic; GUI сохраняет секрет +Дополнительно: журнал больше не замолкает после первых 64 разных событий; +Unicode-секрет больше не вызывает panic; GUI сохраняет секрет в каталоге приложения с миграцией прежнего файла; npm lock обновляет уязвимый транзитивный `nanoid` без смены версии приложения. @@ -25,6 +26,10 @@ - TLS проверяет имя из WebSocket URI даже при подключении к закреплённому IP. Переход на rustls использует встроенные WebPKI roots; системные пользовательские корневые сертификаты автоматически не импортируются. +- Прямой маршрут CDN203 использует обычный MTProto TCP к одному закреплённому + адресу, без TLS-обёртки. Ключи и шифрование содержимого Telegram не меняются; + снимается только transport-obfuscation локального прокси и накладывается + transport-obfuscation сервера. Если провайдер блокирует CDN IP, нужен Worker. - [Telegram WebSocket](https://core.telegram.org/mtproto/transports#websocket) является потоком байтов. Границы TCP read и WS messages не обязаны совпадать с MTProto-пакетами; дополнительный парсер пакетов не добавлялся. @@ -47,7 +52,8 @@ cargo test --locked --no-default-features --features cli --lib --bins ``` CI дополнительно проверяет MSRV 1.88, состав macOS GUI bundle, статическую -ARM64 musl-сборку и подпись/содержимое Android debug APK. Конкретные результаты +ARM64 musl-сборку, подпись/содержимое Android debug APK и запуск на Android 15 +эмуляторе. Пропуск UI lifecycle отмечается отдельно от успешного запуска. Конкретные результаты и ссылки на прогоны фиксируются в описании PR после завершения CI. Worker: `npm run test:worker` исполняет настоящий deployment-файл с подменой @@ -55,6 +61,10 @@ Worker: `npm run test:worker` исполняет настоящий deployment- Это не тест живого Cloudflare-аккаунта. Отдельный тест Rust поднимает локальный WebSocket и проверяет совместимость клиента с контрактом Worker. +Для проверки реального бинаря без Telegram-аккаунта добавлен +[независимый Node.js probe](LIVE_PROBE.md). Он проверяет `req_pq_multi → resPQ`, +nonce и целостность ответа; это ещё не авторизация и не скачивание медиа. + ## Что нужно проверить на устройствах Android: установка APK, старт/стоп, ссылка в Telegram, сохранение секрета после diff --git a/docs/LIVE_PROBE.md b/docs/LIVE_PROBE.md new file mode 100644 index 0000000..d3db2c0 --- /dev/null +++ b/docs/LIVE_PROBE.md @@ -0,0 +1,61 @@ +# Manual protocol probe + +`scripts/probe_proxy.mjs` uses only Node.js built-ins and is independent of the +Rust transport helpers. It connects to an already running TGLock listener at +`127.0.0.1`, sends one unauthenticated `req_pq_multi` through a secret-protected +obfuscated2 padded-intermediate stream, and validates the returned `resPQ`, its +request nonce, and the lengths of its TL fields. It stops before creating an +authorization key. No Telegram account, API ID, API hash or login is needed. + +The secret is read from an explicitly supplied file and is never printed. Use +the same secret file as the running CLI. Do not paste proxy links or secret +values into public logs. + +```sh +# Offline validation first: no network access. +node scripts/probe_proxy.mjs --self-test + +# Start a local CLI separately, using a persistent secret file. +tglock-cli --port 18080 --secret-file /private/path/tglock-secret + +# Ordinary DC, media route, and CDN route; run each explicitly. +node scripts/probe_proxy.mjs --port 18080 --secret-file /private/path/tglock-secret --dc 2 +node scripts/probe_proxy.mjs --port 18080 --secret-file /private/path/tglock-secret --dc -4 --fragment-size 7 +node scripts/probe_proxy.mjs --port 18080 --secret-file /private/path/tglock-secret --dc 203 +``` + +On Windows, supply the downloaded CLI executable and a Windows file path in the +same commands. Supported DC values are `1` through `5` and `203`; a negative +value requests a media route. `--timeout-ms` defaults to 15000 and is bounded at +120000. The response is bounded at 2 MiB. `--fragment-size 7` sends small writes +with 2 ms gaps to exercise stream fragmentation; TCP can still combine writes. + +A successful JSON report contains `response: "resPQ"`, `nonceMatches: true`, the +requested DC, the public RSA fingerprint count, and elapsed time. Exit status 1 +means connection, timeout, decryption/framing or response validation failed. +The ordinary CI suite does not run this live probe. + +Success demonstrates a correctly relayed protocol exchange. It does **not** +authenticate the responding server, prove its DC identity, log into an account, +or verify message sending, media downloads or Android lifecycle. In particular, +`requestedDc` describes the request, not an independently confirmed backend. + +## Isolate CDN transport failures + +To distinguish an unavailable CDN WebSocket endpoint from an unavailable CDN +TCP connection, explicitly run: + +```sh +node scripts/probe_proxy.mjs --direct-cdn --dc 203 +``` + +This optional mode bypasses local TGLock and connects **only** to the pinned +Telegram CDN address `91.105.192.100:443`, using raw obfuscated2 TCP without a +proxy secret or TLS. It supports no arbitrary host. It performs the same single +unauthenticated exchange and still does not prove account or media operation. +The parser accepts trailing bytes after the complete `resPQ` TL object because +live CDN replies can include random padding in the declared message length. + +Protocol references: [handshake initiation](https://core.telegram.org/mtproto/auth_key), +[obfuscated transports](https://core.telegram.org/mtproto/mtproto-transports), +and [unencrypted messages](https://core.telegram.org/mtproto/description#unencrypted-message). diff --git a/scripts/probe_proxy.mjs b/scripts/probe_proxy.mjs new file mode 100644 index 0000000..94e58a3 --- /dev/null +++ b/scripts/probe_proxy.mjs @@ -0,0 +1,191 @@ +#!/usr/bin/env node +// Manual, account-free probe of an ALREADY RUNNING local TGLock instance. +// Protocol sources (this does not import the Rust implementation): +// https://core.telegram.org/mtproto/auth_key +// https://core.telegram.org/mtproto/mtproto-transports#transport-obfuscation +// https://core.telegram.org/mtproto/description#unencrypted-message +import { createCipheriv, createDecipheriv, createHash, randomBytes } from "node:crypto"; +import { readFile } from "node:fs/promises"; +import { createConnection } from "node:net"; +import { setTimeout as sleep } from "node:timers/promises"; +import assert from "node:assert/strict"; + +const MAX_RESPONSE = 2 * 1024 * 1024; +const HELP = `Usage: node scripts/probe_proxy.mjs --secret-file PATH [--port 1080] [--dc 2] [--timeout-ms 15000] [--fragment-size 0] +Or: node scripts/probe_proxy.mjs --direct-cdn --dc 203 +DC: 1..5 or 203; negative values request the media route. +Default connects only to 127.0.0.1. --direct-cdn explicitly probes only 91.105.192.100:443 without a secret. +Reads the local proxy secret from the explicit file; never prints it. +Sends one req_pq_multi and checks resPQ/nonce. No account, API credentials, login or auth key is created. +This proves a protocol response, not DC identity, account operation or media downloads. +Use --self-test for offline parser checks; --fragment-size 7 sends small writes with 2ms gaps.`; + +function options(args) { + const result = { port: 1080, dc: 2, timeoutMs: 15000, fragmentSize: 0 }; + const numeric = { "--port": "port", "--dc": "dc", "--timeout-ms": "timeoutMs", "--fragment-size": "fragmentSize" }; + for (let i = 0; i < args.length; i++) { + const name = args[i]; + if (name === "--direct-cdn") { result.directCdn = true; continue; } + const value = args[++i]; + if (value === undefined) throw new Error(`Missing value for ${name}`); + if (name === "--secret-file") result.secretFile = value; + else if (numeric[name] && /^-?\d+$/.test(value)) result[numeric[name]] = Number(value); + else throw new Error(`Invalid option ${name}`); + } + if (!result.secretFile && !result.directCdn) throw new Error("--secret-file is required"); + if (result.directCdn && result.dc !== 203) throw new Error("Direct CDN probe requires --dc 203"); + if (!Number.isInteger(result.port) || result.port < 1 || result.port > 65535) throw new Error("Invalid port"); + if (![1, 2, 3, 4, 5, 203].includes(Math.abs(result.dc))) throw new Error("Unsupported DC"); + if (result.timeoutMs < 100 || result.timeoutMs > 120000) throw new Error("Timeout must be 100..120000ms"); + if (result.fragmentSize < 0 || result.fragmentSize > 65536) throw new Error("Fragment size must be 0..65536"); + return result; +} + +function makeRequest(secret, dc) { + let header; + do { + header = randomBytes(64); + } while (header[0] === 0xef || ["HEAD", "POST", "GET ", "OPTI"].includes(header.toString("ascii", 0, 4)) + || [0xeeeeeeee, 0xdddddddd, 0x02010316].includes(header.readUInt32LE()) || header.readUInt32LE(4) === 0); + header.fill(0xdd, 56, 60); // padded intermediate transport + header.writeInt16LE(dc, 60); + const salted = (key) => secret ? createHash("sha256").update(key).update(secret).digest() : Buffer.from(key); + const encrypt = createCipheriv("aes-256-ctr", salted(header.subarray(8, 40)), header.subarray(40, 56)); + const reversed = Buffer.from(header.subarray(8, 56)).reverse(); + const decrypt = createDecipheriv("aes-256-ctr", salted(reversed.subarray(0, 32)), reversed.subarray(32)); + const wireHeader = Buffer.from(header); + encrypt.update(header).copy(wireHeader, 56, 56); // advances outgoing CTR by all 64 bytes + + const nonce = randomBytes(16); + const payload = Buffer.alloc(40); + // auth_key_id = 0, client message ID divisible by four, TL payload length = 20. + const now = BigInt(Date.now()); + const messageId = ((now / 1000n << 32n) | ((now % 1000n) * (1n << 32n) / 1000n)) & ~3n; + payload.writeBigUInt64LE(messageId, 8); + payload.writeUInt32LE(20, 16); + payload.writeUInt32LE(0xbe7e8ef1, 20); + nonce.copy(payload, 24); + const padded = Buffer.concat([payload, randomBytes(7)]); + const length = Buffer.alloc(4); + length.writeUInt32LE(padded.length); + return { wire: Buffer.concat([wireHeader, encrypt.update(Buffer.concat([length, padded]))]), decrypt, nonce }; +} + +function parseResPQ(frame, nonce) { + if (frame.length === 4) throw new Error(`MTProto transport error ${frame.readInt32LE()}`); + if (frame.length < 20 || frame.readBigUInt64LE() !== 0n) throw new Error("Expected an unencrypted MTProto response"); + const length = frame.readUInt32LE(16); + if (length < 48 || length % 4 !== 0 || length > frame.length - 20) throw new Error("Invalid MTProto message length"); + const body = frame.subarray(20, 20 + length); + if (body.readUInt32LE() !== 0x05162463) throw new Error("Response is not resPQ"); + if (!body.subarray(4, 20).equals(nonce)) throw new Error("resPQ nonce does not match request"); + // pq is at most eight bytes for this handshake, so its TL string uses the + // one-byte length encoding followed by padding to a four-byte boundary. + const pqLength = body[36]; + if (pqLength < 1 || pqLength > 8) throw new Error("Invalid resPQ pq length"); + const vectorOffset = 36 + Math.ceil((1 + pqLength) / 4) * 4; + if (vectorOffset + 8 > body.length || body.readUInt32LE(vectorOffset) !== 0x1cb5c415) throw new Error("Invalid RSA fingerprint vector"); + const count = body.readUInt32LE(vectorOffset + 4); + // A CDN may include trailing random bytes in its declared message length. + // Validate the complete TL object fits; do not mistake padding for corruption. + if (count < 1 || count > 64 || vectorOffset + 8 + count * 8 > body.length) throw new Error("Invalid RSA fingerprint count"); + // Deliberately do not infer DC identity from these public fingerprints. + return { response: "resPQ", nonceMatches: true, rsaFingerprintCount: count }; +} + +async function probe(config) { + let secret = null; + if (!config.directCdn) { + let hex = (await readFile(config.secretFile, "utf8")).trim(); + if (/^dd[0-9a-f]{32}$/i.test(hex)) hex = hex.slice(2); + if (!/^[0-9a-f]{32}$/i.test(hex)) throw new Error("Secret file must contain 32 hex characters or dd followed by 32 hex characters"); + secret = Buffer.from(hex, "hex"); + } + const { wire, decrypt, nonce } = makeRequest(secret, config.dc); + const started = Date.now(); + return new Promise((resolve, reject) => { + const socket = createConnection(config.directCdn + ? { host: "91.105.192.100", port: 443 } + : { host: "127.0.0.1", port: config.port }); + let pending = Buffer.alloc(0); + let received = 0; + let finished = false; + const finish = (error, result) => { + if (finished) return; + finished = true; + clearTimeout(timer); + socket.destroy(); + if (error) reject(error); + else resolve({ transport: config.directCdn ? "direct-cdn-tcp" : "local-proxy", requestedDc: config.dc, ...result, elapsedMs: Date.now() - started }); + }; + const timer = setTimeout(() => finish(new Error("Timed out waiting for resPQ")), config.timeoutMs); + socket.on("error", (error) => finish(error)); + socket.on("end", () => finish(new Error("Proxy closed before a complete resPQ response"))); + socket.on("data", (chunk) => { + try { + received += chunk.length; + if (received > MAX_RESPONSE) throw new Error("Response exceeded the 2MiB limit"); + pending = Buffer.concat([pending, decrypt.update(chunk)]); + while (pending.length >= 4) { + const length = pending.readUInt32LE(); + if (length & 0x80000000) { // optional intermediate quick acknowledgment + pending = pending.subarray(4); + continue; + } + if (length < 4 || length > MAX_RESPONSE - 4) throw new Error("Invalid intermediate frame length"); + if (pending.length < 4 + length) return; + if (length >= 8 && length <= 16 && pending.readUInt32LE(4) === 0xffffffff) { + pending = pending.subarray(4 + length); // padded intermediate quick ACK + continue; + } + finish(null, parseResPQ(pending.subarray(4, 4 + length), nonce)); + return; + } + } catch (error) { finish(error); } + }); + socket.on("connect", async () => { + try { + socket.setNoDelay(true); + const size = config.fragmentSize || wire.length; + for (let offset = 0; offset < wire.length && !finished; offset += size) { + socket.write(wire.subarray(offset, offset + size)); + if (config.fragmentSize) await sleep(2); + } + } catch (error) { finish(error); } + }); + }); +} + +function selfTest() { + // Fixed TL fixture: resPQ, request nonce 00..0f, server nonce 10..1f, + // eight-byte pq, a one-element vector of public RSA fingerprints. + const body = Buffer.from("63241605000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f08112233445566778800000015c4b51c010000008877665544332211", "hex"); + const envelope = Buffer.alloc(20); + envelope.writeUInt32LE(body.length, 16); + const frame = Buffer.concat([envelope, body, Buffer.from([1, 2, 3])]); + const nonce = Buffer.from("000102030405060708090a0b0c0d0e0f", "hex"); + assert.equal(parseResPQ(frame, nonce).rsaFingerprintCount, 1); + const paddedEnvelope = Buffer.from(envelope); + paddedEnvelope.writeUInt32LE(body.length + 128, 16); + assert.equal(parseResPQ(Buffer.concat([paddedEnvelope, body, Buffer.alloc(128, 0x42)]), nonce).rsaFingerprintCount, 1); + assert.throws(() => parseResPQ(frame, Buffer.alloc(16)), /nonce/); + assert.throws(() => parseResPQ(frame.subarray(0, 30), nonce), /length/); + const wrongConstructor = Buffer.from(frame); + wrongConstructor[20] = 0; + assert.throws(() => parseResPQ(wrongConstructor, nonce), /not resPQ/); + const oversizedVector = Buffer.from(frame); + oversizedVector.writeUInt32LE(65, 72); + assert.throws(() => parseResPQ(oversizedVector, nonce), /count/); + assert.throws(() => parseResPQ(Buffer.from("6cfeffff", "hex"), nonce), /-404/); + console.log("Offline parser checks passed; no network connection made."); +} + +try { + const args = process.argv.slice(2); + if (args.length === 1 && args[0] === "--self-test") selfTest(); + else if (args.length === 1 && ["--help", "-h"].includes(args[0])) console.log(HELP); + else console.log(JSON.stringify(await probe(options(args)), null, 2)); +} catch (error) { + console.error(`Probe failed: ${error.message}`); + process.exitCode = 1; +} diff --git a/src/gui.rs b/src/gui.rs index 491367f..17a3f3c 100644 --- a/src/gui.rs +++ b/src/gui.rs @@ -48,9 +48,8 @@ struct StatusSnapshot { route_failures: u32, /// Запросы, отклонённые политикой «в LAN-режиме только Telegram». /// - /// Ноль при неработающем телефоне означает, что он вообще не дотянулся до - /// этой машины; не ноль — что дотянулся, и разбираться надо с адресами - /// (by-sonic/tglock#42). + /// Ноль означает, что фильтр не отклонял запросы. Доходит ли телефон до + /// машины, показывают соединения и журнал, а не этот счётчик (#42). blocked: u32, /// Клиенты, которые дошли, но не сумели договориться. Почти всегда это /// ссылка `tg://proxy` от прошлого запуска, то есть другой секрет. diff --git a/src/proxy.rs b/src/proxy.rs index edbbe66..2aa2cf1 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -1,5 +1,5 @@ use crate::config::ListenConfig; -use std::collections::{HashSet, VecDeque}; +use std::collections::VecDeque; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::sync::atomic::{AtomicBool, AtomicU16, AtomicU32, Ordering}; use std::sync::{Arc, Mutex}; @@ -71,15 +71,17 @@ pub struct Stats { shutdown: Mutex>>, } -/// Однократные сообщения о том, что происходит с подключениями. +/// Сообщения о подключениях со склейкой недавних повторов. /// -/// Однократные намеренно: отклонённый адрес повторяется десятки раз в минуту, +/// Склейка намеренна: отклонённый адрес повторяется десятки раз в минуту, /// и без склейки журнал превратился бы в одну строку, повторённую сто раз. /// Число повторов при этом не теряется — оно в счётчике `blocked`. #[derive(Default)] struct Events { pending: VecDeque, - seen: HashSet, + // FIFO of recent distinct messages, independent from the consumer queue. + // Once full, evict the oldest entry rather than silencing all future logs. + seen: VecDeque, } /// Сколько разных событий помним, чтобы буфер не рос без границы. @@ -132,13 +134,20 @@ impl Stats { }) } - /// Запомнить событие, если такого ещё не было. + /// Запомнить событие, если его нет среди последних EVENT_LIMIT разных сообщений. pub fn note(&self, message: impl Into) { let message = message.into(); let mut events = self.events.lock().unwrap(); - if events.seen.len() >= EVENT_LIMIT || !events.seen.insert(message.clone()) { + if events.seen.contains(&message) { return; } + if events.seen.len() == EVENT_LIMIT { + events.seen.pop_front(); + } + events.seen.push_back(message.clone()); + if events.pending.len() == EVENT_LIMIT { + events.pending.pop_front(); + } events.pending.push_back(message); } @@ -683,13 +692,9 @@ fn dc_from_init(init: &[u8; 64]) -> Option<(u16, bool)> { // -- WebSocket tunnel ------------------------------------------------------- -/// Keeps `Stats::ws` equal to the number of *established* tunnels. -/// -/// Counting attempts instead would let the interface announce «Telegram на -/// связи» while the WebSocket handshake is still failing over between routes, -/// which takes seconds per route. Reporting a working tunnel that does not -/// exist yet is the whole reason users saw «прокси подключён, а Telegram не -/// работает». +/// Counts opened upstream transports, not pending connection attempts. +/// WebSocket routes have completed their upgrade; native CDN routes have an +/// open TCP socket. Neither alone proves a successful Telegram protocol reply. struct EstablishedTunnel<'a>(&'a Stats); impl<'a> EstablishedTunnel<'a> { @@ -715,7 +720,7 @@ async fn ws_tunnel( ) -> Result<(), Box> { use futures_util::{SinkExt, StreamExt}; - let (ws, connected) = match stats.transport.connect(dc, media).await { + let (connection, connected) = match stats.transport.connect(dc, media).await { Ok(connected) => connected, Err(error) => { // Единственное место, где известно, ПОЧЕМУ туннеля нет. Раньше @@ -729,6 +734,13 @@ async fn ws_tunnel( let _tunnel = EstablishedTunnel::new(stats); stats.note_tunnel(dc, connected.route.kind.ui_code()); + let ws = match connection { + crate::transport::TelegramConnection::WebSocket(ws) => ws, + crate::transport::TelegramConnection::Tcp(remote) => { + return cdn_tcp_tunnel(tcp, remote, init, crypto, stats, &connected, dc).await; + } + }; + let (mut tcp_r, mut tcp_w) = tokio::io::split(tcp); let (mut ws_w, mut ws_r) = ws.split(); let (upstream_crypto, downstream_crypto) = match crypto.map(|crypto| crypto.split()) { @@ -853,6 +865,79 @@ async fn ws_tunnel( Ok(()) } +/// DC203's pinned CDN endpoint speaks obfuscated MTProto over TCP. Preserve the +/// same generated init and independent CTR streams used for a WebSocket route. +/// The transport engine restricts this connection to the exact CDN destination. +async fn cdn_tcp_tunnel( + client: TcpStream, + mut remote: TcpStream, + init: &[u8; 64], + crypto: Option, + stats: &Stats, + connected: &crate::transport::ConnectedRoute, + dc: u16, +) -> Result<(), Box> { + remote + .write_all(init) + .await + .map_err(|error| tunnel_route_error(stats, connected, dc, &error.to_string()))?; + let (mut client_r, mut client_w) = client.into_split(); + let (mut remote_r, mut remote_w) = remote.into_split(); + let (upstream_crypto, downstream_crypto) = match crypto.map(|crypto| crypto.split()) { + Some((upstream, downstream)) => (Some(upstream), Some(downstream)), + None => (None, None), + }; + let upstream = async { + let mut crypto = upstream_crypto; + let mut buffer = vec![0; 65536]; + loop { + let read = client_r.read(&mut buffer).await?; + if read == 0 { + return Ok::<_, Box>(()); + } + if let Some(crypto) = &mut crypto { + crypto.apply(&mut buffer[..read]); + } + remote_w + .write_all(&buffer[..read]) + .await + .map_err(|error| tunnel_route_error(stats, connected, dc, &error.to_string()))?; + } + }; + let downstream = async { + let mut crypto = downstream_crypto; + let mut buffer = vec![0; 65536]; + let mut received_payload = false; + loop { + let read = remote_r + .read(&mut buffer) + .await + .map_err(|error| tunnel_route_error(stats, connected, dc, &error.to_string()))?; + if read == 0 { + return if received_payload { + Ok(()) + } else { + Err(tunnel_route_error( + stats, + connected, + dc, + "CDN TCP закрыт без ответа Telegram", + )) + }; + } + received_payload = true; + if let Some(crypto) = &mut crypto { + crypto.apply(&mut buffer[..read]); + } + client_w.write_all(&buffer[..read]).await?; + } + }; + tokio::select! { + result = upstream => result, + result = downstream => result, + } +} + fn tunnel_route_error( stats: &Stats, connected: &crate::transport::ConnectedRoute, @@ -1100,6 +1185,52 @@ mod tests { ); } + #[test] + fn diagnostics_keep_new_events_after_the_first_sixty_four() { + let stats = Stats::with_secret([0; 16]); + for index in 0..=EVENT_LIMIT { + stats.note(format!("route failure {index}")); + } + { + let events = stats.events.lock().unwrap(); + assert_eq!(events.seen.len(), EVENT_LIMIT); + assert_eq!(events.pending.len(), EVENT_LIMIT); + } + let events = stats.drain_events(); + assert_eq!(events.first().unwrap(), "route failure 1"); + assert_eq!( + events.last().unwrap(), + &format!("route failure {EVENT_LIMIT}") + ); + + // Draining must not discard deduplication history, but must not stop + // the next new error from reaching the operator either. + stats.note(format!("route failure {EVENT_LIMIT}")); + stats.note("a new upstream error"); + assert_eq!(stats.drain_events(), vec!["a new upstream error"]); + } + + #[test] + fn an_evicted_event_can_be_reported_again_without_repeating_recent_events() { + let stats = Stats::with_secret([0; 16]); + stats.note("original error"); + assert_eq!(stats.drain_events(), vec!["original error"]); + stats.note("original error"); + assert!(stats.drain_events().is_empty()); + + // More distinct events than the retained history used to silence + // diagnostics permanently, even when the frontend drained every one. + for index in 0..EVENT_LIMIT * 2 { + let message = format!("error {index}"); + stats.note(message.clone()); + assert_eq!(stats.drain_events(), vec![message]); + } + stats.note("original error"); + stats.note("original error"); + assert_eq!(stats.drain_events(), vec!["original error"]); + assert_eq!(stats.events.lock().unwrap().seen.len(), EVENT_LIMIT); + } + /// Диагностика обязана показывать пару из одного соединения. /// /// Пока это были два независимых поля, при десятках одновременных @@ -1548,9 +1679,53 @@ mod tests { } } - #[allow(clippy::result_large_err)] #[tokio::test] async fn fragmented_encrypted_upload_and_download_preserve_both_streams() { + check_fragmented_encrypted_duplex(false).await; + } + + #[tokio::test] + async fn cdn_tcp_preserves_fragmented_encrypted_duplex_and_clean_close() { + check_fragmented_encrypted_duplex(true).await; + } + + #[tokio::test] + async fn cdn_tcp_closing_before_a_reply_is_reported_as_a_route_failure() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let relay_port = listener.local_addr().unwrap().port(); + let relay = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut init = [0; INIT_LEN]; + stream.read_exact(&mut init).await.unwrap(); + }); + let stats = Stats::new(); + stats.transport.force_local_route_with( + relay_port, + crate::transport::RouteKind::TelegramTcp, + String::new(), + ); + let (port, server) = start_proxy(stats.clone(), false).await; + let mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + client + .write_all(&unambiguous_client_init(&stats.secret, 203)) + .await + .unwrap(); + wait_until("CDN failure is counted", || { + stats.ws_failures.load(Ordering::Relaxed) == 1 + }) + .await; + assert_eq!(stats.transport.route_failures(), 1); + assert!(stats + .drain_events() + .iter() + .any(|event| event.contains("CDN TCP закрыт без ответа"))); + relay.await.unwrap(); + stats.stop(); + server.await.unwrap().unwrap(); + } + + #[allow(clippy::result_large_err)] + async fn check_fragmented_encrypted_duplex(raw_tcp: bool) { use futures_util::{SinkExt, StreamExt}; let request: Vec = (0..131_072).map(|i| (i % 251) as u8).collect(); let response: Vec = (0..262_144).map(|i| (i % 239) as u8).collect(); @@ -1559,7 +1734,30 @@ mod tests { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let relay_port = listener.local_addr().unwrap().port(); let relay = tokio::spawn(async move { - let (tcp, _) = listener.accept().await.unwrap(); + let (mut tcp, _) = listener.accept().await.unwrap(); + if raw_tcp { + let mut init = [0; INIT_LEN]; + tcp.read_exact(&mut init).await.unwrap(); + let mut peer = crate::mtproto::test_relay_peer(&init); + let mut download = relay_response; + peer.encrypt(&mut download); + let (mut reader, mut writer) = tcp.into_split(); + let send = async { + for chunk in download.chunks(8191) { + writer.write_all(chunk).await.unwrap(); + tokio::task::yield_now().await; + } + }; + let receive = async { + let mut received = vec![0; expected_request.len()]; + reader.read_exact(&mut received).await.unwrap(); + peer.decrypt(&mut received); + assert_eq!(received, expected_request); + }; + tokio::join!(send, receive); + writer.shutdown().await.unwrap(); + return; + } let mut ws = tokio_tungstenite::accept_hdr_async(tcp, |_: &Request, mut response: Response| { response @@ -1596,9 +1794,17 @@ mod tests { sink.close().await.unwrap(); }); let stats = Stats::new(); - stats.transport.force_local_route(relay_port); + if raw_tcp { + stats.transport.force_local_route_with( + relay_port, + crate::transport::RouteKind::TelegramTcp, + String::new(), + ); + } else { + stats.transport.force_local_route(relay_port); + } let (port, server) = start_proxy(stats.clone(), false).await; - let init = unambiguous_client_init(&stats.secret, -4); + let init = unambiguous_client_init(&stats.secret, if raw_tcp { 203 } else { -4 }); let mut peer = crate::mtproto::test_client_peer(&init, &stats.secret); let mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); // A mobile TCP stream can split the 64-byte init and subsequent data @@ -1628,6 +1834,14 @@ mod tests { }) .await .expect("both encrypted directions must make progress"); + assert_eq!(stats.ws_failures.load(Ordering::Relaxed), 0); + if raw_tcp { + assert_eq!( + stats.last_route(), + crate::transport::RouteKind::TelegramTcp.ui_code() + ); + assert_eq!(stats.last_dc(), 203); + } stats.stop(); server.await.unwrap().unwrap(); } diff --git a/src/transport.rs b/src/transport.rs index 61f7b57..625464e 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -15,15 +15,24 @@ const MAX_CONNECTING: usize = 3; const FAILURE_BACKOFF_INITIAL: Duration = Duration::from_secs(30); const FAILURE_BACKOFF_MAX: Duration = Duration::from_secs(30 * 60); const HTTPS_PORT: u16 = 443; +const CDN203_IP: &str = "91.105.192.100"; pub type TelegramWebSocket = WebSocketStream>; +#[derive(Debug)] +pub enum TelegramConnection { + WebSocket(Box), + /// Native obfuscated2 transport, currently restricted to the CDN203 IP. + Tcp(TcpStream), +} + #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub enum RouteKind { TelegramIp, AlternateTelegramIp, SystemDns, CloudflareWorker, + TelegramTcp, } impl RouteKind { @@ -33,6 +42,7 @@ impl RouteKind { Self::AlternateTelegramIp => 2, Self::SystemDns => 3, Self::CloudflareWorker => 4, + Self::TelegramTcp => 5, } } @@ -42,6 +52,7 @@ impl RouteKind { 2 => Some(Self::AlternateTelegramIp), 3 => Some(Self::SystemDns), 4 => Some(Self::CloudflareWorker), + 5 => Some(Self::TelegramTcp), _ => None, } } @@ -53,6 +64,7 @@ impl RouteKind { Self::AlternateTelegramIp => "Запасной Telegram IP", Self::SystemDns => "Системный DNS", Self::CloudflareWorker => "Cloudflare Worker", + Self::TelegramTcp => "Telegram CDN TCP", } } } @@ -73,11 +85,22 @@ pub struct Route { pub kind: RouteKind, /// TCP port to dial. Always 443 for Telegram and for Cloudflare Workers. pub port: u16, - /// Wrap the connection in TLS. Always true outside tests. + /// TLS for WebSocket routes; false for native obfuscated2 CDN TCP. pub secure: bool, } impl Route { + fn cdn_tcp() -> Self { + Self { + connect_host: CDN203_IP.to_owned(), + websocket_host: String::new(), + path: String::new(), + kind: RouteKind::TelegramTcp, + port: HTTPS_PORT, + secure: false, + } + } + /// A production route: TLS on 443. fn https(connect_host: String, websocket_host: String, path: String, kind: RouteKind) -> Self { Self { @@ -193,7 +216,7 @@ impl TransportEngine { &self, dc: u16, media: bool, - ) -> Result<(TelegramWebSocket, ConnectedRoute), String> { + ) -> Result<(TelegramConnection, ConnectedRoute), String> { let key = DcKey { dc, media }; self.race_connections(key, |route| async move { connect_route(&route).await }) .await @@ -242,7 +265,12 @@ impl TransportEngine { } Err(error) => { self.record_failure(&route); - errors.push(format!("{} (TLS {}) — {}", route.connect_host, route.websocket_host, error)); + let endpoint = if route.kind == RouteKind::TelegramTcp { + format!("{}:{} (MTProto TCP)", route.connect_host, route.port) + } else { + format!("{} (TLS {})", route.connect_host, route.websocket_host) + }; + errors.push(format!("{endpoint} — {error}")); } } } @@ -305,6 +333,7 @@ impl TransportEngine { candidates.sort_by_key(|route| { let preferred_rank = u8::from(preferred != Some(route)); let kind_rank = match route.kind { + RouteKind::TelegramTcp => 0, RouteKind::TelegramIp if Some(route) == all_routes.first() => 0, // The operator explicitly configured this independent path. // Start it after the primary IP, before more potentially @@ -421,7 +450,7 @@ fn telegram_ips(dc: u16) -> &'static [&'static str] { 3 => &["149.154.175.100"], 4 => &["149.154.167.91", "149.154.167.220"], 5 => &["149.154.171.5"], - 203 => &["91.105.192.100"], + 203 => &[CDN203_IP], _ => &[], } } @@ -440,6 +469,13 @@ pub fn routes_for_dc(dc: u16, media: bool) -> Vec { }; let ips = telegram_ips(dc); let mut routes = Vec::new(); + if dc == 203 { + // CDN203 speaks native obfuscated2 on this exact IP:443. A WebSocket + // handshake on the same address can time out even while MTProto is + // healthy. Keep the destination unchanged and use its native wire + // transport; never redirect CDN authorization to ordinary DC2. + routes.push(Route::cdn_tcp()); + } for websocket_host in &websocket_hosts { for (index, ip) in ips.iter().enumerate() { @@ -469,7 +505,7 @@ pub fn routes_for_dc(dc: u16, media: bool) -> Vec { routes } -async fn connect_route(route: &Route) -> Result { +async fn connect_route(route: &Route) -> Result { connect_route_with_config(route, tls_config()).await } @@ -499,7 +535,12 @@ fn tls_config() -> Arc { async fn connect_route_with_config( route: &Route, config: Arc, -) -> Result { +) -> Result { + // Native TCP is a narrowly scoped CDN route, not a general proxy escape + // hatch. Reject malformed raw routes before dialing anything. + if route.kind == RouteKind::TelegramTcp && !allowed_tcp_route(route) { + return Err("native MTProto TCP разрешён только для закреплённого CDN203".to_owned()); + } let tcp = tokio::time::timeout( CONNECT_TIMEOUT, TcpStream::connect((route.connect_host.as_str(), route.port)), @@ -510,6 +551,10 @@ async fn connect_route_with_config( tcp.set_nodelay(true) .map_err(|error| format!("TCP_NODELAY: {}", error))?; + if route.kind == RouteKind::TelegramTcp { + return Ok(TelegramConnection::Tcp(tcp)); + } + let scheme = if route.secure { "wss" } else { "ws" }; let url = format!("{}://{}{}", scheme, route.websocket_host, route.path); let mut request = url @@ -524,15 +569,15 @@ async fn connect_route_with_config( ); if !route.secure { - // Only reachable from tests, which run a local WebSocket server without - // a certificate. Production routes are always built by `Route::https`. + // Raw TCP has already returned above. Plain WebSocket is only used by + // local fixtures; production WebSocket routes are always HTTPS. return tokio::time::timeout( CONNECT_TIMEOUT, tokio_tungstenite::client_async(request, MaybeTlsStream::Plain(tcp)), ) .await .map_err(|_| "таймаут WebSocket".to_owned())? - .map(|(websocket, _)| websocket) + .map(|(websocket, _)| TelegramConnection::WebSocket(Box::new(websocket))) .map_err(|error| format!("рукопожатие WebSocket: {}", error)); } @@ -547,10 +592,22 @@ async fn connect_route_with_config( ) .await .map_err(|_| "таймаут TLS/WebSocket".to_owned())? - .map(|(websocket, _)| websocket) + .map(|(websocket, _)| TelegramConnection::WebSocket(Box::new(websocket))) .map_err(|error| format!("TLS/WebSocket handshake: {}", error)) } +fn allowed_tcp_route(route: &Route) -> bool { + #[cfg(test)] + if route.connect_host == "127.0.0.1" && !route.secure { + return true; + } + route.connect_host == CDN203_IP + && route.port == HTTPS_PORT + && route.websocket_host.is_empty() + && route.path.is_empty() + && !route.secure +} + fn valid_domain(domain: &str) -> bool { if domain.is_empty() || domain.len() > 253 || !domain.contains('.') { return false; @@ -578,10 +635,13 @@ mod tests { } #[test] - fn dc203_uses_dc2_websocket_and_its_own_ip() { + fn dc203_prefers_native_tcp_and_never_changes_the_cdn_destination() { let routes = routes_for_dc(203, false); - assert_eq!(routes[0].websocket_host, "kws2.web.telegram.org"); + assert_eq!(routes[0].kind, RouteKind::TelegramTcp); + assert!(routes[0].websocket_host.is_empty()); + assert!(!routes[0].secure); assert_eq!(routes[0].connect_host, "91.105.192.100"); + assert_eq!(routes[1].websocket_host, "kws2.web.telegram.org"); assert!(routes .iter() .all(|route| route.connect_host == "91.105.192.100")); @@ -594,6 +654,12 @@ mod tests { dc: 203, media: true, }); + let ordered = engine.ordered_candidates(DcKey { + dc: 203, + media: true, + }); + assert_eq!(ordered[0].kind, RouteKind::TelegramTcp); + assert_eq!(ordered[1].kind, RouteKind::CloudflareWorker); assert!(routes .iter() .filter(|route| route.kind == RouteKind::CloudflareWorker) @@ -648,7 +714,7 @@ mod tests { } #[test] - fn every_production_route_is_tls_on_443() { + fn production_websockets_use_tls_and_native_tcp_is_restricted_to_cdn203() { let engine = TransportEngine::new(); engine.set_worker_domains(&["fallback.workers.dev".to_owned()]); for dc in [1, 2, 3, 4, 5, 203] { @@ -657,7 +723,13 @@ mod tests { assert!(!routes.is_empty(), "DC{dc} must have at least one route"); for route in routes { assert_eq!(route.port, 443, "{route:?}"); - assert!(route.secure, "{route:?}"); + if route.kind == RouteKind::TelegramTcp { + assert_eq!(dc, 203); + assert_eq!(route.connect_host, CDN203_IP); + assert!(allowed_tcp_route(&route)); + } else { + assert!(route.secure, "{route:?}"); + } } } } @@ -902,6 +974,7 @@ mod tests { RouteKind::AlternateTelegramIp, RouteKind::SystemDns, RouteKind::CloudflareWorker, + RouteKind::TelegramTcp, ] { assert_eq!(RouteKind::from_ui_code(kind.ui_code()), Some(kind)); assert_eq!(route_label(kind.ui_code()), kind.label()); @@ -917,6 +990,7 @@ mod tests { RouteKind::AlternateTelegramIp, RouteKind::SystemDns, RouteKind::CloudflareWorker, + RouteKind::TelegramTcp, ] { assert_ne!(route_label(0), kind.label()); } @@ -1131,14 +1205,95 @@ mod tests { ); } + #[tokio::test] + async fn native_tcp_connects_without_sending_a_tls_or_websocket_handshake() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let engine = TransportEngine::new(); + engine.force_local_route_with( + listener.local_addr().unwrap().port(), + RouteKind::TelegramTcp, + String::new(), + ); + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut bytes = [0; 4]; + socket.read_exact(&mut bytes).await.unwrap(); + assert_eq!( + &bytes, b"init", + "first bytes must be native transport bytes" + ); + socket.write_all(b"pong").await.unwrap(); + }); + let (connection, connected) = + tokio::time::timeout(Duration::from_secs(1), engine.connect(203, true)) + .await + .unwrap() + .unwrap(); + assert_eq!(connected.route.kind, RouteKind::TelegramTcp); + let TelegramConnection::Tcp(mut socket) = connection else { + panic!("native CDN route must return a TCP stream"); + }; + assert!(socket.nodelay().unwrap()); + socket.write_all(b"init").await.unwrap(); + let mut reply = [0; 4]; + tokio::time::timeout(Duration::from_secs(1), socket.read_exact(&mut reply)) + .await + .unwrap() + .unwrap(); + assert_eq!(&reply, b"pong"); + server.await.unwrap(); + } + + #[tokio::test] + async fn raw_tcp_failure_is_named_and_enters_cooldown() { + let engine = TransportEngine::new(); + // Port zero cannot accept a TCP connection; no external traffic. + engine.force_local_route_with(0, RouteKind::TelegramTcp, String::new()); + let error = engine.connect(203, false).await.unwrap_err(); + assert!(error.contains("MTProto TCP"), "{error}"); + assert!( + !error.contains("TLS"), + "native transport must not be called TLS" + ); + assert_eq!(engine.route_failures(), 1); + assert!(engine + .ordered_candidates(DcKey { + dc: 203, + media: false + }) + .is_empty()); + } + + #[tokio::test] + async fn raw_tcp_guard_rejects_a_different_destination_before_dialing() { + let mut route = Route::cdn_tcp(); + route.connect_host = "127.0.0.2".to_owned(); + let error = connect_route(&route).await.unwrap_err(); + assert!(error.contains("только для закреплённого CDN203"), "{error}"); + for dc in [1, 2, 3, 4, 5, 999] { + assert!(routes_for_dc(dc, false) + .iter() + .all(|route| route.kind != RouteKind::TelegramTcp)); + } + } + #[tokio::test] #[ignore = "requires live Telegram network access"] async fn connects_to_all_production_data_centers() { let engine = TransportEngine::new(); for dc in [1, 2, 3, 4, 5, 203] { - let (mut websocket, connected) = engine.connect(dc, false).await.unwrap(); - assert!(!connected.route.websocket_host.is_empty()); - websocket.close(None).await.unwrap(); + let (connection, connected) = engine.connect(dc, false).await.unwrap(); + assert!(!connected.route.connect_host.is_empty()); + match connection { + TelegramConnection::WebSocket(mut websocket) => { + websocket.close(None).await.unwrap() + } + TelegramConnection::Tcp(mut socket) => { + use tokio::io::AsyncWriteExt; + socket.shutdown().await.unwrap(); + } + } } } }