395 lines
15 KiB
Rust
395 lines
15 KiB
Rust
use std::{
|
|
fs, net::SocketAddr, os::unix::fs::PermissionsExt, path::Path, process::Command, sync::Arc,
|
|
time::Duration,
|
|
};
|
|
|
|
use rustls::pki_types::pem::PemObject;
|
|
use tempfile::TempDir;
|
|
use tokio::{
|
|
net::UdpSocket,
|
|
sync::mpsc,
|
|
time::{sleep, timeout},
|
|
};
|
|
use wgcloak::{
|
|
client,
|
|
config::{self, Config},
|
|
control, server, stats, transport,
|
|
transport::{Http3ClientConnector, TransportConnector, TransportSession},
|
|
};
|
|
|
|
const SECRET_ONE: &str = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
|
|
const SECRET_TWO: &str = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
|
|
const WRONG_SECRET: &str = "CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC";
|
|
|
|
fn free_udp_port() -> u16 {
|
|
let socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
|
|
socket.local_addr().unwrap().port()
|
|
}
|
|
|
|
fn private_runtime_dir(root: &Path) -> std::path::PathBuf {
|
|
let path = root.join("runtime");
|
|
fs::create_dir(&path).unwrap();
|
|
fs::set_permissions(&path, fs::Permissions::from_mode(0o700)).unwrap();
|
|
path
|
|
}
|
|
|
|
fn server_config(
|
|
directory: &TempDir,
|
|
backend: SocketAddr,
|
|
direct_port: u16,
|
|
http3_port: u16,
|
|
) -> config::ServerConfig {
|
|
let certificate = directory.path().join("server-cert.pem");
|
|
let private_key = directory.path().join("server-key.pem");
|
|
let generated = Command::new("openssl")
|
|
.args(["req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout"])
|
|
.arg(&private_key)
|
|
.args(["-out"])
|
|
.arg(&certificate)
|
|
.args([
|
|
"-subj",
|
|
"/CN=127.0.0.1",
|
|
"-addext",
|
|
"subjectAltName=IP:127.0.0.1",
|
|
"-addext",
|
|
"basicConstraints=critical,CA:FALSE",
|
|
"-addext",
|
|
"keyUsage=critical,digitalSignature,keyEncipherment",
|
|
"-addext",
|
|
"extendedKeyUsage=serverAuth",
|
|
"-days",
|
|
"1",
|
|
])
|
|
.output()
|
|
.expect("openssl is required for the HTTP/3 runtime integration test");
|
|
assert!(
|
|
generated.status.success(),
|
|
"openssl certificate generation failed: {}",
|
|
String::from_utf8_lossy(&generated.stderr)
|
|
);
|
|
|
|
Config::from_str(&format!(
|
|
"[server]\nlisten='127.0.0.1'\n[backend]\ntarget='{backend}'\n[transports.direct_udp]\nenabled=true\nport={direct_port}\n[transports.quic]\nenabled=false\nport={}\n[transports.http3]\nenabled=true\nport={http3_port}\npath='/tunnel'\n[transports.tls_tcp]\nenabled=true\nport={direct_port}\n[tls]\ncertificate='{}'\nprivate_key='{}'\n[[auth.clients]]\nid='one'\nsecret='{SECRET_ONE}'\n[[auth.clients]]\nid='two'\nsecret='{SECRET_TWO}'\n[limits]\nhandshake_timeout_ms=3000\n",
|
|
free_udp_port(),
|
|
certificate.display(),
|
|
private_key.display()
|
|
))
|
|
.unwrap()
|
|
.into_server()
|
|
.unwrap()
|
|
}
|
|
|
|
fn client_config(
|
|
id: &str,
|
|
secret: &str,
|
|
http3_port: u16,
|
|
direct_probe_port: u16,
|
|
local_port: u16,
|
|
) -> config::ClientConfig {
|
|
Config::from_str(&format!(
|
|
"[client]\nserver='127.0.0.1'\nid='{id}'\nsecret='{secret}'\n[local]\nlisten='127.0.0.1:{local_port}'\n[transport]\nmode='auto'\n[transport.auto]\norder=['direct_udp','http3']\nprobe_timeout_ms=700\nretries=1\n[transport.direct_udp]\nport={direct_probe_port}\n[transport.http3]\nport={http3_port}\npath='/tunnel'\n[network]\nmax_datagram_size=1200\n"
|
|
))
|
|
.unwrap()
|
|
.into_client()
|
|
.unwrap()
|
|
}
|
|
|
|
fn test_client_tls(certificate: &[u8]) -> quinn::ClientConfig {
|
|
let _ = rustls::crypto::ring::default_provider().install_default();
|
|
let certs: Vec<_> = rustls::pki_types::CertificateDer::pem_slice_iter(certificate)
|
|
.collect::<Result<_, _>>()
|
|
.unwrap();
|
|
let mut roots = rustls::RootCertStore::empty();
|
|
roots.add(certs[0].clone()).unwrap();
|
|
let mut tls = rustls::ClientConfig::builder_with_provider(Arc::new(
|
|
rustls::crypto::ring::default_provider(),
|
|
))
|
|
.with_protocol_versions(&[&rustls::version::TLS13])
|
|
.unwrap()
|
|
.with_root_certificates(roots)
|
|
.with_no_client_auth();
|
|
tls.alpn_protocols = vec![b"h3".to_vec()];
|
|
let crypto = quinn::crypto::rustls::QuicClientConfig::try_from(tls).unwrap();
|
|
quinn::ClientConfig::new(Arc::new(crypto))
|
|
}
|
|
|
|
async fn wait_control_contains(path: &std::path::Path, command: &str, needle: &str) {
|
|
let mut last_response = None;
|
|
let result = timeout(Duration::from_secs(6), async {
|
|
loop {
|
|
if let Ok(response) = control::send_command(path, command).await {
|
|
if response.contains(needle) {
|
|
return;
|
|
}
|
|
last_response = Some(response);
|
|
}
|
|
sleep(Duration::from_millis(40)).await;
|
|
}
|
|
})
|
|
.await;
|
|
assert!(
|
|
result.is_ok(),
|
|
"control {command} did not contain {needle:?}; latest response: {last_response:?}"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn auto_http3_runtime_relays_isolated_clients_and_rejects_bad_secret() {
|
|
if Command::new("openssl").arg("version").output().is_err() {
|
|
return;
|
|
}
|
|
let directory = tempfile::tempdir().unwrap();
|
|
let runtime_dir = private_runtime_dir(directory.path());
|
|
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
|
|
let backend_addr = backend.local_addr().unwrap();
|
|
let (peer_tx, mut peer_rx) = mpsc::unbounded_channel();
|
|
let backend_task = tokio::spawn(async move {
|
|
let mut packet = vec![0u8; u16::MAX as usize];
|
|
while let Ok((len, peer)) = backend.recv_from(&mut packet).await {
|
|
let _ = peer_tx.send(peer);
|
|
let _ = backend.send_to(&packet[..len], peer).await;
|
|
}
|
|
});
|
|
|
|
let certificate = directory.path().join("server-cert.pem");
|
|
let direct_server_port = free_udp_port();
|
|
let http3_port = free_udp_port();
|
|
let server = server_config(&directory, backend_addr, direct_server_port, http3_port);
|
|
let server_control = runtime_dir.join("server.sock");
|
|
let server_control_for_task = server_control.clone();
|
|
let server_task = tokio::spawn(server::run_with_control_path(
|
|
server,
|
|
server_control_for_task,
|
|
));
|
|
wait_control_contains(
|
|
&server_control,
|
|
"status",
|
|
"transports=direct_udp,http3,tls_tcp",
|
|
)
|
|
.await;
|
|
|
|
let certificate_bytes = fs::read(certificate).unwrap();
|
|
let client_tls = test_client_tls(&certificate_bytes);
|
|
let direct_probe_port = free_udp_port();
|
|
assert_ne!(direct_probe_port, direct_server_port);
|
|
|
|
let local_one = free_udp_port();
|
|
let local_two = free_udp_port();
|
|
let app_one = UdpSocket::bind("127.0.0.1:0").await.unwrap();
|
|
let app_two = UdpSocket::bind("127.0.0.1:0").await.unwrap();
|
|
let client_one_config =
|
|
client_config("one", SECRET_ONE, http3_port, direct_probe_port, local_one);
|
|
let client_two_config =
|
|
client_config("two", SECRET_TWO, http3_port, direct_probe_port, local_two);
|
|
let client_one_control = runtime_dir.join("client-one.sock");
|
|
let client_two_control = runtime_dir.join("client-two.sock");
|
|
let client_one_task = tokio::spawn(client::run_with_http3_connector_and_control_path(
|
|
client_one_config.clone(),
|
|
client_one_control.clone(),
|
|
Some(Arc::new(Http3ClientConnector::with_client_config(
|
|
client_one_config,
|
|
client_tls.clone(),
|
|
)) as Arc<dyn TransportConnector>),
|
|
));
|
|
let client_two_task = tokio::spawn(client::run_with_http3_connector_and_control_path(
|
|
client_two_config.clone(),
|
|
client_two_control.clone(),
|
|
Some(Arc::new(Http3ClientConnector::with_client_config(
|
|
client_two_config,
|
|
client_tls.clone(),
|
|
)) as Arc<dyn TransportConnector>),
|
|
));
|
|
|
|
wait_control_contains(&server_control, "status", "active_sessions=2").await;
|
|
app_one
|
|
.send_to(b"client-one", ("127.0.0.1", local_one))
|
|
.await
|
|
.unwrap();
|
|
app_two
|
|
.send_to(b"client-two", ("127.0.0.1", local_two))
|
|
.await
|
|
.unwrap();
|
|
let mut response = [0u8; 64];
|
|
let (len_one, _) = timeout(Duration::from_secs(4), app_one.recv_from(&mut response))
|
|
.await
|
|
.expect("first HTTP/3 echo timed out")
|
|
.unwrap();
|
|
assert_eq!(&response[..len_one], b"client-one");
|
|
let (len_two, _) = timeout(Duration::from_secs(4), app_two.recv_from(&mut response))
|
|
.await
|
|
.expect("second HTTP/3 echo timed out")
|
|
.unwrap();
|
|
assert_eq!(&response[..len_two], b"client-two");
|
|
let peer_one = timeout(Duration::from_secs(2), peer_rx.recv())
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
let peer_two = timeout(Duration::from_secs(2), peer_rx.recv())
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
assert_ne!(
|
|
peer_one, peer_two,
|
|
"each tunnel must own its backend UDP socket"
|
|
);
|
|
|
|
let bad_local = free_udp_port();
|
|
let bad_client_config = client_config(
|
|
"one",
|
|
WRONG_SECRET,
|
|
http3_port,
|
|
direct_probe_port,
|
|
bad_local,
|
|
);
|
|
let bad_control = runtime_dir.join("client-bad.sock");
|
|
let bad_task = tokio::spawn(client::run_with_http3_connector_and_control_path(
|
|
bad_client_config.clone(),
|
|
bad_control.clone(),
|
|
Some(Arc::new(Http3ClientConnector::with_client_config(
|
|
bad_client_config,
|
|
client_tls,
|
|
)) as Arc<dyn TransportConnector>),
|
|
));
|
|
let bad_result = timeout(Duration::from_secs(6), bad_task)
|
|
.await
|
|
.expect("bad-secret client should finish its rejected startup")
|
|
.unwrap();
|
|
assert!(bad_result.is_err(), "wrong project secret must be rejected");
|
|
wait_control_contains(&server_control, "stats", "\"authentication_rejections\":1").await;
|
|
let aggregate = control::send_command(&server_control, "stats")
|
|
.await
|
|
.unwrap();
|
|
assert!(aggregate.contains("\"transport\":\"multi\""));
|
|
assert!(aggregate.contains("\"direct_udp\":{"));
|
|
assert!(aggregate.contains("\"http3\":{"));
|
|
assert!(aggregate.contains("\"tls_tcp\":{"));
|
|
|
|
for path in [&client_one_control, &client_two_control] {
|
|
control::send_command(path, "shutdown").await.unwrap();
|
|
}
|
|
timeout(Duration::from_secs(4), client_one_task)
|
|
.await
|
|
.unwrap()
|
|
.unwrap()
|
|
.unwrap();
|
|
timeout(Duration::from_secs(4), client_two_task)
|
|
.await
|
|
.unwrap()
|
|
.unwrap()
|
|
.unwrap();
|
|
control::send_command(&server_control, "shutdown")
|
|
.await
|
|
.unwrap();
|
|
timeout(Duration::from_secs(5), server_task)
|
|
.await
|
|
.unwrap()
|
|
.unwrap()
|
|
.unwrap();
|
|
backend_task.abort();
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn conflicting_udp_listeners_are_rejected() {
|
|
let parsed = Config::from_str(&format!(
|
|
"[server]\nlisten='127.0.0.1'\n[backend]\ntarget='127.0.0.1:9'\n[transports.direct_udp]\nenabled=true\nport=54001\n[transports.quic]\nenabled=true\nport=54001\n[transports.http3]\nenabled=false\n[transports.tls_tcp]\nenabled=true\nport=54001\n[tls]\ncertificate='placeholder-cert'\nprivate_key='placeholder-key'\n[[auth.clients]]\nid='one'\nsecret='{SECRET_ONE}'\n"
|
|
))
|
|
.unwrap()
|
|
.into_server()
|
|
.unwrap();
|
|
let directory = tempfile::tempdir().unwrap();
|
|
let runtime_dir = private_runtime_dir(directory.path());
|
|
// Validation happens before binding any listener and reports only the
|
|
// incompatible same-protocol UDP bind; TCP is allowed to reuse the number.
|
|
let result = server::run_with_control_path(parsed, runtime_dir.join("conflict.sock")).await;
|
|
assert!(result.unwrap_err().to_string().contains("UDP listeners"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn control_stats_and_status_merge_multiple_transport_snapshots() {
|
|
let directory = tempfile::tempdir().unwrap();
|
|
let path = private_runtime_dir(directory.path()).join("aggregate.sock");
|
|
let state = control::ControlState::default();
|
|
state
|
|
.set_status("listening; transports=direct_udp,http3")
|
|
.await;
|
|
let direct = stats::StatsSnapshot {
|
|
datagrams_sent: 2,
|
|
bytes_sent: 40,
|
|
..stats::StatsSnapshot::default()
|
|
};
|
|
let http3 = stats::StatsSnapshot {
|
|
datagrams_received: 3,
|
|
authentication_rejections: 1,
|
|
..stats::StatsSnapshot::default()
|
|
};
|
|
state.set_transport_stats("direct_udp", 1, direct).await;
|
|
state.set_transport_stats("http3", 2, http3).await;
|
|
let server = control::ControlServer::bind(&path, state.clone())
|
|
.await
|
|
.unwrap();
|
|
let task = tokio::spawn(server.run());
|
|
|
|
let status = control::send_command(&path, "status").await.unwrap();
|
|
assert!(status.contains("active_sessions=3"));
|
|
assert!(status.contains("active_sessions_by_transport=direct_udp:1,http3:2"));
|
|
let response = control::send_command(&path, "stats").await.unwrap();
|
|
assert!(response.contains("\"datagrams_sent\":2"));
|
|
assert!(response.contains("\"bytes_sent\":40"));
|
|
assert!(response.contains("\"datagrams_received\":3"));
|
|
assert!(response.contains("\"authentication_rejections\":1"));
|
|
assert!(response.contains("\"direct_udp\":{"));
|
|
assert!(response.contains("\"http3\":{"));
|
|
|
|
state.shutdown.cancel();
|
|
timeout(Duration::from_secs(1), task)
|
|
.await
|
|
.unwrap()
|
|
.unwrap()
|
|
.unwrap();
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn direct_udp_runtime_cancellation_runs_session_cleanup() {
|
|
let port = free_udp_port();
|
|
let server = Config::from_str(&format!(
|
|
"[server]\nlisten='127.0.0.1'\n[backend]\ntarget='127.0.0.1:9'\n[transports.direct_udp]\nenabled=true\nport={port}\n[transports.quic]\nenabled=false\n[transports.http3]\nenabled=false\n[transports.tls_tcp]\nenabled=false\n[[auth.clients]]\nid='lifecycle'\nsecret='{SECRET_ONE}'\n"
|
|
))
|
|
.unwrap()
|
|
.into_server()
|
|
.unwrap();
|
|
let client = Config::from_str(&format!(
|
|
"[client]\nserver='127.0.0.1'\nid='lifecycle'\nsecret='{SECRET_ONE}'\n[transport]\nmode='direct_udp'\n[transport.direct_udp]\nport={port}\n"
|
|
))
|
|
.unwrap()
|
|
.into_client()
|
|
.unwrap();
|
|
|
|
let listener = transport::DirectUdpServer::bind(server).await.unwrap();
|
|
let observer = listener.handle();
|
|
let shutdown = tokio_util::sync::CancellationToken::new();
|
|
let run = tokio::spawn(listener.run_with_shutdown(shutdown.clone()));
|
|
let session = timeout(
|
|
Duration::from_secs(3),
|
|
transport::direct_udp::DirectUdpSession::connect(&client),
|
|
)
|
|
.await
|
|
.expect("direct UDP client authentication timed out")
|
|
.unwrap();
|
|
timeout(Duration::from_secs(2), async {
|
|
while observer.active_sessions().await != 1 {
|
|
sleep(Duration::from_millis(10)).await;
|
|
}
|
|
})
|
|
.await
|
|
.expect("server did not register the authenticated direct session");
|
|
|
|
shutdown.cancel();
|
|
timeout(Duration::from_secs(2), run)
|
|
.await
|
|
.expect("direct listener did not complete graceful cleanup")
|
|
.unwrap()
|
|
.unwrap();
|
|
assert_eq!(observer.active_sessions().await, 0);
|
|
session.close().await.unwrap();
|
|
}
|