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

buf sizes changes and socks5 remove to rawcast

Browse Source
This commit is contained in:
Kirill
2026-03-26 15:50:41 +07:00
parent c51c086fbc
commit b0dc3fc0a4
13 changed files with 641 additions and 945 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/lib.rs
+9 -38
View File
@@ -1,3 +1,4 @@
use netrunner_core::net::network::NetworkConfig;
use uniffi;
uniffi::setup_scaffolding!();
@@ -9,8 +10,6 @@ use crate::{
net::engine::{EngineBuilder, EngineConfig},
tun::routing::reset_platform_routing,
};
use netrunner_core::net::ConnectionRole;
use netrunner_core::net::network::Network;
use netrunner_logger::{error, info};
use std::sync::{Arc, OnceLock};
use tokio::runtime::Runtime;
@@ -88,40 +87,9 @@ impl SessionManager {
config = config.with_mtu(1350);
}
// --- 1. ЗАПУСК ЛОКАЛЬНОГО ПРОКСИ (NETWORK) ---
let local_proxy_port = 8080;
let local_proxy_host = "127.0.0.1".to_string();
let proxy_token = cancel_token.clone();
let remote_addr_clone = remote_address.clone();
NetworkConfig::init_global(config.mtu);
runtime.spawn(async move {
info!(
"Starting Local Proxy (Network) on {}:{}",
local_proxy_host, local_proxy_port
);
let network = Network::new(
local_proxy_host,
local_proxy_port,
ConnectionRole::Client,
Some(remote_addr_clone),
);
// Оборачиваем в tokio::select! для жесткой отмены
tokio::select! {
_ = network.run(proxy_token.clone()) => {
info!("Local Proxy (Network) task finished normally.");
}
_ = proxy_token.cancelled() => {
info!("Local Proxy (Network) task forcefully stopped via CancellationToken.");
}
}
});
// Даем прокси немного времени на бинд порта
std::thread::sleep(std::time::Duration::from_millis(100));
// --- 2. ЗАПУСК ENGINE И TUN ---
// --- ЗАПУСК ENGINE И TUN ---
let engine_token = cancel_token.clone();
runtime.spawn(async move {
info!("Starting VPN Engine thread...");
@@ -160,10 +128,13 @@ impl SessionManager {
match builder_result {
Ok((mut engine, tun)) => {
info!("Engine async task started");
info!("Engine built successfully, starting loop...");
// tokio::select! позволяет моментально прервать бесконечный цикл
// engine.run() при вызове session.stop()
tokio::select! {
res = engine.run(tun) => {
info!("Engine loop finished: {:?}", res);
_ = engine.run(tun) => {
info!("Engine loop finished normally.");
},
_ = engine_token.cancelled() => {
info!("Engine task shutting down via token");
client/src/main.rs
+8 -40
View File
@@ -6,9 +6,8 @@ mod tun;
use crate::tun::{routing::reset_platform_routing, tun::Tun};
use net::engine::{EngineBuilder, EngineConfig};
// Импортируем компоненты локального прокси
use netrunner_core::net::ConnectionRole;
use netrunner_core::net::network::Network;
// Импортируем глобальный конфиг сети
use netrunner_core::net::network::NetworkConfig;
use tokio_util::sync::CancellationToken;
#[tokio::main]
@@ -17,45 +16,18 @@ async fn main() -> anyhow::Result<()> {
info!("Initializing NetRunner Stack...");
let remote_address = "147.45.43.70:443".to_string();
let local_proxy_host = "127.0.0.1".to_string();
let local_proxy_port = 8080; // Локальный порт прокси
// Токен для управления жизненным циклом фоновых задач
let cancel_token = CancellationToken::new();
// ==================================================
// 1. ЗАПУСК ЛОКАЛЬНОГО ПРОКСИ (NETWORK)
// ==================================================
let proxy_token = cancel_token.clone();
let remote_addr_clone = remote_address.clone();
tokio::spawn(async move {
info!(
"Starting Local Proxy (Network) on {}:{}",
local_proxy_host, local_proxy_port
);
let network = Network::new(
local_proxy_host,
local_proxy_port,
ConnectionRole::Client,
Some(remote_addr_clone),
);
// Эта функция заблокирует поток, пока не сработает proxy_token
network.run(proxy_token).await;
info!("Local Proxy (Network) task stopped.");
});
// Даем локальному прокси немного времени на бинд порта и установку соединения
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
// ==================================================
// 2. ИНИЦИАЛИЗАЦИЯ ДВИЖКА И TUN
// 1. ИНИЦИАЛИЗАЦИЯ ДВИЖКА И TUN
// ==================================================
let config = EngineConfig::new(&remote_address)
.with_cache_path(".")
.with_mtu(1350);
// ВАЖНО: Инициализируем глобальные настройки сети (MTU, размеры буферов Muxer'а)
NetworkConfig::init_global(config.mtu);
let tun_device = Tun::create(|tun_cfg| {
tun_cfg
.tun_name("netr0")
@@ -75,7 +47,7 @@ async fn main() -> anyhow::Result<()> {
.await;
// ==================================================
// 3. ГЛАВНЫЙ ЦИКЛ ENGINE
// 2. ГЛАВНЫЙ ЦИКЛ ENGINE
// ==================================================
match builder_result {
Ok((mut engine, tun)) => {
@@ -87,7 +59,6 @@ async fn main() -> anyhow::Result<()> {
},
_ = tokio::signal::ctrl_c() => {
info!("Ctrl+C received, shutting down...");
// Отменяем токен, чтобы Network.run завершился
cancel_token.cancel();
}
}
@@ -99,7 +70,7 @@ async fn main() -> anyhow::Result<()> {
}
// ==================================================
// 4. ОЧИСТКА РОУТИНГА
// 3. ОЧИСТКА РОУТИНГА
// ==================================================
info!("Restoring system routing...");
let addr: std::net::SocketAddr = remote_address.parse().expect("Invalid address format");
@@ -111,8 +82,5 @@ async fn main() -> anyhow::Result<()> {
info!("System routing restored successfully.");
}
// Даем таске Network время на graceful shutdown (чтобы сокеты успели закрыться)
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
Ok(())
}
client/src/net/connection.rs
+56 -74
View File
@@ -29,9 +29,9 @@ impl ConnectionCore {
pub fn new(handle: SocketHandle) -> (Self, mpsc::Receiver<Bytes>, mpsc::Sender<Bytes>) {
trace!(%handle, "Creating ConnectionCore channels");
let (tx_to_net, rx_from_smol) =
mpsc::channel::<Bytes>(NetworkConfig::global().channel_capacity);
mpsc::channel::<Bytes>(NetworkConfig::global().stream_capacity);
let (tx_to_smol, rx_from_net) =
mpsc::channel::<Bytes>(NetworkConfig::global().channel_capacity);
mpsc::channel::<Bytes>(NetworkConfig::global().stream_capacity);
let core = Self {
handle,
@@ -42,7 +42,6 @@ impl ConnectionCore {
(core, rx_from_smol, tx_to_smol)
}
}
// ============================================================================
// 2. TCP СОЕДИНЕНИЕ (TcpConnection)
// ============================================================================
@@ -55,14 +54,13 @@ pub enum ConnectionState {
Closed,
}
const MAX_PENDING: usize = 64 * 1024;
const TCP_CHUNK_SIZE: usize = 1024 * 16;
pub struct TcpConnection {
core: ConnectionCore,
state: ConnectionState,
pending_data: BytesMut,
handshake_rx: Option<oneshot::Receiver<()>>,
chunk_buf: Vec<u8>,
server_eof: bool, // <--- ДОБАВЛЕН ФЛАГ ОКОНЧАНИЯ ПЕРЕДАЧИ
}
impl TcpConnection {
@@ -83,18 +81,14 @@ impl TcpConnection {
state: ConnectionState::Handshaking,
pending_data: BytesMut::new(),
handshake_rx: Some(handshake_rx),
chunk_buf: vec![0u8; NetworkConfig::global().tcp_chunk_size],
server_eof: false, // Инициализируем
};
(conn, rx_from_smol, tx_to_smol, handshake_tx)
}
pub fn is_finished(&self, socket: &tcp::Socket) -> bool {
matches!(socket.state(), tcp::State::Closed | tcp::State::TimeWait)
}
pub fn tick(&mut self, socket: &mut tcp::Socket) -> bool {
let state = socket.state();
match self.state {
ConnectionState::Handshaking => {
if let Some(rx) = &mut self.handshake_rx {
@@ -105,9 +99,8 @@ impl TcpConnection {
self.handshake_rx = None;
return true;
}
Err(oneshot::error::TryRecvError::Empty) => return true, // Ждем
Err(oneshot::error::TryRecvError::Empty) => return true,
Err(oneshot::error::TryRecvError::Closed) => {
debug!(%self.core.handle, "TCP Handshake channel dropped/aborted, State -> Closed");
self.state = ConnectionState::Closed;
return false;
}
@@ -120,19 +113,12 @@ impl TcpConnection {
ConnectionState::Active => {
self.poll_and_process(socket);
if state == tcp::State::CloseWait {
debug!(%self.core.handle, "TCP Socket reached CloseWait state, closing");
socket.close();
// Если сокет достиг финальных стадий, убиваем нашу сессию
if matches!(socket.state(), tcp::State::Closed | tcp::State::TimeWait) {
debug!(%self.core.handle, "TCP Socket is finished, state -> Closed");
self.state = ConnectionState::Closed;
return false;
}
if self.is_finished(socket) {
debug!(%self.core.handle, "TCP Socket is finished (Closed/TimeWait), state -> Closed");
self.state = ConnectionState::Closed;
socket.close();
return false;
}
}
ConnectionState::Closed => {
@@ -146,30 +132,29 @@ impl TcpConnection {
}
fn poll_and_process(&mut self, socket: &mut tcp::Socket) {
let max_pending = NetworkConfig::global().tcp_max_pending;
// 1. Вычитываем данные из smoltcp и шлем в Muxer
while socket.can_recv() {
let mut full = false;
let mut temp = [0u8; TCP_CHUNK_SIZE];
if let Ok(n) = socket.peek_slice(&mut temp) {
if let Ok(n) = socket.peek_slice(&mut self.chunk_buf) {
if n == 0 {
break;
}
let chunk = Bytes::copy_from_slice(&temp[..n]);
let chunk = Bytes::copy_from_slice(&self.chunk_buf[..n]);
match self.core.tx.try_send(chunk) {
Ok(_) => {
trace!(%self.core.handle, "Forwarded {} bytes from smoltcp to Muxer", n);
socket.recv_slice(&mut temp[..n]).unwrap();
socket.recv_slice(&mut self.chunk_buf[..n]).unwrap();
}
Err(mpsc::error::TrySendError::Full(_)) => {
debug!(%self.core.handle, "Muxer TX channel full, backpressure applied to smoltcp read");
full = true;
}
Err(_) => {
debug!(%self.core.handle, "Muxer TX channel closed unexpectedly, state -> Closed");
self.state = ConnectionState::Closed;
return;
// Канал Muxer'а закрыт
self.server_eof = true;
break;
}
}
} else {
@@ -181,60 +166,57 @@ impl TcpConnection {
}
}
// 2. Читаем данные из Muxer'а с учетом Backpressure
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 {
warn!(
%self.core.handle,
"Backpressure ACTIVE: Buffer is FULL ({} bytes). Stalling RX channel.",
current_pending
);
} else if fill_ratio > 80.0 {
info!(
%self.core.handle,
"Bufferbloat Warning: Buffer {:.1}% full ({} bytes). Latency increasing.",
fill_ratio, current_pending
);
while let Ok(data) = self.core.rx.try_recv() {
trace!(%self.core.handle, "Received {} bytes from Muxer (high buffer)", data.len());
self.pending_data.extend_from_slice(&data);
if self.pending_data.len() >= MAX_PENDING {
break;
}
}
} else {
while let Ok(data) = self.core.rx.try_recv() {
trace!(%self.core.handle, "Received {} bytes from Muxer", data.len());
self.pending_data.extend_from_slice(&data);
if self.pending_data.len() >= MAX_PENDING {
break;
// 2. Читаем данные из Muxer'а
if !self.server_eof {
loop {
match self.core.rx.try_recv() {
Ok(data) => {
self.pending_data.extend_from_slice(&data);
if self.pending_data.len() >= max_pending {
break;
}
}
Err(mpsc::error::TryRecvError::Empty) => {
break;
}
Err(mpsc::error::TryRecvError::Disconnected) => {
// ВАЖНО: Удаленный сервер прислал EOF!
debug!(%self.core.handle, "Server sent EOF (channel disconnected).");
self.server_eof = true;
break;
}
}
}
}
// 3. Отправляем буферизированные данные в smoltcp
// 3. Пишем данные браузеру
if !self.pending_data.is_empty() && socket.can_send() {
match socket.send_slice(&self.pending_data) {
Ok(n) => {
trace!(%self.core.handle, "Wrote {} bytes from buffer to smoltcp", n);
self.pending_data.advance(n);
if n > 0 && self.pending_data.len() < (MAX_PENDING / 2) && fill_ratio > 90.0 {
info!(
%self.core.handle,
"Backpressure RELIEVED: Buffer drained to {} bytes",
self.pending_data.len()
);
}
}
Err(e) => {
debug!(%self.core.handle, "Smoltcp socket send error: {:?}", e);
debug!(%self.core.handle, "Smoltcp send error: {:?}", e);
}
}
}
// 4. ГРАЦИОЗНОЕ ЗАКРЫТИЕ (Отправка FIN браузеру)
// Сценарий А: Сервер закрыл соединение, и мы отдали все остатки данных браузеру
if self.server_eof && self.pending_data.is_empty() && socket.may_send() {
debug!(%self.core.handle, "All data flushed after server EOF, sending FIN to browser");
socket.close();
}
// Сценарий Б: Браузер сам инициировал закрытие (CloseWait), но мы дожидаемся опустошения буфера
if socket.state() == tcp::State::CloseWait
&& self.pending_data.is_empty()
&& socket.may_send()
{
debug!(%self.core.handle, "Browser in CloseWait and buffer flushed, sending FIN");
socket.close();
}
}
}
client/src/net/connection_manager.rs
+183 -167
View File
@@ -1,14 +1,16 @@
use bytes::Bytes;
use netrunner_core::nrxp::TargetAddress;
use netrunner_logger::{debug, error, info, trace, warn};
use netrunner_core::{
net::network::NetworkConfig,
rawcast::{LocalProtocol, RawCastEvent, RawCastFrame},
};
use netrunner_logger::{debug, error, info, trace};
use smoltcp::{
iface::{SocketHandle, SocketSet},
socket::{AnySocket, icmp, tcp, udp},
wire::{IpListenEndpoint, IpProtocol, Ipv4Packet, TcpPacket, UdpPacket},
};
use std::{collections::HashMap, time::Instant as StdInstant};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpStream, UdpSocket};
use tokio::sync::mpsc;
use crate::net::{
connection::{TcpConnection, UdpConnection},
@@ -19,13 +21,15 @@ use crate::net::{
// ============================================================================
// 1. УПРАВЛЕНИЕ СОСТОЯНИЕМ СЕССИЙ (SessionTracker)
// ============================================================================
struct SessionTracker {
last_activity: HashMap<SocketHandle, StdInstant>,
active_tcp: HashMap<SocketHandle, TcpConnection>,
active_udp: HashMap<SocketHandle, UdpConnection>,
inbound_tx: HashMap<u64, mpsc::Sender<Bytes>>,
failed_until: HashMap<SocketHandle, StdInstant>,
to_remove: Vec<SocketHandle>,
next_socket_id: u64,
handle_to_id: HashMap<SocketHandle, u64>,
}
impl SessionTracker {
@@ -34,11 +38,20 @@ impl SessionTracker {
last_activity: HashMap::new(),
active_tcp: HashMap::new(),
active_udp: HashMap::new(),
inbound_tx: HashMap::new(),
failed_until: HashMap::new(),
to_remove: Vec::new(),
next_socket_id: 1,
handle_to_id: HashMap::new(),
}
}
fn generate_socket_id(&mut self) -> u64 {
let id = self.next_socket_id;
self.next_socket_id = self.next_socket_id.wrapping_add(1);
id
}
fn queue_removal(&mut self, handle: SocketHandle) {
if !self.to_remove.contains(&handle) {
self.to_remove.push(handle);
@@ -53,19 +66,28 @@ impl SessionTracker {
self.failed_until.remove(&handle);
self.active_tcp.remove(&handle);
self.active_udp.remove(&handle);
// ВОТ ТУТ ГЛАВНОЕ ИСПРАВЛЕНИЕ:
// Достаем НАШ socket_id, который мы выдали этому handle при старте сессии
if let Some(socket_id) = self.handle_to_id.remove(&handle) {
self.inbound_tx.remove(&socket_id);
}
}
}
fn has_tcp(
fn has_connection_from(
&self,
dst_addr: smoltcp::wire::IpAddress,
dst_port: u16,
src_addr: smoltcp::wire::IpAddress,
src_port: u16,
socket_set: &SocketSet,
) -> bool {
socket_set.iter().any(|(_, s)| {
if let Some(tcp) = tcp::Socket::downcast(s) {
if let Some(ep) = tcp.local_endpoint() {
return ep.addr == dst_addr && ep.port == dst_port;
// Если сокет уже привязался к клиенту, проверяем его исходные данные
if let Some(remote) = tcp.remote_endpoint() {
if remote.addr == src_addr && remote.port == src_port {
return true;
}
}
}
false
@@ -105,30 +127,6 @@ impl TargetResolver {
}
}
fn resolve_tcp(&self, socket: &tcp::Socket) -> TargetAddress {
let ep = match socket.local_endpoint() {
Some(ep) => ep,
None => {
warn!(handle=?socket, "Target resolution failed: no local endpoint");
return TargetAddress::Domain("disconnected".to_string(), 0);
}
};
match ep.addr {
smoltcp::wire::IpAddress::Ipv4(ip) => {
let std_ip = std::net::Ipv4Addr::from(ip);
if let Some(domain) = self.fake_ip_store.lookup_by_ip(&std_ip) {
TargetAddress::Domain(domain, ep.port)
} else {
TargetAddress::Ipv4(std_ip, ep.port)
}
}
smoltcp::wire::IpAddress::Ipv6(ip) => {
TargetAddress::Ipv6(std::net::Ipv6Addr::from(ip), ep.port)
}
}
}
fn process_dns_query(&mut self, data: &[u8]) -> Option<Vec<u8>> {
self.dns_handler.handle_query(data, &mut self.fake_ip_store)
}
@@ -137,17 +135,21 @@ impl TargetResolver {
// ============================================================================
// 3. ФАБРИКА СОКЕТОВ (SocketFactory)
// ============================================================================
struct SocketFactory;
impl SocketFactory {
fn create_tcp<'a>(port: u16) -> tcp::Socket<'a> {
let max_buf = NetworkConfig::global().smoltcp_socket_buf;
// Для Web-трафика используем максимум из конфига (например, 2 МБ)
// Для остальных урезаем в 4 раза, чтобы сэкономить RAM на фоновых соединениях
let buf_size = match port {
443 | 80 => 1024 * 1024 * 2,
443 | 80 | 8080 => max_buf,
22 => 32 * 1024,
53 => 16 * 1024,
_ => 128 * 1024,
_ => max_buf / 4,
};
let mut socket = tcp::Socket::new(
tcp::SocketBuffer::new(vec![0; buf_size]),
tcp::SocketBuffer::new(vec![0; buf_size]),
@@ -158,11 +160,20 @@ impl SocketFactory {
}
fn create_udp<'a>(port: u16) -> udp::Socket<'a> {
let config = NetworkConfig::global();
let max_buf = config.udp_buffer_size;
let payload_size = config.safe_payload_size.max(1); // Защита от деления на 0
// Вычисляем размер буфера и количество пакетов
let (buf_size, packet_count) = match port {
443 => (512 * 1024, 390),
53 => (64 * 1024, 32),
_ => (128 * 1024, 100),
443 => (max_buf, max_buf / payload_size), // QUIC/HTTP3 трафик
53 => (64 * 1024, (64 * 1024) / payload_size), // DNS
_ => (max_buf / 2, (max_buf / 2) / payload_size),
};
// Гарантируем, что метаданных хватит хотя бы на 10 пакетов
let packet_count = packet_count.max(10);
udp::Socket::new(
udp::PacketBuffer::new(
vec![udp::PacketMetadata::EMPTY; packet_count],
@@ -186,20 +197,45 @@ impl SocketFactory {
// ============================================================================
// 4. ГЛАВНЫЙ КООРДИНАТОР (ConnectionManager)
// ============================================================================
pub struct ConnectionManager {
tracker: SessionTracker,
resolver: TargetResolver,
tx_to_tunnel: mpsc::Sender<RawCastFrame>,
}
impl ConnectionManager {
pub fn new(dns_handler: DnsHandler) -> Self {
pub fn new(dns_handler: DnsHandler, tx_to_tunnel: mpsc::Sender<RawCastFrame>) -> Self {
Self {
tracker: SessionTracker::new(),
resolver: TargetResolver::new(dns_handler),
tx_to_tunnel,
}
}
// ВАЖНО: Метод для получения данных из сети (от VPN) и инъекции их в smoltcp
// Пытается внедрить пакет. Если канал переполнен — возвращает пакет обратно!
pub fn try_inject_inbound(&mut self, frame: RawCastFrame) -> Result<(), RawCastFrame> {
if frame.event != RawCastEvent::Data {
if frame.event == RawCastEvent::Close {
self.tracker.inbound_tx.remove(&frame.socket_id);
}
return Ok(());
}
if let Some(tx) = self.tracker.inbound_tx.get(&frame.socket_id) {
match tx.try_send(frame.payload.clone()) {
Ok(_) => Ok(()),
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
Err(frame) // Возвращаем кадр, чтобы Engine затормозил чтение туннеля
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
Ok(()) // Сокет уже закрыт браузером, дропаем пакет
}
}
} else {
Ok(())
}
}
pub fn setup_sockets(n_icmp: usize) -> SocketSet<'static> {
let mut sockets = SocketSet::new(Vec::with_capacity(48));
for _ in 0..n_icmp {
@@ -238,13 +274,23 @@ impl ConnectionManager {
match ip_packet.next_header() {
IpProtocol::Tcp => {
if let Ok(tcp_packet) = TcpPacket::new_checked(ip_packet.payload()) {
if tcp_packet.syn() && !tcp_packet.ack() {
let dst_port = tcp_packet.dst_port();
trace!(%dst_addr, dst_port, "Received TCP SYN");
let src_addr = ip_packet.src_addr();
let dst_addr = ip_packet.dst_addr();
if !self.tracker.has_tcp(dst_addr.into(), dst_port, socket_set) {
debug!(%dst_addr, dst_port, "No active TCP socket found, allocating new one");
if let Ok(tcp_packet) = TcpPacket::new_checked(ip_packet.payload()) {
// 2. Проверяем, что это пакет инициализации соединения (SYN)
if tcp_packet.syn() && !tcp_packet.ack() {
let src_port = tcp_packet.src_port();
let dst_port = tcp_packet.dst_port();
trace!(%dst_addr, dst_port, src_port, "Received TCP SYN");
// 3. Теперь src_addr доступен в этой области видимости
if !self
.tracker
.has_connection_from(src_addr.into(), src_port, socket_set)
{
debug!(%dst_addr, dst_port, src_port, "Allocating new TCP socket");
let mut socket = SocketFactory::create_tcp(dst_port);
let endpoint = IpListenEndpoint {
@@ -254,19 +300,16 @@ impl ConnectionManager {
match socket.listen(endpoint) {
Ok(_) => {
debug!(%dst_addr, dst_port, "TCP socket successfully listening");
socket_set.add(socket);
}
Err(e) => {
error!(%dst_addr, dst_port, "Failed to listen on TCP socket: {:?}", e);
error!(%dst_addr, dst_port, "Failed to listen: {:?}", e);
}
}
} else {
trace!(%dst_addr, dst_port, "TCP socket already exists, ignoring SYN");
trace!(%dst_addr, dst_port, src_port, "Socket already exists, ignoring SYN");
}
}
} else {
trace!("try_create_socket: Failed to parse TCP payload");
}
}
IpProtocol::Udp => {
@@ -324,7 +367,6 @@ impl ConnectionManager {
}
}
}
fn handle_tcp(&mut self, handle: SocketHandle, socket: &mut tcp::Socket) {
if socket.state() == tcp::State::Closed {
self.tracker.active_tcp.remove(&handle);
@@ -335,67 +377,63 @@ impl ConnectionManager {
if socket.state() == tcp::State::Established
&& !self.tracker.active_tcp.contains_key(&handle)
{
let target = self.resolver.resolve_tcp(socket);
if let TargetAddress::Domain(ref d, _) = target {
if d == "disconnected" {
socket.abort();
return;
}
}
info!(%handle, target = %target, "New TCP session established");
let (conn, mut rx_from_smol, tx_to_smol, handshake_tx) = TcpConnection::new(handle);
self.tracker.active_tcp.insert(handle, conn);
// Конвертируем TargetAddress в строку, понятную для tokio::net
let target_str = match target {
TargetAddress::Domain(d, p) => format!("{}:{}", d, p),
TargetAddress::Ipv4(ip, p) => format!("{}:{}", ip, p),
TargetAddress::Ipv6(ip, p) => format!("{}:{}", ip, p),
let (dst_ip, dst_port) = match socket.local_endpoint() {
Some(ep) => match ep.addr {
smoltcp::wire::IpAddress::Ipv4(ip) => (std::net::Ipv4Addr::from(ip), ep.port),
_ => return,
},
None => return,
};
tokio::spawn(async move {
let mut upstream = match TcpStream::connect(&target_str).await {
Ok(s) => s,
Err(e) => {
error!("Failed to connect to upstream TCP {}: {}", target_str, e);
return;
}
};
info!(%handle, ip = %dst_ip, port = dst_port, "New TCP session intercepted");
let socket_id = self.tracker.generate_socket_id();
self.tracker.handle_to_id.insert(handle, socket_id);
let (conn, mut rx_from_smol, tx_to_smol, handshake_tx) = TcpConnection::new(handle);
self.tracker.active_tcp.insert(handle, conn);
self.tracker.inbound_tx.insert(socket_id, tx_to_smol);
// ИСПРАВЛЕНИЕ ЗДЕСЬ: Переводим IP обратно в домен
let target_str = if let Some(domain) = self.resolver.fake_ip_store.lookup_by_ip(&dst_ip)
{
format!("{}:{}", domain, dst_port)
} else {
format!("{}:{}", dst_ip, dst_port)
};
let tx_tunnel = self.tx_to_tunnel.clone();
tokio::spawn(async move {
// Создаем кадр коннекта
let mut connect_frame =
RawCastFrame::connect(LocalProtocol::Tcp, socket_id, dst_ip, dst_port);
// Кладем доменное имя в payload, чтобы ClientHandler его прочитал
connect_frame.payload = bytes::Bytes::from(target_str);
if tx_tunnel.send(connect_frame).await.is_err() {
return;
}
// Сообщаем соединению smoltcp, что мы готовы (рукопожатие выполнено)
let _ = handshake_tx.send(());
let (mut r, mut w) = upstream.into_split();
// Читаем из tun (smoltcp) и пишем во внешнюю сеть
let to_upstream = async {
while let Some(data) = rx_from_smol.recv().await {
if w.write_all(&data).await.is_err() {
break;
}
while let Some(data) = rx_from_smol.recv().await {
let data_frame = RawCastFrame::data(
LocalProtocol::Tcp,
socket_id,
dst_ip,
dst_port,
data.to_vec(),
);
if tx_tunnel.send(data_frame).await.is_err() {
break;
}
};
}
// Читаем из внешней сети и пишем в tun (smoltcp)
let from_upstream = async {
let mut buf = vec![0u8; 8192]; // Читаем чанками
while let Ok(n) = r.read(&mut buf).await {
if n == 0 {
break;
} // EOF
if tx_to_smol
.send(Bytes::copy_from_slice(&buf[..n]))
.await
.is_err()
{
break;
}
}
};
tokio::select! { _ = to_upstream => {}, _ = from_upstream => {} }
let close_frame =
RawCastFrame::close(LocalProtocol::Tcp, socket_id, dst_ip, dst_port);
let _ = tx_tunnel.send(close_frame).await;
});
}
@@ -404,12 +442,7 @@ impl ConnectionManager {
socket.abort();
}
}
if socket.state() == tcp::State::CloseWait {
socket.close();
}
}
fn handle_udp(&mut self, handle: SocketHandle, socket: &mut udp::Socket) {
self.tracker.last_activity.insert(handle, StdInstant::now());
@@ -427,71 +460,54 @@ impl ConnectionManager {
}
if socket.is_open() && !self.tracker.active_udp.contains_key(&handle) {
let ep = socket.endpoint();
let target = match ep.addr {
let (dst_ip, dst_port) = match socket.endpoint().addr {
Some(smoltcp::wire::IpAddress::Ipv4(ip)) => {
TargetAddress::Ipv4(std::net::Ipv4Addr::from(ip), ep.port)
(std::net::Ipv4Addr::from(ip), socket.endpoint().port)
}
Some(smoltcp::wire::IpAddress::Ipv6(ip)) => {
TargetAddress::Ipv6(std::net::Ipv6Addr::from(ip), ep.port)
}
None => return,
_ => return,
};
// 1. Генерируем уникальный ID
let socket_id = self.tracker.generate_socket_id();
// 2. Сохраняем привязку для будущего cleanup
self.tracker.handle_to_id.insert(handle, socket_id);
let (conn, mut rx_from_smol, tx_to_smol) = UdpConnection::new(handle);
self.tracker.active_udp.insert(handle, conn);
// Конвертируем для tokio::net::UdpSocket
let target_str = match target {
TargetAddress::Domain(d, p) => format!("{}:{}", d, p),
TargetAddress::Ipv4(ip, p) => format!("{}:{}", ip, p),
TargetAddress::Ipv6(ip, p) => format!("{}:{}", ip, p),
self.tracker.active_udp.insert(handle, conn);
self.tracker.inbound_tx.insert(socket_id, tx_to_smol);
let target_str = if let Some(domain) = self.resolver.fake_ip_store.lookup_by_ip(&dst_ip)
{
format!("{}:{}", domain, dst_port)
} else {
format!("{}:{}", dst_ip, dst_port)
};
tokio::spawn(async move {
// Создаем локальный UDP сокет со случайным портом
let upstream = match UdpSocket::bind("0.0.0.0:0").await {
Ok(s) => s,
Err(e) => {
error!("Failed to bind local UDP socket: {}", e);
return;
}
};
let tx_tunnel = self.tx_to_tunnel.clone();
// "Подключаем" UDP сокет к цели (включает фильтр пакетов и позволяет использовать обычные send/recv)
if let Err(e) = upstream.connect(&target_str).await {
error!("Failed to connect UDP to {}: {}", target_str, e);
return;
tokio::spawn(async move {
let mut connect_frame =
RawCastFrame::connect(LocalProtocol::Udp, socket_id, dst_ip, dst_port);
connect_frame.payload = bytes::Bytes::from(target_str); // Кладем домен
let _ = tx_tunnel.send(connect_frame).await;
while let Some(data) = rx_from_smol.recv().await {
let data_frame = RawCastFrame::data(
LocalProtocol::Udp,
socket_id,
dst_ip,
dst_port,
data.to_vec(),
);
if tx_tunnel.send(data_frame).await.is_err() {
break;
}
}
let upstream = std::sync::Arc::new(upstream);
let upstream_rx = upstream.clone();
let upstream_tx = upstream;
// Из smoltcp наружу
let to_upstream = async {
while let Some(data) = rx_from_smol.recv().await {
if upstream_tx.send(&data).await.is_err() {
break;
}
}
};
// Извне в smoltcp
let from_upstream = async {
let mut buf = vec![0u8; 65536]; // Максимальный размер UDP датаграммы
while let Ok(n) = upstream_rx.recv(&mut buf).await {
if tx_to_smol
.send(Bytes::copy_from_slice(&buf[..n]))
.await
.is_err()
{
break;
}
}
};
tokio::select! { _ = to_upstream => {}, _ = from_upstream => {} }
let close_frame =
RawCastFrame::close(LocalProtocol::Udp, socket_id, dst_ip, dst_port);
let _ = tx_tunnel.send(close_frame).await;
});
}
client/src/net/engine.rs
+96 -26
View File
@@ -1,4 +1,6 @@
use netrunner_core::net::ClientHandler;
use netrunner_core::net::network::NetworkConfig;
use netrunner_core::rawcast::RawCastFrame;
use smoltcp::iface::PollResult;
use smoltcp::time::Instant;
use smoltcp::wire::{IpAddress, IpCidr};
@@ -35,17 +37,25 @@ pub struct Engine {
to_smoltcp_tx: UnboundedSender<TokenBuffer>,
from_smoltcp_rx: Option<UnboundedReceiver<TokenBuffer>>,
avail: Arc<AtomicBool>,
rx_from_tunnel: mpsc::Receiver<RawCastFrame>,
}
impl Engine {
pub fn new(config: Config, caps: DeviceCapabilities, dns_handler: DnsHandler) -> Self {
pub fn new(
config: Config,
caps: DeviceCapabilities,
dns_handler: DnsHandler,
tx_to_tunnel: mpsc::Sender<RawCastFrame>, // Куда менеджер будет слать пакеты
rx_from_tunnel: mpsc::Receiver<RawCastFrame>, // Откуда движок будет читать пакеты
) -> Self {
let now = Engine::current_time();
let (mut device, to_smoltcp_tx, from_smoltcp_rx, avail) = VirtTunDevice::new(caps);
let interface = Interface::new(config, &mut device, now);
let socket_set = ConnectionManager::setup_sockets(2);
let manager = ConnectionManager::new(dns_handler);
// Передаем TX-канал в ConnectionManager
let manager = ConnectionManager::new(dns_handler, tx_to_tunnel);
Self {
interface,
@@ -55,6 +65,7 @@ impl Engine {
from_smoltcp_rx: Some(from_smoltcp_rx),
avail,
manager,
rx_from_tunnel,
}
}
@@ -69,17 +80,36 @@ impl Engine {
Self::spawn_tun_writer(writer, from_smoltcp_rx);
let mut last_log = StdInstant::now();
let mut stuck_frame: Option<RawCastFrame> = None;
loop {
// 1. Быстро читаем из TUN
while let Ok(token) = tun_to_engine_rx.try_recv() {
self.manager
.try_create_socket_from_packet(&token, &mut self.socket_set);
if self.to_smoltcp_tx.send(token).is_ok() {
self.device.mark_rx_available();
}
}
// 2. Пытаемся протолкнуть застрявший кадр
if let Some(frame) = stuck_frame.take() {
if let Err(returned_frame) = self.manager.try_inject_inbound(frame) {
stuck_frame = Some(returned_frame);
}
}
// 3. Если затора нет, читаем новые кадры из VPN-туннеля
if stuck_frame.is_none() {
while let Ok(frame) = self.rx_from_tunnel.try_recv() {
if let Err(returned_frame) = self.manager.try_inject_inbound(frame) {
stuck_frame = Some(returned_frame);
break; // СТОП! Локальный сокет переполнен, давим на тормоза!
}
}
}
// 4. Двигаем стейт-машину smoltcp
let result = self.poll();
self.manager.process_sockets(&mut self.socket_set);
@@ -89,16 +119,48 @@ impl Engine {
}
if matches!(result, PollResult::SocketStateChanged) {
continue;
}
if self.avail.swap(false, Ordering::Acquire) {
tokio::task::yield_now().await;
self.manager.cleanup(&mut self.socket_set);
continue;
}
self.manager.cleanup(&mut self.socket_set);
self.poll_delay().await;
let delay = self
.interface
.poll_delay(Self::current_time(), &self.socket_set);
let sleep_duration = match delay {
Some(d) => Duration::from_micros(d.micros()),
None => Duration::from_millis(10),
};
// 5. Умный select!
if stuck_frame.is_none() {
tokio::select! {
_ = sleep(sleep_duration) => {}
Some(token) = tun_to_engine_rx.recv() => {
self.manager.try_create_socket_from_packet(&token, &mut self.socket_set);
if self.to_smoltcp_tx.send(token).is_ok() {
self.device.mark_rx_available();
}
}
Some(frame) = self.rx_from_tunnel.recv() => {
if let Err(returned_frame) = self.manager.try_inject_inbound(frame) {
stuck_frame = Some(returned_frame);
}
}
}
} else {
// Если буфер переполнен, ждем только таймер или данные от браузера
tokio::select! {
_ = sleep(sleep_duration) => {}
Some(token) = tun_to_engine_rx.recv() => {
self.manager.try_create_socket_from_packet(&token, &mut self.socket_set);
if self.to_smoltcp_tx.send(token).is_ok() {
self.device.mark_rx_available();
}
}
}
}
}
}
@@ -108,16 +170,6 @@ impl Engine {
.poll(now, &mut self.device, &mut self.socket_set)
}
async fn poll_delay(&mut self) {
let timestamp = Self::current_time();
let delay = self.interface.poll_delay(timestamp, &self.socket_set);
let sleep_duration = match delay {
Some(d) => Duration::from_micros(d.micros()),
None => Duration::from_millis(10),
};
sleep(sleep_duration).await;
}
fn spawn_tun_reader(
mut reader: DeviceReader,
to_engine: UnboundedSender<TokenBuffer>,
@@ -305,18 +357,36 @@ impl EngineBuilder {
// 3. Конфигурация интерфейса smoltcp
let smol_config = Config::new(smoltcp::wire::HardwareAddress::Ip);
let mut caps = DeviceCapabilities::default();
caps.max_transmission_unit = self.config.mtu; // Берем из конфига
caps.max_transmission_unit = self.config.mtu;
caps.medium = smoltcp::phy::Medium::Ip;
// --- СОЗДАЕМ КАНАЛЫ СВЯЗИ ДВИЖОК <-> ТУННЕЛЬ ---
// tx_to_tunnel: Движок пишет, ClientHandler читает (исходящий трафик)
let (tx_to_tunnel, rx_for_client_handler) =
mpsc::channel(NetworkConfig::global().muxer_capacity);
// tx_to_engine: ClientHandler пишет, Движок читает (входящий трафик)
let (tx_for_client_handler, rx_from_tunnel) =
mpsc::channel(NetworkConfig::global().muxer_capacity);
// 4. Подключение к серверу
info!("Establishing secure tunnel to proxy server...");
ClientHandler::connect(&self.config.remote_address)
.await
.map_err(|e| format!("Failed to establish secure tunnel: {}", e))?;
// ВАЖНО: Тебе нужно добавить эти аргументы в `ClientHandler::connect`,
// чтобы он знал, откуда забирать `RawCastFrame` для отправки на сервер,
// и куда отдавать `RawCastFrame`, прилетевшие от сервера.
ClientHandler::connect(
&self.config.remote_address,
rx_for_client_handler,
tx_for_client_handler,
)
.await
.map_err(|e| format!("Failed to establish secure tunnel: {}", e))?;
info!("Secure tunnel established, Muxer is ready.");
// 5. Инициализация и настройка Engine
let mut engine = Engine::new(smol_config, caps, dns_handler);
// 5. Инициализация и настройка Engine (передаем каналы)
let mut engine = Engine::new(smol_config, caps, dns_handler, tx_to_tunnel, rx_from_tunnel);
engine.set_any_ip(self.config.any_ip);