1.0.1.1

Drafting Upstreams and SOCKS

src/transport/mod.rs

@@ -3,7 +3,11 @@
 pub mod pool;
 pub mod proxy_protocol;
 pub mod socket;
+pub mod socks;
+pub mod upstream;
 
 pub use pool::ConnectionPool;
 pub use proxy_protocol::{ProxyProtocolInfo, parse_proxy_protocol};
-pub use socket::*;
+pub use socket::*;
+pub use socks::*;
+pub use upstream::UpstreamManager;

src/transport/socket.rs

@@ -1,7 +1,7 @@
 //! TCP Socket Configuration
 
 use std::io::Result;
-use std::net::SocketAddr;
+use std::net::{SocketAddr, IpAddr};
 use std::time::Duration;
 use tokio::net::TcpStream;
 use socket2::{Socket, TcpKeepalive, Domain, Type, Protocol};
@@ -93,6 +93,11 @@ pub fn set_linger_zero(stream: &TcpStream) -> Result<()> {
 
 /// Create a new TCP socket for outgoing connections
 pub fn create_outgoing_socket(addr: SocketAddr) -> Result<Socket> {
+    create_outgoing_socket_bound(addr, None)
+}
+
+/// Create a new TCP socket for outgoing connections, optionally bound to a specific interface
+pub fn create_outgoing_socket_bound(addr: SocketAddr, bind_addr: Option<IpAddr>) -> Result<Socket> {
     let domain = if addr.is_ipv4() {
         Domain::IPV4
     } else {
@@ -106,10 +111,17 @@ pub fn create_outgoing_socket(addr: SocketAddr) -> Result<Socket> {
     
     // Disable Nagle
     socket.set_nodelay(true)?;
+
+    if let Some(bind_ip) = bind_addr {
+        let bind_sock_addr = SocketAddr::new(bind_ip, 0);
+        socket.bind(&bind_sock_addr.into())?;
+        debug!("Bound outgoing socket to {}", bind_ip);
+    }
     
     Ok(socket)
 }
 
+
 /// Get local address of a socket
 pub fn get_local_addr(stream: &TcpStream) -> Option<SocketAddr> {
     stream.local_addr().ok()

src/transport/socks.rs

@@ -0,0 +1,145 @@
+//! SOCKS4/5 Client Implementation
+
+use std::net::{IpAddr, SocketAddr};
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use tokio::net::TcpStream;
+use crate::error::{ProxyError, Result};
+
+pub async fn connect_socks4(
+    stream: &mut TcpStream,
+    target: SocketAddr,
+    user_id: Option<&str>,
+) -> Result<()> {
+    let ip = match target.ip() {
+        IpAddr::V4(ip) => ip,
+        IpAddr::V6(_) => return Err(ProxyError::Proxy("SOCKS4 does not support IPv6".to_string())),
+    };
+    
+    let port = target.port();
+    let user = user_id.unwrap_or("").as_bytes();
+    
+    // VN (4) | CD (1) | DSTPORT (2) | DSTIP (4) | USERID (variable) | NULL (1)
+    let mut buf = Vec::with_capacity(9 + user.len());
+    buf.push(4); // VN
+    buf.push(1); // CD (CONNECT)
+    buf.extend_from_slice(&port.to_be_bytes());
+    buf.extend_from_slice(&ip.octets());
+    buf.extend_from_slice(user);
+    buf.push(0); // NULL
+    
+    stream.write_all(&buf).await.map_err(|e| ProxyError::Io(e))?;
+    
+    // Response: VN (1) | CD (1) | DSTPORT (2) | DSTIP (4)
+    let mut resp = [0u8; 8];
+    stream.read_exact(&mut resp).await.map_err(|e| ProxyError::Io(e))?;
+    
+    if resp[1] != 90 {
+        return Err(ProxyError::Proxy(format!("SOCKS4 request rejected: code {}", resp[1])));
+    }
+    
+    Ok(())
+}
+
+pub async fn connect_socks5(
+    stream: &mut TcpStream,
+    target: SocketAddr,
+    username: Option<&str>,
+    password: Option<&str>,
+) -> Result<()> {
+    // 1. Auth negotiation
+    // VER (1) | NMETHODS (1) | METHODS (variable)
+    let mut methods = vec![0u8]; // No auth
+    if username.is_some() {
+        methods.push(2u8); // Username/Password
+    }
+    
+    let mut buf = vec![5u8, methods.len() as u8];
+    buf.extend_from_slice(&methods);
+    
+    stream.write_all(&buf).await.map_err(|e| ProxyError::Io(e))?;
+    
+    let mut resp = [0u8; 2];
+    stream.read_exact(&mut resp).await.map_err(|e| ProxyError::Io(e))?;
+    
+    if resp[0] != 5 {
+        return Err(ProxyError::Proxy("Invalid SOCKS5 version".to_string()));
+    }
+    
+    match resp[1] {
+        0 => {}, // No auth
+        2 => {
+            // Username/Password auth
+            if let (Some(u), Some(p)) = (username, password) {
+                let u_bytes = u.as_bytes();
+                let p_bytes = p.as_bytes();
+                
+                let mut auth_buf = Vec::with_capacity(3 + u_bytes.len() + p_bytes.len());
+                auth_buf.push(1); // VER
+                auth_buf.push(u_bytes.len() as u8);
+                auth_buf.extend_from_slice(u_bytes);
+                auth_buf.push(p_bytes.len() as u8);
+                auth_buf.extend_from_slice(p_bytes);
+                
+                stream.write_all(&auth_buf).await.map_err(|e| ProxyError::Io(e))?;
+                
+                let mut auth_resp = [0u8; 2];
+                stream.read_exact(&mut auth_resp).await.map_err(|e| ProxyError::Io(e))?;
+                
+                if auth_resp[1] != 0 {
+                    return Err(ProxyError::Proxy("SOCKS5 authentication failed".to_string()));
+                }
+            } else {
+                return Err(ProxyError::Proxy("SOCKS5 server requires authentication".to_string()));
+            }
+        },
+        _ => return Err(ProxyError::Proxy("Unsupported SOCKS5 auth method".to_string())),
+    }
+    
+    // 2. Connection request
+    // VER (1) | CMD (1) | RSV (1) | ATYP (1) | DST.ADDR (variable) | DST.PORT (2)
+    let mut req = vec![5u8, 1u8, 0u8]; // CONNECT
+    
+    match target {
+        SocketAddr::V4(v4) => {
+            req.push(1u8); // IPv4
+            req.extend_from_slice(&v4.ip().octets());
+        },
+        SocketAddr::V6(v6) => {
+            req.push(4u8); // IPv6
+            req.extend_from_slice(&v6.ip().octets());
+        },
+    }
+    
+    req.extend_from_slice(&target.port().to_be_bytes());
+    
+    stream.write_all(&req).await.map_err(|e| ProxyError::Io(e))?;
+    
+    // Response
+    let mut head = [0u8; 4];
+    stream.read_exact(&mut head).await.map_err(|e| ProxyError::Io(e))?;
+    
+    if head[1] != 0 {
+        return Err(ProxyError::Proxy(format!("SOCKS5 request failed: code {}", head[1])));
+    }
+    
+    // Skip address part of response
+    match head[3] {
+        1 => { // IPv4
+            let mut addr = [0u8; 4 + 2];
+            stream.read_exact(&mut addr).await.map_err(|e| ProxyError::Io(e))?;
+        },
+        3 => { // Domain
+            let mut len = [0u8; 1];
+            stream.read_exact(&mut len).await.map_err(|e| ProxyError::Io(e))?;
+            let mut addr = vec![0u8; len[0] as usize + 2];
+            stream.read_exact(&mut addr).await.map_err(|e| ProxyError::Io(e))?;
+        },
+        4 => { // IPv6
+            let mut addr = [0u8; 16 + 2];
+            stream.read_exact(&mut addr).await.map_err(|e| ProxyError::Io(e))?;
+        },
+        _ => return Err(ProxyError::Proxy("Invalid address type in SOCKS5 response".to_string())),
+    }
+    
+    Ok(())
+}

src/transport/upstream.rs

@@ -0,0 +1,255 @@
+//! Upstream Management
+
+use std::net::{SocketAddr, IpAddr};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::net::TcpStream;
+use tokio::sync::RwLock;
+use rand::Rng;
+use tracing::{debug, warn, error};
+
+use crate::config::{UpstreamConfig, UpstreamType};
+use crate::error::{Result, ProxyError};
+use crate::transport::socket::create_outgoing_socket_bound;
+use crate::transport::socks::{connect_socks4, connect_socks5};
+
+#[derive(Debug)]
+struct UpstreamState {
+    config: UpstreamConfig,
+    healthy: bool,
+    fails: u32,
+    last_check: std::time::Instant,
+}
+
+#[derive(Clone)]
+pub struct UpstreamManager {
+    upstreams: Arc<RwLock<Vec<UpstreamState>>>,
+}
+
+impl UpstreamManager {
+    pub fn new(configs: Vec<UpstreamConfig>) -> Self {
+        let states = configs.into_iter()
+            .filter(|c| c.enabled)
+            .map(|c| UpstreamState {
+                config: c,
+                healthy: true, // Optimistic start
+                fails: 0,
+                last_check: std::time::Instant::now(),
+            })
+            .collect();
+            
+        Self {
+            upstreams: Arc::new(RwLock::new(states)),
+        }
+    }
+    
+    /// Select an upstream using Weighted Round Robin (simplified)
+    async fn select_upstream(&self) -> Option<usize> {
+        let upstreams = self.upstreams.read().await;
+        if upstreams.is_empty() {
+            return None;
+        }
+
+        let healthy_indices: Vec<usize> = upstreams.iter()
+            .enumerate()
+            .filter(|(_, u)| u.healthy)
+            .map(|(i, _)| i)
+            .collect();
+            
+        if healthy_indices.is_empty() {
+            // If all unhealthy, try any random one
+            return Some(rand::thread_rng().gen_range(0..upstreams.len()));
+        }
+        
+        // Weighted selection
+        let total_weight: u32 = healthy_indices.iter()
+            .map(|&i| upstreams[i].config.weight as u32)
+            .sum();
+            
+        if total_weight == 0 {
+            return Some(healthy_indices[rand::thread_rng().gen_range(0..healthy_indices.len())]);
+        }
+        
+        let mut choice = rand::thread_rng().gen_range(0..total_weight);
+        
+        for &idx in &healthy_indices {
+            let weight = upstreams[idx].config.weight as u32;
+            if choice < weight {
+                return Some(idx);
+            }
+            choice -= weight;
+        }
+        
+        Some(healthy_indices[0])
+    }
+    
+    pub async fn connect(&self, target: SocketAddr) -> Result<TcpStream> {
+        let idx = self.select_upstream().await
+            .ok_or_else(|| ProxyError::Config("No upstreams available".to_string()))?;
+            
+        let upstream = {
+            let guard = self.upstreams.read().await;
+            guard[idx].config.clone()
+        };
+        
+        match self.connect_via_upstream(&upstream, target).await {
+            Ok(stream) => {
+                // Mark success
+                let mut guard = self.upstreams.write().await;
+                if let Some(u) = guard.get_mut(idx) {
+                    if !u.healthy {
+                        debug!("Upstream recovered: {:?}", u.config);
+                    }
+                    u.healthy = true;
+                    u.fails = 0;
+                }
+                Ok(stream)
+            },
+            Err(e) => {
+                // Mark failure
+                let mut guard = self.upstreams.write().await;
+                if let Some(u) = guard.get_mut(idx) {
+                    u.fails += 1;
+                    warn!("Failed to connect via upstream {:?}: {}. Fails: {}", u.config, e, u.fails);
+                    if u.fails > 3 {
+                        u.healthy = false;
+                        warn!("Upstream disabled due to failures: {:?}", u.config);
+                    }
+                }
+                Err(e)
+            }
+        }
+    }
+    
+    async fn connect_via_upstream(&self, config: &UpstreamConfig, target: SocketAddr) -> Result<TcpStream> {
+        match &config.upstream_type {
+            UpstreamType::Direct { interface } => {
+                let bind_ip = interface.as_ref()
+                    .and_then(|s| s.parse::<IpAddr>().ok());
+                
+                let socket = create_outgoing_socket_bound(target, bind_ip)?;
+                
+                // Non-blocking connect logic
+                socket.set_nonblocking(true)?;
+                match socket.connect(&target.into()) {
+                    Ok(()) => {},
+                    Err(err) if err.raw_os_error() == Some(115) || err.kind() == std::io::ErrorKind::WouldBlock => {},
+                    Err(err) => return Err(ProxyError::Io(err)),
+                }
+                
+                let std_stream: std::net::TcpStream = socket.into();
+                let stream = TcpStream::from_std(std_stream)?;
+                
+                // Wait for connection to complete
+                stream.writable().await?;
+                if let Some(e) = stream.take_error()? {
+                    return Err(ProxyError::Io(e));
+                }
+                
+                Ok(stream)
+            },
+            UpstreamType::Socks4 { address, interface, user_id } => {
+                let proxy_addr: SocketAddr = address.parse()
+                    .map_err(|_| ProxyError::Config("Invalid SOCKS4 address".to_string()))?;
+                    
+                let bind_ip = interface.as_ref()
+                    .and_then(|s| s.parse::<IpAddr>().ok());
+                
+                let socket = create_outgoing_socket_bound(proxy_addr, bind_ip)?;
+                
+                // Non-blocking connect logic
+                socket.set_nonblocking(true)?;
+                match socket.connect(&proxy_addr.into()) {
+                    Ok(()) => {},
+                    Err(err) if err.raw_os_error() == Some(115) || err.kind() == std::io::ErrorKind::WouldBlock => {},
+                    Err(err) => return Err(ProxyError::Io(err)),
+                }
+                
+                let std_stream: std::net::TcpStream = socket.into();
+                let mut stream = TcpStream::from_std(std_stream)?;
+                
+                // Wait for connection to complete
+                stream.writable().await?;
+                if let Some(e) = stream.take_error()? {
+                    return Err(ProxyError::Io(e));
+                }
+                
+                connect_socks4(&mut stream, target, user_id.as_deref()).await?;
+                Ok(stream)
+            },
+            UpstreamType::Socks5 { address, interface, username, password } => {
+                let proxy_addr: SocketAddr = address.parse()
+                    .map_err(|_| ProxyError::Config("Invalid SOCKS5 address".to_string()))?;
+                    
+                let bind_ip = interface.as_ref()
+                    .and_then(|s| s.parse::<IpAddr>().ok());
+                
+                let socket = create_outgoing_socket_bound(proxy_addr, bind_ip)?;
+                
+                // Non-blocking connect logic
+                socket.set_nonblocking(true)?;
+                match socket.connect(&proxy_addr.into()) {
+                    Ok(()) => {},
+                    Err(err) if err.raw_os_error() == Some(115) || err.kind() == std::io::ErrorKind::WouldBlock => {},
+                    Err(err) => return Err(ProxyError::Io(err)),
+                }
+                
+                let std_stream: std::net::TcpStream = socket.into();
+                let mut stream = TcpStream::from_std(std_stream)?;
+                
+                // Wait for connection to complete
+                stream.writable().await?;
+                if let Some(e) = stream.take_error()? {
+                    return Err(ProxyError::Io(e));
+                }
+                
+                connect_socks5(&mut stream, target, username.as_deref(), password.as_deref()).await?;
+                Ok(stream)
+            },
+        }
+    }
+    
+    /// Background task to check health
+    pub async fn run_health_checks(&self) {
+        // Simple TCP connect check to a known stable DC (e.g. 149.154.167.50:443 - DC2)
+        let check_target: SocketAddr = "149.154.167.50:443".parse().unwrap();
+        
+        loop {
+            tokio::time::sleep(Duration::from_secs(60)).await;
+            
+            let count = self.upstreams.read().await.len();
+            for i in 0..count {
+                let config = {
+                    let guard = self.upstreams.read().await;
+                    guard[i].config.clone()
+                };
+                
+                let result = tokio::time::timeout(
+                    Duration::from_secs(10),
+                    self.connect_via_upstream(&config, check_target)
+                ).await;
+                
+                let mut guard = self.upstreams.write().await;
+                let u = &mut guard[i];
+                
+                match result {
+                    Ok(Ok(_stream)) => {
+                        if !u.healthy {
+                            debug!("Upstream recovered: {:?}", u.config);
+                        }
+                        u.healthy = true;
+                        u.fails = 0;
+                    }
+                    Ok(Err(e)) => {
+                        debug!("Health check failed for {:?}: {}", u.config, e);
+                        // Don't mark unhealthy immediately in background check
+                    }
+                    Err(_) => {
+                        debug!("Health check timeout for {:?}", u.config);
+                    }
+                }
+                u.last_check = std::time::Instant::now();
+            }
+        }
+    }
+}