test(agent): e2e + registry + routing test suite

testsuite/Cargo.toml

@@ -31,18 +31,29 @@ typed-builder = "0.21"
 tokio-tungstenite = { version = "0.26", 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"] }
+devolutions-gateway-task = { path = "../crates/devolutions-gateway-task" }
 base64 = "0.22"
-proxy-socks = { path = "../crates/proxy-socks" }
+camino = "1"
+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" }
+proxy-socks = { path = "../crates/proxy-socks" }
+quinn = "0.11"
+rcgen = { version = "0.13", features = ["pem", "x509-parser"] }
 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"

testsuite/tests/agent_tunnel/cert.rs

@@ -0,0 +1,74 @@
+//! Unit tests for `agent-tunnel/src/cert.rs`.
+//!
+//! Focus on the identity invariants exercised by enrollment (#1773) and
+//! certificate renewal (#1775): the gateway must encode the agent's UUID in
+//! the issued cert's URN SAN, and recover the same UUID from any cert it
+//! later sees on the wire.
+
+use agent_tunnel::cert::{CaManager, extract_agent_id_from_pem};
+use camino::Utf8PathBuf;
+use uuid::Uuid;
+
+use super::common::{generate_csr_with_cn, generate_test_key_and_csr};
+
+fn fresh_ca() -> std::sync::Arc<CaManager> {
+    let temp_dir = tempfile::tempdir().expect("create tempdir");
+    let data_dir = Utf8PathBuf::from_path_buf(temp_dir.path().to_path_buf()).expect("UTF-8 temp path");
+    // Leak the TempDir for the test's lifetime: CaManager owns the files
+    // already loaded, so dropping the dir while still in use is fine, but
+    // leaking removes any chance of TOCTOU surprises.
+    std::mem::forget(temp_dir);
+    CaManager::load_or_generate(&data_dir).expect("CA generation")
+}
+
+/// Security invariant from #1775 review: when the gateway re-signs (or
+/// initially signs) an agent CSR, the issued cert's URN SAN encodes the
+/// `agent_id` parameter passed in by the caller, **not** anything from the
+/// CSR's own subject. A compromised agent crafting a CSR with someone else's
+/// CN must not be able to impersonate.
+#[test]
+fn sign_agent_csr_ignores_csr_subject_uses_passed_identity() {
+    let 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 cert 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 ca_manager = fresh_ca();
+
+    let known_id = Uuid::new_v4();
+    let (_key, csr_pem) = generate_test_key_and_csr("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 ca_manager = fresh_ca();
+
+    // The CA's own root cert does not carry an `urn:uuid:` SAN.
+    let error = extract_agent_id_from_pem(ca_manager.ca_cert_pem()).expect_err("CA cert has no urn:uuid SAN");
+
+    let msg = format!("{error:#}");
+    assert!(
+        msg.contains("urn:uuid"),
+        "error should reference the missing urn:uuid SAN, got: {msg}"
+    );
+}

testsuite/tests/agent_tunnel/common.rs

@@ -0,0 +1,192 @@
+//! Shared helpers for the agent-tunnel test suite.
+//!
+//! These were originally private to `integration.rs`; consolidated here so
+//! the cert-renewal E2E and the routing tests can reuse them without
+//! duplicating ~80 lines of QUIC + mTLS scaffolding per test.
+
+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 camino::Utf8PathBuf;
+use devolutions_gateway_task::ShutdownHandle;
+use tempfile::TempDir;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpListener;
+use tokio::task::JoinHandle;
+use uuid::Uuid;
+
+/// Start a TCP echo server that echoes back whatever it receives.
+pub(super) async fn start_echo_server() -> (SocketAddr, JoinHandle<()>) {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    let handle = tokio::spawn(async move {
+        loop {
+            let (mut stream, _) = match listener.accept().await {
+                Ok(v) => v,
+                Err(_) => break,
+            };
+
+            tokio::spawn(async move {
+                let mut buf = vec![0u8; 65535];
+                loop {
+                    let n = match stream.read(&mut buf).await {
+                        Ok(0) | Err(_) => break,
+                        Ok(n) => n,
+                    };
+                    if stream.write_all(&buf[..n]).await.is_err() {
+                        break;
+                    }
+                }
+            });
+        }
+    });
+
+    (addr, handle)
+}
+
+/// Generate a key pair and CSR (same as the real agent does during enrollment).
+pub(super) fn generate_test_key_and_csr(agent_name: &str) -> (rcgen::KeyPair, String) {
+    generate_csr_with_cn(agent_name)
+}
+
+/// Generate a key pair and CSR with the given Common Name on the CSR subject.
+///
+/// Useful for the security-invariant test that checks `sign_agent_csr` ignores
+/// the CSR subject in favor of the mTLS-authenticated agent name.
+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("CSR to PEM");
+    (key_pair, csr_pem)
+}
+
+/// Create a Quinn client connection to the gateway with mTLS.
+pub(super) 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<rustls_pki_types::CertificateDer<'static>> =
+        certs(&mut std::io::BufReader::new(client_cert_pem.as_bytes()))
+            .collect::<Result<Vec<_>, _>>()
+            .expect("parse client certs");
+    let client_key = private_key(&mut std::io::BufReader::new(client_key_pem.as_bytes()))
+        .expect("parse private key")
+        .expect("no private key found");
+
+    let mut roots = rustls::RootCertStore::empty();
+    let ca_certs: Vec<rustls_pki_types::CertificateDer<'static>> =
+        certs(&mut std::io::BufReader::new(ca_cert_pem.as_bytes()))
+            .collect::<Result<Vec<_>, _>>()
+            .expect("parse CA certs");
+    for cert in ca_certs {
+        roots.add(cert).expect("add CA cert to root store");
+    }
+
+    // Trust only the test CA. Hostname verification is still on (SNI = "localhost").
+    let verifier = rustls::client::WebPkiServerVerifier::builder(Arc::new(roots))
+        .build()
+        .expect("build verifier");
+
+    let mut client_crypto = rustls::ClientConfig::builder()
+        .dangerous()
+        .with_custom_certificate_verifier(verifier)
+        .with_client_auth_cert(client_certs, client_key)
+        .expect("client auth config");
+
+    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("QUIC client config"),
+    ));
+
+    let mut endpoint = quinn::Endpoint::client("0.0.0.0:0".parse().expect("bind addr")).expect("create endpoint");
+    endpoint.set_default_client_config(client_config);
+
+    endpoint
+        .connect(server_addr, "localhost")
+        .expect("initiate connection")
+        .await
+        .expect("QUIC handshake")
+}
+
+/// Live `AgentTunnelListener` running on a random localhost port, plus the
+/// resources needed to drive and shut it down cleanly.
+pub(super) struct TestListener {
+    pub handle: AgentTunnelHandle,
+    shutdown: ShutdownHandle,
+    task: JoinHandle<()>,
+    _temp_dir: TempDir,
+}
+
+impl TestListener {
+    /// Signal shutdown and wait for the listener task to exit (or time out).
+    pub(super) async fn shutdown(self) {
+        self.shutdown.signal();
+        let _ = tokio::time::timeout(Duration::from_secs(2), self.task).await;
+    }
+}
+
+/// Bring up a fresh `AgentTunnelListener` on `127.0.0.1:0` with a freshly
+/// generated CA in a temp directory.
+pub(super) async fn bind_test_listener() -> TestListener {
+    let temp_dir = tempfile::tempdir().expect("create tempdir");
+    let data_dir = Utf8PathBuf::from_path_buf(temp_dir.path().to_path_buf()).expect("UTF-8 temp path");
+    let ca_manager = CaManager::load_or_generate(&data_dir).expect("CA generation");
+
+    let listen_addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
+    let (listener, handle) = AgentTunnelListener::bind(listen_addr, ca_manager, "localhost")
+        .await
+        .expect("bind QUIC listener");
+
+    let (shutdown, shutdown_signal) = ShutdownHandle::new();
+    let task = tokio::spawn(async move {
+        use devolutions_gateway_task::Task;
+        let _ = listener.run(shutdown_signal).await;
+    });
+
+    // Give listener time to be ready.
+    tokio::time::sleep(Duration::from_millis(50)).await;
+
+    TestListener {
+        handle,
+        shutdown,
+        task,
+        _temp_dir: temp_dir,
+    }
+}
+
+/// Poll the registry until `agent_id` is present and has applied at least one
+/// route advertisement with epoch ≥ `min_epoch`, or panic after 5 seconds.
+///
+/// Replaces the older fixed-sleep pattern that raced on slow CI runners:
+/// `ctrl.send(&RouteAdvertise)` only guarantees the message is on the wire,
+/// not that the gateway has processed it. Default `RouteAdvertisementState`
+/// starts at epoch 0, so any successful RouteAdvertise bumps it to ≥ 1.
+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;
+        }
+        if Instant::now() >= deadline {
+            panic!("agent {agent_id} did not advertise route at epoch >= {min_epoch} within 5s");
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+}