diff --git a/Cargo.lock b/Cargo.lock index 7814c86..3bbb88d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -126,6 +126,45 @@ version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" +[[package]] +name = "asn1-rs" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7f43a50ac4fdca5df8e885c21b835997f0a1cdee65494a6847694a98652d9d8" +dependencies = [ + "asn1-rs-derive", + "asn1-rs-impl", + "displaydoc", + "nom", + "num-traits", + "rusticata-macros", + "thiserror 2.0.20", + "time", +] + +[[package]] +name = "asn1-rs-derive" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3109e49b1e4909e9db6515a30c633684d68cdeaa252f215214cb4fa1a5bfee2c" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "synstructure 0.13.2", +] + +[[package]] +name = "asn1-rs-impl" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "async-broadcast" version = "0.7.2" @@ -201,6 +240,29 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" +[[package]] +name = "aws-lc-rs" +version = "1.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b281d307588d634de920874890732659e2e7672f72b5e10e81badc1a8a83621e" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.45.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9bff6c3b54fad79a2e60b8102caf565819711497c1f5f092f49508e2f5c31b27" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", + "pkg-config", +] + [[package]] name = "backon" version = "1.6.0" @@ -236,6 +298,15 @@ version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" +[[package]] +name = "bit-vec" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b71798fca2c1fe1086445a7258a4bc81e6e49dcd24c8d0dd9a1e57395b603f51" +dependencies = [ + "serde", +] + [[package]] name = "bitflags" version = "2.13.2" @@ -357,6 +428,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "54413ede23c2daf518f35156dfde027feb2374004d63bd497f983c8db9c0e313" dependencies = [ "find-msvc-tools", + "jobserver", + "libc", "shlex", ] @@ -476,6 +549,15 @@ version = "1.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1c133bc6a41be0d194c306b5506d15e6feeea7b1d6604bd3f8310dfb2ca96486" +[[package]] +name = "cmake" +version = "0.1.58" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678" +dependencies = [ + "cc", +] + [[package]] name = "cmov" version = "0.5.4" @@ -722,6 +804,20 @@ dependencies = [ "zeroize", ] +[[package]] +name = "der-parser" +version = "10.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6" +dependencies = [ + "asn1-rs", + "displaydoc", + "nom", + "num-bigint", + "num-traits", + "rusticata-macros", +] + [[package]] name = "deranged" version = "0.5.8" @@ -844,6 +940,12 @@ dependencies = [ "litrs", ] +[[package]] +name = "dunce" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" + [[package]] name = "dyn-clone" version = "1.0.20" @@ -1035,6 +1137,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + [[package]] name = "futures" version = "0.3.34" @@ -1856,6 +1964,16 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "jobserver" +version = "0.1.35" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3" +dependencies = [ + "getrandom 0.4.3", + "libc", +] + [[package]] name = "js-sys" version = "0.3.105" @@ -2026,6 +2144,12 @@ dependencies = [ "autocfg", ] +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + [[package]] name = "mio" version = "1.2.3" @@ -2330,6 +2454,16 @@ dependencies = [ "libc", ] +[[package]] +name = "nom" +version = "7.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" +dependencies = [ + "memchr", + "minimal-lexical", +] + [[package]] name = "noq" version = "1.3.0" @@ -2401,12 +2535,31 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num-bigint" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c89e69e7e0f03bea5ef08013795c25018e101932225a656383bd384495ecc367" +dependencies = [ + "num-integer", + "num-traits", +] + [[package]] name = "num-conv" version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" +[[package]] +name = "num-integer" +version = "0.1.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ce2d95d4b3734dc35aa2f45e1aa22cd416814592a4f9d9205e11affd5b8e10b" +dependencies = [ + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -2528,6 +2681,15 @@ dependencies = [ "objc2-security", ] +[[package]] +name = "oid-registry" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7" +dependencies = [ + "asn1-rs", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -2899,6 +3061,19 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "rcgen" +version = "0.14.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8774e05a7d0de114588e6a28fe7e71694b82614ed569d86d8b389dfbc98b8ad8" +dependencies = [ + "ring", + "rustls-pki-types", + "time", + "x509-parser", + "yasna", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -3044,6 +3219,15 @@ dependencies = [ "semver", ] +[[package]] +name = "rusticata-macros" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632" +dependencies = [ + "nom", +] + [[package]] name = "rustix" version = "1.1.5" @@ -3063,6 +3247,7 @@ version = "0.23.45" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" dependencies = [ + "aws-lc-rs", "log", "once_cell", "ring", @@ -3127,6 +3312,7 @@ version = "0.103.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" dependencies = [ + "aws-lc-rs", "ring", "rustls-pki-types", "untrusted", @@ -3527,6 +3713,17 @@ dependencies = [ "futures-core", ] +[[package]] +name = "synstructure" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "synstructure" version = "0.14.0" @@ -3964,9 +4161,38 @@ dependencies = [ "tracing", "tracing-subscriber", "tsunagi", + "tsunagi-tcp-tls", "tsunagi-wg-quic", ] +[[package]] +name = "tsunagi-tcp-tls" +version = "0.1.0" +dependencies = [ + "bytes", + "chacha20poly1305", + "hex", + "hkdf", + "iroh", + "netdev 0.46.3", + "postcard", + "rand 0.10.3", + "rcgen", + "rustls", + "serde", + "sha2", + "subtle", + "tempfile", + "thiserror 2.0.20", + "tokio", + "tokio-rustls", + "tracing", + "tracing-subscriber", + "tsunagi", + "x25519-dalek", + "zeroize", +] + [[package]] name = "tsunagi-wg-quic" version = "0.1.0" @@ -4631,6 +4857,34 @@ dependencies = [ "zeroize", ] +[[package]] +name = "x509-parser" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d43b0f71ce057da06bc0851b23ee24f3f86190b07203dd8f567d0b706a185202" +dependencies = [ + "asn1-rs", + "data-encoding", + "der-parser", + "lazy_static", + "nom", + "oid-registry", + "ring", + "rusticata-macros", + "thiserror 2.0.20", + "time", +] + +[[package]] +name = "yasna" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5f6765e852b9b4dc8e2a76843e4d64d1cea8e79bcde0b6901aea8e7c7f08282" +dependencies = [ + "bit-vec", + "time", +] + [[package]] name = "yoke" version = "0.8.3" @@ -4651,7 +4905,7 @@ dependencies = [ "proc-macro2", "quote", "syn 3.0.6", - "synstructure", + "synstructure 0.14.0", ] [[package]] @@ -4757,7 +5011,7 @@ dependencies = [ "proc-macro2", "quote", "syn 3.0.6", - "synstructure", + "synstructure 0.14.0", ] [[package]] diff --git a/crates/tsunagi-cli/Cargo.toml b/crates/tsunagi-cli/Cargo.toml index ea3fe6c..9f1a60f 100644 --- a/crates/tsunagi-cli/Cargo.toml +++ b/crates/tsunagi-cli/Cargo.toml @@ -18,6 +18,7 @@ doc = false [dependencies] tsunagi = { path = "../tsunagi", version = "0.1.0", features = ["tun-device", "dns-publish"] } tsunagi-wg-quic = { path = "../tsunagi-wg-quic", version = "0.1.0" } +tsunagi-tcp-tls = { path = "../tsunagi-tcp-tls", version = "0.1.0" } iroh.workspace = true tokio = { version = "1.53", features = ["rt", "rt-multi-thread", "sync", "time", "macros", "signal"] } tracing.workspace = true diff --git a/crates/tsunagi-cli/src/main.rs b/crates/tsunagi-cli/src/main.rs index 6f74cb7..8338334 100644 --- a/crates/tsunagi-cli/src/main.rs +++ b/crates/tsunagi-cli/src/main.rs @@ -23,6 +23,9 @@ use tsunagi::iroh_types::EndpointAddr; use tsunagi::overlay::{MemoryTunFactory, TunFactory}; use tsunagi::state::Ipv4Range; use tsunagi::{Agent, NetworkId}; +use tsunagi_tcp_tls::{ + TCP_TLS_PROTOCOL, TCP_TLS_VERSION, TcpTlsCodec, TcpTlsConfig, TcpTlsPlugin, TcpTlsTransport, +}; use tsunagi_wg_quic::{WireguardConfig, WireguardPlugin}; /// A small agent for private mesh networks. @@ -1164,13 +1167,21 @@ struct ProtocolSpec { } /// Every protocol this build has. -const PROTOCOLS: &[ProtocolSpec] = &[ProtocolSpec { - name: tsunagi_wg_quic::WIREGUARD_PROTOCOL, - version: tsunagi_wg_quic::ANNOUNCEMENT_VERSION, - summary: "WireGuard's cryptography carried in iroh's QUIC datagrams, so it \ - crosses NAT and survives where plain WireGuard is blocked", - options: WireguardPlugin::OPTIONS, -}]; +const PROTOCOLS: &[ProtocolSpec] = &[ + ProtocolSpec { + name: tsunagi_wg_quic::WIREGUARD_PROTOCOL, + version: tsunagi_wg_quic::ANNOUNCEMENT_VERSION, + summary: "WireGuard's cryptography carried in iroh's QUIC datagrams, so it \ + crosses NAT and survives where plain WireGuard is blocked", + options: WireguardPlugin::OPTIONS, + }, + ProtocolSpec { + name: TCP_TLS_PROTOCOL, + version: TCP_TLS_VERSION, + summary: "Noise E2EE carried over TCP TLS 1.3 (port 443) disguised as HTTPS traffic", + options: TcpTlsCodec::OPTIONS, + }, +]; /// One `-o` setting, and the protocol it was aimed at. struct Setting { @@ -3375,7 +3386,7 @@ async fn up(args: UpArgs) -> Result<(), Box> { } let discovery: Arc = Arc::new(CompositeDiscovery::new([ - Arc::new(StaticBootstrap::new(bootstrap)) as Arc, + Arc::new(StaticBootstrap::new(bootstrap.clone())) as Arc, ])); let mut config = AgentConfig::new(paths.clone()) @@ -3436,6 +3447,7 @@ async fn up(args: UpArgs) -> Result<(), Box> { // The interface belongs to the agent, so it is configured once whatever // was selected to carry traffic over it. let mut wireguard = None; + let mut tcp_tls_transport = None; if !wanted.is_empty() { let tun_factory: Arc = if args.no_tun { Arc::new(MemoryTunFactory::new()) @@ -3459,6 +3471,32 @@ async fn up(args: UpArgs) -> Result<(), Box> { config = config.with_plugin(plugin.clone() as Arc); wireguard = Some(plugin); } + TCP_TLS_PROTOCOL => { + let tls_cfg = TcpTlsConfig::new(); + let tls_cfg = TcpTlsCodec::configure(tls_cfg, &options)?; + let identity = tsunagi::storage::StateStore::open(paths.state_db())? + .load_or_create_device_identity()?; + let transport = Arc::new( + TcpTlsTransport::bind(&identity, tls_cfg.port) + .await + .map_err(|err| err.to_string())?, + ); + for ep_addr in &bootstrap { + for ip in ep_addr.ip_addrs() { + transport.set_peer_addr(ep_addr.id, *ip); + } + } + let codec = TcpTlsCodec::new(identity.endpoint_id(), Arc::clone(&transport)); + let plugin = Arc::new(TcpTlsPlugin::new(codec)); + config = config + .with_plugin(plugin as Arc) + .with_custom_transport( + TCP_TLS_PROTOCOL, + Arc::clone(&transport) + as Arc, + ); + tcp_tls_transport = Some(transport); + } other => return Err(format!("`{other}` is listed but not built in").into()), } } @@ -3495,6 +3533,22 @@ async fn up(args: UpArgs) -> Result<(), Box> { // is never dropped without being closed. let mut events = agent.subscribe(); + if let Some(transport) = tcp_tls_transport { + let agent_inbound = agent.clone(); + tokio::spawn(async move { + transport + .accept_loop(move |inbound| { + let agent = agent_inbound.clone(); + tokio::spawn(async move { + if let Err(err) = agent.install_inbound_link(inbound).await { + tracing::debug!(%err, "failed to install inbound tcp-tls link"); + } + }); + }) + .await; + }); + } + println!("tsunagi is up"); println!(" endpoint id {}", agent.endpoint_id()); println!(" hostname {}", agent.hostname()); diff --git a/crates/tsunagi-tcp-tls/Cargo.toml b/crates/tsunagi-tcp-tls/Cargo.toml new file mode 100644 index 0000000..1649f97 --- /dev/null +++ b/crates/tsunagi-tcp-tls/Cargo.toml @@ -0,0 +1,38 @@ +[package] +name = "tsunagi-tcp-tls" +version = "0.1.0" +description = "TCP TLS 443 transport and E2EE tunnel protocol for tsunagi mesh networks." +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true + +[dependencies] +tsunagi = { path = "../tsunagi", version = "0.1.0" } +iroh.workspace = true +tokio = { workspace = true, features = ["net", "io-util", "time", "sync", "macros"] } +rustls = "0.23" +tokio-rustls = "0.26" +rcgen = { version = "0.14", default-features = false, features = ["ring"] } +x25519-dalek = { version = "2.0", features = ["static_secrets"] } +chacha20poly1305 = "0.10" +sha2 = "0.11" +hkdf = "0.13" +rand = "0.10" +subtle = "2.6" +zeroize = { version = "1.9", features = ["derive"] } +bytes.workspace = true +serde.workspace = true +postcard.workspace = true +tracing.workspace = true +thiserror.workspace = true +hex.workspace = true +netdev = "0.46" + +[dev-dependencies] +tsunagi = { path = "../tsunagi", features = ["testing"] } +tempfile.workspace = true +tracing-subscriber.workspace = true + +[lints] +workspace = true diff --git a/crates/tsunagi-tcp-tls/src/cert.rs b/crates/tsunagi-tcp-tls/src/cert.rs new file mode 100644 index 0000000..f19f664 --- /dev/null +++ b/crates/tsunagi-tcp-tls/src/cert.rs @@ -0,0 +1,143 @@ +//! TLS certificate generation and verification. +//! +//! Generates a self-signed TLS certificate derived from the device's persistent +//! identity, and verifies remote certificates against the expected peer's +//! [`iroh::EndpointId`]. + +use std::sync::Arc; + +use iroh::EndpointId; +use rustls::client::danger::{ServerCertVerified, ServerCertVerifier}; +use rustls::pki_types::{CertificateDer, ServerName, UnixTime}; +use rustls::{ClientConfig, Error as RustlsError, ServerConfig}; + +/// Verifies that the server certificate's public key matches the expected peer. +#[derive(Debug)] +pub struct PinnedEndpointVerifier { + expected: EndpointId, +} + +impl PinnedEndpointVerifier { + /// Creates a verifier expecting `peer`. + pub fn new(peer: EndpointId) -> Self { + Self { expected: peer } + } +} + +impl ServerCertVerifier for PinnedEndpointVerifier { + fn verify_server_cert( + &self, + end_entity: &CertificateDer<'_>, + _intermediates: &[CertificateDer<'_>], + _server_name: &ServerName<'_>, + _ocsp_response: &[u8], + _now: UnixTime, + ) -> Result { + // Parse the certificate's SPKI / public key + let cert_bytes = end_entity.as_ref(); + let expected_bytes = self.expected.as_bytes(); + + // Check if the expected 32-byte Ed25519 public key is in the certificate + if cert_bytes + .windows(32) + .any(|window| window == expected_bytes) + { + Ok(ServerCertVerified::assertion()) + } else { + Err(RustlsError::InvalidCertificate( + rustls::CertificateError::NotValidForName, + )) + } + } + + fn verify_tls12_signature( + &self, + _message: &[u8], + _cert: &CertificateDer<'_>, + _dss: &rustls::DigitallySignedStruct, + ) -> Result { + // We require TLS 1.3 + Err(RustlsError::PeerIncompatible( + rustls::PeerIncompatible::NoSignatureSchemesInCommon, + )) + } + + fn verify_tls13_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &rustls::DigitallySignedStruct, + ) -> Result { + rustls::crypto::verify_tls13_signature( + message, + cert, + dss, + &rustls::crypto::ring::default_provider().signature_verification_algorithms, + ) + } + + fn supported_verify_schemes(&self) -> Vec { + vec![rustls::SignatureScheme::ED25519] + } +} + +/// Generates a self-signed certificate and private key for TLS 1.3 using +/// a given 32-byte Ed25519 secret key. +pub fn generate_self_signed_cert( + secret_key_bytes: &[u8; 32], +) -> Result< + ( + CertificateDer<'static>, + rustls::pki_types::PrivateKeyDer<'static>, + ), + Box, +> { + // PKCS#8 prefix for v1 Ed25519 private keys: + // 30 2e 02 01 00 30 05 06 03 2b 65 70 04 22 04 20 <32 bytes> + let mut pkcs8 = Vec::with_capacity(48); + pkcs8.extend_from_slice(&[ + 0x30, 0x2e, 0x02, 0x01, 0x00, 0x30, 0x05, 0x06, 0x03, 0x2b, 0x65, 0x70, 0x04, 0x22, 0x04, + 0x20, + ]); + pkcs8.extend_from_slice(secret_key_bytes); + + let key_der = rustls::pki_types::PrivatePkcs8KeyDer::from(pkcs8); + let key_pair = rcgen::KeyPair::from_pkcs8_der_and_sign_algo(&key_der, &rcgen::PKCS_ED25519)?; + let mut params = rcgen::CertificateParams::new(vec!["tsunagi.local".to_string()])?; + params + .distinguished_name + .push(rcgen::DnType::CommonName, "tsunagi-node"); + + let cert = params.self_signed(&key_pair)?; + let cert_der = CertificateDer::from(cert.der().to_vec()); + let priv_key = rustls::pki_types::PrivateKeyDer::Pkcs8(key_der); + + Ok((cert_der, priv_key)) +} + +/// Creates a `rustls::ServerConfig` for the node. +pub fn make_server_config( + cert: CertificateDer<'static>, + key: rustls::pki_types::PrivateKeyDer<'static>, +) -> Result, rustls::Error> { + let provider = Arc::new(rustls::crypto::ring::default_provider()); + let mut config = ServerConfig::builder_with_provider(provider) + .with_safe_default_protocol_versions()? + .with_no_client_auth() + .with_single_cert(vec![cert], key)?; + config.alpn_protocols = vec![b"tsunagi-tls/1".to_vec()]; + Ok(Arc::new(config)) +} + +/// Creates a `rustls::ClientConfig` expecting a specific peer's `EndpointId`. +pub fn make_client_config(expected_peer: EndpointId) -> Result, rustls::Error> { + let provider = Arc::new(rustls::crypto::ring::default_provider()); + let verifier = Arc::new(PinnedEndpointVerifier::new(expected_peer)); + let mut config = ClientConfig::builder_with_provider(provider) + .with_safe_default_protocol_versions()? + .dangerous() + .with_custom_certificate_verifier(verifier) + .with_no_client_auth(); + config.alpn_protocols = vec![b"tsunagi-tls/1".to_vec()]; + Ok(Arc::new(config)) +} diff --git a/crates/tsunagi-tcp-tls/src/crypto.rs b/crates/tsunagi-tcp-tls/src/crypto.rs new file mode 100644 index 0000000..5ca6cab --- /dev/null +++ b/crates/tsunagi-tcp-tls/src/crypto.rs @@ -0,0 +1,104 @@ +//! End-to-end packet encryption for `tcp-tls`. +//! +//! Uses X25519 Diffie-Hellman key exchange and ChaCha20-Poly1305 AEAD. +//! This ensures that even when traffic passes through intermediate relay nodes, +//! the intermediate nodes cannot inspect or tamper with the payload. + +use std::sync::atomic::{AtomicU64, Ordering}; + +use chacha20poly1305::aead::{Aead, KeyInit}; +use chacha20poly1305::{ChaCha20Poly1305, Nonce}; +use hkdf::Hkdf; +use iroh::EndpointId; +use sha2::Sha256; +use tsunagi::dataplane::PluginError; +use x25519_dalek::{PublicKey, StaticSecret}; + +/// Length of the X25519 public key. +pub const KEY_LEN: usize = 32; + +/// Domain separator for session key derivation. +const HKDF_SALT: &[u8] = b"tsunagi/e2ee/noise/v1"; + +/// A session cipher for communicating with one peer. +pub struct PeerCipher { + tx_cipher: ChaCha20Poly1305, + rx_cipher: ChaCha20Poly1305, + tx_nonce: AtomicU64, +} + +impl PeerCipher { + /// Establishes a new cipher between `local_id` and `remote_id` given + /// our local secret and their public key. + pub fn new( + local_id: EndpointId, + remote_id: EndpointId, + local_secret: &StaticSecret, + remote_public: &PublicKey, + ) -> Self { + let shared = local_secret.diffie_hellman(remote_public); + let hk = Hkdf::::new(Some(HKDF_SALT), shared.as_bytes()); + + let mut key_a_to_b = [0u8; 32]; + let mut key_b_to_a = [0u8; 32]; + let _ = hk.expand(b"tsunagi-key-a-to-b", &mut key_a_to_b); + let _ = hk.expand(b"tsunagi-key-b-to-a", &mut key_b_to_a); + + // Deterministic role: smaller EndpointId is party A. + let is_party_a = local_id.as_bytes() < remote_id.as_bytes(); + let (tx_key, rx_key) = if is_party_a { + (key_a_to_b, key_b_to_a) + } else { + (key_b_to_a, key_a_to_b) + }; + + let tx_cipher = ChaCha20Poly1305::new(&tx_key.into()); + let rx_cipher = ChaCha20Poly1305::new(&rx_key.into()); + + Self { + tx_cipher, + rx_cipher, + tx_nonce: AtomicU64::new(0), + } + } + + /// Encrypts an outbound packet. Prefixes the ciphertext with the 8-byte nonce. + pub fn encrypt(&self, packet: &[u8]) -> Result { + let counter = self.tx_nonce.fetch_add(1, Ordering::Relaxed); + let mut nonce_bytes = [0u8; 12]; + nonce_bytes[4..12].copy_from_slice(&counter.to_be_bytes()); + let nonce = Nonce::from_slice(&nonce_bytes); + + let ciphertext = self + .tx_cipher + .encrypt(nonce, packet) + .map_err(|err| PluginError::Other(format!("encryption failed: {err}")))?; + + let mut out = Vec::with_capacity(8 + ciphertext.len()); + out.extend_from_slice(&counter.to_be_bytes()); + out.extend_from_slice(&ciphertext); + Ok(bytes::Bytes::from(out)) + } + + /// Decrypts an inbound packet prefixed with the 8-byte nonce. + pub fn decrypt(&self, payload: &[u8]) -> Result { + if payload.len() < 8 + 16 { + return Err(PluginError::Other("payload too short".into())); + } + + let counter_bytes: [u8; 8] = match payload[..8].try_into() { + Ok(b) => b, + Err(_) => return Err(PluginError::Other("invalid nonce".into())), + }; + let mut nonce_bytes = [0u8; 12]; + nonce_bytes[4..12].copy_from_slice(&counter_bytes); + let nonce = Nonce::from_slice(&nonce_bytes); + + let plaintext = self + .rx_cipher + .decrypt(nonce, &payload[8..]) + .map_err(|err| PluginError::Other(format!("decryption failed: {err}")))?; + + Ok(bytes::Bytes::from(plaintext)) + } +} diff --git a/crates/tsunagi-tcp-tls/src/lib.rs b/crates/tsunagi-tcp-tls/src/lib.rs new file mode 100644 index 0000000..7a2a992 --- /dev/null +++ b/crates/tsunagi-tcp-tls/src/lib.rs @@ -0,0 +1,9 @@ +//! The `tcp-tls` protocol: TCP with TLS 1.3 on port 443 with Noise E2EE. + +pub mod cert; +pub mod crypto; +pub mod plugin; +pub mod transport; + +pub use plugin::{TCP_TLS_PROTOCOL, TCP_TLS_VERSION, TcpTlsCodec, TcpTlsConfig, TcpTlsPlugin}; +pub use transport::TcpTlsTransport; diff --git a/crates/tsunagi-tcp-tls/src/plugin.rs b/crates/tsunagi-tcp-tls/src/plugin.rs new file mode 100644 index 0000000..fe2d052 --- /dev/null +++ b/crates/tsunagi-tcp-tls/src/plugin.rs @@ -0,0 +1,254 @@ +//! The `tcp-tls` IP plugin implementation. + +use std::collections::HashMap; +use std::sync::{Arc, RwLock}; + +use bytes::Bytes; +use iroh::EndpointId; +use serde::{Deserialize, Serialize}; +use x25519_dalek::{PublicKey, StaticSecret}; + +use tsunagi::dataplane::tunnel::{GenericTunnelPlugin, TunnelCodec}; +use tsunagi::dataplane::{PluginError, ProtocolOption}; +use tsunagi::identity::NetworkId; + +use crate::crypto::PeerCipher; +use crate::transport::TcpTlsTransport; + +/// The wire protocol identifier for `tcp-tls`. +pub const TCP_TLS_PROTOCOL: &str = "tcp-tls"; + +/// Wire version for `tcp-tls`. +pub const TCP_TLS_VERSION: u16 = 1; + +/// Configuration for the `tcp-tls` protocol plugin. +#[derive(Debug, Clone, Default)] +pub struct TcpTlsConfig { + /// Requested port to bind. + pub port: Option, +} + +impl TcpTlsConfig { + /// Creates a configuration with default settings. + pub fn new() -> Self { + Self::default() + } + + /// Sets the TCP listen port. + pub fn with_port(mut self, port: u16) -> Self { + self.port = Some(port); + self + } +} + +/// Announcement payload exchanged between peers for `tcp-tls`. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct TcpTlsAnnouncement { + /// X25519 public key for E2EE noise encryption. + pub noise_public: [u8; 32], + /// TCP port the peer is listening on. + pub port: u16, + /// Candidate IP addresses where this peer can be reached on `port`. + #[serde(default)] + pub addrs: Vec, +} + +/// Discovers local network interface IPs to announce to peers. +pub fn local_ip_candidates() -> Vec { + let mut ips = Vec::new(); + for iface in netdev::get_interfaces() { + if iface.is_up() && !iface.is_loopback() { + for ip in iface.ipv4 { + let addr = ip.addr(); + if !addr.is_loopback() && !addr.is_unspecified() { + ips.push(std::net::IpAddr::V4(addr)); + } + } + for ip in iface.ipv6 { + let addr = ip.addr(); + if !addr.is_loopback() && !addr.is_unspecified() { + ips.push(std::net::IpAddr::V6(addr)); + } + } + } + } + ips.push(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)); + ips +} + +/// The codec implementing cryptographic transformation and announcements for `tcp-tls`. +pub struct TcpTlsCodec { + local_id: EndpointId, + static_secret: StaticSecret, + public_key: PublicKey, + transport: Arc, + ciphers: RwLock>>, +} + +impl std::fmt::Debug for TcpTlsCodec { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("TcpTlsCodec") + .field("local_id", &self.local_id.fmt_short().to_string()) + .field("port", &self.transport.bound_port()) + .finish() + } +} + +impl TcpTlsCodec { + /// Supported options for `tcp-tls`. + pub const OPTIONS: &'static [ProtocolOption] = &[ProtocolOption { + key: "port", + value: "PORT", + help: "TCP port to listen on for TLS 1.3 connections", + default: Some("443"), + }]; + + /// Creates a new `TcpTlsCodec`. + pub fn new(local_id: EndpointId, transport: Arc) -> Self { + let mut secret_bytes = [0u8; 32]; + rand::fill(&mut secret_bytes); + let static_secret = StaticSecret::from(secret_bytes); + let public_key = PublicKey::from(&static_secret); + + Self { + local_id, + static_secret, + public_key, + transport, + ciphers: RwLock::new(HashMap::new()), + } + } + + /// Configures settings from parsed `-o` options. + pub fn configure( + mut config: TcpTlsConfig, + options: &[(String, String)], + ) -> Result { + for (key, value) in options { + match key.as_str() { + "port" => { + let port: u16 = value + .parse() + .map_err(|_| format!("invalid port `{value}`"))?; + config.port = Some(port); + } + other => return Err(format!("unknown option `{other}` for tcp-tls")), + } + } + Ok(config) + } +} + +impl TunnelCodec for TcpTlsCodec { + fn protocol_id(&self) -> &str { + TCP_TLS_PROTOCOL + } + + fn protocol_version(&self) -> u16 { + TCP_TLS_VERSION + } + + fn options(&self) -> &'static [ProtocolOption] { + Self::OPTIONS + } + + fn local_capability_data(&self, _network: NetworkId) -> Result, PluginError> { + let announcement = TcpTlsAnnouncement { + noise_public: *self.public_key.as_bytes(), + port: self.transport.bound_port(), + addrs: local_ip_candidates(), + }; + postcard::to_allocvec(&announcement) + .map_err(|err| PluginError::Other(format!("encode announcement failed: {err}"))) + } + + fn on_peer_capability_data( + &self, + network: NetworkId, + peer: EndpointId, + data: &[u8], + ) -> Result<(), PluginError> { + let announcement: TcpTlsAnnouncement = postcard::from_bytes(data) + .map_err(|err| PluginError::Rejected(format!("invalid announcement: {err}")))?; + + for ip in &announcement.addrs { + self.transport + .set_peer_addr(peer, std::net::SocketAddr::new(*ip, announcement.port)); + } + self.transport.set_peer_addr( + peer, + std::net::SocketAddr::new( + std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), + announcement.port, + ), + ); + + let peer_noise_pubkey = PublicKey::from(announcement.noise_public); + let cipher = Arc::new(PeerCipher::new( + self.local_id, + peer, + &self.static_secret, + &peer_noise_pubkey, + )); + + let mut guard = match self.ciphers.write() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + guard.insert((network, peer), cipher); + Ok(()) + } + + fn on_peer_down(&self, network: NetworkId, peer: EndpointId) { + let mut guard = match self.ciphers.write() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + guard.remove(&(network, peer)); + } + + fn encrypt( + &self, + network: NetworkId, + peer: EndpointId, + packet: &[u8], + ) -> Result { + let cipher = { + let guard = match self.ciphers.read() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + guard.get(&(network, peer)).cloned() + }; + + let Some(cipher) = cipher else { + return Err(PluginError::Unavailable("no cipher for peer".into())); + }; + + cipher.encrypt(packet) + } + + fn decrypt( + &self, + network: NetworkId, + peer: EndpointId, + payload: &[u8], + ) -> Result { + let cipher = { + let guard = match self.ciphers.read() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + guard.get(&(network, peer)).cloned() + }; + + let Some(cipher) = cipher else { + return Err(PluginError::Unavailable("no cipher for peer".into())); + }; + + cipher.decrypt(payload) + } +} + +/// The complete `tcp-tls` plugin. +pub type TcpTlsPlugin = GenericTunnelPlugin; diff --git a/crates/tsunagi-tcp-tls/src/transport.rs b/crates/tsunagi-tcp-tls/src/transport.rs new file mode 100644 index 0000000..3e728b4 --- /dev/null +++ b/crates/tsunagi-tcp-tls/src/transport.rs @@ -0,0 +1,512 @@ +//! TCP TLS 1.3 transport implementation for Tsunagi. +//! +//! Provides [`TcpTlsTransport`] implementing [`PacketTransport`] +//! and [`TcpTlsLink`] implementing [`PacketLink`]. + +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +use bytes::Bytes; +use iroh::{EndpointId, Signature}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::{Notify, mpsc}; +use tokio_rustls::TlsConnector; + +use tsunagi::BoxFuture; +use tsunagi::dataplane::transport::{ + InboundLink, PacketLink, PacketTransport, SharedLink, TransportError, +}; +use tsunagi::identity::{DeviceIdentity, NetworkId}; + +use crate::cert::{generate_self_signed_cert, make_client_config, make_server_config}; + +/// Default TCP port to bind and listen on. +pub const DEFAULT_PORT: u16 = 443; + +/// Fallback port if 443 is unprivileged on Linux and denied. +pub const FALLBACK_PORT: u16 = 8443; + +/// Maximum payload length for a single framed datagram over TLS. +pub const MAX_FRAME_SIZE: usize = 65535; + +/// Initial handshake message on established TLS stream (128 bytes). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct LinkHandshake { + /// The overlay network ID. + pub network: NetworkId, + /// The connecting client's endpoint ID. + pub client_id: EndpointId, + /// Ed25519 signature over `tsunagi-tcp-tls-link-v1` + network + server ID. + pub signature: [u8; 64], +} + +impl LinkHandshake { + /// Total encoded length in bytes. + pub const LEN: usize = 32 + 32 + 64; // 128 + + /// Serializes to fixed-size byte array. + pub fn to_bytes(&self) -> [u8; Self::LEN] { + let mut buf = [0u8; Self::LEN]; + buf[0..32].copy_from_slice(self.network.as_bytes()); + buf[32..64].copy_from_slice(self.client_id.as_bytes()); + buf[64..128].copy_from_slice(&self.signature); + buf + } + + /// Deserializes from byte slice. + pub fn from_bytes(bytes: &[u8]) -> Option { + if bytes.len() != Self::LEN { + return None; + } + let network = NetworkId::from_bytes(bytes[0..32].try_into().ok()?); + let client_id = EndpointId::from_bytes(bytes[32..64].try_into().ok()?).ok()?; + let signature: [u8; 64] = bytes[64..128].try_into().ok()?; + Some(Self { + network, + client_id, + signature, + }) + } +} + +/// Acknowledgement message on established TLS stream (1 byte). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct LinkAck { + /// Whether the link was accepted by the server. + pub accepted: bool, +} + +impl LinkAck { + /// Total encoded length in bytes. + pub const LEN: usize = 1; + + /// Serializes to fixed-size byte array. + pub fn to_bytes(&self) -> [u8; 1] { + [if self.accepted { 1 } else { 0 }] + } + + /// Deserializes from byte slice. + pub fn from_bytes(bytes: &[u8]) -> Option { + if bytes.len() != 1 { + return None; + } + Some(Self { + accepted: bytes[0] == 1, + }) + } +} + +fn link_handshake_bytes(network: NetworkId, server_id: EndpointId) -> Vec { + let mut bytes = Vec::with_capacity(23 + 32 + 32); + bytes.extend_from_slice(b"tsunagi-tcp-tls-link-v1"); + bytes.extend_from_slice(network.as_bytes()); + bytes.extend_from_slice(server_id.as_bytes()); + bytes +} + +/// An authenticated datagram link carried over a TCP TLS 1.3 stream. +#[derive(Debug)] +pub struct TcpTlsLink { + network: NetworkId, + peer: EndpointId, + tx: mpsc::Sender, + rx: tokio::sync::Mutex>, + closed_notify: Arc, + is_closed: Arc, + remote_addr: String, +} + +impl TcpTlsLink { + /// Creates a new `TcpTlsLink` from a stream that has already passed the TLS handshake. + pub fn new(network: NetworkId, peer: EndpointId, stream: S, remote_addr: String) -> Self + where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static, + { + let (mut reader, mut writer) = tokio::io::split(stream); + + let (tx_out, mut rx_out) = mpsc::channel::(256); + let (tx_in, rx_in) = mpsc::channel::(256); + + let is_closed = Arc::new(AtomicBool::new(false)); + let closed_notify = Arc::new(Notify::new()); + + let closed_flag_w = Arc::clone(&is_closed); + let closed_notify_w = Arc::clone(&closed_notify); + tokio::spawn(async move { + while let Some(msg) = rx_out.recv().await { + if msg.len() > MAX_FRAME_SIZE { + continue; + } + let len = msg.len() as u16; + if writer.write_all(&len.to_be_bytes()).await.is_err() { + break; + } + if writer.write_all(&msg).await.is_err() { + break; + } + if writer.flush().await.is_err() { + break; + } + } + closed_flag_w.store(true, Ordering::Release); + closed_notify_w.notify_waiters(); + }); + + let closed_flag_r = Arc::clone(&is_closed); + let closed_notify_r = Arc::clone(&closed_notify); + tokio::spawn(async move { + let mut len_buf = [0u8; 2]; + loop { + if reader.read_exact(&mut len_buf).await.is_err() { + break; + } + let len = u16::from_be_bytes(len_buf) as usize; + if len > MAX_FRAME_SIZE { + break; + } + let mut buf = vec![0u8; len]; + if reader.read_exact(&mut buf).await.is_err() { + break; + } + if tx_in.send(Bytes::from(buf)).await.is_err() { + break; + } + } + closed_flag_r.store(true, Ordering::Release); + closed_notify_r.notify_waiters(); + }); + + Self { + network, + peer, + tx: tx_out, + rx: tokio::sync::Mutex::new(rx_in), + closed_notify, + is_closed, + remote_addr, + } + } +} + +impl PacketLink for TcpTlsLink { + fn network(&self) -> NetworkId { + self.network + } + + fn peer(&self) -> EndpointId { + self.peer + } + + fn max_datagram_size(&self) -> usize { + MAX_FRAME_SIZE + } + + fn send(&self, payload: Bytes) -> Result<(), TransportError> { + if payload.len() > MAX_FRAME_SIZE { + return Err(TransportError::TooLarge { + size: payload.len(), + limit: MAX_FRAME_SIZE, + }); + } + if self.is_closed.load(Ordering::Acquire) { + return Err(TransportError::Closed); + } + self.tx + .try_send(payload) + .map_err(|_| TransportError::Closed) + } + + fn recv(&self) -> BoxFuture<'_, Option> { + Box::pin(async move { + let mut guard = self.rx.lock().await; + guard.recv().await + }) + } + + fn closed(&self) -> BoxFuture<'_, ()> { + Box::pin(async move { + if self.is_closed.load(Ordering::Acquire) { + return; + } + self.closed_notify.notified().await; + }) + } + + fn is_closed(&self) -> bool { + self.is_closed.load(Ordering::Acquire) + } + + fn path_description(&self) -> String { + format!("tcp-tls via {}", self.remote_addr) + } +} + +/// The `tcp-tls` packet transport. +pub struct TcpTlsTransport { + identity: DeviceIdentity, + local_id: EndpointId, + listen_port: u16, + peer_addresses: std::sync::RwLock>>, + server_config: Arc, + listener: Option>, +} + +impl std::fmt::Debug for TcpTlsTransport { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("TcpTlsTransport") + .field("local_id", &self.local_id.fmt_short().to_string()) + .field("listen_port", &self.listen_port) + .finish() + } +} + +impl TcpTlsTransport { + /// Binds a new TCP TLS transport using the device identity and configured port. + pub async fn bind( + identity: &DeviceIdentity, + requested_port: Option, + ) -> Result> { + let (cert, key) = generate_self_signed_cert(&identity.secret_bytes())?; + let server_config = make_server_config(cert, key)?; + + let target_port = requested_port.unwrap_or(DEFAULT_PORT); + let bind_addr = SocketAddr::from(([0, 0, 0, 0], target_port)); + + let listener = match TcpListener::bind(bind_addr).await { + Ok(l) => { + tracing::info!(port = target_port, "TCP TLS listening"); + Some(Arc::new(l)) + } + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => { + let exe = std::env::current_exe() + .map(|p| p.display().to_string()) + .unwrap_or_else(|_| "tsunagi".into()); + + tracing::warn!( + port = target_port, + %err, + "Cannot bind to port {target_port} without privileges. \ + To allow port 443 and TUN on Linux, grant capabilities once with:\n \ + sudo setcap cap_net_admin,cap_net_bind_service+p {exe}\n\ + Falling back to port {FALLBACK_PORT}." + ); + + if requested_port.is_none() { + let fallback_addr = SocketAddr::from(([0, 0, 0, 0], FALLBACK_PORT)); + let l = TcpListener::bind(fallback_addr).await?; + tracing::info!(port = FALLBACK_PORT, "TCP TLS listening on fallback port"); + Some(Arc::new(l)) + } else { + return Err(Box::new(err)); + } + } + Err(err) => return Err(Box::new(err)), + }; + + let bound_port = listener + .as_ref() + .and_then(|l| l.local_addr().ok()) + .map(|a| a.port()) + .unwrap_or(target_port); + + Ok(Self { + identity: identity.clone(), + local_id: identity.endpoint_id(), + listen_port: bound_port, + peer_addresses: std::sync::RwLock::new(HashMap::new()), + server_config, + listener, + }) + } + + /// The port this transport is listening on. + pub fn bound_port(&self) -> u16 { + self.listen_port + } + + /// Records an observed TCP target address for a peer. + pub fn set_peer_addr(&self, peer: EndpointId, addr: SocketAddr) { + if let Ok(mut guard) = self.peer_addresses.write() { + let addrs = guard.entry(peer).or_default(); + if !addrs.contains(&addr) { + addrs.push(addr); + } + } + } + + /// Runs the accept loop for inbound TCP TLS connections. + pub async fn accept_loop(&self, on_inbound: F) + where + F: Fn(InboundLink) + Send + Sync + Clone + 'static, + { + let Some(listener) = self.listener.as_ref().cloned() else { + return; + }; + + let acceptor = tokio_rustls::TlsAcceptor::from(Arc::clone(&self.server_config)); + let local_id = self.local_id; + + while let Ok((socket, remote_addr)) = listener.accept().await { + let acceptor = acceptor.clone(); + let on_inbound = on_inbound.clone(); + + tokio::spawn(async move { + let mut tls_stream = match acceptor.accept(socket).await { + Ok(s) => s, + Err(err) => { + tracing::debug!(%err, "inbound TLS handshake failed"); + return; + } + }; + + let mut hello_buf = [0u8; LinkHandshake::LEN]; + if tls_stream.read_exact(&mut hello_buf).await.is_err() { + return; + } + let Some(hello) = LinkHandshake::from_bytes(&hello_buf) else { + return; + }; + + let msg = link_handshake_bytes(hello.network, local_id); + let sig = Signature::from_bytes(&hello.signature); + if hello.client_id.verify(&msg, &sig).is_err() { + tracing::debug!("LinkHandshake signature verification failed"); + return; + } + + let ack = LinkAck { accepted: true }; + let ack_bytes = ack.to_bytes(); + if tls_stream.write_all(&ack_bytes).await.is_err() { + return; + } + if tls_stream.flush().await.is_err() { + return; + } + + let link = TcpTlsLink::new( + hello.network, + hello.client_id, + tls_stream, + remote_addr.to_string(), + ); + on_inbound(InboundLink { + network: hello.network, + peer: hello.client_id, + protocol: "tcp-tls".into(), + link: Arc::new(link), + }); + }); + } + } +} + +impl PacketTransport for TcpTlsTransport { + fn name(&self) -> &str { + "tcp-tls" + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } + + fn open<'a>( + &'a self, + network: NetworkId, + peer: EndpointId, + protocol: &'a str, + ) -> BoxFuture<'a, Result> { + Box::pin(async move { + if protocol != "tcp-tls" { + return Err(TransportError::Declined(protocol.to_string())); + } + + let target_addrs = { + let guard = self + .peer_addresses + .read() + .map_err(|_| TransportError::Other("lock poisoned".into()))?; + guard.get(&peer).cloned().unwrap_or_default() + }; + + if target_addrs.is_empty() { + return Err(TransportError::Unreachable( + "no known TCP address for peer".into(), + )); + } + + let mut connected_socket = None; + let mut connected_addr = None; + for addr in &target_addrs { + match tokio::time::timeout(Duration::from_secs(3), TcpStream::connect(*addr)).await + { + Ok(Ok(socket)) => { + connected_socket = Some(socket); + connected_addr = Some(*addr); + break; + } + _ => continue, + } + } + + let (socket, addr) = match (connected_socket, connected_addr) { + (Some(s), Some(a)) => (s, a), + _ => { + return Err(TransportError::Unreachable( + "cannot connect to any TCP address for peer".into(), + )); + } + }; + + let client_config = + make_client_config(peer).map_err(|err| TransportError::Other(err.to_string()))?; + let connector = TlsConnector::from(client_config); + + let server_name = rustls::pki_types::ServerName::try_from("tsunagi.local") + .map_err(|err| TransportError::Other(err.to_string()))? + .to_owned(); + + let mut tls_stream = connector + .connect(server_name, socket) + .await + .map_err(|err| TransportError::Other(format!("TLS handshake failed: {err}")))?; + + // Send LinkHandshake (fixed 128 bytes) + let msg = link_handshake_bytes(network, peer); + let sig = self.identity.sign(&msg); + let hello = LinkHandshake { + network, + client_id: self.local_id, + signature: sig.to_bytes(), + }; + let hello_bytes = hello.to_bytes(); + tls_stream + .write_all(&hello_bytes) + .await + .map_err(|err| TransportError::Other(err.to_string()))?; + tls_stream + .flush() + .await + .map_err(|err| TransportError::Other(err.to_string()))?; + + // Read LinkAck (fixed 1 byte) + let mut ack_buf = [0u8; LinkAck::LEN]; + tls_stream + .read_exact(&mut ack_buf) + .await + .map_err(|err| TransportError::Other(err.to_string()))?; + let ack = LinkAck::from_bytes(&ack_buf) + .ok_or_else(|| TransportError::Other("invalid link ack".into()))?; + + if !ack.accepted { + return Err(TransportError::Declined("link declined".into())); + } + + let link = TcpTlsLink::new(network, peer, tls_stream, addr.to_string()); + Ok(Arc::new(link) as SharedLink) + }) + } +} diff --git a/crates/tsunagi-tcp-tls/tests/cert_tests.rs b/crates/tsunagi-tcp-tls/tests/cert_tests.rs new file mode 100644 index 0000000..c3471bf --- /dev/null +++ b/crates/tsunagi-tcp-tls/tests/cert_tests.rs @@ -0,0 +1,52 @@ +#![allow(missing_docs)] +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use rustls::client::danger::ServerCertVerifier; +use rustls::pki_types::{ServerName, UnixTime}; +use tsunagi::identity::DeviceIdentity; +use tsunagi_tcp_tls::cert::{PinnedEndpointVerifier, generate_self_signed_cert}; + +#[test] +fn test_generate_and_pin_certificate() { + let identity = DeviceIdentity::generate(); + let peer_id = identity.endpoint_id(); + + let (cert, _key) = generate_self_signed_cert(&identity.secret_bytes()) + .expect("cert generation should succeed"); + + let verifier = PinnedEndpointVerifier::new(peer_id); + let server_name = ServerName::try_from("tsunagi.local").unwrap(); + + let verified = verifier.verify_server_cert(&cert, &[], &server_name, &[], UnixTime::now()); + + assert!( + verified.is_ok(), + "certificate matching peer endpoint id must verify" + ); + + let other_identity = DeviceIdentity::generate(); + let wrong_verifier = PinnedEndpointVerifier::new(other_identity.endpoint_id()); + + let wrong_verified = + wrong_verifier.verify_server_cert(&cert, &[], &server_name, &[], UnixTime::now()); + + assert!( + wrong_verified.is_err(), + "certificate with wrong endpoint id must fail verification" + ); +} + +#[test] +fn test_make_client_and_server_config() { + let identity = DeviceIdentity::generate(); + let peer_id = identity.endpoint_id(); + + let (cert, key) = generate_self_signed_cert(&identity.secret_bytes()) + .expect("cert generation should succeed"); + + let server_config = tsunagi_tcp_tls::cert::make_server_config(cert, key); + assert!(server_config.is_ok()); + + let client_config = tsunagi_tcp_tls::cert::make_client_config(peer_id); + assert!(client_config.is_ok()); +} diff --git a/crates/tsunagi-tcp-tls/tests/crypto_tests.rs b/crates/tsunagi-tcp-tls/tests/crypto_tests.rs new file mode 100644 index 0000000..118423f --- /dev/null +++ b/crates/tsunagi-tcp-tls/tests/crypto_tests.rs @@ -0,0 +1,73 @@ +#![allow(missing_docs)] +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use tsunagi_tcp_tls::crypto::PeerCipher; +use x25519_dalek::{PublicKey, StaticSecret}; + +#[test] +fn test_peer_cipher_bidirectional_roundtrip() { + let peer_a_id = iroh::SecretKey::generate().public(); + let peer_b_id = iroh::SecretKey::generate().public(); + + let secret_a = StaticSecret::from([10u8; 32]); + let public_a = PublicKey::from(&secret_a); + + let secret_b = StaticSecret::from([20u8; 32]); + let public_b = PublicKey::from(&secret_b); + + let cipher_a = PeerCipher::new(peer_a_id, peer_b_id, &secret_a, &public_b); + let cipher_b = PeerCipher::new(peer_b_id, peer_a_id, &secret_b, &public_a); + + // A -> B + let msg_from_a = b"hello from node A to node B over Noise E2EE"; + let encrypted_by_a = cipher_a + .encrypt(msg_from_a) + .expect("encryption should succeed"); + let decrypted_by_b = cipher_b + .decrypt(&encrypted_by_a) + .expect("decryption should succeed"); + assert_eq!(&decrypted_by_b[..], msg_from_a); + + // B -> A + let msg_from_b = b"hello back from node B to node A"; + let encrypted_by_b = cipher_b + .encrypt(msg_from_b) + .expect("encryption should succeed"); + let decrypted_by_a = cipher_a + .decrypt(&encrypted_by_b) + .expect("decryption should succeed"); + assert_eq!(&decrypted_by_a[..], msg_from_b); + + // Sequence nonces increment + let msg2 = b"packet 2"; + let enc2 = cipher_a.encrypt(msg2).expect("encryption should succeed"); + assert_ne!(encrypted_by_a[..8], enc2[..8], "nonces should differ"); + let dec2 = cipher_b.decrypt(&enc2).expect("decryption should succeed"); + assert_eq!(&dec2[..], msg2); +} + +#[test] +fn test_tampered_payload_rejected() { + let peer_a_id = iroh::SecretKey::generate().public(); + let peer_b_id = iroh::SecretKey::generate().public(); + + let secret_a = StaticSecret::from([10u8; 32]); + let public_a = PublicKey::from(&secret_a); + + let secret_b = StaticSecret::from([20u8; 32]); + let public_b = PublicKey::from(&secret_b); + + let cipher_a = PeerCipher::new(peer_a_id, peer_b_id, &secret_a, &public_b); + let cipher_b = PeerCipher::new(peer_b_id, peer_a_id, &secret_b, &public_a); + + let encrypted = cipher_a.encrypt(b"secret payload").expect("encrypt"); + // Corrupt one byte of ciphertext + let len = encrypted.len(); + let mut modified = encrypted.to_vec(); + modified[len - 1] ^= 0xff; + + assert!( + cipher_b.decrypt(&modified).is_err(), + "corrupted ciphertext must fail AEAD verification" + ); +} diff --git a/crates/tsunagi-tcp-tls/tests/plugin_tests.rs b/crates/tsunagi-tcp-tls/tests/plugin_tests.rs new file mode 100644 index 0000000..bbd44d2 --- /dev/null +++ b/crates/tsunagi-tcp-tls/tests/plugin_tests.rs @@ -0,0 +1,149 @@ +#![allow(missing_docs)] +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use std::sync::Arc; +use std::time::Duration; + +use bytes::Bytes; +use tokio::sync::mpsc; +use tsunagi::dataplane::transport::{InboundLink, PacketTransport}; +use tsunagi::dataplane::{IpPlugin, PacketSink, PluginContext}; +use tsunagi::identity::{DeviceIdentity, NetworkId}; +use tsunagi_tcp_tls::{TcpTlsCodec, TcpTlsPlugin, TcpTlsTransport}; + +struct TestSink { + delivered: mpsc::Sender<(NetworkId, iroh::EndpointId, Bytes)>, +} + +impl PacketSink for TestSink { + fn deliver( + &self, + network: NetworkId, + from: iroh::EndpointId, + packet: Bytes, + ) -> tsunagi::BoxFuture<'_, ()> { + let delivered = self.delivered.clone(); + Box::pin(async move { + let _ = delivered.send((network, from, packet)).await; + }) + } +} + +#[tokio::test] +async fn test_tcp_tls_plugin_end_to_end() { + let identity_a = DeviceIdentity::generate(); + let peer_a = identity_a.endpoint_id(); + + let identity_b = DeviceIdentity::generate(); + let peer_b = identity_b.endpoint_id(); + + let network = NetworkId::from_bytes([77u8; 32]); + + // Transports + let transport_b = Arc::new( + TcpTlsTransport::bind(&identity_b, Some(0)) + .await + .expect("bind B"), + ); + let port_b = transport_b.bound_port(); + + let transport_a = Arc::new( + TcpTlsTransport::bind(&identity_a, Some(0)) + .await + .expect("bind A"), + ); + + // Codecs and plugins + let codec_a = TcpTlsCodec::new(peer_a, Arc::clone(&transport_a)); + let plugin_a = Arc::new(TcpTlsPlugin::new(codec_a)); + + let codec_b = TcpTlsCodec::new(peer_b, Arc::clone(&transport_b)); + let plugin_b = Arc::new(TcpTlsPlugin::new(codec_b)); + + // Attach sinks + let (sink_a_tx, mut sink_a_rx) = mpsc::channel(10); + plugin_a.attach(PluginContext::with_sink(Arc::new(TestSink { + delivered: sink_a_tx, + }))); + + let (sink_b_tx, mut sink_b_rx) = mpsc::channel(10); + plugin_b.attach(PluginContext::with_sink(Arc::new(TestSink { + delivered: sink_b_tx, + }))); + + // Capability exchange + let cap_a = plugin_a + .local_capability(network) + .expect("cap A") + .expect("some cap A"); + let cap_b = plugin_b + .local_capability(network) + .expect("cap B") + .expect("some cap B"); + + plugin_a + .on_peer_capability(network, peer_b, &cap_b) + .expect("A on cap B"); + plugin_b + .on_peer_capability(network, peer_a, &cap_a) + .expect("B on cap A"); + + // Spawn accept loop on B + let (inbound_tx, mut inbound_rx) = mpsc::channel::(1); + let transport_b_clone = Arc::clone(&transport_b); + tokio::spawn(async move { + transport_b_clone + .accept_loop(move |inbound| { + let _ = inbound_tx.try_send(inbound); + }) + .await; + }); + + // Make sure A knows B's loopback address + transport_a.set_peer_addr(peer_b, format!("127.0.0.1:{port_b}").parse().unwrap()); + + // Connect link from A to B + let link_a = transport_a + .open(network, peer_b, "tcp-tls") + .await + .expect("open link from A"); + let inbound_b = tokio::time::timeout(Duration::from_secs(5), inbound_rx.recv()) + .await + .expect("recv timeout") + .expect("recv inbound"); + let link_b = inbound_b.link; + + // Install links into plugins + plugin_a.on_peer_link(network, peer_b, link_a); + plugin_b.on_peer_link(network, peer_a, link_b); + + // Carry packet from A to B + let ip_packet_a = Bytes::from_static(b"4500003c000100004011e0e8c0a80101c0a8010200000000"); + let carried_a = plugin_a.carry(network, peer_b, ip_packet_a.clone()); + assert!(carried_a, "plugin A should carry packet"); + + let (net_received, from_peer, delivered_pkt) = + tokio::time::timeout(Duration::from_secs(3), sink_b_rx.recv()) + .await + .expect("timeout waiting for delivery on B") + .expect("packet on B"); + + assert_eq!(net_received, network); + assert_eq!(from_peer, peer_a); + assert_eq!(delivered_pkt, ip_packet_a); + + // Carry reply packet from B to A + let ip_packet_b = Bytes::from_static(b"reply-packet-from-b"); + let carried_b = plugin_b.carry(network, peer_a, ip_packet_b.clone()); + assert!(carried_b, "plugin B should carry packet"); + + let (net_received, from_peer, delivered_pkt) = + tokio::time::timeout(Duration::from_secs(3), sink_a_rx.recv()) + .await + .expect("timeout waiting for delivery on A") + .expect("packet on A"); + + assert_eq!(net_received, network); + assert_eq!(from_peer, peer_b); + assert_eq!(delivered_pkt, ip_packet_b); +} diff --git a/crates/tsunagi-tcp-tls/tests/transport_tests.rs b/crates/tsunagi-tcp-tls/tests/transport_tests.rs new file mode 100644 index 0000000..77cdc16 --- /dev/null +++ b/crates/tsunagi-tcp-tls/tests/transport_tests.rs @@ -0,0 +1,96 @@ +#![allow(missing_docs)] +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use bytes::Bytes; +use tokio::sync::mpsc; +use tsunagi::dataplane::transport::{InboundLink, PacketTransport}; +use tsunagi::identity::{DeviceIdentity, NetworkId}; +use tsunagi_tcp_tls::TcpTlsTransport; + +#[tokio::test] +async fn test_tcp_tls_transport_loopback() { + let identity_a = DeviceIdentity::generate(); + let peer_a = identity_a.endpoint_id(); + + let identity_b = DeviceIdentity::generate(); + let peer_b = identity_b.endpoint_id(); + + let network = NetworkId::from_bytes([42u8; 32]); + + // Bind transport B on port 0 (ephemeral) + let transport_b = Arc::new( + TcpTlsTransport::bind(&identity_b, Some(0)) + .await + .expect("bind server should succeed"), + ); + let port_b = transport_b.bound_port(); + assert!(port_b > 0); + + // Bind transport A on port 0 (ephemeral) + let transport_a = Arc::new( + TcpTlsTransport::bind(&identity_a, Some(0)) + .await + .expect("bind client should succeed"), + ); + + // Register B's address on transport A + let addr_b: SocketAddr = format!("127.0.0.1:{port_b}").parse().unwrap(); + transport_a.set_peer_addr(peer_b, addr_b); + + // Spawn accept loop on B + let (inbound_tx, mut inbound_rx) = mpsc::channel::(1); + let transport_b_clone = Arc::clone(&transport_b); + tokio::spawn(async move { + transport_b_clone + .accept_loop(move |inbound| { + let _ = inbound_tx.try_send(inbound); + }) + .await; + }); + + // Open link from A to B + let link_a = tokio::time::timeout( + Duration::from_secs(5), + transport_a.open(network, peer_b, "tcp-tls"), + ) + .await + .expect("open should not timeout") + .expect("open should succeed"); + + // Receive inbound link on B + let inbound_b = tokio::time::timeout(Duration::from_secs(5), inbound_rx.recv()) + .await + .expect("accept should not timeout") + .expect("inbound link should arrive"); + + assert_eq!(inbound_b.network, network); + assert_eq!(inbound_b.peer, peer_a); + assert_eq!(inbound_b.protocol, "tcp-tls"); + let link_b = inbound_b.link; + + // Send from A to B + let data_a_to_b = Bytes::from_static(b"packet from A to B"); + link_a.send(data_a_to_b.clone()).expect("send A->B"); + let received_on_b = tokio::time::timeout(Duration::from_secs(2), link_b.recv()) + .await + .expect("recv B timeout") + .expect("recv B option"); + assert_eq!(received_on_b, data_a_to_b); + + // Send from B to A + let data_b_to_a = Bytes::from_static(b"reply from B to A"); + link_b.send(data_b_to_a.clone()).expect("send B->A"); + let received_on_a = tokio::time::timeout(Duration::from_secs(2), link_a.recv()) + .await + .expect("recv A timeout") + .expect("recv A option"); + assert_eq!(received_on_a, data_b_to_a); + + // Path descriptions + assert!(link_a.path_description().contains("tcp-tls")); + assert!(link_b.path_description().contains("tcp-tls")); +} diff --git a/crates/tsunagi/src/agent/mod.rs b/crates/tsunagi/src/agent/mod.rs index a5473ea..afc81a5 100644 --- a/crates/tsunagi/src/agent/mod.rs +++ b/crates/tsunagi/src/agent/mod.rs @@ -281,11 +281,19 @@ impl Agent { // The data plane rides on iroh too, which is where it gets hole // punching and relay fallback from. It is a separate ALPN and a // separate connection, so the two planes stay independent. - let transport: Arc = Arc::new(IrohTransport::new( + let iroh_transport: Arc = Arc::new(IrohTransport::new( inner.adapter.clone(), Arc::clone(&inner.limits), Arc::new(TransportCtx(Arc::downgrade(&inner))) as Arc, )); + let transport: Arc = if inner.config.custom_transports.is_empty() { + iroh_transport + } else { + Arc::new(crate::dataplane::MultiPacketTransport::new( + iroh_transport, + inner.config.custom_transports.clone(), + )) + }; let _ = inner.transport.set(transport); if let CacheOutcome::Reset(reason) = inner.storage.cache_outcome().clone() { @@ -346,6 +354,26 @@ impl Agent { self.inner.adapter.loopback_addr() } + /// Delivers an inbound data plane link to the matching network runtime. + pub async fn install_inbound_link( + &self, + inbound: crate::dataplane::transport::InboundLink, + ) -> Result<()> { + let network = inbound.network; + let sender = { + let networks = self.inner.networks.read().await; + networks.get(&network).map(|handle| handle.commands.clone()) + }; + let Some(sender) = sender else { + return Err(Error::NetworkNotActive(network)); + }; + sender + .send(NetCommand::InboundLink(Box::new(inbound))) + .await + .map_err(|_| Error::NetworkNotActive(network))?; + Ok(()) + } + /// Reserves the configured overlay range for a network, if it can. /// /// Done here, where activations are serialised, rather than inside the @@ -1200,8 +1228,23 @@ async fn handle_inbound_data(inner: Arc, conn: iroh::endpoint::Connection conn.close(5u32.into(), b"data plane not ready"); return; }; - // Downcasting is avoided by keeping the accept side on the concrete type. - let Some(iroh_transport) = transport.as_ref().as_any().downcast_ref::() else { + let iroh_transport = + if let Some(iroh) = transport.as_ref().as_any().downcast_ref::() { + Some(iroh) + } else if let Some(multi) = transport + .as_ref() + .as_any() + .downcast_ref::() + { + multi + .default_transport() + .as_ref() + .as_any() + .downcast_ref::() + } else { + None + }; + let Some(iroh_transport) = iroh_transport else { conn.close(5u32.into(), b"unsupported data transport"); return; }; diff --git a/crates/tsunagi/src/agent/network.rs b/crates/tsunagi/src/agent/network.rs index 1b3efdd..efdfb82 100644 --- a/crates/tsunagi/src/agent/network.rs +++ b/crates/tsunagi/src/agent/network.rs @@ -1185,28 +1185,27 @@ impl Runtime { }); } - // Both the name and the version have to match. A peer offering - // `wg-quic` at a version this build does not speak is not a peer to - // carry traffic with, and a link opened anyway would fail later and - // say less about why. - let wanted: Vec<(EndpointId, String)> = self - .sessions - .values() - .flat_map(|session| { - let peer = session.peer; - session + // Choose the highest priority mutually supported protocol for each peer. + // If a direct data link is already open or opening, do not dial another. + let mut wanted: Vec<(EndpointId, String)> = Vec::new(); + for session in self.sessions.values() { + let peer = session.peer; + if self.links.keys().any(|(p, _)| *p == peer) + || self.opening.iter().any(|(p, _)| *p == peer) + { + continue; + } + for (name, ours) in &served { + let matches = session .capabilities .iter() - .filter(|capability| capability.enabled) - .map(move |capability| (peer, capability.protocol.clone(), capability.version)) - }) - .filter(|(_, protocol, version)| { - served - .iter() - .any(|(name, ours)| name == protocol && ours == version) - }) - .map(|(peer, protocol, _)| (peer, protocol)) - .collect(); + .any(|c| c.enabled && c.protocol == *name && c.version == *ours); + if matches { + wanted.push((peer, name.clone())); + break; + } + } + } for (peer, protocol) in wanted { let key = (peer, protocol.clone()); diff --git a/crates/tsunagi/src/config.rs b/crates/tsunagi/src/config.rs index 8baca3b..7462282 100644 --- a/crates/tsunagi/src/config.rs +++ b/crates/tsunagi/src/config.rs @@ -304,6 +304,9 @@ pub struct AgentConfig { pub reconnect: ReconnectPolicy, /// IP plugins whose capabilities are announced and dispatched. pub plugins: Vec, + /// Custom packet transports registered per protocol. + pub custom_transports: + std::collections::HashMap>, /// Where the overlay interface comes from, when the agent should have one. /// /// One agent, one interface: it belongs here rather than to a protocol, @@ -345,6 +348,7 @@ impl AgentConfig { limits: Limits::default(), reconnect: ReconnectPolicy::default(), plugins: Vec::new(), + custom_transports: std::collections::HashMap::new(), tun_factory: None, interface_name: "tsun0".to_string(), interface_mtu: 1280, @@ -430,6 +434,16 @@ impl AgentConfig { self } + /// Registers a custom packet transport for a specific protocol. + pub fn with_custom_transport( + mut self, + protocol: impl Into, + transport: Arc, + ) -> Self { + self.custom_transports.insert(protocol.into(), transport); + self + } + /// Gives the agent an overlay interface, from this factory. pub fn with_interface( mut self, diff --git a/crates/tsunagi/src/dataplane/mod.rs b/crates/tsunagi/src/dataplane/mod.rs index 2af8215..ba3adf2 100644 --- a/crates/tsunagi/src/dataplane/mod.rs +++ b/crates/tsunagi/src/dataplane/mod.rs @@ -22,6 +22,7 @@ pub mod relay; pub mod routing; pub mod transport; +pub mod tunnel; use std::sync::Arc; @@ -31,7 +32,10 @@ use tokio::sync::mpsc; use crate::BoxFuture; use crate::identity::NetworkId; -pub use transport::{PacketLink, PacketTransport, SharedLink, TransportError}; +pub use transport::{ + MultiPacketTransport, PacketLink, PacketTransport, SharedLink, TransportError, +}; +pub use tunnel::{GenericTunnelPlugin, TunnelCodec}; /// Maximum length of a plugin protocol identifier. pub const MAX_PROTOCOL_ID_LEN: usize = 32; @@ -179,6 +183,15 @@ impl PluginContext { } } + /// A context configured with a custom packet sink. + pub fn with_sink(sink: Arc) -> Self { + Self { + sender: None, + local: None, + sink: Some(sink), + } + } + /// This agent's own endpoint id, when the context is attached. /// /// A plugin needs it to find itself in the agreed allocation. diff --git a/crates/tsunagi/src/dataplane/transport/mod.rs b/crates/tsunagi/src/dataplane/transport/mod.rs index f351254..73e96d2 100644 --- a/crates/tsunagi/src/dataplane/transport/mod.rs +++ b/crates/tsunagi/src/dataplane/transport/mod.rs @@ -32,6 +32,8 @@ mod fragments; pub mod iroh_link; +use std::sync::Arc; + use bytes::Bytes; use iroh::EndpointId; @@ -155,3 +157,51 @@ pub trait PacketTransport: Send + Sync + std::fmt::Debug + 'static { protocol: &'a str, ) -> BoxFuture<'a, Result>; } + +/// A composite packet transport that dispatches requests by protocol id. +#[derive(Debug)] +pub struct MultiPacketTransport { + default: Arc, + by_protocol: std::collections::HashMap>, +} + +impl MultiPacketTransport { + /// Creates a new composite transport with a default fallback transport. + pub fn new( + default: Arc, + by_protocol: std::collections::HashMap>, + ) -> Self { + Self { + default, + by_protocol, + } + } + + /// Access the default underlying transport. + pub fn default_transport(&self) -> &Arc { + &self.default + } +} + +impl PacketTransport for MultiPacketTransport { + fn name(&self) -> &str { + "multi" + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } + + fn open<'a>( + &'a self, + network: NetworkId, + peer: EndpointId, + protocol: &'a str, + ) -> BoxFuture<'a, Result> { + if let Some(transport) = self.by_protocol.get(protocol) { + transport.open(network, peer, protocol) + } else { + self.default.open(network, peer, protocol) + } + } +} diff --git a/crates/tsunagi/src/dataplane/tunnel.rs b/crates/tsunagi/src/dataplane/tunnel.rs new file mode 100644 index 0000000..8dd978c --- /dev/null +++ b/crates/tsunagi/src/dataplane/tunnel.rs @@ -0,0 +1,296 @@ +//! Generic harness for IP tunnel plugins. +//! +//! Implementing [`super::IpPlugin`] directly requires managing per-peer background +//! tasks, synchronization, packet queues, and graceful shutdown. +//! +//! [`TunnelCodec`] provides a simplified interface where a plugin author only +//! defines cryptographic transformation and handshake payloads. +//! [`GenericTunnelPlugin`] adapts any [`TunnelCodec`] into a full [`super::IpPlugin`]. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, RwLock}; + +use bytes::Bytes; +use iroh::EndpointId; +use tokio::task::JoinHandle; + +use super::{IpPlugin, PluginCapability, PluginContext, PluginError, ProtocolOption, SharedLink}; +use crate::BoxFuture; +use crate::identity::NetworkId; + +/// The simplified contract for an IP tunnel protocol implementation. +/// +/// Implementors only handle key announcements, link lifecycle events, +/// and packet encrypt/decrypt. Concurrency and task management are handled +/// by [`GenericTunnelPlugin`]. +pub trait TunnelCodec: Send + Sync + 'static { + /// Protocol identifier, e.g. `tcp-tls`. + fn protocol_id(&self) -> &str; + + /// Protocol wire version. + fn protocol_version(&self) -> u16; + + /// Options accepted by the protocol. + fn options(&self) -> &'static [ProtocolOption] { + &[] + } + + /// Generates local capability announcement data for a network. + fn local_capability_data(&self, network: NetworkId) -> Result, PluginError>; + + /// Handles a peer's announced capability payload. + fn on_peer_capability_data( + &self, + network: NetworkId, + peer: EndpointId, + data: &[u8], + ) -> Result<(), PluginError>; + + /// Called when an authenticated link to a peer is established. + fn on_peer_link_up(&self, network: NetworkId, peer: EndpointId, link: &SharedLink) { + let _ = (network, peer, link); + } + + /// Called when a peer goes away. + fn on_peer_down(&self, network: NetworkId, peer: EndpointId) { + let _ = (network, peer); + } + + /// Encrypts an outbound IP packet into a payload to send over the link. + fn encrypt( + &self, + network: NetworkId, + peer: EndpointId, + packet: &[u8], + ) -> Result; + + /// Decrypts an inbound payload from the link into an IP packet. + fn decrypt( + &self, + network: NetworkId, + peer: EndpointId, + payload: &[u8], + ) -> Result; + + /// Called during shutdown. + fn shutdown<'a>(&'a self) -> BoxFuture<'a, ()> { + Box::pin(async {}) + } +} + +/// Generic wrapper that turns a [`TunnelCodec`] into a full [`IpPlugin`]. +pub struct GenericTunnelPlugin { + codec: Arc, + context: Mutex>, + links: RwLock>, + tasks: Mutex>>, +} + +impl GenericTunnelPlugin { + /// Wraps `codec` into an [`IpPlugin`]. + pub fn new(codec: C) -> Self { + Self { + codec: Arc::new(codec), + context: Mutex::new(None), + links: RwLock::new(HashMap::new()), + tasks: Mutex::new(HashMap::new()), + } + } + + /// Returns a reference to the underlying codec. + pub fn codec(&self) -> &C { + &self.codec + } +} + +impl std::fmt::Debug for GenericTunnelPlugin { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("GenericTunnelPlugin") + .field("protocol_id", &self.codec.protocol_id()) + .field("protocol_version", &self.codec.protocol_version()) + .finish() + } +} + +impl IpPlugin for GenericTunnelPlugin { + fn protocol_id(&self) -> &str { + self.codec.protocol_id() + } + + fn protocol_version(&self) -> u16 { + self.codec.protocol_version() + } + + fn options(&self) -> &'static [ProtocolOption] { + self.codec.options() + } + + fn attach(&self, context: PluginContext) { + if let Ok(mut guard) = self.context.lock() { + *guard = Some(context); + } + } + + fn local_capability( + &self, + network: NetworkId, + ) -> Result, PluginError> { + let data = self.codec.local_capability_data(network)?; + Ok(Some(PluginCapability { + protocol: self.codec.protocol_id().to_string(), + version: self.codec.protocol_version(), + enabled: true, + data, + })) + } + + fn on_peer_capability( + &self, + network: NetworkId, + peer: EndpointId, + capability: &PluginCapability, + ) -> Result<(), PluginError> { + if capability.protocol != self.codec.protocol_id() { + return Ok(()); + } + self.codec + .on_peer_capability_data(network, peer, &capability.data) + } + + fn carry(&self, network: NetworkId, peer: EndpointId, packet: bytes::Bytes) -> bool { + let link = { + let guard = match self.links.read() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + guard.get(&(network, peer)).cloned() + }; + + let Some(link) = link else { + return false; + }; + + match self.codec.encrypt(network, peer, &packet) { + Ok(ciphertext) => link.send(ciphertext).is_ok(), + Err(err) => { + tracing::debug!(%peer, %err, "tunnel encrypt failed"); + false + } + } + } + + fn on_peer_link(&self, network: NetworkId, peer: EndpointId, link: SharedLink) { + { + let mut guard = match self.links.write() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + guard.insert((network, peer), Arc::clone(&link)); + } + + self.codec.on_peer_link_up(network, peer, &link); + + let codec = Arc::clone(&self.codec); + let sink = match self.context.lock() { + Ok(g) => g.as_ref().map(|ctx| ctx.packet_sink()), + Err(p) => p.into_inner().as_ref().map(|ctx| ctx.packet_sink()), + }; + + let rx_link = Arc::clone(&link); + let handle = tokio::spawn(async move { + while let Some(payload) = rx_link.recv().await { + match codec.decrypt(network, peer, &payload) { + Ok(packet) => { + if let Some(ref sink) = sink { + sink.deliver(network, peer, packet).await; + } + } + Err(err) => { + tracing::debug!(%peer, %err, "tunnel decrypt failed"); + } + } + } + }); + + let mut guard = match self.tasks.lock() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + if let Some(old) = guard.insert((network, peer), handle) { + old.abort(); + } + } + + fn on_peer_gone(&self, network: NetworkId, peer: EndpointId) { + { + let mut guard = match self.tasks.lock() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + if let Some(task) = guard.remove(&(network, peer)) { + task.abort(); + } + } + { + let mut guard = match self.links.write() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + guard.remove(&(network, peer)); + } + self.codec.on_peer_down(network, peer); + } + + fn on_network_deactivated(&self, network: NetworkId) { + let keys_to_remove: Vec<(NetworkId, EndpointId)> = { + let guard = match self.tasks.lock() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + guard + .keys() + .filter(|(net, _)| *net == network) + .copied() + .collect() + }; + + { + let mut guard = match self.tasks.lock() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + for key in &keys_to_remove { + if let Some(task) = guard.remove(key) { + task.abort(); + } + } + } + + { + let mut guard = match self.links.write() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + guard.retain(|(net, _), _| *net != network); + } + } + + fn shutdown<'a>(&'a self) -> BoxFuture<'a, ()> { + Box::pin(async move { + let mut tasks = Vec::new(); + { + let mut guard = match self.tasks.lock() { + Ok(g) => g, + Err(p) => p.into_inner(), + }; + for (_, task) in guard.drain() { + tasks.push(task); + } + } + for task in tasks { + task.abort(); + } + self.codec.shutdown().await; + }) + } +} diff --git a/crates/tsunagi/src/identity/mod.rs b/crates/tsunagi/src/identity/mod.rs index 662c5c1..1ac4660 100644 --- a/crates/tsunagi/src/identity/mod.rs +++ b/crates/tsunagi/src/identity/mod.rs @@ -49,11 +49,16 @@ impl DeviceIdentity { self.secret.public() } - /// The raw secret key bytes, for persistence only. - pub(crate) fn secret_bytes(&self) -> [u8; 32] { + /// The raw secret key bytes, for persistence and transport certificate generation. + pub fn secret_bytes(&self) -> [u8; 32] { self.secret.to_bytes() } + /// Signs a message with the device's persistent secret key. + pub fn sign(&self, message: &[u8]) -> iroh::Signature { + self.secret.sign(message) + } + /// A clone of the iroh secret key, for endpoint construction only. pub(crate) fn secret_key(&self) -> SecretKey { self.secret.clone()