h3cx/wgcloak
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

checkpoint: integrate authenticated transports

Browse Source
This commit is contained in:
h3cx committed 2026-09-15 10:57:11 +02:00
1 parent 313ab00765
commit 586ff8f223
42 files changed
+6277 -771
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/control.rs
+64 -9
View File
@@ -1,26 +1,38 @@
#[path = "../src/control.rs"]
mod control;
use wgcloak::control;
use std::os::unix::fs::PermissionsExt;
use tempfile::tempdir;
use tokio::time::{sleep, Duration};
#[tokio::test]
async fn control_commands_and_socket_permissions() {
let dir = tempdir().unwrap();
let path = dir.path().join("control.sock");
let runtime_dir = dir.path().join("runtime");
let path = runtime_dir.join("control.sock");
let state = control::ControlState::default();
state.enable_reconnect();
state.set_status("ready").await;
state.set_stats_json("{\"packets\":3}").await;
let server = control::ControlServer::bind(&path, state.clone())
.await
.unwrap();
assert_eq!(
std::fs::metadata(&runtime_dir)
.unwrap()
.permissions()
.mode()
& 0o777,
0o700
);
assert_eq!(
std::fs::metadata(&path).unwrap().permissions().mode() & 0o777,
0o600
);
let task = tokio::spawn(server.run());
sleep(Duration::from_millis(10)).await;
let second_bind = control::ControlServer::bind(&path, control::ControlState::default()).await;
assert!(matches!(
second_bind,
Err(error) if error.kind() == std::io::ErrorKind::AddrInUse
));
assert!(control::send_command(&path, "status")
.await
.unwrap()
@@ -33,13 +45,56 @@ async fn control_commands_and_socket_permissions() {
.await
.unwrap()
.contains("true"));
assert!(control::send_command(&path, "reload")
.await
.unwrap()
.contains("true"));
let reload = control::send_command(&path, "reload").await.unwrap();
assert!(reload.contains("\"ok\":false"));
assert!(reload.contains("not supported"));
assert!(control::send_command(&path, "shutdown")
.await
.unwrap()
.contains("true"));
task.await.unwrap().unwrap();
}
#[tokio::test]
async fn replaces_only_a_verified_stale_socket() {
let dir = tempdir().unwrap();
let private_dir = dir.path().join("runtime");
std::fs::create_dir(&private_dir).unwrap();
std::fs::set_permissions(&private_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
let path = private_dir.join("stale.sock");
let stale = std::os::unix::net::UnixListener::bind(&path).unwrap();
drop(stale);
let server = control::ControlServer::bind(&path, control::ControlState::default())
.await
.expect("verified stale socket should be replaced");
assert_eq!(
std::fs::metadata(&path).unwrap().permissions().mode() & 0o777,
0o600
);
drop(server);
assert!(!path.exists());
}
#[tokio::test]
async fn refuses_to_replace_a_non_socket_path() {
let dir = tempdir().unwrap();
let private_dir = dir.path().join("runtime");
std::fs::create_dir(&private_dir).unwrap();
std::fs::set_permissions(&private_dir, std::fs::Permissions::from_mode(0o700)).unwrap();
let path = private_dir.join("control.sock");
std::fs::write(&path, b"keep this file").unwrap();
let result = control::ControlServer::bind(&path, control::ControlState::default()).await;
assert!(matches!(
result,
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists
));
assert_eq!(std::fs::read(&path).unwrap(), b"keep this file");
}
#[test]
fn default_socket_path_never_targets_the_shared_tmp_socket() {
let path = control::default_socket_path();
assert_eq!(path.file_name().unwrap(), "wgcloak.sock");
assert_ne!(path.parent().unwrap(), std::path::Path::new("/tmp"));
}
tests/direct_udp.rs
+322 -13
View File
@@ -1,13 +1,4 @@
#[path = "../src/config.rs"]
mod config;
#[path = "../src/datagram.rs"]
mod datagram;
#[path = "../src/protocol/mod.rs"]
mod protocol;
#[path = "../src/stats.rs"]
mod stats;
#[path = "../src/transport/mod.rs"]
mod transport;
use wgcloak::{config, protocol, transport};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
@@ -16,7 +7,7 @@ use bytes::Bytes;
use config::Config;
use tokio::{net::UdpSocket, time::timeout};
use transport::{
direct_udp::{DirectUdpServer, DirectUdpSession},
direct_udp::{DirectUdpServer, DirectUdpServerHandle, DirectUdpSession},
TransportError, TransportSession,
};
@@ -51,6 +42,19 @@ fn multi_configs(
(server, client("one"), client("two"))
}
async fn wait_for_active_sessions(server: &DirectUdpServerHandle, expected: usize) {
timeout(Duration::from_secs(2), async {
loop {
if server.active_sessions().await == expected {
return;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.unwrap_or_else(|_| panic!("active session count did not become {expected}"));
}
#[tokio::test]
async fn direct_udp_round_trip_and_auth_rejection() {
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
@@ -66,6 +70,11 @@ async fn direct_udp_round_trip_and_auth_rejection() {
backend.send_to(&buf[..len], peer).await.unwrap();
});
let session = DirectUdpSession::connect(&client_config).await.unwrap();
assert_eq!(
session.authenticated_keepalive_interval(),
Some(Duration::from_secs(15))
);
session.send_keepalive().await.unwrap();
session
.send_datagram(Bytes::from_static(b"opaque"))
.await
@@ -99,19 +108,42 @@ async fn direct_udp_enforces_payload_limit_and_expires_idle_session() {
let (mut server_config, client_config) = configs(45782, backend_addr.port());
server_config.session.idle_timeout_secs = 1;
let server = DirectUdpServer::bind(server_config).await.unwrap();
let stats = server.stats();
let observer = server.handle();
let stats = observer.stats();
let server_task = tokio::spawn(server.run());
let session = DirectUdpSession::connect(&client_config).await.unwrap();
wait_for_active_sessions(&observer, 1).await;
assert!(matches!(
session.send_datagram(Bytes::from(vec![0u8; 1201])).await,
Err(TransportError::Oversized { maximum: 1200 })
));
tokio::time::sleep(Duration::from_secs(3)).await;
assert_eq!(stats.snapshot().expired_sessions, 1);
wait_for_active_sessions(&observer, 0).await;
session.close().await.unwrap();
server_task.abort();
}
#[tokio::test]
async fn direct_udp_observer_tracks_authenticated_session_close() {
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let backend_addr = backend.local_addr().unwrap();
let (server_config, client_config) = configs(45787, backend_addr.port());
let server = DirectUdpServer::bind(server_config).await.unwrap();
let observer = server.handle();
let stats = observer.stats();
let server_task = tokio::spawn(server.run());
let session = DirectUdpSession::connect(&client_config).await.unwrap();
wait_for_active_sessions(&observer, 1).await;
assert_eq!(stats.snapshot().expired_sessions, 0);
session.close().await.unwrap();
wait_for_active_sessions(&observer, 0).await;
assert_eq!(stats.snapshot().expired_sessions, 0);
server_task.abort();
}
#[tokio::test]
async fn concurrent_clients_use_independent_backend_sockets() {
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
@@ -178,6 +210,14 @@ async fn malformed_and_unauthenticated_packets_do_not_kill_server() {
let raw = UdpSocket::bind("127.0.0.1:0").await.unwrap();
raw.send_to(&[1, 2, 3], "127.0.0.1:45784").await.unwrap();
raw.send_to(&[0u8; 64], "127.0.0.1:45784").await.unwrap();
let oversized = protocol::DataPacket::new(1, 1, vec![0u8; 1201], [0; 16])
.encode()
.unwrap();
raw.send_to(&oversized, "127.0.0.1:45784").await.unwrap();
let unexpected = protocol::RejectPacket::new(protocol::RejectReason::AuthenticationFailed)
.encode()
.unwrap();
raw.send_to(&unexpected, "127.0.0.1:45784").await.unwrap();
let mut wrong = client_config.clone();
wrong.client.secret.secret = Some("B".repeat(64));
assert!(matches!(
@@ -201,7 +241,8 @@ async fn malformed_and_unauthenticated_packets_do_not_kill_server() {
.unwrap(),
Bytes::from_static(b"still-alive")
);
assert!(stats.snapshot().datagrams_dropped >= 3);
assert!(stats.snapshot().datagrams_dropped >= 5);
assert!(stats.snapshot().oversized_datagrams >= 1);
session.close().await.unwrap();
backend_task.await.unwrap();
server_task.abort();
@@ -275,6 +316,274 @@ async fn authenticated_source_port_change_updates_return_peer() {
server_task.abort();
}
#[tokio::test]
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 server = DirectUdpServer::bind(server_config).await.unwrap();
let observer = server.handle();
let server_task = tokio::spawn(server.run());
let client = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let other_source = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let secret = base64::engine::general_purpose::STANDARD
.decode(SECRET)
.unwrap();
let mut nonce = [0u8; 16];
nonce[0] = 44;
let hello = protocol::HelloPacket::authenticated(
&secret,
Bytes::from_static(b"test"),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs(),
nonce,
)
.unwrap();
let hello_bytes = hello.encode().unwrap();
client
.send_to(&hello_bytes, "127.0.0.1:45788")
.await
.unwrap();
let mut response = [0u8; 2048];
let (len, _) = timeout(Duration::from_secs(2), client.recv_from(&mut response))
.await
.unwrap()
.unwrap();
let first_accept = match protocol::parse_packet(&response[..len]).unwrap() {
protocol::Packet::Accept(accept) => accept,
other => panic!("unexpected handshake response: {other:?}"),
};
client
.send_to(&hello_bytes, "127.0.0.1:45788")
.await
.unwrap();
let (len, _) = timeout(Duration::from_secs(2), client.recv_from(&mut response))
.await
.unwrap()
.unwrap();
let retry_accept = match protocol::parse_packet(&response[..len]).unwrap() {
protocol::Packet::Accept(accept) => accept,
other => panic!("identical HELLO should recover ACCEPT, got {other:?}"),
};
assert_eq!(retry_accept, first_accept);
wait_for_active_sessions(&observer, 1).await;
other_source
.send_to(&hello_bytes, "127.0.0.1:45788")
.await
.unwrap();
let (len, _) = timeout(
Duration::from_secs(2),
other_source.recv_from(&mut response),
)
.await
.unwrap()
.unwrap();
assert!(matches!(
protocol::parse_packet(&response[..len]),
Ok(protocol::Packet::Reject(protocol::RejectPacket {
reason: protocol::RejectReason::ReplayDetected,
..
}))
));
let incompatible = protocol::HelloPacket::authenticated(
&secret,
Bytes::from_static(b"test"),
hello.timestamp + 1,
nonce,
)
.unwrap();
client
.send_to(&incompatible.encode().unwrap(), "127.0.0.1:45788")
.await
.unwrap();
let (len, _) = timeout(Duration::from_secs(2), client.recv_from(&mut response))
.await
.unwrap()
.unwrap();
assert!(matches!(
protocol::parse_packet(&response[..len]),
Ok(protocol::Packet::Reject(protocol::RejectPacket {
reason: protocol::RejectReason::ReplayDetected,
..
}))
));
let session_key = protocol::derive_session_key(
&secret,
&nonce,
&first_accept.server_nonce,
first_accept.session_id,
);
let (peer_tx, mut peer_rx) = tokio::sync::mpsc::unbounded_channel();
let backend_task = tokio::spawn(async move {
let mut buffer = [0u8; 2048];
for _ in 0..2 {
let (len, peer) = backend.recv_from(&mut buffer).await.unwrap();
peer_tx.send(peer).unwrap();
backend.send_to(&buffer[..len], peer).await.unwrap();
}
});
for (sequence, payload) in [(1, b"one".as_slice()), (2, b"two".as_slice())] {
let packet = protocol::DataPacket::authenticated(
&session_key,
first_accept.session_id,
sequence,
Bytes::copy_from_slice(payload),
);
client
.send_to(&packet.encode().unwrap(), "127.0.0.1:45788")
.await
.unwrap();
let (len, _) = timeout(Duration::from_secs(2), client.recv_from(&mut response))
.await
.unwrap()
.unwrap();
assert!(matches!(
protocol::parse_packet(&response[..len]),
Ok(protocol::Packet::Data(_))
));
}
backend_task.await.unwrap();
assert_eq!(peer_rx.recv().await, peer_rx.recv().await);
assert_eq!(observer.active_sessions().await, 1);
let close = protocol::SessionPacket::authenticated(
protocol::PacketType::Close,
&session_key,
first_accept.session_id,
3,
);
client
.send_to(&close.encode().unwrap(), "127.0.0.1:45788")
.await
.unwrap();
wait_for_active_sessions(&observer, 0).await;
client
.send_to(&hello_bytes, "127.0.0.1:45788")
.await
.unwrap();
let (len, _) = timeout(Duration::from_secs(2), client.recv_from(&mut response))
.await
.unwrap()
.unwrap();
assert!(matches!(
protocol::parse_packet(&response[..len]),
Ok(protocol::Packet::Reject(protocol::RejectPacket {
reason: protocol::RejectReason::ReplayDetected,
..
}))
));
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();
let backend_addr = backend.local_addr().unwrap();
let (server_config, _) = configs(45789, backend_addr.port());
let server = DirectUdpServer::bind(server_config).await.unwrap();
let observer = server.handle();
let server_task = tokio::spawn(server.run());
let client = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let secret = base64::engine::general_purpose::STANDARD
.decode(SECRET)
.unwrap();
let mut nonce = [0u8; 16];
nonce[0] = 45;
let hello = protocol::HelloPacket::authenticated(
&secret,
Bytes::from_static(b"test"),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs(),
nonce,
)
.unwrap();
client
.send_to(&hello.encode().unwrap(), "127.0.0.1:45789")
.await
.unwrap();
let mut response = [0u8; 2048];
let (len, _) = timeout(Duration::from_secs(2), client.recv_from(&mut response))
.await
.unwrap()
.unwrap();
let accept = match protocol::parse_packet(&response[..len]).unwrap() {
protocol::Packet::Accept(accept) => accept,
other => panic!("unexpected handshake response: {other:?}"),
};
let key =
protocol::derive_session_key(&secret, &nonce, &accept.server_nonce, accept.session_id);
let data = protocol::DataPacket::authenticated(&key, accept.session_id, 1, b"valid".as_slice());
let mut mutated_flags = data.encode().unwrap();
mutated_flags[2] = 1;
client
.send_to(&mutated_flags, "127.0.0.1:45789")
.await
.unwrap();
assert!(timeout(
Duration::from_millis(100),
backend.recv_from(&mut [0u8; 32])
)
.await
.is_err());
let invalid_close = protocol::SessionPacket::authenticated(
protocol::PacketType::Close,
&key,
accept.session_id,
1,
);
let mut mutated_close_flags = invalid_close.encode().unwrap();
mutated_close_flags[2] = 1;
client
.send_to(&mutated_close_flags, "127.0.0.1:45789")
.await
.unwrap();
tokio::time::sleep(Duration::from_millis(50)).await;
assert_eq!(observer.active_sessions().await, 1);
let valid_data = protocol::DataPacket::authenticated(
&key,
accept.session_id,
1,
Bytes::from_static(b"still-open"),
);
client
.send_to(&valid_data.encode().unwrap(), "127.0.0.1:45789")
.await
.unwrap();
let mut backend_buffer = [0u8; 64];
let (len, _) = timeout(
Duration::from_secs(2),
backend.recv_from(&mut backend_buffer),
)
.await
.unwrap()
.unwrap();
assert_eq!(&backend_buffer[..len], b"still-open");
let valid_close = protocol::SessionPacket::authenticated(
protocol::PacketType::Close,
&key,
accept.session_id,
2,
);
client
.send_to(&valid_close.encode().unwrap(), "127.0.0.1:45789")
.await
.unwrap();
wait_for_active_sessions(&observer, 0).await;
server_task.abort();
}
#[tokio::test]
async fn unavailable_backend_does_not_terminate_listener() {
let (server_config, client_config) = configs(45786, 9);
tests/http3.rs
+32 -7
View File
@@ -1,12 +1,10 @@
#[path = "../src/transport/http3.rs"]
mod http3;
use std::collections::HashMap;
use http3::{
decode_capsule, decode_http_datagram, encode_auth_challenge, encode_auth_response,
encode_capsule, encode_http_datagram, validate_connect_udp_request, AuthChallenge,
ConnectUdpRequest, Http3Error, CONNECT_UDP_CONTEXT_ID,
use wgcloak::transport::http3;
use wgcloak::transport::http3::{
decode_auth_challenge, decode_capsule, decode_http_datagram, encode_auth_challenge,
encode_auth_response, encode_capsule, encode_http_datagram, validate_connect_udp_request,
AuthChallenge, ConnectUdpRequest, Http3Error, CONNECT_UDP_CONTEXT_ID,
};
#[test]
@@ -78,6 +76,10 @@ fn authentication_is_completed_before_session_activation() {
let nonce = [5u8; 16];
let secret = [7u8; 32];
let challenge_capsule = encode_auth_challenge(&challenge).unwrap();
assert_eq!(
decode_auth_challenge(&challenge_capsule).unwrap(),
challenge
);
assert_eq!(
decode_capsule(&challenge_capsule).unwrap().0,
http3::AUTH_CHALLENGE_CAPSULE
@@ -94,3 +96,26 @@ fn authentication_is_completed_before_session_activation() {
assert!(http3::is_auth_accept(&http3::encode_auth_accept().unwrap()).unwrap());
assert!(!http3::is_auth_accept(&http3::encode_auth_reject().unwrap()).unwrap());
}
#[test]
fn datagram_limit_includes_http_context_and_request_stream_overhead() {
use h3::quic::StreamId;
let limit = http3::maximum_application_datagram(
1200,
StreamId::try_from(0).unwrap(),
CONNECT_UDP_CONTEXT_ID,
1200,
)
.unwrap();
assert_eq!(limit, 1198);
assert!(matches!(
http3::maximum_application_datagram(
2,
StreamId::try_from(0).unwrap(),
CONNECT_UDP_CONTEXT_ID,
1200
),
Ok(0)
));
}
tests/http3_integration.rs
+272
View File
@@ -0,0 +1,272 @@
use std::{collections::HashMap, fs, process::Command, sync::Arc, time::Duration};
use base64::Engine;
use bytes::Bytes;
use quinn::{ClientConfig, Endpoint, ServerConfig};
use rustls::pki_types::{pem::PemObject, CertificateDer, PrivateKeyDer};
use tokio::time::timeout;
use wgcloak::config;
use wgcloak::transport::{
http3::{Http3ClientConnection, Http3Error, Http3Server, Http3ServerConnection},
AdmissionController, TransportSession,
};
const SECRET: &str = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
const WRONG_SECRET: &str = "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB";
fn test_endpoints() -> (
tempfile::TempDir,
Endpoint,
ClientConfig,
std::net::SocketAddr,
) {
let directory = tempfile::tempdir().unwrap();
let certificate = directory.path().join("localhost-cert.pem");
let private_key = directory.path().join("localhost-key.pem");
let result = Command::new("openssl")
.args([
"req",
"-x509",
"-newkey",
"rsa:2048",
"-nodes",
"-keyout",
private_key.to_str().unwrap(),
"-out",
certificate.to_str().unwrap(),
"-subj",
"/CN=localhost",
"-addext",
"subjectAltName=DNS:localhost",
"-addext",
"basicConstraints=critical,CA:FALSE",
"-addext",
"keyUsage=critical,digitalSignature,keyEncipherment",
"-addext",
"extendedKeyUsage=serverAuth",
"-days",
"1",
])
.output()
.unwrap();
assert!(
result.status.success(),
"openssl failed: {}",
String::from_utf8_lossy(&result.stderr)
);
let cert_pem = fs::read(&certificate).unwrap();
let certs: Vec<CertificateDer<'static>> = CertificateDer::pem_slice_iter(&cert_pem)
.collect::<Result<_, _>>()
.unwrap();
let key_pem = fs::read(&private_key).unwrap();
let key = PrivateKeyDer::from_pem_slice(&key_pem).unwrap();
let mut server_config = ServerConfig::with_single_cert(certs.clone(), key).unwrap();
let mut server_transport = quinn::TransportConfig::default();
server_transport.datagram_send_buffer_size(1024 * 1024);
server_config.transport_config(Arc::new(server_transport));
let server = Endpoint::server(server_config, "127.0.0.1:0".parse().unwrap()).unwrap();
let server_address = server.local_addr().unwrap();
let mut roots = rustls::RootCertStore::empty();
roots.add(certs[0].clone()).unwrap();
let mut client_config = ClientConfig::with_root_certificates(Arc::new(roots)).unwrap();
let mut client_transport = quinn::TransportConfig::default();
client_transport.datagram_send_buffer_size(1024 * 1024);
client_config.transport_config(Arc::new(client_transport));
(directory, server, client_config, server_address)
}
async fn connect_client(address: std::net::SocketAddr, config: ClientConfig) -> quinn::Connection {
let mut endpoint = Endpoint::client("[::]:0".parse().unwrap()).unwrap();
endpoint.set_default_client_config(config);
endpoint
.connect(address, "localhost")
.unwrap()
.await
.unwrap()
}
#[tokio::test]
async fn connect_udp_authenticates_then_exchanges_contextual_http_datagrams() {
let (_certificate_dir, server_endpoint, client_tls, address) = test_endpoints();
let mut clients = HashMap::new();
clients.insert(
"laptop".to_owned(),
base64::engine::general_purpose::STANDARD
.decode(SECRET)
.unwrap(),
);
let server_task = tokio::spawn(async move {
let incoming = server_endpoint.accept().await.unwrap();
let quic_connection = incoming.await.unwrap();
Http3ServerConnection::new(quic_connection)
.await
.unwrap()
.accept_connect_udp("/tunnel", &clients, 1200, Duration::from_secs(3))
.await
.unwrap()
});
let quic_connection = connect_client(address, client_tls).await;
let client = Http3ClientConnection::new(quic_connection).await.unwrap();
let client_session = client
.connect_udp(
&format!("localhost:{}", address.port()),
"/tunnel",
"laptop",
&base64::engine::general_purpose::STANDARD
.decode(SECRET)
.unwrap(),
1200,
Duration::from_secs(3),
)
.await
.unwrap();
let server_session = timeout(Duration::from_secs(3), server_task)
.await
.unwrap()
.unwrap();
assert_eq!(
client_session.max_datagram_size(),
server_session.max_datagram_size()
);
assert!(client_session.max_datagram_size() < 1200);
client_session
.send_datagram(b"opaque-upstream")
.await
.unwrap();
assert_eq!(
timeout(Duration::from_secs(2), server_session.recv_datagram())
.await
.unwrap()
.unwrap(),
Bytes::from_static(b"opaque-upstream")
);
server_session
.send_datagram(b"opaque-downstream")
.await
.unwrap();
assert_eq!(
timeout(Duration::from_secs(2), client_session.recv_datagram())
.await
.unwrap()
.unwrap(),
Bytes::from_static(b"opaque-downstream")
);
let too_large = vec![0; client_session.max_datagram_size() + 1];
assert!(matches!(
client_session.send_datagram(&too_large).await,
Err(Http3Error::Oversized { .. })
));
client_session.close(Duration::from_secs(2)).await.unwrap();
assert!(matches!(
client_session.send_datagram(b"after-close").await,
Err(Http3Error::Closed)
));
}
#[tokio::test]
async fn connect_udp_rejects_bad_secret_before_returning_session() {
let (_certificate_dir, server_endpoint, client_tls, address) = test_endpoints();
let mut clients = HashMap::new();
clients.insert(
"laptop".to_owned(),
base64::engine::general_purpose::STANDARD
.decode(SECRET)
.unwrap(),
);
let server_task = tokio::spawn(async move {
let incoming = server_endpoint.accept().await.unwrap();
let quic_connection = incoming.await.unwrap();
Http3ServerConnection::new(quic_connection)
.await
.unwrap()
.accept_connect_udp("/tunnel", &clients, 1200, Duration::from_secs(3))
.await
});
let quic_connection = connect_client(address, client_tls).await;
let client = Http3ClientConnection::new(quic_connection).await.unwrap();
let client_result = client
.connect_udp(
&format!("localhost:{}", address.port()),
"/tunnel",
"laptop",
&base64::engine::general_purpose::STANDARD
.decode(WRONG_SECRET)
.unwrap(),
1200,
Duration::from_secs(3),
)
.await;
assert!(
matches!(client_result, Err(Http3Error::AuthenticationRejected)),
"unexpected client error: {:?}",
client_result.as_ref().err()
);
let server_result = timeout(Duration::from_secs(3), server_task)
.await
.unwrap()
.unwrap();
assert!(
matches!(server_result, Err(Http3Error::AuthenticationRejected)),
"unexpected server error: {:?}",
server_result.as_ref().err()
);
}
#[tokio::test]
async fn http3_server_admission_permit_tracks_authenticated_session_lifetime() {
let (_certificate_dir, endpoint, client_tls, address) = test_endpoints();
let mut clients = HashMap::new();
clients.insert(
"laptop".to_owned(),
base64::engine::general_purpose::STANDARD
.decode(SECRET)
.unwrap(),
);
let limits = config::LimitsConfig {
max_sessions: 1,
max_sessions_per_ip: 1,
..config::LimitsConfig::default()
};
let admission = AdmissionController::new(&limits);
let server = Arc::new(Http3Server::from_endpoint_with_admission(
endpoint,
"/tunnel",
clients,
1200,
Duration::from_secs(3),
admission.clone(),
));
let accept_server = server.clone();
let server_task = tokio::spawn(async move { accept_server.accept_session().await.unwrap() });
let quic_connection = connect_client(address, client_tls).await;
let client = Http3ClientConnection::new(quic_connection).await.unwrap();
let client_session = client
.connect_udp(
&format!("localhost:{}", address.port()),
"/tunnel",
"laptop",
&base64::engine::general_purpose::STANDARD
.decode(SECRET)
.unwrap(),
1200,
Duration::from_secs(3),
)
.await
.unwrap();
let server_session = timeout(Duration::from_secs(3), server_task)
.await
.unwrap()
.unwrap();
assert_eq!(admission.active_sessions(), 1);
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();
}
tests/http3_runtime.rs
+394
View File
@@ -0,0 +1,394 @@
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();
}
tests/integration_transports.rs
+95 -18
View File
@@ -1,22 +1,16 @@
#[path = "../src/config.rs"]
mod config;
#[path = "../src/datagram.rs"]
mod datagram;
#[path = "../src/protocol/mod.rs"]
mod protocol;
#[path = "../src/stats.rs"]
mod stats;
#[path = "../src/transport/mod.rs"]
mod transport;
use std::{process::Command, sync::Arc, time::Duration};
use bytes::Bytes;
use config::Config;
use rustls::RootCertStore;
use tempfile::tempdir;
use tokio::time::timeout;
use transport::{TlsTcpConnector, TlsTcpServer, TransportConnector, TransportSession};
use wgcloak::{
config::Config,
protocol, transport,
transport::{
AdmissionController, TlsTcpConnector, TlsTcpServer, TransportConnector, TransportError,
},
};
const SECRET: &str = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
@@ -29,7 +23,7 @@ fn free_port() -> u16 {
}
#[tokio::test]
async fn tls_tcp_authenticates_and_round_trips_opaque_datagrams() {
async fn tls_tcp_authenticates_round_trips_and_classifies_failures() {
if Command::new("openssl").arg("version").output().is_err() {
return;
}
@@ -45,6 +39,10 @@ async fn tls_tcp_authenticates_and_round_trips_opaque_datagrams() {
"-nodes",
"-subj",
"/CN=localhost",
"-addext",
"subjectAltName=DNS:localhost",
"-addext",
"basicConstraints=critical,CA:FALSE",
"-days",
"1",
"-keyout",
@@ -61,7 +59,7 @@ async fn tls_tcp_authenticates_and_round_trips_opaque_datagrams() {
);
let port = free_port();
let server_config = Config::from_str(&format!(
"[server]\nlisten = \"127.0.0.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",
"[server]\nlisten = \"127.0.0.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[limits]\nhandshake_timeout_ms = 500\n[tls]\ncertificate = \"{}\"\nprivate_key = \"{}\"\n[[auth.clients]]\nid = \"test\"\nsecret = \"{SECRET}\"\n",
cert.display(),
key.display()
))
@@ -69,23 +67,47 @@ async fn tls_tcp_authenticates_and_round_trips_opaque_datagrams() {
.into_server()
.unwrap();
let client_config = Config::from_str(&format!(
"[client]\nserver = \"localhost\"\nid = \"test\"\nsecret = \"{SECRET}\"\n[transport]\nmode = \"tls_tcp\"\n[transport.tls_tcp]\nport = {port}\n[network]\nmax_datagram_size = 1200\n"
"[client]\nserver = \"localhost\"\nid = \"test\"\nsecret = \"{SECRET}\"\n[transport]\nmode = \"tls_tcp\"\n[transport.tls_tcp]\nport = {port}\nkeepalive_secs = 1\n[network]\nmax_datagram_size = 1200\n"
))
.unwrap()
.into_client()
.unwrap();
let mismatch_config = client_config.clone();
let certs = transport::tls_tcp::load_certificates(&cert).unwrap();
let mut roots = RootCertStore::empty();
roots.add(certs[0].clone()).unwrap();
let client_tls = (*transport::tls_tcp::client_tls_config(roots)).clone();
let server = Arc::new(TlsTcpServer::bind(&server_config).await.unwrap());
let admission = AdmissionController::new(&server_config.limits);
let server = Arc::new(
TlsTcpServer::bind_with_admission(&server_config, admission.clone())
.await
.unwrap(),
);
let slow_server = server.clone();
let slow_handshake = tokio::spawn(async move { slow_server.accept_session().await });
let slow_peer = tokio::net::TcpStream::connect(server.local_addr().unwrap())
.await
.unwrap();
tokio::time::sleep(Duration::from_millis(50)).await;
let accept = {
let server = server.clone();
tokio::spawn(async move { server.accept_session().await })
};
let client = TlsTcpConnector::with_client_config(client_config, client_tls);
let client = TlsTcpConnector::with_client_config(client_config, client_tls.clone());
let client_session = client.connect().await.unwrap();
let server_session = accept.await.unwrap().unwrap();
assert_eq!(
client_session.authenticated_keepalive_interval(),
Some(Duration::from_secs(1))
);
assert_eq!(client_session.rtt(), None);
assert_eq!(
client_session.remote_addr().unwrap().ip(),
"127.0.0.1".parse::<std::net::IpAddr>().unwrap()
);
assert_eq!(admission.active_sessions(), 1);
assert_eq!(admission.pending_handshakes(), 1);
client_session.send_keepalive().await.unwrap();
client_session
.send_datagram(Bytes::from_static(b"tls-opaque"))
.await
@@ -99,4 +121,59 @@ async fn tls_tcp_authenticates_and_round_trips_opaque_datagrams() {
);
client_session.close().await.unwrap();
server_session.close().await.unwrap();
assert!(matches!(
timeout(Duration::from_secs(1), slow_handshake)
.await
.unwrap()
.unwrap(),
Err(TransportError::HandshakeTimeout)
));
drop(slow_peer);
assert_eq!(admission.active_sessions(), 0);
assert_eq!(admission.pending_handshakes(), 0);
let mut wrong_credentials = mismatch_config.clone();
wrong_credentials.client.secret.secret =
Some("BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB".to_owned());
let server_accept = {
let server = server.clone();
tokio::spawn(async move { server.accept_session().await })
};
let wrong_client = TlsTcpConnector::with_client_config(wrong_credentials, client_tls.clone());
assert!(matches!(
timeout(Duration::from_secs(2), wrong_client.connect())
.await
.unwrap(),
Err(TransportError::AuthenticationRejected(
protocol::RejectReason::AuthenticationFailed
))
));
assert!(matches!(
timeout(Duration::from_secs(2), server_accept)
.await
.unwrap()
.unwrap(),
Err(TransportError::AuthenticationRejected(
protocol::RejectReason::AuthenticationFailed
))
));
let server_accept = {
let server = server.clone();
tokio::spawn(async move { server.accept_session().await })
};
let mut mismatch_config = mismatch_config;
mismatch_config.client.server = "127.0.0.1".to_owned();
let mismatch_client = TlsTcpConnector::with_client_config(mismatch_config, client_tls);
assert!(matches!(
timeout(Duration::from_secs(2), mismatch_client.connect())
.await
.unwrap(),
Err(TransportError::Io(_))
));
assert!(timeout(Duration::from_secs(2), server_accept)
.await
.unwrap()
.unwrap()
.is_err());
}
tests/limits.rs
+141
View File
@@ -0,0 +1,141 @@
use std::{net::SocketAddr, sync::Arc, thread, time::Duration};
use wgcloak::{
config::{self, LimitsConfig},
transport::{AdmissionController, AdmissionError},
};
fn limits(
max_sessions: usize,
max_sessions_per_ip: usize,
max_pending_handshakes: usize,
max_auth_attempts_per_ip: usize,
auth_attempt_window_secs: u64,
) -> LimitsConfig {
LimitsConfig {
max_sessions,
max_sessions_per_ip,
max_pending_handshakes,
max_auth_attempts_per_ip,
auth_attempt_window_secs,
handshake_timeout_ms: 1_000,
}
}
fn address(ip: &str) -> SocketAddr {
format!("{ip}:40000").parse().unwrap()
}
#[test]
fn pending_limit_and_failed_handshake_release_reservation() {
let controller = AdmissionController::new(&limits(8, 4, 1, 100, 60));
let first = controller.begin(address("192.0.2.1")).unwrap();
assert_eq!(controller.pending_handshakes(), 1);
assert!(matches!(
controller.begin(address("192.0.2.2")),
Err(AdmissionError::Busy)
));
// Dropping the pending guard models any failed or timed-out handshake.
drop(first);
assert_eq!(controller.pending_handshakes(), 0);
let second = controller.begin(address("192.0.2.2")).unwrap();
drop(second);
assert_eq!(controller.pending_handshakes(), 0);
}
#[test]
fn active_and_per_ip_limits_release_on_close_or_drop() {
let controller = AdmissionController::new(&limits(2, 1, 2, 100, 60));
let first = controller.begin(address("192.0.2.1")).unwrap().promote();
assert_eq!(controller.active_sessions(), 1);
assert!(matches!(
controller.begin(address("192.0.2.1")),
Err(AdmissionError::Busy)
));
let second = controller.begin(address("192.0.2.2")).unwrap().promote();
assert_eq!(controller.active_sessions(), 2);
assert!(matches!(
controller.begin(address("192.0.2.3")),
Err(AdmissionError::Busy)
));
drop(first);
assert_eq!(controller.active_sessions(), 1);
drop(second);
assert_eq!(controller.active_sessions(), 0);
assert!(controller.begin(address("192.0.2.3")).is_ok());
}
#[test]
fn concurrent_admissions_never_exceed_global_or_per_ip_capacity() {
let controller = AdmissionController::new(&limits(3, 3, 32, 100, 60));
let barrier = Arc::new(std::sync::Barrier::new(33));
let (sender, receiver) = std::sync::mpsc::channel();
let mut workers = Vec::new();
for _ in 0..32 {
let controller = controller.clone();
let barrier = barrier.clone();
let sender = sender.clone();
workers.push(thread::spawn(move || {
barrier.wait();
let permit = controller
.begin(address("192.0.2.10"))
.ok()
.map(|pending| pending.promote());
barrier.wait();
sender.send(permit).unwrap();
}));
}
barrier.wait();
barrier.wait();
drop(sender);
let permits: Vec<_> = receiver.into_iter().flatten().collect();
for worker in workers {
worker.join().unwrap();
}
assert_eq!(permits.len(), 3);
assert_eq!(controller.active_sessions(), 3);
}
#[tokio::test]
async fn authentication_attempt_limit_is_per_ip_and_expires() {
let controller = AdmissionController::new(&limits(8, 4, 8, 2, 1));
for _ in 0..2 {
drop(controller.begin(address("192.0.2.1")).unwrap());
}
assert!(matches!(
controller.begin(address("192.0.2.1")),
Err(AdmissionError::RateLimited)
));
assert!(controller.begin(address("192.0.2.2")).is_ok());
tokio::time::sleep(Duration::from_millis(1_050)).await;
assert!(controller.begin(address("192.0.2.1")).is_ok());
}
#[test]
fn server_limits_have_safe_defaults_and_reject_out_of_bounds_values() {
let defaults = LimitsConfig::default();
assert_eq!(defaults.max_sessions, 1024);
assert_eq!(defaults.max_sessions_per_ip, 32);
assert_eq!(defaults.max_pending_handshakes, 256);
assert_eq!(defaults.max_auth_attempts_per_ip, 20);
assert_eq!(defaults.auth_attempt_window_secs, 60);
assert_eq!(defaults.handshake_timeout_ms, 3_000);
let valid = "[server]\nlisten='127.0.0.1'\n[backend]\ntarget='127.0.0.1:51820'\n[[auth.clients]]\nid='one'\nsecret='AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'\n";
assert!(config::Config::from_str(valid).unwrap().validate().is_ok());
let invalid = format!("{valid}\n[limits]\nmax_auth_attempts_per_ip=0\n");
assert!(config::Config::from_str(&invalid)
.unwrap()
.validate()
.is_err());
let too_long = format!("{valid}\n[limits]\nhandshake_timeout_ms=300001\n");
assert!(config::Config::from_str(&too_long)
.unwrap()
.validate()
.is_err());
}
tests/quic.rs
+32 -16
View File
@@ -1,14 +1,3 @@
#[path = "../src/config.rs"]
mod config;
#[path = "../src/datagram.rs"]
mod datagram;
#[path = "../src/protocol/mod.rs"]
mod protocol;
#[path = "../src/stats.rs"]
mod stats;
#[path = "../src/transport/mod.rs"]
mod transport;
use std::{collections::HashMap, sync::Arc, time::Duration};
use base64::Engine;
@@ -16,9 +5,12 @@ use bytes::Bytes;
use quinn::{ClientConfig as QuinnClientConfig, Endpoint, ServerConfig as QuinnServerConfig};
use rustls::pki_types::{CertificateDer, PrivatePkcs8KeyDer};
use tokio::time::timeout;
use transport::{
quic::{QuicClientConnector, QuicServer},
TransportError, TransportSession,
use wgcloak::{
config,
transport::{
quic::{QuicClientConnector, QuicServer},
AdmissionController, TransportError, TransportSession,
},
};
const SECRET: &str = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
@@ -71,12 +63,34 @@ async fn authenticated_datagram_round_trip_and_limit() {
.decode(SECRET)
.unwrap(),
);
let server = QuicServer::from_endpoint(endpoint, clients, 1200, Duration::from_secs(2));
let port = server.local_addr().unwrap().port();
let admission = AdmissionController::new(&config::LimitsConfig::default());
let server = QuicServer::from_endpoint_with_admission(
endpoint,
clients,
1200,
Duration::from_secs(2),
admission.clone(),
);
let server_addr = server.local_addr().unwrap();
let port = server_addr.port();
let client = QuicClientConnector::with_client_config(client_config(port, SECRET), client_tls);
let server_task = tokio::spawn(async move { server.accept().await });
let client_session = client.connect_session().await.unwrap();
let server_session = server_task.await.unwrap().unwrap();
let remote_addr = client_session.remote_addr().unwrap();
assert_eq!(remote_addr.port(), server_addr.port());
assert!(match remote_addr.ip() {
std::net::IpAddr::V4(address) => address.is_loopback(),
std::net::IpAddr::V6(address) => {
address.is_loopback()
|| address
.to_ipv4_mapped()
.map(|mapped| mapped.is_loopback())
.unwrap_or(false)
}
});
assert!(client_session.rtt().is_some());
assert_eq!(client_session.authenticated_keepalive_interval(), None);
client_session
.send_datagram(Bytes::from_static(b"opaque"))
.await
@@ -88,6 +102,7 @@ async fn authenticated_datagram_round_trip_and_limit() {
.unwrap(),
Bytes::from_static(b"opaque")
);
assert_eq!(admission.active_sessions(), 1);
assert!(matches!(
client_session
.send_datagram(Bytes::from(vec![0; 1201]))
@@ -96,6 +111,7 @@ async fn authenticated_datagram_round_trip_and_limit() {
));
client_session.close().await.unwrap();
server_session.close().await.unwrap();
assert_eq!(admission.active_sessions(), 0);
}
#[tokio::test]
tests/reconnect.rs
+161
View File
@@ -0,0 +1,161 @@
use std::{path::Path, time::Duration};
use tempfile::tempdir;
use tokio::{
net::UdpSocket,
time::{sleep, timeout},
};
use wgcloak::{
client,
config::{self, Config},
control, server,
};
const SECRET: &str = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
fn free_udp_port() -> u16 {
let socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
socket.local_addr().unwrap().port()
}
fn server_config(port: u16, backend: std::net::SocketAddr) -> config::ServerConfig {
Config::from_str(&format!(
"[server]\nlisten='127.0.0.1'\n[backend]\ntarget='{backend}'\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='reconnect-test'\nsecret='{SECRET}'\n"
))
.unwrap()
.into_server()
.unwrap()
}
fn client_config(port: u16, local_port: u16) -> config::ClientConfig {
Config::from_str(&format!(
"[client]\nserver='127.0.0.1'\nid='reconnect-test'\nsecret='{SECRET}'\n[local]\nlisten='127.0.0.1:{local_port}'\n[transport]\nmode='direct_udp'\n[transport.direct_udp]\nport={port}\n[transport.auto]\nprobe_timeout_ms=150\nretries=1\n"
))
.unwrap()
.into_client()
.unwrap()
}
async fn wait_for_contains(path: &Path, command: &str, needle: &str) -> String {
timeout(Duration::from_secs(10), async {
loop {
if let Ok(response) = control::send_command(path, command).await {
if response.contains(needle) {
return response;
}
}
sleep(Duration::from_millis(20)).await;
}
})
.await
.unwrap_or_else(|_| panic!("control {command} did not contain {needle:?}"))
}
async fn wait_for_reconnected_stats(path: &Path) -> String {
timeout(Duration::from_secs(10), async {
loop {
if let Ok(response) = control::send_command(path, "stats").await {
if response.contains("\"state\":\"connected\"")
&& response.contains("\"reconnections\":1")
{
return response;
}
}
sleep(Duration::from_millis(20)).await;
}
})
.await
.expect("client did not publish its connected reconnect state")
}
async fn wait_for_echo(app: &UdpSocket, local_port: u16, payload: &[u8]) {
let mut response = [0u8; 128];
app.send_to(payload, ("127.0.0.1", local_port))
.await
.unwrap();
let (len, _) = timeout(Duration::from_secs(5), app.recv_from(&mut response))
.await
.expect("application echo timed out")
.unwrap();
assert_eq!(&response[..len], payload);
}
#[tokio::test]
async fn client_recovers_through_backoff_after_server_restart() {
let directory = tempdir().unwrap();
let backend = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let backend_addr = backend.local_addr().unwrap();
let backend_task = tokio::spawn(async move {
let mut packet = vec![0u8; 2048];
while let Ok((len, peer)) = backend.recv_from(&mut packet).await {
let _ = backend.send_to(&packet[..len], peer).await;
}
});
let port = free_udp_port();
let local_port = free_udp_port();
let runtime_dir = directory.path().join("runtime");
let server_control = runtime_dir.join("server.sock");
let client_control = runtime_dir.join("client.sock");
let server_task = tokio::spawn(server::run_with_control_path(
server_config(port, backend_addr),
server_control.clone(),
));
wait_for_contains(&server_control, "status", "transports=direct_udp").await;
let client_task = tokio::spawn(client::run_with_http3_connector_and_control_path(
client_config(port, local_port),
client_control.clone(),
None,
));
wait_for_contains(&client_control, "status", "connected; server=127.0.0.1").await;
let app = UdpSocket::bind("127.0.0.1:0").await.unwrap();
wait_for_echo(&app, local_port, b"before-restart").await;
let stopped = control::send_command(&server_control, "shutdown")
.await
.unwrap();
assert!(stopped.contains("\"ok\":true"));
timeout(Duration::from_secs(5), server_task)
.await
.expect("first server instance should stop")
.unwrap()
.unwrap();
let requested = control::send_command(&client_control, "reconnect")
.await
.unwrap();
assert!(requested.contains("\"ok\":true"));
// Wait until the client has made a failed connection attempt and entered
// jittered backoff before starting the replacement server. This exercises
// the daemon loop without relying on a guessed sleep duration.
wait_for_contains(&client_control, "status", "error=").await;
let replacement = tokio::spawn(server::run_with_control_path(
server_config(port, backend_addr),
server_control.clone(),
));
wait_for_contains(&server_control, "status", "transports=direct_udp").await;
let _stats = wait_for_reconnected_stats(&client_control).await;
wait_for_echo(&app, local_port, b"after-restart").await;
let stopped = control::send_command(&server_control, "shutdown")
.await
.unwrap();
assert!(stopped.contains("\"ok\":true"));
timeout(Duration::from_secs(5), replacement)
.await
.expect("replacement server should stop")
.unwrap()
.unwrap();
let stopped = control::send_command(&client_control, "shutdown")
.await
.unwrap();
assert!(stopped.contains("\"ok\":true"));
timeout(Duration::from_secs(5), client_task)
.await
.expect("client should stop")
.unwrap()
.unwrap();
backend_task.abort();
}
tests/runtime_lifecycle.rs
+146
View File
@@ -0,0 +1,146 @@
use std::{os::unix::fs::PermissionsExt, path::Path};
use tempfile::tempdir;
use tokio::time::Duration;
use tokio::{net::UdpSocket, time::timeout};
use wgcloak::{
config::Config,
control, server,
transport::{direct_udp::DirectUdpSession, TransportSession},
};
const DIRECT_SECRET: &str = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
fn private_runtime_dir(root: &Path) -> std::path::PathBuf {
let path = root.join("runtime");
std::fs::create_dir(&path).unwrap();
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700)).unwrap();
path
}
#[tokio::test]
async fn control_commands_drive_only_supported_runtime_actions() {
let dir = tempdir().unwrap();
let path = private_runtime_dir(dir.path()).join("runtime.sock");
let state = control::ControlState::default();
state.enable_reconnect();
let server = control::ControlServer::bind(&path, state.clone())
.await
.unwrap();
let task = tokio::spawn(server.run());
state.set_status("connected; transport=direct_udp").await;
state
.set_stats_json("{\"datagrams_sent\":7,\"datagrams_dropped\":1}")
.await;
let status = control::send_command(&path, "status").await.unwrap();
assert!(status.contains("connected; transport=direct_udp"));
let stats = control::send_command(&path, "stats").await.unwrap();
assert!(stats.contains("\"datagrams_sent\":7"));
let reconnect = state.reconnect.notified();
tokio::pin!(reconnect);
let response = control::send_command(&path, "reconnect").await.unwrap();
assert!(response.contains("\"ok\":true"));
timeout(Duration::from_secs(1), &mut reconnect)
.await
.expect("reconnect should reach the running client loop");
let response = control::send_command(&path, "reload").await.unwrap();
assert!(response.contains("\"ok\":false"));
assert!(response.contains("not supported"));
let response = control::send_command(&path, "shutdown").await.unwrap();
assert!(response.contains("\"ok\":true"));
timeout(Duration::from_secs(1), task)
.await
.expect("control server should stop on shutdown")
.unwrap()
.unwrap();
}
#[tokio::test]
async fn server_control_rejects_client_only_reconnect() {
let dir = tempdir().unwrap();
let path = private_runtime_dir(dir.path()).join("server.sock");
let state = control::ControlState::default();
let server = control::ControlServer::bind(&path, state.clone())
.await
.unwrap();
let task = tokio::spawn(server.run());
let response = control::send_command(&path, "reconnect").await.unwrap();
assert!(response.contains("\"ok\":false"));
assert!(response.contains("not supported"));
state.shutdown.cancel();
timeout(Duration::from_secs(1), task)
.await
.expect("control server should stop")
.unwrap()
.unwrap();
}
#[tokio::test]
async fn direct_server_control_stats_track_authenticated_session_lifecycle() {
let probe = UdpSocket::bind("127.0.0.1:0").await.unwrap();
let port = probe.local_addr().unwrap().port();
drop(probe);
let config = Config::from_str(&format!(
"[server]\nlisten='127.0.0.1'\n[backend]\ntarget='127.0.0.1:9'\n[transports.direct_udp]\nport={port}\n[[auth.clients]]\nid='runtime-test'\nsecret='{DIRECT_SECRET}'\n"
))
.unwrap()
.into_server()
.unwrap();
let client_config = Config::from_str(&format!(
"[client]\nserver='127.0.0.1'\nid='runtime-test'\nsecret='{DIRECT_SECRET}'\n[transport.direct_udp]\nport={port}\n"
))
.unwrap()
.into_client()
.unwrap();
let dir = tempdir().unwrap();
let control_path = private_runtime_dir(dir.path()).join("direct-runtime.sock");
let server_task = tokio::spawn(server::run_with_control_path(config, control_path.clone()));
let session = timeout(
Duration::from_secs(3),
DirectUdpSession::connect(&client_config),
)
.await
.expect("direct handshake should complete")
.unwrap();
wait_for_active_count(&control_path, 1).await;
let status = control::send_command(&control_path, "status")
.await
.unwrap();
assert!(status.contains("active_sessions=1"));
session.close().await.unwrap();
wait_for_active_count(&control_path, 0).await;
let response = control::send_command(&control_path, "shutdown")
.await
.unwrap();
assert!(response.contains("\"ok\":true"));
timeout(Duration::from_secs(3), server_task)
.await
.expect("server runtime should stop")
.unwrap()
.unwrap();
}
async fn wait_for_active_count(path: &std::path::Path, expected: usize) {
timeout(Duration::from_secs(4), async {
loop {
if let Ok(response) = control::send_command(path, "stats").await {
if response.contains(&format!("\"active_sessions\":{expected}")) {
return;
}
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await
.unwrap_or_else(|_| panic!("server control stats did not report {expected} active sessions"));
}
tests/supervisor.rs
+6 -16
View File
@@ -1,18 +1,4 @@
#[path = "../src/config.rs"]
mod config;
#[path = "../src/datagram.rs"]
mod datagram;
#[path = "../src/protocol/mod.rs"]
mod protocol;
#[path = "../src/stats.rs"]
mod stats;
#[path = "../src/client/supervisor.rs"]
mod supervisor;
#[path = "../src/transport/mod.rs"]
mod transport;
use bytes::Bytes;
use config::{Config, TransportMode};
use std::{
sync::{
atomic::{AtomicUsize, Ordering},
@@ -20,9 +6,13 @@ use std::{
},
time::Duration,
};
use supervisor::{ProbeFuture, TransportProbe, TransportSupervisor};
use tokio::{sync::Mutex, time::timeout};
use transport::{TransportError, TransportFuture, TransportSession};
use wgcloak::{
client::supervisor::{self as supervisor, ProbeFuture, TransportProbe, TransportSupervisor},
config::{self, Config, TransportMode},
protocol,
transport::{TransportError, TransportFuture, TransportSession},
};
struct Session;
impl TransportSession for Session {
tests/tls_tcp.rs
+3 -5
View File
@@ -1,13 +1,11 @@
#[path = "../src/transport/tls_tcp.rs"]
mod tls_tcp;
use std::{collections::HashMap, time::Duration};
use tls_tcp::{
use tokio::io::{duplex, split};
use wgcloak::transport::tls_tcp;
use wgcloak::transport::tls_tcp::{
client_authenticate, server_authenticate, AuthChallenge, FrameCodec, FrameError, FramedReader,
FramedWriter,
};
use tokio::io::{duplex, split};
#[test]
fn codec_handles_split_and_coalesced_frames() {