identity(spire): provide named pipe support for spire (#3970)

This change does two things: 
- adds support for `NamedPipes` to our SPIRE client. This will allow the client to connect to spire agents running on Windows hosts
- renames the `LINKERD2_PROXY_IDENTITY_SPIRE_SOCKET` to `LINKERD2_PROXY_IDENTITY_SPIRE_WORKLOAD_API_ADDRESS` and deprecates the former.

Signed-off-by: Zahari Dichev <zaharidichev@gmail.com>

Author: zaharidichev

linkerd/app/src/env.rs

@@ -233,8 +233,12 @@ pub const ENV_IDENTITY_IDENTITY_SERVER_ID: &str = "LINKERD2_PROXY_IDENTITY_SERVE
 pub const ENV_IDENTITY_IDENTITY_SERVER_NAME: &str = "LINKERD2_PROXY_IDENTITY_SERVER_NAME";
 
 // If this config is set, then the proxy will be configured to use Spire as identity
-// provider
+// provider. On Unix systems this needs to be a path to a UDS while on Windows - a
+// named pipe path.
+pub const ENV_IDENTITY_SPIRE_WORKLOAD_API_ADDRESS: &str =
+    "LINKERD2_PROXY_IDENTITY_SPIRE_WORKLOAD_API_ADDRESS";
 pub const ENV_IDENTITY_SPIRE_SOCKET: &str = "LINKERD2_PROXY_IDENTITY_SPIRE_SOCKET";
+
 pub const IDENTITY_SPIRE_BASE: &str = "LINKERD2_PROXY_IDENTITY_SPIRE";
 const DEFAULT_SPIRE_BACKOFF: ExponentialBackoff =
     ExponentialBackoff::new_unchecked(Duration::from_millis(100), Duration::from_secs(1), 0.1);
@@ -909,8 +913,13 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
     let identity = {
         let tls = tls?;
 
-        match strings.get(ENV_IDENTITY_SPIRE_SOCKET)? {
-            Some(socket) => match &tls.id {
+        match parse_deprecated(
+            strings,
+            ENV_IDENTITY_SPIRE_WORKLOAD_API_ADDRESS,
+            ENV_IDENTITY_SPIRE_SOCKET,
+            |s| Ok(s.to_string()),
+        )? {
+            Some(workload_api_addr) => match &tls.id {
                 // TODO: perform stricter SPIFFE ID validation following:
                 // https://github.com/spiffe/spiffe/blob/27b59b81ba8c56885ac5d4be73b35b9b3305fd7a/standards/SPIFFE-ID.md
                 identity::Id::Uri(uri)
@@ -919,7 +928,7 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
                     identity::Config::Spire {
                         tls,
                         client: spire::Config {
-                            socket_addr: std::sync::Arc::new(socket),
+                            workload_api_addr: std::sync::Arc::new(workload_api_addr),
                             backoff: parse_backoff(
                                 strings,
                                 IDENTITY_SPIRE_BASE,

linkerd/app/src/identity.rs

@@ -109,7 +109,7 @@ impl Config {
                 }
             }
             Self::Spire { client, tls } => {
-                let addr = client.socket_addr.clone();
+                let addr = client.workload_api_addr.clone();
                 let spire = spire::client::Spire::new(tls.id.clone());
 
                 let (store, receiver, ready) = watch(tls, metrics.cert)?;

linkerd/app/src/spire.rs

@@ -4,40 +4,26 @@ use tokio::sync::watch;
 
 pub use linkerd_app_core::identity::client::spire as client;
 
-#[cfg(target_os = "linux")]
-const UNIX_PREFIX: &str = "unix:";
-#[cfg(target_os = "linux")]
 const TONIC_DEFAULT_URI: &str = "http://[::]:50051";
 
 #[derive(Clone, Debug)]
 pub struct Config {
-    pub socket_addr: Arc<String>,
+    pub workload_api_addr: Arc<String>,
     pub backoff: ExponentialBackoff,
 }
 
 // Connects to SPIRE workload API via Unix Domain Socket
 pub struct Client {
-    #[cfg_attr(not(target_os = "linux"), allow(dead_code))]
     config: Config,
 }
 
 // === impl Client ===
-
-#[cfg(target_os = "linux")]
 impl From<Config> for Client {
     fn from(config: Config) -> Self {
         Self { config }
     }
 }
 
-#[cfg(not(target_os = "linux"))]
-impl From<Config> for Client {
-    fn from(_: Config) -> Self {
-        panic!("Spire is supported on Linux only")
-    }
-}
-
-#[cfg(target_os = "linux")]
 impl tower::Service<()> for Client {
     type Response = tonic::Response<watch::Receiver<client::SvidUpdate>>;
     type Error = Error;
@@ -51,25 +37,43 @@ impl tower::Service<()> for Client {
     }
 
     fn call(&mut self, _req: ()) -> Self::Future {
-        let socket = self.config.socket_addr.clone();
+        let addr = self.config.workload_api_addr.clone();
         let backoff = self.config.backoff;
         Box::pin(async move {
-            use tokio::net::UnixStream;
             use tonic::transport::{Endpoint, Uri};
 
-            // Strip the 'unix:' prefix for tonic compatibility.
-            let stripped_path = socket
-                .strip_prefix(UNIX_PREFIX)
-                .unwrap_or(socket.as_str())
-                .to_string();
-
             // We will ignore this uri because uds do not use it
             // if your connector does use the uri it will be provided
             // as the request to the `MakeConnection`.
             let chan = Endpoint::try_from(TONIC_DEFAULT_URI)?
                 .connect_with_connector(tower::util::service_fn(move |_: Uri| {
-                    use futures::TryFutureExt;
-                    UnixStream::connect(stripped_path.clone()).map_ok(hyper_util::rt::TokioIo::new)
+                    #[cfg(unix)]
+                    {
+                        use futures::TryFutureExt;
+
+                        // The 'unix:' scheme must be stripped from socket paths.
+                        let path = addr.strip_prefix("unix:").unwrap_or(addr.as_str());
+
+                        tokio::net::UnixStream::connect(path.to_string())
+                            .map_ok(hyper_util::rt::TokioIo::new)
+                    }
+
+                    #[cfg(windows)]
+                    {
+                        use tokio::net::windows::named_pipe;
+                        let named_pipe_path = addr.clone();
+                        let client = named_pipe::ClientOptions::new()
+                            .open(named_pipe_path.as_str())
+                            .map(hyper_util::rt::TokioIo::new);
+
+                        futures::future::ready(client)
+                    }
+
+                    #[cfg(not(any(unix, windows)))]
+                    {
+                        compile_error!("Spire is supported only on Windows and Unix systems.");
+                        futures::future::pending()
+                    }
                 }))
                 .await?;
 
@@ -80,21 +84,3 @@ impl tower::Service<()> for Client {
         })
     }
 }
-
-#[cfg(not(target_os = "linux"))]
-impl tower::Service<()> for Client {
-    type Response = tonic::Response<watch::Receiver<client::SvidUpdate>>;
-    type Error = Error;
-    type Future = futures::future::BoxFuture<'static, Result<Self::Response, Self::Error>>;
-
-    fn poll_ready(
-        &mut self,
-        _cx: &mut std::task::Context<'_>,
-    ) -> std::task::Poll<Result<(), Self::Error>> {
-        unimplemented!("Spire is supported on Linux only")
-    }
-
-    fn call(&mut self, _req: ()) -> Self::Future {
-        unimplemented!("Spire is supported on Linux only")
-    }
-}