nineap/netrunner-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

clean comments

Browse Source
This commit is contained in:
Kirill
2026-04-03 12:06:58 +07:00
parent b90083539c
commit e53a95692e
18 changed files with 100 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/net/connection/bridge.rs
+8 -15
View File
@@ -10,11 +10,8 @@ use tokio::net::UdpSocket;
use tokio::sync::mpsc;
use tokio::time::timeout;
/// Тайм-аут бездействия. Если данных нет 5 минут — закрываем мост.
const IDLE_TIMEOUT: Duration = Duration::from_secs(300);
/// RAII Guard: гарантирует удаление стрима из Muxer при любом выходе из функции,
/// даже если произошла паника (panic) или ранняя ошибка.
struct StreamGuard {
stream_id: u32,
muxer: Arc<Muxer>,
@@ -37,7 +34,6 @@ pub(crate) async fn run_tcp_bridge<R, W>(
R: tokio::io::AsyncReadExt + Unpin,
W: tokio::io::AsyncWriteExt + Unpin,
{
// Инициализируем защитника ресурсов
let _guard = StreamGuard {
stream_id,
muxer: muxer.clone(),
@@ -46,15 +42,14 @@ pub(crate) async fn run_tcp_bridge<R, W>(
let mut buf = BytesMut::with_capacity(NetworkConfig::global().tcp_buffer_size);
loop {
// Оборачиваем весь select в таймаут, чтобы убивать "мертвые" задачи
let select_res = timeout(IDLE_TIMEOUT, async {
tokio::select! {
// Данные из реального TCP сокета -> в Muxer
res = reader.read_buf(&mut buf) => {
match res {
Ok(0) => {
debug!(stream_id, "TCP Socket reached EOF");
return Ok(false); // Выход из цикла
return Ok(false);
}
Ok(_) => {
let msg = MuxMessage {
@@ -62,7 +57,7 @@ pub(crate) async fn run_tcp_bridge<R, W>(
frame_type: FrameType::Data,
data: buf.split().freeze(),
};
// Если туннель закрыт — выходим
if muxer.send_to_network(msg).is_err() {
return Ok(false);
}
@@ -75,7 +70,7 @@ pub(crate) async fn run_tcp_bridge<R, W>(
}
}
// Данные из туннеля -> в реальный TCP сокет
maybe_data = v_rx.recv() => {
match maybe_data {
Some(data) => {
@@ -98,7 +93,7 @@ pub(crate) async fn run_tcp_bridge<R, W>(
match select_res {
Ok(Ok(true)) => continue,
Ok(Ok(false)) => break, // Штатное закрытие
Ok(Ok(false)) => break,
Ok(Err(e)) => {
debug!(stream_id, "Bridge closing due to error: {}", e);
break;
@@ -110,14 +105,12 @@ pub(crate) async fn run_tcp_bridge<R, W>(
}
}
// Попытка вежливо сказать удаленной стороне, что мы закрылись
let _ = muxer.send_to_network(MuxMessage {
stream_id,
frame_type: FrameType::Close,
data: Bytes::new(),
});
// Небольшая пауза, чтобы пакет Close успел уйти в сеть перед тем, как Guard удалит стрим
tokio::time::sleep(Duration::from_millis(50)).await;
}
@@ -140,7 +133,7 @@ pub(crate) async fn run_udp_bridge(
loop {
let select_res = timeout(IDLE_TIMEOUT, async {
tokio::select! {
// Из интернета в туннель
res = socket.recv(&mut buf) => {
match res {
Ok(n) if n > 0 => {
@@ -159,7 +152,7 @@ pub(crate) async fn run_udp_bridge(
}
}
// Из туннеля в интернет
maybe_data = v_rx.recv() => {
match maybe_data {
Some(data) => {
@@ -178,7 +171,7 @@ pub(crate) async fn run_udp_bridge(
match select_res {
Ok(Ok(true)) => continue,
_ => break, // Любая ошибка или таймаут закрывают UDP мост
_ => break,
}
}
core/src/net/connection/connection.rs
+4 -15
View File
@@ -41,7 +41,6 @@ impl SessionManager {
pub fn get_or_create(&self, session_id: &str) -> Arc<Muxer> {
self.sessions
.entry(session_id.to_string())
// ОБНОВЛЕНО: Передаем session_id в конструктор Муксера
.or_insert_with(|| Arc::new(Muxer::new(false, session_id.to_string())))
.clone()
}
@@ -121,7 +120,6 @@ impl ClientHandler {
let (inbound, outbound) = stream.into_split();
let mut conn = Connection::new_raw(inbound, outbound);
// TLS Handshake
let ch = conn
.codec
.make_client_handshake(&BrowserProfile::CHROME_131, "ubuntu.com")
@@ -152,7 +150,6 @@ impl ClientHandler {
info!("{} TLS Handshake complete.", leg_name);
let handshake_payload = Bytes::from(format!("{}:{}", session_id, leg_id));
// Шифруем кастомный Handshake
let encrypted_handshake = conn
.codec
.encrypt_data(0, FrameType::Handshake, handshake_payload)
@@ -177,7 +174,7 @@ impl ClientHandler {
std::sync::Arc::new(StreamHandler::new(muxer.clone(), ConnectionRole::Client));
let engine = TunnelEngine {
leg_id, // ОБНОВЛЕНО: Передаем leg_id, чтобы движок писал стату
leg_id,
inbound: conn.inbound,
outbound: conn.outbound,
codec: conn.codec,
@@ -185,7 +182,7 @@ impl ClientHandler {
control_rx,
data_rx,
handler,
muxer: muxer.clone(), // Не забудь передать muxer (он нужен движку для record_leg_rx)
muxer: muxer.clone(),
};
engine.run().await.map_err(|e| e.to_string())?;
@@ -200,20 +197,17 @@ impl ClientHandler {
let session_id = SessionManager::generate_id();
info!("🔑 Generated Master Session ID: {}", session_id);
// ОБНОВЛЕНО: Передаем session_id в Муксер
let muxer = Arc::new(Muxer::new(true, session_id.clone()));
let registry: Arc<DashMap<u32, StreamContext>> = Arc::new(DashMap::new());
info!("🚀 Netrunner Multi-Path Tunnel Initializing (10 Legs max).");
// === ДИНАМИЧЕСКИЙ ПУЛ НА 10 НОГ ===
for id in 0..10 {
let addr = remote_proxy_addr.to_string();
let m = muxer.clone();
let sid = session_id.clone();
tokio::spawn(async move {
// Плавный запуск (staggered start), чтобы не задудосить свой же сервер
tokio::time::sleep(std::time::Duration::from_millis(id as u64 * 250)).await;
loop {
if let Err(e) = Self::establish_leg(&addr, id, m.clone(), &sid).await {
@@ -224,7 +218,6 @@ impl ClientHandler {
});
}
// === ДЕМОН СТАТИСТИКИ (КЛИЕНТ) ===
let m_weak = Arc::downgrade(&muxer);
tokio::spawn(async move {
while let Some(m_stats) = m_weak.upgrade() {
@@ -236,7 +229,6 @@ impl ClientHandler {
}
});
// === ЛОКАЛЬНАЯ МАРШРУТИЗАЦИЯ (RawCast -> Nrxp) ===
let muxer_inner = muxer.clone();
tokio::spawn(async move {
while let Some(raw_frame) = rx_from_engine.recv().await {
@@ -472,7 +464,6 @@ impl TunnelHandler for ServerHandler {
}
};
// РЕГИСТРИРУЕМ НОГУ
let muxer = self.session_manager.get_or_create(&session_id);
let (control_tx, control_rx) = mpsc::channel(NetworkConfig::global().muxer_capacity);
@@ -494,9 +485,8 @@ impl TunnelHandler for ServerHandler {
}
});
// ЗАПУСКАЕМ ДВИЖОК
let engine = TunnelEngine {
leg_id, // ОБНОВЛЕНО
leg_id,
inbound: self.conn.inbound,
outbound: self.conn.outbound,
codec: self.conn.codec,
@@ -504,12 +494,11 @@ impl TunnelHandler for ServerHandler {
control_rx,
data_rx,
handler,
muxer: muxer.clone(), // ОБНОВЛЕНО
muxer: muxer.clone(),
};
let _ = engine.run().await;
// ОЧИСТКА ПОСЛЕ ОТКЛЮЧЕНИЯ НОГИ
muxer.remove_leg(leg_id);
if muxer.active_legs_count() == 0 {
self.session_manager.remove(&session_id);
core/src/net/connection/engine.rs
+1 -2
View File
@@ -41,7 +41,6 @@ impl TunnelEngine {
let data_rx = self.data_rx;
let handler = self.handler;
// === ИЗВЛЕКАЕМ НОВЫЕ ПОЛЯ ===
let leg_id = self.leg_id;
let muxer = self.muxer.clone();
@@ -76,7 +75,7 @@ impl TunnelEngine {
return Err::<(), String>("EOF".into());
}
// === РЕГИСТРИРУЕМ ВХОДЯЩИЙ ТРАФИК ДЛЯ СТАТИСТИКИ ===
muxer.record_leg_rx(leg_id, n as u64);
let mut frames = Vec::new();
core/src/net/connection/handler.rs
-1
View File
@@ -149,7 +149,6 @@ impl StreamHandler {
}
async fn on_data(&self, stream_id: u32, payload: Bytes) {
// Здесь info может быть избыточным при большой нагрузке, но для отладки полезно
debug!(stream_id, "📦 [TCP Data] Size: {} bytes", payload.len());
self.muxer.dispatch_to_local(stream_id, payload);
}
core/src/net/connection/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
mod bridge; // Все модули делаем приватными
mod bridge;
mod connection;
mod engine;
mod handler;
core/src/net/connection/muxer.rs
+26 -26
View File
@@ -9,7 +9,7 @@ use tokio::sync::mpsc::Sender;
use crate::net::network::NetworkConfig;
use crate::nrxp::FrameType;
// === СТРУКТУРЫ СТАТИСТИКИ ===
#[derive(Default, Debug)]
pub struct LegStats {
pub tx_bytes: AtomicU64,
@@ -56,14 +56,14 @@ pub struct MuxMessage {
#[derive(Clone)]
pub struct Muxer {
legs: Arc<DashMap<u32, MuxLeg>>,
// Храним канал + статистику стрима
streams: Arc<DashMap<u32, (Sender<Bytes>, Arc<StreamStats>)>>,
id_gen: Arc<IdGenerator>,
session_id: Arc<String>,
}
impl Muxer {
// Конструктор теперь принимает session_id
pub fn new(is_client: bool, session_id: String) -> Self {
Self {
legs: Arc::new(DashMap::new()),
@@ -79,7 +79,7 @@ impl Muxer {
control_tx: Sender<MuxMessage>,
data_tx: Sender<MuxMessage>,
) {
// Ограничение: максимум 10 физических ног
if self.legs.len() >= 10 {
warn!(
leg_id,
@@ -115,43 +115,43 @@ impl Muxer {
let is_udp = matches!(frame_type, FrameType::UdpData | FrameType::UdpConnect);
// Собираем доступные ноги
let mut candidates: Vec<(u32, MuxLeg)> = self.legs
.iter()
.map(|kv| (*kv.key(), kv.value().clone()))
.filter(|(id, _)| {
// Фильтр по протоколу, если ноги разделены
if is_udp { id % 2 != 0 } else { id % 2 == 0 }
})
.collect();
// Если по протоколу ничего не нашли, берем любые активные
if candidates.is_empty() {
candidates = self.legs.iter().map(|kv| (*kv.key(), kv.value().clone())).collect();
}
// --- МАГИЯ ЗДЕСЬ ---
// Сортируем кандидатов по RTT (от меньшего к большему)
// Ноги с RTT = 0 (еще не проверенные) или таймаутом считаем медленными
candidates.sort_by_key(|(_, leg)| {
let rtt = leg.stats.rtt_ms.load(Ordering::Relaxed);
if rtt == 0 { 9999 } else { rtt }
});
// Выбираем из ТОП-2 лучших ног по RTT, используя stream_id для аффинити
// Это защитит от ситуации, когда все стримы прыгнут на одну ногу с RTT 10мс
let pool_size = std::cmp::min(candidates.len(), 2);
let target = &candidates[stream_id as usize % pool_size];
Some(target.clone())
}
// СИНХРОННАЯ ФУНКЦИЯ (с отбрасыванием пакетов при перегрузке)
pub fn send_to_network(&self, message: MuxMessage) -> Result<(), String> {
let stream_id = message.stream_id;
let size = message.data.len() as u64;
// Передаем stream_id в select_leg
let (leg_id, leg) = self
.select_leg(&message.frame_type, stream_id)
.ok_or_else(|| "MUXER: No active legs available".to_string())?;
@@ -166,7 +166,7 @@ impl Muxer {
match target_tx.try_send(message) {
Ok(_) => {
// Пишем стату только если пакет успешно ушел в канал
leg.stats.tx_bytes.fetch_add(size, Ordering::Relaxed);
if let Some(stream_ref) = self.streams.get(&stream_id) {
stream_ref
@@ -178,7 +178,7 @@ impl Muxer {
Ok(())
}
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
// Если очередь переполнена - просто дропаем пакет (Backpressure)
warn!(
leg_id,
"MUXER: Network queue full! Dropping outbound packet."
@@ -241,7 +241,7 @@ impl Muxer {
}
pub fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
// Получаем клон Sender и Stats из DashMap
let stream_opt = self
.streams
.get(&stream_id)
@@ -250,7 +250,7 @@ pub fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
if let Some((tx, stats)) = stream_opt {
let size = data.len() as u64;
// Используем try_send вместо send().await
match tx.try_send(data) {
Ok(_) => {
stats.rx_bytes.fetch_add(size, Ordering::Relaxed);
@@ -275,7 +275,7 @@ pub fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
}
pub async fn perform_health_check(&self) {
// Создаем снапшот ног, чтобы не держать lock DashMap слишком долго
let legs: Vec<(u32, Sender<MuxMessage>)> = self.legs.iter()
.map(|k| (*k.key(), k.value().control_tx.clone()))
.collect();
@@ -284,7 +284,7 @@ pub fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
let probe_stream_id = self.id_gen.next();
let (probe_tx, mut probe_rx) = tokio::sync::mpsc::channel(2);
// Регистрируем временный стрим для ответа на PING
self.register_stream(probe_stream_id, probe_tx);
let msg = MuxMessage {
@@ -295,9 +295,9 @@ pub fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
let start = std::time::Instant::now();
// Отправляем напрямую в ногу
if tx.try_send(msg).is_ok() {
// Ждем ответа с жестким таймаутом
match tokio::time::timeout(std::time::Duration::from_secs(2), probe_rx.recv()).await {
Ok(Some(_)) => {
let rtt = start.elapsed().as_millis() as u32;
@@ -307,7 +307,7 @@ pub fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
}
}
_ => {
// Если таймаут — задираем RTT до небес, чтобы select_leg ее не выбирал
if let Some(leg) = self.legs.get(&leg_id) {
leg.stats.rtt_ms.store(5000, Ordering::Relaxed);
warn!(leg_id, "❌ Leg Health Check Timeout (marked as slow)");
@@ -316,12 +316,12 @@ pub fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
}
}
// Обязательная очистка, чтобы не было утечки памяти в DashMap streams
self.remove_stream(probe_stream_id);
}
}
// ФОРМАТИРОВАНИЕ БАЙТ
fn format_size(bytes: u64) -> String {
const KB: u64 = 1024;
const MB: u64 = KB * 1024;
@@ -337,7 +337,7 @@ pub fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
}
}
// ВЫВОД КРАСИВОЙ СТАТИСТИКИ
pub fn print_topology_tree(&self) {
println!(
"\n🌐 Netrunner Tunnel Topology [Session: {}]",
core/src/net/network.rs
+29 -17
View File
@@ -22,13 +22,10 @@ impl Network {
pub async fn run(&self, token: CancellationToken) {
let addr = format!("{}:{}", self.host, self.port);
// Инициализируем глобальный конфиг сети (MTU, размеры буферов)
NetworkConfig::init_global(1500);
match self.role {
ConnectionRole::Client => {
// В новой архитектуре клиент запускается через EngineBuilder + TUN.
// Структура Network теперь используется только для запуска Сервера.
error!("Client mode cannot be run via Network::run anymore.");
error!("Please use EngineBuilder to initialize the TUN client.");
panic!("Legacy SOCKS5 client mode has been removed.");
@@ -47,7 +44,7 @@ impl Network {
if let Ok((stream, client_addr)) = res {
info!("New connection from {}", client_addr);
// Создаем соединение (init = true для сервера)
let conn = Connection::new(stream, true);
let handler = ServerHandler::new(conn);
@@ -72,40 +69,55 @@ pub struct NetworkConfig {
pub max_wire_frame_size: usize,
pub safe_payload_size: usize,
// --- ИЗМЕНЕНИЯ ЗДЕСЬ ---
pub tcp_buffer_size: usize, // Размер буфера для системного tokio::TcpStream
pub tcp_buffer_size: usize,
pub udp_buffer_size: usize,
pub muxer_capacity: usize, // Глобальные каналы (Muxer <-> Engine)
pub tcp_stream_capacity: usize, // Локальные каналы 1 сокета (Engine <-> Muxer)
pub udp_stream_capacity: usize, // Локальные каналы 1 сокета (Engine <-> Muxer)
pub muxer_capacity: usize,
pub tcp_stream_capacity: usize,
pub udp_stream_capacity: usize,
pub smoltcp_socket_buf: usize,
pub tcp_chunk_size: usize,
}
pub tcp_buf_heavy: usize,
pub tcp_buf_light: usize,
pub udp_buf_heavy: usize,
pub udp_meta_heavy: usize,
pub udp_buf_light: usize,
pub udp_meta_light: usize,
}
impl NetworkConfig {
pub fn new(system_mtu: usize) -> Self {
let transport_overhead = 48;
let max_wire_frame = system_mtu.saturating_sub(transport_overhead);
let safe_payload = max_wire_frame.saturating_sub(64);
let chunk_size = 16 * 1024;
let heavy_buf = 1024 * 1024;
Self {
mtu: system_mtu,
max_wire_frame_size: max_wire_frame,
safe_payload_size: safe_payload,
// Буферы ОС/Tokio (держим с запасом под быстрые всплески)
tcp_buffer_size: 1024 * 1024,
udp_buffer_size: 64 * 1024,
tcp_buffer_size: heavy_buf,
udp_buffer_size: 256 * 1024,
muxer_capacity: 512,
tcp_stream_capacity: 16,
udp_stream_capacity: 128,
// Настройки smoltcp
smoltcp_socket_buf: 256 * 1024,
tcp_chunk_size: 16 * 1024,
smoltcp_socket_buf: heavy_buf,
tcp_chunk_size: chunk_size,
tcp_buf_heavy: heavy_buf,
tcp_buf_light: 64 * 1024,
udp_buf_heavy: heavy_buf,
udp_meta_heavy: 512,
udp_buf_light: 32 * 1024,
udp_meta_light: 16,
}
}