tcp sockets dynamic

client/src/tun/connection_manager.rs

@@ -3,7 +3,7 @@ use netrunner_logger::{debug, error, info, warn};
 use smoltcp::{
     iface::{SocketHandle, SocketSet},
     socket::{AnySocket, icmp, tcp, udp},
-    wire::IpListenEndpoint,
+    wire::{IpListenEndpoint, IpProtocol, Ipv4Packet, TcpPacket},
 };
 use std::{
     collections::HashMap,
@@ -22,6 +22,7 @@ pub struct ConnectionManager {
     fake_ip_store: FakeIpStore,
     proxy_ip: String,
     failed_until: HashMap<SocketHandle, StdInstant>,
+    sockets_to_remove: Vec<SocketHandle>,
 }
 
 impl ConnectionManager {
@@ -33,7 +34,8 @@ impl ConnectionManager {
             proxy_ip: ip,
             fake_ip_store: FakeIpStore::new(),
             failed_until: HashMap::new(),
-            dns_handler, // Просто сохраняем готовый объект
+            dns_handler,
+            sockets_to_remove: Vec::new(),
         }
     }
     pub fn start_listening(&mut self, socket_set: &mut SocketSet) {
@@ -112,19 +114,22 @@ impl ConnectionManager {
     fn handle_tcp(&mut self, handle: SocketHandle, socket: &mut tcp::Socket) {
         use tcp::State;
 
+        // 1. Очистка закрытых сокетов
         if socket.state() == State::Closed {
-            if let Some(until) = self.failed_until.get(&handle) {
+            // Если сокет закрыт, удаляем его сессию и помечаем на удаление из сета
-                if StdInstant::now() < *until {
+            if self.active_tcp_sessions.contains_key(&handle) {
-                    return;
+                debug!(%handle, "TCP session closed, removing from active sessions");
-                }
+                self.active_tcp_sessions.remove(&handle);
             }
 
-            self.active_tcp_sessions.remove(&handle);
+            // Добавляем в очередь на удаление из SocketSet (чтобы освободить память)
-            socket.abort();
+            if !self.sockets_to_remove.contains(&handle) {
-            let _ = socket.listen(443);
+                self.sockets_to_remove.push(handle);
+            }
             return;
         }
 
+        // 2. Инициализация сессии при установке соединения
         if socket.state() == State::Established && !self.active_tcp_sessions.contains_key(&handle) {
             let target = self.resolve_target(socket);
 
@@ -135,22 +140,18 @@ impl ConnectionManager {
                 }
             }
 
+            info!(%handle, "New TCP session established for target: {:?}", target);
             let conn = TcpConnection::new(handle, self.proxy_ip.clone(), target);
             self.active_tcp_sessions.insert(handle, conn);
         }
 
+        // 3. Тик активной сессии (проброс данных в прокси)
         if let Some(conn) = self.active_tcp_sessions.get_mut(&handle) {
             if !conn.tick(socket) {
-                debug!(%handle, "Connection handshake failed or closed, aborting socket.");
+                debug!(%handle, "Connection tick failed, aborting.");
-                self.active_tcp_sessions.remove(&handle);
                 socket.abort();
-                self.failed_until
+                // Сессия удалится на следующем проходе, когда стейт станет Closed
-                    .insert(handle, StdInstant::now() + Duration::from_secs(5));
             }
-        } else if socket.state() == State::Established {
-            self.failed_until
-                .insert(handle, StdInstant::now() + Duration::from_secs(5));
-            socket.abort();
         }
 
         if socket.state() == State::CloseWait {
@@ -187,6 +188,56 @@ impl ConnectionManager {
         socket
     }
 
+    fn create_dynamic_tcp_socket<'a>(port: u16) -> tcp::Socket<'a> {
+        let buf_size = match port {
+            443 | 80 => 256 * 1024,     // 256 KB для веба
+            22 | 53 | 123 => 16 * 1024, // 16 KB для мелких протоколов (SSH, DNS over TCP, NTP)
+            _ => 64 * 1024,             // Дефолт
+        };
+
+        let mut socket = tcp::Socket::new(
+            tcp::SocketBuffer::new(vec![0; buf_size]),
+            tcp::SocketBuffer::new(vec![0; buf_size]),
+        );
+
+        socket.set_nagle_enabled(false); // Для отзывчивости (особенно в играх типа Silent Hill, если через VPN)
+        socket.set_ack_delay(None);
+        socket
+    }
+
+    pub fn try_create_socket_from_packet(&mut self, packet: &[u8], socket_set: &mut SocketSet) {
+        let Ok(ip_packet) = Ipv4Packet::new_checked(packet) else {
+            return;
+        };
+        if ip_packet.next_header() != IpProtocol::Tcp {
+            return;
+        };
+        let Ok(tcp_packet) = TcpPacket::new_checked(ip_packet.payload()) else {
+            return;
+        };
+
+        // Ищем только SYN (начало соединения)
+        if tcp_packet.syn() && !tcp_packet.ack() {
+            let dst_port = tcp_packet.dst_port();
+            let dst_addr = ip_packet.dst_addr();
+
+            // Проверяем, не создали ли мы уже такой сокет на предыдущем шаге
+            if !self.has_active_session(socket_set, dst_addr.into(), dst_port) {
+                debug!(target: "netrunner", "Dynamic TCP: Creating socket for {}:{}", dst_addr, dst_port);
+
+                let mut socket = Self::create_dynamic_tcp_socket(dst_port);
+                let endpoint = IpListenEndpoint {
+                    addr: Some(dst_addr.into()),
+                    port: dst_port,
+                };
+
+                if let Ok(_) = socket.listen(endpoint) {
+                    socket_set.add(socket);
+                }
+            }
+        }
+    }
+
     fn create_udp_socket<'a>() -> udp::Socket<'a> {
         const BUF_SIZE: usize = 1024 * 64;
         const PACKET_COUNT: usize = 128;
@@ -211,12 +262,34 @@ impl ConnectionManager {
         icmp::Socket::new(icmp_rx_buffer, icmp_tx_buffer)
     }
 
-    pub fn setup_sockets(n_tcp: usize, n_udp: usize, n_icmp: usize) -> SocketSet<'static> {
+    fn has_active_session(
-        let mut sockets = SocketSet::new(Vec::with_capacity(n_tcp + n_udp + n_icmp));
+        &self,
-
+        socket_set: &SocketSet,
-        for _ in 0..n_tcp {
+        dst_addr: smoltcp::wire::IpAddress,
-            sockets.add(Self::create_tcp_socket());
+        dst_port: u16,
+    ) -> bool {
+        for (_, socket) in socket_set.iter() {
+            if let Some(tcp) = tcp::Socket::downcast(socket) {
+                if let Some(endpoint) = tcp.local_endpoint() {
+                    if endpoint.addr == dst_addr && endpoint.port == dst_port {
+                        return true;
+                    }
+                }
+            }
         }
+        false
+    }
+
+    pub fn cleanup(&mut self, socket_set: &mut SocketSet) {
+        for handle in self.sockets_to_remove.drain(..) {
+            debug!(%handle, "Memory free: removing socket from set");
+            socket_set.remove(handle);
+            self.last_activity.remove(&handle);
+            self.failed_until.remove(&handle);
+        }
+    }
+    pub fn setup_sockets(n_udp: usize, n_icmp: usize) -> SocketSet<'static> {
+        let mut sockets = SocketSet::new(Vec::with_capacity(n_udp + n_icmp + 10));
 
         for _ in 0..n_udp {
             sockets.add(Self::create_udp_socket());
@@ -228,7 +301,6 @@ impl ConnectionManager {
 
         sockets
     }
-
     pub fn log_status(&self, socket_set: &SocketSet) {
         let mut established = 0;
         let mut total_tcp = 0;

client/src/tun/device.rs

@@ -15,43 +15,86 @@ use smoltcp::{
 };
 use tokio::sync::mpsc;
 
+// --- TokenBuffer (без изменений, он у тебя отличный) ---
+const TOKEN_BUFFER_LIST_MAX_SIZE: usize = 64;
+static TOKEN_BUFFER_LIST: LazyLock<Mutex<Vec<BytesMut>>> = LazyLock::new(|| Mutex::new(Vec::new()));
+
+pub struct TokenBuffer {
+    buffer: BytesMut,
+}
+
+impl Drop for TokenBuffer {
+    fn drop(&mut self) {
+        let mut list = TOKEN_BUFFER_LIST.lock().unwrap();
+        if list.len() >= TOKEN_BUFFER_LIST_MAX_SIZE {
+            return;
+        }
+        let empty_buffer = BytesMut::new();
+        let mut buffer = mem::replace(&mut self.buffer, empty_buffer);
+        buffer.clear();
+        list.push(buffer);
+    }
+}
+
+impl TokenBuffer {
+    pub fn with_capacity(cap: usize) -> Self {
+        let mut list = TOKEN_BUFFER_LIST.lock().unwrap();
+        if let Some(mut buffer) = list.pop() {
+            buffer.reserve(cap);
+            return Self { buffer };
+        }
+        Self {
+            buffer: BytesMut::with_capacity(cap),
+        }
+    }
+}
+
+impl Deref for TokenBuffer {
+    type Target = BytesMut;
+    fn deref(&self) -> &Self::Target {
+        &self.buffer
+    }
+}
+impl DerefMut for TokenBuffer {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.buffer
+    }
+}
+
+// --- VirtTunDevice ---
 pub struct VirtTunDevice {
     capabilities: DeviceCapabilities,
-    in_buf: mpsc::UnboundedReceiver<TokenBuffer>,
+    rx_queue: mpsc::UnboundedReceiver<TokenBuffer>, // smoltcp читает отсюда
-    out_buf: mpsc::UnboundedSender<TokenBuffer>,
+    tx_queue: mpsc::UnboundedSender<TokenBuffer>,   // smoltcp пишет сюда
-    in_buf_avail: Arc<AtomicBool>,
+    rx_avail: Arc<AtomicBool>,
 }
 
 impl VirtTunDevice {
-    #[allow(clippy::type_complexity)]
     pub fn new(
         capabilities: DeviceCapabilities,
     ) -> (
         Self,
-        mpsc::UnboundedReceiver<TokenBuffer>,
+        mpsc::UnboundedSender<TokenBuffer>, // Канал, чтобы закидывать пакеты в smoltcp
-        mpsc::UnboundedSender<TokenBuffer>,
+        mpsc::UnboundedReceiver<TokenBuffer>, // Канал, чтобы забирать готовые пакеты из smoltcp
         Arc<AtomicBool>,
     ) {
-        let (iface_tx, iface_output) = mpsc::unbounded_channel();
+        let (to_smoltcp_tx, to_smoltcp_rx) = mpsc::unbounded_channel();
-        let (iface_input, iface_rx) = mpsc::unbounded_channel();
+        let (from_smoltcp_tx, from_smoltcp_rx) = mpsc::unbounded_channel();
-        let in_buf_avail = Arc::new(AtomicBool::new(false));
+        let rx_avail = Arc::new(AtomicBool::new(false));
 
-        (
+        let device = Self {
-            Self {
+            capabilities,
-                capabilities,
+            rx_queue: to_smoltcp_rx,
-                in_buf: iface_rx,
+            tx_queue: from_smoltcp_tx,
-                out_buf: iface_tx,
+            rx_avail: rx_avail.clone(),
-                in_buf_avail: in_buf_avail.clone(),
+        };
-            },
+
-            iface_output,
+        (device, to_smoltcp_tx, from_smoltcp_rx, rx_avail)
-            iface_input,
-            in_buf_avail,
-        )
     }
 
     #[inline]
-    pub fn recv_available(&self) -> bool {
+    pub fn mark_rx_available(&self) {
-        self.in_buf_avail.load(Ordering::Acquire)
+        self.rx_avail.store(true, Ordering::Release);
     }
 }
 
@@ -60,7 +103,7 @@ impl Device for VirtTunDevice {
     type TxToken<'a> = VirtTxToken<'a>;
 
     fn receive(&mut self, _timestamp: Instant) -> Option<(Self::RxToken<'_>, Self::TxToken<'_>)> {
-        if let Ok(buffer) = self.in_buf.try_recv() {
+        if let Ok(buffer) = self.rx_queue.try_recv() {
             let rx = Self::RxToken {
                 buffer,
                 phantom_device: PhantomData,
@@ -68,7 +111,7 @@ impl Device for VirtTunDevice {
             let tx = VirtTxToken(self);
             return Some((rx, tx));
         }
-        self.in_buf_avail.store(false, Ordering::Release);
+        self.rx_avail.store(false, Ordering::Release);
         None
     }
 
@@ -106,61 +149,8 @@ impl phy::TxToken for VirtTxToken<'_> {
         unsafe {
             buffer.set_len(len);
         }
-
         let result = f(&mut buffer);
-        self.0
+        let _ = self.0.tx_queue.send(buffer);
-            .out_buf
-            .send(buffer)
-            .expect("channel closed unexpectedly");
         result
     }
 }
-
-const TOKEN_BUFFER_LIST_MAX_SIZE: usize = 64;
-static TOKEN_BUFFER_LIST: LazyLock<Mutex<Vec<BytesMut>>> = LazyLock::new(|| Mutex::new(Vec::new()));
-
-pub struct TokenBuffer {
-    buffer: BytesMut,
-}
-
-impl Drop for TokenBuffer {
-    fn drop(&mut self) {
-        let mut list = TOKEN_BUFFER_LIST.lock().unwrap();
-        if list.len() >= TOKEN_BUFFER_LIST_MAX_SIZE {
-            return;
-        }
-
-        let empty_buffer = BytesMut::new();
-        let mut buffer = mem::replace(&mut self.buffer, empty_buffer);
-        buffer.clear();
-
-        list.push(buffer);
-    }
-}
-
-impl TokenBuffer {
-    pub fn with_capacity(cap: usize) -> Self {
-        let mut list = TOKEN_BUFFER_LIST.lock().unwrap();
-        if let Some(mut buffer) = list.pop() {
-            buffer.reserve(cap);
-            return Self { buffer };
-        }
-        Self {
-            buffer: BytesMut::with_capacity(cap),
-        }
-    }
-}
-
-impl Deref for TokenBuffer {
-    type Target = BytesMut;
-
-    fn deref(&self) -> &Self::Target {
-        &self.buffer
-    }
-}
-
-impl DerefMut for TokenBuffer {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.buffer
-    }
-}

client/src/tun/engine.rs

@@ -7,7 +7,6 @@ use smoltcp::{
 };
 use std::sync::atomic::Ordering;
 use std::{
-    mem,
     sync::{Arc, LazyLock, atomic::AtomicBool},
     time::Instant as StdInstant,
 };
@@ -18,19 +17,25 @@ use tun::{DeviceReader, DeviceWriter};
 
 use netrunner_logger::{debug, info, warn};
 
-use crate::connections::dns::{self, DnsHandler};
+use crate::connections::dns::DnsHandler;
 use crate::tun::connection_manager::ConnectionManager;
 use crate::tun::device::{TokenBuffer, VirtTunDevice};
 use crate::tun::tun::Tun;
 
 pub static START_TIME: LazyLock<StdInstant> = LazyLock::new(StdInstant::now);
+
 pub struct Engine {
     interface: Interface,
     socket_set: SocketSet<'static>,
     manager: ConnectionManager,
     device: VirtTunDevice,
-    bridge_rx: UnboundedReceiver<TokenBuffer>,
+
-    bridge_tx: UnboundedSender<TokenBuffer>,
+    // Канал для инъекции пакетов в smoltcp
+    to_smoltcp_tx: UnboundedSender<TokenBuffer>,
+
+    // Временное хранилище канала выхода из smoltcp (отдадим воркеру в run)
+    from_smoltcp_rx: Option<UnboundedReceiver<TokenBuffer>>,
+
     avail: Arc<AtomicBool>,
 }
 
@@ -42,17 +47,19 @@ impl Engine {
         dns_handler: DnsHandler,
     ) -> Self {
         let now = Engine::current_time();
-        let (mut device, bridge_rx, bridge_tx, avail) = VirtTunDevice::new(caps);
+        // Смотри, как чисто мы получаем компоненты:
+        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(128, 128, 4);
+        let socket_set = ConnectionManager::setup_sockets(64, 4);
         let manager = ConnectionManager::new(ip, dns_handler);
+
         Self {
             interface,
             socket_set,
             device,
-            bridge_tx,
+            to_smoltcp_tx,
-            bridge_rx,
+            from_smoltcp_rx: Some(from_smoltcp_rx),
             avail,
             manager,
         }
@@ -60,17 +67,33 @@ impl Engine {
 
     pub async fn run(&mut self, tun: Tun) {
         info!("Current routes: {:?}", self.interface.routes());
-
         let (writer, reader) = tun.split().expect("Failed to split TUN");
 
-        let from_engine = mem::replace(&mut self.bridge_rx, mpsc::unbounded_channel().1);
+        // 1. Создаем трубу для ВХОДЯЩЕГО трафика: TUN -> Engine
+        let (tun_to_engine_tx, mut tun_to_engine_rx) = mpsc::unbounded_channel();
+        Self::spawn_tun_reader(reader, tun_to_engine_tx, self.avail.clone());
+
+        // 2. Запускаем трубу ИСХОДЯЩЕГО трафика: smoltcp -> TUN
+        let from_smoltcp_rx = self.from_smoltcp_rx.take().expect("Engine started twice");
+        Self::spawn_tun_writer(writer, from_smoltcp_rx);
 
-        Self::spawn_tun_to_engine(reader, self.bridge_tx.clone(), self.avail.clone());
-        Self::spawn_engine_to_tun(writer, from_engine);
         let mut last_log = StdInstant::now();
-        loop {
-            let result = self.poll();
 
+        // 3. Главный цикл (ДИСПЕТЧЕР)
+        loop {
+            // Читаем всё, что прилетело из операционной системы
+            while let Ok(token) = tun_to_engine_rx.try_recv() {
+                // АНАЛИЗ: Создаем сокет, если это новый SYN
+                self.manager
+                    .try_create_socket_from_packet(&token, &mut self.socket_set);
+
+                // ПЕРЕДАЧА: Закидываем пакет в виртуальную сетевую карту smoltcp
+                if self.to_smoltcp_tx.send(token).is_ok() {
+                    self.device.mark_rx_available();
+                }
+            }
+
+            let result = self.poll();
             self.manager.process_sockets(&mut self.socket_set);
 
             if last_log.elapsed() >= Duration::from_secs(5) {
@@ -82,11 +105,12 @@ impl Engine {
                 continue;
             }
 
-            if self.avail.load(Ordering::Acquire) {
+            if self.avail.swap(false, Ordering::Acquire) {
                 tokio::task::yield_now().await;
                 continue;
             }
 
+            self.manager.cleanup(&mut self.socket_set);
             self.poll_delay().await;
         }
     }
@@ -107,13 +131,14 @@ impl Engine {
         sleep(sleep_duration).await;
     }
 
-    fn spawn_tun_to_engine(
+    // Воркер: Читает ОС -> Шлет в Engine
+    fn spawn_tun_reader(
         mut reader: DeviceReader,
         to_engine: UnboundedSender<TokenBuffer>,
         is_avail: Arc<AtomicBool>,
     ) {
         tokio::spawn(async move {
-            debug!("TUN-to-Engine bridge task started");
+            debug!("TUN Reader task started");
             let mut buf = [0u8; 65536];
             while let Ok(n) = reader.read(&mut buf).await {
                 if n == 0 {
@@ -122,17 +147,14 @@ impl Engine {
 
                 let first_byte = buf[0];
                 let version = first_byte >> 4;
-
                 if version == 4 {
-                    // IPv4 Multicast: 224.0.0.0 - 239.255.255.255
                     if buf[12..16] == [0, 0, 0, 0] {
                         continue;
-                    } // Маршрутизация
+                    }
                     if buf[16] >= 224 && buf[16] <= 239 {
                         continue;
                     }
                 } else if version == 6 {
-                    // IPv6 Multicast: ff00::/8 (первый байт 0xff)
                     if first_byte == 0xff {
                         continue;
                     }
@@ -142,27 +164,28 @@ impl Engine {
                 token.extend_from_slice(&buf[..n]);
 
                 if to_engine.send(token).is_ok() {
-                    is_avail.store(true, std::sync::atomic::Ordering::Release);
+                    is_avail.store(true, Ordering::Release);
                 } else {
                     break;
                 }
             }
-            warn!("TUN-to-Engine bridge task stopped");
+            warn!("TUN Reader task stopped");
         });
     }
 
-    fn spawn_engine_to_tun(
+    // Воркер: Читает smoltcp -> Шлет в ОС
+    fn spawn_tun_writer(
         mut writer: DeviceWriter,
-        mut from_engine: UnboundedReceiver<TokenBuffer>,
+        mut from_smoltcp: UnboundedReceiver<TokenBuffer>,
     ) {
         tokio::spawn(async move {
-            debug!("Engine-to-TUN bridge task started");
+            debug!("TUN Writer task started");
-            while let Some(token) = from_engine.recv().await {
+            while let Some(token) = from_smoltcp.recv().await {
                 if writer.write_all(&token).await.is_err() {
                     break;
                 }
             }
-            warn!("Engine-to-TUN bridge task stopped");
+            warn!("TUN Writer task stopped");
         });
     }
 
@@ -171,19 +194,17 @@ impl Engine {
         Instant::from_micros(duration.as_micros() as i64)
     }
 
-    pub fn set_any_ip(&mut self, state: bool) -> () {
+    pub fn set_any_ip(&mut self, state: bool) {
         self.interface.set_any_ip(state)
     }
 
     pub fn set_transparent_mode(&mut self) {
         self.interface.update_ip_addrs(|addrs| {
             addrs.clear();
-
             addrs
                 .push(IpCidr::new(IpAddress::v4(10, 0, 0, 2), 24))
                 .unwrap();
         });
-
         self.interface.routes_mut().remove_default_ipv4_route();
     }
 

core/src/proxy/connection/handler.rs

@@ -36,7 +36,7 @@ impl StreamHandler {
             let muxer = self.muxer.clone();
 
             // Канал для передачи данных из мультиплексора в мост (bridge)
-            let (v_tx, v_rx) = tokio::sync::mpsc::channel(2048);
+            let (v_tx, v_rx) = tokio::sync::mpsc::channel(512);
             muxer.register_stream(stream_id, v_tx).await;
 
             tokio::spawn(async move {