diff --git a/Cargo.lock b/Cargo.lock index 935e804..d14d6a0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3822,7 +3822,7 @@ dependencies = [ [[package]] name = "tglock" -version = "2.0.0-beta.18" +version = "2.0.0-beta.19" dependencies = [ "aes", "cipher", diff --git a/Cargo.toml b/Cargo.toml index 280c5e6..743a9e9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tglock" -version = "2.0.0-beta.18" +version = "2.0.0-beta.19" edition = "2021" rust-version = "1.88" description = "Telegram unblock via local WebSocket tunnel" diff --git a/README.md b/README.md index 69c001d..bdd4915 100644 --- a/README.md +++ b/README.md @@ -217,6 +217,8 @@ worker = ["my-name.workers.dev"] Полный пример со всеми полями и пояснениями — [`tglock.example.toml`](tglock.example.toml). +Журнал можно писать в файл, а не в stdout: `log_file` и `log_max_mb` в настройках или `--log-file` и `--log-max-mb`. Заполненный файл переименовывается в `<имя>.1`, больше двух файлов не бывает. + Приоритеты: значения по умолчанию → файл → флаги. Флаг всегда сильнее файла. Флаги-переключатели (`--lan`, `--allow-direct`, `--quiet`) могут только включать: их отсутствие означает «взять из файла», а не «выключить». Опечатка в файле — ошибка при старте с перечислением допустимых полей, а не молча проигнорированная строка. Сервис, который из-за `porrt = 1443` слушает 1080, хуже сервиса, который не запустился. @@ -433,7 +435,7 @@ xattr -cr /Applications/TGLock.app Это реальный риск, и TGLock 2.0 к нему подготовлен настолько, насколько может. -Маршрут не один: пробуются закреплённые IP Telegram, дублёры `kwsN-1` и системный DNS. Пока жив хотя бы один — туннель поднимается. +Маршрут не один: пробуются закреплённые WebSocket-фронты Telegram, для медиа ещё и его отдельный кластер `kwsN-1`, затем системный DNS, WebSocket на порту 80 и родной MTProto TCP. Пока жив хотя бы один — туннель поднимается. Если у твоего провайдера легли **все** маршруты, есть запасной выход — **свой Cloudflare Worker**. Тогда соединение идёт на твой домен `*.workers.dev`, а воркер доводит его до Telegram; блокировать его провайдеру придётся отдельно. Готовый скрипт и пошаговая установка: **[docs/CLOUDFLARE_WORKER.md](docs/CLOUDFLARE_WORKER.md)**. Нужен только аккаунт Cloudflare, бесплатного тарифа хватает, свой сервер и домен не нужны. diff --git a/docs/ARCHITECTURE_V2.md b/docs/ARCHITECTURE_V2.md index 46aadb6..f74ec16 100644 --- a/docs/ARCHITECTURE_V2.md +++ b/docs/ARCHITECTURE_V2.md @@ -200,25 +200,55 @@ Ping приходит в читающую половину, а отвечает Для каждого DC и отдельно для media-соединений строится список маршрутов: 1. сохранённый успешный маршрут; -2. точный Telegram IP с `kwsN` или `kwsN-1` в TLS SNI и WebSocket Host; -3. явно настроенный пользователем Cloudflare Worker; -4. дополнительные Telegram IP и варианты хоста; -5. системный DNS (кроме CDN DC203). +2. основной WebSocket-фронт DC (`web_fronts`) с настоящим именем хоста в TLS SNI + и WebSocket Host; +3. явно настроенные пользователем CF-прокси и Cloudflare Worker; +4. запасные фронты DC; +5. системный DNS; +6. WebSocket без TLS на порту 80 адреса DC и родной MTProto TCP, только если всё + остальное не ответило. -Поддерживаются 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. +Адреса фронтов и родные адреса DC разные. `149.154.167.51`, `.91`, `149.154.175.50`, +`.100`, `149.154.171.5` и `91.105.192.100` говорят только на MTProto и на TLS не +отвечают, в том числе из сети без блокировок. WebSocket принимают `149.154.167.220` +и `.99` (DC2, DC4), `149.154.174.100` и `.200`, `149.154.175.209` (DC1, DC3), +`149.154.170.100`, `.200` и `.96` (DC5). + +Фронт выбирает дата-центр по имени хоста, а не по заголовку клиента, поэтому имя и +адрес нельзя менять местами. Какой DC стоит за маршрутом, проверяет +`probe_proxy.mjs --identity`: сервер сверяет номер DC внутри `req_DH_params` с собой +и отвечает `-444`, если номер чужой. `resPQ` этого не показывает, на него отвечает +любой сервер при любом номере в заголовке. + +Кластер `kwsN-1` только для медиа: обычное соединение он отвергает с `-444`, а +Telegram Desktop на этот код выключает прокси как настроенный неверно. Основной +кластер `kwsN` принимает оба вида соединений и остаётся запасным для медиа. + +Фронт разбирает один транспортный пакет на входящий кадр, остаток ждёт следующего +кадра. Клиенты пишут запросы пачками, поэтому поток режется на пакеты +(`PacketSplitter`), и каждый уходит отдельным кадром. Для Worker это не нужно: он +пишет кадры в обычный TCP. + +На порту 80 родные адреса принимают WebSocket без TLS. Там DC определяет адрес, а +не имя хоста, поэтому в `Host` идёт сам адрес и имя Telegram в открытом виде не +появляется. Защита та же, что у родного TCP: внутри тот же обфусцированный MTProto, +а сервер подтверждает себя в самом протоколе. Маршрут разрешён только до +закреплённых адресов. + +Поддерживаются DC1–5 и CDN DC203. У DC203 нет фронта на 443, а `kws2` на любом +фронте это обычный DC2, поэтому его маршруты: MTProto TCP к закреплённому +`91.105.192.100:443`, Worker, CF-прокси и порт 80 того же адреса. После ошибки маршрут получает exponential cooldown от 30 секунд до 30 минут. Успешный маршрут становится первым для следующего соединения того же DC и типа трафика. Попытки запускаются с интервалом 250 мс, не больше трёх одновременно на соединение; остальные отменяются после первого успеха. Если все маршруты на паузе, возвращается причина и время до повторной -попытки. Новое подключение клиента не обходит cooldown. Ошибка upstream -после handshake также снимает предпочтение маршрута и добавляет cooldown. +попытки. Новое подключение клиента не обходит cooldown. Маршрут тоже теряет +предпочтение и получает cooldown, если Telegram закрыл соединение с кодом ошибки, +туннель оборвался, не проработав 15 секунд, или клиент спросил, не дождался ответа +за 5 секунд и ушёл сам. Обрыв туннеля, который работал дольше и получал ответы, +маршрут не наказывает: это обычное дело, а пауза оставила бы весь DC без связи. ## TLS policy diff --git a/docs/LIVE_PROBE.md b/docs/LIVE_PROBE.md index d3db2c0..e92893e 100644 --- a/docs/LIVE_PROBE.md +++ b/docs/LIVE_PROBE.md @@ -40,6 +40,26 @@ 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. +## Confirm which data centre answers + +`--identity` continues the handshake with `req_DH_params`. The DC number sealed +inside it is checked by the server against itself, and a mismatch is answered +with transport error `-444`, so this does prove the backend. `resPQ` alone proves +nothing here: every server answers it for any DC in the header. The handshake +stops before an authorization key exists. CDN DC203 uses its own RSA keys, so +the report says `identity: "skipped: CDN keys"` for it. + +`--batched` implies `--identity` and sends `req_DH_params` glued behind another +packet in one write, the way real clients batch requests. A WebSocket frontend +reads one packet per frame, so a proxy that forwards the write as a single frame +times out here while a lone request still passes. Run it before every release: + +```sh +for dc in 1 2 -2 3 4 -4 5 203; do + node scripts/probe_proxy.mjs --port 18080 --secret-file /private/path/tglock-secret --dc $dc --batched +done +``` + ## Isolate CDN transport failures To distinguish an unavailable CDN WebSocket endpoint from an unavailable CDN diff --git a/docs/OPENWRT.md b/docs/OPENWRT.md index b716f5c..9a91a49 100644 --- a/docs/OPENWRT.md +++ b/docs/OPENWRT.md @@ -77,6 +77,21 @@ chmod 755 /etc/init.d/tglock logread -e tglock ``` +Состояние меняется часто, и stdout сервиса уходит в системный журнал. Чтобы TGLock +его не забивал, задайте отдельный файл в `/etc/tglock/tglock.toml`: + +```toml +# /tmp лежит в оперативной памяти: журнал не изнашивает flash и не переживает +# перезагрузку. Заполненный файл переименовывается в tglock.log.1, больше двух +# файлов на диске не бывает. +log_file = "/tmp/tglock.log" +log_max_mb = 1 +``` + +Смотреть его: `tail -f /tmp/tglock.log`. То же самое флагами: `--log-file`, +`--log-max-mb`. Ошибки запуска по-прежнему попадают в `logread`. Если журнал не +нужен совсем, есть `quiet = true`. + Обновление: остановите сервис, замените бинарь после проверки контрольной суммы, сохраните `/etc/tglock`, запустите сервис снова. Для отмены автозапуска используйте `/etc/init.d/tglock stop` и `/etc/init.d/tglock disable`. diff --git a/package-lock.json b/package-lock.json index 3c4c0b0..742b332 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "tglock-ui", - "version": "2.0.0-beta.18", + "version": "2.0.0-beta.19", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "tglock-ui", - "version": "2.0.0-beta.18", + "version": "2.0.0-beta.19", "dependencies": { "@tauri-apps/api": "^2.8.0" }, diff --git a/package.json b/package.json index 1f297ed..df61298 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "tglock-ui", "private": true, - "version": "2.0.0-beta.18", + "version": "2.0.0-beta.19", "type": "module", "scripts": { "test:worker": "node --experimental-vm-modules --test worker/tglock-worker.test.js", diff --git a/scripts/probe_proxy.mjs b/scripts/probe_proxy.mjs index 94e58a3..2e8357d 100644 --- a/scripts/probe_proxy.mjs +++ b/scripts/probe_proxy.mjs @@ -4,28 +4,34 @@ // 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 { createCipheriv, createDecipheriv, createHash, createPublicKey, 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] +const HELP = `Usage: node scripts/probe_proxy.mjs --secret-file PATH [--port 1080] [--dc 2] [--timeout-ms 15000] [--fragment-size 0] [--identity] [--batched] 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. +Without --identity this proves a protocol response, not DC identity, account operation or media downloads. +--identity continues with req_DH_params: the server checks the DC number sealed inside it against itself +and answers -444 on a mismatch, so this proves WHICH data centre is behind the route. The key is never finished. +--batched sends req_DH_params glued behind another packet in one write, the way real clients batch requests. +A WebSocket frontend reads one packet per frame, so a proxy that does not split the stream times out here. 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 result = { port: 1080, dc: 2, timeoutMs: 15000, fragmentSize: 0, identity: false, batched: false }; 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; } + if (name === "--identity") { result.identity = true; continue; } + if (name === "--batched") { result.identity = result.batched = true; continue; } const value = args[++i]; if (value === undefined) throw new Error(`Missing value for ${name}`); if (name === "--secret-file") result.secretFile = value; @@ -41,6 +47,110 @@ function options(args) { return result; } +// Production key from Telegram Desktop (mtproto_dc_options.cpp). CDN data centres +// use their own keys, so a CDN simply does not list this fingerprint. +const TELEGRAM_RSA_KEY = `-----BEGIN RSA PUBLIC KEY----- +MIIBCgKCAQEA6LszBcC1LGzyr992NzE0ieY+BSaOW622Aa9Bd4ZHLl+TuFQ4lo4g +5nKaMBwK/BIb9xUfg0Q29/2mgIR6Zr9krM7HjuIcCzFvDtr+L0GQjae9H0pRB2OO +62cECs5HKhT5DZ98K33vmWiLowc621dQuwKWSQKjWf50XYFw42h21P2KXUGyp2y/ ++aEyZ+uVgLLQbRA1dEjSDZ2iGRy12Mk5gpYc397aYp438fsJoHIgJ2lgMv5h7WY9 +t6N/byY9Nw9p21Og3AoXSL2q/2IJ1WRUhebgAdGVMlV1fkuOQoEzR7EdpqtQD9Cs +5+bfo3Nhmcyvk5ftB0WkJ9z6bNZ7yxrP8wIDAQAB +-----END RSA PUBLIC KEY-----`; + +const tlBytes = (value) => { + const prefixed = Buffer.concat([Buffer.from([value.length]), value]); + const long = Buffer.concat([Buffer.from([254, value.length & 255, value.length >> 8, 0]), value]); + const raw = value.length < 254 ? prefixed : long; + return Buffer.concat([raw, Buffer.alloc((4 - raw.length % 4) % 4)]); +}; +const toBigInt = (bytes) => BigInt(`0x${bytes.toString("hex")}`); +const rsaKey = () => { + const { n, e } = createPublicKey({ key: TELEGRAM_RSA_KEY, format: "pem", type: "pkcs1" }).export({ format: "jwk" }); + const modulus = Buffer.from(n, "base64url"); + const exponent = Buffer.from(e, "base64url"); + const fingerprint = createHash("sha1").update(tlBytes(modulus)).update(tlBytes(exponent)).digest().subarray(12); + return { modulus: toBigInt(modulus), exponent: toBigInt(exponent), fingerprint }; +}; +function modPow(base, power, modulus) { + let result = 1n; + for (let b = base % modulus, e = power; e > 0n; e >>= 1n, b = b * b % modulus) if (e & 1n) result = result * b % modulus; + return result; +} +function factorize(pq) { + const gcd = (a, b) => { while (b) [a, b] = [b, a % b]; return a; }; + for (let c = 1n; ; c++) { + const step = (x) => (x * x + c) % pq; + let slow = 2n, fast = 2n, divisor = 1n; + while (divisor === 1n) { slow = step(slow); fast = step(step(fast)); divisor = gcd(slow > fast ? slow - fast : fast - slow, pq); } + if (divisor !== pq) { const other = pq / divisor; return divisor < other ? [divisor, other] : [other, divisor]; } + } +} +function aesIgeEncrypt(data, key) { + const output = Buffer.alloc(data.length); + let previousCipher = Buffer.alloc(16), previousPlain = Buffer.alloc(16); + for (let offset = 0; offset < data.length; offset += 16) { + const block = data.subarray(offset, offset + 16), mixed = Buffer.alloc(16); + for (let i = 0; i < 16; i++) mixed[i] = block[i] ^ previousCipher[i]; + const encrypted = createCipheriv("aes-256-ecb", key, null).setAutoPadding(false).update(mixed); + for (let i = 0; i < 16; i++) output[offset + i] = encrypted[i] ^ previousPlain[i]; + previousCipher = output.subarray(offset, offset + 16); + previousPlain = block; + } + return output; +} +// RSA_PAD from https://core.telegram.org/mtproto/auth_key#41-rsa-paddata-server-public-key-mentioned-above-is-implemented-as-follows +function rsaPad(data, key) { + const padded = Buffer.concat([data, randomBytes(192 - data.length)]); + for (;;) { + const tempKey = randomBytes(32); + const hashed = Buffer.concat([Buffer.from(padded).reverse(), createHash("sha256").update(tempKey).update(padded).digest()]); + const encrypted = aesIgeEncrypt(hashed, tempKey); + const mask = createHash("sha256").update(encrypted).digest(); + const masked = Buffer.from(tempKey.map((byte, index) => byte ^ mask[index])); + const value = toBigInt(Buffer.concat([masked, encrypted])); + if (value < key.modulus) return Buffer.from(modPow(value, key.exponent, key.modulus).toString(16).padStart(512, "0"), "hex"); + } +} +function unencryptedPacket(body) { + const message = Buffer.alloc(20 + body.length); + const now = BigInt(Date.now()); + message.writeBigUInt64LE(((now / 1000n << 32n) | ((now % 1000n) * (1n << 32n) / 1000n)) & ~3n, 8); + message.writeUInt32LE(body.length, 16); + body.copy(message, 20); + const padded = Buffer.concat([message, randomBytes(7)]); + const length = Buffer.alloc(4); + length.writeUInt32LE(padded.length); + return Buffer.concat([length, padded]); +} +const reqPqBody = (nonce) => { + const constructor = Buffer.alloc(4); + constructor.writeUInt32LE(0xbe7e8ef1); + return Buffer.concat([constructor, nonce]); +}; +function reqDhParams(resPQ, nonce, dc, key) { + const [p, q] = factorize(toBigInt(resPQ.pq)); + const bigEndian = (value) => Buffer.from(value.toString(16).padStart(8, "0"), "hex"); + const dcField = Buffer.alloc(4); + dcField.writeInt32LE(dc); + const inner = Buffer.concat([Buffer.from("955ff5a9", "hex"), tlBytes(resPQ.pq), tlBytes(bigEndian(p)), tlBytes(bigEndian(q)), + nonce, resPQ.serverNonce, randomBytes(32), dcField]); + return Buffer.concat([Buffer.from("bee412d7", "hex"), nonce, resPQ.serverNonce, tlBytes(bigEndian(p)), tlBytes(bigEndian(q)), + key.fingerprint, tlBytes(rsaPad(inner, key))]); +} +function parseDhReply(frame) { + if (frame.length < 24 || frame.readBigUInt64LE() !== 0n) { + const code = frame.readInt32LE(); + throw new Error(code === -444 + ? "MTProto transport error -444: this route leads to the wrong data centre" + : `MTProto transport error ${code}`); + } + const constructor = frame.readUInt32LE(20); + if (constructor === 0x05162463) return null; // the reply to the packet batched in front + if (constructor !== 0xd0e8075c) throw new Error(`Unexpected reply 0x${constructor.toString(16)} to req_DH_params`); + return "accepted"; +} + function makeRequest(secret, dc) { let header; do { @@ -68,7 +178,7 @@ function makeRequest(secret, dc) { 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 }; + return { wire: Buffer.concat([wireHeader, encrypt.update(Buffer.concat([length, padded]))]), encrypt, decrypt, nonce }; } function parseResPQ(frame, nonce) { @@ -90,7 +200,11 @@ function parseResPQ(frame, nonce) { // 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 }; + const fingerprints = []; + for (let i = 0; i < count; i++) fingerprints.push(body.subarray(vectorOffset + 8 + i * 8, vectorOffset + 16 + i * 8)); + const handshake = { serverNonce: body.subarray(20, 36), pq: body.subarray(37, 37 + pqLength), fingerprints }; + return Object.defineProperty({ response: "resPQ", nonceMatches: true, rsaFingerprintCount: count }, + "handshake", { value: handshake, enumerable: false }); } async function probe(config) { @@ -101,7 +215,8 @@ async function probe(config) { 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 { wire, encrypt, decrypt, nonce } = makeRequest(secret, config.dc); + let resPQ = null; const started = Date.now(); return new Promise((resolve, reject) => { const socket = createConnection(config.directCdn @@ -118,7 +233,7 @@ async function probe(config) { 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); + const timer = setTimeout(() => finish(new Error(resPQ ? "Timed out waiting for the reply to req_DH_params" : "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) => { @@ -138,8 +253,23 @@ async function probe(config) { pending = pending.subarray(4 + length); // padded intermediate quick ACK continue; } - finish(null, parseResPQ(pending.subarray(4, 4 + length), nonce)); - return; + const frame = pending.subarray(4, 4 + length); + pending = pending.subarray(4 + length); + if (resPQ) { + const verdict = parseDhReply(frame); + if (verdict) { finish(null, { ...resPQ, identity: verdict, signedDc: config.dc, batched: config.batched }); return; } + continue; + } + resPQ = parseResPQ(frame, nonce); + if (!config.identity) { finish(null, resPQ); return; } + const key = rsaKey(); + if (!resPQ.handshake.fingerprints.some((fingerprint) => fingerprint.equals(key.fingerprint))) { + finish(null, { ...resPQ, identity: "skipped: CDN keys" }); + return; + } + const request = unencryptedPacket(reqDhParams(resPQ.handshake, nonce, config.dc, key)); + const batch = config.batched ? [unencryptedPacket(reqPqBody(randomBytes(16))), request] : [request]; + socket.write(encrypt.update(Buffer.concat(batch))); } } catch (error) { finish(error); } }); @@ -177,6 +307,9 @@ function selfTest() { oversizedVector.writeUInt32LE(65, 72); assert.throws(() => parseResPQ(oversizedVector, nonce), /count/); assert.throws(() => parseResPQ(Buffer.from("6cfeffff", "hex"), nonce), /-404/); + assert.throws(() => parseDhReply(Buffer.from("44feffff", "hex")), /wrong data centre/); + assert.equal(rsaKey().fingerprint.toString("hex"), Buffer.from("d09d1d85de64fd85", "hex").reverse().toString("hex")); + assert.deepEqual(factorize(0x17ED48941A08F981n), [0x494C553Bn, 0x53911073n]); console.log("Offline parser checks passed; no network connection made."); } diff --git a/src/bin/cli.rs b/src/bin/cli.rs index 90ad75c..a27a2be 100644 --- a/src/bin/cli.rs +++ b/src/bin/cli.rs @@ -6,12 +6,14 @@ //! (by-sonic/tglock#10, by-sonic/tglock#17). use clap::Parser; +use std::fs::File; +use std::io::Write; use std::net::IpAddr; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::process::ExitCode; use std::sync::atomic::Ordering; -use std::sync::Arc; -use std::time::Duration; +use std::sync::{Arc, Mutex, OnceLock}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tglock_lib::cli_settings as config; use tglock_lib::{proxy, transport}; @@ -23,11 +25,144 @@ const STATUS_POLL: Duration = Duration::from_secs(1); /// пускают в `head`, закрывают терминал, перезапускают сборщик логов. Падать из /// за этого туннель не должен — он продолжает работать молча. fn say(text: &str) -> bool { - use std::io::Write; + if let Some(log) = LOG_FILE.get() { + // `false` здесь значило бы «писать больше некому», а это не так: диск + // могло временно заполнить. Журнал продолжается со следующей строки. + log.lock().unwrap().record(text); + return true; + } let mut out = std::io::stdout().lock(); writeln!(out, "{text}").and_then(|()| out.flush()).is_ok() } +/// Журнал вместо stdout, если задан `--log-file`. +static LOG_FILE: OnceLock> = OnceLock::new(); + +/// Файл журнала с ротацией. +/// +/// На роутере stdout сервиса уходит в системный журнал, и TGLock забивал его +/// строками состояния (by-sonic/tglock#53). Заполненный файл переименовывается +/// в `<имя>.1`, поэтому на диске лежит не больше двух файлов: flash и память +/// роутера не резиновые. +struct LogFile { + path: PathBuf, + /// `None` только внутри ротации: Windows не переименует открытый файл. + file: Option, + written: u64, + limit: u64, + /// Об ошибке записи сообщается один раз, а не на каждой строке. + failing: bool, +} + +impl LogFile { + fn open(path: PathBuf, limit: u64) -> std::io::Result { + let file = open_log(&path)?; + let written = file.metadata()?.len(); + Ok(Self { + path, + file: Some(file), + written, + limit, + failing: false, + }) + } + + fn record(&mut self, text: &str) { + match self.write_line(text) { + Ok(()) => self.failing = false, + Err(error) if !self.failing => { + self.failing = true; + eprintln!( + "tglock-cli: журнал {} не пишется: {error}", + self.path.display() + ); + } + Err(_) => {} + } + } + + fn write_line(&mut self, text: &str) -> std::io::Result<()> { + let line = format!("{} {text}\n", timestamp(now_seconds())); + if self.written > 0 && self.written + line.len() as u64 > self.limit { + self.rotate()?; + } + if self.file.is_none() { + self.file = Some(open_log(&self.path)?); + } + let file = self.file.as_mut().ok_or(std::io::ErrorKind::NotFound)?; + file.write_all(line.as_bytes())?; + file.flush()?; + self.written += line.len() as u64; + Ok(()) + } + + fn rotate(&mut self) -> std::io::Result<()> { + let mut backup = self.path.clone().into_os_string(); + backup.push(".1"); + self.file = None; + // rename заменяет старый бэкап сам, и если он не удался, бэкап цел. + let renamed = std::fs::rename(&self.path, &backup); + let file = open_log(&self.path)?; + // Файл могли удалить или держать другой программой. Тогда предел + // важнее старых строк: журнал не должен съесть память роутера. + if renamed.is_err() { + // Отдельным открытием: файл для дозаписи на Windows нельзя обрезать. + File::options() + .write(true) + .truncate(true) + .open(&self.path)?; + } + self.file = Some(file); + self.written = 0; + Ok(()) + } +} + +/// В журнал попадает ссылка `tg://proxy` с секретом: читать его может только +/// владелец, даже если файл появился раньше и с другими правами. +fn open_log(path: &Path) -> std::io::Result { + let mut options = std::fs::OpenOptions::new(); + options.create(true).append(true); + #[cfg(unix)] + std::os::unix::fs::OpenOptionsExt::mode(&mut options, 0o600); + let file = options.open(path)?; + #[cfg(unix)] + file.set_permissions(std::os::unix::fs::PermissionsExt::from_mode(0o600))?; + Ok(file) +} + +fn now_seconds() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +/// `2026-10-31 09:30:00` в UTC. Перевод дней в дату — civil_from_days из +/// https://howardhinnant.github.io/date_algorithms.html +fn timestamp(seconds: u64) -> String { + let days = (seconds / 86_400) as i64 + 719_468; + let era = days.div_euclid(146_097); + let day_of_era = days.rem_euclid(146_097); + let year_of_era = + (day_of_era - day_of_era / 1_460 + day_of_era / 36_524 - day_of_era / 146_096) / 365; + let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100); + let shifted_month = (5 * day_of_year + 2) / 153; + let day = day_of_year - (153 * shifted_month + 2) / 5 + 1; + let month = if shifted_month < 10 { + shifted_month + 3 + } else { + shifted_month - 9 + }; + let year = year_of_era + era * 400 + i64::from(month <= 2); + format!( + "{year:04}-{month:02}-{day:02} {:02}:{:02}:{:02}", + (seconds / 3600) % 24, + (seconds / 60) % 60, + seconds % 60 + ) +} + #[derive(Debug, Parser)] #[command( name = "tglock-cli", @@ -72,6 +207,15 @@ struct Args { /// Печатать только ошибки #[arg(short, long)] quiet: bool, + + /// Писать журнал в этот файл, а не в stdout. Ошибки запуска всё равно идут + /// в stderr. Заполненный файл переименовывается в <имя>.1 + #[arg(long, value_name = "PATH")] + log_file: Option, + + /// Размер файла журнала в МБ, после которого он сменяется. По умолчанию 5 + #[arg(long, value_name = "MB")] + log_max_mb: Option, } impl Args { @@ -85,6 +229,8 @@ impl Args { cf_proxy: self.cf_proxy.clone(), secret_file: self.secret_file.clone(), quiet: self.quiet, + log_file: self.log_file.clone(), + log_max_mb: self.log_max_mb, } } @@ -129,6 +275,11 @@ async fn serve(args: Args) -> Result<(), String> { let listen = settings.listen; let stats = settings.stats(); let quiet = settings.quiet; + if let Some(path) = &settings.log_file { + let log = LogFile::open(path.clone(), u64::from(settings.log_max_mb) * 1024 * 1024) + .map_err(|error| format!("не удалось открыть журнал {}: {error}", path.display()))?; + let _ = LOG_FILE.set(Mutex::new(log)); + } // Bind before printing anything: a busy port must be an error, not a // daemon that reports success and silently does nothing. @@ -291,6 +442,114 @@ mod tests { config::resolve(config::FileConfig::default(), parse(args).overrides()).unwrap() } + /// Пустой каталог для одного теста, удаляется вместе с ним. + struct TempDir(PathBuf); + + impl TempDir { + fn new(name: &str) -> Self { + let path = std::env::temp_dir().join(format!( + "tglock-cli-{name}-{}-{:?}", + std::process::id(), + std::thread::current().id() + )); + let _ = std::fs::remove_dir_all(&path); + std::fs::create_dir_all(&path).unwrap(); + Self(path) + } + } + + impl Drop for TempDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } + } + + #[test] + fn log_rotation_keeps_the_newest_lines_in_two_bounded_files() { + let dir = TempDir::new("rotation"); + let path = dir.0.join("tglock.log"); + let mut log = LogFile::open(path.clone(), 200).unwrap(); + for line in 0..40 { + log.write_line(&format!("line {line}")).unwrap(); + } + let current = std::fs::read_to_string(&path).unwrap(); + let backup = std::fs::read_to_string(dir.0.join("tglock.log.1")).unwrap(); + assert!(current.contains("line 39"), "{current}"); + assert!(current.len() <= 200 && backup.len() <= 200); + assert_eq!(std::fs::read_dir(&dir.0).unwrap().count(), 2); + } + + #[test] + fn a_log_deleted_by_hand_is_recreated_and_stays_bounded() { + let dir = TempDir::new("deleted"); + let path = dir.0.join("tglock.log"); + let mut log = LogFile::open(path.clone(), 200).unwrap(); + log.write_line("before").unwrap(); + // На Windows открытый файл не удалить, поэтому он закрывается как при + // ротации. На unix запись шла бы в удалённый файл до предела. + log.file = None; + std::fs::remove_file(&path).unwrap(); + for line in 0..40 { + log.write_line(&format!("line {line}")).unwrap(); + } + let current = std::fs::read_to_string(&path).unwrap(); + assert!(current.contains("line 39"), "{current}"); + assert!(current.len() <= 200); + } + + #[test] + fn a_failed_rotation_keeps_the_old_backup_and_the_size_limit() { + let dir = TempDir::new("failed-rotation"); + let path = dir.0.join("tglock.log"); + let mut log = LogFile::open(path.clone(), 200).unwrap(); + std::fs::write(dir.0.join("tglock.log.1"), "old backup").unwrap(); + log.write_line("x").unwrap(); + // Файл исчез до ротации: переименовывать нечего. + log.file = None; + std::fs::remove_file(&path).unwrap(); + log.rotate().unwrap(); + assert_eq!( + std::fs::read_to_string(dir.0.join("tglock.log.1")).unwrap(), + "old backup" + ); + assert_eq!(log.written, 0); + } + + #[cfg(unix)] + #[test] + fn an_existing_log_with_wide_permissions_is_tightened() { + use std::os::unix::fs::PermissionsExt; + let dir = TempDir::new("permissions"); + let path = dir.0.join("tglock.log"); + std::fs::write(&path, "").unwrap(); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap(); + LogFile::open(path.clone(), 200).unwrap(); + let mode = std::fs::metadata(&path).unwrap().permissions().mode(); + assert_eq!(mode & 0o777, 0o600); + } + + #[test] + fn a_restarted_daemon_continues_the_log_and_counts_what_is_there() { + let dir = TempDir::new("restart"); + let path = dir.0.join("tglock.log"); + LogFile::open(path.clone(), 10_000) + .unwrap() + .write_line("first run") + .unwrap(); + let mut log = LogFile::open(path.clone(), 10_000).unwrap(); + assert!(log.written > 0); + log.write_line("second run").unwrap(); + let text = std::fs::read_to_string(&path).unwrap(); + assert!(text.contains("first run") && text.contains("second run")); + } + + #[test] + fn timestamps_are_calendar_dates_in_utc() { + assert_eq!(timestamp(0), "1970-01-01 00:00:00"); + assert_eq!(timestamp(951_782_400), "2000-02-29 00:00:00"); + assert_eq!(timestamp(1_793_442_645), "2026-10-31 10:30:45"); + } + #[test] fn command_definition_is_valid() { Args::command().debug_assert(); diff --git a/src/cli_settings.rs b/src/cli_settings.rs index 3a01c3c..a4afacc 100644 --- a/src/cli_settings.rs +++ b/src/cli_settings.rs @@ -20,6 +20,9 @@ use std::path::{Path, PathBuf}; /// Имя файла, который ищется рядом с бинарём, если `--config` не задан. pub const DEFAULT_FILE_NAME: &str = "tglock.toml"; +/// Размер журнала, после которого он сменяется. Как в Flowseal/tg-ws-proxy. +pub const DEFAULT_LOG_MAX_MB: u32 = 5; + /// Содержимое файла конфигурации. /// /// `deny_unknown_fields` намеренно: опечатка вроде `porrt = 1080` должна быть @@ -38,6 +41,9 @@ pub struct FileConfig { pub secret: Option, pub secret_file: Option, pub quiet: Option, + /// Писать журнал в файл, а не в stdout (by-sonic/tglock#53). + pub log_file: Option, + pub log_max_mb: Option, } impl FileConfig { @@ -63,6 +69,8 @@ pub struct Overrides { pub cf_proxy: Vec, pub secret_file: Option, pub quiet: bool, + pub log_file: Option, + pub log_max_mb: Option, } /// Откуда взять секрет прокси. @@ -86,6 +94,8 @@ pub struct Resolved { pub cf_proxies: String, pub secret: SecretSource, pub quiet: bool, + pub log_file: Option, + pub log_max_mb: u32, } impl Resolved { @@ -139,6 +149,14 @@ pub fn resolve(file: FileConfig, cli: Overrides) -> Result { cli.cf_proxy }; + let log_max_mb = cli + .log_max_mb + .or(file.log_max_mb) + .unwrap_or(DEFAULT_LOG_MAX_MB); + if log_max_mb == 0 { + return Err("log_max_mb должен быть не меньше 1".to_owned()); + } + // Секрет из файла конфигурации важнее отдельного файла: если человек вписал // его сюда, значит хотел держать всё в одном месте. let secret = match (&file.secret, cli.secret_file.or(file.secret_file)) { @@ -156,6 +174,8 @@ pub fn resolve(file: FileConfig, cli: Overrides) -> Result { cf_proxies: cf_proxies.join(","), secret, quiet: cli.quiet || file.quiet.unwrap_or(false), + log_file: cli.log_file.or(file.log_file), + log_max_mb, }) } @@ -189,6 +209,31 @@ mod tests { assert!(!resolved.quiet); } + #[test] + fn log_file_comes_from_the_file_and_a_flag_overrides_it() { + let from_file = + || FileConfig::parse("log_file = \"/tmp/tglock.log\"\nlog_max_mb = 2").unwrap(); + let resolved = resolve(from_file(), cli()).unwrap(); + assert_eq!(resolved.log_file, Some(PathBuf::from("/tmp/tglock.log"))); + assert_eq!(resolved.log_max_mb, 2); + + let flags = Overrides { + log_file: Some(PathBuf::from("/var/log/t.log")), + log_max_mb: Some(1), + ..cli() + }; + let resolved = resolve(from_file(), flags).unwrap(); + assert_eq!(resolved.log_file, Some(PathBuf::from("/var/log/t.log"))); + assert_eq!(resolved.log_max_mb, 1); + } + + #[test] + fn a_zero_log_limit_is_refused() { + // Иначе файл менялся бы на каждой строке. + let file = FileConfig::parse("log_max_mb = 0").unwrap(); + assert!(resolve(file, cli()).unwrap_err().contains("log_max_mb")); + } + #[test] fn file_values_are_used_when_no_flags_are_given() { let file = FileConfig::parse( diff --git a/src/mtproto.rs b/src/mtproto.rs index f5b626c..d7fab47 100644 --- a/src/mtproto.rs +++ b/src/mtproto.rs @@ -354,6 +354,138 @@ fn build_crypto_context( }) } +/// Длина, больше которой пакет считается не настоящим. Файлы грузятся частями +/// по 512 КиБ. Предел держит и память: в LAN-режиме чужое устройство могло бы +/// заставить копить «пакет» на роутере. Если настоящий пакет окажется длиннее, +/// поток просто пойдёт дальше без нарезки. +const MAX_PACKET: usize = 2 * 1024 * 1024; + +#[derive(Clone, Copy)] +enum Framing { + Abridged, + Intermediate, +} + +/// Режет поток клиента на транспортные пакеты MTProto. +/// +/// WebSocket-фронт Telegram разбирает один пакет на входящий кадр, а остаток +/// кадра держит до следующего. Клиент пишет запросы пачками, и всё, что шло +/// вторым в той же записи TCP, застревало на фронте: текст проходил, а файлы +/// вставали. Проверено на живом фронте: `req_DH_params` вторым пакетом в кадре +/// остаётся без ответа, пока не придёт следующий кадр, а отдельным кадром +/// принимается сразу. Так же режет поток Flowseal/tg-ws-proxy. +/// +/// Заголовок читается по копии: ключи `init` открыты, поэтому режим с секретом +/// и SOCKS5 разбираются одинаково, а наружу уходят исходные байты. +pub struct PacketSplitter { + framing: Option, + decrypt: AesCtr, + wire: Vec, + plain: Vec, +} + +impl PacketSplitter { + /// Разбирать поток, начатый этим `init`. Неизвестный протокол не режется. + pub fn from_init(init: &[u8; INIT_LEN]) -> Self { + let key: [u8; 32] = init[KEY_START..KEY_END].try_into().unwrap(); + let iv: [u8; 16] = init[KEY_END..IV_END].try_into().unwrap(); + let mut decrypt = AesCtr::new((&key).into(), (&iv).into()); + let mut plain_init = *init; + decrypt.apply_keystream(&mut plain_init); + let framing = match [ + plain_init[TAG_START], + plain_init[TAG_START + 1], + plain_init[TAG_START + 2], + plain_init[TAG_START + 3], + ] { + ABRIDGED => Some(Framing::Abridged), + INTERMEDIATE | PADDED_INTERMEDIATE => Some(Framing::Intermediate), + _ => None, + }; + Self { + framing, + decrypt, + wire: Vec::new(), + plain: Vec::new(), + } + } + + /// Принять зашифрованные байты, вернуть готовые к отправке пакеты. + pub fn push(&mut self, wire: &[u8]) -> Vec> { + let Some(framing) = self.framing else { + return vec![wire.to_vec()]; + }; + let start = self.plain.len(); + self.wire.extend_from_slice(wire); + self.plain.extend_from_slice(wire); + self.decrypt.apply_keystream(&mut self.plain[start..]); + + let mut packets = Vec::new(); + let mut offset = 0; + while offset < self.plain.len() { + match packet_len(framing, &self.plain[offset..]) { + PacketLen::Incomplete => break, + PacketLen::Complete(length) => { + if self.plain.len() - offset < length { + break; + } + packets.push(self.wire[offset..offset + length].to_vec()); + offset += length; + } + // Длина не похожа на настоящую: поток не наш. Дальше он идёт + // как есть, иначе данные встанут в ожидании несуществующего конца. + PacketLen::Invalid => { + packets.push(self.wire[offset..].to_vec()); + offset = self.plain.len(); + self.framing = None; + } + } + } + self.wire.drain(..offset); + self.plain.drain(..offset); + packets + } + + /// Хвост незаконченного пакета: клиент закрыл соединение на середине. + pub fn finish(&mut self) -> Option> { + self.plain.clear(); + (!self.wire.is_empty()).then(|| std::mem::take(&mut self.wire)) + } +} + +enum PacketLen { + Incomplete, + Complete(usize), + Invalid, +} + +fn packet_len(framing: Framing, plain: &[u8]) -> PacketLen { + let (header, payload) = match framing { + Framing::Abridged => match plain { + [] => return PacketLen::Incomplete, + [0x7f | 0xff, length @ ..] => match length { + [a, b, c, ..] => (4, u32::from_le_bytes([*a, *b, *c, 0]) as usize * 4), + _ => return PacketLen::Incomplete, + }, + [first, ..] => (1, (*first & 0x7f) as usize * 4), + }, + Framing::Intermediate => match plain { + [a, b, c, d, ..] => { + // Старший бит просит quick ack и к длине не относится. + ( + 4, + (u32::from_le_bytes([*a, *b, *c, *d]) & 0x7fff_ffff) as usize, + ) + } + _ => return PacketLen::Incomplete, + }, + }; + if payload == 0 || payload > MAX_PACKET { + return PacketLen::Invalid; + } + PacketLen::Complete(header + payload) +} + #[cfg(test)] pub(crate) fn test_client_init(secret: &[u8; 16], dc_index: i16) -> [u8; INIT_LEN] { tests::generate_client_init(secret, PADDED_INTERMEDIATE, dc_index) @@ -423,6 +555,118 @@ pub(crate) fn test_relay_peer(relay_init: &[u8; INIT_LEN]) -> TestPeer { mod tests { use super::*; + /// Поток, каким его увидел бы фронт Telegram, и разделитель для него. + fn splitter_for(tag: [u8; 4], plain: &[u8]) -> (PacketSplitter, Vec) { + let init = generate_relay_init(tag, 2); + let mut wire = plain.to_vec(); + test_relay_peer(&init).decrypt(&mut wire); + (PacketSplitter::from_init(&init), wire) + } + + fn intermediate(payload: &[u8]) -> Vec { + let mut packet = (payload.len() as u32).to_le_bytes().to_vec(); + packet.extend_from_slice(payload); + packet + } + + #[test] + fn glued_packets_leave_one_per_frame() { + let first = intermediate(&[1; 40]); + let second = intermediate(&[2; 340]); + let plain = [first.clone(), second.clone()].concat(); + for tag in [INTERMEDIATE, PADDED_INTERMEDIATE] { + let (mut splitter, wire) = splitter_for(tag, &plain); + let packets = splitter.push(&wire); + assert_eq!(packets.len(), 2); + assert_eq!(packets[0], wire[..first.len()]); + assert_eq!(packets[1], wire[first.len()..]); + assert_eq!(splitter.finish(), None); + } + } + + #[test] + fn a_packet_is_held_until_it_is_complete() { + let packet = intermediate(&[7; 300]); + let (mut splitter, wire) = splitter_for(INTERMEDIATE, &packet); + // Границы TCP не совпадают с пакетами: режем в том числе внутри длины. + assert!(splitter.push(&wire[..2]).is_empty()); + assert!(splitter.push(&wire[2..150]).is_empty()); + assert_eq!(splitter.push(&wire[150..]), vec![wire.clone()]); + } + + #[test] + fn a_tail_of_one_packet_and_the_start_of_the_next_stay_separate() { + let first = intermediate(&[1; 40]); + let second = intermediate(&[2; 80]); + let plain = [first.clone(), second.clone()].concat(); + let (mut splitter, wire) = splitter_for(INTERMEDIATE, &plain); + let packets = splitter.push(&wire[..first.len() + 10]); + assert_eq!(packets, vec![wire[..first.len()].to_vec()]); + let packets = splitter.push(&wire[first.len() + 10..]); + assert_eq!(packets, vec![wire[first.len()..].to_vec()]); + } + + #[test] + fn quick_ack_bit_is_not_part_of_the_length() { + let mut packet = intermediate(&[3; 16]); + packet[3] |= 0x80; + let (mut splitter, wire) = splitter_for(INTERMEDIATE, &packet); + assert_eq!(splitter.push(&wire), vec![wire.clone()]); + } + + #[test] + fn abridged_lengths_count_in_words_with_either_header() { + let short = [vec![2], vec![9; 8]].concat(); + let long = [vec![0x7f, 0x80, 0, 0], vec![9; 512]].concat(); + let quick_ack = [vec![0x81], vec![9; 4]].concat(); + let plain = [short.clone(), long.clone(), quick_ack.clone()].concat(); + let (mut splitter, wire) = splitter_for(ABRIDGED, &plain); + let lengths: Vec<_> = splitter.push(&wire).iter().map(Vec::len).collect(); + assert_eq!(lengths, [short.len(), long.len(), quick_ack.len()]); + } + + #[test] + fn a_length_that_cannot_be_real_stops_the_splitting_without_losing_bytes() { + // Если длина мусорная, ждать конца пакета значит заморозить соединение. + let mut garbage = 0x7fff_fff0_u32.to_le_bytes().to_vec(); + garbage.extend_from_slice(&[5; 32]); + let (mut splitter, wire) = splitter_for(INTERMEDIATE, &garbage); + assert_eq!(splitter.push(&wire), vec![wire.clone()]); + assert_eq!(splitter.push(b"more"), vec![b"more".to_vec()]); + } + + #[test] + fn a_length_over_the_limit_is_never_buffered() { + let mut oversized = ((MAX_PACKET + 1) as u32).to_le_bytes().to_vec(); + oversized.extend_from_slice(&[5; 32]); + let (mut splitter, wire) = splitter_for(INTERMEDIATE, &oversized); + assert_eq!(splitter.push(&wire), vec![wire.clone()]); + assert_eq!(splitter.finish(), None); + } + + #[test] + fn the_largest_real_upload_part_still_arrives_whole() { + let packet = intermediate(&vec![7; 512 * 1024 + 256]); + let (mut splitter, wire) = splitter_for(INTERMEDIATE, &packet); + assert!(splitter.push(&wire[..65536]).is_empty()); + assert_eq!(splitter.push(&wire[65536..]), vec![wire.clone()]); + } + + #[test] + fn an_unknown_protocol_passes_through_untouched() { + let (mut splitter, wire) = splitter_for([0x11; 4], &[0; 16]); + assert_eq!(splitter.push(&wire), vec![wire.clone()]); + } + + #[test] + fn the_unfinished_tail_is_returned_once() { + let packet = intermediate(&[1; 40]); + let (mut splitter, wire) = splitter_for(INTERMEDIATE, &packet); + assert!(splitter.push(&wire[..10]).is_empty()); + assert_eq!(splitter.finish(), Some(wire[..10].to_vec())); + assert_eq!(splitter.finish(), None); + } + pub(super) fn generate_client_init( secret: &[u8; 16], protocol_tag: [u8; 4], diff --git a/src/proxy.rs b/src/proxy.rs index 3dcf7d5..174f25a 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -772,6 +772,7 @@ async fn ws_tunnel( ws_w.send(tungstenite::Message::Binary(init.to_vec())) .await .map_err(|error| tunnel_route_error(stats, &connected, dc, &error.to_string()))?; + let health = TunnelHealth::new(); // Ping приходит в половину, которая читает, а отвечать на него должна та, // которая пишет: владелец у отправляющей половины строго один. @@ -785,11 +786,12 @@ async fn ws_tunnel( // сессия встаёт при живом туннеле (by-sonic/tglock#42, #32). let downstream = async { let mut crypto = downstream_crypto; - let mut received_payload = false; while let Some(message) = ws_r.next().await { match message { Ok(tungstenite::Message::Binary(mut data)) => { - received_payload |= !data.is_empty(); + if !data.is_empty() { + health.note_reply(); + } if let Some(crypto) = &mut crypto { crypto.apply(data.as_mut()); } @@ -813,7 +815,7 @@ async fn ws_tunnel( .is_ok_and(|code| code != 0) || frame.code != tungstenite::protocol::frame::coding::CloseCode::Normal }); - if !received_payload || transport_error { + if !health.replied() || transport_error { let reason = match frame { Some(frame) => describe_ws_close(frame.code.into(), &frame.reason), None => "WebSocket закрыт без ответа Telegram".to_owned(), @@ -823,17 +825,18 @@ async fn ws_tunnel( return Ok(()); } Err(error) => { - return Err(tunnel_route_error( + return Err(tunnel_broke( stats, &connected, dc, &error.to_string(), + &health, )); } Ok(_) => {} } } - if !received_payload { + if !health.replied() { return Err(tunnel_route_error( stats, &connected, @@ -846,32 +849,54 @@ async fn ws_tunnel( let upstream = async { let mut crypto = upstream_crypto; + // Worker пишет кадры в обычный TCP до Telegram, там они ничего не значат. + // Отдельный кадр на пакет нужен только WebSocket Telegram, на 443 и на 80. + let mut splitter = (connected.route.kind != crate::transport::RouteKind::CloudflareWorker) + .then(|| crate::mtproto::PacketSplitter::from_init(init)); let mut buf = vec![0u8; 65536]; loop { tokio::select! { read = tcp_r.read(&mut buf) => match read { - Ok(0) => break, + Ok(0) => { + if let Some(tail) = splitter.as_mut().and_then(|splitter| splitter.finish()) { + let _ = ws_w.send(tungstenite::Message::Binary(tail)).await; + } + break; + } Err(error) => return Err(error.into()), Ok(read) => { + health.note_request(); if let Some(crypto) = &mut crypto { crypto.apply(&mut buf[..read]); } - ws_w - .send(tungstenite::Message::Binary(buf[..read].to_vec())) - .await - .map_err(|error| tunnel_route_error(stats, &connected, dc, &error.to_string()))?; + let packets = match &mut splitter { + Some(splitter) => splitter.push(&buf[..read]), + None => vec![buf[..read].to_vec()], + }; + for packet in packets { + ws_w.feed(tungstenite::Message::Binary(packet)).await.map_err(|error| { + tunnel_broke(stats, &connected, dc, &error.to_string(), &health) + })?; + } + ws_w.flush().await.map_err(|error| { + tunnel_broke(stats, &connected, dc, &error.to_string(), &health) + })?; } }, payload = pong_rx.recv() => match payload { Some(payload) => { - ws_w.send(tungstenite::Message::Pong(payload)).await - .map_err(|error| tunnel_route_error(stats, &connected, dc, &error.to_string()))?; + ws_w.send(tungstenite::Message::Pong(payload)).await.map_err(|error| { + tunnel_broke(stats, &connected, dc, &error.to_string(), &health) + })?; } None => break, }, } } let _ = ws_w.close().await; + if health.went_silent() { + return Err(tunnel_route_error(stats, &connected, dc, SILENT_ROUTE)); + } Ok::<(), Box>(()) }; @@ -905,34 +930,38 @@ async fn cdn_tcp_tunnel( Some((upstream, downstream)) => (Some(upstream), Some(downstream)), None => (None, None), }; + let health = TunnelHealth::new(); 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 { + if health.went_silent() { + return Err(tunnel_route_error(stats, connected, dc, SILENT_ROUTE)); + } return Ok::<_, Box>(()); } + health.note_request(); 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()))?; + .map_err(|error| tunnel_broke(stats, connected, dc, &error.to_string(), &health))?; } }; 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()))?; + .map_err(|error| tunnel_broke(stats, connected, dc, &error.to_string(), &health))?; if read == 0 { - return if received_payload { + return if health.replied() { Ok(()) } else { Err(tunnel_route_error( @@ -943,7 +972,7 @@ async fn cdn_tcp_tunnel( )) }; } - received_payload = true; + health.note_reply(); if let Some(crypto) = &mut crypto { crypto.apply(&mut buffer[..read]); } @@ -989,6 +1018,92 @@ fn tunnel_route_error( message.into() } +const SILENT_ROUTE: &str = "Telegram не ответил на запрос, клиент ушёл сам"; + +/// Сколько туннель должен проработать, чтобы его обрыв не считался поломкой +/// маршрута. Один ответ ничего не доказывает: DPI режет и после первых байт. +const PROVEN_TUNNEL: Duration = if cfg!(test) { + Duration::from_millis(500) +} else { + Duration::from_secs(15) +}; + +/// Сколько Telegram может молчать в ответ на запрос. Он отвечает за доли секунды. +const SILENT_TUNNEL: Duration = if cfg!(test) { + Duration::from_millis(500) +} else { + Duration::from_secs(5) +}; + +/// Что туннель показал о своём маршруте. +struct TunnelHealth { + opened: std::time::Instant, + requested: AtomicBool, + replied: AtomicBool, +} + +impl TunnelHealth { + fn new() -> Self { + Self { + opened: std::time::Instant::now(), + requested: AtomicBool::new(false), + replied: AtomicBool::new(false), + } + } + + fn note_request(&self) { + self.requested.store(true, Ordering::Relaxed); + } + + fn note_reply(&self) { + self.replied.store(true, Ordering::Relaxed); + } + + fn replied(&self) -> bool { + self.replied.load(Ordering::Relaxed) + } + + /// Маршрут работал по-настоящему: Telegram отвечал, и не один миг. + fn proven(&self) -> bool { + self.replied() && self.opened.elapsed() >= PROVEN_TUNNEL + } + + /// Клиент спросил, Telegram промолчал, и клиент сдался. + /// + /// Раньше такой маршрут оставался предпочтительным: TCP открылся, значит + /// успех, а отказа никто не зафиксировал. Запасные маршруты до очереди так + /// и не доходили. Быстрый уход не считается: клиент сам закрывает лишние + /// соединения из своей гонки. + fn went_silent(&self) -> bool { + self.requested.load(Ordering::Relaxed) + && !self.replied() + && self.opened.elapsed() >= SILENT_TUNNEL + } +} + +/// Соединение оборвалось на полпути. +/// +/// Обрыв долгоживущего туннеля обычен, и пауза на рабочем маршруте оставила бы +/// весь дата-центр без связи. Клиент сам откроет новое соединение. Маршрут +/// наказывается, если он не успел доказать, что работает. +fn tunnel_broke( + stats: &Stats, + connected: &crate::transport::ConnectedRoute, + dc: u16, + reason: &str, + health: &TunnelHealth, +) -> Box { + if !health.proven() { + return tunnel_route_error(stats, connected, dc, reason); + } + let message = format!( + "Обрыв туннеля DC{dc} через {}: {reason}", + connected.route.connect_host + ); + stats.note(message.clone()); + message.into() +} + async fn tcp_relay(a: TcpStream, b: TcpStream) { let (mut ar, mut aw) = tokio::io::split(a); let (mut br, mut bw) = tokio::io::split(b); @@ -1726,6 +1841,225 @@ mod tests { } } + /// Точка фронта Telegram: принять WebSocket и вернуть его вместе с `init`. + #[allow(clippy::result_large_err)] + async fn accept_relay( + listener: TcpListener, + ) -> ( + tokio_tungstenite::WebSocketStream, + [u8; INIT_LEN], + ) { + use futures_util::StreamExt; + let (tcp, _) = listener.accept().await.unwrap(); + let mut ws = + tokio_tungstenite::accept_hdr_async(tcp, |_: &Request, mut response: Response| { + response + .headers_mut() + .insert("Sec-WebSocket-Protocol", "binary".parse().unwrap()); + Ok(response) + }) + .await + .unwrap(); + let init = match ws.next().await.unwrap().unwrap() { + Message::Binary(init) => init, + other => panic!("expected init: {other:?}"), + }; + (ws, init.try_into().unwrap()) + } + + async fn check_packets_per_frame(kind: crate::transport::RouteKind, expected_frames: usize) { + use futures_util::StreamExt; + let first = [&44_u32.to_le_bytes()[..], &[1; 44]].concat(); + let second = [&300_u32.to_le_bytes()[..], &[2; 300]].concat(); + let request = [first.clone(), second.clone()].concat(); + let total = request.len(); + + 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 ws, init) = accept_relay(listener).await; + let mut peer = crate::mtproto::test_relay_peer(&init); + let mut frames = Vec::new(); + let mut received = 0; + while received < total { + if let Message::Binary(mut data) = ws.next().await.unwrap().unwrap() { + peer.decrypt(&mut data); + received += data.len(); + frames.push(data); + } + } + frames + }); + + let stats = Stats::new(); + stats + .transport + .force_local_route_with(relay_port, kind, "/apiws".to_owned()); + let (port, server) = start_proxy(stats.clone(), false).await; + let init = unambiguous_client_init(&stats.secret, 2); + let mut peer = crate::mtproto::test_client_peer(&init, &stats.secret); + let mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + client.write_all(&init).await.unwrap(); + // Одна запись TCP: так клиент отправляет msgs_ack вместе с запросом. + let mut wire = request.clone(); + peer.encrypt(&mut wire); + client.write_all(&wire).await.unwrap(); + + let frames = tokio::time::timeout(Duration::from_secs(5), relay) + .await + .expect("every packet must reach the relay") + .unwrap(); + assert_eq!(frames.len(), expected_frames); + assert_eq!(frames.concat(), request); + if expected_frames == 2 { + assert_eq!(frames, vec![first, second]); + } + stats.stop(); + drop(client); + server.await.unwrap().unwrap(); + } + + #[tokio::test] + async fn websocket_frontend_gets_one_packet_per_frame() { + // Фронт разбирает один пакет на кадр, остальное застревает до следующего. + check_packets_per_frame(crate::transport::RouteKind::TelegramIp, 2).await; + check_packets_per_frame(crate::transport::RouteKind::TelegramPlainWs, 2).await; + } + + #[tokio::test] + async fn cloudflare_worker_needs_no_splitting() { + check_packets_per_frame(crate::transport::RouteKind::CloudflareWorker, 1).await; + } + + /// Туннель, который ответил и через `lifetime` оборвался без закрывающего + /// кадра: так выглядит «Connection reset without closing handshake». + async fn break_tunnel_after(lifetime: Duration) -> Arc { + use futures_util::SinkExt; + 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 ws, _) = accept_relay(listener).await; + ws.send(Message::Binary(vec![1, 2, 3, 4])).await.unwrap(); + tokio::time::sleep(lifetime).await; + drop(ws); + }); + let stats = Stats::new(); + stats.transport.force_local_route(relay_port); + 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, 2)) + .await + .unwrap(); + let mut reply = Vec::new(); + tokio::time::timeout(Duration::from_secs(5), client.read_to_end(&mut reply)) + .await + .unwrap() + .unwrap(); + relay.await.unwrap(); + wait_until("connection task finished", || { + stats.active.load(Ordering::Relaxed) == 0 + }) + .await; + stats.stop(); + server.await.unwrap().unwrap(); + stats + } + + #[tokio::test] + async fn a_tunnel_that_proved_itself_does_not_penalize_its_route_when_it_breaks() { + // Иначе один обрыв рабочего туннеля ставил маршрут на паузу, и DC + // оставался без связи. + let stats = break_tunnel_after(PROVEN_TUNNEL + Duration::from_millis(100)).await; + assert_eq!(stats.transport.route_failures(), 0); + assert_eq!(stats.ws_failures.load(Ordering::Relaxed), 1); + assert!(stats + .drain_events() + .iter() + .any(|event| event.contains("Обрыв туннеля DC2"))); + } + + #[tokio::test] + async fn a_route_that_answers_once_and_dies_is_still_penalized() { + // Иначе он навсегда оставался бы предпочтительным: DPI режет и после + // первых байт, и запасные маршруты до очереди не доходили. + let stats = break_tunnel_after(Duration::ZERO).await; + assert_eq!(stats.transport.route_failures(), 1); + assert!(stats + .drain_events() + .iter() + .any(|event| event.contains("Сбой туннеля DC2"))); + } + + /// Telegram молчит, клиент спрашивает и уходит через `patience`. + async fn leave_silent_route_after(patience: Duration, raw_tcp: bool) -> Arc { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let relay_port = listener.local_addr().unwrap().port(); + let (done_tx, done_rx) = tokio::sync::oneshot::channel::<()>(); + let relay = tokio::spawn(async move { + // Соединение держится открытым и без ответа, пока клиент не уйдёт. + if raw_tcp { + let (stream, _) = listener.accept().await.unwrap(); + let _ = done_rx.await; + drop(stream); + } else { + let (ws, _) = accept_relay(listener).await; + let _ = done_rx.await; + drop(ws); + } + }); + let stats = Stats::new(); + 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 mut client = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + let init = unambiguous_client_init(&stats.secret, if raw_tcp { 203 } else { 2 }); + client.write_all(&init).await.unwrap(); + client.write_all(&[0; 16]).await.unwrap(); + tokio::time::sleep(patience).await; + drop(client); + wait_until("connection task finished", || { + stats.active.load(Ordering::Relaxed) == 0 + }) + .await; + let _ = done_tx.send(()); + relay.await.unwrap(); + stats.stop(); + server.await.unwrap().unwrap(); + stats + } + + #[tokio::test] + async fn a_silent_route_is_penalized_when_the_client_gives_up() { + // Иначе CDN, который принял TCP и молчит, оставался первым в списке, и + // запасной маршрут через порт 80 не получал ни одного шанса. + for raw_tcp in [true, false] { + let wait = SILENT_TUNNEL + Duration::from_millis(100); + let stats = leave_silent_route_after(wait, raw_tcp).await; + assert_eq!(stats.transport.route_failures(), 1, "raw_tcp {raw_tcp}"); + assert!(stats + .drain_events() + .iter() + .any(|event| event.contains(SILENT_ROUTE))); + } + } + + #[tokio::test] + async fn a_client_that_leaves_at_once_does_not_penalize_the_route() { + // Клиент сам закрывает лишние соединения из своей гонки. + for raw_tcp in [true, false] { + let stats = leave_silent_route_after(Duration::ZERO, raw_tcp).await; + assert_eq!(stats.transport.route_failures(), 0, "raw_tcp {raw_tcp}"); + } + } + #[tokio::test] async fn fragmented_encrypted_upload_and_download_preserve_both_streams() { check_fragmented_encrypted_duplex(false).await; diff --git a/src/transport.rs b/src/transport.rs index 9fe48f2..5e042f2 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -9,12 +9,15 @@ use tokio::time::Instant; use tokio_tungstenite::tungstenite::client::IntoClientRequest; use tokio_tungstenite::{MaybeTlsStream, WebSocketStream}; -const CONNECT_TIMEOUT: Duration = Duration::from_secs(4); +// Живой адрес под нагрузкой не всегда укладывался в 4 с, а упавший маршрут уходит +// на паузу. Столько же ждёт Flowseal/tg-ws-proxy. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(8); const FALLBACK_DELAY: Duration = Duration::from_millis(250); 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 HTTP_PORT: u16 = 80; const CDN203_IP: &str = "91.105.192.100"; pub type TelegramWebSocket = WebSocketStream>; @@ -37,6 +40,9 @@ pub enum RouteKind { /// и Cloudflare сам доводит WebSocket до дата-центра. Worker и его лимиты /// не нужны — так работает CF-прокси в Flowseal/tg-ws-proxy. CloudflareProxy, + /// WebSocket без TLS на порту 80 родного адреса DC. Единственный прямой + /// WebSocket к CDN203, у которого нет фронта на 443. + TelegramPlainWs, } impl RouteKind { @@ -48,6 +54,7 @@ impl RouteKind { Self::CloudflareWorker => 4, Self::TelegramTcp => 5, Self::CloudflareProxy => 6, + Self::TelegramPlainWs => 7, } } @@ -59,6 +66,7 @@ impl RouteKind { 4 => Some(Self::CloudflareWorker), 5 => Some(Self::TelegramTcp), 6 => Some(Self::CloudflareProxy), + 7 => Some(Self::TelegramPlainWs), _ => None, } } @@ -72,6 +80,7 @@ impl RouteKind { Self::CloudflareWorker => "Cloudflare Worker", Self::TelegramTcp => "Telegram TCP (MTProto)", Self::CloudflareProxy => "Cloudflare-прокси", + Self::TelegramPlainWs => "Telegram WebSocket (порт 80)", } } } @@ -113,6 +122,23 @@ impl Route { } } + /// WebSocket to a native Telegram address on port 80. + /// + /// There is no TLS here, but no less protection than on `telegram_tcp`: the + /// payload is the same obfuscated MTProto stream, and MTProto authenticates + /// the server itself. Here the address picks the DC and the Host header does + /// not matter, so the address is all that goes in it. + fn plain_ws(ip: &str) -> Self { + Self { + connect_host: ip.to_owned(), + websocket_host: ip.to_owned(), + path: "/apiws".to_owned(), + kind: RouteKind::TelegramPlainWs, + port: HTTP_PORT, + secure: false, + } + } + /// A production route: TLS on 443. fn https(connect_host: String, websocket_host: String, path: String, kind: RouteKind) -> Self { Self { @@ -327,10 +353,17 @@ impl TransportEngine { } Err(error) => { self.record_failure(&route); - let endpoint = if route.kind == RouteKind::TelegramTcp { - format!("{}:{} (MTProto TCP)", route.connect_host, route.port) - } else { - format!("{} (TLS {})", route.connect_host, route.websocket_host) + let endpoint = match route.kind { + RouteKind::TelegramTcp => { + format!("{}:{} (MTProto TCP)", route.connect_host, route.port) + } + RouteKind::TelegramPlainWs => { + format!("{}:{} (WebSocket)", route.connect_host, route.port) + } + _ => format!( + "{} (TLS {})", + route.connect_host, route.websocket_host + ), }; errors.push(format!("{endpoint} — {error}")); } @@ -379,8 +412,11 @@ impl TransportEngine { RouteKind::CloudflareWorker | RouteKind::CloudflareProxy => 1, RouteKind::TelegramIp | RouteKind::AlternateTelegramIp => 2, RouteKind::SystemDns => 3, - // Last resort, raced separately; see race_connections(). - RouteKind::TelegramTcp => 4, + // Last resort, raced separately; see race_connections(). The + // WebSocket goes first: its upgrade is an answer from Telegram, + // while TCP only shows that the address accepts connections. + RouteKind::TelegramPlainWs => 4, + RouteKind::TelegramTcp => 5, }; (preferred_rank, kind_rank) }); @@ -479,10 +515,11 @@ pub(crate) fn cf_proxy_host(dc: u16, domain: &str) -> String { format!("kws{}.{}", dc, domain) } -/// Native TCP to an ordinary data centre — every pinned address except the -/// CDN, whose native transport is its primary route. +/// Routes tried only once everything else failed: port 80, and native TCP to +/// an ordinary data centre. The CDN's native TCP is its primary route. fn is_last_resort(route: &Route) -> bool { - route.kind == RouteKind::TelegramTcp && route.connect_host != CDN203_IP + route.kind == RouteKind::TelegramPlainWs + || (route.kind == RouteKind::TelegramTcp && route.connect_host != CDN203_IP) } /// Every address a Worker may be asked to reach, so a deployment can refuse @@ -497,14 +534,118 @@ pub fn worker_allowed_destinations() -> Vec<&'static str> { all } -fn canonical_dc(dc: u16) -> u16 { - if dc == 203 { - 2 - } else { - dc +/// Настоящий WebSocket-фронт Telegram: адрес и имена, которые он обслуживает. +struct WebFront { + ip: &'static str, + /// Основной кластер. Принимает и обычные соединения, и медиа. + host: &'static str, + /// Медиа-кластер (`-1`), только для медиа; см. `routes_for_dc`. + media_host: &'static str, +} + +const fn front(ip: &'static str, host: &'static str, media_host: &'static str) -> WebFront { + WebFront { + ip, + host, + media_host, } } +/// Где Telegram принимает WebSocket. Первый фронт каждого DC — тот, на который +/// указывает `kwsN.web.telegram.org`. +/// +/// Это не адреса из `telegram_ips`: те говорят только на родном MTProto и на +/// TLS не отвечают из любой сети. Фронт выбирает DC по имени хоста, а не по +/// заголовку клиента, так что имя и адрес здесь нельзя менять местами: `.220` +/// на `kws1` отвечает редиректом. Каждая пара проверена запросом `req_DH_params`: +/// сервер принимает только свой номер DC, поэтому за фронтом стоит именно он. +fn web_fronts(dc: u16) -> &'static [WebFront] { + const DC1: &[WebFront] = &[ + front( + "149.154.174.100", + "kws1.web.telegram.org", + "kws1-1.web.telegram.org", + ), + front( + "149.154.174.200", + "zws1.web.telegram.org", + "zws1-1.web.telegram.org", + ), + front( + "149.154.175.209", + "pluto.web.telegram.org", + "pluto-1.web.telegram.org", + ), + ]; + const DC2: &[WebFront] = &[ + front( + "149.154.167.220", + "kws2.web.telegram.org", + "kws2-1.web.telegram.org", + ), + front( + "149.154.167.99", + "kws2.web.telegram.org", + "kws2-1.web.telegram.org", + ), + ]; + const DC3: &[WebFront] = &[ + front( + "149.154.174.100", + "kws3.web.telegram.org", + "kws3-1.web.telegram.org", + ), + front( + "149.154.174.200", + "zws3.web.telegram.org", + "zws3-1.web.telegram.org", + ), + front( + "149.154.175.209", + "aurora.web.telegram.org", + "aurora-1.web.telegram.org", + ), + ]; + const DC4: &[WebFront] = &[ + front( + "149.154.167.220", + "kws4.web.telegram.org", + "kws4-1.web.telegram.org", + ), + front( + "149.154.167.99", + "kws4.web.telegram.org", + "kws4-1.web.telegram.org", + ), + ]; + const DC5: &[WebFront] = &[ + front( + "149.154.170.100", + "kws5.web.telegram.org", + "kws5-1.web.telegram.org", + ), + front( + "149.154.170.200", + "zws5.web.telegram.org", + "zws5-1.web.telegram.org", + ), + front( + "149.154.170.96", + "flora.web.telegram.org", + "flora-1.web.telegram.org", + ), + ]; + match dc { + 1 => DC1, + 2 => DC2, + 3 => DC3, + 4 => DC4, + 5 => DC5, + _ => &[], + } +} + +/// Родные адреса MTProto: сюда ходит прямой TCP и Worker. TLS здесь нет. fn telegram_ips(dc: u16) -> &'static [&'static str] { match dc { 1 => &["149.154.175.50"], @@ -518,32 +659,33 @@ fn telegram_ips(dc: u16) -> &'static [&'static str] { } pub fn routes_for_dc(dc: u16, media: bool) -> Vec { - if telegram_ips(dc).is_empty() { + if dc == 203 { + // CDN203 has no WebSocket frontend of its own: its address speaks only + // native obfuscated2 on 443, and kws2 on any frontend is plain DC2, where + // a CDN session has neither its keys nor its data. Port 80 of the same + // address does lead to the CDN. Worker and CF proxy routes are added on + // top in routes_for_key. + return vec![Route::cdn_tcp(), Route::plain_ws(CDN203_IP)]; + } + let fronts = web_fronts(dc); + if fronts.is_empty() { return Vec::new(); } - let websocket_dc = canonical_dc(dc); - let primary = format!("kws{}.web.telegram.org", websocket_dc); - let secondary = format!("kws{}-1.web.telegram.org", websocket_dc); - let websocket_hosts = if media { - [secondary, primary] - } else { - [primary, secondary] - }; - 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() { + // Only media may use the `-1` hosts. That cluster rejects an ordinary + // connection's key exchange with -444 (measured on kws2-1 and kws4-1), and + // Telegram Desktop answers -444 by turning the proxy off as misconfigured. + let clusters: &[fn(&WebFront) -> &'static str] = if media { + &[|front| front.media_host, |front| front.host] + } else { + &[|front| front.host] + }; + let mut routes = Vec::new(); + for host in clusters { + for (index, front) in fronts.iter().enumerate() { routes.push(Route::https( - (*ip).to_owned(), - websocket_host.clone(), + front.ip.to_owned(), + host(front).to_owned(), "/apiws".to_owned(), if index == 0 { RouteKind::TelegramIp @@ -552,28 +694,18 @@ pub fn routes_for_dc(dc: u16, media: bool) -> Vec { }, )); } - // DC203 uses the DC2 hostname only for TLS/HTTP virtual hosting. - // Resolving it would connect to DC2 itself and send CDN sessions to - // the wrong data centre. Its fallback must retain the CDN address. - if dc != 203 { - routes.push(Route::https( - websocket_host.clone(), - websocket_host.clone(), - "/apiws".to_owned(), - RouteKind::SystemDns, - )); - } - } - // Where the WebSocket frontends are filtered but the address itself is - // reachable, plain MTProto to the DC's primary address still works — - // measured from a Russian ISP where TLS to .51/.91/.50/.100/.5 was cut - // while native TCP to each of them answered req_pq (main and media). - // 149.154.167.220 is a WebSocket frontend only and does not speak native - // MTProto, so it gets no TCP route. Raced only after everything above - // failed (race_connections). - if dc != 203 { - routes.push(Route::telegram_tcp(ips[0])); + let name = host(&fronts[0]).to_owned(); + routes.push(Route::https( + name.clone(), + name, + "/apiws".to_owned(), + RouteKind::SystemDns, + )); } + // Where the frontends are out of reach, the DC's own address may still + // work. Raced only after everything above failed (race_connections). + routes.push(Route::plain_ws(telegram_ips(dc)[0])); + routes.push(Route::telegram_tcp(telegram_ips(dc)[0])); routes } @@ -615,6 +747,11 @@ async fn connect_route_with_config( "native MTProto TCP разрешён только до закреплённых адресов Telegram".to_owned(), ); } + if route.kind == RouteKind::TelegramPlainWs && !allowed_plain_ws_route(route) { + return Err( + "WebSocket без TLS разрешён только до закреплённых адресов Telegram".to_owned(), + ); + } let tcp = tokio::time::timeout( CONNECT_TIMEOUT, TcpStream::connect((route.connect_host.as_str(), route.port)), @@ -643,8 +780,8 @@ async fn connect_route_with_config( ); if !route.secure { - // Raw TCP has already returned above. Plain WebSocket is only used by - // local fixtures; production WebSocket routes are always HTTPS. + // Raw TCP has already returned above. Besides local fixtures, plain + // WebSocket is only allowed to pinned Telegram addresses, checked above. return tokio::time::timeout( CONNECT_TIMEOUT, tokio_tungstenite::client_async(request, MaybeTlsStream::Plain(tcp)), @@ -670,6 +807,17 @@ async fn connect_route_with_config( .map_err(|error| format!("TLS/WebSocket handshake: {}", error)) } +fn allowed_plain_ws_route(route: &Route) -> bool { + #[cfg(test)] + if route.connect_host == "127.0.0.1" { + return !route.secure; + } + worker_allowed_destinations().contains(&route.connect_host.as_str()) + && route.port == HTTP_PORT + && route.websocket_host == route.connect_host + && !route.secure +} + fn allowed_tcp_route(route: &Route) -> bool { #[cfg(test)] if route.connect_host == "127.0.0.1" && !route.secure { @@ -724,7 +872,73 @@ mod tests { fn media_prefers_dash_one_websocket_host() { let routes = routes_for_dc(2, true); assert_eq!(routes[0].websocket_host, "kws2-1.web.telegram.org"); - assert_eq!(routes[0].connect_host, "149.154.167.51"); + assert_eq!(routes[0].connect_host, "149.154.167.220"); + } + + #[test] + fn only_media_may_reach_the_dash_one_cluster() { + // That cluster answers an ordinary key exchange with -444, and Telegram + // Desktop reacts by disabling the proxy as misconfigured. + for dc in [1, 2, 3, 4, 5] { + let dash_one = |route: &Route| route.websocket_host.contains("-1."); + assert!( + !routes_for_dc(dc, false).iter().any(dash_one), + "DC{dc} ordinary connection on a media-only cluster" + ); + let media = routes_for_dc(dc, true); + assert!(dash_one(&media[0]), "DC{dc}: {:?}", media[0]); + // The main cluster accepts media too, so it stays as a fallback. + assert!(media + .iter() + .any(|route| route.websocket_host == format!("kws{dc}.web.telegram.org"))); + } + } + + #[test] + fn websocket_routes_only_dial_real_frontends() { + // The native addresses speak MTProto and answer TLS with garbage from + // any network, so a WebSocket route there can never connect. + for dc in [1, 2, 3, 4, 5, 203] { + for media in [false, true] { + for route in routes_for_dc(dc, media) { + if matches!( + route.kind, + RouteKind::TelegramTcp | RouteKind::TelegramPlainWs | RouteKind::SystemDns + ) { + continue; + } + assert!( + web_fronts(dc) + .iter() + .any(|front| front.ip == route.connect_host), + "DC{dc}: {route:?}" + ); + assert!(route.secure, "DC{dc}: {route:?}"); + assert_ne!(route.connect_host, telegram_ips(dc)[0], "DC{dc}"); + } + } + } + } + + #[test] + fn frontends_are_only_dialled_under_a_name_they_serve() { + // A frontend picks the DC from the Host header, not from the client's + // init: kws2 is DC2 whatever the header says, and .220 redirects kws1. + for (dc, name) in [ + (1, "kws1."), + (2, "kws2."), + (3, "kws3."), + (4, "kws4."), + (5, "kws5."), + ] { + let routes = routes_for_dc(dc, false); + assert!(routes[0].websocket_host.starts_with(name), "{routes:?}"); + let front = &web_fronts(dc)[0]; + assert_eq!(routes[0].connect_host, front.ip); + } + assert!(routes_for_dc(1, false) + .iter() + .all(|route| route.connect_host != "149.154.167.220")); } #[test] @@ -734,7 +948,8 @@ mod tests { 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_eq!(routes[1], Route::plain_ws("91.105.192.100")); + assert_eq!(routes.len(), 2, "CDN203 has no WebSocket frontend on 443"); assert!(routes .iter() .all(|route| route.connect_host == "91.105.192.100")); @@ -816,15 +1031,27 @@ mod tests { let routes = engine.routes_for_key(DcKey { dc, media }); assert!(!routes.is_empty(), "DC{dc} must have at least one route"); for route in routes { - assert_eq!(route.port, 443, "{route:?}"); - if route.kind == RouteKind::TelegramTcp { - assert!( - telegram_ips(dc).contains(&route.connect_host.as_str()), - "{route:?}" - ); - assert!(allowed_tcp_route(&route)); - } else { - assert!(route.secure, "{route:?}"); + match route.kind { + RouteKind::TelegramTcp => { + assert_eq!(route.port, 443, "{route:?}"); + assert!( + telegram_ips(dc).contains(&route.connect_host.as_str()), + "{route:?}" + ); + assert!(allowed_tcp_route(&route)); + } + // The only route without TLS: pinned address, and no + // Telegram name on the wire. + RouteKind::TelegramPlainWs => { + assert_eq!(route.port, 80, "{route:?}"); + assert_eq!(route.connect_host, telegram_ips(dc)[0]); + assert_eq!(route.websocket_host, route.connect_host); + assert!(allowed_plain_ws_route(&route)); + } + _ => { + assert_eq!(route.port, 443, "{route:?}"); + assert!(route.secure, "{route:?}"); + } } } } @@ -837,17 +1064,13 @@ mod tests { engine.set_cf_proxy_domains(&["example.com".to_owned()]); for dc in [1, 2, 3, 4, 5] { let ordered = engine.ordered_candidates(DcKey { dc, media: true }); - let first_tcp = ordered + let first_fallback = ordered .iter() - .position(|route| route.kind == RouteKind::TelegramTcp) - .expect("every DC has a native TCP fallback"); - assert!( - ordered[first_tcp..].iter().all(is_last_resort), - "DC{dc}: {ordered:?}" - ); - assert!(ordered[..first_tcp] - .iter() - .all(|route| !is_last_resort(route))); + .position(is_last_resort) + .expect("every DC has a last-resort route"); + assert_eq!(ordered[first_fallback].kind, RouteKind::TelegramPlainWs); + assert_eq!(ordered[first_fallback + 1].kind, RouteKind::TelegramTcp); + assert_eq!(ordered.len(), first_fallback + 2, "DC{dc}: {ordered:?}"); } let cdn = engine.ordered_candidates(DcKey { dc: 203, @@ -855,6 +1078,25 @@ mod tests { }); assert_eq!(cdn[0], Route::cdn_tcp()); assert!(!is_last_resort(&cdn[0])); + assert_eq!(cdn.last(), Some(&Route::plain_ws(CDN203_IP))); + } + + #[tokio::test] + async fn plain_websocket_refuses_anything_but_a_pinned_address() { + for route in [ + Route::plain_ws("203.0.113.5"), + Route { + websocket_host: "kws2.web.telegram.org".to_owned(), + ..Route::plain_ws("149.154.167.51") + }, + Route { + port: 8080, + ..Route::plain_ws("149.154.167.51") + }, + ] { + let error = connect_route(&route).await.unwrap_err(); + assert!(error.contains("без TLS"), "{route:?}: {error}"); + } } #[tokio::test] @@ -1163,6 +1405,8 @@ mod tests { RouteKind::SystemDns, RouteKind::CloudflareWorker, RouteKind::TelegramTcp, + RouteKind::CloudflareProxy, + RouteKind::TelegramPlainWs, ] { assert_eq!(RouteKind::from_ui_code(kind.ui_code()), Some(kind)); assert_eq!(route_label(kind.ui_code()), kind.label()); @@ -1471,6 +1715,20 @@ mod tests { } } + #[tokio::test] + #[ignore = "requires live Telegram network access"] + async fn port_80_answers_on_every_native_address() { + for dc in [1, 2, 3, 4, 5, 203] { + let route = Route::plain_ws(telegram_ips(dc)[0]); + match connect_route(&route).await { + Ok(TelegramConnection::WebSocket(mut websocket)) => { + websocket.close(None).await.unwrap() + } + other => panic!("DC{dc}: {other:?}"), + } + } + } + #[tokio::test] #[ignore = "requires live Telegram network access"] async fn connects_to_all_production_data_centers() { diff --git a/tauri.conf.json b/tauri.conf.json index 5fcda31..3cefebc 100644 --- a/tauri.conf.json +++ b/tauri.conf.json @@ -1,7 +1,7 @@ { "$schema": "https://schema.tauri.app/config/2", "productName": "TGLock", - "version": "2.0.0-beta.18", + "version": "2.0.0-beta.19", "identifier": "com.bysonic.tglock", "mainBinaryName": "tglock", "build": { diff --git a/tglock.example.toml b/tglock.example.toml index c5ec0ba..01a7113 100644 --- a/tglock.example.toml +++ b/tglock.example.toml @@ -44,3 +44,8 @@ # Печатать только ошибки. # quiet = false + +# Журнал в файл вместо stdout: на роутере stdout сервиса уходит в системный журнал. +# Заполненный файл переименовывается в <имя>.1, больше двух файлов не бывает. +# log_file = "/tmp/tglock.log" +# log_max_mb = 5 diff --git a/ui/main.ts b/ui/main.ts index cdf8b1e..3c36b05 100644 --- a/ui/main.ts +++ b/ui/main.ts @@ -513,12 +513,29 @@ async function copyTelegramLink(): Promise { } } +// Отчёт публикуют в issue: домен Worker'а и свой домен CF-прокси туда не идут, +// Worker можно задать и адресом, поэтому оставляются только адреса Telegram. +// Они и имена Telegram нужны, чтобы понять, какой маршрут не ответил. +function publicLogLine(message: string): string { + return message.replace(/[\w-]+(?:\.[\w-]+)+/g, (host) => + /^(149\.154|91\.105|91\.108)\.[\d.]+$/.test(host) || host.endsWith(".telegram.org") ? host : "<домен>"); +} + +function failureLog(): string[] { + const failure = /^(Не поднялся туннель|Сбой туннеля|Обрыв туннеля)/; + const paused = /на паузе после ошибок/; + return status.logs + .filter(({ message }) => failure.test(message) && !paused.test(message)) + .slice(-20) + .map(({ timestamp, message }) => `${timestamp} ${publicLogLine(message)}`); +} + async function copyDiagnostics(): Promise { try { - // Public reports exclude proxy links, worker domains and free-form logs. + // Public reports exclude proxy links and secrets. const report = { application: "TGLock", - version: "2.0.0-beta.18", + version: "2.0.0-beta.19", running: status.running, activeConnections: status.activeConnections, tunnels: status.tunnels, @@ -532,9 +549,10 @@ async function copyDiagnostics(): Promise { uptimeSeconds: status.uptimeSeconds, port: status.port, lanMode: settings.lanMode, + failureLog: failureLog(), }; await navigator.clipboard.writeText(JSON.stringify(report, null, 2)); - showToast("Диагностика скопирована без адресов и секретов"); + showToast("Диагностика скопирована без секретов и своих доменов"); } catch { showToast("Буфер обмена недоступен", true); } }