use anyhow::{Context, Result};
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use wispers_connect::{Node, NodeState};
use tracing::{debug, error, info, warn};
use crate::proxy_common::{
CLEANUP_INTERVAL, ConnectionPool, ProxyError, REQUEST_TIMEOUT, parse_wispers_host, send_command,
};
const SOCKS_VERSION: u8 = 0x05;
const AUTH_NOAUTH: u8 = 0x00;
const CMD_CONNECT: u8 = 0x01;
const ATYP_IPV4: u8 = 0x01;
const ATYP_DOMAIN: u8 = 0x03;
const ATYP_IPV6: u8 = 0x04;
const REP_SUCCESS: u8 = 0x00;
const REP_GENERAL_FAILURE: u8 = 0x01;
const REP_NOT_ALLOWED: u8 = 0x02;
const REP_HOST_UNREACHABLE: u8 = 0x04;
const REP_CONNECTION_REFUSED: u8 = 0x05;
const REP_TTL_EXPIRED: u8 = 0x06;
const REP_COMMAND_NOT_SUPPORTED: u8 = 0x07;
const REP_ADDRESS_TYPE_NOT_SUPPORTED: u8 = 0x08;
pub async fn run(
hub_override: Option<&str>,
profile: &str,
bind_addr: &str,
egress_node: Option<i32>,
) -> Result<()> {
let storage = super::get_storage(hub_override, profile)?;
let node = super::load_node(&storage).await?;
if node.state() != NodeState::Activated {
anyhow::bail!(
"Node must be activated to use SOCKS5 proxy. Current state: {:?}",
node.state()
);
}
let listener = TcpListener::bind(bind_addr)
.await
.with_context(|| format!("failed to bind to {}", bind_addr))?;
println!("SOCKS5 proxy listening on {}", bind_addr);
if let Some(egress) = egress_node {
println!(" Internet egress: enabled via node {}", egress);
println!(
"Example: curl --proxy socks5h://{} https://example.com/",
bind_addr
);
} else {
println!(" Internet egress: disabled (wispers.link only)");
println!(
"Example: curl --proxy socks5h://{} http://3.wispers.link/",
bind_addr
);
}
println!(" (Use socks5h:// so the proxy resolves hostnames, not curl)");
let node = Arc::new(node);
let pool = ConnectionPool::new();
let cleanup_pool = pool.clone();
tokio::spawn(async move {
loop {
tokio::time::sleep(CLEANUP_INTERVAL).await;
cleanup_pool.cleanup_idle().await;
}
});
loop {
match listener.accept().await {
Ok((stream, addr)) => {
debug!(%addr, "Accepted connection");
let node = Arc::clone(&node);
let pool = pool.clone();
tokio::spawn(async move {
if let Err(e) = handle_client_connection(stream, node, pool, egress_node).await
{
warn!(error = %e, "Connection error");
}
});
}
Err(e) => {
error!(error = %e, "Accept error");
}
}
}
}
#[derive(Debug)]
struct ConnectRequest {
host: String,
port: u16,
}
async fn handle_client_connection(
mut stream: TcpStream,
node: Arc<Node>,
pool: ConnectionPool,
egress_node: Option<i32>,
) -> Result<()> {
let peer = stream.peer_addr()?;
if let Err(e) = handle_auth(&mut stream).await {
debug!(%peer, error = %e, "Auth failed");
return Ok(());
}
let request = match handle_connect_request(&mut stream).await {
Ok(req) => req,
Err(e) => {
debug!(%peer, error = %e, "Connect request failed");
return Ok(());
}
};
info!(host = %request.host, port = request.port, "CONNECT");
match route_connection(&mut stream, &node, &pool, &request, egress_node).await {
Ok(()) => {}
Err(e) => {
info!(host = %request.host, error = %e, "Routing failed");
}
}
debug!(%peer, "Connection closed");
Ok(())
}
async fn handle_auth(stream: &mut TcpStream) -> Result<(), ProxyError> {
let mut buf = [0u8; 258]; let n = stream
.read(&mut buf)
.await
.map_err(|e| ProxyError::BadRequest(format!("failed to read auth request: {}", e)))?;
if n < 2 {
return Err(ProxyError::BadRequest("auth request too short".to_string()));
}
let version = buf[0];
if version != SOCKS_VERSION {
return Err(ProxyError::BadRequest(format!(
"unsupported SOCKS version: {}",
version
)));
}
let nmethods = buf[1] as usize;
if n < 2 + nmethods {
return Err(ProxyError::BadRequest("auth request truncated".to_string()));
}
let methods = &buf[2..2 + nmethods];
if !methods.contains(&AUTH_NOAUTH) {
let _ = stream.write_all(&[SOCKS_VERSION, 0xFF]).await;
return Err(ProxyError::BadRequest(
"client does not support NOAUTH".to_string(),
));
}
stream
.write_all(&[SOCKS_VERSION, AUTH_NOAUTH])
.await
.map_err(|e| ProxyError::BadRequest(format!("failed to send auth response: {}", e)))?;
Ok(())
}
async fn handle_connect_request(stream: &mut TcpStream) -> Result<ConnectRequest, ProxyError> {
let mut header = [0u8; 4];
stream
.read_exact(&mut header)
.await
.map_err(|e| ProxyError::BadRequest(format!("failed to read request header: {}", e)))?;
let version = header[0];
let cmd = header[1];
let atyp = header[3];
if version != SOCKS_VERSION {
return Err(ProxyError::BadRequest(format!(
"unsupported SOCKS version: {}",
version
)));
}
if cmd != CMD_CONNECT {
send_reply(stream, REP_COMMAND_NOT_SUPPORTED).await;
return Err(ProxyError::BadRequest(format!(
"unsupported command: {}",
cmd
)));
}
let host = match atyp {
ATYP_IPV4 => {
let mut addr = [0u8; 4];
stream.read_exact(&mut addr).await.map_err(|e| {
ProxyError::BadRequest(format!("failed to read IPv4 address: {}", e))
})?;
format!("{}.{}.{}.{}", addr[0], addr[1], addr[2], addr[3])
}
ATYP_DOMAIN => {
let mut len_buf = [0u8; 1];
stream.read_exact(&mut len_buf).await.map_err(|e| {
ProxyError::BadRequest(format!("failed to read domain length: {}", e))
})?;
let len = len_buf[0] as usize;
let mut domain = vec![0u8; len];
stream
.read_exact(&mut domain)
.await
.map_err(|e| ProxyError::BadRequest(format!("failed to read domain: {}", e)))?;
String::from_utf8(domain)
.map_err(|_| ProxyError::BadRequest("invalid domain name encoding".to_string()))?
}
ATYP_IPV6 => {
let mut addr = [0u8; 16];
stream.read_exact(&mut addr).await.map_err(|e| {
ProxyError::BadRequest(format!("failed to read IPv6 address: {}", e))
})?;
std::net::Ipv6Addr::from(addr).to_string()
}
_ => {
send_reply(stream, REP_ADDRESS_TYPE_NOT_SUPPORTED).await;
return Err(ProxyError::BadRequest(format!(
"unsupported address type: {}",
atyp
)));
}
};
let mut port_buf = [0u8; 2];
stream
.read_exact(&mut port_buf)
.await
.map_err(|e| ProxyError::BadRequest(format!("failed to read port: {}", e)))?;
let port = u16::from_be_bytes(port_buf);
Ok(ConnectRequest { host, port })
}
async fn route_connection(
stream: &mut TcpStream,
node: &Node,
pool: &ConnectionPool,
request: &ConnectRequest,
egress_node: Option<i32>,
) -> Result<()> {
match parse_wispers_host(&request.host) {
Ok(wispers_host) => {
forward_to_node(stream, node, pool, wispers_host.node_number, request.port).await
}
Err(None) => {
match egress_node {
Some(egress) => {
debug!(egress_node = egress, "Routing via egress");
egress_to_node(stream, node, pool, egress, &request.host, request.port).await
}
None => {
warn!(host = %request.host, "Rejected: egress not enabled");
send_reply(stream, REP_NOT_ALLOWED).await;
Err(anyhow::anyhow!("egress not enabled"))
}
}
}
Err(Some(e)) => {
warn!(host = %request.host, error = %e, "Rejected: invalid wispers.link host");
send_reply(stream, REP_GENERAL_FAILURE).await;
Err(anyhow::anyhow!("{}", e))
}
}
}
async fn forward_to_node(
stream: &mut TcpStream,
node: &Node,
pool: &ConnectionPool,
target_node: i32,
port: u16,
) -> Result<()> {
let quic_stream =
match tokio::time::timeout(REQUEST_TIMEOUT, pool.open_stream(node, target_node)).await {
Ok(Ok(s)) => s,
Ok(Err(e)) => {
warn!(target_node, error = %e, "Failed to open stream");
send_reply(stream, REP_HOST_UNREACHABLE).await;
return Err(anyhow::anyhow!("{}", e));
}
Err(_) => {
warn!(target_node, "Timeout opening stream");
send_reply(stream, REP_TTL_EXPIRED).await;
return Err(anyhow::anyhow!("open_stream timeout"));
}
};
let command = format!("FORWARD {}\n", port);
if let Err(e) = send_command(&quic_stream, &command).await {
warn!(target_node, port, error = %e, "FORWARD failed");
send_reply(stream, REP_CONNECTION_REFUSED).await;
return Err(anyhow::anyhow!("{}", e));
}
send_reply(stream, REP_SUCCESS).await;
relay(stream, quic_stream).await;
Ok(())
}
async fn egress_to_node(
stream: &mut TcpStream,
node: &Node,
pool: &ConnectionPool,
egress_node: i32,
host: &str,
port: u16,
) -> Result<()> {
let quic_stream =
match tokio::time::timeout(REQUEST_TIMEOUT, pool.open_stream(node, egress_node)).await {
Ok(Ok(s)) => s,
Ok(Err(e)) => {
warn!(egress_node, error = %e, "Failed to open stream to egress");
send_reply(stream, REP_HOST_UNREACHABLE).await;
return Err(anyhow::anyhow!("{}", e));
}
Err(_) => {
warn!(egress_node, "Timeout opening stream to egress");
send_reply(stream, REP_TTL_EXPIRED).await;
return Err(anyhow::anyhow!("open_stream timeout"));
}
};
let command = format!("CONNECT {}:{}\n", host, port);
if let Err(e) = send_command(&quic_stream, &command).await {
warn!(egress_node, host, port, error = %e, "CONNECT failed");
send_reply(stream, REP_CONNECTION_REFUSED).await;
return Err(anyhow::anyhow!("{}", e));
}
send_reply(stream, REP_SUCCESS).await;
relay(stream, quic_stream).await;
Ok(())
}
async fn relay(tcp_stream: &mut TcpStream, quic_stream: wispers_connect::QuicStream) {
let quic_stream = Arc::new(quic_stream);
let (mut tcp_read, mut tcp_write) = tcp_stream.split();
let quic_read = Arc::clone(&quic_stream);
let quic_write = Arc::clone(&quic_stream);
let tcp_to_quic = async move {
let mut buf = [0u8; 8192];
loop {
match tcp_read.read(&mut buf).await {
Ok(0) => break,
Ok(n) => {
if let Err(e) = quic_write.write_all(&buf[..n]).await {
debug!(error = %e, "QUIC write error");
break;
}
}
Err(e) => {
debug!(error = %e, "TCP read error");
break;
}
}
}
let _ = quic_write.finish().await;
};
let quic_to_tcp = async move {
let mut buf = [0u8; 8192];
loop {
match quic_read.read(&mut buf).await {
Ok(0) => break,
Ok(n) => {
if let Err(e) = tcp_write.write_all(&buf[..n]).await {
debug!(error = %e, "TCP write error");
break;
}
}
Err(e) => {
debug!(error = %e, "QUIC read error");
break;
}
}
}
let _ = tcp_write.shutdown().await;
};
tokio::join!(tcp_to_quic, quic_to_tcp);
}
async fn send_reply(stream: &mut TcpStream, reply_code: u8) {
let reply = [
SOCKS_VERSION, reply_code, 0x00, ATYP_IPV4, 0,
0,
0,
0, 0,
0, ];
let _ = stream.write_all(&reply).await;
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_socks5_constants() {
assert_eq!(SOCKS_VERSION, 0x05);
assert_eq!(AUTH_NOAUTH, 0x00);
assert_eq!(CMD_CONNECT, 0x01);
assert_eq!(ATYP_IPV4, 0x01);
assert_eq!(ATYP_DOMAIN, 0x03);
assert_eq!(ATYP_IPV6, 0x04);
}
#[test]
fn test_reply_codes() {
assert_eq!(REP_SUCCESS, 0x00);
assert_eq!(REP_GENERAL_FAILURE, 0x01);
assert_eq!(REP_NOT_ALLOWED, 0x02);
assert_eq!(REP_CONNECTION_REFUSED, 0x05);
assert_eq!(REP_TTL_EXPIRED, 0x06);
assert_eq!(REP_COMMAND_NOT_SUPPORTED, 0x07);
assert_eq!(REP_ADDRESS_TYPE_NOT_SUPPORTED, 0x08);
}
fn build_auth_request(methods: &[u8]) -> Vec<u8> {
let mut data = vec![SOCKS_VERSION, methods.len() as u8];
data.extend_from_slice(methods);
data
}
fn build_connect_ipv4(ip: [u8; 4], port: u16) -> Vec<u8> {
let mut data = vec![SOCKS_VERSION, CMD_CONNECT, 0x00, ATYP_IPV4];
data.extend_from_slice(&ip);
data.extend_from_slice(&port.to_be_bytes());
data
}
fn build_connect_domain(domain: &str, port: u16) -> Vec<u8> {
let mut data = vec![SOCKS_VERSION, CMD_CONNECT, 0x00, ATYP_DOMAIN];
data.push(domain.len() as u8);
data.extend_from_slice(domain.as_bytes());
data.extend_from_slice(&port.to_be_bytes());
data
}
fn build_connect_ipv6(ip: [u8; 16], port: u16) -> Vec<u8> {
let mut data = vec![SOCKS_VERSION, CMD_CONNECT, 0x00, ATYP_IPV6];
data.extend_from_slice(&ip);
data.extend_from_slice(&port.to_be_bytes());
data
}
#[test]
fn test_auth_response_noauth() {
let expected_response = [SOCKS_VERSION, AUTH_NOAUTH];
assert_eq!(expected_response, [0x05, 0x00]);
}
#[test]
fn test_auth_request_multiple_methods() {
let data = build_auth_request(&[0x02, AUTH_NOAUTH, 0x01]); assert_eq!(data, [0x05, 0x03, 0x02, 0x00, 0x01]);
}
#[test]
fn test_auth_request_structure() {
let req = build_auth_request(&[AUTH_NOAUTH]);
assert_eq!(req[0], SOCKS_VERSION);
assert_eq!(req[1], 1); assert_eq!(req[2], AUTH_NOAUTH);
}
#[test]
fn test_connect_request_ipv4_structure() {
let req = build_connect_ipv4([192, 168, 1, 1], 8080);
assert_eq!(req[0], SOCKS_VERSION);
assert_eq!(req[1], CMD_CONNECT);
assert_eq!(req[2], 0x00); assert_eq!(req[3], ATYP_IPV4);
assert_eq!(&req[4..8], &[192, 168, 1, 1]);
assert_eq!(&req[8..10], &8080u16.to_be_bytes());
}
#[test]
fn test_connect_request_domain_structure() {
let req = build_connect_domain("example.com", 443);
assert_eq!(req[0], SOCKS_VERSION);
assert_eq!(req[1], CMD_CONNECT);
assert_eq!(req[2], 0x00); assert_eq!(req[3], ATYP_DOMAIN);
assert_eq!(req[4], 11); assert_eq!(&req[5..16], b"example.com");
assert_eq!(&req[16..18], &443u16.to_be_bytes());
}
#[test]
fn test_connect_request_wispers_domain() {
let req = build_connect_domain("3.wispers.link", 80);
assert_eq!(req[4], 14); assert_eq!(&req[5..19], b"3.wispers.link");
}
#[test]
fn test_connect_request_ipv6_structure() {
let ip = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1];
let req = build_connect_ipv6(ip, 8080);
assert_eq!(req[0], SOCKS_VERSION);
assert_eq!(req[1], CMD_CONNECT);
assert_eq!(req[2], 0x00); assert_eq!(req[3], ATYP_IPV6);
assert_eq!(&req[4..20], &ip);
assert_eq!(&req[20..22], &8080u16.to_be_bytes());
}
#[test]
fn test_reply_structure() {
let expected_success = [
SOCKS_VERSION, REP_SUCCESS, 0x00, ATYP_IPV4, 0,
0,
0,
0, 0,
0, ];
assert_eq!(expected_success.len(), 10);
assert_eq!(expected_success[0], 0x05);
assert_eq!(expected_success[1], 0x00);
}
#[test]
fn test_reply_error_codes() {
let error_reply = [
SOCKS_VERSION,
REP_CONNECTION_REFUSED,
0x00,
ATYP_IPV4,
0,
0,
0,
0,
0,
0,
];
assert_eq!(error_reply[1], 0x05); }
#[test]
fn test_port_big_endian() {
let port: u16 = 8080;
let bytes = port.to_be_bytes();
assert_eq!(bytes, [0x1F, 0x90]);
let port: u16 = 443;
let bytes = port.to_be_bytes();
assert_eq!(bytes, [0x01, 0xBB]);
let port: u16 = 80;
let bytes = port.to_be_bytes();
assert_eq!(bytes, [0x00, 0x50]);
}
#[test]
fn test_unsupported_commands() {
const CMD_BIND: u8 = 0x02;
const CMD_UDP_ASSOCIATE: u8 = 0x03;
assert_ne!(CMD_BIND, CMD_CONNECT);
assert_ne!(CMD_UDP_ASSOCIATE, CMD_CONNECT);
}
#[test]
fn test_ipv4_address_formatting() {
let addr = [192u8, 168, 1, 1];
let formatted = format!("{}.{}.{}.{}", addr[0], addr[1], addr[2], addr[3]);
assert_eq!(formatted, "192.168.1.1");
let addr = [127u8, 0, 0, 1];
let formatted = format!("{}.{}.{}.{}", addr[0], addr[1], addr[2], addr[3]);
assert_eq!(formatted, "127.0.0.1");
}
#[test]
fn test_ipv6_address_formatting() {
use std::net::Ipv6Addr;
let addr = [0u8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1];
let formatted = Ipv6Addr::from(addr).to_string();
assert_eq!(formatted, "::1");
let addr = [0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1];
let formatted = Ipv6Addr::from(addr).to_string();
assert_eq!(formatted, "2001:db8::1");
let addr = [0xfe, 0x80, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1];
let formatted = Ipv6Addr::from(addr).to_string();
assert_eq!(formatted, "fe80::1");
let addr = [
0x20, 0x01, 0x0d, 0xb8, 0x85, 0xa3, 0x00, 0x00, 0x00, 0x00, 0x8a, 0x2e, 0x03, 0x70,
0x73, 0x34,
];
let formatted = Ipv6Addr::from(addr).to_string();
assert_eq!(formatted, "2001:db8:85a3::8a2e:370:7334");
}
#[test]
fn test_wispers_domain_parsing() {
let result = parse_wispers_host("3.wispers.link");
assert!(result.is_ok());
assert_eq!(result.unwrap().node_number, 3);
let result = parse_wispers_host("123.wispers.link");
assert!(result.is_ok());
assert_eq!(result.unwrap().node_number, 123);
}
#[test]
fn test_non_wispers_domain() {
let result = parse_wispers_host("example.com");
assert!(result.is_err());
assert!(result.unwrap_err().is_none());
let result = parse_wispers_host("google.com");
assert!(result.is_err());
assert!(result.unwrap_err().is_none());
}
#[test]
fn test_invalid_wispers_domain() {
let result = parse_wispers_host("abc.wispers.link");
assert!(result.is_err());
assert!(result.unwrap_err().is_some()); }
}