diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 113ebd31d..928eb54be 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -239,6 +239,37 @@ jobs: shell: pwsh run: sccache --show-stats + agent-tunnel-e2e: + name: agent tunnel end-to-end test + needs: [preflight] + runs-on: ubuntu-24.04 + timeout-minutes: 30 + + steps: + - name: Checkout ${{ github.repository }} + uses: actions/checkout@v6 + with: + ref: ${{ needs.preflight.outputs.ref }} + + - name: Setup Rust cache + uses: ./.github/actions/setup-rust-cache + with: + sccache-enabled: ${{ needs.preflight.outputs.sccache }} + + - name: Configure runner + run: docker info + + - name: Test agent tunnel + shell: pwsh + run: cargo test --locked -p testsuite --test integration_tests cli::agent::tunnel::docker_isolates_real_agent_dns_and_ip_routes -- --ignored --exact --nocapture + env: + AWS_LC_SYS_NO_ASM: true + + - name: Show sccache stats + if: ${{ needs.preflight.outputs.sccache == 'true' && !cancelled() }} + shell: pwsh + run: sccache --show-stats + check-dependencies: name: Check no ${{ matrix.banned }} in ${{ matrix.package }} [${{ matrix.target }}] needs: [preflight] @@ -1267,7 +1298,7 @@ jobs: success: name: Success if: ${{ always() }} - needs: [tests, lints, check-dependencies, jetsocat-lipo, devolutions-gateway-powershell, devolutions-gateway, devolutions-gateway-merge, devolutions-pedm-desktop, devolutions-agent, devolutions-agent-merge, devolutions-pedm-client, dotnet-utils-tests, winapi-sanitizer-tests, winapi-miri, pedm-simulator, secure-memory-verifier] + needs: [tests, agent-tunnel-e2e, lints, check-dependencies, jetsocat-lipo, devolutions-gateway-powershell, devolutions-gateway, devolutions-gateway-merge, devolutions-pedm-desktop, devolutions-agent, devolutions-agent-merge, devolutions-pedm-client, dotnet-utils-tests, winapi-sanitizer-tests, winapi-miri, pedm-simulator, secure-memory-verifier] runs-on: ubuntu-latest steps: diff --git a/Cargo.lock b/Cargo.lock index f2fa1f931..77a9ab858 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7509,19 +7509,34 @@ dependencies = [ name = "testsuite" version = "0.0.0" dependencies = [ + "agent-tunnel", + "agent-tunnel-proto", "anyhow", "assert_cmd", "base64 0.23.1", + "camino", + "devolutions-gateway", + "devolutions-gateway-task", "dynosaur", "escargot", "expect-test", "fastrand", + "futures-util", + "ipnetwork", "libsql", "mcp-proxy", "network-scanner", "network-scanner-proto", + "nonempty", + "picky", "proxy-socks", + "quinn", + "rcgen", + "reqwest", "rstest", + "rustls 0.23.43", + "rustls-pemfile 2.2.0", + "rustls-pki-types", "serde", "serde_json", "sysevent", @@ -7534,6 +7549,7 @@ dependencies = [ "tokio-tungstenite", "tokio-util", "typed-builder", + "uuid", ] [[package]] diff --git a/testsuite/Cargo.toml b/testsuite/Cargo.toml index 895e12531..7961358b6 100644 --- a/testsuite/Cargo.toml +++ b/testsuite/Cargo.toml @@ -31,18 +31,34 @@ typed-builder = "0.21" tokio-tungstenite = { version = "0.29", features = ["rustls-tls-native-roots"] } [dev-dependencies] +agent-tunnel = { path = "../crates/agent-tunnel", features = ["test-utils"] } +agent-tunnel-proto = { path = "../crates/agent-tunnel-proto", features = ["serde"] } base64 = "0.23" -proxy-socks = { path = "../crates/proxy-socks" } +camino = "1" +devolutions-gateway-task = { path = "../crates/devolutions-gateway-task" } +devolutions-gateway = { path = "../devolutions-gateway" } +futures-util = "0.3" +ipnetwork = "0.20" libsql = { version = "0.9", default-features = false, features = ["core"] } mcp-proxy.path = "../crates/mcp-proxy" network-scanner = { path = "../crates/network-scanner", features = ["test-utils"] } network-scanner-proto = { path = "../crates/network-scanner-proto" } +nonempty = "0.12" +picky = { version = "7.0.0-rc.25", default-features = false, features = ["jose"] } +proxy-socks = { path = "../crates/proxy-socks" } +quinn = "0.11" +rcgen = { version = "0.13", features = ["pem", "x509-parser"] } +reqwest = { version = "0.12", default-features = false, features = ["json"] } rstest = "0.25" +rustls = { version = "0.23", default-features = false, features = ["ring", "logging", "std", "tls12"] } +rustls-pemfile = "2" +rustls-pki-types = "1" serde_json = "1" sysevent.path = "../crates/sysevent" tempfile = "3" test-utils.path = "../crates/test-utils" tokio-rustls = { version = "0.26", features = ["ring"] } +uuid = { version = "1", features = ["v4"] } [target.'cfg(unix)'.dev-dependencies] sysevent-syslog.path = "../crates/sysevent-syslog" diff --git a/testsuite/src/cli.rs b/testsuite/src/cli.rs index cfd818347..4279c32dc 100644 --- a/testsuite/src/cli.rs +++ b/testsuite/src/cli.rs @@ -83,6 +83,12 @@ pub fn agent_assert_cmd() -> assert_cmd::Command { cmd } +pub fn agent_tokio_cmd() -> tokio::process::Command { + let mut cmd = tokio::process::Command::new(&*AGENT_BIN_PATH); + cmd.env("RUST_BACKTRACE", "0"); + cmd +} + pub fn assert_stderr_eq(output: &assert_cmd::assert::Assert, expected: expect_test::Expect) { let stderr = std::str::from_utf8(&output.get_output().stderr).unwrap(); expected.assert_eq(stderr); diff --git a/testsuite/src/dgw_config.rs b/testsuite/src/dgw_config.rs index f5fc7a441..eb8baa0de 100644 --- a/testsuite/src/dgw_config.rs +++ b/testsuite/src/dgw_config.rs @@ -34,6 +34,12 @@ pub struct AgentTunnelConfig { #[derive(TypedBuilder)] pub struct DgwConfig { + #[builder(default, setter(into))] + hostname: Option, + #[builder(default = "127.0.0.1".to_owned(), setter(into))] + listener_host: String, + #[builder(default, setter(into))] + provisioner_public_key_data: Option, #[builder(default, setter(into))] tcp_port: Option, #[builder(default, setter(into))] @@ -64,6 +70,15 @@ fn find_unused_port() -> u16 { .port() } +fn find_unused_udp_port() -> u16 { + std::net::UdpSocket::bind((std::net::Ipv6Addr::UNSPECIFIED, 0)) + .or_else(|_| std::net::UdpSocket::bind((std::net::Ipv4Addr::UNSPECIFIED, 0))) + .unwrap() + .local_addr() + .unwrap() + .port() +} + impl DgwConfig { pub fn init(self) -> anyhow::Result { DgwConfigHandle::init(self) @@ -79,6 +94,9 @@ pub struct DgwConfigHandle { impl DgwConfigHandle { pub fn init(config: DgwConfig) -> anyhow::Result { let DgwConfig { + hostname, + listener_host, + provisioner_public_key_data, tcp_port, http_port, disable_token_validation, @@ -106,7 +124,7 @@ impl DgwConfigHandle { }; let agent_tunnel_json = if let Some(at_config) = agent_tunnel { - let listen_port = at_config.listen_port.unwrap_or_else(find_unused_port); + let listen_port = at_config.listen_port.unwrap_or_else(find_unused_udp_port); format!( r#", "AgentTunnel": {{ @@ -119,19 +137,31 @@ impl DgwConfigHandle { String::new() }; + let hostname_json = hostname + .map(|hostname| { + format!( + r#" "Hostname": "{hostname}", +"# + ) + }) + .unwrap_or_default(); + let provisioner_public_key_data = provisioner_public_key_data.unwrap_or_else(|| { + "mMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA4vuqLOkl1pWobt6su1XO9VskgCAwevEGs6kkNjJQBwkGnPKYLmNF1E/af1yCocfVn/OnPf9e4x+lXVyZ6LMDJxFxu+axdgOq3Ld392J1iAEbfvwlyRFnEXFOJNyylqg3bY6LvnWHL/XZczVdMD9xYfq2sO9bg3xjRW4s7r9EEYOFjqVT3VFznH9iWJVtcSEKukmS/3uKoO6lGhacvu0HgjXXdgq0R8zvR4XRJ9Fcnf0f9Ypoc+i6L80NVjrRCeVOH+Ld/2fA9bocpfLarcVqG3RjS+qgOtpyCc0jWVFF4zaGQ7LUDFkEIYILkICeMMn2ll29hmZNzsJzZJ9s6NocgQIDAQAB".to_owned() + }); + let config = format!( r#"{{ - "ProvisionerPublicKeyData": {{ - "Value": "mMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA4vuqLOkl1pWobt6su1XO9VskgCAwevEGs6kkNjJQBwkGnPKYLmNF1E/af1yCocfVn/OnPf9e4x+lXVyZ6LMDJxFxu+axdgOq3Ld392J1iAEbfvwlyRFnEXFOJNyylqg3bY6LvnWHL/XZczVdMD9xYfq2sO9bg3xjRW4s7r9EEYOFjqVT3VFznH9iWJVtcSEKukmS/3uKoO6lGhacvu0HgjXXdgq0R8zvR4XRJ9Fcnf0f9Ypoc+i6L80NVjrRCeVOH+Ld/2fA9bocpfLarcVqG3RjS+qgOtpyCc0jWVFF4zaGQ7LUDFkEIYILkICeMMn2ll29hmZNzsJzZJ9s6NocgQIDAQAB" +{hostname_json} "ProvisionerPublicKeyData": {{ + "Value": "{provisioner_public_key_data}" }}, "Listeners": [ {{ - "InternalUrl": "tcp://127.0.0.1:{tcp_port}", - "ExternalUrl": "tcp://127.0.0.1:{tcp_port}" + "InternalUrl": "tcp://{listener_host}:{tcp_port}", + "ExternalUrl": "tcp://{listener_host}:{tcp_port}" }}, {{ - "InternalUrl": "http://127.0.0.1:{http_port}", - "ExternalUrl": "http://127.0.0.1:{http_port}" + "InternalUrl": "http://{listener_host}:{http_port}", + "ExternalUrl": "http://{listener_host}:{http_port}" }} ], "VerbosityProfile": "{verbosity_profile}", diff --git a/testsuite/tests/agent_tunnel/cert.rs b/testsuite/tests/agent_tunnel/cert.rs new file mode 100644 index 000000000..578297648 --- /dev/null +++ b/testsuite/tests/agent_tunnel/cert.rs @@ -0,0 +1,58 @@ +use agent_tunnel::cert::{CaManager, extract_agent_id_from_pem}; +use camino::Utf8PathBuf; +use tempfile::TempDir; +use uuid::Uuid; + +use super::common::generate_csr_with_cn; + +fn fresh_ca() -> (TempDir, std::sync::Arc) { + let temp_dir = tempfile::tempdir().expect("create temporary directory"); + let data_dir = Utf8PathBuf::from_path_buf(temp_dir.path().to_path_buf()).expect("use utf-8 temporary path"); + let manager = CaManager::load_or_generate(&data_dir).expect("generate test ca"); + (temp_dir, manager) +} + +#[test] +fn sign_agent_csr_ignores_csr_subject_uses_passed_identity() { + let (_temp_dir, ca_manager) = fresh_ca(); + + let real_agent_id = Uuid::new_v4(); + let (_evil_key, evil_csr_pem) = generate_csr_with_cn("evil-impersonator"); + + let signed = ca_manager + .sign_agent_csr(real_agent_id, "legit-name", &evil_csr_pem, None) + .expect("sign agent csr"); + + let recovered = extract_agent_id_from_pem(&signed.client_cert_pem).expect("issued certificate has urn:uuid san"); + assert_eq!( + recovered, real_agent_id, + "issued cert must encode the agent_id passed by the caller, not the CSR subject" + ); +} + +#[test] +fn extract_agent_id_from_pem_round_trips() { + let (_temp_dir, ca_manager) = fresh_ca(); + + let known_id = Uuid::new_v4(); + let (_key, csr_pem) = generate_csr_with_cn("round-trip-agent"); + + let signed = ca_manager + .sign_agent_csr(known_id, "round-trip-agent", &csr_pem, None) + .expect("sign agent csr"); + + let recovered = extract_agent_id_from_pem(&signed.client_cert_pem).expect("urn:uuid san present"); + assert_eq!(recovered, known_id); +} + +#[test] +fn extract_agent_id_from_pem_rejects_cert_without_san() { + let (_temp_dir, ca_manager) = fresh_ca(); + let error = extract_agent_id_from_pem(ca_manager.ca_cert_pem()).expect_err("ca certificate has no urn:uuid san"); + + let msg = format!("{error:#}"); + assert!( + msg.contains("urn:uuid"), + "error should reference the missing urn:uuid SAN, got: {msg}" + ); +} diff --git a/testsuite/tests/agent_tunnel/common.rs b/testsuite/tests/agent_tunnel/common.rs new file mode 100644 index 000000000..627384d0c --- /dev/null +++ b/testsuite/tests/agent_tunnel/common.rs @@ -0,0 +1,195 @@ +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use agent_tunnel::AgentTunnelHandle; +use agent_tunnel::cert::CaManager; +use agent_tunnel::listener::AgentTunnelListener; +use agent_tunnel::registry::AgentRegistry; +use agent_tunnel_proto::{ControlMessage, ControlStream, DomainAdvertisement, SessionStream}; +use camino::Utf8PathBuf; +use devolutions_gateway_task::ShutdownHandle; +use ipnetwork::Ipv4Network; +use tempfile::TempDir; +use tokio::net::TcpListener; +use tokio::task::JoinHandle; +use uuid::Uuid; + +pub(super) async fn start_echo_server() -> (SocketAddr, JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind echo server"); + let addr = listener.local_addr().expect("read echo server address"); + let task = tokio::spawn(async move { + let (stream, _) = listener.accept().await.expect("accept echo connection"); + let (mut read, mut write) = stream.into_split(); + tokio::io::copy(&mut read, &mut write).await.expect("echo data"); + }); + + (addr, task) +} + +pub(super) fn generate_csr_with_cn(cn: &str) -> (rcgen::KeyPair, String) { + let key_pair = rcgen::KeyPair::generate_for(&rcgen::PKCS_ECDSA_P256_SHA256).expect("generate test key pair"); + let mut params = rcgen::CertificateParams::default(); + params.distinguished_name.push(rcgen::DnType::CommonName, cn); + let csr = params.serialize_request(&key_pair).expect("serialize test csr"); + let csr_pem = csr.pem().expect("encode test csr"); + (key_pair, csr_pem) +} + +async fn connect_quinn_client( + ca_cert_pem: &str, + client_cert_pem: &str, + client_key_pem: &str, + server_addr: SocketAddr, +) -> quinn::Connection { + use rustls_pemfile::{certs, private_key}; + + let _ = rustls::crypto::ring::default_provider().install_default(); + + let client_certs: Vec> = + certs(&mut std::io::BufReader::new(client_cert_pem.as_bytes())) + .collect::, _>>() + .expect("parse client certificates"); + let client_key = private_key(&mut std::io::BufReader::new(client_key_pem.as_bytes())) + .expect("parse client key") + .expect("find client key"); + + let mut roots = rustls::RootCertStore::empty(); + let ca_certs: Vec> = + certs(&mut std::io::BufReader::new(ca_cert_pem.as_bytes())) + .collect::, _>>() + .expect("parse ca certificates"); + for cert in ca_certs { + roots.add(cert).expect("add ca certificate"); + } + + let verifier = rustls::client::WebPkiServerVerifier::builder(Arc::new(roots)) + .build() + .expect("build server verifier"); + let mut client_crypto = rustls::ClientConfig::builder() + .dangerous() + .with_custom_certificate_verifier(verifier) + .with_client_auth_cert(client_certs, client_key) + .expect("configure client authentication"); + client_crypto.alpn_protocols = vec![agent_tunnel_proto::ALPN_PROTOCOL.to_vec()]; + + let client_config = quinn::ClientConfig::new(Arc::new( + quinn::crypto::rustls::QuicClientConfig::try_from(client_crypto).expect("configure quic client"), + )); + let mut endpoint = quinn::Endpoint::client("0.0.0.0:0".parse().expect("parse client bind address")) + .expect("create client endpoint"); + endpoint.set_default_client_config(client_config); + + endpoint + .connect(server_addr, "localhost") + .expect("start quic connection") + .await + .expect("complete quic handshake") +} + +pub(super) struct TestListener { + pub handle: AgentTunnelHandle, + pub server_addr: SocketAddr, + shutdown: ShutdownHandle, + task: JoinHandle>, + _temp_dir: TempDir, +} + +impl TestListener { + pub(super) async fn connect_agent(&self, agent_name: &str) -> (Uuid, quinn::Connection) { + let agent_id = Uuid::new_v4(); + let (key_pair, csr_pem) = generate_csr_with_cn(agent_name); + let signed = self + .handle + .ca_manager() + .sign_agent_csr(agent_id, agent_name, &csr_pem, Some("localhost")) + .expect("sign agent csr"); + let connection = connect_quinn_client( + &signed.ca_cert_pem, + &signed.client_cert_pem, + &key_pair.serialize_pem(), + self.server_addr, + ) + .await; + + (agent_id, connection) + } + + pub(super) async fn shutdown(self) { + self.shutdown.signal(); + tokio::time::timeout(Duration::from_secs(2), self.task) + .await + .expect("listener shutdown timed out") + .expect("listener task panicked") + .expect("listener shutdown failed"); + } +} + +pub(super) async fn bind_test_listener() -> TestListener { + let temp_dir = tempfile::tempdir().expect("create temporary directory"); + let data_dir = Utf8PathBuf::from_path_buf(temp_dir.path().to_path_buf()).expect("use utf-8 temporary path"); + let ca_manager = CaManager::load_or_generate(&data_dir).expect("generate test ca"); + let listen_addr: SocketAddr = "127.0.0.1:0".parse().expect("parse listener address"); + let (listener, handle) = AgentTunnelListener::bind(listen_addr, ca_manager, "localhost") + .await + .expect("bind quic listener"); + let server_addr = listener.local_addr(); + let (shutdown, shutdown_signal) = ShutdownHandle::new(); + let task = tokio::spawn(async move { + use devolutions_gateway_task::Task; + listener.run(shutdown_signal).await + }); + + TestListener { + handle, + server_addr, + shutdown, + task, + _temp_dir: temp_dir, + } +} + +pub(super) async fn wait_for_route_advertised(registry: &AgentRegistry, agent_id: Uuid, min_epoch: u64) { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Some(peer) = registry.get(&agent_id).await + && peer.route_state().epoch >= min_epoch + { + return; + } + assert!( + Instant::now() < deadline, + "agent {agent_id} did not advertise route at epoch >= {min_epoch} within 5s" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } +} + +pub(super) async fn advertise_routes( + connection: &quinn::Connection, + registry: &AgentRegistry, + agent_id: Uuid, + epoch: u64, + subnets: Vec, + domains: Vec, +) -> ControlStream { + let mut ctrl: ControlStream<_, _> = connection.open_bi().await.expect("open control stream").into(); + ctrl.send(&ControlMessage::route_advertise(epoch, subnets, domains)) + .await + .expect("send route advertisement"); + wait_for_route_advertised(registry, agent_id, epoch).await; + ctrl +} + +pub(super) async fn accept_session_request( + connection: &quinn::Connection, + session_id: Uuid, + expected_target: &str, +) -> SessionStream { + let (send, recv) = connection.accept_bi().await.expect("accept session stream"); + let mut session: SessionStream<_, _> = (send, recv).into(); + let request = session.recv_request().await.expect("receive connect request"); + assert_eq!(request.session_id(), session_id); + assert_eq!(request.target(), expected_target); + session +} diff --git a/testsuite/tests/agent_tunnel/integration.rs b/testsuite/tests/agent_tunnel/integration.rs new file mode 100644 index 000000000..1a6dc342f --- /dev/null +++ b/testsuite/tests/agent_tunnel/integration.rs @@ -0,0 +1,358 @@ +use std::time::Duration; + +use agent_tunnel::AgentTunnelHandle; +use agent_tunnel::cert::extract_agent_id_from_pem; +use agent_tunnel::registry::AgentRegistry; +use agent_tunnel_proto::{ + CertRenewalResult, ConnectResponse, ControlMessage, ControlStream, DomainAdvertisement, DomainName, +}; +use devolutions_gateway::target_addr::TargetAddr; +use devolutions_gateway::upstream::{ConnectedUpstream, UpstreamLeg, connect_upstream}; +use nonempty::NonEmpty; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use uuid::Uuid; + +use super::common::{ + accept_session_request, advertise_routes, bind_test_listener, generate_csr_with_cn, start_echo_server, + wait_for_route_advertised, +}; + +fn target(host: &str, port: u16) -> TargetAddr { + TargetAddr::from_components("tcp", host, port).expect("build target address") +} + +async fn advertise_domain( + connection: &quinn::Connection, + registry: &AgentRegistry, + agent_id: Uuid, + epoch: u64, + domain: &str, +) -> ControlStream { + advertise_routes( + connection, + registry, + agent_id, + epoch, + vec![], + vec![DomainAdvertisement { + domain: DomainName::new(domain), + auto_detected: false, + }], + ) + .await +} + +async fn set_route_order(registry: &AgentRegistry, older: Uuid, newer: Uuid) { + registry + .get(&older) + .await + .expect("find older agent") + .set_received_at_for_test(std::time::UNIX_EPOCH + Duration::from_secs(1)); + registry + .get(&newer) + .await + .expect("find newer agent") + .set_received_at_for_test(std::time::UNIX_EPOCH + Duration::from_secs(2)); +} + +async fn connect( + handle: AgentTunnelHandle, + target: TargetAddr, + explicit_agent_id: Option, + session_id: Uuid, +) -> anyhow::Result { + connect_upstream(&NonEmpty::new(target), explicit_agent_id, session_id, Some(&handle)).await +} + +async fn assert_round_trip(mut upstream: UpstreamLeg, payload: &[u8]) { + upstream.write_all(payload).await.expect("write upstream payload"); + let mut response = vec![0; payload.len()]; + upstream + .read_exact(&mut response) + .await + .expect("read upstream response"); + assert_eq!(response, payload); +} + +#[tokio::test] +async fn gateway_connect_upstream_routes_wildcard_domain_without_subnets() { + let listener = bind_test_listener().await; + let (agent_id, connection) = listener.connect_agent("test-agent").await; + let (echo_addr, echo_task) = start_echo_server().await; + let _ctrl = advertise_domain(&connection, listener.handle.registry(), agent_id, 1, "*.echo.test").await; + + let session_id = Uuid::new_v4(); + let expected_target = format!("service.echo.test:{}", echo_addr.port()); + let handle = listener.handle.clone(); + let connect_task = tokio::spawn(connect( + handle, + target("service.echo.test", echo_addr.port()), + None, + session_id, + )); + + let mut session = accept_session_request(&connection, session_id, &expected_target).await; + + let mut tcp_stream = TcpStream::connect(echo_addr).await.expect("connect to echo server"); + session + .send_response(&ConnectResponse::success()) + .await + .expect("send connection success"); + + let connected = tokio::time::timeout(Duration::from_secs(5), connect_task) + .await + .expect("upstream connection timed out") + .expect("upstream task panicked") + .expect("connect through agent"); + assert!(matches!(connected.leg, UpstreamLeg::Tunnel(_))); + + let payload = b"agent tunnel payload"; + let (mut tunnel_read, mut tunnel_write) = tokio::io::split(connected.leg); + tunnel_write.write_all(payload).await.expect("write tunnel payload"); + + let (mut session_send, mut session_recv) = session.into_inner(); + let mut relay = vec![0; payload.len()]; + session_recv.read_exact(&mut relay).await.expect("read agent payload"); + tcp_stream.write_all(&relay).await.expect("write echo payload"); + tcp_stream.read_exact(&mut relay).await.expect("read echo payload"); + session_send.write_all(&relay).await.expect("write agent response"); + + let mut response = vec![0; payload.len()]; + tunnel_read + .read_exact(&mut response) + .await + .expect("read tunnel response"); + assert_eq!(response, payload); + + connection.close(0u32.into(), b"test done"); + echo_task.abort(); + listener.shutdown().await; +} + +#[tokio::test] +async fn gateway_connect_upstream_falls_back_to_direct_tcp_without_a_route() { + let listener = bind_test_listener().await; + let (agent_id, connection) = listener.connect_agent("unmatched-agent").await; + let _ctrl = advertise_domain(&connection, listener.handle.registry(), agent_id, 1, "unused.example").await; + let (echo_addr, echo_task) = start_echo_server().await; + + let connected = connect( + listener.handle.clone(), + target("127.0.0.1", echo_addr.port()), + None, + Uuid::new_v4(), + ) + .await + .expect("connect directly"); + assert!(matches!(connected.leg, UpstreamLeg::Tcp(_))); + assert_round_trip(connected.leg, b"direct payload").await; + assert!( + tokio::time::timeout(Duration::from_millis(100), connection.accept_bi()) + .await + .is_err() + ); + + connection.close(0u32.into(), b"test done"); + echo_task.abort(); + listener.shutdown().await; +} + +#[tokio::test] +async fn gateway_connect_upstream_uses_explicit_agent_without_a_matching_route() { + let listener = bind_test_listener().await; + let (agent_id, connection) = listener.connect_agent("explicit-agent").await; + let _ctrl = advertise_domain(&connection, listener.handle.registry(), agent_id, 1, "unused.example").await; + let (echo_addr, echo_task) = start_echo_server().await; + let session_id = Uuid::new_v4(); + let target_addr = format!("127.0.0.1:{}", echo_addr.port()); + let connect_task = tokio::spawn(connect( + listener.handle.clone(), + target("127.0.0.1", echo_addr.port()), + Some(agent_id), + session_id, + )); + + let mut session = accept_session_request(&connection, session_id, &target_addr).await; + let tcp_stream = TcpStream::connect(echo_addr).await.expect("connect to echo server"); + session + .send_response(&ConnectResponse::success()) + .await + .expect("send connection success"); + let connected = connect_task + .await + .expect("upstream task panicked") + .expect("connect through explicit agent"); + let relay_task = tokio::spawn(async move { + let (mut send, mut recv) = session.into_inner(); + let (mut read, mut write) = tcp_stream.into_split(); + tokio::try_join!( + tokio::io::copy(&mut recv, &mut write), + tokio::io::copy(&mut read, &mut send) + ) + }); + assert_round_trip(connected.leg, b"explicit payload").await; + + relay_task.abort(); + connection.close(0u32.into(), b"test done"); + echo_task.abort(); + listener.shutdown().await; +} + +#[tokio::test] +async fn gateway_connect_upstream_tries_the_next_matching_agent() { + let listener = bind_test_listener().await; + let (fallback_id, fallback_connection) = listener.connect_agent("fallback-agent").await; + let _fallback_ctrl = advertise_domain( + &fallback_connection, + listener.handle.registry(), + fallback_id, + 1, + "service.example", + ) + .await; + let (first_id, first_connection) = listener.connect_agent("first-agent").await; + let _first_ctrl = advertise_domain( + &first_connection, + listener.handle.registry(), + first_id, + 1, + "service.example", + ) + .await; + set_route_order(listener.handle.registry(), fallback_id, first_id).await; + let session_id = Uuid::new_v4(); + let target_addr = "service.example:443"; + let connect_task = tokio::spawn(connect( + listener.handle.clone(), + target("service.example", 443), + None, + session_id, + )); + + let mut first_session = accept_session_request(&first_connection, session_id, target_addr).await; + first_session + .send_response(&ConnectResponse::error("connection refused")) + .await + .expect("send connection error"); + let mut fallback_session = accept_session_request(&fallback_connection, session_id, target_addr).await; + fallback_session + .send_response(&ConnectResponse::success()) + .await + .expect("send connection success"); + let connected = connect_task + .await + .expect("upstream task panicked") + .expect("connect through fallback agent"); + assert!(matches!(connected.leg, UpstreamLeg::Tunnel(_))); + + first_connection.close(0u32.into(), b"test done"); + fallback_connection.close(0u32.into(), b"test done"); + listener.shutdown().await; +} + +#[tokio::test] +async fn gateway_connect_upstream_does_not_bypass_failed_agent_routes() { + let listener = bind_test_listener().await; + let (first_id, first_connection) = listener.connect_agent("first-agent").await; + let _first_ctrl = advertise_routes( + &first_connection, + listener.handle.registry(), + first_id, + 1, + vec!["127.0.0.0/8".parse().expect("parse test subnet")], + vec![], + ) + .await; + let (second_id, second_connection) = listener.connect_agent("second-agent").await; + let _second_ctrl = advertise_routes( + &second_connection, + listener.handle.registry(), + second_id, + 1, + vec!["127.0.0.0/8".parse().expect("parse test subnet")], + vec![], + ) + .await; + set_route_order(listener.handle.registry(), first_id, second_id).await; + let direct_listener = TcpListener::bind("127.0.0.1:0").await.expect("bind direct target"); + let target_port = direct_listener.local_addr().expect("read direct target address").port(); + let session_id = Uuid::new_v4(); + let target_addr = format!("127.0.0.1:{target_port}"); + let connect_task = tokio::spawn(connect( + listener.handle.clone(), + target("127.0.0.1", target_port), + None, + session_id, + )); + + let mut second_session = accept_session_request(&second_connection, session_id, &target_addr).await; + second_session + .send_response(&ConnectResponse::error("connection refused")) + .await + .expect("send connection error"); + let mut first_session = accept_session_request(&first_connection, session_id, &target_addr).await; + first_session + .send_response(&ConnectResponse::error("connection refused")) + .await + .expect("send connection error"); + let error = match connect_task.await.expect("upstream task panicked") { + Ok(_) => panic!("all routed connections should fail"), + Err(error) => error, + }; + assert!(format!("{error:#}").contains("connection refused")); + assert!( + tokio::time::timeout(Duration::from_millis(100), direct_listener.accept()) + .await + .is_err() + ); + + first_connection.close(0u32.into(), b"test done"); + second_connection.close(0u32.into(), b"test done"); + listener.shutdown().await; +} + +#[tokio::test] +async fn gateway_listener_renews_authenticated_agent_identity() { + let listener = bind_test_listener().await; + let (agent_id, connection) = listener.connect_agent("renewal-agent").await; + let expected_ca = listener.handle.ca_manager().ca_cert_pem().to_owned(); + let mut ctrl: ControlStream<_, _> = connection.open_bi().await.expect("open control stream").into(); + + ctrl.send(&ControlMessage::route_advertise(1, vec![], vec![])) + .await + .expect("send route advertisement"); + wait_for_route_advertised(listener.handle.registry(), agent_id, 1).await; + + let (_, csr_pem) = generate_csr_with_cn("evil-impersonator"); + ctrl.send(&ControlMessage::cert_renewal_request(csr_pem)) + .await + .expect("send renewal request"); + + let response = tokio::time::timeout(Duration::from_secs(5), ctrl.recv()) + .await + .expect("renewal response timed out") + .expect("receive renewal response"); + let renewed_pem = match response { + ControlMessage::CertRenewalResponse { + result: + CertRenewalResult::Success { + client_cert_pem, + gateway_ca_cert_pem, + }, + .. + } => { + assert_eq!(gateway_ca_cert_pem, expected_ca); + client_cert_pem + } + other => panic!("expected successful renewal, got {other:?}"), + }; + + assert_eq!( + extract_agent_id_from_pem(&renewed_pem).expect("read renewed agent identity"), + agent_id + ); + + connection.close(0u32.into(), b"test done"); + listener.shutdown().await; +} diff --git a/testsuite/tests/agent_tunnel/kdc.rs b/testsuite/tests/agent_tunnel/kdc.rs new file mode 100644 index 000000000..1812535a2 --- /dev/null +++ b/testsuite/tests/agent_tunnel/kdc.rs @@ -0,0 +1,140 @@ +use std::sync::Arc; +use std::time::Duration; + +use agent_tunnel_proto::{ConnectResponse, ControlStream}; +use devolutions_gateway::kdc_connector::KdcConnector; +use devolutions_gateway::target_addr::TargetAddr; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; +use uuid::Uuid; + +use super::common::{accept_session_request, advertise_routes, bind_test_listener}; + +const KDC_REQUEST: &[u8] = b"\0\0\0\x04test"; +const KDC_REPLY: &[u8] = b"\0\0\0\x04data"; + +fn target(scheme: &str, port: u16) -> TargetAddr { + TargetAddr::from_components(scheme, "127.0.0.1", port).expect("build kdc target") +} + +async fn advertise_loopback( + connection: &quinn::Connection, + listener: &super::common::TestListener, + agent_id: Uuid, +) -> ControlStream { + advertise_routes( + connection, + listener.handle.registry(), + agent_id, + 1, + vec!["127.0.0.0/8".parse().expect("parse test subnet")], + vec![], + ) + .await +} + +#[tokio::test] +async fn kdc_connector_relays_tcp_through_a_matching_agent() { + let listener = bind_test_listener().await; + let (agent_id, connection) = listener.connect_agent("kdc-agent").await; + let _ctrl = advertise_loopback(&connection, &listener, agent_id).await; + let session_id = Uuid::new_v4(); + let kdc_target = target("tcp", 88); + let connector = KdcConnector::new(session_id, None, Some(Arc::new(listener.handle.clone()))); + let send_task = tokio::spawn(async move { connector.send(&kdc_target, KDC_REQUEST).await }); + + let mut session = accept_session_request(&connection, session_id, "127.0.0.1:88").await; + session + .send_response(&ConnectResponse::success()) + .await + .expect("send connection success"); + let (mut send, mut recv) = session.into_inner(); + let mut request = vec![0; KDC_REQUEST.len()]; + recv.read_exact(&mut request).await.expect("read kdc request"); + assert_eq!(request, KDC_REQUEST); + send.write_all(KDC_REPLY).await.expect("write kdc reply"); + + let reply = match send_task.await.expect("kdc task panicked") { + Ok(reply) => reply, + Err(error) => panic!("relay kdc request through agent: {error}"), + }; + assert_eq!(reply, KDC_REPLY); + + connection.close(0u32.into(), b"test done"); + listener.shutdown().await; +} + +#[tokio::test] +async fn kdc_connector_falls_back_to_direct_tcp_without_a_matching_route() { + let listener = bind_test_listener().await; + let (agent_id, connection) = listener.connect_agent("unmatched-kdc-agent").await; + let _ctrl = advertise_routes(&connection, listener.handle.registry(), agent_id, 1, vec![], vec![]).await; + let kdc = TcpListener::bind("127.0.0.1:0").await.expect("bind fake kdc"); + let port = kdc.local_addr().expect("read fake kdc address").port(); + let kdc_task = tokio::spawn(async move { + let (mut stream, _) = kdc.accept().await.expect("accept kdc connection"); + let mut request = vec![0; KDC_REQUEST.len()]; + stream.read_exact(&mut request).await.expect("read kdc request"); + assert_eq!(request, KDC_REQUEST); + stream.write_all(KDC_REPLY).await.expect("write kdc reply"); + }); + let connector = KdcConnector::new(Uuid::new_v4(), None, Some(Arc::new(listener.handle.clone()))); + + let reply = match connector.send(&target("tcp", port), KDC_REQUEST).await { + Ok(reply) => reply, + Err(error) => panic!("send directly to kdc: {error}"), + }; + assert_eq!(reply, KDC_REPLY); + assert!( + tokio::time::timeout(Duration::from_millis(100), connection.accept_bi()) + .await + .is_err() + ); + + kdc_task.await.expect("kdc task panicked"); + connection.close(0u32.into(), b"test done"); + listener.shutdown().await; +} + +#[tokio::test] +async fn kdc_connector_rejects_udp_selected_for_an_agent_route() { + let listener = bind_test_listener().await; + let (agent_id, connection) = listener.connect_agent("udp-kdc-agent").await; + let _ctrl = advertise_loopback(&connection, &listener, agent_id).await; + let session_id = Uuid::new_v4(); + let connector = KdcConnector::new(session_id, None, Some(Arc::new(listener.handle.clone()))); + let send_task = tokio::spawn(async move { connector.send(&target("udp", 88), KDC_REQUEST).await }); + + let mut session = accept_session_request(&connection, session_id, "127.0.0.1:88").await; + session + .send_response(&ConnectResponse::success()) + .await + .expect("send connection success"); + let error = send_task + .await + .expect("kdc task panicked") + .expect_err("udp agent route should be rejected"); + assert!(format!("{error}").contains("does not yet support UDP")); + + connection.close(0u32.into(), b"test done"); + listener.shutdown().await; +} + +#[tokio::test] +async fn kdc_connector_rejects_an_explicit_missing_agent() { + let listener = bind_test_listener().await; + let missing_agent_id = Uuid::new_v4(); + let connector = KdcConnector::new( + Uuid::new_v4(), + Some(missing_agent_id), + Some(Arc::new(listener.handle.clone())), + ); + + let error = connector + .send(&target("tcp", 88), KDC_REQUEST) + .await + .expect_err("missing explicit agent should be rejected"); + assert!(format!("{error}").contains(&format!("agent {missing_agent_id} specified in token not found"))); + + listener.shutdown().await; +} diff --git a/testsuite/tests/agent_tunnel/mod.rs b/testsuite/tests/agent_tunnel/mod.rs new file mode 100644 index 000000000..be84468c4 --- /dev/null +++ b/testsuite/tests/agent_tunnel/mod.rs @@ -0,0 +1,5 @@ +mod cert; +mod common; +mod integration; +mod kdc; +mod routing; diff --git a/testsuite/tests/agent_tunnel/routing.rs b/testsuite/tests/agent_tunnel/routing.rs new file mode 100644 index 000000000..635d23508 --- /dev/null +++ b/testsuite/tests/agent_tunnel/routing.rs @@ -0,0 +1,272 @@ +use std::net::{IpAddr, Ipv4Addr}; +use std::sync::Arc; + +use agent_tunnel::registry::{AgentPeer, AgentRegistry}; +use agent_tunnel::routing::{RouteTarget, RoutingDecision, resolve_route, route_and_connect, try_route}; +use agent_tunnel_proto::{DomainAdvertisement, DomainName}; +use ipnetwork::Ipv4Network; +use uuid::Uuid; + +use super::common::bind_test_listener; + +fn ip(s: &str) -> RouteTarget { + RouteTarget::Ip(IpAddr::V4(s.parse::().expect("valid test ipv4"))) +} + +fn host(s: &str) -> RouteTarget { + RouteTarget::hostname(s) +} + +fn make_peer(name: &str) -> Arc { + Arc::new(AgentPeer::new( + Uuid::new_v4(), + name.to_owned(), + "sha256:test".to_owned(), + )) +} + +fn domain(name: &str) -> DomainAdvertisement { + DomainAdvertisement { + domain: DomainName::new(name), + auto_detected: false, + } +} + +#[tokio::test] +async fn route_explicit_agent_id() { + let registry = AgentRegistry::new(); + let peer = make_peer("agent-a"); + let agent_id = peer.agent_id; + registry.register(Arc::clone(&peer)).await; + + match resolve_route(®istry, Some(agent_id), &host("anything")).await { + RoutingDecision::ViaAgent(agents) => { + assert_eq!(agents.len(), 1); + assert_eq!(agents[0].agent_id, agent_id); + } + other => panic!("expected agent route, got {other:?}"), + } +} + +#[tokio::test] +async fn route_explicit_agent_id_not_found() { + let registry = AgentRegistry::new(); + let bogus_id = Uuid::new_v4(); + + match resolve_route(®istry, Some(bogus_id), &host("anything")).await { + RoutingDecision::ExplicitAgentNotFound(id) => { + assert_eq!(id, bogus_id); + } + other => panic!("expected missing explicit agent, got {other:?}"), + } +} + +#[tokio::test] +async fn route_ip_target_via_subnet() { + let registry = AgentRegistry::new(); + let peer = make_peer("agent-a"); + let agent_id = peer.agent_id; + let subnet: Ipv4Network = "10.1.0.0/16".parse().expect("valid test subnet"); + peer.update_routes(1, vec![subnet], vec![]); + registry.register(peer).await; + + match resolve_route(®istry, None, &ip("10.1.5.50")).await { + RoutingDecision::ViaAgent(agents) => { + assert_eq!(agents[0].agent_id, agent_id); + } + other => panic!("expected agent route, got {other:?}"), + } +} + +#[tokio::test] +async fn route_hostname_via_domain() { + let registry = AgentRegistry::new(); + let peer = make_peer("agent-a"); + let agent_id = peer.agent_id; + let subnet: Ipv4Network = "10.1.0.0/16".parse().expect("valid test subnet"); + peer.update_routes(1, vec![subnet], vec![domain("*.contoso.local")]); + registry.register(peer).await; + + match resolve_route(®istry, None, &host("dc01.contoso.local")).await { + RoutingDecision::ViaAgent(agents) => { + assert_eq!(agents[0].agent_id, agent_id); + } + other => panic!("expected agent route, got {other:?}"), + } +} + +#[tokio::test] +async fn route_no_match_returns_direct() { + let registry = AgentRegistry::new(); + let peer = make_peer("agent-a"); + let subnet: Ipv4Network = "10.1.0.0/16".parse().expect("valid test subnet"); + peer.update_routes(1, vec![subnet], vec![domain("*.contoso.local")]); + registry.register(peer).await; + + assert!(matches!( + resolve_route(®istry, None, &host("external.example.com")).await, + RoutingDecision::Direct + )); +} + +#[tokio::test] +async fn route_ip_no_match_returns_direct() { + let registry = AgentRegistry::new(); + let peer = make_peer("agent-a"); + let subnet: Ipv4Network = "10.1.0.0/16".parse().expect("valid test subnet"); + peer.update_routes(1, vec![subnet], vec![]); + registry.register(peer).await; + + assert!(matches!( + resolve_route(®istry, None, &ip("192.168.1.1")).await, + RoutingDecision::Direct + )); +} + +#[tokio::test] +async fn route_skips_offline_agents() { + let registry = AgentRegistry::new(); + let peer = make_peer("offline-agent"); + let subnet: Ipv4Network = "10.1.0.0/16".parse().expect("valid test subnet"); + peer.update_routes(1, vec![subnet], vec![domain("*.contoso.local")]); + peer.set_last_seen_for_test(0); + registry.register(peer).await; + + assert!(matches!( + resolve_route(®istry, None, &host("dc01.contoso.local")).await, + RoutingDecision::Direct + )); +} + +#[tokio::test] +async fn route_domain_match_returns_multiple_agents_ordered() { + let registry = AgentRegistry::new(); + + let peer_a = make_peer("agent-a"); + let subnet_a: Ipv4Network = "10.1.0.0/16".parse().expect("valid test subnet"); + peer_a.update_routes(1, vec![subnet_a], vec![domain("*.contoso.local")]); + peer_a.set_received_at_for_test(std::time::UNIX_EPOCH + std::time::Duration::from_secs(1)); + registry.register(Arc::clone(&peer_a)).await; + + let peer_b = make_peer("agent-b"); + let id_b = peer_b.agent_id; + let subnet_b: Ipv4Network = "10.2.0.0/16".parse().expect("valid test subnet"); + peer_b.update_routes(1, vec![subnet_b], vec![domain("*.contoso.local")]); + peer_b.set_received_at_for_test(std::time::UNIX_EPOCH + std::time::Duration::from_secs(2)); + registry.register(Arc::clone(&peer_b)).await; + + match resolve_route(®istry, None, &host("dc01.contoso.local")).await { + RoutingDecision::ViaAgent(agents) => { + assert_eq!(agents.len(), 2); + assert_eq!(agents[0].agent_id, id_b, "most recent first"); + } + other => panic!("expected agent route, got {other:?}"), + } +} + +#[tokio::test] +async fn try_route_rejects_explicit_agent_when_handle_missing() { + let result = try_route( + None, + Some(Uuid::new_v4()), + &host("host.example.com"), + Uuid::new_v4(), + "host.example.com:443", + ) + .await; + + assert!( + result.is_err(), + "expected an error for an explicit agent without a handle" + ); +} + +#[tokio::test] +async fn try_route_without_explicit_agent_falls_through_when_handle_missing() { + let result = try_route( + None, + None, + &host("host.example.com"), + Uuid::new_v4(), + "host.example.com:443", + ) + .await; + + match result { + Ok(None) => {} + Ok(Some(_)) => panic!("expected direct fallback, got a tunnel"), + Err(error) => panic!("expected direct fallback, got {error:#}"), + } +} + +#[tokio::test] +async fn route_and_connect_with_empty_candidates_errors() { + let listener = bind_test_listener().await; + + let err = match route_and_connect(&listener.handle, &[], Uuid::new_v4(), "10.1.1.1:22").await { + Ok(_) => panic!("expected an error for an empty candidate list"), + Err(e) => e, + }; + + let msg = format!("{err:#}"); + assert!( + msg.contains("empty candidates"), + "error should mention empty candidates, got: {msg}" + ); + + listener.shutdown().await; +} + +#[tokio::test] +async fn try_route_falls_through_when_no_agent_matches() { + let listener = bind_test_listener().await; + + let peer = make_peer("agent-a"); + let subnet: Ipv4Network = "10.0.0.0/8".parse().expect("valid cidr"); + peer.update_routes(1, vec![subnet], vec![domain("contoso.local")]); + listener.handle.registry().register(peer).await; + + let result = try_route( + Some(&listener.handle), + None, + &host("external.example.com"), + Uuid::new_v4(), + "external.example.com:443", + ) + .await; + + match result { + Ok(None) => {} + Ok(Some(_)) => panic!("expected direct fallback, got a tunnel"), + Err(error) => panic!("expected direct fallback, got {error:#}"), + } + + listener.shutdown().await; +} + +#[tokio::test] +async fn try_route_errors_on_explicit_agent_not_found() { + let listener = bind_test_listener().await; + + let bogus_id = Uuid::new_v4(); + let err = match try_route( + Some(&listener.handle), + Some(bogus_id), + &host("anywhere.example.com"), + Uuid::new_v4(), + "anywhere.example.com:443", + ) + .await + { + Ok(_) => panic!("expected an error for an explicit agent missing from the registry"), + Err(e) => e, + }; + + let msg = format!("{err:#}"); + assert!( + msg.contains("not found in registry"), + "error should mention the missing agent, got: {msg}" + ); + + listener.shutdown().await; +} diff --git a/testsuite/tests/cli/agent/mod.rs b/testsuite/tests/cli/agent/mod.rs index 786e3668c..0ed56ed3b 100644 --- a/testsuite/tests/cli/agent/mod.rs +++ b/testsuite/tests/cli/agent/mod.rs @@ -1 +1,2 @@ +mod tunnel; mod up; diff --git a/testsuite/tests/cli/agent/tunnel.rs b/testsuite/tests/cli/agent/tunnel.rs new file mode 100644 index 000000000..c0856a54e --- /dev/null +++ b/testsuite/tests/cli/agent/tunnel.rs @@ -0,0 +1,840 @@ +use std::ffi::{OsStr, OsString}; +use std::net::Ipv4Addr; +use std::path::{Path, PathBuf}; +use std::process::{Command, Output, Stdio}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use base64::Engine as _; +use devolutions_gateway::target_addr::TargetAddr; +use devolutions_gateway::token::{ + AccessScope, ApplicationProtocol, AssociationTokenClaims, ConnectionMode, EnrollmentTokenClaims, + ReconnectionPolicy, RecordingPolicy, ScopeTokenClaims, SessionTtl, +}; +use futures_util::{SinkExt as _, StreamExt as _}; +use ipnetwork::Ipv4Network; +use nonempty::NonEmpty; +use picky::jose::jws::JwsAlg; +use picky::jose::jwt::CheckedJwtSig; +use picky::key::PrivateKey; +use serde::Serialize; +use testsuite::cli::{agent_assert_cmd, agent_tokio_cmd, dgw_tokio_cmd, wait_for_tcp_port}; +use testsuite::dgw_config::{AgentTunnelConfig, DgwConfig}; +use tokio::net::TcpListener; +use tokio::process::Child; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::tungstenite::client::IntoClientRequest as _; +use tokio_tungstenite::tungstenite::http::header::AUTHORIZATION; +use uuid::Uuid; + +const DOCKER_TARGET_PORT: u16 = 9000; +const DOCKER_EXACT_ROUTE: &str = "exact.route.test"; +const DOCKER_WILDCARD_ROUTE: &str = "*.wild.route.test"; +const DOCKER_WILDCARD_TARGET: &str = "child.wild.route.test"; +const DOCKER_GATEWAY_HOST: &str = "gateway.test"; +const AGENT_BINARY: &str = "/opt/devolutions/agent/devolutions-agent"; +const POWERSHELL_BINARY: &str = "/var/lib/devolutions-agent/.pwsh/bin/pwsh"; + +fn docker(args: I) -> anyhow::Result +where + I: IntoIterator, + S: AsRef, +{ + let args: Vec = args.into_iter().map(|arg| arg.as_ref().to_owned()).collect(); + let output = Command::new("docker").args(&args).output()?; + anyhow::ensure!( + output.status.success(), + "docker {args:?} failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + ); + Ok(output) +} + +fn docker_stdout(args: I) -> anyhow::Result +where + I: IntoIterator, + S: AsRef, +{ + let output = docker(args)?; + Ok(String::from_utf8(output.stdout)?.trim().to_owned()) +} + +fn remove_container(name: &str) { + let _ = Command::new("docker") + .args(["rm", "--force", name]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); +} + +fn container_logs(name: &str) -> String { + docker_stdout(["logs", name]).unwrap_or_default() +} + +fn assert_container_running(name: &str) { + let running = docker_stdout(["inspect", "--format", "{{.State.Running}}", name]).unwrap_or_default(); + assert_eq!(running, "true", "container stopped: {}", container_logs(name)); +} + +struct DockerEnvironment { + image: String, + network: String, + target_container: String, + dns_agent_container: String, + ip_agent_container: String, + dns_volume: String, + ip_volume: String, + target_ip: String, + dns_agent_ip: String, + ip_agent_ip: String, +} + +impl DockerEnvironment { + fn create(repo_root: &Path) -> anyhow::Result { + docker(["info", "--format", "{{.ServerVersion}}"])?; + + let id = Uuid::new_v4(); + let subnet = format!("10.{}.{}.0/24", id.as_bytes()[0], id.as_bytes()[1]); + let suffix = id.simple().to_string(); + let mut environment = Self { + image: format!("dgw-agent-tunnel-e2e:{suffix}"), + network: format!("dgw-agent-tunnel-e2e-{suffix}"), + target_container: format!("dgw-agent-tunnel-target-{suffix}"), + dns_agent_container: format!("dgw-agent-tunnel-dns-agent-{suffix}"), + ip_agent_container: format!("dgw-agent-tunnel-ip-agent-{suffix}"), + dns_volume: format!("dgw-agent-tunnel-dns-{suffix}"), + ip_volume: format!("dgw-agent-tunnel-ip-{suffix}"), + target_ip: String::new(), + dns_agent_ip: String::new(), + ip_agent_ip: String::new(), + }; + + docker([ + OsStr::new("build"), + OsStr::new("--build-arg"), + OsStr::new("BUILD_TARGET=local"), + OsStr::new("--file"), + repo_root.join("package/AgentLinux/Dockerfile").as_os_str(), + OsStr::new("--tag"), + environment.image.as_ref(), + repo_root.as_os_str(), + ])?; + + docker([ + "network", + "create", + "--driver", + "bridge", + "--subnet", + &subnet, + &environment.network, + ])?; + let subnet: Ipv4Network = docker_stdout([ + "network", + "inspect", + "--format", + "{{(index .IPAM.Config 0).Subnet}}", + &environment.network, + ])? + .parse()?; + let network = u32::from(subnet.network()); + let address = |offset| Ipv4Addr::from(network + offset); + let target_ip = address(10); + let dns_agent_ip = address(20); + let ip_agent_ip = address(21); + anyhow::ensure!( + [target_ip, dns_agent_ip, ip_agent_ip] + .into_iter() + .all(|address| subnet.contains(address) && address != subnet.broadcast()), + "docker network subnet is too small for the test fixture" + ); + environment.target_ip = target_ip.to_string(); + environment.dns_agent_ip = dns_agent_ip.to_string(); + environment.ip_agent_ip = ip_agent_ip.to_string(); + docker(["volume", "create", &environment.dns_volume])?; + docker(["volume", "create", &environment.ip_volume])?; + Ok(environment) + } + + fn start_target(&self, allowed_agent_ip: &str) -> anyhow::Result<()> { + let script = r#" +$allowed = [Net.IPAddress]::Parse($env:ALLOWED_CLIENT) +$listener = [Net.Sockets.TcpListener]::new([Net.IPAddress]::Any, 9000) +$listener.Start() +Write-Output 'READY' +while ($true) { + $client = $listener.AcceptTcpClient() + if (-not $client.Client.RemoteEndPoint.Address.Equals($allowed)) { + $client.Dispose() + continue + } + $stream = $client.GetStream() + $buffer = [byte[]]::new(8192) + $count = $stream.Read($buffer, 0, $buffer.Length) + if ($count -gt 0) { + $stream.Write($buffer, 0, $count) + } + $client.Dispose() +} +"#; + docker([ + "run", + "--detach", + "--name", + &self.target_container, + "--network", + &self.network, + "--ip", + &self.target_ip, + "--network-alias", + DOCKER_EXACT_ROUTE, + "--network-alias", + DOCKER_WILDCARD_TARGET, + "--env", + &format!("ALLOWED_CLIENT={allowed_agent_ip}"), + "--entrypoint", + POWERSHELL_BINARY, + &self.image, + "-NoLogo", + "-NoProfile", + "-Command", + script, + ])?; + Ok(()) + } + + fn wait_for_target(&self) -> anyhow::Result<()> { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + if docker_stdout(["logs", &self.target_container])?.contains("READY") { + return Ok(()); + } + anyhow::ensure!(Instant::now() < deadline, "docker target did not become ready"); + std::thread::sleep(Duration::from_millis(100)); + } + } + + fn enroll_agent( + &self, + volume: &str, + agent_ip: &str, + token_path: &Path, + route_flag: &str, + routes: &str, + ) -> anyhow::Result<()> { + let token_mount = format!( + "type=bind,src={},dst=/run/enrollment-token,readonly", + token_path.to_string_lossy() + ); + let config_mount = format!("type=volume,src={volume},dst=/etc/devolutions-agent"); + let command = format!( + "{AGENT_BINARY} up --enrollment-string \"$(cat /run/enrollment-token)\" {route_flag} '{routes}' && \ + chown -R devolutions-agent:devolutions-agent /etc/devolutions-agent" + ); + docker([ + "run", + "--rm", + "--user", + "0", + "--network", + &self.network, + "--ip", + agent_ip, + "--add-host", + &format!("{DOCKER_GATEWAY_HOST}:host-gateway"), + "--mount", + &config_mount, + "--mount", + &token_mount, + "--entrypoint", + "/bin/sh", + &self.image, + "-c", + &command, + ])?; + Ok(()) + } + + fn start_agent(&self, name: &str, volume: &str, agent_ip: &str) -> anyhow::Result<()> { + let config_mount = format!("type=volume,src={volume},dst=/etc/devolutions-agent"); + docker([ + "run", + "--detach", + "--name", + name, + "--network", + &self.network, + "--ip", + agent_ip, + "--add-host", + &format!("{DOCKER_GATEWAY_HOST}:host-gateway"), + "--mount", + &config_mount, + "--entrypoint", + AGENT_BINARY, + &self.image, + "run", + ])?; + Ok(()) + } +} + +impl Drop for DockerEnvironment { + fn drop(&mut self) { + for container in [ + &self.target_container, + &self.dns_agent_container, + &self.ip_agent_container, + ] { + remove_container(container); + } + for volume in [&self.dns_volume, &self.ip_volume] { + let _ = Command::new("docker") + .args(["volume", "rm", "--force", volume]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + } + let _ = Command::new("docker") + .args(["network", "rm", &self.network]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + let _ = Command::new("docker") + .args(["image", "rm", "--force", &self.image]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + } +} + +fn unix_timestamp() -> i64 { + i64::try_from( + SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("read system time") + .as_secs(), + ) + .expect("system time should fit in an i64") +} + +fn sign(key: &PrivateKey, content_type: &str, claims: &T) -> String { + let now = unix_timestamp(); + let mut claims = serde_json::to_value(claims).expect("serialize test claims"); + let claims = claims.as_object_mut().expect("test claims should be an object"); + claims.insert("iat".to_owned(), serde_json::json!(now)); + claims.insert("nbf".to_owned(), serde_json::json!(now - 1)); + CheckedJwtSig::new_with_cty(JwsAlg::RS256, content_type, claims) + .encode(key) + .expect("sign test token") +} + +fn association_token(key: &PrivateKey, target: TargetAddr, agent_id: Option) -> (Uuid, String) { + let session_id = Uuid::new_v4(); + let claims = AssociationTokenClaims { + jet_aid: session_id, + jet_ap: ApplicationProtocol::unknown(), + jet_cm: ConnectionMode::Fwd { + targets: NonEmpty::new(target), + }, + jet_rec: RecordingPolicy::None, + jet_flt: false, + jet_ttl: SessionTtl::Unlimited, + jet_reuse: ReconnectionPolicy::Disallowed, + exp: unix_timestamp() + 60, + jti: Uuid::new_v4(), + cert_thumb256: None, + jet_agent_id: agent_id, + }; + (session_id, sign(key, "ASSOCIATION", &claims)) +} + +fn forwarding_request( + http_port: u16, + key: &PrivateKey, + target: TargetAddr, + agent_id: Option, +) -> tokio_tungstenite::tungstenite::http::Request<()> { + let (session_id, token) = association_token(key, target, agent_id); + let mut request = format!("ws://127.0.0.1:{http_port}/jet/fwd/tcp/{session_id}") + .into_client_request() + .expect("build forwarding request"); + request.headers_mut().insert( + AUTHORIZATION, + format!("Bearer {token}").parse().expect("build authorization header"), + ); + request +} + +fn start_gateway(config_dir: &Path) -> Child { + dgw_tokio_cmd() + .env("DGATEWAY_CONFIG_PATH", config_dir) + .kill_on_drop(true) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("start gateway") +} + +async fn round_trip( + http_port: u16, + key: &PrivateKey, + target: TargetAddr, + agent_id: Option, + payload: &[u8], +) -> anyhow::Result<()> { + let (mut socket, _) = + tokio_tungstenite::connect_async(forwarding_request(http_port, key, target, agent_id)).await?; + socket.send(Message::Binary(payload.to_vec().into())).await?; + let message = tokio::time::timeout(Duration::from_secs(3), socket.next()) + .await? + .ok_or_else(|| anyhow::anyhow!("forwarding socket closed"))??; + anyhow::ensure!( + message.into_data().as_ref() == payload, + "forwarding response did not match request" + ); + Ok(()) +} + +async fn wait_for_round_trip( + http_port: u16, + key: &PrivateKey, + target: TargetAddr, + payload: &[u8], +) -> anyhow::Result<()> { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + let error = match round_trip(http_port, key, target.clone(), None, payload).await { + Ok(()) => return Ok(()), + Err(error) => error, + }; + if Instant::now() >= deadline { + return Err(error); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + +async fn wait_for_registered_agent( + http_port: u16, + key: &PrivateKey, + agent_name: &str, + expected_subnets: &[&str], + expected_domains: &[&str], + timeout: Duration, + container_logs: impl Fn() -> String, +) { + let client = reqwest::Client::new(); + let deadline = Instant::now() + timeout; + let mut last_response = serde_json::Value::Null; + + loop { + let claims = ScopeTokenClaims { + scope: AccessScope::AgentRead, + exp: unix_timestamp() + 60, + jti: Uuid::new_v4(), + }; + let response = client + .get(format!("http://127.0.0.1:{http_port}/jet/tunnel/agents")) + .bearer_auth(sign(key, "SCOPE", &claims)) + .send() + .await; + + if let Ok(response) = response + && response.status().is_success() + && let Ok(agents) = response.json::().await + { + last_response = agents.clone(); + if let Some(agent) = agents.as_array().and_then(|agents| { + agents + .iter() + .find(|agent| agent.get("name").and_then(serde_json::Value::as_str) == Some(agent_name)) + }) { + let subnets = agent + .get("subnets") + .and_then(serde_json::Value::as_array) + .into_iter() + .flatten() + .filter_map(serde_json::Value::as_str) + .collect::>(); + let domains = agent + .get("domains") + .and_then(serde_json::Value::as_array) + .into_iter() + .flatten() + .filter_map(|domain| domain.get("domain").and_then(serde_json::Value::as_str)) + .collect::>(); + let domains_are_explicit = + agent + .get("domains") + .and_then(serde_json::Value::as_array) + .is_some_and(|domains| { + domains.iter().all(|domain| { + domain.get("auto_detected").and_then(serde_json::Value::as_bool) == Some(false) + }) + }); + + if agent.get("is_online").and_then(serde_json::Value::as_bool) == Some(true) + && agent.get("route_epoch").and_then(serde_json::Value::as_u64) == Some(1) + && subnets == expected_subnets + && domains == expected_domains + && domains_are_explicit + { + return; + } + } + } + + assert!( + Instant::now() < deadline, + "agent registration did not match within {timeout:?}; response={last_response}; logs={}", + container_logs() + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + +fn enrollment_token_file(key: &PrivateKey, http_port: u16, agent_name: &str) -> tempfile::NamedTempFile { + let claims = EnrollmentTokenClaims { + exp: unix_timestamp() + 60, + jti: Uuid::new_v4(), + jet_gw_url: format!("http://{DOCKER_GATEWAY_HOST}:{http_port}"), + jet_agent_name: agent_name.to_owned(), + }; + let file = tempfile::NamedTempFile::new().expect("create enrollment token file"); + std::fs::write(file.path(), sign(key, "ENROLLMENT", &claims)).expect("write enrollment token file"); + file +} + +async fn start_echo_server() -> (u16, tokio::task::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind echo server"); + let port = listener.local_addr().expect("read echo server address").port(); + let task = tokio::spawn(async move { + loop { + let (stream, _) = listener.accept().await.expect("accept echo connection"); + tokio::spawn(async move { + let (mut read, mut write) = stream.into_split(); + let _ = tokio::io::copy(&mut read, &mut write).await; + }); + } + }); + (port, task) +} + +fn enrolled_agent_id(config_dir: &Path) -> Uuid { + std::fs::read_dir(config_dir.join("certs")) + .expect("read agent certificate directory") + .filter_map(Result::ok) + .find_map(|entry| { + entry + .file_name() + .to_str() + .and_then(|name| name.strip_suffix("-cert.pem")) + .and_then(|id| Uuid::parse_str(id).ok()) + }) + .expect("find enrolled agent identity") +} + +async fn assert_explicit_ip_is_refused(http_port: u16, key: &PrivateKey, agent_id: Uuid) { + let target_listener = TcpListener::bind("127.0.0.1:0").await.expect("bind refused target"); + let port = target_listener + .local_addr() + .expect("read refused target address") + .port(); + let (mut socket, _) = tokio_tungstenite::connect_async(forwarding_request( + http_port, + key, + TargetAddr::from_components("tcp", "127.0.0.1", port).expect("build refused target"), + Some(agent_id), + )) + .await + .expect("open authenticated forwarding socket"); + socket + .send(Message::Binary(b"must not connect".to_vec().into())) + .await + .expect("send refused payload"); + let response = tokio::time::timeout(Duration::from_secs(5), socket.next()) + .await + .expect("forwarding refusal timed out"); + assert!(!matches!(response, Some(Ok(Message::Binary(_))))); + assert!( + tokio::time::timeout(Duration::from_millis(200), target_listener.accept()) + .await + .is_err() + ); +} + +#[tokio::test] +async fn enrolled_agent_forwards_domain_only_route_and_reconnects() { + let provisioner_key = PrivateKey::generate_rsa(2048).expect("generate provisioner key"); + let public_key_data = format!( + "m{}", + base64::engine::general_purpose::STANDARD.encode( + provisioner_key + .to_public_key() + .expect("derive provisioner public key") + .to_der() + .expect("encode provisioner public key") + ) + ); + let config = DgwConfig::builder() + .hostname("localhost".to_owned()) + .provisioner_public_key_data(public_key_data) + .agent_tunnel(AgentTunnelConfig::builder().build()) + .enable_unstable(true) + .build() + .init() + .expect("initialize gateway config"); + let mut gateway = start_gateway(config.config_dir()); + wait_for_tcp_port(config.http_port()) + .await + .expect("gateway http port ready"); + + let agent_config = tempfile::tempdir().expect("create agent config directory"); + let enrollment_claims = EnrollmentTokenClaims { + exp: unix_timestamp() + 60, + jti: Uuid::new_v4(), + jet_gw_url: format!("http://127.0.0.1:{}", config.http_port()), + jet_agent_name: "smoke-agent".to_owned(), + }; + let enrollment_token = sign(&provisioner_key, "ENROLLMENT", &enrollment_claims); + agent_assert_cmd() + .env("DAGENT_CONFIG_PATH", agent_config.path()) + .args([ + "up", + "--enrollment-string", + &enrollment_token, + "--advertise-domains", + "localhost", + ]) + .assert() + .success(); + let agent_id = enrolled_agent_id(agent_config.path()); + let mut agent = agent_tokio_cmd() + .env("DAGENT_CONFIG_PATH", agent_config.path()) + .arg("run") + .kill_on_drop(true) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("start agent"); + let (echo_port, echo_task) = start_echo_server().await; + let echo_target = TargetAddr::from_components("tcp", "localhost", echo_port).expect("build echo target"); + + wait_for_round_trip( + config.http_port(), + &provisioner_key, + echo_target.clone(), + b"before restart", + ) + .await + .expect("forward before restart"); + gateway.kill().await.expect("stop gateway"); + gateway.wait().await.expect("wait for gateway shutdown"); + gateway = start_gateway(config.config_dir()); + wait_for_tcp_port(config.http_port()) + .await + .expect("restarted gateway http port ready"); + wait_for_round_trip(config.http_port(), &provisioner_key, echo_target, b"after restart") + .await + .expect("forward after restart"); + assert_explicit_ip_is_refused(config.http_port(), &provisioner_key, agent_id).await; + + agent.kill().await.expect("stop agent"); + gateway.kill().await.expect("stop gateway"); + echo_task.abort(); +} + +#[tokio::test] +#[ignore = "requires a running Docker daemon"] +async fn docker_isolates_real_agent_dns_and_ip_routes() { + let repo_root = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .parent() + .expect("testsuite should have a repository parent") + .to_owned(); + let docker = DockerEnvironment::create(&repo_root).expect("create Docker environment"); + let provisioner_key = PrivateKey::generate_rsa(2048).expect("generate provisioner key"); + let public_key_data = format!( + "m{}", + base64::engine::general_purpose::STANDARD.encode( + provisioner_key + .to_public_key() + .expect("derive provisioner public key") + .to_der() + .expect("encode provisioner public key") + ) + ); + let config = DgwConfig::builder() + .hostname(DOCKER_GATEWAY_HOST.to_owned()) + .listener_host("0.0.0.0") + .provisioner_public_key_data(public_key_data) + .agent_tunnel(AgentTunnelConfig::builder().build()) + .enable_unstable(true) + .build() + .init() + .expect("initialize gateway config"); + let mut gateway = start_gateway(config.config_dir()); + wait_for_tcp_port(config.http_port()) + .await + .expect("gateway http port ready"); + + docker + .start_target(&docker.dns_agent_ip) + .expect("start dns target container"); + docker.wait_for_target().expect("wait for dns target container"); + let exact_target = + TargetAddr::from_components("tcp", DOCKER_EXACT_ROUTE, DOCKER_TARGET_PORT).expect("build exact dns target"); + assert!( + round_trip( + config.http_port(), + &provisioner_key, + exact_target.clone(), + None, + b"must fail without agent" + ) + .await + .is_err() + ); + + let dns_agent_name = "docker-dns-agent"; + let dns_token = enrollment_token_file(&provisioner_key, config.http_port(), dns_agent_name); + docker + .enroll_agent( + &docker.dns_volume, + &docker.dns_agent_ip, + dns_token.path(), + "--advertise-domains", + &format!("{DOCKER_EXACT_ROUTE},{DOCKER_WILDCARD_ROUTE}"), + ) + .expect("enroll dns-only agent"); + docker + .start_agent(&docker.dns_agent_container, &docker.dns_volume, &docker.dns_agent_ip) + .expect("start dns-only agent"); + wait_for_registered_agent( + config.http_port(), + &provisioner_key, + dns_agent_name, + &[], + &[DOCKER_EXACT_ROUTE, DOCKER_WILDCARD_ROUTE], + Duration::from_secs(40), + || container_logs(&docker.dns_agent_container), + ) + .await; + wait_for_round_trip( + config.http_port(), + &provisioner_key, + exact_target.clone(), + b"exact DNS route", + ) + .await + .unwrap_or_else(|error| { + panic!( + "exact dns route failed: {error:#}; agent logs={}; target logs={}", + container_logs(&docker.dns_agent_container), + container_logs(&docker.target_container) + ) + }); + assert_container_running(&docker.target_container); + let wildcard_target = TargetAddr::from_components("tcp", DOCKER_WILDCARD_TARGET, DOCKER_TARGET_PORT) + .expect("build wildcard dns target"); + wait_for_round_trip( + config.http_port(), + &provisioner_key, + wildcard_target.clone(), + b"wildcard DNS route", + ) + .await + .unwrap_or_else(|error| { + panic!( + "wildcard dns route failed: {error:#}; agent logs={}; target logs={}", + container_logs(&docker.dns_agent_container), + container_logs(&docker.target_container) + ) + }); + + gateway.kill().await.expect("stop gateway"); + gateway.wait().await.expect("wait for gateway shutdown"); + gateway = start_gateway(config.config_dir()); + wait_for_tcp_port(config.http_port()) + .await + .expect("restarted gateway http port ready"); + wait_for_registered_agent( + config.http_port(), + &provisioner_key, + dns_agent_name, + &[], + &[DOCKER_EXACT_ROUTE, DOCKER_WILDCARD_ROUTE], + Duration::from_secs(150), + || container_logs(&docker.dns_agent_container), + ) + .await; + wait_for_round_trip( + config.http_port(), + &provisioner_key, + wildcard_target, + b"DNS route after restart", + ) + .await + .unwrap_or_else(|error| { + panic!( + "dns route after restart failed: {error:#}; agent logs={}; target logs={}", + container_logs(&docker.dns_agent_container), + container_logs(&docker.target_container) + ) + }); + + remove_container(&docker.dns_agent_container); + remove_container(&docker.target_container); + docker + .start_target(&docker.ip_agent_ip) + .expect("start ip target container"); + docker.wait_for_target().expect("wait for ip target container"); + let ip_target = TargetAddr::from_components("tcp", &docker.target_ip, DOCKER_TARGET_PORT).expect("build ip target"); + assert!( + round_trip( + config.http_port(), + &provisioner_key, + ip_target.clone(), + None, + b"must fail without IP agent" + ) + .await + .is_err() + ); + + let ip_agent_name = "docker-ip-agent"; + let ip_route = format!("{}/32", docker.target_ip); + let ip_token = enrollment_token_file(&provisioner_key, config.http_port(), ip_agent_name); + docker + .enroll_agent( + &docker.ip_volume, + &docker.ip_agent_ip, + ip_token.path(), + "--advertise-subnets", + &ip_route, + ) + .expect("enroll ip agent"); + docker + .start_agent(&docker.ip_agent_container, &docker.ip_volume, &docker.ip_agent_ip) + .expect("start ip agent"); + wait_for_registered_agent( + config.http_port(), + &provisioner_key, + ip_agent_name, + &[&ip_route], + &[], + Duration::from_secs(40), + || container_logs(&docker.ip_agent_container), + ) + .await; + wait_for_round_trip(config.http_port(), &provisioner_key, ip_target, b"IP route") + .await + .unwrap_or_else(|error| { + panic!( + "ip route failed: {error:#}; agent logs={}; target logs={}", + container_logs(&docker.ip_agent_container), + container_logs(&docker.target_container) + ) + }); + + gateway.kill().await.expect("stop gateway"); +} diff --git a/testsuite/tests/main.rs b/testsuite/tests/main.rs index 5d8b1e6c5..c2a339878 100644 --- a/testsuite/tests/main.rs +++ b/testsuite/tests/main.rs @@ -2,6 +2,7 @@ #![allow(clippy::print_stdout, reason = "test code uses print for diagnostics")] #![allow(clippy::print_stderr, reason = "test code uses print for diagnostics")] +mod agent_tunnel; mod cli; mod mcp_proxy; mod network_scanner;