diff --git a/Cargo.lock b/Cargo.lock index 6175722..9f0cb4a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -199,6 +199,41 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe" +[[package]] +name = "aead" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d122413f284cf2d62fb1b7db97e02edb8cda96d769b16e443a4f6195e35662b0" +dependencies = [ + "crypto-common", + "generic-array", +] + +[[package]] +name = "aes" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" +dependencies = [ + "cfg-if", + "cipher", + "cpufeatures", +] + +[[package]] +name = "aes-gcm" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "831010a0f742e1209b3bcea8fab6a8e149051ba6099432c8cb2cc117dec3ead1" +dependencies = [ + "aead", + "aes", + "cipher", + "ctr", + "ghash", + "subtle", +] + [[package]] name = "ahash" version = "0.8.11" @@ -350,6 +385,15 @@ version = "2.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf4b9d6a944f767f8e5e0db018570623c85f3d925ac718db4e06d0187adb21c1" +[[package]] +name = "blake2" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "46502ad458c9a52b69d4d4d32775c788b7a1b85e8bc9d482d92250fc0e3f8efe" +dependencies = [ + "digest", +] + [[package]] name = "block-buffer" version = "0.10.4" @@ -418,6 +462,41 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" +[[package]] +name = "chacha20" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3613f74bd2eac03dad61bd53dbe620703d4371614fe0bc3b9f04dd36fe4e818" +dependencies = [ + "cfg-if", + "cipher", + "cpufeatures", +] + +[[package]] +name = "chacha20poly1305" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35" +dependencies = [ + "aead", + "chacha20", + "cipher", + "poly1305", + "zeroize", +] + +[[package]] +name = "cipher" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad" +dependencies = [ + "crypto-common", + "inout", + "zeroize", +] + [[package]] name = "clap" version = "4.5.4" @@ -522,9 +601,45 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3" dependencies = [ "generic-array", + "rand_core", "typenum", ] +[[package]] +name = "ctr" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0369ee1ad671834580515889b80f2ea915f23b8be8d0daa4bbaf2ac5c7590835" +dependencies = [ + "cipher", +] + +[[package]] +name = "curve25519-dalek" +version = "4.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97fb8b7c4503de7d6ae7b42ab72a5a59857b4c937ec27a3d4539dba95b5ab2be" +dependencies = [ + "cfg-if", + "cpufeatures", + "curve25519-dalek-derive", + "fiat-crypto", + "rustc_version", + "subtle", + "zeroize", +] + +[[package]] +name = "curve25519-dalek-derive" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.65", +] + [[package]] name = "deranged" version = "0.3.11" @@ -555,6 +670,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer", "crypto-common", + "subtle", ] [[package]] @@ -626,6 +742,12 @@ version = "2.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9fc0510504f03c51ada170672ac806f1f105a88aa97a5281117e1ddc3368e51a" +[[package]] +name = "fiat-crypto" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" + [[package]] name = "flate2" version = "1.0.30" @@ -688,6 +810,17 @@ version = "0.3.30" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a44623e20b9681a318efdd71c299b6b222ed6f231972bfe2f224ebad6311f0c1" +[[package]] +name = "futures-macro" +version = "0.3.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87750cf4b7a4c0625b1529e4c543c2182106e4dedc60a2a6455e00d212c489ac" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.65", +] + [[package]] name = "futures-sink" version = "0.3.30" @@ -708,6 +841,7 @@ checksum = "3d6401deb83407ab3da39eba7e33987a73c3df0c82b4bb5813ee871c19c41d48" dependencies = [ "futures-core", "futures-io", + "futures-macro", "futures-sink", "futures-task", "memchr", @@ -737,6 +871,16 @@ dependencies = [ "wasi", ] +[[package]] +name = "ghash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0d8a4362ccb29cb0b265253fb0a2728f592895ee6854fd9bc13f2ffda266ff1" +dependencies = [ + "opaque-debug", + "polyval", +] + [[package]] name = "gimli" version = "0.28.1" @@ -799,6 +943,12 @@ version = "0.3.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d231dfb89cfffdbc30e7fc41579ed6066ad03abda9e567ccafae602b97ec5024" +[[package]] +name = "hex" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" + [[package]] name = "http" version = "0.2.12" @@ -956,6 +1106,15 @@ dependencies = [ "hashbrown", ] +[[package]] +name = "inout" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01" +dependencies = [ + "generic-array", +] + [[package]] name = "ipnet" version = "2.9.0" @@ -1035,7 +1194,7 @@ checksum = "4d873d7c67ce09b42110d801813efbc9364414e356be9935700d368351657487" [[package]] name = "localtunnel" -version = "0.1.7" +version = "0.2.0" dependencies = [ "anyhow", "clap", @@ -1048,32 +1207,38 @@ dependencies = [ [[package]] name = "localtunnel-client" -version = "0.1.7" +version = "0.2.0" dependencies = [ "anyhow", + "hex", "log", "reqwest", "serde", + "snowstorm", "socket2", "tokio", ] [[package]] name = "localtunnel-server" -version = "0.1.7" +version = "0.2.0" dependencies = [ "actix-web", "anyhow", "async-trait", "dotenv", "envy", + "hex", "hyper", "hyper-util", "lazy_static", + "localtunnel-client", "log", + "rand", "regex", "reqwest", "serde", + "snowstorm", "socket2", "thiserror", "tokio", @@ -1177,6 +1342,12 @@ version = "1.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3fdb12b2476b595f9358c5161aa467c2438859caa136dec86c26fdd2efe17b92" +[[package]] +name = "opaque-debug" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" + [[package]] name = "openssl" version = "0.10.64" @@ -1304,6 +1475,29 @@ version = "0.3.30" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d231b230927b5e4ad203db57bbcbee2802f6bce620b1e4a9024a07d94e2907ec" +[[package]] +name = "poly1305" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8159bd90725d2df49889a078b54f4f79e87f1f8a8444194cdca81d38f5393abf" +dependencies = [ + "cpufeatures", + "opaque-debug", + "universal-hash", +] + +[[package]] +name = "polyval" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1fe60d06143b2430aa532c94cfe9e29783047f06c0d7fd359a9a51b729fa25" +dependencies = [ + "cfg-if", + "cpufeatures", + "opaque-debug", + "universal-hash", +] + [[package]] name = "powerfmt" version = "0.2.0" @@ -1644,6 +1838,17 @@ dependencies = [ "digest", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "signal-hook-registry" version = "1.4.2" @@ -1668,6 +1873,38 @@ version = "1.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67" +[[package]] +name = "snow" +version = "0.9.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "850948bee068e713b8ab860fe1adc4d109676ab4c3b621fd8147f06b261f2f85" +dependencies = [ + "aes-gcm", + "blake2", + "chacha20poly1305", + "curve25519-dalek", + "rand_core", + "rustc_version", + "sha2", + "subtle", +] + +[[package]] +name = "snowstorm" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d594c1d69c0a1d316829b8241ce930d651196df03ba15e1eceaf7df71f70ee58" +dependencies = [ + "bytes", + "futures-util", + "log", + "pin-project", + "rand", + "snow", + "thiserror", + "tokio", +] + [[package]] name = "socket2" version = "0.5.7" @@ -1968,6 +2205,16 @@ dependencies = [ "tinyvec", ] +[[package]] +name = "universal-hash" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc1de2c688dc15305988b563c3854064043356019f97a4b46276fe734c4f07ea" +dependencies = [ + "crypto-common", + "subtle", +] + [[package]] name = "untrusted" version = "0.9.0" diff --git a/README.md b/README.md index 7382d59..252c689 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,18 @@ Localtunnel exposes your localhost endpoint to the world, user cases are: - multiple devices access to single data store - peer to peer connection, workaround for NAT hole punching. +## Encrypted tunnel + +Since v0.2.0 all tunnel connections between client and server are encrypted +with the [Noise protocol](https://noiseprotocol.org/) (`NK` pattern, +X25519 + ChaCha20-Poly1305), with fresh session keys per connection. The +server hands its public key and a per-tunnel session token to the client in +the registration response, and only connections that complete the handshake +and present the token can join the tunnel pool. + +This is a breaking protocol change: v0.2.0 clients and servers do not +interoperate with older releases — upgrade both sides. + ## Client Usage Known issue: *the public proxy server is down, please setup your own server.* @@ -40,6 +52,7 @@ let config = ClientConfig { shutdown_signal: notify_shutdown.clone(), max_conn: 10, credential: None, + reregister_after: None, }; let result = open_tunnel(config).await?; diff --git a/cli/Cargo.toml b/cli/Cargo.toml index 2c2a023..8bae3b0 100644 --- a/cli/Cargo.toml +++ b/cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "localtunnel" -version = "0.1.7" +version = "0.2.0" edition.workspace = true description = "A CLI to proxy with localtunnel server." license.workspace = true @@ -10,8 +10,8 @@ repository.workspace = true [dependencies] clap = { version = "4.5", features = ["derive"] } -localtunnel-client = { path = "../client", version = "0.1.6" } -localtunnel-server = { path = "../server", version = "0.1.6" } +localtunnel-client = { path = "../client", version = "0.2.0" } +localtunnel-server = { path = "../server", version = "0.2.0" } tokio = { workspace = true } log = { workspace = true } env_logger = "0.11" diff --git a/client/Cargo.toml b/client/Cargo.toml index 83fabe9..a4374af 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "localtunnel-client" -version = "0.1.7" +version = "0.2.0" edition.workspace = true description = "A client to connect with localtunnel server." license.workspace = true @@ -15,6 +15,8 @@ tokio = { workspace = true } anyhow = { workspace = true } log = { workspace = true } socket2 = { workspace = true } +snowstorm = "0.4" +hex = "0.4" [features] default = ["reqwest/default"] diff --git a/client/src/lib.rs b/client/src/lib.rs index 8b40b23..98beea2 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -4,18 +4,29 @@ use std::sync::{ }; use std::time::Instant; -use anyhow::Result; +use anyhow::{anyhow, Context, Result}; use serde::{Deserialize, Serialize}; +use snowstorm::{Builder, NoiseStream}; use socket2::{SockRef, TcpKeepalive}; -use tokio::io; +use tokio::io::{self, AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; pub use tokio::sync::broadcast; use tokio::sync::{mpsc, Semaphore}; -use tokio::time::{sleep, Duration}; +use tokio::time::{sleep, timeout, Duration}; pub const PROXY_SERVER: &str = "https://your-domain.com"; pub const LOCAL_HOST: &str = "127.0.0.1"; +/// Noise protocol parameters for the encrypted tunnel between client and +/// server. Must match the server exactly; NK authenticates the server by its +/// static key (from the registration response) and derives fresh session keys +/// per connection. +pub const NOISE_PARAMS: &str = "Noise_NK_25519_ChaChaPoly_BLAKE2s"; + +/// How long the tunnel handshake may take before the connection attempt is +/// treated as a remote failure. +const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10); + // See https://tldp.org/HOWTO/html_single/TCP-Keepalive-HOWTO to understand how keepalive work. const TCP_KEEPALIVE_TIME: Duration = Duration::from_secs(30); const TCP_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(10); @@ -36,6 +47,8 @@ struct ProxyResponse { port: u16, max_conn_count: u8, url: String, + server_public_key: String, + session_token: String, } /// The server detail for client to connect @@ -45,6 +58,11 @@ pub struct TunnelServerInfo { pub port: u16, pub max_conn_count: u8, pub url: String, + /// The server's static Noise public key, used to authenticate the tunnel + /// handshake. + pub server_public_key: Vec, + /// Per-tunnel token presented on every tunnel connection. + pub session_token: Vec, } pub struct ClientConfig { @@ -250,8 +268,7 @@ fn start_tunnel_connections( max_conn: u8, health: RoundHealth, ) { - let server_host = server.host.clone(); - let server_port = server.port; + let server = Arc::new(server.clone()); let local_host = local_host.unwrap_or_else(|| LOCAL_HOST.to_string()); let count = std::cmp::min(server.max_conn_count, max_conn); @@ -271,7 +288,7 @@ fn start_tunnel_connections( return; }, }; - let server_host = server_host.clone(); + let server = server.clone(); let local_host = local_host.clone(); let health = health.clone(); let mut shutdown_receiver = shutdown_signal.subscribe(); @@ -279,7 +296,7 @@ fn start_tunnel_connections( tokio::spawn(async move { tokio::select! { _ = tunnel_one_connection( - &server_host, server_port, + &server, &local_host, local_port, &health, ) => {} @@ -301,21 +318,47 @@ fn start_tunnel_connections( } async fn tunnel_one_connection( - server_host: &str, - server_port: u16, + server: &TunnelServerInfo, local_host: &str, local_port: u16, health: &RoundHealth, ) { - log::debug!("Connecting to remote: {}:{}", server_host, server_port); - let remote_stream = match TcpStream::connect(format!("{server_host}:{server_port}")).await { - Ok(stream) => { + log::debug!("Connecting to remote: {}:{}", server.host, server.port); + let remote_stream = match TcpStream::connect(format!("{}:{}", server.host, server.port)).await + { + Ok(stream) => stream, + Err(err) => { + let down_for = health.record_failure(); + log::error!("Remote connect failed (down for {:?}): {:?}", down_for, err); + sleep(Duration::from_secs(10)).await; + return; + } + }; + + // Keepalive has to be configured on the raw TCP socket, before the + // encrypted stream takes ownership of it. + if let Err(err) = set_keepalive(&remote_stream) { + log::warn!("failed to enable TCP keepalive: {err:?}"); + } + + // A completed handshake (server key verified, session token acknowledged) + // is the success signal for re-registration health: a TCP connect alone + // can succeed against a stale endpoint whose key or token no longer match. + let remote_stream = match timeout(HANDSHAKE_TIMEOUT, secure_connect(remote_stream, server)).await + { + Ok(Ok(stream)) => { health.record_success(); stream } - Err(err) => { + Ok(Err(err)) => { let down_for = health.record_failure(); - log::error!("Remote connect failed (down for {:?}): {:?}", down_for, err); + log::error!("Tunnel handshake failed (down for {:?}): {:?}", down_for, err); + sleep(Duration::from_secs(10)).await; + return; + } + Err(_) => { + let down_for = health.record_failure(); + log::error!("Tunnel handshake timed out (down for {:?})", down_for); sleep(Duration::from_secs(10)).await; return; } @@ -334,21 +377,51 @@ async fn tunnel_one_connection( } } -async fn proxy_through( - mut remote_stream: TcpStream, - local_host: &str, - local_port: u16, -) -> Result<()> { - log::debug!("Connecting to local: {}:{}", local_host, local_port); - let mut local_stream = TcpStream::connect(format!("{local_host}:{local_port}")).await?; +/// Establish the encrypted tunnel: Noise NK handshake pinned to the server's +/// static public key, then authenticate with the session token and wait for +/// the server's one-byte acknowledgement. Without the ack, a rejected token +/// would only surface later as a mysteriously dead proxied request. +async fn secure_connect( + stream: TcpStream, + server: &TunnelServerInfo, +) -> Result> { + let initiator = Builder::new(NOISE_PARAMS.parse()?) + .remote_public_key(&server.server_public_key) + .build_initiator()?; + let mut stream = NoiseStream::handshake(stream, initiator) + .await + .map_err(|err| anyhow!("noise handshake failed: {err:?}"))?; + + stream.write_all(&server.session_token).await?; + stream.flush().await?; + + let mut ack = [0u8; 1]; + stream + .read_exact(&mut ack) + .await + .context("server rejected the session token")?; + + Ok(stream) +} +fn set_keepalive(stream: &TcpStream) -> Result<()> { let ka = TcpKeepalive::new() .with_time(TCP_KEEPALIVE_TIME) .with_interval(TCP_KEEPALIVE_INTERVAL); #[cfg(not(target_os = "windows"))] let ka = ka.with_retries(TCP_KEEPALIVE_RETRIES); - let sf = SockRef::from(&remote_stream); + let sf = SockRef::from(stream); sf.set_tcp_keepalive(&ka)?; + Ok(()) +} + +async fn proxy_through( + mut remote_stream: NoiseStream, + local_host: &str, + local_port: u16, +) -> Result<()> { + log::debug!("Connecting to local: {}:{}", local_host, local_port); + let mut local_stream = TcpStream::connect(format!("{local_host}:{local_port}")).await?; io::copy_bidirectional(&mut remote_stream, &mut local_stream).await?; Ok(()) @@ -377,11 +450,18 @@ async fn get_tunnel_endpoint( None => host, }; + let server_public_key = hex::decode(&resp.server_public_key) + .context("invalid server_public_key in registration response")?; + let session_token = hex::decode(&resp.session_token) + .context("invalid session_token in registration response")?; + let tunnel_info = TunnelServerInfo { host: host.to_string(), port: resp.port, max_conn_count: resp.max_conn_count, url: resp.url, + server_public_key, + session_token, }; Ok(tunnel_info) diff --git a/client/tests/reregistration.rs b/client/tests/reregistration.rs index f508308..1fb22b1 100644 --- a/client/tests/reregistration.rs +++ b/client/tests/reregistration.rs @@ -3,12 +3,15 @@ use std::sync::{ Arc, }; -use localtunnel_client::{broadcast, open_tunnel, ClientConfig}; +use localtunnel_client::{broadcast, open_tunnel, ClientConfig, NOISE_PARAMS}; +use snowstorm::{Builder, Keypair, NoiseStream}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; use tokio::time::{sleep, Duration}; -async fn mock_api_server(listener: TcpListener, endpoint_port: Arc) { +const SESSION_TOKEN: [u8; 32] = [7u8; 32]; + +async fn mock_api_server(listener: TcpListener, endpoint_port: Arc, public_key: String) { loop { let (mut stream, _) = match listener.accept().await { Ok(v) => v, @@ -18,8 +21,9 @@ async fn mock_api_server(listener: TcpListener, endpoint_port: Arc) { let _ = stream.read(&mut buf).await; let port = endpoint_port.load(Ordering::Relaxed); + let token = hex::encode(SESSION_TOKEN); let body = format!( - r#"{{"id":"test","port":{port},"max_conn_count":10,"url":"http://test.127.0.0.1:{port}"}}"#, + r#"{{"id":"test","port":{port},"max_conn_count":10,"url":"http://test.127.0.0.1:{port}","server_public_key":"{public_key}","session_token":"{token}"}}"#, ); let response = format!( "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", @@ -40,8 +44,44 @@ async fn accept_and_count(listener: TcpListener, counter: Arc) { } } +// A tunnel endpoint that completes the encrypted handshake (so the client +// counts the endpoint as healthy) and then drops the stream, mirroring the +// old drop-on-accept behaviour of the plain-TCP mock. +async fn noise_accept_and_count(listener: TcpListener, counter: Arc, key: Arc) { + loop { + match listener.accept().await { + Ok((stream, _)) => { + counter.fetch_add(1, Ordering::Relaxed); + let key = key.clone(); + tokio::spawn(async move { + let responder = Builder::new(NOISE_PARAMS.parse().unwrap()) + .local_private_key(&key.private) + .build_responder() + .unwrap(); + if let Ok(mut stream) = NoiseStream::handshake(stream, responder).await { + let mut token = [0u8; 32]; + if stream.read_exact(&mut token).await.is_ok() { + let _ = stream.write_all(&[1]).await; + let _ = stream.flush().await; + } + } + }); + } + Err(_) => return, + } + } +} + #[tokio::test] async fn reregistration_on_remote_failure() { + // Server-side Noise identity, shared by both mock tunnel endpoints. + let key = Arc::new( + Builder::new(NOISE_PARAMS.parse().unwrap()) + .generate_keypair() + .unwrap(), + ); + let public_key = hex::encode(&key.public); + // Local server (simulates the application behind the tunnel) let local = TcpListener::bind("127.0.0.1:0").await.unwrap(); let local_port = local.local_addr().unwrap().port(); @@ -51,19 +91,27 @@ async fn reregistration_on_remote_failure() { let remote1 = TcpListener::bind("127.0.0.1:0").await.unwrap(); let remote1_port = remote1.local_addr().unwrap().port(); let remote1_count = Arc::new(AtomicU32::new(0)); - let remote1_task = tokio::spawn(accept_and_count(remote1, remote1_count.clone())); + let remote1_task = tokio::spawn(noise_accept_and_count( + remote1, + remote1_count.clone(), + key.clone(), + )); // Remote endpoint 2 (ready before remote1 goes down) let remote2 = TcpListener::bind("127.0.0.1:0").await.unwrap(); let remote2_port = remote2.local_addr().unwrap().port(); let remote2_count = Arc::new(AtomicU32::new(0)); - tokio::spawn(accept_and_count(remote2, remote2_count.clone())); + tokio::spawn(noise_accept_and_count( + remote2, + remote2_count.clone(), + key.clone(), + )); // Mock API server (returns whichever port endpoint_port holds) let endpoint_port = Arc::new(AtomicU16::new(remote1_port)); let api = TcpListener::bind("127.0.0.1:0").await.unwrap(); let api_port = api.local_addr().unwrap().port(); - tokio::spawn(mock_api_server(api, endpoint_port.clone())); + tokio::spawn(mock_api_server(api, endpoint_port.clone(), public_key)); // Start the tunnel client with a zero re-registration window: the first // remote-connect failure then triggers re-registration immediately, which diff --git a/server/Cargo.toml b/server/Cargo.toml index 857dda5..64990a5 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "localtunnel-server" -version = "0.1.7" +version = "0.2.0" edition.workspace = true description = "Server implementation of localtunnel." license.workspace = true @@ -28,3 +28,9 @@ thiserror = "1.0" async-trait = "0.1" regex = "1.7.0" socket2 = { workspace = true } +snowstorm = "0.4" +hex = "0.4" +rand = "0.8" + +[dev-dependencies] +localtunnel-client = { path = "../client" } diff --git a/server/src/api.rs b/server/src/api.rs index 67f4440..b4bf0ef 100644 --- a/server/src/api.rs +++ b/server/src/api.rs @@ -65,13 +65,15 @@ pub async fn request_endpoint( let mut manager = state.manager.lock().await; match manager.put(endpoint.to_string()).await { - Ok(port) => { + Ok((port, session_token)) => { let schema = if state.secure { "https" } else { "http" }; let info = ProxyInfo { id: endpoint.to_string(), port, max_conn_count: state.max_sockets, url: format!("{}://{}.{}", schema, endpoint, state.domain), + server_public_key: state.public_key.clone(), + session_token, }; log::debug!("Proxy info, {:?}", info); @@ -107,6 +109,12 @@ struct ProxyInfo { port: u16, max_conn_count: u8, url: String, + /// Hex-encoded static Noise public key of the server, used by the client + /// to authenticate the tunnel handshake. + server_public_key: String, + /// Hex-encoded per-tunnel token the client must present on every tunnel + /// connection. + session_token: String, } #[cfg(test)] diff --git a/server/src/lib.rs b/server/src/lib.rs index ad5c6b1..4e6b10f 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -66,13 +66,23 @@ pub async fn start(config: ServerConfig) -> Result<()> { require_auth ); - let manager = Arc::new(Mutex::new(ClientManager::new(max_sockets))); + // Static Noise keypair for this server process. Clients receive the + // public key in every registration response (over the HTTPS API), so a + // restart invalidating old keys is fine: clients re-register on failure. + let keypair = snowstorm::Builder::new(state::NOISE_PARAMS.parse()?).generate_keypair()?; + let public_key = hex::encode(&keypair.public); + + let manager = Arc::new(Mutex::new(ClientManager::new( + max_sockets, + Arc::new(keypair), + ))); let api_state = web::Data::new(State { manager: manager.clone(), max_sockets, require_auth, secure, domain, + public_key, }); let proxy_addr: SocketAddr = ([0, 0, 0, 0], proxy_port).into(); diff --git a/server/src/state.rs b/server/src/state.rs index 1075140..3ab8658 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -5,9 +5,11 @@ use std::{ time::{Duration, Instant}, }; +use anyhow::{anyhow, Result}; +use snowstorm::{Builder, Keypair, NoiseStream}; use socket2::{SockRef, TcpKeepalive}; use tokio::{ - io::Interest, + io::{AsyncReadExt, AsyncWriteExt, Interest}, net::{TcpListener, TcpStream}, sync::Mutex, task::JoinHandle, @@ -23,6 +25,24 @@ const TCP_KEEPALIVE_RETRIES: u32 = 5; /// How long before an unused client is cleaned up. const CLEANUP_TIMEOUT: Duration = Duration::from_secs(60 * 60); +/// Noise protocol parameters for the encrypted tunnel between client and +/// server. Must match the client exactly; NK authenticates the server by its +/// static key (delivered to the client in the registration response) and +/// derives fresh session keys per connection. +pub const NOISE_PARAMS: &str = "Noise_NK_25519_ChaChaPoly_BLAKE2s"; + +/// Length of the per-tunnel session token the client must present after the +/// handshake before its connection joins the pool. +pub const SESSION_TOKEN_LEN: usize = 32; + +/// How long a connecting peer gets to finish handshake + token before being +/// dropped, so half-open or non-speaking connections can't pile up. +const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10); + +/// A pooled tunnel connection: encrypted and already authenticated by the +/// session token. +pub type TunnelStream = NoiseStream; + /// App state holds all the client connection and status info. pub struct State { pub manager: Arc>, @@ -30,29 +50,41 @@ pub struct State { pub require_auth: bool, pub secure: bool, pub domain: String, + /// Hex-encoded static Noise public key, handed to clients at registration. + pub public_key: String, } pub struct ClientManager { pub clients: HashMap>>, pub _tunnels: u16, pub default_max_sockets: u8, + key: Arc, } impl ClientManager { - pub fn new(max_sockets: u8) -> Self { + pub fn new(max_sockets: u8, key: Arc) -> Self { ClientManager { clients: HashMap::new(), _tunnels: 0, default_max_sockets: max_sockets, + key, } } - pub async fn put(&mut self, url: String) -> io::Result { - let client = Arc::new(Mutex::new(Client::new(self.default_max_sockets))); + /// Registers a tunnel and returns the assigned port together with the + /// hex-encoded session token the client must present on every connection. + pub async fn put(&mut self, url: String) -> io::Result<(u16, String)> { + let session_token: [u8; SESSION_TOKEN_LEN] = rand::random(); + let client = Arc::new(Mutex::new(Client::new( + self.default_max_sockets, + self.key.clone(), + session_token, + ))); self.clients.insert(url, client.clone()); let mut client = client.lock().await; - client.listen().await + let port = client.listen().await?; + Ok((port, hex::encode(session_token))) } /// clean up old unused clients @@ -74,20 +106,24 @@ impl ClientManager { } pub struct Client { - pub available_sockets: Arc>>, + pub available_sockets: Arc>>, pub port: Option, pub max_sockets: u8, + key: Arc, + session_token: [u8; SESSION_TOKEN_LEN], listen_task: Option>, /// last time a new connection was established last_connection_time: Instant, } impl Client { - pub fn new(max_sockets: u8) -> Self { + pub fn new(max_sockets: u8, key: Arc, session_token: [u8; SESSION_TOKEN_LEN]) -> Self { Client { available_sockets: Arc::new(Mutex::new(vec![])), port: None, max_sockets, + key, + session_token, listen_task: None, last_connection_time: std::time::Instant::now(), } @@ -100,19 +136,25 @@ impl Client { let sockets = self.available_sockets.clone(); let max_sockets = self.max_sockets; + let key = self.key.clone(); + let session_token = self.session_token; let listen_task = tokio::spawn(async move { - // TODO check client is authenticated for the port loop { match timeout(Duration::from_secs(20), listener.accept()).await { Ok(Ok((socket, addr))) => { log::info!("new client connection: {:?}", addr); - let mut sockets = sockets.lock().await; - let sockets_len = sockets.len(); + let sockets = sockets.clone(); + let key = key.clone(); - if sockets_len < max_sockets as usize { - log::debug!("Add a new socket {}/{max_sockets}", sockets_len + 1,); + // Handshake in its own task so a slow or hostile peer + // can't stall the accept loop. + tokio::spawn(async move { + if sockets.lock().await.len() >= max_sockets as usize { + log::warn!("Reached sockets max: {max_sockets}, dropping connection"); + return; + } let ka = TcpKeepalive::new() .with_time(TCP_KEEPALIVE_TIME) @@ -124,10 +166,32 @@ impl Client { log::warn!("failed to enable TCP keepalive: {err}"); } - sockets.push(socket) - } else { - log::warn!("Reached sockets max: {sockets_len}/{max_sockets}"); - } + let stream = match timeout( + HANDSHAKE_TIMEOUT, + secure_accept(socket, &key, &session_token), + ) + .await + { + Ok(Ok(stream)) => stream, + Ok(Err(err)) => { + log::warn!("rejected tunnel connection from {addr:?}: {err:?}"); + return; + } + Err(_) => { + log::warn!("tunnel handshake with {addr:?} timed out"); + return; + } + }; + + let mut sockets = sockets.lock().await; + let sockets_len = sockets.len(); + if sockets_len < max_sockets as usize { + log::debug!("Add a new socket {}/{max_sockets}", sockets_len + 1); + sockets.push(stream); + } else { + log::warn!("Reached sockets max: {sockets_len}/{max_sockets}"); + } + }); } Ok(Err(e)) => log::info!("Couldn't get client: {:?}", e), Err(_) => { @@ -136,7 +200,7 @@ impl Client { let sockets_len = sockets.len(); let mut connected_sockets = vec![]; while let Some(s) = sockets.pop() { - if socket_is_writable(&s).await { + if socket_is_writable(s.get_inner()).await { connected_sockets.push(s); } } @@ -157,7 +221,7 @@ impl Client { Ok(port) } - pub async fn take(&mut self) -> Option { + pub async fn take(&mut self) -> Option { self.last_connection_time = Instant::now(); let mut sockets = self.available_sockets.lock().await; @@ -169,7 +233,7 @@ impl Client { self.max_sockets ); - if socket_is_writable(&socket).await { + if socket_is_writable(socket.get_inner()).await { return Some(socket); } @@ -199,6 +263,41 @@ impl Drop for Client { } } +/// Complete the Noise handshake as responder, then require the tunnel's +/// session token as the first encrypted message. Only authenticated +/// connections may join the pool — this is what stops an arbitrary peer that +/// found the port from receiving proxied traffic. A one-byte ack is sent back +/// so the client can distinguish "accepted" from "rejected" instead of +/// discovering it later through a dead proxied request. +async fn secure_accept( + socket: TcpStream, + key: &Keypair, + session_token: &[u8; SESSION_TOKEN_LEN], +) -> Result { + let responder = Builder::new(NOISE_PARAMS.parse()?) + .local_private_key(&key.private) + .build_responder()?; + let mut stream = NoiseStream::handshake(socket, responder) + .await + .map_err(|err| anyhow!("noise handshake failed: {err:?}"))?; + + let mut received = [0u8; SESSION_TOKEN_LEN]; + stream.read_exact(&mut received).await?; + if !token_matches(&received, session_token) { + return Err(anyhow!("session token mismatch")); + } + + stream.write_all(&[1]).await?; + stream.flush().await?; + + Ok(stream) +} + +fn token_matches(a: &[u8; SESSION_TOKEN_LEN], b: &[u8; SESSION_TOKEN_LEN]) -> bool { + // Constant-time comparison, no dependence on where the first mismatch is. + a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0 +} + async fn socket_is_writable(socket: &TcpStream) -> bool { socket .ready(Interest::WRITABLE) diff --git a/server/tests/e2e.rs b/server/tests/e2e.rs new file mode 100644 index 0000000..95314a6 --- /dev/null +++ b/server/tests/e2e.rs @@ -0,0 +1,166 @@ +use std::time::Duration; + +use localtunnel_client::{broadcast, open_tunnel, ClientConfig, NOISE_PARAMS}; +use localtunnel_server::{start, ServerConfig}; +use snowstorm::{Builder, NoiseStream}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::time::{sleep, Instant}; + +async fn free_port() -> u16 { + TcpListener::bind("127.0.0.1:0") + .await + .unwrap() + .local_addr() + .unwrap() + .port() +} + +/// Minimal HTTP/1.1 app behind the tunnel. +async fn spawn_local_http(body: &'static str) -> u16 { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + return; + }; + tokio::spawn(async move { + let mut buf = [0u8; 4096]; + let _ = stream.read(&mut buf).await; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len(), + ); + let _ = stream.write_all(response.as_bytes()).await; + }); + } + }); + port +} + +async fn spawn_server() -> (u16, u16) { + let api_port = free_port().await; + let proxy_port = free_port().await; + // `start` drives actix's non-Send server future, so it gets its own + // runtime on a dedicated thread instead of `tokio::spawn`. + std::thread::spawn(move || { + tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .unwrap() + .block_on(start(ServerConfig { + domain: "127.0.0.1".to_string(), + api_port, + secure: false, + max_sockets: 4, + proxy_port, + require_auth: false, + })) + .unwrap(); + }); + + // Wait until the API server accepts connections. + let deadline = Instant::now() + Duration::from_secs(10); + while TcpStream::connect(("127.0.0.1", api_port)).await.is_err() { + assert!(Instant::now() < deadline, "server did not start"); + sleep(Duration::from_millis(50)).await; + } + (api_port, proxy_port) +} + +#[tokio::test] +async fn proxies_request_through_encrypted_tunnel() { + let body = "hello through the tunnel"; + let local_port = spawn_local_http(body).await; + let (api_port, proxy_port) = spawn_server().await; + + let (shutdown_tx, _) = broadcast::channel(1); + let url = open_tunnel(ClientConfig { + server: Some(format!("http://127.0.0.1:{api_port}")), + subdomain: Some("e2e".to_string()), + local_host: Some("127.0.0.1".to_string()), + local_port, + shutdown_signal: shutdown_tx.clone(), + max_conn: 4, + credential: None, + reregister_after: None, + }) + .await + .unwrap(); + assert_eq!(url, "http://e2e.127.0.0.1"); + + // Tunnel connections are pooled asynchronously; retry until one serves us. + let deadline = Instant::now() + Duration::from_secs(10); + let response = loop { + let mut conn = TcpStream::connect(("127.0.0.1", proxy_port)).await.unwrap(); + conn.write_all(b"GET / HTTP/1.1\r\nHost: e2e.127.0.0.1\r\nConnection: close\r\n\r\n") + .await + .unwrap(); + let mut buf = String::new(); + let _ = conn.read_to_string(&mut buf).await; + if buf.contains("200 OK") { + break buf; + } + assert!( + Instant::now() < deadline, + "no successful proxied response, last: {buf:?}" + ); + sleep(Duration::from_millis(100)).await; + }; + assert!(response.ends_with(body), "unexpected response: {response:?}"); + + let _ = shutdown_tx.send(()); +} + +#[derive(serde::Deserialize)] +struct Registered { + port: u16, + server_public_key: String, + session_token: String, +} + +#[tokio::test] +async fn unauthenticated_tunnel_connections_are_rejected() { + let (api_port, _proxy_port) = spawn_server().await; + + // Register a tunnel directly against the API to learn the assigned port. + let reg: Registered = reqwest::get(format!("http://127.0.0.1:{api_port}/reject")) + .await + .unwrap() + .json() + .await + .unwrap(); + + // A peer that speaks framed garbage instead of a Noise handshake must be + // dropped without ever joining the connection pool. + let mut conn = TcpStream::connect(("127.0.0.1", reg.port)).await.unwrap(); + conn.write_all(&[8u8, 0]).await.unwrap(); // frame length: 8 + conn.write_all(&[0u8; 8]).await.unwrap(); // not a valid handshake message + let mut buf = [0u8; 16]; + assert!( + matches!(conn.read(&mut buf).await, Ok(0) | Err(_)), + "garbage connection should be closed" + ); + + // A peer with the right server key but the wrong session token completes + // the handshake yet must be rejected before receiving the ack byte. + let server_key = hex::decode(®.server_public_key).unwrap(); + let mut wrong_token = hex::decode(®.session_token).unwrap(); + wrong_token[0] ^= 0xff; + + let initiator = Builder::new(NOISE_PARAMS.parse().unwrap()) + .remote_public_key(&server_key) + .build_initiator() + .unwrap(); + let conn = TcpStream::connect(("127.0.0.1", reg.port)).await.unwrap(); + let mut stream = NoiseStream::handshake(conn, initiator).await.unwrap(); + stream.write_all(&wrong_token).await.unwrap(); + stream.flush().await.unwrap(); + + let mut ack = [0u8; 1]; + assert!( + stream.read_exact(&mut ack).await.is_err(), + "server must not ack a wrong session token" + ); +}