fix: review agent findings — SAN type, UUID dedup, proxy safety

- SanType::Rfc822Name → SanType::URI for urn:uuid: (correct X.509 type)
- GeneralName::RFC822Name → GeneralName::URI in extraction
- Reject duplicate agent UUID on enrollment (409 Conflict)
- tokio::join! instead of select! for session proxy (prevents data loss)
- JoinSet instead of Vec<JoinHandle> (prevents unbounded growth)
- Timeout (30s) on session handshake recv_request/recv_response
- Fix typos: "redeemd" → "redeemed"

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: irvingoujAtDevolution

devolutions-agent/src/tunnel.rs

@@ -447,7 +447,10 @@ async fn run_session_proxy(advertise_subnets: Vec<Ipv4Network>, send: quinn::Sen
     let _: anyhow::Result<()> = async {
         let mut session: SessionStream<_, _> = (send, recv).into();
 
-        let connect_msg = session.recv_request().await.context("recv ConnectRequest")?;
+        let connect_msg = tokio::time::timeout(Duration::from_secs(30), session.recv_request())
+            .await
+            .context("session handshake timeout")?
+            .context("recv ConnectRequest")?;
 
         info!(
             session_id = %connect_msg.session_id,
@@ -483,14 +486,14 @@ async fn run_session_proxy(advertise_subnets: Vec<Ipv4Network>, send: quinn::Sen
         let (mut send, mut recv) = session.into_inner();
         let (mut tcp_read, mut tcp_write) = tcp_stream.into_split();
 
-        tokio::select! {
-            r = tokio::io::copy(&mut recv, &mut tcp_write) => {
-                r.inspect_err(|e| debug!(%e, "QUIC->TCP copy ended"))?;
-            }
-            r = tokio::io::copy(&mut tcp_read, &mut send) => {
-                r.inspect_err(|e| debug!(%e, "TCP->QUIC copy ended"))?;
-            }
-        }
+        // Use join! (not select!) to wait for BOTH directions to finish.
+        // select! would cancel in-flight data when one direction closes first.
+        let (r1, r2) = tokio::join!(
+            tokio::io::copy(&mut recv, &mut tcp_write),
+            tokio::io::copy(&mut tcp_read, &mut send),
+        );
+        r1.inspect_err(|e| debug!(%e, "QUIC->TCP copy ended"))?;
+        r2.inspect_err(|e| debug!(%e, "TCP->QUIC copy ended"))?;
 
         Ok(())
     }

devolutions-gateway/src/agent_tunnel/cert.rs

@@ -141,7 +141,7 @@ impl CaManager {
         agent_params
             .distinguished_name
             .push(DnType::OrganizationName, CA_ORG_NAME);
-        agent_params.subject_alt_names.push(SanType::Rfc822Name(
+        agent_params.subject_alt_names.push(SanType::URI(
             format!("urn:uuid:{agent_id}").try_into().context("SAN URI")?,
         ));
         if let Some(hostname) = agent_hostname {
@@ -364,7 +364,7 @@ pub fn extract_agent_id_from_der(der_bytes: &[u8]) -> anyhow::Result<Uuid> {
     for ext in cert.extensions() {
         if let x509_parser::extensions::ParsedExtension::SubjectAlternativeName(san) = ext.parsed_extension() {
             for name in &san.general_names {
-                if let x509_parser::extensions::GeneralName::RFC822Name(val) = name
+                if let x509_parser::extensions::GeneralName::URI(val) = name
                     && let Some(uuid_str) = val.strip_prefix("urn:uuid:")
                 {
                     return uuid_str.parse().context("parse UUID from SAN");

devolutions-gateway/src/agent_tunnel/enrollment_store.rs

@@ -1,6 +1,6 @@
 //! In-memory store for one-time enrollment tokens.
 //!
-//! Tokens are generated by the webapp enrollment-string endpoint and redeemd
+//! Tokens are generated by the webapp enrollment-string endpoint and redeemed
 //! by the agent enrollment endpoint. Each token is single-use and has an expiry.
 
 use std::time::{SystemTime, UNIX_EPOCH};
@@ -50,7 +50,7 @@ impl EnrollmentTokenStore {
 
     /// Consumes a token if it exists and is not expired.
     ///
-    /// Returns `true` if the token was valid and has been redeemd (removed).
+    /// Returns `true` if the token was valid and has been redeemed (removed).
     /// Returns `false` if the token doesn't exist or is expired.
     pub fn redeem(&self, token: &str) -> bool {
         let now = current_time_secs();

devolutions-gateway/src/agent_tunnel/listener.rs

@@ -75,10 +75,10 @@ impl AgentTunnelHandle {
             .await
             .map_err(|e| anyhow::anyhow!("send ConnectRequest: {e}"))?;
 
-        // Read ConnectResponse.
-        let response = session
-            .recv_response()
+        // Read ConnectResponse (with timeout to prevent stalled peers).
+        let response = tokio::time::timeout(Duration::from_secs(30), session.recv_response())
             .await
+            .map_err(|_| anyhow::anyhow!("session handshake timeout (30s)"))?
             .map_err(|e| anyhow::anyhow!("recv ConnectResponse: {e}"))?;
 
         if !response.is_success() {
@@ -174,7 +174,7 @@ impl devolutions_gateway_task::Task for AgentTunnelListener {
         let local_addr = self.endpoint.local_addr()?;
         info!(%local_addr, "Agent tunnel listener started");
 
-        let mut conn_handles: Vec<tokio::task::JoinHandle<()>> = Vec::new();
+        let mut conn_handles = tokio::task::JoinSet::new();
 
         loop {
             tokio::select! {
@@ -195,19 +195,17 @@ impl devolutions_gateway_task::Task for AgentTunnelListener {
                     let registry = Arc::clone(&self.registry);
                     let agent_connections = Arc::clone(&self.agent_connections);
 
-                    conn_handles.push(tokio::spawn(
+                    conn_handles.spawn(
                         run_agent_connection(registry, agent_connections, incoming),
-                    ));
+                    );
                 }
+
+                // Reap completed connection tasks to prevent unbounded growth.
+                Some(_) = conn_handles.join_next() => {}
             }
         }
 
-        for handle in &conn_handles {
-            handle.abort();
-        }
-        for handle in conn_handles {
-            let _ = handle.await;
-        }
+        conn_handles.shutdown().await;
 
         Ok(())
     }

devolutions-gateway/src/api/tunnel.rs

@@ -114,6 +114,13 @@ async fn enroll_agent(
 
     let agent_id = req.agent_id;
 
+    // Reject duplicate agent IDs to prevent identity shadowing.
+    if handle.registry().get(&agent_id).is_some() {
+        return Err(
+            crate::http::HttpErrorBuilder::new(axum::http::StatusCode::CONFLICT).msg("agent ID already registered")
+        );
+    }
+
     let signed = handle
         .ca_manager()
         .sign_agent_csr(agent_id, &req.agent_name, &req.csr_pem, req.agent_hostname.as_deref())