feat(dgw): route KDC traffic through agent tunnel

When an agent advertises the KDC's subnet or DNS domain, route Kerberos
traffic through the QUIC tunnel just like every other proxy path. Closes
the last gap left after the transparent routing PR (#1741).

Two paths now use the same routing pipeline as connection forwarding:

- `/jet/KdcProxy` HTTP endpoint -- the handler builds a `KdcConnector`
  and forwards through it. When an agent advertises the KDC subnet, the
  request goes through the agent tunnel; otherwise it falls back to a
  direct TCP/UDP connection.

- RDP CredSSP/NLA -- `rdp_proxy.rs::send_network_request` previously
  hard-coded `None` for the agent handle. `RdpProxy` now carries a
  `KdcConnector` field that the CredSSP machinery
  (`perform_credssp_as_*` -> `resolve_*_generator` -> `send_network_request`)
  uses for every Kerberos sub-request. The same change reaches the
  credential-injection clean path (`rd_clean_path.rs`).

`KdcConnector` (new `src/kdc_connector.rs`) bundles the three inputs the
routing pipeline needs (`session_id`, `explicit_agent_id`,
`agent_tunnel_handle`) into a single value and always defers the
routing decision to `agent_tunnel::routing::try_route`. Callers never
pre-decide "direct" vs "via tunnel": the routing pipeline does, and
its existing `explicit_agent_id` enforcement (pin without tunnel handle
must error, never silently fall back to direct) is preserved end-to-end.

Session correlation:

- RDP CredSSP callers pass the parent association's `claims.jet_aid`
  as `session_id`, so KDC sub-traffic ties back to its parent RDP
  session in agent-side logs.
- The HTTP `/jet/KdcProxy` handler passes the KDC token's own `jti`,
  the most persistent identifier available without a parent
  association. `KdcTokenClaims` now exposes `jti` through its serde
  helper, matching how every other `*TokenClaims` type surfaces `jti`.

Explicit-agent routing (matches every other proxy path):

- The parent association's `jet_agent_id`, when set, is forwarded to
  `try_route`. KDC traffic must route via that agent or fail -- never
  silently fall back to a different agent or to direct connect. The
  HTTP handler passes `None` (no parent association).
- A new UDP-via-agent guard rejects `udp://` KDC targets whenever the
  routing pipeline selects an agent. Without it, an explicit
  `jet_agent_id` pin could be downgraded to direct UDP, since the
  agent tunnel currently carries only TCP.

Hardening (came along since they share the file):

- 64 KiB `MAX_KDC_REPLY_MESSAGE_LEN` DoS cap on the announced TCP-framed
  KDC reply length, with overflow-safe length math.
- UDP scheme guard at the direct-connect branch (preserved).

Tests:

- `kdc_connector` unit tests pin the routing-decision contract:
  pin-without-tunnel must error, no-pin-no-tunnel directs, pin-with-
  missing-agent errors, no-match falls back to direct. The success
  path and the UDP-via-agent guard both require a live `connect_via_agent`
  fixture and are tracked as follow-up TODOs in the same file.
- A `pub(hidden) AgentTunnelHandle::for_testing(registry, ca_manager)`
  constructor was added to the `agent-tunnel` crate so the gateway
  crate's tests can build a handle without standing up a real QUIC
  endpoint.

Issue: DGW-384

Author: irvingoujAtDevolution

crates/agent-tunnel/src/listener.rs

@@ -43,6 +43,20 @@ impl AgentTunnelHandle {
         &self.ca_manager
     }
 
+    /// Build an `AgentTunnelHandle` without spinning up a real QUIC endpoint, for tests
+    /// that need to exercise routing decisions against a hand-built registry.
+    ///
+    /// Not gated behind `#[cfg(test)]` because tests in dependent crates
+    /// (`devolutions-gateway::kdc_connector`) need to call it.
+    #[doc(hidden)]
+    pub fn for_testing(registry: Arc<AgentRegistry>, ca_manager: Arc<CaManager>) -> Self {
+        Self {
+            registry,
+            agent_connections: Arc::new(RwLock::new(HashMap::new())),
+            ca_manager,
+        }
+    }
+
     /// Open a proxy stream through a connected agent.
     // TODO: Emit TrafficEvent for connections routed through the agent tunnel.
     pub async fn connect_via_agent(

crates/agent-tunnel/src/routing.rs

@@ -1,7 +1,9 @@
 //! Shared routing pipeline for agent tunnel.
 //!
 //! Consumed by the upstream connection paths (forwarding, RDP clean path,
-//! generic client) to ensure consistent routing behavior and error messages.
+//! generic client) and by the KDC proxy (HTTP endpoint plus the CredSSP/NLA
+//! sub-flow inside `rdp_proxy.rs`) to ensure consistent routing behavior and
+//! error messages.
 
 use std::net::IpAddr;
 use std::sync::Arc;

devolutions-gateway/src/api/kdc_proxy.rs

@@ -1,20 +1,17 @@
-use std::io;
-
 use axum::Router;
 use axum::extract::State;
-use axum::http::StatusCode;
 use axum::routing::post;
 use picky_krb::messages::KdcProxyMessage;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::net::{TcpStream, UdpSocket};
+use uuid::Uuid;
 
 use crate::DgwState;
 use crate::credential_injection_kdc::{
     CredentialInjectionKdcInterception, CredentialInjectionKdcRequest, CredentialInjectionKdcResolveError,
     kdc_proxy_message_realm,
 };
 use crate::extract::KdcToken;
-use crate::http::{HttpError, HttpErrorBuilder};
+use crate::http::HttpError;
+use crate::kdc_connector::KdcConnector;
 use crate::target_addr::TargetAddr;
 use crate::token::{KdcDestination, KdcTokenClaims};
 
@@ -26,9 +23,13 @@ async fn kdc_proxy(
     State(DgwState {
         conf_handle,
         credentials,
+        agent_tunnel_handle,
         ..
     }): State<DgwState>,
-    KdcToken(KdcTokenClaims { destination }): KdcToken,
+    KdcToken(KdcTokenClaims {
+        destination,
+        jti: token_jti,
+    }): KdcToken,
     body: axum::body::Bytes,
 ) -> Result<Vec<u8>, HttpError> {
     let conf = conf_handle.get_conf();
@@ -70,13 +71,21 @@ async fn kdc_proxy(
         }
         KdcDestination::Real { krb_realm, krb_kdc } => {
             let envelope_realm = kdc_proxy_message_realm(&kdc_proxy_message);
+
+            // session_id: the HTTP /jet/KdcProxy endpoint has no parent association token, so we
+            // use the KDC token's own `jti` for log correlation (the RDP CredSSP/NLA caller
+            // passes `claims.jet_aid` so KDC sub-traffic correlates with its parent RDP session).
+            // explicit_agent_id: HTTP has no parent association, hence no `jet_agent_id` pin.
+            let kdc_connector = KdcConnector::new(token_jti, None, agent_tunnel_handle);
+
             forward_to_real_kdc(
                 kdc_proxy_message,
                 envelope_realm,
                 &krb_realm,
                 &krb_kdc,
                 conf.debug.override_kdc.as_ref(),
                 conf.debug.disable_token_validation,
+                &kdc_connector,
             )
             .await
         }
@@ -107,6 +116,7 @@ async fn forward_to_real_kdc(
     token_kdc_addr: &TargetAddr,
     override_kdc: Option<&TargetAddr>,
     bypass_realm_check: bool,
+    kdc_connector: &KdcConnector,
 ) -> Result<Vec<u8>, HttpError> {
     let realm = envelope_realm.ok_or_else(|| HttpError::bad_request().msg("realm is missing from KDC request"))?;
     debug!(resolved_realm = %realm, "Forward-to-real-KDC realm resolved");
@@ -120,7 +130,9 @@ async fn forward_to_real_kdc(
         None => token_kdc_addr,
     };
 
-    let kdc_reply_bytes = send_krb_message(kdc_addr, &kdc_proxy_message.kerb_message.0.0).await?;
+    let kdc_reply_bytes = kdc_connector
+        .send(kdc_addr, &kdc_proxy_message.kerb_message.0.0)
+        .await?;
 
     let reply = KdcProxyMessage::from_raw_kerb_message(&kdc_reply_bytes)
         .map_err(HttpError::internal().with_msg("couldn't create KDC proxy reply").err())?;
@@ -130,7 +142,7 @@ async fn forward_to_real_kdc(
     reply.to_vec().map_err(HttpError::internal().err())
 }
 
-fn enforce_credential_injection_enabled(jet_cred_id: uuid::Uuid, enable_unstable: bool) -> Result<(), HttpError> {
+fn enforce_credential_injection_enabled(jet_cred_id: Uuid, enable_unstable: bool) -> Result<(), HttpError> {
     if enable_unstable {
         return Ok(());
     }
@@ -165,104 +177,6 @@ fn enforce_realm_token_match(token_realm: &str, request_realm: &str, bypass: boo
         .err()(format!("expected: {token_realm}, got: {request_realm}")))
 }
 
-async fn read_kdc_reply_message(connection: &mut TcpStream) -> io::Result<Vec<u8>> {
-    let len = connection.read_u32().await?;
-    let mut buf = vec![0; (len + 4).try_into().expect("u32-to-usize")];
-    buf[0..4].copy_from_slice(&(len.to_be_bytes()));
-    connection.read_exact(&mut buf[4..]).await?;
-    Ok(buf)
-}
-
-#[track_caller]
-fn unable_to_reach_kdc_server_err(error: io::Error) -> HttpError {
-    use io::ErrorKind;
-
-    let builder = match error.kind() {
-        ErrorKind::TimedOut => HttpErrorBuilder::new(StatusCode::GATEWAY_TIMEOUT),
-        ErrorKind::ConnectionRefused => HttpError::bad_gateway(),
-        ErrorKind::ConnectionAborted => HttpError::bad_gateway(),
-        ErrorKind::ConnectionReset => HttpError::bad_gateway(),
-        ErrorKind::BrokenPipe => HttpError::bad_gateway(),
-        ErrorKind::OutOfMemory => HttpError::internal(),
-        // FIXME: once stabilized use new IO error variants
-        // - https://github.com/rust-lang/rust/pull/106375
-        // - https://github.com/rust-lang/rust/issues/86442
-        // ErrorKind::NetworkDown => HttpErrorBuilder::new(StatusCode::SERVICE_UNAVAILABLE),
-        // ErrorKind::NetworkUnreachable => HttpError::bad_gateway(),
-        // ErrorKind::HostUnreachable => HttpError::bad_gateway(),
-        // TODO: When the above is applied, we can return an internal error in the fallback branch.
-        _ => HttpError::bad_gateway(),
-    };
-
-    builder.with_msg("unable to reach KDC server").build(error)
-}
-
-/// Sends the Kerberos message to the specified KDC address.
-pub async fn send_krb_message(kdc_addr: &TargetAddr, message: &[u8]) -> Result<Vec<u8>, HttpError> {
-    let protocol = kdc_addr.scheme();
-
-    debug!("Connecting to KDC server located at {kdc_addr} using protocol {protocol}...");
-
-    if protocol == "tcp" {
-        #[allow(clippy::redundant_closure)] // We get a better caller location for the error by using a closure.
-        let mut connection = TcpStream::connect(kdc_addr.as_addr()).await.map_err(|e| {
-            error!(%kdc_addr, "failed to connect to KDC server");
-            unable_to_reach_kdc_server_err(e)
-        })?;
-
-        trace!("Connected! Forwarding KDC message...");
-
-        connection.write_all(message).await.map_err(
-            HttpError::bad_gateway()
-                .with_msg("unable to send the message to the KDC server")
-                .err(),
-        )?;
-
-        trace!("Reading KDC reply...");
-
-        Ok(read_kdc_reply_message(&mut connection).await.map_err(
-            HttpError::bad_gateway()
-                .with_msg("unable to read KDC reply message")
-                .err(),
-        )?)
-    } else {
-        // We assume that ticket length is not bigger than 2048 bytes.
-        let mut buf = [0; 2048];
-
-        let udp_socket = UdpSocket::bind("127.0.0.1:0")
-            .await
-            .map_err(HttpError::internal().with_msg("unable to bind UDP socket").err())?;
-
-        let port = udp_socket
-            .local_addr()
-            .map_err(HttpError::internal().with_msg("unable to get UDP socket address").err())?
-            .port();
-
-        trace!("Binded UDP listener to 127.0.0.1:{port}, forwarding KDC message...");
-
-        // First 4 bytes contains message length. We don't need it for UDP.
-        #[allow(clippy::redundant_closure)] // We get a better caller location for the error by using a closure.
-        udp_socket
-            .send_to(&message[4..], kdc_addr.as_addr())
-            .await
-            .map_err(|e| unable_to_reach_kdc_server_err(e))?;
-
-        trace!("Reading KDC reply...");
-
-        let n = udp_socket.recv(&mut buf).await.map_err(
-            HttpError::bad_gateway()
-                .with_msg("unable to read reply from the KDC server")
-                .err(),
-        )?;
-
-        let mut reply_buf = Vec::new();
-        reply_buf.extend_from_slice(&u32::try_from(n).expect("n not too big").to_be_bytes());
-        reply_buf.extend_from_slice(&buf[0..n]);
-
-        Ok(reply_buf)
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -288,11 +202,11 @@ mod tests {
 
     #[test]
     fn credential_injection_gate_allows_jet_cred_id_when_enabled() {
-        assert!(enforce_credential_injection_enabled(uuid::Uuid::new_v4(), true).is_ok());
+        assert!(enforce_credential_injection_enabled(Uuid::new_v4(), true).is_ok());
     }
 
     #[test]
     fn credential_injection_gate_rejects_jet_cred_id_when_disabled() {
-        assert!(enforce_credential_injection_enabled(uuid::Uuid::new_v4(), false).is_err());
+        assert!(enforce_credential_injection_enabled(Uuid::new_v4(), false).is_err());
     }
 }

devolutions-gateway/src/api/webapp.rs

@@ -390,6 +390,7 @@ pub(crate) async fn sign_session_token(
                     krb_realm: krb_realm.into(),
                     krb_kdc: krb_kdc.clone(),
                 },
+                jti,
             }
             .pipe(serde_json::to_value)
             .map(|mut claims| {

devolutions-gateway/src/generic_client.rs

@@ -163,6 +163,12 @@ where
                         "RDP-TLS forwarding with credential injection"
                     );
 
+                    let kdc_connector = crate::kdc_connector::KdcConnector::new(
+                        claims.jet_aid,
+                        claims.jet_agent_id,
+                        agent_tunnel_handle.clone(),
+                    );
+
                     // NOTE: In the future, we could imagine performing proxy-based recording as well using RdpProxy.
                     return crate::rdp_proxy::RdpProxy::builder()
                         .conf(conf)
@@ -177,6 +183,7 @@ where
                         .client_stream_leftover_bytes(leftover_bytes)
                         .server_dns_name(selected_target.host().to_owned())
                         .disconnect_interest(disconnect_interest)
+                        .kdc_connector(kdc_connector)
                         .build()
                         .run()
                         .await