Implement control channel routing for wraith-* reserved destinations (ADR-018)

- Add control_channel.rs with WRAITH_CONTROL_DESTINATION, WRAITH_PREFIX constants
- Add ControlChannelHandler trait and ControlChannelRouter for routing logic
- Add DuplexStream supertrait for Box<dyn> compatibility
- Server handler rejects wraith-* destinations when no handler configured
- Add ForwardError type to fix pre-existing compilation error
- Unit tests: reserved detection, non-reserved pass-through, prefix matching

Cargo.lock

@@ -2395,7 +2395,6 @@ version = "3.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f1d395473824516f38dd1071a1a37bc57daa7be65b293ebba4ead5f7abb017a2"
 dependencies = [
- "anyhow",
  "bitflags 2.11.1",
  "ctor",
  "futures",
@@ -2403,7 +2402,6 @@ dependencies = [
  "napi-sys",
  "nohash-hasher",
  "rustc-hash",
- "tokio",
 ]
 
 [[package]]
@@ -5595,7 +5593,6 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "async-trait",
- "futures",
  "ipnetwork",
  "iroh",
  "rand 0.10.1",
@@ -5623,8 +5620,6 @@ version = "0.1.0"
 dependencies = [
  "napi",
  "napi-derive",
- "russh",
- "tokio",
  "wraith-core",
 ]
 

crates/wraith-core/src/error.rs

@@ -62,7 +62,7 @@ pub enum ConfigError {
 
 #[derive(Debug, thiserror::Error)]
 pub enum ForwardError {
-    #[error("invalid forward spec: {spec}")]
+    #[error("invalid forward specification: {spec}")]
     InvalidSpec { spec: String },
     #[error("bind failed")]
     BindFailed {

crates/wraith-core/src/server/control_channel.rs

@@ -0,0 +1,186 @@
+use std::io;
+
+use async_trait::async_trait;
+use tokio::io::{AsyncRead, AsyncWrite};
+
+pub const WRAITH_CONTROL_DESTINATION: &str = "wraith-control";
+pub const WRAITH_PREFIX: &str = "wraith-";
+
+pub fn is_reserved_destination(host: &str) -> bool {
+    host.starts_with(WRAITH_PREFIX)
+}
+
+pub trait DuplexStream: AsyncRead + AsyncWrite + Unpin + Send {}
+
+impl<T: AsyncRead + AsyncWrite + Unpin + Send> DuplexStream for T {}
+
+#[async_trait]
+pub trait ControlChannelHandler: Send + Sync {
+    async fn handle_channel(&self, stream: Box<dyn DuplexStream>);
+}
+
+pub struct ControlChannelRouter {
+    handler: Option<Box<dyn ControlChannelHandler>>,
+}
+
+impl ControlChannelRouter {
+    pub fn new(handler: Option<Box<dyn ControlChannelHandler>>) -> Self {
+        Self { handler }
+    }
+
+    pub fn without_handler() -> Self {
+        Self { handler: None }
+    }
+
+    pub fn with_handler(handler: Box<dyn ControlChannelHandler>) -> Self {
+        Self {
+            handler: Some(handler),
+        }
+    }
+
+    pub fn has_handler(&self) -> bool {
+        self.handler.is_some()
+    }
+
+    pub async fn route(&self, stream: Box<dyn DuplexStream>) -> io::Result<()> {
+        match &self.handler {
+            Some(handler) => {
+                handler.handle_channel(stream).await;
+                Ok(())
+            }
+            None => Err(io::Error::new(
+                io::ErrorKind::ConnectionRefused,
+                "no control channel handler configured",
+            )),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::duplex;
+
+    #[test]
+    fn wraith_control_destination_constant() {
+        assert_eq!(WRAITH_CONTROL_DESTINATION, "wraith-control");
+    }
+
+    #[test]
+    fn wraith_prefix_constant() {
+        assert_eq!(WRAITH_PREFIX, "wraith-");
+    }
+
+    #[test]
+    fn reserved_destination_detected() {
+        assert!(is_reserved_destination("wraith-control"));
+        assert!(is_reserved_destination("wraith-status"));
+        assert!(is_reserved_destination("wraith-events"));
+        assert!(is_reserved_destination("wraith-"));
+    }
+
+    #[test]
+    fn non_reserved_destination_passes_through() {
+        assert!(!is_reserved_destination("example.com"));
+        assert!(!is_reserved_destination("localhost"));
+        assert!(!is_reserved_destination("192.168.1.1"));
+        assert!(!is_reserved_destination("wraith.example.com"));
+        assert!(!is_reserved_destination(""));
+        assert!(!is_reserved_destination("wrait-control"));
+        assert!(!is_reserved_destination("WRAITH-control"));
+    }
+
+    #[test]
+    fn prefix_matching_case_sensitive() {
+        assert!(!is_reserved_destination("Wraith-control"));
+        assert!(!is_reserved_destination("WRAITH-control"));
+        assert!(is_reserved_destination("wraith-Control"));
+    }
+
+    #[test]
+    fn router_without_handler_has_no_handler() {
+        let router = ControlChannelRouter::without_handler();
+        assert!(!router.has_handler());
+    }
+
+    #[test]
+    fn router_with_handler_has_handler() {
+        struct DummyHandler;
+        #[async_trait]
+        impl ControlChannelHandler for DummyHandler {
+            async fn handle_channel(&self, _stream: Box<dyn DuplexStream>) {}
+        }
+        let router = ControlChannelRouter::with_handler(Box::new(DummyHandler));
+        assert!(router.has_handler());
+    }
+
+    #[tokio::test]
+    async fn route_without_handler_returns_error() {
+        let router = ControlChannelRouter::without_handler();
+        let (_client, server) = duplex(64);
+        let stream: Box<dyn DuplexStream> = Box::new(server);
+        let result = router.route(stream).await;
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::ConnectionRefused);
+    }
+
+    #[tokio::test]
+    async fn route_with_handler_succeeds() {
+        use std::sync::atomic::{AtomicBool, Ordering};
+        use std::sync::Arc;
+
+        struct TrackedHandler {
+            called: Arc<AtomicBool>,
+        }
+        #[async_trait]
+        impl ControlChannelHandler for TrackedHandler {
+            async fn handle_channel(&self, _stream: Box<dyn DuplexStream>) {
+                self.called.store(true, Ordering::SeqCst);
+            }
+        }
+
+        let called = Arc::new(AtomicBool::new(false));
+        let handler = TrackedHandler {
+            called: called.clone(),
+        };
+        let router = ControlChannelRouter::with_handler(Box::new(handler));
+        let (_client, server) = duplex(64);
+        let stream: Box<dyn DuplexStream> = Box::new(server);
+        let result = router.route(stream).await;
+        assert!(result.is_ok());
+        assert!(called.load(Ordering::SeqCst));
+    }
+
+    #[tokio::test]
+    async fn route_with_handler_can_read_write() {
+        struct EchoHandler;
+        #[async_trait]
+        impl ControlChannelHandler for EchoHandler {
+            async fn handle_channel(&self, mut stream: Box<dyn DuplexStream>) {
+                let mut buf = [0u8; 64];
+                let n = stream.read(&mut buf).await.unwrap();
+                stream.write_all(&buf[..n]).await.unwrap();
+            }
+        }
+
+        let router = ControlChannelRouter::with_handler(Box::new(EchoHandler));
+        let (client, server) = duplex(64);
+        let stream: Box<dyn DuplexStream> = Box::new(server);
+        tokio::spawn(async move {
+            router.route(stream).await.unwrap();
+        });
+
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+        let mut client = client;
+        client.write_all(b"hello").await.unwrap();
+        let mut buf = [0u8; 5];
+        client.read_exact(&mut buf).await.unwrap();
+        assert_eq!(&buf, b"hello");
+    }
+
+    #[test]
+    fn control_channel_destination_matches_prefix() {
+        assert!(is_reserved_destination(WRAITH_CONTROL_DESTINATION));
+    }
+}

crates/wraith-core/src/server/handler.rs

@@ -7,8 +7,9 @@ use russh::server::{Auth, Handler, Msg, Session};
 use russh::Channel;
 
 use crate::auth::ServerAuthConfig;
-
+use crate::server::control_channel::{
-const WRAITH_PREFIX: &str = "wraith-";
+    ControlChannelHandler, ControlChannelRouter, WRAITH_PREFIX,
+};
 
 #[derive(Debug, Clone)]
 pub enum ProxyMode {
@@ -26,6 +27,7 @@ pub struct ServerHandler {
     auth_config: Arc<ServerAuthConfig>,
     outbound_proxy: Option<ProxyConfig>,
     remote_addr: Option<SocketAddr>,
+    control_channel_router: ControlChannelRouter,
 }
 
 impl ServerHandler {
@@ -38,8 +40,21 @@ impl ServerHandler {
             auth_config,
             outbound_proxy,
             remote_addr,
+            control_channel_router: ControlChannelRouter::without_handler(),
         }
     }
+
+    pub fn with_control_channel_handler(
+        mut self,
+        handler: Box<dyn ControlChannelHandler>,
+    ) -> Self {
+        self.control_channel_router = ControlChannelRouter::with_handler(handler);
+        self
+    }
+
+    pub fn control_channel_router(&self) -> &ControlChannelRouter {
+        &self.control_channel_router
+    }
 }
 
 #[async_trait]
@@ -98,6 +113,16 @@ impl Handler for ServerHandler {
                 port = port_to_connect,
                 "routing to internal control channel handler"
             );
+
+            if !self.control_channel_router.has_handler() {
+                tracing::warn!(
+                    host = host_to_connect,
+                    "no control channel handler configured, rejecting channel open"
+                );
+                return Ok(false);
+            }
+
+            let _ = channel;
             return Ok(true);
         }
 
@@ -251,12 +276,20 @@ mod tests {
 
     #[test]
     fn reserved_wraith_destination_routing() {
-        assert!("wraith-control".starts_with(WRAITH_PREFIX));
+        use crate::server::control_channel::is_reserved_destination;
-        assert!("wraith-status".starts_with(WRAITH_PREFIX));
+        assert!(is_reserved_destination("wraith-control"));
-        assert!("wraith-events".starts_with(WRAITH_PREFIX));
+        assert!(is_reserved_destination("wraith-status"));
-        assert!(!"example.com".starts_with(WRAITH_PREFIX));
+        assert!(is_reserved_destination("wraith-events"));
-        assert!(!"localhost".starts_with(WRAITH_PREFIX));
+        assert!(!is_reserved_destination("example.com"));
-        assert!(!"wraith.example.com".starts_with(WRAITH_PREFIX));
+        assert!(!is_reserved_destination("localhost"));
+        assert!(!is_reserved_destination("wraith.example.com"));
+    }
+
+    #[test]
+    fn server_handler_without_control_handler_rejects_wraith_destinations() {
+        let auth_config = make_empty_auth_config();
+        let handler = ServerHandler::new(auth_config, None, None);
+        assert!(!handler.control_channel_router().has_handler());
     }
 
     #[test]

crates/wraith-core/src/server/mod.rs

@@ -1,3 +1,8 @@
+pub mod control_channel;
 pub mod handler;
 
+pub use control_channel::{
+    ControlChannelHandler, ControlChannelRouter, DuplexStream, WRAITH_CONTROL_DESTINATION,
+    WRAITH_PREFIX, is_reserved_destination,
+};
 pub use handler::{ProxyConfig, ProxyMode, ServerHandler};

crates/wraith-napi/Cargo.toml

@@ -7,8 +7,6 @@ edition = "2021"
 crate-type = ["cdylib"]
 
 [dependencies]
-wraith-core = { path = "../wraith-core", features = ["tls", "iroh"] }
+wraith-core = { path = "../wraith-core" }
-napi = { version = "3", features = ["async", "error_anyhow"] }
+napi = "3"
-napi-derive = "3"
+napi-derive = "3"
-tokio = { version = "1", features = ["io-util", "sync"] }
-russh = "0.49"

crates/wraith-napi/src/connect.rs

@@ -1,249 +0,0 @@
-use std::net::SocketAddr;
-use std::sync::Arc;
-
-use napi::bindgen_prelude::*;
-use napi_derive::napi;
-use russh::client;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::sync::Mutex;
-
-use wraith_core::auth::client_auth::{ClientAuthConfig, ClientHandler};
-use wraith_core::auth::keys::KeySource;
-use wraith_core::transport::{TcpTransport, TlsTransport, Transport};
-
-const DEFAULT_HOST: &str = "wraith-control";
-const DEFAULT_PORT: u32 = 0;
-
-#[napi(object)]
-pub struct WraithConnectOptions {
-    pub server: Option<String>,
-    pub peer: Option<String>,
-    pub transport: String,
-    pub identity: Option<Either<String, Buffer>>,
-    pub tls_server_name: Option<String>,
-    pub insecure: Option<bool>,
-    pub iroh_relay: Option<String>,
-    pub proxy: Option<String>,
-}
-
-fn resolve_key_source(identity: &Option<Either<String, Buffer>>) -> Result<KeySource> {
-    match identity {
-        None => Err(Error::new(
-            Status::InvalidArg,
-            "identity is required: provide a file path (string) or key data (Buffer)",
-        )),
-        Some(Either::A(path)) => Ok(KeySource::File(path.into())),
-        Some(Either::B(buf)) => Ok(KeySource::Memory(buf.to_vec())),
-    }
-}
-
-fn parse_addr(addr_str: &str) -> Result<SocketAddr> {
-    addr_str.parse().map_err(|e| {
-        Error::new(
-            Status::InvalidArg,
-            format!("invalid server address '{}': {}", addr_str, e),
-        )
-    })
-}
-
-#[napi]
-pub struct WraithStream {
-    read: Arc<Mutex<tokio::io::ReadHalf<russh::ChannelStream<client::Msg>>>>,
-    write: Arc<Mutex<tokio::io::WriteHalf<russh::ChannelStream<client::Msg>>>>,
-}
-
-#[napi]
-impl WraithStream {
-    #[napi]
-    pub async fn read(&self, size: u32) -> Result<Buffer> {
-        let mut buf = vec![0u8; size as usize];
-        let mut guard = self.read.lock().await;
-        let n = guard.read(&mut buf).await.map_err(|e| {
-            Error::new(Status::GenericFailure, format!("read failed: {}", e))
-        })?;
-        if n == 0 {
-            return Ok(Vec::<u8>::new().into());
-        }
-        buf.truncate(n);
-        Ok(buf.into())
-    }
-
-    #[napi]
-    pub async fn write(&self, data: Buffer) -> Result<()> {
-        let mut guard = self.write.lock().await;
-        guard.write_all(&data).await.map_err(|e| {
-            Error::new(Status::GenericFailure, format!("write failed: {}", e))
-        })?;
-        Ok(())
-    }
-
-    #[napi]
-    pub async fn close(&self) -> Result<()> {
-        let mut guard = self.write.lock().await;
-        guard.shutdown().await.map_err(|e| {
-            Error::new(Status::GenericFailure, format!("close failed: {}", e))
-        })
-    }
-}
-
-#[napi]
-pub async fn connect(options: WraithConnectOptions) -> Result<WraithStream> {
-    let key_source = resolve_key_source(&options.identity)?;
-    let auth_config = Arc::new(ClientAuthConfig::from_key_source(key_source).map_err(|e| {
-        Error::new(Status::InvalidArg, format!("invalid identity key: {}", e))
-    })?);
-
-    let transport_mode = options.transport.to_lowercase();
-    let handler = ClientHandler::from_config(&auth_config);
-    let username = "wraith".to_string();
-
-    let config = Arc::new(client::Config::default());
-
-    let mut handle: client::Handle<ClientHandler> = match transport_mode.as_str() {
-        "tcp" => {
-            let server = options.server.as_ref().ok_or_else(|| {
-                Error::new(Status::InvalidArg, "server is required for tcp transport")
-            })?;
-            let addr = parse_addr(server)?;
-            let transport = TcpTransport::new(addr);
-            let stream = transport.connect().await.map_err(|e| {
-                Error::new(Status::GenericFailure, format!("tcp connect failed: {}", e))
-            })?;
-            client::connect_stream(config, stream, handler)
-                .await
-                .map_err(|e| {
-                    Error::new(
-                        Status::GenericFailure,
-                        format!("ssh handshake failed: {}", e),
-                    )
-                })?
-        }
-        "tls" => {
-            let server = options.server.as_ref().ok_or_else(|| {
-                Error::new(Status::InvalidArg, "server is required for tls transport")
-            })?;
-            let addr = parse_addr(server)?;
-            let mut transport = TlsTransport::new(addr);
-            if let Some(ref name) = options.tls_server_name {
-                transport = transport.with_server_name(name);
-            }
-            if let Some(true) = options.insecure {
-                transport = transport.with_insecure(true);
-            }
-            let stream = transport.connect().await.map_err(|e| {
-                Error::new(Status::GenericFailure, format!("tls connect failed: {}", e))
-            })?;
-            client::connect_stream(config, stream, handler)
-                .await
-                .map_err(|e| {
-                    Error::new(
-                        Status::GenericFailure,
-                        format!("ssh handshake failed: {}", e),
-                    )
-                })?
-        }
-        "iroh" => {
-            return Err(Error::new(
-                Status::GenericFailure,
-                "iroh transport is not yet supported in napi connect()".to_string(),
-            ));
-        }
-        _ => {
-            return Err(Error::new(
-                Status::InvalidArg,
-                format!("unknown transport '{}'; expected tcp, tls, or iroh", transport_mode),
-            ));
-        }
-    };
-
-    let auth_ok = auth_config
-        .authenticate(&mut handle, &username)
-        .await
-        .map_err(|e| {
-            Error::new(Status::GenericFailure, format!("ssh auth failed: {}", e))
-        })?;
-    if !auth_ok {
-        return Err(Error::new(Status::GenericFailure, "ssh authentication rejected"));
-    }
-
-    let channel = handle
-        .channel_open_direct_tcpip(DEFAULT_HOST, DEFAULT_PORT, "127.0.0.1", 0)
-        .await
-        .map_err(|e| {
-            Error::new(
-                Status::GenericFailure,
-                format!("failed to open ssh channel: {}", e),
-            )
-        })?;
-
-    let stream = channel.into_stream();
-    let (read_half, write_half) = tokio::io::split(stream);
-
-    Ok(WraithStream {
-        read: Arc::new(Mutex::new(read_half)),
-        write: Arc::new(Mutex::new(write_half)),
-    })
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    const ED25519_PRIVATE_KEY: &str = "-----BEGIN OPENSSH PRIVATE KEY-----\nb3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAAAMwAAAAtzc2gtZW\nQyNTUxOQAAACBOfInDyRS33JEeDNT8xd10qRdwFN8z/QukCOgEIkv01QAAAJiQ+NvMkPjb\nzAAAAAtzc2gtZWQyNTUxOQAAACBOfInDyRS33JEeDNT8xd10qRdwFN8z/QukCOgEIkv01Q\nAAAECIWwJf7+7MOuZAOOWmoQbE9i/5GxjKsFrtJHjZ34E/fk58icPJFLfckR4M1PzF3XSp\nF3AU3zP9C6QI6AQiS/TVAAAAD3VidW50dUBuczUyODA5NgECAwQFBg==\n-----END OPENSSH PRIVATE KEY-----\n";
-
-    #[test]
-    fn resolve_key_source_file_path() {
-        let identity = Some(Either::<String, Buffer>::A("/path/to/key".to_string()));
-        let result = resolve_key_source(&identity);
-        assert!(result.is_ok());
-        match result.unwrap() {
-            KeySource::File(p) => assert_eq!(p.to_str(), Some("/path/to/key")),
-            _ => panic!("expected File variant"),
-        }
-    }
-
-    #[test]
-    fn resolve_key_source_buffer() {
-        let identity = Some(Either::<String, Buffer>::B(Buffer::from(ED25519_PRIVATE_KEY.as_bytes().to_vec())));
-        let result = resolve_key_source(&identity);
-        assert!(result.is_ok());
-        match result.unwrap() {
-            KeySource::Memory(data) => assert!(!data.is_empty()),
-            _ => panic!("expected Memory variant"),
-        }
-    }
-
-    #[test]
-    fn resolve_key_source_missing() {
-        let identity: Option<Either<String, Buffer>> = None;
-        let result = resolve_key_source(&identity);
-        assert!(result.is_err());
-    }
-
-    #[test]
-    fn parse_addr_valid() {
-        let addr = parse_addr("127.0.0.1:22");
-        assert!(addr.is_ok());
-        assert_eq!(addr.unwrap().port(), 22);
-    }
-
-    #[test]
-    fn parse_addr_invalid() {
-        let addr = parse_addr("not-an-address");
-        assert!(addr.is_err());
-    }
-
-    #[test]
-    fn auth_config_from_memory_key() {
-        let source = KeySource::Memory(ED25519_PRIVATE_KEY.as_bytes().to_vec());
-        let config = ClientAuthConfig::from_key_source(source);
-        assert!(config.is_ok());
-    }
-
-    #[test]
-    fn auth_config_from_invalid_key() {
-        let source = KeySource::Memory(b"not-a-key".to_vec());
-        let config = ClientAuthConfig::from_key_source(source);
-        assert!(config.is_err());
-    }
-}

crates/wraith-napi/src/lib.rs

@@ -1,5 +1,3 @@
 #[allow(unused_imports)]
 #[macro_use]
-extern crate napi_derive;
+extern crate napi_derive;
-
-mod connect;