Add support for Windows Named Pipes

.github/workflows/ci.yml

@@ -22,23 +22,41 @@ jobs:
 
   formatting:
     name: Check formatting
-    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        include:
+        - os: ubuntu-latest
+        - os: macos-latest
+        - os: windows-latest
+    runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v4
       - name: Check formatting
         run: cargo fmt --all -- --check
 
   tests:
     name: Unit tests
-    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        include:
+        - os: ubuntu-latest
+        - os: macos-latest
+        - os: windows-latest
+    runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v4
       - name: Build and test
         run: cargo build --verbose --all && cargo test --verbose --all
 
   lints:
     name: Clippy lints
-    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        include:
+        - os: ubuntu-latest
+        - os: macos-latest
+        - os: windows-latest
+    runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v4
       - name: Check for lints

Cargo.toml

@@ -24,7 +24,7 @@ futures = { version = "0.3.30", optional = true }
 log = { version = "0.4.6", optional = true }
 tokio = { version = "1", optional = true, features = ["rt", "net"] }
 tokio-util = { version = "0.7.1", optional = true, features = ["codec"] }
-service-binding = "1.1.0"
+service-binding = { version = "^2" }
 
 [features]
 default = ["agent"]

README.md

@@ -9,10 +9,14 @@ This makes it possible to utilize remote keys not supported by the default OpenS
 
 ## Example
 
-This example starts listening on a Unix socket `ssh-agent.sock` and processes requests.
+The following example starts listening on a socket and processing requests.
+On Unix it uses `ssh-agent.sock` Unix domain socket while on Windows it uses a named pipe `\\.\pipe\agent`.
 
 ```rust,no_run
+#[cfg(not(windows))]
 use tokio::net::UnixListener;
+#[cfg(windows)]
+use ssh_agent_lib::agent::NamedPipeListener;
 
 use ssh_agent_lib::agent::{Session, Agent};
 use ssh_agent_lib::proto::message::Message;
@@ -35,13 +39,21 @@ impl Session for MyAgent {
 }
 
 #[tokio::main]
+#[cfg(not(windows))]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let socket = "ssh-agent.sock";
     let _ = std::fs::remove_file(socket); // remove the socket if exists
 
     MyAgent.listen(UnixListener::bind(socket)?).await?;
     Ok(())
 }
+
+#[tokio::main]
+#[cfg(windows)]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    MyAgent.listen(NamedPipeListener::new(r"\\.\pipe\agent")?).await?;
+    Ok(())
+}
 ```
 
 Now, point your OpenSSH client to this socket using `SSH_AUTH_SOCK` environment variable and it will transparently use the agent:
@@ -50,6 +62,12 @@ Now, point your OpenSSH client to this socket using `SSH_AUTH_SOCK` environment
 SSH_AUTH_SOCK=ssh-agent.sock ssh [email protected]
 ```
 
+On Windows the path of the pipe has to be used:
+
+```sh
+SSH_AUTH_SOCK=\\.\pipe\agent ssh [email protected]
+```
+
 For more elaborate example see the `examples` directory or [crates using `ssh-agent-lib`](https://crates.io/crates/ssh-agent-lib/reverse_dependencies).
 
 ## Note

src/agent.rs

@@ -4,7 +4,11 @@ use bytes::{Buf, BufMut, BytesMut};
 use futures::{SinkExt, TryStreamExt};
 use log::{error, info};
 use tokio::io::{AsyncRead, AsyncWrite};
-use tokio::net::{TcpListener, TcpStream, UnixListener, UnixStream};
+#[cfg(windows)]
+use tokio::net::windows::named_pipe::{NamedPipeServer, ServerOptions};
+use tokio::net::{TcpListener, TcpStream};
+#[cfg(unix)]
+use tokio::net::{UnixListener, UnixStream};
 use tokio_util::codec::{Decoder, Encoder, Framed};
 
 use std::fmt;
@@ -56,25 +60,55 @@ impl Encoder<Message> for MessageCodec {
 pub trait ListeningSocket {
     type Stream: fmt::Debug + AsyncRead + AsyncWrite + Send + Unpin + 'static;
 
-    async fn accept(&self) -> io::Result<Self::Stream>;
+    async fn accept(&mut self) -> io::Result<Self::Stream>;
 }
 
+#[cfg(unix)]
 #[async_trait]
 impl ListeningSocket for UnixListener {
     type Stream = UnixStream;
-    async fn accept(&self) -> io::Result<Self::Stream> {
+    async fn accept(&mut self) -> io::Result<Self::Stream> {
         UnixListener::accept(self).await.map(|(s, _addr)| s)
     }
 }
 
 #[async_trait]
 impl ListeningSocket for TcpListener {
     type Stream = TcpStream;
-    async fn accept(&self) -> io::Result<Self::Stream> {
+    async fn accept(&mut self) -> io::Result<Self::Stream> {
         TcpListener::accept(self).await.map(|(s, _addr)| s)
     }
 }
 
+#[cfg(windows)]
+#[derive(Debug)]
+pub struct NamedPipeListener(NamedPipeServer, std::ffi::OsString);
+
+#[cfg(windows)]
+impl NamedPipeListener {
+    pub fn new(pipe: std::ffi::OsString) -> std::io::Result<Self> {
+        Ok(NamedPipeListener(
+            ServerOptions::new()
+                .first_pipe_instance(true)
+                .create(&pipe)?,
+            pipe,
+        ))
+    }
+}
+
+#[cfg(windows)]
+#[async_trait]
+impl ListeningSocket for NamedPipeListener {
+    type Stream = NamedPipeServer;
+    async fn accept(&mut self) -> io::Result<Self::Stream> {
+        self.0.connect().await?;
+        Ok(std::mem::replace(
+            &mut self.0,
+            ServerOptions::new().create(&self.1)?,
+        ))
+    }
+}
+
 #[async_trait]
 pub trait Session: 'static + Sync + Send + Sized {
     async fn handle(&mut self, message: Message) -> Result<Message, Box<dyn std::error::Error>>;
@@ -109,7 +143,7 @@ pub trait Session: 'static + Sync + Send + Sized {
 #[async_trait]
 pub trait Agent: 'static + Sync + Send + Sized {
     fn new_session(&mut self) -> impl Session;
-    async fn listen<S>(mut self, socket: S) -> Result<(), AgentError>
+    async fn listen<S>(mut self, mut socket: S) -> Result<(), AgentError>
     where
         S: ListeningSocket + fmt::Debug + Send,
     {
@@ -134,12 +168,21 @@ pub trait Agent: 'static + Sync + Send + Sized {
     }
     async fn bind(mut self, listener: service_binding::Listener) -> Result<(), AgentError> {
         match listener {
+            #[cfg(unix)]
             service_binding::Listener::Unix(listener) => {
                 self.listen(UnixListener::from_std(listener)?).await
             }
             service_binding::Listener::Tcp(listener) => {
                 self.listen(TcpListener::from_std(listener)?).await
             }
+            #[cfg(windows)]
+            service_binding::Listener::NamedPipe(pipe) => {
+                self.listen(NamedPipeListener::new(pipe)?).await
+            }
+            #[cfg(not(windows))]
+            service_binding::Listener::NamedPipe(_) => Err(AgentError::IO(std::io::Error::other(
+                "Named pipes supported on Windows only",
+            ))),
         }
     }
 }