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

remove comments & rewrite client connection

Browse Source
This commit is contained in:
Kirill
2026-03-24 12:42:32 +07:00
parent cf9f225969
commit 9177b61a02
13 changed files with 166 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/connections/tcp_connection.rs
+1 -11
View File
@@ -109,10 +109,8 @@ impl TcpConnection {
while socket.can_recv() {
let mut full = false;
// Используем recv_slice для контроля размера чанка
let mut temp = [0u8; TCP_CHUNK_SIZE];
// Peek, чтобы проверить, сможем ли мы отправить данные, прежде чем извлечь их
if let Ok(n) = socket.peek_slice(&mut temp) {
if n == 0 {
break;
@@ -121,11 +119,10 @@ impl TcpConnection {
let chunk = Bytes::copy_from_slice(&temp[..n]);
match self.tx.try_send(chunk) {
Ok(_) => {
// Только если успешно отправили в канал, удаляем данные из сокета
socket.recv_slice(&mut temp[..n]).unwrap();
}
Err(mpsc::error::TrySendError::Full(_)) => {
full = true; // Канал забит, сработает Backpressure в smoltcp
full = true;
}
Err(_) => {
self.state = ConnectionState::Closed;
@@ -143,25 +140,21 @@ impl TcpConnection {
let current_pending = self.pending_data.len();
// Считаем % заполненности для логирования
let fill_ratio = (current_pending as f32 / MAX_PENDING as f32) * 100.0;
if current_pending >= MAX_PENDING {
// Состояние активного Backpressure
netrunner_logger::warn!(
%self.handle,
"Backpressure ACTIVE: Buffer is FULL ({} bytes). Stalling RX channel.",
current_pending
);
} else if fill_ratio > 80.0 {
// Состояние Bufferbloat (буфер почти полон, пакеты задерживаются)
netrunner_logger::info!(
%self.handle,
"Bufferbloat Warning: Buffer {:.1}% full ({} bytes). Latency increasing.",
fill_ratio, current_pending
);
// Продолжаем читать, пока есть хоть какое-то место
while let Ok(data) = self.rx.try_recv() {
self.pending_data.extend_from_slice(&data);
if self.pending_data.len() >= MAX_PENDING {
@@ -169,7 +162,6 @@ impl TcpConnection {
}
}
} else {
// Обычный режим
while let Ok(data) = self.rx.try_recv() {
self.pending_data.extend_from_slice(&data);
if self.pending_data.len() >= MAX_PENDING {
@@ -178,13 +170,11 @@ impl TcpConnection {
}
}
// 3. DOWNLOAD: Отправка накопленного буфера в smoltcp
if !self.pending_data.is_empty() && socket.can_send() {
match socket.send_slice(&self.pending_data) {
Ok(n) => {
self.pending_data.advance(n);
// Лог освобождения (опционально, чтобы видеть, что пробка рассасывается)
if n > 0 && self.pending_data.len() < (MAX_PENDING / 2) && fill_ratio > 90.0 {
netrunner_logger::info!(%self.handle, "Backpressure RELIEVED: Buffer drained to {} bytes", self.pending_data.len());
}
client/src/connections/udp_connection.rs
+2 -14
View File
@@ -9,9 +9,9 @@ use bytes::Bytes;
pub struct UdpConnection {
pub handle: SocketHandle,
// UPLOAD: Ограниченный канал для передачи данных наружу
tx: mpsc::Sender<Bytes>,
// DOWNLOAD: Канал для приема данных из сети
rx: mpsc::Receiver<Bytes>,
client_endpoint: Option<IpEndpoint>,
last_activity: Instant,
@@ -20,10 +20,6 @@ pub struct UdpConnection {
const UDP_TIMEOUT: Duration = Duration::from_secs(60);
impl UdpConnection {
/// Возвращает:
/// 1. Экземпляр UdpConnection
/// 2. Receiver (для чтения данных, отправляемых ИЗ smoltcp В сеть)
/// 3. Sender (для записи данных ИЗ сети В smoltcp)
pub fn new(handle: SocketHandle) -> (Self, mpsc::Receiver<Bytes>, mpsc::Sender<Bytes>) {
let (tx_to_net, rx_from_smol) = mpsc::channel::<Bytes>(CHANNEL_CAPACITY);
let (tx_to_smol, rx_from_net) = mpsc::channel::<Bytes>(CHANNEL_CAPACITY);
@@ -40,31 +36,25 @@ impl UdpConnection {
}
pub fn tick(&mut self, socket: &mut udp::Socket) -> bool {
// Проверка таймаута бездействия
if self.last_activity.elapsed() > UDP_TIMEOUT {
netrunner_logger::debug!(%self.handle, "UDP Session closed due to timeout");
socket.close();
return false;
}
// 1. UPLOAD: Читаем из smoltcp и отправляем в виртуальный канал
if socket.can_recv() {
while let Ok((data, metadata)) = socket.recv() {
self.client_endpoint = Some(metadata.endpoint);
// Копируем данные в Bytes и пытаемся протолкнуть.
// Если канал забит (Full), пакет дропается — это штатное поведение UDP.
if self.tx.try_send(Bytes::copy_from_slice(data)).is_ok() {
self.last_activity = Instant::now();
}
}
}
// 2. DOWNLOAD: Читаем из виртуального канала и пишем в smoltcp
if socket.can_send() {
if let Some(endpoint) = self.client_endpoint {
while let Ok(data) = self.rx.try_recv() {
// Сигнал закрытия стрима
if data.is_empty() {
socket.close();
return false;
@@ -75,8 +65,6 @@ impl UdpConnection {
self.last_activity = Instant::now();
}
Err(_) => {
// Если сокет smoltcp переполнен, прерываем цикл.
// Пакет теряется, что опять же является нормой для UDP.
break;
}
}
client/src/lib.rs
+4 -29
View File
@@ -1,35 +1,10 @@
uniffi::setup_scaffolding!();
pub mod connections;
pub mod session;
pub mod tun;
use netrunner_logger::info;
use std::sync::OnceLock;
use tokio::runtime::Runtime;
use tokio_util::sync::CancellationToken;
use crate::tun::routing::reset_platform_routing;
pub mod session;
static RUNTIME: OnceLock<Runtime> = OnceLock::new();
#[derive(uniffi::Object)]
pub struct Session {
pub(crate) cancel_token: CancellationToken,
pub(crate) proxy_ip: String,
}
#[uniffi::export]
impl Session {
pub fn stop(&self) {
info!("Stopping session...");
self.cancel_token.cancel();
let _ = reset_platform_routing(Some(&self.proxy_ip));
}
}
impl Drop for Session {
fn drop(&mut self) {
info!("Session dropped, stopping all tasks...");
self.cancel_token.cancel();
let _ = reset_platform_routing(Some(&self.proxy_ip));
}
}
pub static RUNTIME: OnceLock<Runtime> = OnceLock::new();
client/src/main.rs
+6 -17
View File
@@ -6,7 +6,7 @@ use netrunner_client::{
tun::Tun,
},
};
use netrunner_core::proxy::{connection::connection::ConnectionRole, network::Network};
use netrunner_core::proxy::connection::connection::ClientHandler;
use netrunner_logger::{error, info};
use smoltcp::{iface::Config, phy::DeviceCapabilities};
use std::net::Ipv4Addr;
@@ -42,26 +42,15 @@ async fn main() -> anyhow::Result<()> {
caps.max_transmission_unit = 1350;
caps.medium = smoltcp::phy::Medium::Ip;
let network = Network::new(
"127.0.0.1".into(),
8080,
ConnectionRole::Client,
Some(remote_address.clone()),
);
let network_token = CancellationToken::new();
let net_token_for_spawn = network_token.clone();
info!("Establishing secure tunnel to proxy server...");
let muxer = match network.initialize_client_tunnel(net_token_for_spawn).await {
let muxer = match ClientHandler::connect(&remote_address).await {
Ok(m) => m,
Err(e) => {
error!("Failed to establish secure tunnel: {}", e);
let _ = reset_platform_routing(Some(
&remote_address.split(':').next().unwrap().to_string(),
));
return Err(anyhow::anyhow!("Tunnel initialization failed: {}", e));
error!("Failed to establish secure tunnel to server: {}", e);
return Err(anyhow::anyhow!("Failed to establish secure tunnel: {}", e));
}
};
info!("Secure tunnel established, Muxer is ready.");
@@ -96,7 +85,7 @@ async fn main() -> anyhow::Result<()> {
error!("Failed to restore routing: {}", e);
} else {
info!("System routing restored successfully.");
}
};
Ok(())
}
client/src/session/mod.rs
+53 -63
View File
@@ -1,22 +1,18 @@
use crate::{
RUNTIME,
tun::{engine::EngineBuilder, routing::reset_platform_routing, tun::Tun},
};
use netrunner_logger::{error, info};
use std::sync::Arc;
use tokio::runtime::Runtime;
use tokio_util::sync::CancellationToken;
#[cfg(any(target_os = "linux", target_os = "windows"))]
pub mod desktop;
#[cfg(any(target_os = "android", target_os = "ios"))]
pub mod mobile;
use crate::{
RUNTIME, Session,
connections::dns::DnsHandler,
tun::{engine::Engine, routing::setup_platform_routing, tun::Tun},
};
use netrunner_core::proxy::{connection::connection::ConnectionRole, network::Network};
use netrunner_logger::{error, info};
use smoltcp::{iface::Config, phy::DeviceCapabilities};
use std::net::Ipv4Addr;
use std::sync::Arc;
use tokio::runtime::Runtime;
use tokio_util::sync::CancellationToken;
fn get_runtime() -> &'static Runtime {
RUNTIME.get_or_init(|| {
tokio::runtime::Builder::new_multi_thread()
@@ -26,6 +22,29 @@ fn get_runtime() -> &'static Runtime {
})
}
#[derive(uniffi::Object)]
pub struct Session {
pub(crate) cancel_token: CancellationToken,
pub(crate) proxy_ip: String,
}
#[uniffi::export]
impl Session {
pub fn stop(&self) {
info!("Stopping session...");
self.cancel_token.cancel();
let _ = reset_platform_routing(Some(&self.proxy_ip));
}
}
impl Drop for Session {
fn drop(&mut self) {
info!("Session dropped, stopping all tasks...");
self.cancel_token.cancel();
let _ = reset_platform_routing(Some(&self.proxy_ip));
}
}
#[derive(uniffi::Object)]
pub struct SessionManager;
@@ -48,8 +67,7 @@ impl SessionManager {
) -> Arc<Session> {
let runtime = get_runtime();
let cancel_token = CancellationToken::new();
let sesison_token = cancel_token.clone();
let net_token = cancel_token.clone();
let session_token = cancel_token.clone();
let addr: std::net::SocketAddr = remote_address.parse().expect("Invalid address format");
let remote_proxy_ip = addr.ip().to_string();
@@ -68,11 +86,6 @@ impl SessionManager {
}
};
let mut dns_handler = DnsHandler::new(&cache_path);
if let Err(e) = dns_handler.init().await {
error!("Failed to initialize DNS blocklist: {}", e);
}
let tun_device = {
#[cfg(any(target_os = "android", target_os = "ios"))]
{
@@ -100,55 +113,32 @@ impl SessionManager {
}
};
let _ = setup_platform_routing(&remote_address);
let builder_result = EngineBuilder::new(&remote_address)
.with_cache_path(&cache_path)
.with_tun(tun_device)
.build()
.await;
let config = Config::new(smoltcp::wire::HardwareAddress::Ip);
let mut caps = DeviceCapabilities::default();
caps.max_transmission_unit = 1350;
caps.medium = smoltcp::phy::Medium::Ip;
let network = Network::new(
"0.0.0.0".into(),
8080,
ConnectionRole::Client,
Some(remote_address.clone()),
);
let muxer = match network.initialize_client_tunnel(net_token).await {
Ok(m) => m,
Err(e) => {
error!("Failed to establish secure tunnel to server: {}", e);
return;
}
};
info!("Secure tunnel established, Muxer is ready.");
let mut engine = Engine::new(config, caps, dns_handler, muxer);
engine.set_any_ip(true);
engine.set_transparent_mode();
engine.set_default_gateway(Ipv4Addr::new(10, 0, 0, 2));
engine.activate();
let cancel_token_for_engine = cancel_token.clone();
tokio::spawn(async move {
info!("Engine async task started");
tokio::select! {
res = engine.run(tun_device) => {
info!("Engine loop finished: {:?}", res);
},
_ = cancel_token_for_engine.cancelled() => {
info!("Engine task shutting down via token");
match builder_result {
Ok((mut engine, tun)) => {
info!("Engine async task started");
tokio::select! {
res = engine.run(tun) => {
info!("Engine loop finished: {:?}", res);
},
_ = cancel_token.cancelled() => {
info!("Engine task shutting down via token");
}
}
}
});
Err(e) => {
error!("Failed to build VPN Engine: {}", e);
}
}
});
Arc::new(Session {
cancel_token: sesison_token,
cancel_token: session_token,
proxy_ip: remote_proxy_ip,
})
}
client/src/tun/connection_manager.rs
+3 -9
View File
@@ -101,10 +101,7 @@ impl ConnectionManager {
}
pub fn process_sockets(&mut self, socket_set: &mut SocketSet) {
// Добавим лог в начало цикла, если сессий много, чтобы видеть нагрузку
if !self.active_tcp_sessions.is_empty() || !self.active_udp_sessions.is_empty() {
// trace!("Processing sockets: TCP={}, UDP={}", self.active_tcp_sessions.len(), self.active_udp_sessions.len());
}
if !self.active_tcp_sessions.is_empty() || !self.active_udp_sessions.is_empty() {}
for (handle, socket) in socket_set.iter_mut() {
if let Some(tcp) = tcp::Socket::downcast_mut(socket) {
@@ -170,7 +167,6 @@ impl ConnectionManager {
return;
}
// Ждем подтверждения от прокси (handshake)
match tokio::time::timeout(Duration::from_secs(10), v_rx.recv()).await {
Ok(Some(data)) => {
if data.len() >= 2 && data[1] == 0x00 {
@@ -281,11 +277,9 @@ impl ConnectionManager {
netrunner_logger::info!(%handle, target = %target, "New UDP proxied session established");
// 1. Создаем UDP соединение и получаем каналы
let (conn, mut rx_from_smol, tx_to_smol) = UdpConnection::new(handle);
self.active_udp_sessions.insert(handle, conn);
// 2. Фоновая задача для Muxer'а
let muxer = self.muxer.clone();
let stream_id = muxer.next_id();
let connect_payload = target.to_string();
@@ -441,8 +435,8 @@ impl ConnectionManager {
fn create_dynamic_udp_socket<'a>(port: u16) -> udp::Socket<'a> {
let (buf_size, packet_count) = match port {
443 => (512 * 1024, 390), // Большой буфер для QUIC (YouTube)
53 => (64 * 1024, 32), // DNS
443 => (512 * 1024, 390),
53 => (64 * 1024, 32),
_ => (128 * 1024, 100),
};
client/src/tun/device.rs
+4 -14
View File
@@ -6,11 +6,11 @@ use std::{
Arc, LazyLock, Mutex,
atomic::{AtomicBool, Ordering},
},
time::Instant as StdInstant, // Добавлено для расчёта скорости
time::Instant as StdInstant,
};
use bytes::BytesMut;
use netrunner_logger::info; // Предполагается, что у тебя есть этот логгер
use netrunner_logger::info;
use smoltcp::{
phy::{self, Device, DeviceCapabilities},
time::Instant,
@@ -20,7 +20,6 @@ use tokio::sync::mpsc;
const TOKEN_BUFFER_LIST_MAX_SIZE: usize = 64;
static TOKEN_BUFFER_LIST: LazyLock<Mutex<Vec<BytesMut>>> = LazyLock::new(|| Mutex::new(Vec::new()));
// Структура, которую мы будем возвращать пользователю
#[derive(Debug, Clone, Copy)]
pub struct TrafficStats {
pub rx_bytes: u64,
@@ -79,20 +78,17 @@ pub struct VirtTunDevice {
tx_queue: mpsc::UnboundedSender<TokenBuffer>,
rx_avail: Arc<AtomicBool>,
// === Статистика трафика ===
rx_bytes: u64,
tx_bytes: u64,
rx_packets: u64,
tx_packets: u64,
// Внутренние переменные для расчёта скорости
last_speed_calc: StdInstant,
last_rx_bytes: u64,
last_tx_bytes: u64,
cached_rx_speed: f64,
cached_tx_speed: f64,
// Для периодического логирования
last_log_time: StdInstant,
}
@@ -147,7 +143,6 @@ impl VirtTunDevice {
let rx_diff = self.rx_bytes.saturating_sub(self.last_rx_bytes);
let tx_diff = self.tx_bytes.saturating_sub(self.last_tx_bytes);
// Переводим в МегаБайты в секунду (MB/s)
self.cached_rx_speed = (rx_diff as f64 / 1_048_576.0) / elapsed_speed;
self.cached_tx_speed = (tx_diff as f64 / 1_048_576.0) / elapsed_speed;
@@ -166,12 +161,11 @@ impl VirtTunDevice {
}
}
/// Внутренний метод проверки таймера для вывода логов
fn check_and_log_stats(&mut self) {
let now = StdInstant::now();
// Логируем каждые 5 секунд
if now.duration_since(self.last_log_time).as_secs() >= 5 {
let stats = self.get_stats(); // Заодно обновляем скорости
let stats = self.get_stats();
info!(
"TunDevice Traffic: RX: {:.2} MB ({} pkts) | TX: {:.2} MB ({} pkts) | Speed: ↓{:.2} MB/s, ↑{:.2} MB/s",
@@ -193,12 +187,9 @@ impl Device for VirtTunDevice {
type TxToken<'a> = VirtTxToken<'a>;
fn receive(&mut self, _timestamp: Instant) -> Option<(Self::RxToken<'_>, Self::TxToken<'_>)> {
// Проверяем, не пора ли записать стату в лог. receive() вызывается очень часто,
// поэтому это отличное место для heartbeat таймера.
self.check_and_log_stats();
if let Ok(buffer) = self.rx_queue.try_recv() {
// Учет входящего трафика (DOWNLOAD)
self.rx_bytes += buffer.len() as u64;
self.rx_packets += 1;
@@ -250,7 +241,6 @@ impl phy::TxToken for VirtTxToken<'_> {
let result = f(&mut buffer);
// Учет исходящего трафика (UPLOAD)
self.0.tx_bytes += len as u64;
self.0.tx_packets += 1;
client/src/tun/engine.rs
+66 -1
View File
@@ -1,3 +1,4 @@
use netrunner_core::proxy::connection::connection::ClientHandler;
use netrunner_core::proxy::connection::muxer::Muxer;
use smoltcp::iface::PollResult;
use smoltcp::time::Instant;
@@ -6,6 +7,7 @@ use smoltcp::{
iface::{Config, Interface, SocketSet},
phy::DeviceCapabilities,
};
use std::net::Ipv4Addr;
use std::sync::atomic::Ordering;
use std::{
sync::{Arc, LazyLock, atomic::AtomicBool},
@@ -16,11 +18,12 @@ use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
use tokio::time::{Duration, sleep};
use tun::{DeviceReader, DeviceWriter};
use netrunner_logger::{debug, info, warn};
use netrunner_logger::{debug, error, info, warn};
use crate::connections::dns::DnsHandler;
use crate::tun::connection_manager::ConnectionManager;
use crate::tun::device::{TokenBuffer, VirtTunDevice};
use crate::tun::routing::setup_platform_routing;
use crate::tun::tun::Tun;
pub static START_TIME: LazyLock<StdInstant> = LazyLock::new(StdInstant::now);
@@ -212,3 +215,65 @@ impl Engine {
self.manager.start_listening(&mut self.socket_set);
}
}
pub struct EngineBuilder {
remote_address: String,
cache_path: String,
tun_device: Option<Tun>,
}
impl EngineBuilder {
pub fn new(remote_address: impl Into<String>) -> Self {
Self {
remote_address: remote_address.into(),
cache_path: ".".to_string(),
tun_device: None,
}
}
pub fn with_cache_path(mut self, path: impl Into<String>) -> Self {
self.cache_path = path.into();
self
}
pub fn with_tun(mut self, tun: Tun) -> Self {
self.tun_device = Some(tun);
self
}
pub async fn build(self) -> Result<(Engine, Tun), String> {
let tun = self.tun_device.ok_or("TUN device is required")?;
info!("Initializing Engine components...");
let mut dns_handler = DnsHandler::new(&self.cache_path);
if let Err(e) = dns_handler.init().await {
error!("Failed to initialize DNS blocklist: {}", e);
}
if let Err(e) = setup_platform_routing(&self.remote_address) {
return Err(format!("Routing setup failed: {}", e));
}
let config = Config::new(smoltcp::wire::HardwareAddress::Ip);
let mut caps = DeviceCapabilities::default();
caps.max_transmission_unit = 1350;
caps.medium = smoltcp::phy::Medium::Ip;
info!("Establishing secure tunnel to proxy server...");
let muxer = ClientHandler::connect(&self.remote_address)
.await
.map_err(|e| format!("Failed to establish secure tunnel: {}", e))?;
info!("Secure tunnel established, Muxer is ready.");
let mut engine = Engine::new(config, caps, dns_handler, muxer);
engine.set_any_ip(true);
engine.set_transparent_mode();
engine.set_default_gateway(Ipv4Addr::new(10, 0, 0, 2));
engine.activate();
info!("Stack IP initialized: 10.0.0.2");
Ok((engine, tun))
}
}