feat: complete wgcloak v1 transports

This commit is contained in:
h3cx committed 2026-09-19 19:47:16 +02:00
1 parent 49ec9d8c33
commit cedab4910e
47 files changed
+2849 -400

No files matched your search

+91
View File
@@ -2,6 +2,9 @@ use wgcloak::control;
use std::os::unix::fs::PermissionsExt;
use tempfile::tempdir;
use tokio::io::AsyncReadExt;
use tokio::net::UnixStream;
use tokio::time::{sleep, timeout, Duration};
#[tokio::test]
async fn control_commands_and_socket_permissions() {
@@ -98,3 +101,91 @@ fn default_socket_path_never_targets_the_shared_tmp_socket() {
assert_eq!(path.file_name().unwrap(), "wgcloak.sock");
assert_ne!(path.parent().unwrap(), std::path::Path::new("/tmp"));
}
#[tokio::test]
async fn control_serializes_unusual_status_and_stats_keys_as_json() {
let dir = tempdir().unwrap();
let path = dir.path().join("runtime/control.sock");
let state = control::ControlState::default();
state.set_status("ready\"\n\t\0").await;
let mut clients = std::collections::BTreeMap::new();
clients.insert("client\"\n\t\0".to_owned(), 1);
state
.set_transport_stats_by_client_id("transport\"\n\t\0", 1, Default::default(), clients)
.await;
let server = control::ControlServer::bind(&path, state.clone())
.await
.unwrap();
let task = tokio::spawn(server.run());
let status: serde_json::Value =
serde_json::from_str(&control::send_command(&path, "status").await.unwrap()).unwrap();
assert_eq!(status["status"], "ready\"\n\t\0");
let stats: serde_json::Value =
serde_json::from_str(&control::send_command(&path, "stats").await.unwrap()).unwrap();
assert_eq!(
stats["stats"]["active_sessions_by_client_id"]["client\"\n\t\0"],
1
);
assert_eq!(
stats["stats"]["active_sessions_by_transport"]["transport\"\n\t\0"],
1
);
state.shutdown.cancel();
task.await.unwrap().unwrap();
}
#[tokio::test]
async fn idle_control_clients_are_closed_and_do_not_pin_shutdown() {
let dir = tempdir().unwrap();
let path = dir.path().join("runtime/control.sock");
let state = control::ControlState::default();
let server = control::ControlServer::bind(&path, state.clone())
.await
.unwrap();
let task = tokio::spawn(server.run());
let mut idle = UnixStream::connect(&path).await.unwrap();
let mut response = Vec::new();
timeout(
control::CONTROL_REQUEST_TIMEOUT + Duration::from_secs(1),
idle.read_to_end(&mut response),
)
.await
.expect("idle control client must be closed")
.unwrap();
assert!(response.is_empty());
state.shutdown.cancel();
timeout(Duration::from_secs(1), task)
.await
.unwrap()
.unwrap()
.unwrap();
}
#[tokio::test]
async fn control_rejects_clients_beyond_the_concurrency_bound() {
let dir = tempdir().unwrap();
let path = dir.path().join("runtime/control.sock");
let state = control::ControlState::default();
let server = control::ControlServer::bind(&path, state.clone())
.await
.unwrap();
let task = tokio::spawn(server.run());
let mut held = Vec::new();
for _ in 0..control::CONTROL_MAX_CLIENTS {
held.push(UnixStream::connect(&path).await.unwrap());
}
sleep(Duration::from_millis(25)).await;
let mut extra = UnixStream::connect(&path).await.unwrap();
let mut response = Vec::new();
timeout(Duration::from_secs(1), extra.read_to_end(&mut response))
.await
.expect("connection over the bound must be closed")
.unwrap();
assert!(response.is_empty());
drop(held);
state.shutdown.cancel();
task.await.unwrap().unwrap();
}
+118 -1
View File
@@ -248,6 +248,84 @@ async fn malformed_and_unauthenticated_packets_do_not_kill_server() {
server_task.abort();
}
#[tokio::test]
async fn direct_auth_attempt_limit_expires_and_recovers() {
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let backend_addr = backend.local_addr().unwrap();
let (mut server_config, _) = configs(45790, backend_addr.port());
server_config.limits.max_auth_attempts_per_ip = 2;
server_config.limits.auth_attempt_window_secs = 1;
let server = DirectUdpServer::bind(server_config).await.unwrap();
let stats = server.stats();
let server_task = tokio::spawn(server.run());
let raw = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let wrong_secret = [0xabu8; 32];
let mut response = [0u8; 2048];
for nonce_value in [1u8, 2] {
let mut nonce = [0u8; 16];
nonce[0] = nonce_value;
let hello = protocol::HelloPacket::authenticated(
&wrong_secret,
Bytes::from_static(b"test"),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs(),
nonce,
)
.unwrap();
raw.send_to(&hello.encode().unwrap(), "127.0.0.1:45790")
.await
.unwrap();
let (len, _) = timeout(Duration::from_secs(2), raw.recv_from(&mut response))
.await
.unwrap()
.unwrap();
assert!(matches!(
protocol::parse_packet(&response[..len]),
Ok(protocol::Packet::Reject(protocol::RejectPacket {
reason: protocol::RejectReason::AuthenticationFailed,
..
}))
));
}
let mut limited_nonce = [0u8; 16];
limited_nonce[0] = 3;
let limited = protocol::HelloPacket::authenticated(
&wrong_secret,
Bytes::from_static(b"test"),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs(),
limited_nonce,
)
.unwrap();
raw.send_to(&limited.encode().unwrap(), "127.0.0.1:45790")
.await
.unwrap();
let (len, _) = timeout(Duration::from_secs(2), raw.recv_from(&mut response))
.await
.unwrap()
.unwrap();
assert!(matches!(
protocol::parse_packet(&response[..len]),
Ok(protocol::Packet::Reject(protocol::RejectPacket {
reason: protocol::RejectReason::RateLimited,
..
}))
));
tokio::time::sleep(Duration::from_millis(1_100)).await;
let client_config = configs(45790, backend_addr.port()).1;
let session = DirectUdpSession::connect(&client_config).await.unwrap();
assert!(stats.snapshot().rate_limited_attempts >= 1);
session.close().await.unwrap();
server_task.abort();
}
#[tokio::test]
async fn authenticated_source_port_change_updates_return_peer() {
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
@@ -320,7 +398,9 @@ async fn authenticated_source_port_change_updates_return_peer() {
async fn identical_hello_retransmission_recovers_original_accept_and_session() {
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let backend_addr = backend.local_addr().unwrap();
let (server_config, _) = configs(45788, backend_addr.port());
let (mut server_config, _) = configs(45788, backend_addr.port());
// The cached exact retry must not consume a second auth-attempt slot.
server_config.limits.max_auth_attempts_per_ip = 1;
let server = DirectUdpServer::bind(server_config).await.unwrap();
let observer = server.handle();
let server_task = tokio::spawn(server.run());
@@ -481,6 +561,43 @@ async fn identical_hello_retransmission_recovers_original_accept_and_session() {
server_task.abort();
}
#[tokio::test]
async fn direct_udp_supports_ipv6_loopback_listen() {
let probe = match UdpSocket::bind("[::1]:0").await {
Ok(socket) => socket,
Err(_) => return,
};
let port = probe.local_addr().unwrap().port();
drop(probe);
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let backend_addr = backend.local_addr().unwrap();
let (mut server_config, mut client_config) = configs(port, backend_addr.port());
server_config.server.listen = "::1".to_owned();
client_config.client.server = "::1".to_owned();
let server = DirectUdpServer::bind(server_config).await.unwrap();
let server_task = tokio::spawn(server.run());
let backend_task = tokio::spawn(async move {
let mut buffer = [0u8; 2048];
let (len, peer) = backend.recv_from(&mut buffer).await.unwrap();
backend.send_to(&buffer[..len], peer).await.unwrap();
});
let session = DirectUdpSession::connect(&client_config).await.unwrap();
session
.send_datagram(Bytes::from_static(b"ipv6"))
.await
.unwrap();
assert_eq!(
timeout(Duration::from_secs(2), session.recv_datagram())
.await
.unwrap()
.unwrap(),
Bytes::from_static(b"ipv6")
);
session.close().await.unwrap();
backend_task.await.unwrap();
server_task.abort();
}
#[tokio::test]
async fn flags_are_authenticated_and_invalid_close_cannot_remove_session() {
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+568
View File
@@ -0,0 +1,568 @@
//! Daemon-level encrypted transport proofs.
//!
//! These tests deliberately use the public client/server runtime entrypoints:
//! an application UDP socket talks to a client runtime, which talks through an
//! authenticated encrypted listener to a per-session UDP backend socket.
use std::{
fs,
net::SocketAddr,
os::unix::fs::PermissionsExt,
path::Path,
process::Command,
sync::{
atomic::{AtomicU16, Ordering},
Arc,
},
time::Duration,
};
use tempfile::TempDir;
use tokio::{
net::UdpSocket,
sync::mpsc,
time::{sleep, timeout},
};
use wgcloak::{
client,
config::{self, Config, TransportMode},
control, server,
transport::{
quic::QuicClientConnector, tls_tcp::client_tls_config, TlsTcpConnector, TransportConnector,
TransportFuture, TransportSession,
},
};
const SECRET_ONE: &str = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
const SECRET_TWO: &str = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
const WRONG_SECRET: &str = "CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC";
fn free_udp_port() -> u16 {
std::net::UdpSocket::bind("127.0.0.1:0")
.unwrap()
.local_addr()
.unwrap()
.port()
}
fn free_tcp_port() -> u16 {
std::net::TcpListener::bind("127.0.0.1:0")
.unwrap()
.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 make_identity(dir: &TempDir) -> (std::path::PathBuf, std::path::PathBuf) {
let cert = dir.path().join("cert.der");
let key = dir.path().join("key.der");
let pem_cert = dir.path().join("cert.pem");
let pem_key = dir.path().join("key.pem");
let result = Command::new("openssl")
.args(["req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout"])
.arg(&pem_key)
.args(["-out"])
.arg(&pem_cert)
.args([
"-subj",
"/CN=localhost",
"-addext",
"subjectAltName=DNS:localhost,IP:127.0.0.1",
"-addext",
"basicConstraints=critical,CA:FALSE",
"-addext",
"extendedKeyUsage=serverAuth",
"-days",
"1",
])
.output()
.unwrap();
assert!(result.status.success(), "openssl failed: {:?}", result);
let result = Command::new("openssl")
.args(["x509", "-in"])
.arg(&pem_cert)
.args(["-outform", "DER", "-out"])
.arg(&cert)
.output()
.unwrap();
assert!(result.status.success());
let result = Command::new("openssl")
.args(["pkcs8", "-topk8", "-nocrypt", "-in"])
.arg(&pem_key)
.args(["-outform", "DER", "-out"])
.arg(&key)
.output()
.unwrap();
assert!(result.status.success());
(cert, key)
}
fn server_config(
cert: &Path,
key: &Path,
backend: SocketAddr,
quic_port: u16,
tls_port: u16,
quic: bool,
tls_tcp: bool,
) -> config::ServerConfig {
Config::from_str(&format!(
"[server]\nlisten='127.0.0.1'\n[backend]\ntarget='{backend}'\n[transports.direct_udp]\nenabled=false\n[transports.quic]\nenabled={quic}\nport={quic_port}\n[transports.http3]\nenabled=false\n[transports.tls_tcp]\nenabled={tls_tcp}\nport={tls_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=2000\n",
cert.display(),
key.display(),
))
.unwrap()
.into_server()
.unwrap()
}
fn client_config(
id: &str,
secret: &str,
mode: &str,
port: u16,
local_port: u16,
) -> config::ClientConfig {
Config::from_str(&format!(
"[client]\nserver='localhost'\nid='{id}'\nsecret='{secret}'\n[local]\nlisten='127.0.0.1:{local_port}'\n[transport]\nmode='{mode}'\n[transport.auto]\nprobe_timeout_ms=500\nretries=1\n[transport.quic]\nport={port}\n[transport.tls_tcp]\nport={port}\n[network]\nmax_datagram_size=1200\n"
))
.unwrap()
.into_client()
.unwrap()
}
fn quic_connector(config: config::ClientConfig, cert: &Path, key: &Path) -> QuicClientConnector {
let certs = wgcloak::transport::quic::load_tls_identity(cert, key)
.map(|(certs, _)| certs)
.unwrap();
let mut roots = rustls::RootCertStore::empty();
roots.add(certs[0].clone()).unwrap();
let tls = quinn::ClientConfig::with_root_certificates(Arc::new(roots)).unwrap();
QuicClientConnector::with_client_config(config, tls)
}
fn tls_connector(config: config::ClientConfig, cert: &Path) -> TlsTcpConnector {
let certs = wgcloak::transport::tls_tcp::load_certificates(cert).unwrap();
let mut roots = rustls::RootCertStore::empty();
roots.add(certs[0].clone()).unwrap();
TlsTcpConnector::with_client_config(config, (*client_tls_config(roots)).clone())
}
struct RotatingQuicConnector {
config: config::ClientConfig,
cert: std::path::PathBuf,
key: std::path::PathBuf,
port: Arc<AtomicU16>,
}
impl TransportConnector for RotatingQuicConnector {
fn connect<'a>(&'a self) -> TransportFuture<'a, Box<dyn TransportSession>> {
Box::pin(async move {
let mut config = self.config.clone();
config.transport.quic.port = self.port.load(Ordering::Relaxed);
quic_connector(config, &self.cert, &self.key)
.connect()
.await
})
}
}
async fn wait_for(path: &Path, command: &str, predicate: impl Fn(&str) -> bool) -> String {
let mut last = None;
timeout(Duration::from_secs(10), async {
loop {
if let Ok(response) = control::send_command(path, command).await {
if predicate(&response) {
return response;
}
last = Some(response);
}
sleep(Duration::from_millis(25)).await;
}
})
.await
.unwrap_or_else(|_| {
panic!(
"control {} {command} did not reach expected state: {last:?}",
path.display()
)
})
}
async fn echo_backend(socket: UdpSocket) {
echo_backend_observed(socket, None).await;
}
async fn echo_backend_observed(
socket: UdpSocket,
observed: Option<mpsc::UnboundedSender<SocketAddr>>,
) {
let mut buf = [0u8; 2048];
while let Ok((len, peer)) = socket.recv_from(&mut buf).await {
if let Some(observed) = &observed {
let _ = observed.send(peer);
}
let _ = socket.send_to(&buf[..len], peer).await;
}
}
async fn stop(path: &Path, task: tokio::task::JoinHandle<anyhow::Result<()>>) {
let response = control::send_command(path, "shutdown").await.unwrap();
assert!(response.contains("\"ok\":true"));
timeout(Duration::from_secs(5), task)
.await
.expect("runtime did not stop")
.unwrap()
.unwrap();
}
#[tokio::test]
async fn explicit_quic_runtime_forwards_two_clients_and_reports_identity() {
let dir = tempfile::tempdir().unwrap();
let (cert, key) = make_identity(&dir);
let runtime = private_runtime_dir(dir.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(echo_backend_observed(backend, Some(peer_tx)));
let quic_port = free_udp_port();
let server_path = runtime.join("server.sock");
let server_task = tokio::spawn(server::run_with_control_path(
server_config(
&cert,
&key,
backend_addr,
quic_port,
free_udp_port(),
true,
false,
),
server_path.clone(),
));
wait_for(&server_path, "status", |s| s.contains("transports=quic")).await;
let mut clients = Vec::new();
for (id, secret) in [("one", SECRET_ONE), ("two", SECRET_TWO)] {
let local = free_udp_port();
let app = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let cfg = client_config(id, secret, "quic", quic_port, local);
let control_path = runtime.join(format!("client-{id}.sock"));
let connector =
Arc::new(quic_connector(cfg.clone(), &cert, &key)) as Arc<dyn TransportConnector>;
let task = tokio::spawn(client::run_with_transport_connector_and_control_path(
cfg,
control_path.clone(),
TransportMode::Quic,
connector,
));
wait_for(&control_path, "stats", |s| {
s.contains("\"state\":\"connected\"")
})
.await;
clients.push((local, app, control_path, task));
}
wait_for(&server_path, "stats", |s| {
s.contains("\"active_sessions_by_transport\":{\"quic\":2")
&& s.contains("\"one\":1")
&& s.contains("\"two\":1")
})
.await;
for (local, app, _, _) in &clients {
app.send_to(b"encrypted-echo", ("127.0.0.1", *local))
.await
.unwrap();
let mut reply = [0u8; 64];
let (len, _) = timeout(Duration::from_secs(5), app.recv_from(&mut reply))
.await
.unwrap()
.unwrap();
assert_eq!(&reply[..len], b"encrypted-echo");
}
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 encrypted session needs its own backend socket"
);
for (_, _, control_path, task) in clients {
stop(&control_path, task).await;
}
stop(&server_path, server_task).await;
backend_task.abort();
}
#[tokio::test]
async fn explicit_tls_runtime_forwards_and_rejects_hostname_mismatch() {
let dir = tempfile::tempdir().unwrap();
let (cert, key) = make_identity(&dir);
let runtime = private_runtime_dir(dir.path());
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let backend_addr = backend.local_addr().unwrap();
let backend_task = tokio::spawn(echo_backend(backend));
let tls_port = free_tcp_port();
let server_path = runtime.join("server.sock");
let server_task = tokio::spawn(server::run_with_control_path(
server_config(
&cert,
&key,
backend_addr,
free_udp_port(),
tls_port,
false,
true,
),
server_path.clone(),
));
wait_for(&server_path, "status", |s| s.contains("transports=tls_tcp")).await;
let local = free_udp_port();
let app = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let cfg = client_config("one", SECRET_ONE, "tls_tcp", tls_port, local);
let client_path = runtime.join("client.sock");
let connector = Arc::new(tls_connector(cfg.clone(), &cert)) as Arc<dyn TransportConnector>;
let client_task = tokio::spawn(client::run_with_transport_connector_and_control_path(
cfg.clone(),
client_path.clone(),
TransportMode::TlsTcp,
connector,
));
wait_for(&client_path, "stats", |s| {
s.contains("\"state\":\"connected\"")
})
.await;
app.send_to(b"tls-echo", ("127.0.0.1", local))
.await
.unwrap();
let mut reply = [0u8; 64];
let (len, _) = timeout(Duration::from_secs(5), app.recv_from(&mut reply))
.await
.unwrap()
.unwrap();
assert_eq!(&reply[..len], b"tls-echo");
wait_for(&server_path, "stats", |s| {
s.contains("\"active_sessions_by_transport\":{\"tls_tcp\":1")
&& s.contains("\"active_sessions_by_client_id\":{\"one\":1")
})
.await;
// A trusted certificate with an unrelated hostname must still fail normal
// hostname verification; trusting the issuing certificate is not enough.
let mut mismatch = cfg.clone();
mismatch.client.server = "not-localhost".to_owned();
let result = tls_connector(mismatch, &cert).connect().await;
assert!(matches!(
result,
Err(wgcloak::transport::TransportError::Io(_))
));
stop(&client_path, client_task).await;
stop(&server_path, server_task).await;
backend_task.abort();
}
#[tokio::test]
async fn encrypted_runtime_cancellation_releases_server_sessions() {
let dir = tempfile::tempdir().unwrap();
let (cert, key) = make_identity(&dir);
let runtime = private_runtime_dir(dir.path());
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let backend_addr = backend.local_addr().unwrap();
let backend_task = tokio::spawn(echo_backend(backend));
let quic_port = free_udp_port();
let server_path = runtime.join("server.sock");
let server_task = tokio::spawn(server::run_with_control_path(
server_config(
&cert,
&key,
backend_addr,
quic_port,
free_udp_port(),
true,
false,
),
server_path.clone(),
));
wait_for(&server_path, "status", |s| s.contains("transports=quic")).await;
let local = free_udp_port();
let cfg = client_config("one", SECRET_ONE, "quic", quic_port, local);
let client_path = runtime.join("client.sock");
let connector =
Arc::new(quic_connector(cfg.clone(), &cert, &key)) as Arc<dyn TransportConnector>;
let client_task = tokio::spawn(client::run_with_transport_connector_and_control_path(
cfg,
client_path.clone(),
TransportMode::Quic,
connector,
));
wait_for(&client_path, "stats", |s| {
s.contains("\"state\":\"connected\"")
})
.await;
wait_for(&server_path, "stats", |s| {
s.contains("\"active_sessions\":1")
})
.await;
stop(&client_path, client_task).await;
wait_for(&server_path, "stats", |s| {
s.contains("\"active_sessions\":0")
})
.await;
stop(&server_path, server_task).await;
backend_task.abort();
}
#[tokio::test]
async fn wrong_secret_is_authentication_rejection_without_fallback() {
let dir = tempfile::tempdir().unwrap();
let (cert, key) = make_identity(&dir);
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let backend_addr = backend.local_addr().unwrap();
let backend_task = tokio::spawn(echo_backend(backend));
let quic_port = free_udp_port();
let server_path = private_runtime_dir(dir.path()).join("server.sock");
let server_task = tokio::spawn(server::run_with_control_path(
server_config(
&cert,
&key,
backend_addr,
quic_port,
free_udp_port(),
true,
false,
),
server_path.clone(),
));
wait_for(&server_path, "status", |s| s.contains("transports=quic")).await;
let cfg = client_config("one", WRONG_SECRET, "quic", quic_port, free_udp_port());
let result = timeout(
Duration::from_secs(5),
QuicClientConnector::with_client_config(cfg.clone(), {
let certs = wgcloak::transport::quic::load_tls_identity(&cert, &key)
.unwrap()
.0;
let mut roots = rustls::RootCertStore::empty();
roots.add(certs[0].clone()).unwrap();
quinn::ClientConfig::with_root_certificates(Arc::new(roots)).unwrap()
})
.connect(),
)
.await
.unwrap();
assert!(matches!(
result,
Err(wgcloak::transport::TransportError::AuthenticationRejected(
_
))
));
stop(&server_path, server_task).await;
backend_task.abort();
}
#[tokio::test]
async fn quic_runtime_reconnects_after_server_restart_and_cleans_on_cancel() {
let dir = tempfile::tempdir().unwrap();
let (cert, key) = make_identity(&dir);
let runtime = private_runtime_dir(dir.path());
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let backend_addr = backend.local_addr().unwrap();
let backend_task = tokio::spawn(echo_backend(backend));
let quic_port = free_udp_port();
let server_path = runtime.join("server.sock");
let first_server = tokio::spawn(server::run_with_control_path(
server_config(
&cert,
&key,
backend_addr,
quic_port,
free_udp_port(),
true,
false,
),
server_path.clone(),
));
wait_for(&server_path, "status", |s| s.contains("transports=quic")).await;
let local = free_udp_port();
let app = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let cfg = client_config("one", SECRET_ONE, "quic", quic_port, local);
let client_path = runtime.join("client.sock");
let rotating_port = Arc::new(AtomicU16::new(quic_port));
let connector = Arc::new(RotatingQuicConnector {
config: cfg.clone(),
cert: cert.clone(),
key: key.clone(),
port: rotating_port.clone(),
}) as Arc<dyn TransportConnector>;
let client_task = tokio::spawn(client::run_with_transport_connector_and_control_path(
cfg,
client_path.clone(),
TransportMode::Quic,
connector,
));
wait_for(&client_path, "stats", |s| {
s.contains("\"state\":\"connected\"")
})
.await;
app.send_to(b"before-restart", ("127.0.0.1", local))
.await
.unwrap();
let mut reply = [0u8; 64];
let (len, _) = timeout(Duration::from_secs(5), app.recv_from(&mut reply))
.await
.unwrap()
.unwrap();
assert_eq!(&reply[..len], b"before-restart");
control::send_command(&server_path, "shutdown")
.await
.unwrap();
timeout(Duration::from_secs(5), first_server)
.await
.unwrap()
.unwrap()
.unwrap();
let replacement_port = free_udp_port();
rotating_port.store(replacement_port, Ordering::Relaxed);
let replacement_path = runtime.join("server-replacement.sock");
let replacement = tokio::spawn(server::run_with_control_path(
server_config(
&cert,
&key,
backend_addr,
replacement_port,
free_udp_port(),
true,
false,
),
replacement_path.clone(),
));
wait_for(&replacement_path, "status", |s| {
s.contains("transports=quic")
})
.await;
wait_for(&client_path, "stats", |s| {
s.contains("\"state\":\"connected\"") && s.contains("\"reconnections\":1")
})
.await;
app.send_to(b"after-restart", ("127.0.0.1", local))
.await
.unwrap();
let (len, _) = timeout(Duration::from_secs(5), app.recv_from(&mut reply))
.await
.unwrap()
.unwrap();
assert_eq!(&reply[..len], b"after-restart");
stop(&client_path, client_task).await;
stop(&replacement_path, replacement).await;
backend_task.abort();
}
+22 -4
View File
@@ -180,11 +180,13 @@ async fn connect_udp_rejects_bad_secret_before_returning_session() {
let server_task = tokio::spawn(async move {
let incoming = server_endpoint.accept().await.unwrap();
let quic_connection = incoming.await.unwrap();
Http3ServerConnection::new(quic_connection)
let observer = quic_connection.clone();
let result = Http3ServerConnection::new(quic_connection)
.await
.unwrap()
.accept_connect_udp("/tunnel", &clients, 1200, Duration::from_secs(3))
.await
.await;
(result, observer)
});
let quic_connection = connect_client(address, client_tls).await;
@@ -206,7 +208,7 @@ async fn connect_udp_rejects_bad_secret_before_returning_session() {
"unexpected client error: {:?}",
client_result.as_ref().err()
);
let server_result = timeout(Duration::from_secs(3), server_task)
let (server_result, server_connection) = timeout(Duration::from_secs(3), server_task)
.await
.unwrap()
.unwrap();
@@ -215,6 +217,16 @@ async fn connect_udp_rejects_bad_secret_before_returning_session() {
"unexpected server error: {:?}",
server_result.as_ref().err()
);
timeout(Duration::from_secs(2), async {
loop {
if server_connection.close_reason().is_some() {
break;
}
tokio::task::yield_now().await;
}
})
.await
.expect("authentication rejection must close the QUIC connection");
}
#[tokio::test]
@@ -268,5 +280,11 @@ async fn http3_server_admission_permit_tracks_authenticated_session_lifetime() {
assert_eq!(admission.pending_handshakes(), 0);
server_session.close().await.unwrap();
assert_eq!(admission.active_sessions(), 0);
client_session.close(Duration::from_secs(2)).await.unwrap();
timeout(
Duration::from_secs(2),
client_session.close(Duration::from_millis(250)),
)
.await
.expect("client cleanup after peer close must be bounded")
.unwrap_or_else(|error| panic!("client cleanup returned an unexpected error: {error}"));
}
+50
View File
@@ -177,3 +177,53 @@ async fn tls_tcp_authenticates_round_trips_and_classifies_failures() {
.unwrap()
.is_err());
}
#[tokio::test]
async fn tls_tcp_ipv6_loopback_listener_accepts_bracketed_configuration() {
if Command::new("openssl").arg("version").output().is_err() {
return;
}
let probe = match tokio::net::TcpListener::bind("[::1]:0").await {
Ok(listener) => listener,
Err(_) => return,
};
let port = probe.local_addr().unwrap().port();
drop(probe);
let dir = tempdir().unwrap();
let cert = dir.path().join("cert.pem");
let key = dir.path().join("key.pem");
let generated = Command::new("openssl")
.args([
"req",
"-x509",
"-newkey",
"rsa:2048",
"-nodes",
"-subj",
"/CN=localhost",
"-addext",
"subjectAltName=IP:::1",
"-days",
"1",
"-keyout",
])
.arg(&key)
.args(["-out"])
.arg(&cert)
.output()
.unwrap();
assert!(generated.status.success(), "openssl failed: {generated:?}");
let config = Config::from_str(&format!(
"[server]\nlisten = \"::1\"\n[backend]\ntarget = \"127.0.0.1:51820\"\n[transports.direct_udp]\nenabled = false\n[transports.quic]\nenabled = false\n[transports.http3]\nenabled = false\n[transports.tls_tcp]\nenabled = true\nport = {port}\n[tls]\ncertificate = \"{}\"\nprivate_key = \"{}\"\n[[auth.clients]]\nid = \"test\"\nsecret = \"{SECRET}\"\n",
cert.display(),
key.display()
))
.unwrap()
.into_server()
.unwrap();
let server =
TlsTcpServer::bind_with_admission(&config, AdmissionController::new(&config.limits))
.await
.unwrap();
assert!(server.local_addr().unwrap().ip().is_ipv6());
}
+50
View File
@@ -0,0 +1,50 @@
use wgcloak::config::{Config, ServerConfig, MAX_APPLICATION_DATAGRAM_SIZE};
const SECRET: &str = "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=";
fn server_text(max_datagram_size: Option<u16>) -> String {
let network = max_datagram_size
.map(|value| format!("[network]\nmax_datagram_size = {value}\n"))
.unwrap_or_default();
format!(
"[server]\nlisten = \"::1\"\n[backend]\ntarget = \"127.0.0.1:51820\"\n{network}[transports.direct_udp]\nport = 51830\n[[auth.clients]]\nid = \"test\"\nsecret = \"{SECRET}\"\n"
)
}
#[test]
fn server_datagram_capability_defaults_and_accepts_boundary() {
let Config::Server(defaults) = Config::from_str(&server_text(None)).unwrap() else {
panic!("expected server configuration");
};
assert_eq!(defaults.network.max_datagram_size, 1200);
let Config::Server(config) =
Config::from_str(&server_text(Some(MAX_APPLICATION_DATAGRAM_SIZE))).unwrap()
else {
panic!("expected server configuration");
};
config.validate().unwrap();
assert_eq!(
config.network.max_datagram_size,
MAX_APPLICATION_DATAGRAM_SIZE
);
}
#[test]
fn server_datagram_capability_rejects_unsupported_size() {
let Config::Server(config) =
Config::from_str(&server_text(Some(MAX_APPLICATION_DATAGRAM_SIZE + 1))).unwrap()
else {
panic!("expected server configuration");
};
let error = config.validate().unwrap_err().to_string();
assert!(error.contains("network.max_datagram_size"));
}
#[test]
fn server_config_type_remains_constructible_from_toml() {
let Config::Server(config): Config = Config::from_str(&server_text(Some(1400))).unwrap() else {
panic!("expected server configuration");
};
let _: ServerConfig = config;
}
+165 -1
View File
@@ -29,6 +29,37 @@ impl TransportSession for Session {
1200
}
}
struct EchoSession {
payload: Mutex<Option<Bytes>>,
}
impl TransportSession for EchoSession {
fn send_datagram<'a>(&'a self, data: Bytes) -> TransportFuture<'a, ()> {
Box::pin(async move {
*self.payload.lock().await = Some(data);
Ok(())
})
}
fn recv_datagram<'a>(&'a self) -> TransportFuture<'a, Bytes> {
Box::pin(async move {
self.payload
.lock()
.await
.take()
.ok_or(TransportError::Closed)
})
}
fn close<'a>(&'a self) -> TransportFuture<'a, ()> {
Box::pin(async { Ok(()) })
}
fn max_datagram_size(&self) -> usize {
1200
}
}
struct Probe {
calls: Arc<Mutex<Vec<TransportMode>>>,
auth: bool,
@@ -48,9 +79,113 @@ impl TransportProbe for Probe {
}
})
}
fn wait_for_session_failure<'a>(
&'a self,
_session: &'a dyn TransportSession,
) -> supervisor::SessionWatchFuture<'a> {
Box::pin(async {
Err(TransportError::Unsupported(
"test probe has no session lifecycle".into(),
))
})
}
}
fn config(mode: &str) -> config::ClientConfig {
Config::from_str(&format!("[client]\nserver=\"127.0.0.1\"\nid=\"c\"\nsecret=\"{}\"\n[transport]\nmode=\"{}\"\n[transport.auto]\norder=[\"direct_udp\",\"quic\"]\nretries=2\nprobe_timeout_ms=20\n", "A".repeat(64), mode)).unwrap().into_client().unwrap()
config_with_order(mode, "\"direct_udp\",\"quic\"")
}
fn config_with_order(mode: &str, order: &str) -> config::ClientConfig {
Config::from_str(&format!("[client]\nserver=\"127.0.0.1\"\nid=\"c\"\nsecret=\"{}\"\n[transport]\nmode=\"{}\"\n[transport.auto]\norder=[{}]\nretries=2\nprobe_timeout_ms=20\n", "A".repeat(64), mode, order)).unwrap().into_client().unwrap()
}
struct SelectingProbe {
calls: Arc<Mutex<Vec<TransportMode>>>,
selected: TransportMode,
}
impl TransportProbe for SelectingProbe {
fn connect(&self, mode: TransportMode) -> ProbeFuture {
let calls = self.calls.clone();
let selected = self.selected;
Box::pin(async move {
calls.lock().await.push(mode);
if mode == selected {
Ok(Box::new(EchoSession {
payload: Mutex::new(None),
}) as Box<dyn TransportSession>)
} else {
Err(TransportError::HandshakeTimeout)
}
})
}
fn wait_for_session_failure<'a>(
&'a self,
_session: &'a dyn TransportSession,
) -> supervisor::SessionWatchFuture<'a> {
Box::pin(async { Err(TransportError::Unsupported("test probe".into())) })
}
}
async fn assert_selected_with_roundtrip(
config: config::ClientConfig,
selected: TransportMode,
expected_calls: Vec<TransportMode>,
) {
let calls = Arc::new(Mutex::new(Vec::new()));
let supervisor = TransportSupervisor::new(
config,
SelectingProbe {
calls: calls.clone(),
selected,
},
);
let session = supervisor.connect_once().await.unwrap();
assert_eq!(*calls.lock().await, expected_calls);
let status = supervisor.status().await;
assert_eq!(status.state, supervisor::SupervisorState::Connected);
assert_eq!(status.selected_transport, Some(selected));
session
.send_datagram(Bytes::from_static(b"auto-payload"))
.await
.unwrap();
assert_eq!(
session.recv_datagram().await.unwrap(),
Bytes::from_static(b"auto-payload")
);
}
#[tokio::test]
async fn auto_selects_quic_after_direct_udp_is_unavailable() {
assert_selected_with_roundtrip(
config_with_order("auto", "\"direct_udp\",\"quic\""),
TransportMode::Quic,
vec![
TransportMode::DirectUdp,
TransportMode::DirectUdp,
TransportMode::Quic,
],
)
.await;
}
#[tokio::test]
async fn auto_selects_tls_tcp_after_udp_carriers_are_unavailable() {
assert_selected_with_roundtrip(
config_with_order("auto", "\"direct_udp\",\"quic\",\"http3\",\"tls_tcp\""),
TransportMode::TlsTcp,
vec![
TransportMode::DirectUdp,
TransportMode::DirectUdp,
TransportMode::Quic,
TransportMode::Quic,
TransportMode::Http3,
TransportMode::Http3,
TransportMode::TlsTcp,
],
)
.await;
}
#[tokio::test]
@@ -89,6 +224,10 @@ async fn selection_order_retries_and_auth_stop() {
Err(TransportError::AuthenticationRejected(_))
));
assert_eq!(*calls.lock().await, vec![TransportMode::DirectUdp]);
assert_eq!(
supervisor.status().await.state,
supervisor::SupervisorState::AuthenticationRejected
);
}
#[test]
@@ -154,3 +293,28 @@ async fn established_failure_reprobes_without_reading_datagrams() {
assert!(result.is_ok());
assert!(calls.load(Ordering::SeqCst) >= 2);
}
struct UnsupportedLifecycleProbe;
impl TransportProbe for UnsupportedLifecycleProbe {
fn connect(&self, _mode: TransportMode) -> ProbeFuture {
Box::pin(async { Ok(Box::new(Session) as Box<dyn TransportSession>) })
}
fn wait_for_session_failure<'a>(
&'a self,
_session: &'a dyn TransportSession,
) -> supervisor::SessionWatchFuture<'a> {
Box::pin(async { Err(TransportError::Unsupported("lifecycle unavailable".into())) })
}
}
#[tokio::test]
async fn unsupported_session_lifecycle_is_explicitly_terminal() {
let supervisor = TransportSupervisor::new(config("direct_udp"), UnsupportedLifecycleProbe);
let result = timeout(Duration::from_secs(1), supervisor.run())
.await
.expect("unsupported lifecycle should not hang")
.unwrap_err();
assert!(matches!(result, TransportError::Unsupported(_)));
}
+42
View File
@@ -36,6 +36,14 @@ fn codec_rejects_oversized_before_payload() {
assert!(decoder.finish().is_ok());
}
#[test]
fn codec_rejects_maximum_above_u16_wire_limit() {
assert_eq!(
FrameCodec::new(65_536).unwrap_err(),
FrameError::MaxTooLarge
);
}
#[test]
fn codec_reports_truncated_header_or_payload() {
let mut decoder = FrameCodec::new(32).unwrap();
@@ -95,6 +103,40 @@ async fn tls_stream_authentication_exchange_uses_challenge_response() {
assert_eq!(authenticated.id, "laptop");
}
#[tokio::test]
async fn tls_authentication_accepts_validated_maximum_datagram_size() {
let (left, right) = duplex(4096);
let (mut server_reader, mut server_writer) = split(left);
let (mut client_reader, mut client_writer) = split(right);
let mut clients = HashMap::new();
clients.insert("laptop".to_owned(), vec![7u8; 32]);
let server_clients = clients.clone();
let server = tokio::spawn(async move {
server_authenticate(
&mut server_reader,
&mut server_writer,
&server_clients,
65_503 + 512,
Duration::from_secs(1),
)
.await
});
let client = tokio::spawn(async move {
client_authenticate(
&mut client_reader,
&mut client_writer,
"laptop",
&[7u8; 32],
65_503 + 512,
Duration::from_secs(1),
)
.await
});
assert!(client.await.unwrap().is_ok());
assert_eq!(server.await.unwrap().unwrap().id, "laptop");
}
#[tokio::test]
async fn bad_secret_is_rejected_without_authenticating() {
let (left, right) = duplex(4096);