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

crates/agent-tunnel/src/cert.rs

@@ -8,7 +8,7 @@ use std::time::Duration;
 
 use anyhow::{Context as _, bail};
 use camino::{Utf8Path, Utf8PathBuf};
-use picky::pem::parse_pem;
+use picky::pem::{PemError, parse_pem, read_pem};
 use picky::x509::Cert;
 use picky_asn1_x509::{ExtensionView, GeneralName};
 use rcgen::{CertificateParams, DnType, ExtendedKeyUsagePurpose, IsCa, KeyPair, KeyUsagePurpose, SanType};
@@ -32,29 +32,28 @@ fn cert_pem_to_der(pem_str: &str) -> anyhow::Result<Vec<u8>> {
 /// Parse one or more PEM-encoded certificates into `rustls` certificate types.
 ///
 /// A PEM file can carry multiple concatenated CERTIFICATE blocks (chain). We
-/// iterate block-by-block with [`parse_pem`], check each label, and wrap the
-/// DER bytes in [`rustls_pki_types::CertificateDer`] — the only type the
-/// rustls/quinn TLS builders accept.
+/// use [`read_pem`] in a loop — each call consumes one block; `HeaderNotFound`
+/// signals "no more blocks left", which is the termination condition. Each
+/// block's label is verified, then the DER bytes are wrapped in
+/// [`rustls_pki_types::CertificateDer`] — the type the rustls/quinn TLS
+/// builders accept.
 fn read_cert_chain(pem_str: &str) -> anyhow::Result<Vec<rustls::pki_types::CertificateDer<'static>>> {
+    use std::io::BufReader;
+
+    let mut reader = BufReader::new(pem_str.as_bytes());
     let mut chain = Vec::new();
-    let mut remaining = pem_str;
-    while let Some(start) = remaining.find("-----BEGIN ") {
-        let block_end = remaining[start..]
-            .find("-----END ")
-            .and_then(|e| {
-                remaining[start + e..]
-                    .find("-----\n")
-                    .map(|n| start + e + n + "-----\n".len())
-            })
-            .context("malformed PEM block (no END tag)")?;
-
-        let block = &remaining[start..block_end];
-        let pem = parse_pem(block).context("parse PEM block")?;
-        if pem.label() != PEM_LABEL_CERTIFICATE {
-            bail!("expected {PEM_LABEL_CERTIFICATE} PEM, got {}", pem.label());
+
+    loop {
+        match read_pem(&mut reader) {
+            Ok(pem) => {
+                if pem.label() != PEM_LABEL_CERTIFICATE {
+                    bail!("expected {PEM_LABEL_CERTIFICATE} PEM, got {}", pem.label());
+                }
+                chain.push(rustls::pki_types::CertificateDer::from(pem.data().to_vec()));
+            }
+            Err(PemError::HeaderNotFound) => break,
+            Err(e) => return Err(anyhow::Error::new(e).context("parse PEM block")),
         }
-        chain.push(rustls::pki_types::CertificateDer::from(pem.data().to_vec()));
-        remaining = &remaining[block_end..];
     }
 
     if chain.is_empty() {

crates/agent-tunnel/src/lib.rs

@@ -10,6 +10,7 @@ pub mod cert;
 pub mod enrollment_store;
 pub mod listener;
 pub mod registry;
+pub mod routing;
 pub mod stream;
 
 pub use enrollment_store::EnrollmentTokenStore;

crates/agent-tunnel/src/listener.rs

@@ -153,7 +153,7 @@ impl AgentTunnelListener {
         let handle = AgentTunnelHandle {
             registry: Arc::clone(&registry),
             agent_connections: Arc::clone(&agent_connections),
-            ca_manager,
+            ca_manager: Arc::clone(&ca_manager),
             enrollment_token_store,
         };
 
@@ -165,6 +165,11 @@ impl AgentTunnelListener {
 
         Ok((listener, handle))
     }
+
+    /// Returns the local address the QUIC endpoint is bound to.
+    pub fn local_addr(&self) -> SocketAddr {
+        self.endpoint.local_addr().expect("endpoint has local addr")
+    }
 }
 
 #[async_trait]
@@ -197,9 +202,7 @@ impl devolutions_gateway_task::Task for AgentTunnelListener {
                     let registry = Arc::clone(&self.registry);
                     let agent_connections = Arc::clone(&self.agent_connections);
 
-                    conn_handles.spawn(
-                        run_agent_connection(registry, agent_connections, incoming),
-                    );
+                    conn_handles.spawn(run_agent_connection(registry, agent_connections, incoming));
                 }
 
                 // Reap completed connection tasks to prevent unbounded growth.
@@ -248,7 +251,7 @@ async fn run_agent_connection(
 
         info!(%agent_id, %agent_name, %peer_addr, "Agent authenticated via mTLS");
 
-        let peer = Arc::new(AgentPeer::new(agent_id, agent_name, fingerprint));
+        let peer = Arc::new(AgentPeer::new(agent_id, agent_name.clone(), fingerprint));
         registry.register(Arc::clone(&peer)).await;
         agent_connections.write().await.insert(agent_id, conn.clone());
 

crates/agent-tunnel/src/registry.rs

@@ -10,6 +10,8 @@ use serde::Serialize;
 use tokio::sync::RwLock as TokioRwLock;
 use uuid::Uuid;
 
+use crate::routing::RouteTarget;
+
 /// Duration after which an agent is considered offline if no heartbeat has been received.
 pub const AGENT_OFFLINE_TIMEOUT: Duration = Duration::from_secs(90);
 
@@ -32,6 +34,33 @@ pub struct RouteAdvertisementState {
     pub updated_at: SystemTime,
 }
 
+impl RouteAdvertisementState {
+    /// Match this route set against a parsed target host.
+    ///
+    /// Returns a specificity score if matched, or `None` if no match.
+    /// IP subnet matches return `usize::MAX` (always highest priority).
+    /// Domain matches return the matched domain length (longer = more specific).
+    pub fn matches_target(&self, target: &RouteTarget) -> Option<usize> {
+        use std::net::IpAddr;
+
+        match target {
+            // Only IPv4 subnets are currently tracked; only match IPv4 target IPs.
+            RouteTarget::Ip(IpAddr::V4(ipv4)) => self
+                .subnets
+                .iter()
+                .any(|subnet| subnet.contains(*ipv4))
+                .then_some(usize::MAX),
+            RouteTarget::Ip(IpAddr::V6(_)) => None,
+            RouteTarget::Hostname(hostname) => self
+                .domains
+                .iter()
+                .filter(|adv| adv.domain.matches_hostname(hostname.as_str()))
+                .map(|adv| adv.domain.as_str().len())
+                .max(),
+        }
+    }
+}
+
 impl Default for RouteAdvertisementState {
     fn default() -> Self {
         let now = SystemTime::now();
@@ -79,6 +108,34 @@ impl AgentPeer {
         self.last_seen.store(now_ms, Ordering::Release);
     }
 
+    /// Set `last_seen` to an explicit timestamp (milliseconds since UNIX epoch).
+    ///
+    /// Test-only API — the `_for_test` suffix is the project's signal that
+    /// production code must not call this. Used by integration tests in
+    /// `devolutions-gateway/tests/agent_tunnel_*` to force an agent into the
+    /// "offline" state without waiting for the real timeout to elapse;
+    /// production code should use [`touch`](Self::touch) instead. We deliberately
+    /// keep this `pub` instead of `cfg(test)` because the consumer is a
+    /// different crate's tests, and `cfg(test)` is only set within the
+    /// declaring crate's own test build.
+    #[doc(hidden)]
+    pub fn set_last_seen_for_test(&self, last_seen_ms: u64) {
+        self.last_seen.store(last_seen_ms, Ordering::Release);
+    }
+
+    /// Overwrite `received_at` on the current route state.
+    ///
+    /// Test-only API. Intended for tests that need to assert ordering by
+    /// arrival time without relying on wall-clock `thread::sleep` — which is
+    /// flaky on platforms with coarse timer resolution (e.g. Windows ~16 ms).
+    /// See the note on [`set_last_seen_for_test`](Self::set_last_seen_for_test)
+    /// for why this is `pub` rather than `cfg(test)`.
+    #[doc(hidden)]
+    pub fn set_received_at_for_test(&self, received_at: SystemTime) {
+        let mut state = self.route_state.write();
+        state.received_at = received_at;
+    }
+
     /// Returns the last-seen timestamp as milliseconds since UNIX epoch.
     pub fn last_seen_ms(&self) -> u64 {
         self.last_seen.load(Ordering::Acquire)
@@ -223,6 +280,39 @@ impl AgentRegistry {
     pub async fn agent_infos(&self) -> Vec<AgentInfo> {
         self.agents.read().await.values().map(AgentInfo::from).collect()
     }
+
+    /// Find all online agents that can route to the given parsed target host.
+    ///
+    /// For IP targets: matches against advertised subnets.
+    /// For domain targets: uses longest suffix match (more specific domain wins).
+    ///
+    /// Results with equal specificity are sorted by `received_at` descending (most recent first).
+    pub async fn find_agents_for(&self, target: &RouteTarget) -> Vec<Arc<AgentPeer>> {
+        let mut best_specificity: usize = 0;
+        let mut candidates: Vec<(SystemTime, Arc<AgentPeer>)> = Vec::new();
+
+        let agents = self.agents.read().await;
+        for agent in agents.values() {
+            if !agent.is_online(AGENT_OFFLINE_TIMEOUT) {
+                continue;
+            }
+
+            let route_state = agent.route_state();
+
+            if let Some(specificity) = route_state.matches_target(target) {
+                if specificity > best_specificity {
+                    best_specificity = specificity;
+                    candidates.clear();
+                    candidates.push((route_state.received_at, Arc::clone(agent)));
+                } else if specificity == best_specificity {
+                    candidates.push((route_state.received_at, Arc::clone(agent)));
+                }
+            }
+        }
+
+        candidates.sort_by(|a, b| b.0.cmp(&a.0));
+        candidates.into_iter().map(|(_, agent)| agent).collect()
+    }
 }
 
 impl Default for AgentRegistry {