docs [AI]
This commit is contained in:
+47
-433
@@ -1,448 +1,62 @@
|
||||
# Network Module Documentation
|
||||
# Блок `net` — сетевое ядро и оркестровка туннеля
|
||||
|
||||
## Overview
|
||||
Самый верхний блок крейта: здесь синхронный разбор протокола ([`nrxp`](../nrxp)) и
|
||||
асинхронная сеть `tokio` соединяются в работающий мультиплексированный туннель.
|
||||
Всё, что ниже — крипта, кадры, маскировка — лишь «кирпичи», а этот блок строит из
|
||||
них дом: слушает соединения, держит ноги туннеля, балансирует потоки и проксирует
|
||||
трафик к целям.
|
||||
|
||||
Данный модуль реализует сетевую часть VPN ядра:
|
||||
> Детали — в rustdoc, модуль `net`.
|
||||
|
||||
- входные TCP соединения (SOCKS / туннель)
|
||||
- мультиплексирование потоков поверх одного соединения
|
||||
- транспорт кадров (Frame)
|
||||
- проксирование TCP/UDP
|
||||
- управление туннелем (TunnelEngine)
|
||||
## Файлы
|
||||
|
||||
Архитектура:
|
||||
| Файл/подмодуль | Что внутри |
|
||||
|------------------|-----------------------------------------------------------------|
|
||||
| `config.rs` | `NetworkConfig` — MTU-зависимые размеры буферов и каналов. |
|
||||
| `constants.rs` | Тайм-ауты, лимиты, порты, анти-bufferbloat тюнинг. |
|
||||
| `diagnostics.rs` | События, метрики, снапшоты состояния туннеля. |
|
||||
| `connection/` | Ядро: соединения, мультиплексор, движок, мосты, хендлеры. |
|
||||
|
||||
- network — точка входа (listener + режим)
|
||||
- connection — логика соединений
|
||||
- muxer — мультиплексирование потоков
|
||||
- handler — обработка входящих фреймов
|
||||
- engine — основной цикл туннеля
|
||||
- bridge — проксирование сокетов
|
||||
### Подмодуль `connection/`
|
||||
|
||||
---
|
||||
| Файл | Роль |
|
||||
|-----------------|-----------------------------------------------------------------|
|
||||
| `muxer.rs` | Мультиплексор: реестр потоков/ног, балансировка, failover. |
|
||||
| `engine.rs` | `TunnelEngine`: reader/writer-задачи одной ноги, heartbeat, реконнект. |
|
||||
| `connection.rs` | Точки входа: `ClientHandler`, `ServerHandler`, `SessionManager`. |
|
||||
| `handler.rs` | Диспетчеризация входящих кадров по типу/`stream_id`. |
|
||||
| `bridge.rs` | Проксирование TCP/UDP между потоком туннеля и целью. |
|
||||
|
||||
## High-Level Architecture
|
||||
## Ключевая идея
|
||||
|
||||
Система работает по схеме:
|
||||
**Одно TCP-соединение (нога) = много логических потоков (`stream_id`).** Ради
|
||||
устойчивости ног несколько (`MAX_TUNNEL_LEGS`); поток «прилипает» к ноге, а при её
|
||||
падении бесшовно переезжает на соседнюю.
|
||||
|
||||
Client (SOCKS)
|
||||
→ Muxer
|
||||
→ TunnelEngine (шифрованный канал)
|
||||
→ Server
|
||||
→ StreamHandler
|
||||
→ TCP/UDP target
|
||||
## Сквозной поток данных
|
||||
|
||||
И обратно.
|
||||
```text
|
||||
Клиент: браузер → SOCKS/rawcast → ClientHandler → Muxer → TunnelEngine → шифр TCP
|
||||
Сервер: шифр TCP → TunnelEngine → StreamHandler → bridge → TCP/UDP цель
|
||||
... и обратно тем же путём
|
||||
```
|
||||
|
||||
Ключевая идея:
|
||||
ОДНО TCP соединение = МНОГО логических потоков (stream_id)
|
||||
## Что делает этот блок устойчивым (мотивация констант)
|
||||
|
||||
---
|
||||
- **Anti-domino failover** (`muxer::send_to_network`) — падение ноги не закрывает
|
||||
поток: дохлая нога эвиктится, кадр переотправляется на соседнюю.
|
||||
- **Graceful pause** (`bridge::run_tcp_bridge`) — если легли все ноги, мост
|
||||
держит чанк и ретраит (TCP backpressure), а не рвёт стрим.
|
||||
- **Anti-bufferbloat** — маленькая ёмкость каналов, адаптивные по RTT таймауты и
|
||||
размер батча, неблокирующая раздача `dispatch_to_local`.
|
||||
- **Stealth-fallback** (`ServerHandler`) — «не наш» клиент прозрачно проксируется
|
||||
на `ubuntu.com:443`, маскируясь под обычный TLS.
|
||||
|
||||
## Network
|
||||
Почти каждая константа в `constants.rs` снабжена `///` с объяснением **почему**
|
||||
именно это число — читайте обоснование рядом, прежде чем менять.
|
||||
|
||||
### Назначение
|
||||
## Связи
|
||||
|
||||
Точка входа:
|
||||
|
||||
- запускает TCP listener
|
||||
- определяет режим (Client / Server)
|
||||
- управляет lifecycle
|
||||
|
||||
### Режимы
|
||||
|
||||
#### Client
|
||||
|
||||
- Подключается к удалённому proxy
|
||||
- Поднимает SOCKS сервер
|
||||
- Каждый входящий клиент → новый stream
|
||||
|
||||
#### Server
|
||||
|
||||
- Принимает туннельные соединения
|
||||
- Обрабатывает handshake
|
||||
- Создаёт TunnelEngine
|
||||
|
||||
---
|
||||
|
||||
## NetworkConfig
|
||||
|
||||
Глобальная конфигурация (OnceLock)
|
||||
|
||||
### Основные параметры
|
||||
|
||||
- mtu — системный MTU
|
||||
- max_wire_frame_size — MTU минус overhead
|
||||
- safe_payload_size — безопасный payload
|
||||
- tcp_buffer_size — буфер TCP
|
||||
- udp_buffer_size — буфер UDP
|
||||
- channel_capacity — размер каналов
|
||||
|
||||
### Расчёт
|
||||
|
||||
safe_payload = MTU
|
||||
|
||||
- IPv4/UDP overhead (~28)
|
||||
- frame header
|
||||
- padding
|
||||
|
||||
Цель:
|
||||
|
||||
- не фрагментировать пакеты
|
||||
- держаться в ~1400 байт
|
||||
|
||||
---
|
||||
|
||||
## Connection
|
||||
|
||||
Обёртка над TCP соединением
|
||||
|
||||
### Содержит:
|
||||
|
||||
- inbound (read half)
|
||||
- outbound (write half)
|
||||
- read buffer
|
||||
- codec (шифрование + framing)
|
||||
|
||||
### Основные функции
|
||||
|
||||
#### read_socks_request
|
||||
|
||||
- читает из буфера
|
||||
- использует Parser trait
|
||||
- ждёт пока данных достаточно
|
||||
|
||||
#### send_socks_reply
|
||||
|
||||
- сериализует ответ
|
||||
- отправляет клиенту
|
||||
|
||||
---
|
||||
|
||||
## ClientHandler
|
||||
|
||||
### Назначение
|
||||
|
||||
- реализует SOCKS5 сервер
|
||||
- преобразует запросы в multiplexed streams
|
||||
|
||||
### Flow
|
||||
|
||||
1. SOCKS handshake
|
||||
2. SOCKS request (CONNECT / UDP)
|
||||
3. Создание stream_id
|
||||
4. Регистрация канала
|
||||
5. Отправка FrameType::Connect
|
||||
6. Ожидание ответа от сервера
|
||||
7. Запуск bridge
|
||||
|
||||
---
|
||||
|
||||
## ServerHandler
|
||||
|
||||
### Назначение
|
||||
|
||||
- принимает туннель
|
||||
- делает TLS-like handshake
|
||||
- запускает TunnelEngine
|
||||
|
||||
### Особенность: Stealth Fallback
|
||||
|
||||
Если:
|
||||
|
||||
- handshake не прошёл
|
||||
- таймаут
|
||||
|
||||
→ соединение проксируется как обычный TLS (на ubuntu.com:443)
|
||||
|
||||
Это:
|
||||
|
||||
- маскирует трафик
|
||||
- снижает вероятность блокировки
|
||||
|
||||
---
|
||||
|
||||
## Muxer
|
||||
|
||||
### Назначение
|
||||
|
||||
Мультиплексирование потоков
|
||||
|
||||
### Компоненты
|
||||
|
||||
- control_tx — управляющие сообщения
|
||||
- data_tx — данные
|
||||
- streams — map stream_id → channel
|
||||
- id_gen — генератор ID
|
||||
|
||||
---
|
||||
|
||||
## Stream ID
|
||||
|
||||
Генерация:
|
||||
|
||||
- клиент: 1, 3, 5, ...
|
||||
- сервер: 2, 4, 6, ...
|
||||
|
||||
Это гарантирует:
|
||||
|
||||
- отсутствие коллизий
|
||||
- определение стороны
|
||||
|
||||
---
|
||||
|
||||
## MuxMessage
|
||||
|
||||
Структура:
|
||||
|
||||
- stream_id
|
||||
- frame_type
|
||||
- data
|
||||
|
||||
---
|
||||
|
||||
## Отправка данных
|
||||
|
||||
### send_data_safe
|
||||
|
||||
Если payload большой:
|
||||
|
||||
- режется на куски (~1300 байт)
|
||||
- отправляется последовательно
|
||||
|
||||
Причина:
|
||||
|
||||
- не превышать MTU
|
||||
- избежать фрагментации
|
||||
|
||||
---
|
||||
|
||||
## StreamHandler
|
||||
|
||||
### Назначение
|
||||
|
||||
Обработка входящих Frame
|
||||
|
||||
### Типы Frame
|
||||
|
||||
- Connect
|
||||
- UdpConnect
|
||||
- Data
|
||||
- UdpData
|
||||
- Close
|
||||
|
||||
---
|
||||
|
||||
## TCP Flow (Server)
|
||||
|
||||
1. Получен Connect
|
||||
2. Парсится target
|
||||
3. Открывается TCP соединение
|
||||
4. Отправляется SOCKS reply
|
||||
5. Запускается run_tcp_bridge
|
||||
|
||||
---
|
||||
|
||||
## UDP Flow
|
||||
|
||||
1. bind UDP socket
|
||||
2. connect(target)
|
||||
3. запуск run_udp_bridge
|
||||
|
||||
---
|
||||
|
||||
## TunnelEngine
|
||||
|
||||
### Центральный компонент
|
||||
|
||||
Отвечает за:
|
||||
|
||||
- чтение из сети
|
||||
- декодирование
|
||||
- dispatch
|
||||
- шифрование
|
||||
- запись
|
||||
|
||||
---
|
||||
|
||||
## Reader Task
|
||||
|
||||
Цикл:
|
||||
|
||||
1. read_buf
|
||||
2. codec.inbound()
|
||||
3. извлечение frame
|
||||
4. handler.handle(frame)
|
||||
|
||||
### Ошибки
|
||||
|
||||
- Wait → ждём данных
|
||||
- Drop → критическая ошибка (tampering)
|
||||
|
||||
---
|
||||
|
||||
## Writer Task
|
||||
|
||||
Обрабатывает:
|
||||
|
||||
- control сообщения (приоритет)
|
||||
- heartbeat
|
||||
- data сообщения
|
||||
|
||||
### Heartbeat
|
||||
|
||||
Каждые 15 секунд:
|
||||
FrameType::Heartbeat
|
||||
|
||||
---
|
||||
|
||||
## handle_outbound
|
||||
|
||||
Flow:
|
||||
|
||||
1. chunking (4KB)
|
||||
2. encrypt_data()
|
||||
3. write_all()
|
||||
|
||||
---
|
||||
|
||||
## Bridge Layer
|
||||
|
||||
### run_tcp_bridge
|
||||
|
||||
Два направления:
|
||||
|
||||
1. socket → muxer
|
||||
2. muxer → socket
|
||||
|
||||
Особенности:
|
||||
|
||||
- неблокирующий select
|
||||
- graceful close
|
||||
- отправка Close frame
|
||||
|
||||
---
|
||||
|
||||
### run_udp_bridge
|
||||
|
||||
Аналог TCP, но:
|
||||
|
||||
- recv/send
|
||||
- без stream (datagram)
|
||||
|
||||
---
|
||||
|
||||
## Поток данных (End-to-End)
|
||||
|
||||
Client:
|
||||
|
||||
browser → SOCKS → ClientHandler
|
||||
→ Muxer → TunnelEngine
|
||||
→ encrypted TCP
|
||||
|
||||
Server:
|
||||
|
||||
→ TunnelEngine → StreamHandler
|
||||
→ TCP connect → target
|
||||
|
||||
Ответ обратно тем же путём.
|
||||
|
||||
---
|
||||
|
||||
## Error Handling
|
||||
|
||||
Критические ошибки:
|
||||
|
||||
- codec Drop → разрыв туннеля
|
||||
- write error → закрытие stream
|
||||
- read EOF → завершение
|
||||
|
||||
Некритические:
|
||||
|
||||
- закрыт канал → удаление stream
|
||||
- неизвестный stream → игнор
|
||||
|
||||
---
|
||||
|
||||
## Concurrency Model
|
||||
|
||||
Используется:
|
||||
|
||||
- tokio tasks
|
||||
- mpsc каналы
|
||||
- Arc + DashMap
|
||||
|
||||
Параллелизм:
|
||||
|
||||
- каждый stream независим
|
||||
- reader / writer разделены
|
||||
|
||||
---
|
||||
|
||||
## Design Decisions
|
||||
|
||||
### 1. Один туннель
|
||||
|
||||
Плюсы:
|
||||
|
||||
- меньше соединений
|
||||
- проще маскировка
|
||||
|
||||
Минусы:
|
||||
|
||||
- single point of failure
|
||||
|
||||
---
|
||||
|
||||
### 2. Multiplexing
|
||||
|
||||
Плюсы:
|
||||
|
||||
- высокая эффективность
|
||||
- меньше latency
|
||||
|
||||
---
|
||||
|
||||
### 3. Chunking
|
||||
|
||||
Причины:
|
||||
|
||||
- MTU ограничения
|
||||
- стабильность
|
||||
|
||||
---
|
||||
|
||||
### 4. Stealth fallback
|
||||
|
||||
Критично для:
|
||||
|
||||
- обход DPI
|
||||
- маскировка под TLS
|
||||
|
||||
---
|
||||
|
||||
## Limitations
|
||||
|
||||
- Нет QoS между потоками
|
||||
- Нет backpressure контроля
|
||||
- Нет retransmission (TCP полагается на underlying)
|
||||
- Нет stream prioritization
|
||||
- UDP без гарантии доставки
|
||||
|
||||
---
|
||||
|
||||
## Summary
|
||||
|
||||
Модуль реализует:
|
||||
|
||||
- TCP listener (client/server)
|
||||
- multiplexed туннель
|
||||
- безопасную передачу данных
|
||||
- проксирование TCP/UDP
|
||||
- fallback механизм маскировки
|
||||
|
||||
Это полноценное сетевое ядро VPN:
|
||||
|
||||
- scalable
|
||||
- асинхронное
|
||||
- расширяемое
|
||||
Использует [`nrxp`](../nrxp) (кадры/кодек), [`tlseng`](../tlseng) (хендшейк),
|
||||
[`crypto`](../crypto) (косвенно через кодек) и [`rawcast`](../rawcast) (вход с
|
||||
локальной стороны клиента).
|
||||
|
||||
+34
-2
@@ -1,28 +1,56 @@
|
||||
//! Глобальная сетевая конфигурация, выводимая из MTU.
|
||||
//!
|
||||
//! Все размеры буферов и ёмкости каналов считаются один раз из системного MTU
|
||||
//! ([`NetworkConfig::new`]) и кладутся в глобальный [`OnceLock`]. Логика проста:
|
||||
//! буферы TCP-сокетов smoltcp масштабируются так, чтобы окно вмещало нужное число
|
||||
//! сегментов подряд, а ёмкость mpsc-каналов держится **намеренно маленькой** —
|
||||
//! это главный рычаг против bufferbloat (см. комментарий к `CHANNEL_PACKETS`).
|
||||
//!
|
||||
//! Деление буферов на `heavy`/`light` — это «толстые» потоки (bulk download) против
|
||||
//! «тонких» (DNS, интерактив): первым нужен большой буфер для throughput, вторым —
|
||||
//! маленький для низкой задержки.
|
||||
|
||||
use netrunner_logger::warn;
|
||||
use std::sync::OnceLock;
|
||||
|
||||
/// Глобально инициализируемая сетевая конфигурация (одна на процесс).
|
||||
pub static GLOBAL_NET_CONFIG: OnceLock<NetworkConfig> = OnceLock::new();
|
||||
|
||||
/// Набор размеров буферов и каналов, выведенных из MTU.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct NetworkConfig {
|
||||
/// Эффективный MTU (не ниже 576).
|
||||
pub mtu: usize,
|
||||
/// Размер буфера чтения соединения туннеля.
|
||||
pub connection_buf_size: usize,
|
||||
/// Базовый размер TCP-буфера (для «толстых» потоков).
|
||||
pub tcp_buffer_size: usize,
|
||||
/// Базовый размер UDP-буфера.
|
||||
pub udp_buffer_size: usize,
|
||||
/// Сколько байт читать из локального TCP-сокета за один проход.
|
||||
pub tcp_chunk_size: usize,
|
||||
|
||||
// 🔥 Единый конфиг для всех каналов Tokio
|
||||
/// Единая ёмкость всех Tokio-каналов (в пакетах). Маленькая — против bufferbloat.
|
||||
pub channel_capacity: usize,
|
||||
|
||||
// Буферы сокетов smoltcp
|
||||
// ── Буферы TCP-сокетов smoltcp: heavy (bulk) и light (интерактив) ──
|
||||
/// RX-буфер «толстого» TCP-сокета.
|
||||
pub tcp_rx_heavy: usize,
|
||||
/// TX-буфер «толстого» TCP-сокета.
|
||||
pub tcp_tx_heavy: usize,
|
||||
/// RX-буфер «тонкого» TCP-сокета.
|
||||
pub tcp_rx_light: usize,
|
||||
/// TX-буфер «тонкого» TCP-сокета.
|
||||
pub tcp_tx_light: usize,
|
||||
|
||||
// ── Буферы UDP-сокетов smoltcp: данные + слоты метаданных датаграмм ──
|
||||
/// Буфер данных «толстого» UDP-сокета.
|
||||
pub udp_buf_heavy: usize,
|
||||
/// Слотов метаданных датаграмм у «толстого» UDP-сокета.
|
||||
pub udp_meta_heavy: usize,
|
||||
/// Буфер данных «тонкого» UDP-сокета.
|
||||
pub udp_buf_light: usize,
|
||||
/// Слотов метаданных датаграмм у «тонкого» UDP-сокета.
|
||||
pub udp_meta_light: usize,
|
||||
}
|
||||
|
||||
@@ -84,6 +112,8 @@ impl NetworkConfig {
|
||||
}
|
||||
}
|
||||
|
||||
/// Инициализирует глобальный конфиг из MTU. Повторный вызов безвреден, но
|
||||
/// логирует предупреждение (конфиг неизменяем после первой установки).
|
||||
pub fn init_global(system_mtu: usize) {
|
||||
let config = Self::new(system_mtu);
|
||||
if GLOBAL_NET_CONFIG.set(config).is_err() {
|
||||
@@ -91,6 +121,8 @@ impl NetworkConfig {
|
||||
}
|
||||
}
|
||||
|
||||
/// Доступ к глобальному конфигу. Паникует, если [`init_global`](Self::init_global)
|
||||
/// ещё не вызывали — это ошибка порядка инициализации, а не рантайм-ситуация.
|
||||
pub fn global() -> &'static Self {
|
||||
GLOBAL_NET_CONFIG
|
||||
.get()
|
||||
|
||||
@@ -1,3 +1,22 @@
|
||||
//! Мосты: перекачка данных между логическим потоком туннеля и реальным сокетом.
|
||||
//!
|
||||
//! Когда сервер открыл соединение к цели, его обслуживает один из мостов:
|
||||
//! [`run_tcp_bridge`] или [`run_udp_bridge`]. Каждый качает данные в обе стороны:
|
||||
//!
|
||||
//! - **upload** (интернет → туннель): читает из локального сокета и шлёт в muxer;
|
||||
//! - **download** (туннель → интернет): принимает из канала потока (`v_rx`) и
|
||||
//! пишет в локальный сокет.
|
||||
//!
|
||||
//! Ключевые свойства устойчивости (детали — в inline-комментариях):
|
||||
//! - **Graceful pause (anti-domino).** Если в upload все ноги одновременно легли,
|
||||
//! мост НЕ закрывается: чанк удерживается и переотправляется в пределах
|
||||
//! [`STREAM_PAUSE_BUDGET`]. Пока мы не читаем дальше — работает TCP
|
||||
//! backpressure, источник сам притормаживает, данные не теряются.
|
||||
//! - **Адаптивный write-timeout в download.** Медленный локальный сокет под
|
||||
//! высоким RTT получает больше времени на слив, прежде чем поток закроют.
|
||||
//! - **Гарантированная уборка.** [`StreamGuard`] на `Drop` снимает регистрацию
|
||||
//! потока в muxer — что бы ни завершило мост.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use crate::net::connection::muxer::{adaptive_write_timeout, Muxer};
|
||||
@@ -13,6 +32,8 @@ use tokio::sync::mpsc;
|
||||
use tokio::time::timeout;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
/// RAII-страж: при выходе из моста (любым путём) снимает регистрацию потока,
|
||||
/// гарантируя, что в muxer не останется «зомби»-записи.
|
||||
struct StreamGuard {
|
||||
stream_id: u32,
|
||||
muxer: Arc<Muxer>,
|
||||
@@ -24,6 +45,11 @@ impl Drop for StreamGuard {
|
||||
self.muxer.remove_stream(self.stream_id);
|
||||
}
|
||||
}
|
||||
|
||||
/// TCP-мост: гоняет данные между потоком туннеля и TCP-сокетом цели.
|
||||
///
|
||||
/// upload и download крутятся конкурентно; завершение любой половины через
|
||||
/// общий [`CancellationToken`] немедленно гасит вторую и запускает уборку.
|
||||
pub(crate) async fn run_tcp_bridge<R, W>(
|
||||
stream_id: u32,
|
||||
reader: R,
|
||||
@@ -140,6 +166,12 @@ pub(crate) async fn run_tcp_bridge<R, W>(
|
||||
}
|
||||
token.cancel();
|
||||
}
|
||||
/// UDP-мост: аналог TCP, но датаграммами и в одном `select`-цикле.
|
||||
///
|
||||
/// Отличия от TCP: нет потокового упорядочивания (датаграммы), есть idle-таймаут
|
||||
/// ([`BRIDGE_IDLE_TIMEOUT`]) на закрытие неактивной сессии, и приём идёт zero-copy
|
||||
/// прямо в `BytesMut` (`recv_buf` + `split().freeze()` без копии датаграммы).
|
||||
/// Мёртвый туннель не рвёт мост — датаграмма просто дропается (UDP без гарантий).
|
||||
pub(crate) async fn run_udp_bridge(
|
||||
stream_id: u32,
|
||||
socket: UdpSocket,
|
||||
|
||||
@@ -1,3 +1,22 @@
|
||||
//! Точки входа туннеля: установка соединений на стороне клиента и сервера.
|
||||
//!
|
||||
//! Здесь собирается всё ядро в две роли:
|
||||
//!
|
||||
//! - [`ClientHandler`] — клиентская сторона. [`connect`](ClientHandler::connect)
|
||||
//! поднимает [`MAX_TUNNEL_LEGS`] ног (с разбежкой по времени), сторожит смену
|
||||
//! сети и переводит локальный трафик ([`RawCastFrame`]) в потоки туннеля.
|
||||
//! Каждая нога делает [`perform_handshake`](ClientHandler::perform_handshake)
|
||||
//! (поддельный TLS + обмен ключами + auth-кадр) и крутится в [`TunnelEngine`].
|
||||
//! - [`ServerHandler`] — серверная сторона. Принимает соединение, проверяет, что
|
||||
//! это валидный Netrunner-`ClientHello`; если нет — **stealth-fallback**:
|
||||
//! прозрачно проксирует трафик на безобидный хост (`ubuntu.com:443`), маскируясь
|
||||
//! под обычный TLS и не выдавая себя сканерам/DPI.
|
||||
//! - [`SessionManager`] — реестр сессий сервера (одна сессия = один [`Muxer`],
|
||||
//! несколько ног).
|
||||
//!
|
||||
//! Обе роли сходятся на [`TunnelEngine`]: клиент задаёт `remote_addr`
|
||||
//! (реконнектит), сервер оставляет его пустым (нога просто завершается).
|
||||
|
||||
use std::{net::Ipv4Addr, sync::Arc};
|
||||
|
||||
use crate::{
|
||||
@@ -32,6 +51,7 @@ use tokio::{
|
||||
sync::mpsc,
|
||||
};
|
||||
|
||||
/// Реестр активных сессий сервера: `session_id` → общий на сессию [`Muxer`].
|
||||
pub struct SessionManager {
|
||||
sessions: DashMap<String, Arc<Muxer>>,
|
||||
}
|
||||
@@ -52,6 +72,8 @@ impl SessionManager {
|
||||
&self.sessions
|
||||
}
|
||||
|
||||
/// Возвращает muxer сессии, создавая его при первом обращении. Так вторая и
|
||||
/// последующие ноги одной сессии цепляются к тому же мультиплексору.
|
||||
pub fn get_or_create(&self, session_id: &str) -> Arc<Muxer> {
|
||||
self.sessions
|
||||
.entry(session_id.to_string())
|
||||
@@ -79,11 +101,14 @@ impl SessionManager {
|
||||
}
|
||||
}
|
||||
|
||||
/// Общий контракт обработчика входящего туннельного соединения (реализует сервер).
|
||||
#[async_trait::async_trait]
|
||||
pub trait TunnelHandler {
|
||||
/// Обрабатывает соединение до его завершения.
|
||||
async fn run(self) -> Result<(), AppError>;
|
||||
}
|
||||
|
||||
/// Обёртка над TCP-соединением: половинки сокета + накопительный буфер чтения.
|
||||
pub struct Connection {
|
||||
pub(crate) inbound: OwnedReadHalf,
|
||||
pub(crate) outbound: OwnedWriteHalf,
|
||||
@@ -101,14 +126,24 @@ impl Connection {
|
||||
}
|
||||
}
|
||||
|
||||
/// Клиентская сторона туннеля (набор статических операций).
|
||||
pub struct ClientHandler;
|
||||
impl ClientHandler {
|
||||
/// Узнаёт локальный IP «трюком с UDP-connect»: соединение к 8.8.8.8 без
|
||||
/// отправки заставляет ОС выбрать исходящий интерфейс, чей адрес мы и читаем.
|
||||
/// Нужно для детектора смены сети (Wi-Fi↔LTE).
|
||||
fn get_local_ip() -> Option<std::net::IpAddr> {
|
||||
let socket = std::net::UdpSocket::bind("0.0.0.0:0").ok()?;
|
||||
socket.connect("8.8.8.8:80").ok()?;
|
||||
socket.local_addr().ok().map(|a| a.ip())
|
||||
}
|
||||
|
||||
/// Проводит полный клиентский хендшейк по уже установленному TCP-сокету.
|
||||
///
|
||||
/// Шаги: послать поддельный `ClientHello` (профиль Chrome, SNI=`ubuntu.com`) →
|
||||
/// дождаться `ServerHello` и вывести ключи → зарядить шифр и кодек → отправить
|
||||
/// первый зашифрованный auth-кадр `Heartbeat` с `"session_id:leg_id"`.
|
||||
/// Возвращает половинки сокета и готовые кодеки.
|
||||
pub async fn perform_handshake(
|
||||
stream: tokio::net::TcpStream,
|
||||
session_id: &str,
|
||||
@@ -208,6 +243,12 @@ impl ClientHandler {
|
||||
Ok((conn.inbound, conn.outbound, rx_codec, tx_codec))
|
||||
}
|
||||
|
||||
/// Устанавливает одну ногу и крутит её движок до остановки.
|
||||
///
|
||||
/// Резолвит адрес (с тайм-аутом), создаёт TCP-сокет с анти-bufferbloat
|
||||
/// тюнингом буферов, делает хендшейк, регистрирует ногу в muxer и запускает
|
||||
/// [`TunnelEngine::run`]. Возвращается только при остановке движка; снаружи
|
||||
/// (в [`connect`](ClientHandler::connect)) это уводит ногу на переподключение.
|
||||
async fn establish_leg(
|
||||
remote_proxy_addr: &str,
|
||||
leg_id: u32,
|
||||
@@ -294,6 +335,18 @@ impl ClientHandler {
|
||||
))
|
||||
}
|
||||
|
||||
/// Точка входа клиента: поднимает весь туннель и возвращает его [`Muxer`].
|
||||
///
|
||||
/// Запускает три группы фоновых задач:
|
||||
/// 1. **Ноги** — [`MAX_TUNNEL_LEGS`] задач, каждая в вечном цикле
|
||||
/// establish→disconnect→reconnect (со сдвигом старта [`LEG_STAGGER_DELAY`]).
|
||||
/// 2. **Сторож сети** — следит за сменой локального IP и при переключении
|
||||
/// сети сбрасывает все ноги (быстрый реконнект вместо зависших сокетов).
|
||||
/// 3. **Здоровье/топология** — периодический health-check и печать топологии.
|
||||
///
|
||||
/// Плюс главный цикл, который переводит локальные [`RawCastFrame`]
|
||||
/// (`rx_from_engine`) в потоки/данные туннеля и возвращает ответы обратно
|
||||
/// (`tx_to_engine`), с пер-сокетными буферами выгрузки против HOL-блокировки.
|
||||
pub async fn connect(
|
||||
remote_proxy_addr: &str,
|
||||
mut rx_from_engine: mpsc::Receiver<RawCastFrame>,
|
||||
@@ -594,6 +647,7 @@ impl ClientHandler {
|
||||
}
|
||||
}
|
||||
|
||||
/// Серверная сторона: обрабатывает одно входящее соединение.
|
||||
pub struct ServerHandler {
|
||||
pub(crate) conn: Connection,
|
||||
pub(crate) session_manager: Arc<SessionManager>,
|
||||
@@ -607,6 +661,13 @@ impl ServerHandler {
|
||||
}
|
||||
}
|
||||
|
||||
/// Stealth-fallback: прозрачно проксирует соединение на безобидный хост
|
||||
/// (`ubuntu.com:443`), когда клиент оказался «не наш».
|
||||
///
|
||||
/// Уже прочитанные байты (`initial_data`) пересылаются первыми, затем
|
||||
/// соединение склеивается в обе стороны через `tokio::io::copy`. Снаружи это
|
||||
/// выглядит как обычный визит на публичный сайт — сервер не выдаёт себя
|
||||
/// сканерам и активным пробам DPI.
|
||||
async fn handle_stealth_fallback(
|
||||
mut client_inbound: OwnedReadHalf,
|
||||
mut client_outbound: OwnedWriteHalf,
|
||||
@@ -639,6 +700,14 @@ impl ServerHandler {
|
||||
}
|
||||
}
|
||||
|
||||
/// Серверный жизненный цикл соединения: хендшейк → аутентификация → движок.
|
||||
///
|
||||
/// Три фазы, на каждой при малейшем несоответствии — stealth-fallback или отказ:
|
||||
/// 1. Принять `ClientHello` и собрать `ServerHello`; невалидный/чужой → fallback.
|
||||
/// 2. Расшифровать первый кадр и проверить auth-payload `"session_id:leg_id"`;
|
||||
/// неверный → [`ERR_AUTH_FAILED`].
|
||||
/// 3. Прицепить ногу к muxer сессии и крутить [`TunnelEngine`]; по завершении —
|
||||
/// эвикт ноги и отложенная уборка сессии, если ног не осталось.
|
||||
#[async_trait::async_trait]
|
||||
impl TunnelHandler for ServerHandler {
|
||||
async fn run(self) -> Result<(), AppError> {
|
||||
|
||||
@@ -1,3 +1,22 @@
|
||||
//! Движок одной ноги туннеля: жизненный цикл TCP-соединения и его reader/writer.
|
||||
//!
|
||||
//! [`TunnelEngine`] владеет одним физическим TCP+TLS-соединением и крутит его в
|
||||
//! [`run`](TunnelEngine::run), пока нога жива. Внутри одной итерации соединение
|
||||
//! расщепляется на две параллельные задачи tokio:
|
||||
//!
|
||||
//! - **Reader** — читает байты из сокета, прогоняет через [`RxCodec`]
|
||||
//! (расшифровка + сборка кадров), PONG'и инлайн обновляют RTT, остальные кадры
|
||||
//! уходят в [`StreamHandler`].
|
||||
//! - **Writer** — `biased`-`select!` по приоритету: heartbeat → control → data.
|
||||
//! Данные режутся на interleave-чанки (адаптивно под RTT) и шифруются
|
||||
//! [`TxCodec`] в [`handle_outbound`](TunnelEngine::handle_outbound); несколько
|
||||
//! кадров коалесятся в один `write_all` (экономия syscalls).
|
||||
//!
|
||||
//! При обрыве (EOF/ошибка) задачи останавливаются, их состояние (кодеки,
|
||||
//! приёмники, буфер) возвращается в `self`, и — если это клиент — нога идёт на
|
||||
//! переподключение с экспоненциальным backoff+jitter. Сервер (`remote_addr`
|
||||
//! пуст) при обрыве просто завершает задачу: реконнект инициирует клиент.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use bytes::{Bytes, BytesMut};
|
||||
@@ -22,30 +41,52 @@ use crate::{
|
||||
nrxp::{ErrorAction, FrameType, RxCodec, TxCodec, MAX_FRAME_PAYLOAD},
|
||||
};
|
||||
|
||||
/// Состояние ноги: работает или в процессе переподключения.
|
||||
#[derive(Debug, PartialEq, Clone, Copy)]
|
||||
pub enum LegStatus {
|
||||
Active,
|
||||
Reconnecting,
|
||||
}
|
||||
|
||||
/// Состояние и ресурсы одной ноги туннеля.
|
||||
///
|
||||
/// Половинки сокета, кодеки, приёмники каналов и буфер чтения хранятся в
|
||||
/// [`Option`], потому что на время работы reader/writer они «выдаются» в задачи
|
||||
/// через `take()`, а по завершении итерации возвращаются обратно — это позволяет
|
||||
/// переиспользовать кодеки (с их счётчиками nonce) между итерациями без Arc/Mutex.
|
||||
pub(crate) struct TunnelEngine {
|
||||
/// Читающая половина TCP-сокета (выдаётся reader-задаче).
|
||||
pub inbound: Option<OwnedReadHalf>,
|
||||
/// Пишущая половина TCP-сокета (выдаётся writer-задаче).
|
||||
pub outbound: Option<OwnedWriteHalf>,
|
||||
/// Адрес удалённой стороны; **пустой у сервера** (сервер не реконнектит).
|
||||
pub remote_addr: String,
|
||||
/// Идентификатор сессии (для логов и хендшейка реконнекта).
|
||||
pub session_id: String,
|
||||
/// Текущий статус ноги.
|
||||
pub leg_status: LegStatus,
|
||||
// 💡 ИЗМЕНЕНО: Кодеки теперь хранятся как Option без Arc/Mutex
|
||||
/// Кодек расшифровки входящего потока.
|
||||
pub rx_codec: Option<RxCodec>,
|
||||
/// Кодек шифрования исходящего потока.
|
||||
pub tx_codec: Option<TxCodec>,
|
||||
/// Накопительный буфер чтения из сокета.
|
||||
pub read_buf: BytesMut,
|
||||
/// Приёмник управляющих сообщений от muxer (Close/Heartbeat).
|
||||
pub control_rx: Option<Receiver<MuxMessage>>,
|
||||
/// Приёмник сообщений данных от muxer.
|
||||
pub data_rx: Option<Receiver<MuxMessage>>,
|
||||
/// Обработчик входящих кадров.
|
||||
pub handler: Arc<StreamHandler>,
|
||||
/// Идентификатор этой ноги.
|
||||
pub leg_id: u32,
|
||||
/// Общий мультиплексор туннеля.
|
||||
pub muxer: Arc<crate::net::connection::muxer::Muxer>,
|
||||
}
|
||||
|
||||
impl TunnelEngine {
|
||||
/// Переподключает ногу: заново резолвит хост (подхватывает смену IP/DNS),
|
||||
/// создаёт TCP-сокет с тюнингом буферов и проводит хендшейк заново. Возвращает
|
||||
/// свежие половинки сокета и кодеки.
|
||||
pub async fn attempt_reconnect(
|
||||
&mut self,
|
||||
) -> Result<(OwnedReadHalf, OwnedWriteHalf, RxCodec, TxCodec), AppError> {
|
||||
@@ -77,6 +118,13 @@ impl TunnelEngine {
|
||||
crate::net::ClientHandler::perform_handshake(stream, &self.session_id, self.leg_id).await
|
||||
}
|
||||
|
||||
/// Главный цикл ноги: переподключение (при нужде) → запуск reader/writer →
|
||||
/// ожидание завершения одной из задач → сбор состояния обратно → повтор.
|
||||
///
|
||||
/// Возвращает `Ok(())` при штатном завершении (например, сервер словил EOF);
|
||||
/// `Err` — когда исчерпан внутренний лимит реконнектов
|
||||
/// ([`MAX_INTERNAL_RECONNECT_ATTEMPTS`]) и управление надо вернуть внешнему
|
||||
/// циклу `establish_leg` (он перерезолвит DNS и сбросит счётчики).
|
||||
#[instrument(skip_all, fields(leg_id = self.leg_id))]
|
||||
pub async fn run(mut self) -> Result<(), AppError> {
|
||||
// Tracks consecutive internal reconnect failures. Resets to 0 on
|
||||
@@ -403,7 +451,14 @@ impl TunnelEngine {
|
||||
}
|
||||
}
|
||||
|
||||
// 💡 ИЗМЕНЕНО: Принимает &mut TxCodec, синхронное и сверхбыстрое шифрование
|
||||
/// Шифрует сообщение в один или несколько кадров и пишет их в сокет.
|
||||
///
|
||||
/// `Data` режется на кадры по [`MAX_FRAME_PAYLOAD`]; управляющие/UDP идут одним
|
||||
/// кадром. Срабатывает адаптивный по RTT дедлайн записи
|
||||
/// ([`adaptive_write_timeout`](super::muxer::adaptive_write_timeout)) — чтобы
|
||||
/// медленная, но живая нога не убивалась по жёсткому тайм-ауту. Несколько
|
||||
/// кадров коалесятся в один `write_all` (аналог sendmmsg для байт-потока:
|
||||
/// меньше syscalls); одиночный кадр пишется напрямую без лишней копии.
|
||||
async fn handle_outbound(
|
||||
outbound: &mut OwnedWriteHalf,
|
||||
tx_codec: &mut TxCodec,
|
||||
|
||||
@@ -1,3 +1,17 @@
|
||||
//! Диспетчеризация входящих кадров туннеля по их типу и `stream_id`.
|
||||
//!
|
||||
//! [`StreamHandler`] — это «маршрутизатор» на приёмной стороне: один кадр входит,
|
||||
//! и в зависимости от типа происходит одно из:
|
||||
//! - `Heartbeat` → ответить PONG / измерить RTT / переслать локально;
|
||||
//! - `Connect`/`UdpConnect` → (только сервер) открыть соединение к цели;
|
||||
//! - `Data`/`UdpData` → доставить данные в локальный поток (с backpressure);
|
||||
//! - `Close` → закрыть поток.
|
||||
//!
|
||||
//! Открытием реальных соединений к целям занимается [`RemoteOpener`] (есть только
|
||||
//! на сервере: у клиента `opener == None`, поэтому входящие `Connect` отвергаются).
|
||||
//! Каждое открытое соединение защищено [`CancellationToken`] — при эвикте/закрытии
|
||||
//! потока мост и установка соединения мгновенно обрываются.
|
||||
|
||||
use bytes::Bytes;
|
||||
use netrunner_logger::{debug, error, info, trace, warn};
|
||||
use std::sync::Arc;
|
||||
@@ -15,11 +29,21 @@ use crate::net::{
|
||||
};
|
||||
use crate::nrxp::{Frame, FrameType};
|
||||
|
||||
/// Открыватель реальных соединений к целям (серверная сторона туннеля).
|
||||
///
|
||||
/// На каждый входящий `Connect`/`UdpConnect` поднимает TCP/UDP-сокет к цели и
|
||||
/// запускает соответствующий мост, прокачивающий данные между туннелем и целью.
|
||||
pub struct RemoteOpener {
|
||||
pub muxer: Arc<Muxer>,
|
||||
}
|
||||
|
||||
impl RemoteOpener {
|
||||
/// Открывает TCP-соединение к `target` и запускает TCP-мост.
|
||||
///
|
||||
/// Всё происходит в отдельной задаче. Установка соединения (тайм-аут 7 с) и
|
||||
/// сам мост обёрнуты в `select!` с `token.cancelled()` — эвикт обрывает их
|
||||
/// немедленно. При неудаче подключения шлёт `Close` обратно в туннель. По
|
||||
/// завершении всегда снимает регистрацию потока.
|
||||
pub async fn open_tcp(
|
||||
&self,
|
||||
stream_id: u32,
|
||||
@@ -60,6 +84,8 @@ impl RemoteOpener {
|
||||
});
|
||||
}
|
||||
|
||||
/// Биндит UDP-сокет, «подключает» его к `target` и запускает UDP-мост.
|
||||
/// Так же защищено токеном отмены; по завершении снимает регистрацию потока.
|
||||
pub async fn open_udp(
|
||||
&self,
|
||||
stream_id: u32,
|
||||
@@ -86,6 +112,9 @@ impl RemoteOpener {
|
||||
}
|
||||
}
|
||||
|
||||
/// Маршрутизатор входящих кадров. Наличие `opener` определяет роль:
|
||||
/// `Some` — серверная сторона (умеет открывать соединения к целям),
|
||||
/// `None` — клиентская (входящие `Connect` отвергаются).
|
||||
pub(crate) struct StreamHandler {
|
||||
muxer: Arc<Muxer>,
|
||||
opener: Option<Arc<RemoteOpener>>,
|
||||
@@ -96,6 +125,9 @@ impl StreamHandler {
|
||||
Self { muxer, opener }
|
||||
}
|
||||
|
||||
/// Диспетчеризует один кадр по типу. Для `Data`/`UdpData` доставка идёт через
|
||||
/// `await` (backpressure ради сохранения порядка), для управляющих —
|
||||
/// в отдельных задачах, чтобы не блокировать reader ноги.
|
||||
pub(crate) async fn handle(&self, frame: Frame) {
|
||||
let stream_id = frame.header.stream_id;
|
||||
|
||||
@@ -152,6 +184,9 @@ impl StreamHandler {
|
||||
}
|
||||
}
|
||||
|
||||
/// Обрабатывает `Connect`/`UdpConnect`: регистрирует поток (получая токен
|
||||
/// отмены) и просит [`RemoteOpener`] открыть соединение. На клиенте (нет
|
||||
/// opener) — отказ с `Close`. `payload` несёт адрес цели строкой `"ip:port"`.
|
||||
async fn handle_conn_request(&self, stream_id: u32, payload: Bytes, is_udp: bool) {
|
||||
let target = String::from_utf8_lossy(&payload).to_string();
|
||||
|
||||
|
||||
@@ -1,3 +1,18 @@
|
||||
//! Подмодуль `connection` — собственно машинерия туннеля.
|
||||
//!
|
||||
//! Самый плотный по логике участок крейта. Делится по ролям:
|
||||
//!
|
||||
//! - [`muxer`] — **мультиплексор**: реестр потоков и ног, балансировка
|
||||
//! (`select_leg`), эвикт упавших ног, failover потоков, оценка RTT
|
||||
//! ([`GLOBAL_MIN_RTT`]).
|
||||
//! - [`engine`] — **движок ноги**: пара задач reader/writer на одно TCP-соединение,
|
||||
//! шифрование/дешифрование кадров, heartbeat, переподключение.
|
||||
//! - [`connection`] — обёртки над TCP/SOCKS, [`SessionManager`] и хендлеры:
|
||||
//! [`ClientHandler`] (SOCKS→потоки), [`ServerHandler`] (приём туннеля + stealth-fallback),
|
||||
//! [`TunnelHandler`].
|
||||
//! - [`handler`] — диспетчеризация входящих кадров по `stream_id`/типу к нужному мосту.
|
||||
//! - [`bridge`] — проксирование данных между потоком туннеля и реальным TCP/UDP-сокетом цели.
|
||||
|
||||
mod bridge;
|
||||
mod connection;
|
||||
mod engine;
|
||||
|
||||
@@ -1,3 +1,31 @@
|
||||
//! Мультиплексор: распределение логических потоков по физическим ногам туннеля.
|
||||
//!
|
||||
//! Сердце сетевого ядра и единственный по-настоящему конкурентный компонент.
|
||||
//! [`Muxer`] держит реестр ног (TCP-соединений) и потоков (`stream_id`) и решает,
|
||||
//! по какой ноге отправить каждый кадр. Спроектирован под высокую нагрузку:
|
||||
//!
|
||||
//! - **Lock-free горячий путь.** Реестры — это [`DashMap`] (шардированный), а
|
||||
//! снапшот ног для выбора — [`ArcSwap`] (чтение = атомарный bump `Arc`, без
|
||||
//! read-guard). См. поле `active_legs_cache`.
|
||||
//! - **Sticky-привязка + ребаланс.** Поток «прилипает» к ноге
|
||||
//! (`stream_bindings`), но при её падении мгновенно переезжает на лучшую из
|
||||
//! оставшихся (`select_leg`). Среди равных по качеству ног — round-robin, чтобы
|
||||
//! всплеск новых потоков не сел на одну «лучшую» ногу (thundering herd).
|
||||
//! - **Anti-domino failover.** Падение ноги НЕ закрывает поток: дохлая нога
|
||||
//! эвиктится, кадр переотправляется на соседнюю; `Err` только когда живых ног
|
||||
//! нет вовсе — и тогда мост делает паузу с буфером, а не сброс (см.
|
||||
//! `send_to_network` и `run_tcp_bridge`).
|
||||
//! - **Анти-bufferbloat доставка.** Входящие кадры доставляются неблокирующим
|
||||
//! `try_send` (`dispatch_to_local`), чтобы один медленный потребитель не
|
||||
//! блокировал общий reader ноги (head-of-line).
|
||||
//!
|
||||
//! ## Адаптация под RTT
|
||||
//!
|
||||
//! [`GLOBAL_MIN_RTT`] обновляется по heartbeat'ам (EWMA). От него зависят
|
||||
//! [`adaptive_write_timeout`] (не убивать медленную, но живую ногу) и
|
||||
//! [`adaptive_batch_chunk`] (под высоким RTT слать кадры большими пачками,
|
||||
//! экономя syscalls).
|
||||
|
||||
use arc_swap::ArcSwap;
|
||||
use bytes::Bytes;
|
||||
use dashmap::DashMap;
|
||||
@@ -13,6 +41,7 @@ use crate::net::{DISPATCH_TO_LOCAL_TIMEOUT, MAX_TUNNEL_LEGS};
|
||||
use crate::net::INITIAL_RTT_MS;
|
||||
use crate::nrxp::FrameType;
|
||||
|
||||
/// Атомарная статистика одной ноги: переданные/принятые байты и сглаженный RTT.
|
||||
#[derive(Default, Debug)]
|
||||
pub struct LegStats {
|
||||
pub tx_bytes: AtomicU64,
|
||||
@@ -20,12 +49,18 @@ pub struct LegStats {
|
||||
pub rtt_ms: AtomicU32,
|
||||
}
|
||||
|
||||
/// Атомарная статистика одного потока: переданные/принятые байты.
|
||||
#[derive(Default, Debug)]
|
||||
pub struct StreamStats {
|
||||
pub tx_bytes: AtomicU64,
|
||||
pub rx_bytes: AtomicU64,
|
||||
}
|
||||
|
||||
/// Одна нога туннеля = одно физическое TCP+TLS-соединение.
|
||||
///
|
||||
/// Два раздельных канала к writer-задаче ноги: `control_tx` (Close/Heartbeat,
|
||||
/// приоритетные) и `data_tx` (данные, с backpressure). `Clone` дёшев — внутри
|
||||
/// `Arc`/`Sender`, поэтому ногу можно копировать из кэша без затрат.
|
||||
#[derive(Clone)]
|
||||
struct MuxLeg {
|
||||
id: u32,
|
||||
@@ -35,6 +70,8 @@ struct MuxLeg {
|
||||
}
|
||||
|
||||
impl MuxLeg {
|
||||
/// Степень загруженности `data`-канала: 0.0 — пусто, 1.0 — канал полностью
|
||||
/// забит. Используется в скоринге ног при выборе (`select_leg`).
|
||||
fn congestion_factor(&self) -> f64 {
|
||||
let max = self.data_tx.max_capacity();
|
||||
let current_capacity = self.data_tx.capacity();
|
||||
@@ -43,6 +80,10 @@ impl MuxLeg {
|
||||
}
|
||||
}
|
||||
|
||||
/// Генератор `stream_id`, разводящий клиента и сервер по чётности.
|
||||
///
|
||||
/// Клиент выдаёт нечётные id (1,3,5…), сервер — чётные (2,4,6…). Так две стороны
|
||||
/// независимо открывают потоки, не споря за номера. Шаг — `+2`, атомарно.
|
||||
struct IdGenerator {
|
||||
counter: AtomicU32,
|
||||
}
|
||||
@@ -58,6 +99,8 @@ impl IdGenerator {
|
||||
}
|
||||
}
|
||||
|
||||
/// Единица передачи через muxer: что отправить (`data`), какого типа и в какой
|
||||
/// поток. Передаётся по каналам ноги к её writer-задаче.
|
||||
#[derive(Clone)]
|
||||
pub struct MuxMessage {
|
||||
pub(crate) stream_id: u32,
|
||||
@@ -96,8 +139,11 @@ pub fn adaptive_batch_chunk(base: usize) -> usize {
|
||||
base.saturating_mul(factor)
|
||||
}
|
||||
|
||||
/// Мультиплексор туннеля. Дёшево клонируется (всё внутри `Arc`) и шарится между
|
||||
/// всеми задачами ног и потоков.
|
||||
#[derive(Clone)]
|
||||
pub struct Muxer {
|
||||
/// Источник истины по ногам (id → нога). Шардированная карта, lock-free.
|
||||
legs: Arc<DashMap<u32, MuxLeg>>,
|
||||
// 🔥 ОПТИМИЗАЦИЯ: полностью lock-free кэш горячего пути.
|
||||
// ArcSwap: чтение (load_full) — атомарный bump Arc без блокировок; запись
|
||||
@@ -105,11 +151,16 @@ pub struct Muxer {
|
||||
// чтение брало read-guard.
|
||||
active_legs_cache: Arc<ArcSwap<Vec<MuxLeg>>>,
|
||||
|
||||
// Добавили CancellationToken для предотвращения утечек памяти (Зомби-задач)
|
||||
/// Реестр потоков: id → (канал доставки данных, статистика, токен отмены).
|
||||
/// Токен мгновенно убивает связанные с потоком задачи при `remove_stream`.
|
||||
streams: Arc<DashMap<u32, (Sender<Bytes>, Arc<StreamStats>, CancellationToken)>>,
|
||||
/// Sticky-привязка потока к ноге (`stream_id` → `leg_id`).
|
||||
stream_bindings: Arc<DashMap<u32, u32>>,
|
||||
/// Время отправки PING по каждой ноге — для измерения RTT по PONG.
|
||||
pending_pings: Arc<DashMap<u32, Instant>>,
|
||||
/// Генератор `stream_id` (чётность по роли).
|
||||
id_gen: Arc<IdGenerator>,
|
||||
/// Идентификатор сессии (для логов/топологии).
|
||||
session_id: Arc<String>,
|
||||
/// Rotating cursor for round-robin leg selection among similar-quality legs,
|
||||
/// so a burst of new streams spreads across legs instead of all binding to
|
||||
@@ -131,11 +182,16 @@ impl Muxer {
|
||||
}
|
||||
}
|
||||
|
||||
/// Пересобирает lock-free снапшот ног из источника истины (`legs`) и
|
||||
/// атомарно публикует его в `active_legs_cache`. Вызывается при любом
|
||||
/// изменении набора ног (add/remove).
|
||||
fn update_legs_cache(&self) {
|
||||
let new_cache: Vec<MuxLeg> = self.legs.iter().map(|kv| kv.value().clone()).collect();
|
||||
self.active_legs_cache.store(Arc::new(new_cache));
|
||||
}
|
||||
|
||||
/// Регистрирует новую ногу (после установки TCP+TLS). Если лимит
|
||||
/// [`MAX_TUNNEL_LEGS`] достигнут и это не обновление существующей — игнор.
|
||||
pub fn add_leg(
|
||||
&self,
|
||||
leg_id: u32,
|
||||
@@ -171,6 +227,10 @@ impl Muxer {
|
||||
self.stream_bindings.retain(|_, bound_leg| *bound_leg != leg_id);
|
||||
}
|
||||
|
||||
/// Безопасно эвиктит ногу, но только если её текущий `control_tx` совпадает с
|
||||
/// `tx` (защита от удаления ноги, уже переподключённой под тем же id). Сначала
|
||||
/// снимает привязки, потом обновляет кэш — чтобы конкурентный `select_leg` не
|
||||
/// привязался к эвиктируемой ноге.
|
||||
pub fn remove_leg(&self, leg_id: u32, tx: &Sender<MuxMessage>) {
|
||||
let should_remove = self
|
||||
.legs
|
||||
@@ -189,6 +249,7 @@ impl Muxer {
|
||||
}
|
||||
}
|
||||
|
||||
/// Безусловно удаляет ногу (без сверки канала) — при выходе её движка.
|
||||
pub fn force_remove_leg(&self, leg_id: u32) {
|
||||
if self.legs.remove(&leg_id).is_some() {
|
||||
self.clear_bindings_for_leg(leg_id);
|
||||
@@ -197,16 +258,25 @@ impl Muxer {
|
||||
}
|
||||
}
|
||||
|
||||
/// Сбрасывает все ноги и привязки (полная остановка туннеля).
|
||||
pub fn remove_all_legs(&self) {
|
||||
self.legs.clear();
|
||||
self.stream_bindings.clear();
|
||||
self.update_legs_cache();
|
||||
}
|
||||
|
||||
/// Число активных ног.
|
||||
pub fn active_legs_count(&self) -> usize {
|
||||
self.legs.len()
|
||||
}
|
||||
|
||||
/// Выбирает ногу для отправки кадра потока `stream_id`.
|
||||
///
|
||||
/// Двухуровнево: (1) горячий путь — привязанный поток резолвит ногу по id
|
||||
/// прямо из `legs` (без скана и клонирования кэша); (2) новый/осиротевший
|
||||
/// поток скорится по всем ногам (RTT доминирует, congestion лишь модулирует),
|
||||
/// из ног в пределах 2× от лучшего скора выбирается round-robin, и привязка
|
||||
/// фиксируется. Подробности скоринга — в inline-комментариях ниже.
|
||||
fn select_leg(&self, stream_id: u32) -> Option<MuxLeg> {
|
||||
// 1. FAST PATH (hot, per data frame): a bound stream resolves its leg by
|
||||
// id straight from the legs map — no full-cache Arc clone and no vector
|
||||
@@ -267,10 +337,13 @@ impl Muxer {
|
||||
None
|
||||
}
|
||||
|
||||
/// Запоминает момент отправки PING по ноге (для замера RTT по PONG).
|
||||
pub fn record_ping_sent(&self, leg_id: u32) {
|
||||
self.pending_pings.insert(leg_id, Instant::now());
|
||||
}
|
||||
|
||||
/// Обрабатывает PONG: считает RTT и обновляет сглаженную оценку (EWMA, α=0.25),
|
||||
/// затем пересчитывает глобальный минимум [`GLOBAL_MIN_RTT`] по всем ногам.
|
||||
pub async fn record_pong(&self, leg_id: u32) {
|
||||
if let Some((_, start_time)) = self.pending_pings.remove(&leg_id) {
|
||||
let measured = start_time.elapsed().as_millis() as u32;
|
||||
@@ -298,6 +371,15 @@ impl Muxer {
|
||||
}
|
||||
}
|
||||
|
||||
/// Отправляет кадр в сеть, выбирая ногу и применяя стратегию по типу кадра.
|
||||
///
|
||||
/// - **Данные** (`Data`/`UdpData`): `send().await` (backpressure) с
|
||||
/// anti-domino failover в цикле — при мёртвой ноге эвикт + переотправка на
|
||||
/// другую; `Err` лишь когда живых ног нет.
|
||||
/// - **Критичные** (`Close`/`Heartbeat`): надёжно через `send().await`
|
||||
/// (потеря Close течёт ресурсы, потеря PONG валит health-check).
|
||||
/// - **Прочий контроль**: `try_send`; при переполнении кадр дропается с
|
||||
/// сигналом `ControlChannelFull`, не блокируя.
|
||||
#[instrument(skip(self, message), fields(session_id = %self.session_id, stream_id = message.stream_id, frame = ?message.frame_type))]
|
||||
pub async fn send_to_network(&self, mut message: MuxMessage) -> Result<(), AppError> {
|
||||
let is_data = matches!(message.frame_type, FrameType::Data | FrameType::UdpData);
|
||||
@@ -428,6 +510,7 @@ impl Muxer {
|
||||
}
|
||||
}
|
||||
|
||||
/// Удобная обёртка для отправки данных потока (выбирает `Data`/`UdpData`).
|
||||
pub async fn send_data_safe(
|
||||
&self,
|
||||
stream_id: u32,
|
||||
@@ -446,6 +529,7 @@ impl Muxer {
|
||||
.await
|
||||
}
|
||||
|
||||
/// Удобная обёртка для отправки управляющего кадра заданного типа.
|
||||
pub(crate) async fn send_control(
|
||||
&self,
|
||||
stream_id: u32,
|
||||
@@ -460,6 +544,8 @@ impl Muxer {
|
||||
.await
|
||||
}
|
||||
|
||||
/// Регистрирует поток и возвращает его [`CancellationToken`]. Канал `tx`
|
||||
/// используется для доставки входящих данных потоку (`dispatch_to_local`).
|
||||
pub fn register_stream(&self, stream_id: u32, tx: Sender<Bytes>) -> CancellationToken {
|
||||
let token = CancellationToken::new();
|
||||
self.streams.insert(
|
||||
@@ -469,6 +555,8 @@ impl Muxer {
|
||||
token
|
||||
}
|
||||
|
||||
/// Удаляет поток, отменяя его токен (мгновенно гасит связанные задачи) и
|
||||
/// снимая привязку к ноге.
|
||||
pub fn remove_stream(&self, stream_id: u32) {
|
||||
// 🔥 Мгновенно убиваем "зомби-задачи", привязанные к стриму!
|
||||
if let Some((_, (_, _, token))) = self.streams.remove(&stream_id) {
|
||||
@@ -546,16 +634,24 @@ impl Muxer {
|
||||
}
|
||||
}
|
||||
|
||||
/// Учитывает принятые ногой байты в её статистике.
|
||||
pub fn record_leg_rx(&self, leg_id: u32, bytes: u64) {
|
||||
if let Some(leg) = self.legs.get(&leg_id) {
|
||||
leg.stats.rx_bytes.fetch_add(bytes, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
/// Следующий свободный `stream_id` (с учётом чётности роли).
|
||||
pub fn next_stream_id(&self) -> u32 {
|
||||
self.id_gen.next()
|
||||
}
|
||||
|
||||
/// Прогоняет health-check по всем ногам: PING с уникальным probe-потоком и
|
||||
/// ожидание PONG в пределах [`HEALTH_CHECK_TIMEOUT`](crate::net::HEALTH_CHECK_TIMEOUT).
|
||||
///
|
||||
/// Тонкости (см. inline): закрытый канал → немедленный эвикт; временно
|
||||
/// полный → пропуск цикла (нога жива, просто занята); перед эвиктом по
|
||||
/// тайм-ауту сверяется, что нога не переподключилась под тем же id.
|
||||
pub async fn perform_health_check(&self) {
|
||||
let leg_ids: Vec<u32> = self.legs.iter().map(|kv| *kv.key()).collect();
|
||||
|
||||
@@ -642,6 +738,8 @@ impl Muxer {
|
||||
}
|
||||
}
|
||||
|
||||
/// Печатает в лог дерево топологии туннеля: ноги (трафик/RTT), виртуальные
|
||||
/// потоки и кумулятивные счётчики здоровья пайплайна. Чисто диагностика.
|
||||
pub fn print_topology_tree(&self) {
|
||||
let mut out = String::new();
|
||||
out.push_str(&format!(
|
||||
|
||||
@@ -1,8 +1,19 @@
|
||||
//! Все «магические числа» сетевого ядра в одном месте.
|
||||
//!
|
||||
//! Сгруппированы по назначению (пулы, тайм-ауты, аутентификация, порты, stealth,
|
||||
//! кодек, тюнинг сокетов). Многие значения — результат борьбы с конкретными
|
||||
//! проблемами (bufferbloat, «эффект домино» при падении ноги, рассинхрон часов);
|
||||
//! у таких констант в `///`-комментарии объяснено, **почему** именно это число, а
|
||||
//! не просто что оно значит. Меняя их, читайте обоснование рядом.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
// ── Connection pool ──────────────────────────────────────────────────────────
|
||||
/// Максимум одновременных smoltcp-сокетов (виртуальных соединений) на клиенте.
|
||||
pub const MAX_SOCKETS: usize = 256;
|
||||
/// Сколько параллельных TCP-ног держит туннель (для throughput и отказоустойчивости).
|
||||
pub const MAX_TUNNEL_LEGS: u32 = 4;
|
||||
/// Размер пула мультиплексоров.
|
||||
pub const MUXER_POOL_SIZE: usize = 3;
|
||||
/// Weight applied to observed congestion when scoring tunnel legs.
|
||||
pub const MUXER_CONGESTION_WEIGHT: f64 = 2000.0;
|
||||
@@ -10,12 +21,19 @@ pub const MUXER_CONGESTION_WEIGHT: f64 = 2000.0;
|
||||
pub const INITIAL_RTT_MS: u32 = 250;
|
||||
|
||||
// ── Timeouts ─────────────────────────────────────────────────────────────────
|
||||
/// Тайм-аут TCP-хендшейка к целевому хосту (серверная сторона).
|
||||
pub const TCP_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(20);
|
||||
/// Простой UDP-сессии, после которого она считается завершённой.
|
||||
pub const UDP_IDLE_TIMEOUT: Duration = Duration::from_secs(15);
|
||||
/// Глобальный простой соединения до его закрытия.
|
||||
pub const GLOBAL_IDLE_TIMEOUT: Duration = Duration::from_secs(120);
|
||||
/// Период health-check'ов ног туннеля (heartbeat/проверка живости).
|
||||
pub const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(3);
|
||||
/// Сколько ждать ответа на health-check, прежде чем счесть ногу мёртвой.
|
||||
pub const HEALTH_CHECK_TIMEOUT: Duration = Duration::from_secs(20);
|
||||
/// Пауза перед переподключением упавшей ноги.
|
||||
pub const LEG_RECONNECT_DELAY: Duration = Duration::from_secs(2);
|
||||
/// Простой моста (стрима) до его закрытия.
|
||||
pub const BRIDGE_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
/// Max time to wait for a local app socket to accept downloaded data.
|
||||
/// If the app's receive buffer stays full longer than this, the connection
|
||||
@@ -62,10 +80,15 @@ pub const TOPOLOGY_PRINT_INTERVAL: Duration = Duration::from_secs(10);
|
||||
pub const STATS_LOG_INTERVAL: Duration = Duration::from_secs(5);
|
||||
|
||||
// ── Authentication ───────────────────────────────────────────────────────────
|
||||
/// Длительность одного шага time-based auth-тега в секундах (TOTP-«окно»).
|
||||
/// Тег меняется раз в 60 с — см. [`SessionAuth`](crate::nrxp).
|
||||
pub const AUTH_TIME_STEP: u64 = 60;
|
||||
/// Допуск на рассинхрон часов при проверке тега: ±2 шага (~±2 минуты).
|
||||
/// Сужает окно replay, оставляя запас под дрейф NTP и сетевые задержки.
|
||||
pub const AUTH_WINDOW_SIZE: u64 = 2;
|
||||
|
||||
// ── Well-known ports ─────────────────────────────────────────────────────────
|
||||
// Известные порты для эвристик классификации трафика (heavy/light, спец-обработка).
|
||||
pub const DNS_PORT: u16 = 53;
|
||||
pub const HTTP_PORT: u16 = 80;
|
||||
pub const HTTPS_PORT: u16 = 443;
|
||||
|
||||
@@ -1,3 +1,20 @@
|
||||
//! Диагностика туннеля: события, метрики, счётчики и снапшоты.
|
||||
//!
|
||||
//! Холодный путь, не влияющий на горячую обработку пакетов. Состоит из трёх
|
||||
//! независимых механизмов:
|
||||
//!
|
||||
//! 1. **Канал событий** ([`DiagnosticsEvent`]). Любой код из любого места делает
|
||||
//! fire-and-forget [`send_diag_event`]; потребитель (на клиенте — движок, на
|
||||
//! сервере — `Network::run`) держит приёмник из [`init_diagnostics`].
|
||||
//! 2. **Атомарные счётчики** ([`DIAG_COUNTERS`]). Глобальные накопители событий
|
||||
//! (сбои загрузки, отвалы ног, переполнения каналов и т.п.), инкрементируются
|
||||
//! прямо в местах событий через `Relaxed`.
|
||||
//! 3. **Снапшоты** ([`DiagnosticsSnapshot`] + [`DiagnosticsStore`]). По триггеру
|
||||
//! собирается полный срез состояния (движок, ноги, сокеты, счётчики) и
|
||||
//! кольцевым буфером хранятся последние N для выгрузки в JSON.
|
||||
//!
|
||||
//! Все структуры `Serialize` — снапшоты отдаются наружу как JSON для отладки.
|
||||
|
||||
use std::{
|
||||
collections::VecDeque,
|
||||
sync::{
|
||||
@@ -39,7 +56,11 @@ pub fn send_diag_event(event: DiagnosticsEvent) {
|
||||
|
||||
// ── Event types ───────────────────────────────────────────────────────────────
|
||||
|
||||
/// What triggered this diagnostics snapshot.
|
||||
/// Событие, которое стало триггером снапшота (и единица потока событий).
|
||||
///
|
||||
/// Каждый вариант — это «что-то пошло не так или примечательно» на горячем пути:
|
||||
/// сбой выгрузки, backpressure, отвал/переподключение ноги, переполнение
|
||||
/// управляющего канала, застрявшая запись в туннель. Сериализуется с тегом `kind`.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
#[serde(tag = "kind", rename_all = "snake_case")]
|
||||
pub enum DiagnosticsEvent {
|
||||
@@ -79,6 +100,8 @@ pub enum DiagnosticsEvent {
|
||||
|
||||
// ── Per-snapshot sub-structs (all Serialize) ──────────────────────────────────
|
||||
|
||||
/// Метрики движка (только клиент): трафик, глубины очередей устройства, свободное
|
||||
/// место в каналах TUN↔engine.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct EngineMetrics {
|
||||
pub rx_total_mb: f64,
|
||||
@@ -97,6 +120,7 @@ pub struct EngineMetrics {
|
||||
pub tun_rx_channel_free: usize,
|
||||
}
|
||||
|
||||
/// Метрики одной ноги туннеля: RTT, объёмы, фактор перегруженности канала.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct LegMetrics {
|
||||
pub leg_id: u32,
|
||||
@@ -109,6 +133,7 @@ pub struct LegMetrics {
|
||||
pub data_channel_capacity: usize,
|
||||
}
|
||||
|
||||
/// Сводка по туннелю в целом: глобальный мин. RTT, метрики всех ног, число потоков.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct TunnelMetrics {
|
||||
pub global_min_rtt_ms: u32,
|
||||
@@ -116,6 +141,7 @@ pub struct TunnelMetrics {
|
||||
pub total_streams: usize,
|
||||
}
|
||||
|
||||
/// Метрики одного smoltcp-сокета (только клиент): состояние, очереди, congestion.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct SocketMetrics {
|
||||
pub stream_id: u32,
|
||||
@@ -143,6 +169,7 @@ pub struct ErrorCounters {
|
||||
pub stream_errors: u64,
|
||||
}
|
||||
|
||||
/// Полный срез состояния системы на момент триггер-события.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct DiagnosticsSnapshot {
|
||||
pub timestamp_ms: u64,
|
||||
@@ -159,6 +186,9 @@ pub struct DiagnosticsSnapshot {
|
||||
|
||||
// ── Atomic error counters (global, updated at event sites) ───────────────────
|
||||
|
||||
/// Глобальные атомарные счётчики событий, инкрементируемые в местах их
|
||||
/// возникновения. Включают «воронку» доставки загрузки (`mux_dispatch_*`),
|
||||
/// по которой видно, куда деваются входящие кадры.
|
||||
pub struct DiagnosticsCounters {
|
||||
pub upload_fails: AtomicU64,
|
||||
pub download_backpressure: AtomicU64,
|
||||
@@ -210,6 +240,8 @@ pub static DIAG_COUNTERS: DiagnosticsCounters = DiagnosticsCounters::new();
|
||||
|
||||
// ── DiagnosticsStore — holds the last N snapshots ─────────────────────────────
|
||||
|
||||
/// Кольцевой буфер последних N снапшотов под мьютексом (холодный путь —
|
||||
/// блокировка не вредит). Отдаёт их наружу как JSON для отладки.
|
||||
pub struct DiagnosticsStore {
|
||||
snapshots: Mutex<VecDeque<DiagnosticsSnapshot>>,
|
||||
max_snapshots: usize,
|
||||
|
||||
@@ -1,3 +1,40 @@
|
||||
//! # Сетевое ядро (`net`) — оркестровка живого туннеля
|
||||
//!
|
||||
//! Самый верхний блок крейта: здесь синхронный разбор протокола ([`nrxp`](crate::nrxp))
|
||||
//! и асинхронная сеть `tokio` соединяются в работающий мультиплексированный
|
||||
//! туннель. Всё, что ниже — крипта, кадры, маскировка — лишь «кирпичи», а этот
|
||||
//! блок строит из них дом: слушает соединения, держит ноги туннеля, балансирует
|
||||
//! потоки и проксирует трафик к целям.
|
||||
//!
|
||||
//! ## Состав блока
|
||||
//!
|
||||
//! | Файл/подмодуль | Ответственность |
|
||||
//! |-----------------------|---------------------------------------------------------------------|
|
||||
//! | [`config`] | [`NetworkConfig`] — MTU-зависимые размеры буферов и каналов. |
|
||||
//! | `constants` | Тайм-ауты, лимиты, порты, тюнинг анти-bufferbloat (re-export `*`). |
|
||||
//! | [`diagnostics`] | Сбор и снапшоты метрик туннеля (RTT, очереди, события ног). |
|
||||
//! | `connection` | Ядро: соединения, мультиплексор, движок туннеля, мосты, хендлеры. |
|
||||
//!
|
||||
//! Подмодуль `connection` — самый крупный; его части:
|
||||
//! - **muxer** — мультиплексор: распределяет потоки по ногам, балансирует, эвиктит;
|
||||
//! - **engine** — `TunnelEngine`: reader/writer-задачи одной ноги, шифр/дешифр, heartbeat;
|
||||
//! - **connection** — обёртки над TCP, SOCKS, `SessionManager`, хендлеры клиента/сервера;
|
||||
//! - **handler** — обработка входящих кадров (диспетчеризация по `stream_id`/типу);
|
||||
//! - **bridge** — проксирование TCP/UDP между потоком туннеля и реальной целью.
|
||||
//!
|
||||
//! ## Ключевая идея
|
||||
//!
|
||||
//! **Одно TCP-соединение (нога) = много логических потоков (`stream_id`).** Ради
|
||||
//! устойчивости ног может быть несколько ([`MAX_TUNNEL_LEGS`]); поток «прилипает»
|
||||
//! к ноге, а при её падении бесшовно переезжает на соседнюю (failover вместо
|
||||
//! каскадного сброса). Подробности балансировки — в `connection::muxer`.
|
||||
//!
|
||||
//! ## Что экспортируется
|
||||
//!
|
||||
//! Наружу крейта выходят высокоуровневые сущности: [`NetworkConfig`],
|
||||
//! хендлеры/менеджер сессий (через `connection`) и [`Muxer`] с глобальной оценкой
|
||||
//! [`GLOBAL_MIN_RTT`], а также все константы.
|
||||
|
||||
mod config;
|
||||
mod connection;
|
||||
mod constants;
|
||||
|
||||
Reference in New Issue
Block a user