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

remove comments

Browse Source
This commit is contained in:
Kirill
2026-03-20 17:06:13 +07:00
parent 2b0de62bc0
commit e234bd4006
6 changed files with 20 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/connections/tcp_connection.rs
+4 -14
View File
@@ -35,11 +35,9 @@ impl TcpConnection {
let stream_id = muxer.next_id(); let stream_id = muxer.next_id();
tokio::spawn(async move { tokio::spawn(async move {
// 1. Регистрируем виртуальный стрим
let (v_tx, mut v_rx) = mpsc::channel(1024); let (v_tx, mut v_rx) = mpsc::channel(1024);
muxer.register_stream(stream_id, v_tx).await; muxer.register_stream(stream_id, v_tx).await;
// 2. Отправляем запрос на соединение на удаленный сервер
let connect_payload = target_addr.to_string(); let connect_payload = target_addr.to_string();
if muxer if muxer
.send_to_netwrok(MuxMessage { .send_to_netwrok(MuxMessage {
@@ -57,9 +55,8 @@ impl TcpConnection {
let first_payload = tokio::time::timeout(Duration::from_secs(10), v_rx.recv()).await; let first_payload = tokio::time::timeout(Duration::from_secs(10), v_rx.recv()).await;
match first_payload { match first_payload {
Ok(Some(data)) => { Ok(Some(data)) => {
// Успешный SOCKS-ответ сервера имеет код 0x00 во втором байте
if data.len() >= 2 && data[1] == 0x00 { if data.len() >= 2 && data[1] == 0x00 {
let _ = handshake_tx.send(()); // Даем отмашку smoltcp, что можно слать данные let _ = handshake_tx.send(());
} else { } else {
netrunner_logger::warn!(stream_id, "Server rejected TCP connection"); netrunner_logger::warn!(stream_id, "Server rejected TCP connection");
muxer.remove_stream(stream_id).await; muxer.remove_stream(stream_id).await;
@@ -73,12 +70,11 @@ impl TcpConnection {
} }
} }
// 4. Запускаем мост данных
let to_proxy = async { let to_proxy = async {
while let Some(data) = rx_from_smol.recv().await { while let Some(data) = rx_from_smol.recv().await {
let msg = MuxMessage { let msg = MuxMessage {
stream_id, stream_id,
frame_type: FrameType::Data, // Для TCP используем обычный Data frame_type: FrameType::Data,
data: Bytes::from(data), data: Bytes::from(data),
}; };
if muxer.send_to_netwrok(msg).await.is_err() { if muxer.send_to_netwrok(msg).await.is_err() {
@@ -91,7 +87,7 @@ impl TcpConnection {
while let Some(data) = v_rx.recv().await { while let Some(data) = v_rx.recv().await {
if data.is_empty() { if data.is_empty() {
break; break;
} // EOF }
if tx_to_smol.send(data.to_vec()).is_err() { if tx_to_smol.send(data.to_vec()).is_err() {
break; break;
} }
@@ -103,7 +99,6 @@ impl TcpConnection {
_ = from_proxy => {} _ = from_proxy => {}
} }
// 5. Корректно закрываем стрим на сервере
let _ = muxer let _ = muxer
.send_to_netwrok(MuxMessage { .send_to_netwrok(MuxMessage {
stream_id, stream_id,
@@ -116,7 +111,7 @@ impl TcpConnection {
Self { Self {
handle, handle,
state: ConnectionState::Handshaking, // Начинаем с ожидания handshake_tx state: ConnectionState::Handshaking,
tx: tx_to_mux, tx: tx_to_mux,
rx: rx_from_proxy, rx: rx_from_proxy,
pending_data: vec![], pending_data: vec![],
@@ -182,7 +177,6 @@ impl TcpConnection {
} }
fn poll_and_process(&mut self, socket: &mut tcp::Socket) { fn poll_and_process(&mut self, socket: &mut tcp::Socket) {
// 1. Читаем ИЗ виртуального сокета -> В прокси-сервер
if socket.can_recv() { if socket.can_recv() {
let mut total_read = 0; let mut total_read = 0;
let _ = socket.recv(|data| { let _ = socket.recv(|data| {
@@ -195,7 +189,6 @@ impl TcpConnection {
}); });
} }
// 2. Проверка лимита буфера перед тем, как брать новые данные
if self.pending_data.len() > MAX_PENDING { if self.pending_data.len() > MAX_PENDING {
netrunner_logger::error!(%self.handle, "TCP Buffer overflow ({} bytes). Dropping connection.", self.pending_data.len()); netrunner_logger::error!(%self.handle, "TCP Buffer overflow ({} bytes). Dropping connection.", self.pending_data.len());
socket.abort(); socket.abort();
@@ -203,7 +196,6 @@ impl TcpConnection {
return; return;
} }
// 3. Сбрасываем то, что накопилось в pending_data
if !self.pending_data.is_empty() && socket.can_send() { if !self.pending_data.is_empty() && socket.can_send() {
match socket.send_slice(&self.pending_data) { match socket.send_slice(&self.pending_data) {
Ok(n) => { Ok(n) => {
@@ -213,7 +205,6 @@ impl TcpConnection {
} }
} }
// 4. Читаем новые данные из прокси, ТОЛЬКО если старые ушли
if self.pending_data.is_empty() && socket.can_send() { if self.pending_data.is_empty() && socket.can_send() {
if let Ok(data) = self.rx.try_recv() { if let Ok(data) = self.rx.try_recv() {
match socket.send_slice(&data) { match socket.send_slice(&data) {
@@ -222,7 +213,6 @@ impl TcpConnection {
} }
Ok(_) => {} Ok(_) => {}
Err(_) => { Err(_) => {
// Если сокет внезапно отказал, сохраняем всё в буфер
self.pending_data = data; self.pending_data = data;
} }
} }
client/src/connections/udp_connection.rs
+5 -17
View File
@@ -14,15 +14,15 @@ use netrunner_core::{
pub struct UdpConnection { pub struct UdpConnection {
pub handle: SocketHandle, pub handle: SocketHandle,
stream_id: u32, stream_id: u32,
tx_to_net: mpsc::Sender<MuxMessage>, // Прямой канал кадров tx_to_net: mpsc::Sender<MuxMessage>,
rx_from_net: mpsc::Receiver<Bytes>, // Прямое чтение из муксера rx_from_net: mpsc::Receiver<Bytes>,
client_endpoint: Option<IpEndpoint>, client_endpoint: Option<IpEndpoint>,
last_activity: Instant, last_activity: Instant,
token: CancellationToken, token: CancellationToken,
} }
const UDP_TIMEOUT: Duration = Duration::from_secs(60); const UDP_TIMEOUT: Duration = Duration::from_secs(60);
// Ограничиваем очередь: для UDP дроп пакета при перегрузке — это норма
const CHANNEL_CAPACITY: usize = 1024; const CHANNEL_CAPACITY: usize = 1024;
impl UdpConnection { impl UdpConnection {
@@ -31,10 +31,8 @@ impl UdpConnection {
let token = CancellationToken::new(); let token = CancellationToken::new();
let task_token = token.clone(); let task_token = token.clone();
// Канал из tick в асинхронную таску (сразу в формате MuxMessage)
let (tx_to_net, mut rx_from_smol) = mpsc::channel::<MuxMessage>(CHANNEL_CAPACITY); let (tx_to_net, mut rx_from_smol) = mpsc::channel::<MuxMessage>(CHANNEL_CAPACITY);
// Канал из муксера напрямую в tick
let (v_tx, v_rx) = mpsc::channel::<Bytes>(CHANNEL_CAPACITY); let (v_tx, v_rx) = mpsc::channel::<Bytes>(CHANNEL_CAPACITY);
let m_clone = muxer.clone(); let m_clone = muxer.clone();
@@ -42,7 +40,6 @@ impl UdpConnection {
tokio::spawn(async move { tokio::spawn(async move {
m_clone.register_stream(stream_id, v_tx).await; m_clone.register_stream(stream_id, v_tx).await;
// 1. Устанавливаем соединение
let _ = m_clone let _ = m_clone
.send_to_netwrok(MuxMessage { .send_to_netwrok(MuxMessage {
stream_id, stream_id,
@@ -51,7 +48,6 @@ impl UdpConnection {
}) })
.await; .await;
// 2. Слушаем только один канал (на отправку в сеть)
let to_proxy = async { let to_proxy = async {
while let Some(msg) = rx_from_smol.recv().await { while let Some(msg) = rx_from_smol.recv().await {
if m_clone.send_to_netwrok(msg).await.is_err() { if m_clone.send_to_netwrok(msg).await.is_err() {
@@ -65,7 +61,6 @@ impl UdpConnection {
_ = task_token.cancelled() => {} _ = task_token.cancelled() => {}
} }
// 3. Закрываем стрим
let _ = m_clone let _ = m_clone
.send_to_netwrok(MuxMessage { .send_to_netwrok(MuxMessage {
stream_id, stream_id,
@@ -80,7 +75,7 @@ impl UdpConnection {
handle, handle,
stream_id, stream_id,
tx_to_net, tx_to_net,
rx_from_net: v_rx, // Сохраняем Receiver напрямую rx_from_net: v_rx,
client_endpoint: None, client_endpoint: None,
last_activity: Instant::now(), last_activity: Instant::now(),
token, token,
@@ -95,33 +90,27 @@ impl UdpConnection {
return false; return false;
} }
// 1. Читаем из TUN (приложение -> прокси)
if socket.can_recv() { if socket.can_recv() {
while let Ok((data, metadata)) = socket.recv() { while let Ok((data, metadata)) = socket.recv() {
self.client_endpoint = Some(metadata.endpoint); self.client_endpoint = Some(metadata.endpoint);
// Избегаем двойной аллокации, сразу копируем в Bytes
let msg = MuxMessage { let msg = MuxMessage {
stream_id: self.stream_id, stream_id: self.stream_id,
frame_type: FrameType::UdpData, frame_type: FrameType::UdpData,
data: Bytes::copy_from_slice(data), data: Bytes::copy_from_slice(data),
}; };
// try_send не блокирует цикл. Если буфер забит - пакет отбрасывается.
// QUIC мгновенно поймет потерю и адаптирует битрейт видео.
if self.tx_to_net.try_send(msg).is_ok() { if self.tx_to_net.try_send(msg).is_ok() {
self.last_activity = Instant::now(); self.last_activity = Instant::now();
} }
} }
} }
// 2. Пишем в TUN (прокси -> приложение)
if socket.can_send() { if socket.can_send() {
if let Some(endpoint) = self.client_endpoint { if let Some(endpoint) = self.client_endpoint {
// Читаем напрямую из канала муксера, минуя промежуточные таски
while let Ok(data) = self.rx_from_net.try_recv() { while let Ok(data) = self.rx_from_net.try_recv() {
if data.is_empty() { if data.is_empty() {
self.token.cancel(); // Сервер прислал сигнал закрытия self.token.cancel();
break; break;
} }
@@ -130,7 +119,6 @@ impl UdpConnection {
self.last_activity = Instant::now(); self.last_activity = Instant::now();
} }
Err(_) => { Err(_) => {
// Буфер smoltcp переполнен
break; break;
} }
} }
client/src/main.rs
+2 -4
View File
@@ -57,7 +57,7 @@ async fn main() -> anyhow::Result<()> {
Ok(m) => m, Ok(m) => m,
Err(e) => { Err(e) => {
error!("Failed to establish secure tunnel: {}", e); error!("Failed to establish secure tunnel: {}", e);
// Восстанавливаем роутинг перед выходом
let _ = reset_platform_routing(Some( let _ = reset_platform_routing(Some(
&remote_address.split(':').next().unwrap().to_string(), &remote_address.split(':').next().unwrap().to_string(),
)); ));
@@ -66,7 +66,6 @@ async fn main() -> anyhow::Result<()> {
}; };
info!("Secure tunnel established, Muxer is ready."); info!("Secure tunnel established, Muxer is ready.");
// Передаем Muxer в Engine
let mut engine = Engine::new(config, caps, dns_handler, muxer); let mut engine = Engine::new(config, caps, dns_handler, muxer);
engine.set_any_ip(true); engine.set_any_ip(true);
engine.set_transparent_mode(); engine.set_transparent_mode();
@@ -78,7 +77,7 @@ async fn main() -> anyhow::Result<()> {
let ctrl_c = tokio::signal::ctrl_c(); let ctrl_c = tokio::signal::ctrl_c();
tokio::select! { tokio::select! {
// Запускаем главный цикл обработки пакетов
res = engine.run(tun_device) => { res = engine.run(tun_device) => {
error!("Engine loop error: {:?}", res); error!("Engine loop error: {:?}", res);
}, },
@@ -87,7 +86,6 @@ async fn main() -> anyhow::Result<()> {
} }
} }
// Отменяем токен, что завершит фоновые задачи мультиплексора (TunnelEngine)
network_token.cancel(); network_token.cancel();
info!("Restoring system routing..."); info!("Restoring system routing...");
client/src/tun/connection_manager.rs
+3 -8
View File
@@ -154,7 +154,6 @@ impl ConnectionManager {
let local_port = socket.endpoint().port; let local_port = socket.endpoint().port;
// 1. ЛОКАЛЬНЫЙ DNS РЕЗОЛВЕР (FakeIP)
if local_port == 53 { if local_port == 53 {
while socket.can_recv() { while socket.can_recv() {
let (data, meta) = match socket.recv() { let (data, meta) = match socket.recv() {
@@ -162,17 +161,15 @@ impl ConnectionManager {
Err(_) => break, Err(_) => break,
}; };
// Передаем запрос в наш Fake DNS
if let Some(response) = self.dns_handler.handle_query(data, &mut self.fake_ip_store) if let Some(response) = self.dns_handler.handle_query(data, &mut self.fake_ip_store)
{ {
netrunner_logger::debug!(to = %meta.endpoint, "Sending DNS response (FakeIP/Filtered)"); netrunner_logger::debug!(to = %meta.endpoint, "Sending DNS response (FakeIP/Filtered)");
let _ = socket.send_slice(&response, meta); let _ = socket.send_slice(&response, meta);
} }
} }
return; // Дальше не идем, для DNS сессии не нужны return;
} }
// 2. ПРОКСИРУЕМЫЙ UDP (Игры, звонки и т.д.)
if socket.is_open() && !self.active_udp_sessions.contains_key(&handle) { if socket.is_open() && !self.active_udp_sessions.contains_key(&handle) {
let endpoint = socket.endpoint(); let endpoint = socket.endpoint();
let target = match endpoint.addr { let target = match endpoint.addr {
@@ -267,7 +264,6 @@ impl ConnectionManager {
let dst_port = udp_packet.dst_port(); let dst_port = udp_packet.dst_port();
let dst_addr = ip_packet.dst_addr(); let dst_addr = ip_packet.dst_addr();
// Игнорируем широковещательный мусор
if dst_port == 0 || dst_port == 137 || dst_port == 138 { if dst_port == 0 || dst_port == 137 || dst_port == 138 {
return; return;
} }
@@ -291,10 +287,9 @@ impl ConnectionManager {
} }
fn create_dynamic_udp_socket<'a>(port: u16) -> udp::Socket<'a> { fn create_dynamic_udp_socket<'a>(port: u16) -> udp::Socket<'a> {
// Для QUIC (443) выделяем жирный буфер, для остального - умеренный
let (buf_size, packet_count) = match port { let (buf_size, packet_count) = match port {
443 => (1024 * 1024, 512), // 1MB буфер, 512 метаданных для пакетов 443 => (1024 * 1024, 512),
_ => (128 * 1024, 128), // 128KB для обычного трафика _ => (128 * 1024, 128),
}; };
udp::Socket::new( udp::Socket::new(
core/src/proxy/connection/bridge.rs
+4 -4
View File
@@ -75,11 +75,11 @@ pub async fn run_udp_bridge(
muxer: Muxer, muxer: Muxer,
mut v_rx: mpsc::Receiver<Bytes>, mut v_rx: mpsc::Receiver<Bytes>,
) { ) {
let mut buf = [0u8; 65536]; // Максимальный размер UDP пакета let mut buf = [0u8; 65536];
loop { loop {
tokio::select! { tokio::select! {
// Читаем из реального интернета и шлем в туннель
res = socket.recv(&mut buf) => { res = socket.recv(&mut buf) => {
match res { match res {
Ok(0) => break, Ok(0) => break,
@@ -98,11 +98,11 @@ pub async fn run_udp_bridge(
} }
} }
// Читаем из туннеля и шлем в реальный интернет
maybe_data = v_rx.recv() => { maybe_data = v_rx.recv() => {
match maybe_data { match maybe_data {
Some(data) => { Some(data) => {
if data.is_empty() { break; } // Сигнал закрытия if data.is_empty() { break; }
if let Err(e) = socket.send(&data).await { if let Err(e) = socket.send(&data).await {
error!(stream_id, error = %e, "UDP socket write error"); error!(stream_id, error = %e, "UDP socket write error");
break; break;
core/src/proxy/connection/handler.rs
+2 -8
View File
@@ -28,9 +28,9 @@ impl StreamHandler {
match frame.header.frame_type { match frame.header.frame_type {
FrameType::Connect => self.on_connect(stream_id, frame.payload).await, FrameType::Connect => self.on_connect(stream_id, frame.payload).await,
FrameType::UdpConnect => self.on_udp_connect(stream_id, frame.payload).await, // НОВОЕ FrameType::UdpConnect => self.on_udp_connect(stream_id, frame.payload).await,
FrameType::Data => self.on_data(stream_id, frame.payload).await, FrameType::Data => self.on_data(stream_id, frame.payload).await,
FrameType::UdpData => self.on_udp_data(stream_id, frame.payload).await, // НОВОЕ FrameType::UdpData => self.on_udp_data(stream_id, frame.payload).await,
FrameType::Close => self.on_close(stream_id).await, FrameType::Close => self.on_close(stream_id).await,
_ => debug!(stream_id, "Unhandled frame type"), _ => debug!(stream_id, "Unhandled frame type"),
} }
@@ -90,7 +90,6 @@ impl StreamHandler {
} }
} }
// --- НОВЫЙ БЛОК ДЛЯ UDP ---
async fn on_udp_connect(&self, stream_id: u32, payload: Bytes) { async fn on_udp_connect(&self, stream_id: u32, payload: Bytes) {
if self.role == ConnectionRole::Server { if self.role == ConnectionRole::Server {
let target_str = String::from_utf8_lossy(&payload).to_string(); let target_str = String::from_utf8_lossy(&payload).to_string();
@@ -102,7 +101,6 @@ impl StreamHandler {
tokio::spawn(async move { tokio::spawn(async move {
info!(stream_id, target = %target_str, "Attempting remote UDP connection"); info!(stream_id, target = %target_str, "Attempting remote UDP connection");
// Для UDP мы просто биндим сокет на любой свободный порт и "коннектим" к цели
match tokio::net::UdpSocket::bind("0.0.0.0:0").await { match tokio::net::UdpSocket::bind("0.0.0.0:0").await {
Ok(socket) => { Ok(socket) => {
if let Err(e) = socket.connect(&target_str).await { if let Err(e) = socket.connect(&target_str).await {
@@ -125,7 +123,6 @@ impl StreamHandler {
.send_control(stream_id, FrameType::UdpConnect, reply_buf.freeze()) .send_control(stream_id, FrameType::UdpConnect, reply_buf.freeze())
.await; .await;
// Запускаем бридж для перекачки UDP датаграмм
run_udp_bridge(stream_id, socket, muxer, v_rx).await; run_udp_bridge(stream_id, socket, muxer, v_rx).await;
} }
Err(e) => { Err(e) => {
@@ -136,16 +133,13 @@ impl StreamHandler {
} }
}); });
} else { } else {
// Если это клиент, отдаем payload локальному обработчику
self.muxer.dispatch_to_local(stream_id, payload).await; self.muxer.dispatch_to_local(stream_id, payload).await;
} }
} }
async fn on_udp_data(&self, stream_id: u32, payload: Bytes) { async fn on_udp_data(&self, stream_id: u32, payload: Bytes) {
// Логика идентична TCP: пробрасываем датаграмму в локальный v_rx
self.muxer.dispatch_to_local(stream_id, payload).await; self.muxer.dispatch_to_local(stream_id, payload).await;
} }
// --------------------------
async fn send_error_reply(muxer: &Muxer, stream_id: u32, code: u8, frame_type: FrameType) { async fn send_error_reply(muxer: &Muxer, stream_id: u32, code: u8, frame_type: FrameType) {
muxer.remove_stream(stream_id).await; muxer.remove_stream(stream_id).await;