renames and tauri app
This commit is contained in:
@@ -0,0 +1,71 @@
|
||||
use crate::protocol::codec::frame::FrameType;
|
||||
use crate::proxy::connection::connection::BUF_SIZE;
|
||||
use crate::proxy::connection::muxer::{MuxMessage, Muxer};
|
||||
use bytes::{Bytes, BytesMut};
|
||||
use tokio::sync::mpsc;
|
||||
use tracing::{debug, error};
|
||||
pub async fn run_proxy_bridge<R, W>(
|
||||
stream_id: u32,
|
||||
mut reader: R,
|
||||
mut writer: W,
|
||||
muxer: Muxer,
|
||||
mut v_rx: mpsc::Receiver<Bytes>,
|
||||
) where
|
||||
R: tokio::io::AsyncReadExt + Unpin,
|
||||
W: tokio::io::AsyncWriteExt + Unpin,
|
||||
{
|
||||
let mut buf = BytesMut::with_capacity(BUF_SIZE);
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
res = reader.read_buf(&mut buf) => {
|
||||
match res {
|
||||
Ok(0) => {
|
||||
debug!(stream_id, "Socket closed (EOF)");
|
||||
break;
|
||||
}
|
||||
Ok(_) => {
|
||||
let msg = MuxMessage {
|
||||
stream_id,
|
||||
frame_type: FrameType::Data,
|
||||
data: buf.split().freeze(),
|
||||
};
|
||||
if muxer.to_network.send(msg).await.is_err() { break; }
|
||||
}
|
||||
Err(e) => {
|
||||
error!(stream_id, error = %e, "Socket read error");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
// Читаем из туннеля (v_rx) -> шлем в сокет
|
||||
maybe_data = v_rx.recv() => {
|
||||
match maybe_data {
|
||||
Some(data) => {
|
||||
if data.is_empty() { break; } // EOF от другой стороны
|
||||
if let Err(e) = writer.write_all(&data).await {
|
||||
error!(stream_id, error = %e, "Socket write error");
|
||||
break;
|
||||
}
|
||||
}
|
||||
None => {
|
||||
debug!(stream_id, "Virtual channel closed");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Финализация (общая для всех)
|
||||
let _ = muxer
|
||||
.to_network
|
||||
.send(MuxMessage {
|
||||
stream_id,
|
||||
frame_type: FrameType::Close,
|
||||
data: Bytes::new(),
|
||||
})
|
||||
.await;
|
||||
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
|
||||
muxer.remove_stream(stream_id).await;
|
||||
}
|
||||
@@ -0,0 +1,247 @@
|
||||
use bytes::{Bytes, BytesMut};
|
||||
use tracing::{instrument, info, debug, error, trace, warn};
|
||||
use std::{net::SocketAddr};
|
||||
use tokio::{
|
||||
io::{AsyncReadExt, AsyncWriteExt},
|
||||
net::{
|
||||
tcp::{OwnedReadHalf, OwnedWriteHalf},
|
||||
TcpStream,
|
||||
},
|
||||
sync::mpsc::{self},
|
||||
};
|
||||
|
||||
use crate::{
|
||||
protocol::{
|
||||
codec::{
|
||||
codec::Codec,
|
||||
frame::FrameType,
|
||||
socks::{SocksReply, SocksRequest, SocksTarget},
|
||||
},
|
||||
errors::ErrorAction,
|
||||
parser::parser::Parser,
|
||||
},
|
||||
proxy::connection::{
|
||||
bridge::run_proxy_bridge, engine::TunnelEngine, handler::StreamHandler, muxer::{MuxMessage, Muxer}
|
||||
},
|
||||
};
|
||||
|
||||
pub const BUF_SIZE: usize = 16384;
|
||||
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq)]
|
||||
pub enum ConnectionRole {
|
||||
Client,
|
||||
Server,
|
||||
}
|
||||
|
||||
pub struct Connection {
|
||||
addr: SocketAddr,
|
||||
pub inbound: OwnedReadHalf,
|
||||
pub outbound: OwnedWriteHalf,
|
||||
pub read_buf: BytesMut,
|
||||
pub codec: Codec,
|
||||
}
|
||||
|
||||
impl Connection {
|
||||
pub fn new(
|
||||
stream: TcpStream,
|
||||
addr: SocketAddr,
|
||||
init: bool,
|
||||
) -> Self {
|
||||
let (inbound, outbound) = stream.into_split();
|
||||
Self {
|
||||
addr,
|
||||
inbound,
|
||||
outbound,
|
||||
read_buf: BytesMut::with_capacity(BUF_SIZE),
|
||||
codec: Codec::new(init),
|
||||
}
|
||||
}
|
||||
|
||||
/// Читает и парсит запрос SOCKS5 из входящего потока
|
||||
async fn read_socks_request(&mut self) -> Result<SocksRequest, String> {
|
||||
loop {
|
||||
// Попытка парсинга из текущего буфера
|
||||
match SocksRequest::parse(&mut self.read_buf) {
|
||||
Ok(Some(req)) => {
|
||||
// Используем Debug-вывод (?req), так как SocksRequest обычно Enum
|
||||
info!(client = %self.addr, request = ?req, "SOCKS request successfully parsed");
|
||||
return Ok(req);
|
||||
}
|
||||
Ok(None) => {
|
||||
// Это не ошибка, просто данных в сокете пока меньше, чем размер структуры SOCKS
|
||||
trace!(client = %self.addr, buffer_len = self.read_buf.len(), "SOCKS parse: need more data");
|
||||
}
|
||||
Err(e) => {
|
||||
error!(client = %self.addr, error = %e, "SOCKS protocol violation");
|
||||
return Err(format!("Socks parse error: {}", e));
|
||||
}
|
||||
}
|
||||
|
||||
// Чтение новых данных из сокета
|
||||
let n = self
|
||||
.inbound
|
||||
.read_buf(&mut self.read_buf)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
error!(client = %self.addr, error = %e, "Failed to read from socket during SOCKS handshake");
|
||||
e.to_string()
|
||||
})?;
|
||||
|
||||
if n == 0 {
|
||||
warn!(client = %self.addr, "Client closed connection prematurely during SOCKS handshake");
|
||||
return Err("Client closed connection during SOCKS handshake".into());
|
||||
}
|
||||
|
||||
trace!(client = %self.addr, read_bytes = n, "Read data from client for SOCKS handshake");
|
||||
}
|
||||
}
|
||||
|
||||
/// Отправляет SOCKS ответ
|
||||
async fn send_socks_reply(&mut self, reply: SocksReply) -> Result<(), String> {
|
||||
let mut buf = BytesMut::with_capacity(24);
|
||||
debug!(client = %self.addr, reply = ?reply, "Sending SOCKS reply to client");
|
||||
reply.write_to(&mut buf);
|
||||
|
||||
|
||||
|
||||
self.outbound
|
||||
.write_all(&buf)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
error!(client = %self.addr, error = %e, "Failed to send SOCKS reply");
|
||||
e.to_string()
|
||||
})?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[instrument(
|
||||
name = "socks_handler",
|
||||
skip(self, muxer),
|
||||
fields(addr = %self.addr)
|
||||
)]
|
||||
pub async fn handle_socks_client(mut self, muxer: Muxer) -> Result<(), String> {
|
||||
info!("Starting SOCKS multiplexed handling");
|
||||
|
||||
// 1. SOCKS Handshake
|
||||
debug!("Reading SOCKS handshake request");
|
||||
let _ = self.read_socks_request().await.map_err(|e| {
|
||||
error!("SOCKS handshake failed: {}", e);
|
||||
e
|
||||
})?;
|
||||
|
||||
self.send_socks_reply(SocksReply::HandshakeSelect { method: 0x00 }).await?;
|
||||
|
||||
// 2. SOCKS Connect
|
||||
// 2. SOCKS Connect - читаем, КУДА хочет браузер
|
||||
let req = self.read_socks_request().await?;
|
||||
let target = if let SocksRequest::Connect { target, .. } = req {
|
||||
target
|
||||
} else {
|
||||
return Err("Expected Connect".into());
|
||||
};
|
||||
|
||||
let stream_id = muxer.next_id();
|
||||
let target_str = target.to_string();
|
||||
|
||||
// --- НОВАЯ ЛОГИКА ОЖИДАНИЯ ---
|
||||
// Регистрируем временный канал, чтобы получить Connect-подтверждение от сервера
|
||||
let (v_tx, mut v_rx) = mpsc::channel::<Bytes>(1024);
|
||||
muxer.register_stream(stream_id, v_tx).await;
|
||||
|
||||
// Отправляем Connect-кадр на сервер
|
||||
muxer.to_network.send(MuxMessage {
|
||||
stream_id,
|
||||
frame_type: FrameType::Connect,
|
||||
data: Bytes::from(target_str),
|
||||
}).await.map_err(|e| e.to_string())?;
|
||||
|
||||
let first_payload = match tokio::time::timeout(std::time::Duration::from_secs(10), v_rx.recv()).await {
|
||||
Ok(Some(data)) => data,
|
||||
_ => {
|
||||
error!(stream_id, "Server timeout or failed to send Connect confirmation");
|
||||
// Шлем браузеру ошибку, если сервер промолчал
|
||||
self.send_socks_reply(SocksReply::ConnectResult {
|
||||
reply_code: 0x01, atyp: 0x01, addr: [0, 0, 0, 0], port: 0,
|
||||
}).await.ok();
|
||||
return Err("Target connection failed".into());
|
||||
}
|
||||
};
|
||||
|
||||
// Проверяем код ответа (второй байт в SOCKS5)
|
||||
if first_payload.len() >= 2 && first_payload[1] == 0x00 {
|
||||
debug!(stream_id, "Server confirmed connection, forwarding SOCKS reply to browser");
|
||||
|
||||
// ВАЖНО: Отправляем браузеру ТО, что прислал сервер (те самые 10 байт)
|
||||
// Не создаем новый SocksReply вручную, а пробрасываем байты сервера
|
||||
self.outbound.write_all(&first_payload).await.map_err(|e| e.to_string())?;
|
||||
} else {
|
||||
// Если сервер прислал ошибку (reply_code != 0), тоже пробрасываем её браузеру и выходим
|
||||
self.outbound.write_all(&first_payload).await.ok();
|
||||
return Err("Server rejected connection".into());
|
||||
}
|
||||
|
||||
// 4. Разбираем self и запускаем хендлер
|
||||
let Self { inbound: browser_in, outbound: browser_out, .. } = self;
|
||||
|
||||
let muxer_clone = muxer.clone();
|
||||
tokio::spawn(async move {
|
||||
run_proxy_bridge(stream_id, browser_in, browser_out, muxer_clone, v_rx).await;
|
||||
});
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[instrument(
|
||||
name = "server_tunnel",
|
||||
skip(self),
|
||||
fields(addr = %self.addr)
|
||||
)]
|
||||
pub async fn handle_server_tunnel(mut self) -> Result<(), String> {
|
||||
info!("Acting as TLS Server, waiting for ClientHello");
|
||||
|
||||
// Создаем Muxer для сервера
|
||||
let (mux_tx, mux_rx) = mpsc::channel(BUF_SIZE);
|
||||
let muxer = Muxer::new(mux_tx.clone(), false); // false, так как это Сервер
|
||||
|
||||
// 1. TLS Handshake
|
||||
let server_hello_bytes = loop {
|
||||
match self.codec.make_server_handshake(&mut self.read_buf) {
|
||||
Ok(bytes) => {
|
||||
info!("ClientHello received, sending ServerHello");
|
||||
break bytes;
|
||||
},
|
||||
Err(e) if e.action == ErrorAction::Wait => {
|
||||
let n = self.inbound.read_buf(&mut self.read_buf).await
|
||||
.map_err(|err| format!("Read error: {}", err))?;
|
||||
|
||||
if n == 0 { return Err("Client closed connection".into()); }
|
||||
}
|
||||
Err(e) => return Err(format!("TLS error: {:?}", e)),
|
||||
}
|
||||
};
|
||||
|
||||
self.outbound.write_all(&server_hello_bytes).await.map_err(|e| e.to_string())?;
|
||||
info!("TLS Tunnel established as server");
|
||||
|
||||
let handler = std::sync::Arc::new(StreamHandler::new(muxer.clone(), ConnectionRole::Server));
|
||||
|
||||
// 2. Передача управления в TunnelEngine
|
||||
debug!("Handover to TunnelEngine");
|
||||
let engine = TunnelEngine {
|
||||
inbound: self.inbound,
|
||||
outbound: self.outbound,
|
||||
codec: self.codec,
|
||||
read_buf: self.read_buf,
|
||||
mux_rx,
|
||||
handler
|
||||
};
|
||||
|
||||
engine.run().await.map_err(|e| {
|
||||
error!("TunnelEngine error: {}", e);
|
||||
e
|
||||
})
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use bytes::BytesMut;
|
||||
use tokio::{
|
||||
io::{AsyncReadExt, AsyncWriteExt},
|
||||
net::tcp::{OwnedReadHalf, OwnedWriteHalf},
|
||||
sync::mpsc::Receiver,
|
||||
};
|
||||
use tracing::{debug, error};
|
||||
|
||||
use crate::{
|
||||
protocol::{codec::codec::Codec, errors::ErrorAction},
|
||||
proxy::connection::{handler::StreamHandler, muxer::MuxMessage},
|
||||
};
|
||||
|
||||
pub struct TunnelEngine {
|
||||
pub inbound: OwnedReadHalf,
|
||||
pub outbound: OwnedWriteHalf,
|
||||
pub codec: Codec,
|
||||
pub read_buf: BytesMut,
|
||||
pub mux_rx: Receiver<MuxMessage>,
|
||||
pub handler: Arc<StreamHandler>, // Добавь это вместо прямого вызова логики
|
||||
}
|
||||
|
||||
impl TunnelEngine {
|
||||
pub async fn run(self) -> Result<(), String> {
|
||||
let mut inbound = self.inbound;
|
||||
let mut outbound = self.outbound;
|
||||
let mut codec = self.codec;
|
||||
let mut read_buf = self.read_buf;
|
||||
let mut mux_rx = self.mux_rx;
|
||||
let handler = self.handler;
|
||||
loop {
|
||||
tokio::select! {
|
||||
res = Self::process_inbound(&mut inbound, &mut codec, &mut read_buf, &handler) => {
|
||||
res?
|
||||
}
|
||||
|
||||
// НУЖНО ОТПРАВИТЬ В СЕТЬ (В сторону удаленного прокси)
|
||||
Some(msg) = mux_rx.recv() => {
|
||||
Self::handle_outbound( &mut outbound, &mut codec, msg).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn process_inbound(
|
||||
inbound: &mut OwnedReadHalf,
|
||||
codec: &mut Codec,
|
||||
read_buf: &mut BytesMut,
|
||||
handler: &Arc<StreamHandler>,
|
||||
) -> Result<(), String> {
|
||||
let n = inbound
|
||||
.read_buf(read_buf)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
if n == 0 && read_buf.is_empty() {
|
||||
return Err("EOF".into());
|
||||
}
|
||||
|
||||
loop {
|
||||
match codec.inbound(read_buf) {
|
||||
// 1. Успешно достали фрейм
|
||||
Ok(Some(frame)) => {
|
||||
handler.handle(frame).await;
|
||||
}
|
||||
// 2. Данных в буфере недостаточно (нужно подождать еще)
|
||||
Ok(None) => break,
|
||||
|
||||
// 3. Ошибка кодека
|
||||
Err(e) => {
|
||||
// Если кодек говорит "подожди", выходим из цикла парсинга
|
||||
if e.action == ErrorAction::Wait {
|
||||
break;
|
||||
}
|
||||
// Иначе — это реальная проблема (кривой TLS и т.д.)
|
||||
error!(error = ?e, "Codec inbound failed");
|
||||
return Err(format!("Codec error: {:?}", e));
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn handle_outbound(
|
||||
outbound: &mut OwnedWriteHalf,
|
||||
codec: &mut Codec,
|
||||
msg: MuxMessage,
|
||||
) -> Result<(), String> {
|
||||
// 1. Шифруем данные, используя только кодек
|
||||
match codec.encrypt_data(msg.stream_id, msg.frame_type, msg.data) {
|
||||
Ok(pkt) => {
|
||||
// 2. Пишем в сокет, используя только outbound
|
||||
outbound
|
||||
.write_all(&pkt)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
error!(stream_id = msg.stream_id, error = %e, "Failed to write encrypted data to network");
|
||||
e.to_string()
|
||||
})?;
|
||||
debug!(stream_id = msg.stream_id, "Outbound packet sent");
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => {
|
||||
error!(stream_id = msg.stream_id, error = ?e, "Encryption failed for outbound message");
|
||||
Err(format!("Encryption error: {:?}", e))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
use bytes::{Bytes, BytesMut};
|
||||
use tracing::{debug, error, info};
|
||||
|
||||
use crate::{
|
||||
protocol::codec::{
|
||||
frame::{Frame, FrameType},
|
||||
socks::SocksReply,
|
||||
},
|
||||
proxy::connection::{bridge::run_proxy_bridge, connection::ConnectionRole, muxer::Muxer},
|
||||
};
|
||||
|
||||
// proxy/connection/stream_handler.rs
|
||||
pub struct StreamHandler {
|
||||
muxer: Muxer,
|
||||
role: ConnectionRole,
|
||||
}
|
||||
|
||||
impl StreamHandler {
|
||||
pub fn new(muxer: Muxer, role: ConnectionRole) -> Self {
|
||||
Self { muxer, role }
|
||||
}
|
||||
|
||||
pub async fn handle(&self, frame: Frame) {
|
||||
let stream_id = frame.header.stream_id;
|
||||
|
||||
match frame.header.frame_type {
|
||||
FrameType::Connect => self.on_connect(stream_id, frame.payload).await,
|
||||
FrameType::Data => self.on_data(stream_id, frame.payload).await,
|
||||
FrameType::Close => self.on_close(stream_id).await,
|
||||
_ => debug!(stream_id, "Unhandled frame type"),
|
||||
}
|
||||
}
|
||||
|
||||
async fn on_connect(&self, stream_id: u32, payload: Bytes) {
|
||||
if self.role == ConnectionRole::Server {
|
||||
let target_str = String::from_utf8_lossy(&payload).to_string();
|
||||
let muxer = self.muxer.clone();
|
||||
|
||||
let (v_tx, v_rx) = tokio::sync::mpsc::channel(100);
|
||||
muxer.register_stream(stream_id, v_tx).await;
|
||||
|
||||
tokio::spawn(async move {
|
||||
info!(stream_id, target = %target_str, "Attempting remote connection");
|
||||
|
||||
match tokio::net::TcpStream::connect(&target_str).await {
|
||||
Ok(stream) => {
|
||||
// --- ШАГ 2: ШЛЕМ ПОДТВЕРЖДЕНИЕ ---
|
||||
let mut reply_buf = BytesMut::with_capacity(10);
|
||||
let reply = SocksReply::ConnectResult {
|
||||
reply_code: 0x00,
|
||||
atyp: 0x01,
|
||||
addr: [0, 0, 0, 0],
|
||||
port: 0,
|
||||
};
|
||||
reply.write_to(&mut reply_buf);
|
||||
|
||||
let _ = muxer
|
||||
.send_control(stream_id, FrameType::Connect, reply_buf.freeze())
|
||||
.await;
|
||||
|
||||
// --- ШАГ 3: ЗАПУСКАЕМ МОСТ ---
|
||||
let (r, w) = stream.into_split();
|
||||
run_proxy_bridge(stream_id, r, w, muxer, v_rx).await;
|
||||
}
|
||||
Err(e) => {
|
||||
error!(stream_id, error = %e, "Connection failed");
|
||||
// Если не подключились — удаляем стрим, чтобы не висел в мапе
|
||||
muxer.remove_stream(stream_id).await;
|
||||
|
||||
let mut reply_buf = BytesMut::with_capacity(10);
|
||||
let reply = SocksReply::ConnectResult {
|
||||
reply_code: 0x01,
|
||||
atyp: 0x01,
|
||||
addr: [0, 0, 0, 0],
|
||||
port: 0,
|
||||
};
|
||||
reply.write_to(&mut reply_buf);
|
||||
let _ = muxer
|
||||
.send_control(stream_id, FrameType::Connect, reply_buf.freeze())
|
||||
.await;
|
||||
}
|
||||
}
|
||||
});
|
||||
} else {
|
||||
// Логика для клиента (проброс ответа сервера браузеру)
|
||||
self.muxer.dispatch_to_local(stream_id, payload).await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn on_data(&self, stream_id: u32, payload: Bytes) {
|
||||
self.muxer.dispatch_to_local(stream_id, payload).await;
|
||||
}
|
||||
|
||||
async fn on_close(&self, stream_id: u32) {
|
||||
self.muxer.dispatch_to_local(stream_id, Bytes::new()).await;
|
||||
self.muxer.remove_stream(stream_id).await;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
pub mod bridge;
|
||||
pub mod connection;
|
||||
pub mod engine;
|
||||
pub mod handler;
|
||||
pub mod muxer;
|
||||
@@ -0,0 +1,115 @@
|
||||
use crate::protocol::codec::frame::FrameType;
|
||||
use bytes::Bytes;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc::Sender;
|
||||
use tokio::sync::RwLock;
|
||||
use tracing::{debug, error};
|
||||
|
||||
pub struct IdGenerator {
|
||||
counter: AtomicU32,
|
||||
}
|
||||
|
||||
impl IdGenerator {
|
||||
pub fn new(is_client: bool) -> Self {
|
||||
let start = if is_client { 1 } else { 2 };
|
||||
Self {
|
||||
counter: AtomicU32::new(start),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn next(&self) -> u32 {
|
||||
self.counter.fetch_add(2, Ordering::Relaxed)
|
||||
}
|
||||
}
|
||||
|
||||
pub struct MuxMessage {
|
||||
pub stream_id: u32,
|
||||
pub frame_type: FrameType,
|
||||
pub data: Bytes,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct Muxer {
|
||||
pub to_network: Sender<MuxMessage>,
|
||||
streams: Arc<RwLock<HashMap<u32, Sender<Bytes>>>>,
|
||||
id_gen: Arc<IdGenerator>,
|
||||
}
|
||||
|
||||
impl Muxer {
|
||||
pub fn new(to_network: Sender<MuxMessage>, is_client: bool) -> Self {
|
||||
Self {
|
||||
to_network,
|
||||
streams: Arc::new(RwLock::new(HashMap::new())),
|
||||
id_gen: Arc::new(IdGenerator::new(is_client)),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn next_id(&self) -> u32 {
|
||||
self.id_gen.next()
|
||||
}
|
||||
|
||||
pub async fn register_stream(&self, stream_id: u32, tx: Sender<Bytes>) {
|
||||
let mut lock = self.streams.write().await;
|
||||
lock.insert(stream_id, tx);
|
||||
tracing::debug!(
|
||||
stream_id,
|
||||
total_active = lock.len(),
|
||||
"MUXER: [REGISTER] Stream added"
|
||||
);
|
||||
}
|
||||
|
||||
pub async fn remove_stream(&self, stream_id: u32) {
|
||||
let mut lock = self.streams.write().await;
|
||||
lock.remove(&stream_id); // Просто удаляем, если есть
|
||||
}
|
||||
|
||||
pub async fn send_control(
|
||||
&self,
|
||||
stream_id: u32,
|
||||
f_type: FrameType,
|
||||
data: Bytes,
|
||||
) -> Result<(), String> {
|
||||
self.to_network
|
||||
.send(MuxMessage {
|
||||
stream_id,
|
||||
frame_type: f_type,
|
||||
data,
|
||||
})
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub async fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
|
||||
let tx = {
|
||||
let lock = self.streams.read().await;
|
||||
lock.get(&stream_id).cloned()
|
||||
};
|
||||
|
||||
if let Some(tx) = tx {
|
||||
if data.is_empty() {
|
||||
tracing::debug!(stream_id, "MUXER: [EOF] Forwarding EOF to local handler");
|
||||
} else {
|
||||
tracing::trace!(
|
||||
stream_id,
|
||||
len = data.len(),
|
||||
"MUXER: [DISPATCH] Sending data"
|
||||
);
|
||||
}
|
||||
|
||||
if let Err(_e) = tx.send(data).await {
|
||||
tracing::debug!(
|
||||
stream_id,
|
||||
"MUXER: [WARN] Local channel closed, dropping packet"
|
||||
);
|
||||
self.remove_stream(stream_id).await;
|
||||
}
|
||||
} else {
|
||||
tracing::trace!(
|
||||
stream_id,
|
||||
"MUXER: [IGNORE] Packet for already closed stream"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
pub mod connection;
|
||||
pub mod network;
|
||||
@@ -0,0 +1,157 @@
|
||||
use crate::{
|
||||
protocol::errors::ErrorAction,
|
||||
proxy::connection::{
|
||||
connection::{Connection, ConnectionRole, BUF_SIZE},
|
||||
engine::TunnelEngine,
|
||||
muxer::Muxer,
|
||||
},
|
||||
tlseng::profile::BrowserProfile,
|
||||
};
|
||||
use bytes::BytesMut;
|
||||
use tokio::{
|
||||
io::{AsyncReadExt, AsyncWriteExt},
|
||||
net::{TcpListener, TcpStream},
|
||||
};
|
||||
use tracing::{error, info, instrument}; // Импортируем макросы
|
||||
|
||||
pub struct Network {
|
||||
host: String,
|
||||
port: u16,
|
||||
role: ConnectionRole,
|
||||
remote_proxy_addr: Option<String>,
|
||||
}
|
||||
|
||||
impl Network {
|
||||
pub fn new(
|
||||
host: String,
|
||||
port: u16,
|
||||
role: ConnectionRole,
|
||||
remote_proxy_addr: Option<String>,
|
||||
) -> Self {
|
||||
Self {
|
||||
host,
|
||||
port,
|
||||
role,
|
||||
remote_proxy_addr,
|
||||
}
|
||||
}
|
||||
|
||||
// Добавляем инструмент, чтобы видеть параметры запуска сети в логах
|
||||
#[instrument(skip(self), fields(role = ?self.role, port = self.port))]
|
||||
pub async fn run(&self) {
|
||||
let addr = format!("{}:{}", self.host, self.port);
|
||||
|
||||
match self.role {
|
||||
ConnectionRole::Client => {
|
||||
info!("Starting Client mode: Initializing persistent tunnel to proxy...");
|
||||
|
||||
let muxer = match self.initialize_client_tunnel().await {
|
||||
Ok(m) => m,
|
||||
Err(e) => {
|
||||
error!(error = %e, "Global tunnel failed. Exit.");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let listener = TcpListener::bind(&addr).await.expect("SOCKS bind failed");
|
||||
info!(socks_addr = %addr, "SOCKS5 ready");
|
||||
|
||||
loop {
|
||||
if let Ok((stream, client_addr)) = listener.accept().await {
|
||||
let current_muxer = muxer.clone();
|
||||
tokio::spawn(async move {
|
||||
// Здесь мы просто создаем Connection и сразу в SOCKS
|
||||
let connection = Connection::new(stream, client_addr, false);
|
||||
let _ = connection.handle_socks_client(current_muxer).await;
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ConnectionRole::Server => {
|
||||
// --- ЛОГИКА СЕРВЕРА ---
|
||||
let listener = TcpListener::bind(&addr)
|
||||
.await
|
||||
.expect("Failed to bind Server port");
|
||||
info!(listen_addr = %addr, "Proxy Server listening for incoming tunnels");
|
||||
|
||||
loop {
|
||||
if let Ok((stream, client_addr)) = listener.accept().await {
|
||||
tokio::spawn(async move {
|
||||
// Сервер использует handle_server_tunnel
|
||||
let connection = Connection::new(stream, client_addr, true);
|
||||
if let Err(e) = connection.handle_server_tunnel().await {
|
||||
error!(client = %client_addr, error = %e, "Tunnel error");
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Вспомогательный метод для Клиента: создает TLS туннель и запускает TunnelEngine
|
||||
pub async fn initialize_client_tunnel(&self) -> Result<Muxer, String> {
|
||||
let server_addr = self.remote_proxy_addr.as_ref().ok_or("No proxy addr")?;
|
||||
|
||||
// Вместо создания Connection (который нужен для обработки клиентов),
|
||||
// работаем напрямую с TcpStream для первичного TLS-хендшейка.
|
||||
let stream = TcpStream::connect(server_addr)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
let (mut inbound, mut outbound) = stream.into_split();
|
||||
|
||||
// Кодек создаем «с чистого листа»
|
||||
let mut codec = crate::protocol::codec::codec::Codec::new(false);
|
||||
|
||||
// --- TLS Handshake ---
|
||||
let ch = codec
|
||||
.make_client_handshake(&BrowserProfile::CHROME_131, "google.com")
|
||||
.map_err(|e| format!("{:?}", e))?;
|
||||
outbound.write_all(&ch).await.map_err(|e| e.to_string())?;
|
||||
|
||||
let mut sh_buf = BytesMut::with_capacity(2048);
|
||||
loop {
|
||||
// Пытаемся обработать то, что уже есть в буфере
|
||||
match codec.process_handshake(&mut sh_buf) {
|
||||
Ok(_) => break, // Готово!
|
||||
Err(e) if e.action == ErrorAction::Wait => {
|
||||
let n = inbound
|
||||
.read_buf(&mut sh_buf)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
if n == 0 {
|
||||
return Err("EOF during handshake".into());
|
||||
}
|
||||
}
|
||||
Err(e) => return Err(format!("TLS error: {:?}", e)),
|
||||
}
|
||||
}
|
||||
|
||||
// --- Запуск инфраструктуры ---
|
||||
let (mux_tx, mux_rx) = tokio::sync::mpsc::channel(BUF_SIZE);
|
||||
let muxer = Muxer::new(mux_tx, true);
|
||||
|
||||
let handler = std::sync::Arc::new(crate::proxy::connection::handler::StreamHandler::new(
|
||||
muxer.clone(),
|
||||
ConnectionRole::Client,
|
||||
));
|
||||
|
||||
let engine = TunnelEngine {
|
||||
inbound,
|
||||
outbound,
|
||||
codec,
|
||||
read_buf: sh_buf, // Передаем остатки данных из буфера хендшейка в движок!
|
||||
mux_rx,
|
||||
handler,
|
||||
};
|
||||
|
||||
tokio::spawn(async move { engine.run().await });
|
||||
|
||||
Ok(muxer)
|
||||
}
|
||||
|
||||
pub fn get_self_local_address(&self) -> String {
|
||||
format!("127.0.0.1:{}", self.port)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user