feat(unstable): support named pipes on Windows for the permission broker

cli/main.rs

@@ -562,20 +562,17 @@ pub(crate) fn unstable_exit_cb(feature: &str, api_name: &str) {
   deno_runtime::exit(70);
 }
 
-#[cfg(not(unix))]
-fn maybe_setup_permission_broker() {}
-
-#[cfg(unix)]
 fn maybe_setup_permission_broker() {
-  if let Ok(socket_path) = std::env::var("DENO_PERMISSION_BROKER_PATH") {
-    log::warn!(
-      "{} Permission broker is an experimental feature",
-      colors::yellow("Warning")
-    );
-    let broker =
-      deno_runtime::deno_permissions::PermissionBroker::new(socket_path);
-    deno_runtime::deno_permissions::set_broker(broker);
-  }
+  let Ok(socket_path) = std::env::var("DENO_PERMISSION_BROKER_PATH") else {
+    return;
+  };
+  log::warn!(
+    "{} Permission broker is an experimental feature",
+    colors::yellow("Warning")
+  );
+  let broker =
+    deno_runtime::deno_permissions::broker::PermissionBroker::new(socket_path);
+  deno_runtime::deno_permissions::broker::set_broker(broker);
 }
 
 pub fn main() {
@@ -593,7 +590,7 @@ pub fn main() {
     deno_subprocess_windows::disable_stdio_inheritance();
     colors::enable_ansi(); // For Windows 10
   }
-  deno_runtime::deno_permissions::set_prompt_callbacks(
+  deno_runtime::deno_permissions::prompter::set_prompt_callbacks(
     Box::new(util::draw_thread::DrawThread::hide),
     Box::new(util::draw_thread::DrawThread::show),
   );

runtime/permissions/broker.rs

@@ -3,14 +3,14 @@
 use std::io::BufRead;
 use std::io::BufReader;
 use std::io::Write;
-use std::os::unix::net::UnixStream;
 use std::path::PathBuf;
 use std::sync::OnceLock;
 use std::sync::atomic::AtomicU32;
 
 use parking_lot::Mutex;
 
 use super::BrokerResponse;
+use crate::ipc_pipe::LocalStream;
 
 // TODO(bartlomieju): currently randomly selected exit code, it should
 // be documented
@@ -42,13 +42,14 @@ struct PermissionBrokerResponse {
 }
 
 pub struct PermissionBroker {
-  stream: Mutex<UnixStream>,
+  stream: Mutex<LocalStream>,
   next_id: AtomicU32,
 }
 
 impl PermissionBroker {
   pub fn new(socket_path: impl Into<PathBuf>) -> Self {
-    let stream = match UnixStream::connect(socket_path.into()) {
+    let socket_path = socket_path.into();
+    let stream = match LocalStream::connect(&socket_path) {
       Ok(s) => s,
       Err(err) => {
         log::error!("Failed to create permission broker: {:?}", err);
@@ -57,7 +58,7 @@ impl PermissionBroker {
     };
     Self {
       stream: Mutex::new(stream),
-      next_id: AtomicU32::new(1),
+      next_id: std::sync::atomic::AtomicU32::new(1),
     }
   }
 

runtime/permissions/ipc_pipe.rs

@@ -0,0 +1,126 @@
+// Copyright 2018-2025 the Deno authors. MIT license.
+
+use std::ffi::OsStr;
+use std::io::Read;
+use std::io::Write;
+use std::io::{self};
+
+pub struct LocalStream(Inner);
+
+#[cfg(unix)]
+type Inner = std::os::unix::net::UnixStream;
+
+#[cfg(not(unix))]
+type Inner = std::fs::File;
+
+impl LocalStream {
+  /// Connect to a local IPC endpoint.
+  /// - Unix: `addr` like `/tmp/deno.sock`
+  /// - Windows: `addr` like `\\.\pipe\deno-permission-broker`
+  pub fn connect(addr: impl AsRef<OsStr>) -> io::Result<Self> {
+    Self::connect_impl(addr.as_ref())
+  }
+}
+
+#[cfg(unix)]
+impl LocalStream {
+  fn connect_impl(addr: &OsStr) -> io::Result<Self> {
+    use std::os::unix::net::UnixStream;
+    use std::path::Path;
+    let s = UnixStream::connect(Path::new(addr))?;
+    s.set_nonblocking(false)?;
+    Ok(Self(s))
+  }
+}
+
+#[cfg(windows)]
+impl LocalStream {
+  fn connect_impl(addr: &OsStr) -> io::Result<Self> {
+    use std::os::windows::ffi::OsStrExt;
+    use std::os::windows::io::FromRawHandle;
+
+    use windows_sys::Win32::Foundation::ERROR_PIPE_BUSY;
+    use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE;
+    use windows_sys::Win32::Storage::FileSystem::CreateFileW;
+    use windows_sys::Win32::Storage::FileSystem::FILE_ATTRIBUTE_NORMAL;
+    use windows_sys::Win32::Storage::FileSystem::FILE_GENERIC_READ;
+    use windows_sys::Win32::Storage::FileSystem::FILE_GENERIC_WRITE;
+    use windows_sys::Win32::Storage::FileSystem::OPEN_EXISTING;
+    use windows_sys::Win32::System::Pipes::NMPWAIT_WAIT_FOREVER;
+    use windows_sys::Win32::System::Pipes::PIPE_READMODE_BYTE;
+    use windows_sys::Win32::System::Pipes::SetNamedPipeHandleState;
+    use windows_sys::Win32::System::Pipes::WaitNamedPipeW;
+
+    // OsStr -> UTF-16 + NUL
+    let mut wide: Vec<u16> = addr.encode_wide().collect();
+    wide.push(0);
+
+    // Try to open; if the pipe is busy, wait and retry.
+    let handle = loop {
+      // SAFETY: WinAPI call
+      let h = unsafe {
+        CreateFileW(
+          wide.as_ptr(),
+          FILE_GENERIC_READ | FILE_GENERIC_WRITE,
+          0, // no sharing
+          std::ptr::null(),
+          OPEN_EXISTING,
+          FILE_ATTRIBUTE_NORMAL, // blocking
+          std::ptr::null_mut(),
+        )
+      };
+      if h != INVALID_HANDLE_VALUE {
+        break h;
+      }
+      let err = io::Error::last_os_error();
+      if err.raw_os_error() == Some(ERROR_PIPE_BUSY as i32) {
+        // SAFETY: WinAPI call
+        unsafe { WaitNamedPipeW(wide.as_ptr(), NMPWAIT_WAIT_FOREVER) };
+        continue;
+      } else {
+        return Err(err);
+      }
+    };
+
+    // Ensure byte mode to mirror Unix stream semantics.
+    // SAFETY: WinAPI call
+    unsafe {
+      let _ = SetNamedPipeHandleState(
+        handle,
+        &PIPE_READMODE_BYTE,
+        std::ptr::null_mut(),
+        std::ptr::null_mut(),
+      );
+    }
+
+    // SAFETY: Passing WinAPI handle
+    let file = unsafe { std::fs::File::from_raw_handle(handle as _) };
+    Ok(Self(file))
+  }
+}
+
+#[cfg(all(not(unix), not(windows)))]
+impl LocalStream {
+  fn connect_impl(_addr: &OsStr) -> io::Result<Self> {
+    Err(io::Error::new(
+      io::ErrorKind::Unsupported,
+      "Platform not supported.",
+    ))
+  }
+}
+
+impl Read for LocalStream {
+  fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+    self.0.read(buf)
+  }
+}
+
+impl Write for LocalStream {
+  fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+    self.0.write(buf)
+  }
+
+  fn flush(&mut self) -> io::Result<()> {
+    self.0.flush()
+  }
+}

runtime/permissions/lib.rs

@@ -31,25 +31,16 @@ use serde::Serialize;
 use serde::de;
 use url::Url;
 
-#[cfg(unix)]
 pub mod broker;
+mod ipc_pipe;
 pub mod prompter;
 pub mod which;
-#[cfg(unix)]
-pub use broker::PermissionBroker;
-#[cfg(unix)]
-pub use broker::set_broker;
-pub use prompter::DeniedPrompter;
-pub use prompter::GetFormattedStackFn;
+
 use prompter::MAYBE_CURRENT_STACKTRACE;
 use prompter::PERMISSION_EMOJI;
-pub use prompter::PermissionPrompter;
-pub use prompter::PromptCallback;
-pub use prompter::PromptResponse;
 use prompter::permission_prompt;
-pub use prompter::set_prompt_callbacks;
-pub use prompter::set_prompter;
 
+use self::prompter::PromptResponse;
 use self::which::WhichSys;
 
 #[derive(Debug, Eq, PartialEq)]
@@ -58,25 +49,9 @@ pub enum BrokerResponse {
   Deny,
 }
 
-#[cfg(unix)]
 use self::broker::has_broker;
-
-#[cfg(not(unix))]
-fn has_broker() -> bool {
-  false
-}
-
-#[cfg(unix)]
 use self::broker::maybe_check_with_broker;
 
-#[cfg(not(unix))]
-fn maybe_check_with_broker(
-  _name: &str,
-  _stringified_value_fn: impl Fn() -> Option<String>,
-) -> Option<BrokerResponse> {
-  None
-}
-
 pub static AUDIT_FILE: OnceLock<Mutex<std::fs::File>> = OnceLock::new();
 
 #[derive(Debug, thiserror::Error, deno_error::JsError)]
@@ -4815,6 +4790,7 @@ mod tests {
   use serde_json::json;
 
   use super::*;
+  use crate::prompter::set_prompter;
 
   // Creates vector of strings, Vec<String>
   macro_rules! svec {

tests/integration/run_tests.rs

@@ -3505,11 +3505,14 @@ fn handle_invalid_path_error() {
   assert_contains!(String::from_utf8_lossy(&output.stderr), "Module not found");
 }
 
-#[cfg(unix)]
 #[tokio::test]
 async fn test_permission_broker_doesnt_exit() {
   let context = TestContext::default();
-  let socket_path = context.temp_dir().path().join("broker.sock");
+  let socket_path = if cfg!(windows) {
+    PathRef::new(r"\\.\pipe\deno-permission-broker")
+  } else {
+    context.temp_dir().path().join("broker.sock")
+  };
 
   let output = context
     .new_command()
@@ -3522,14 +3525,17 @@ async fn test_permission_broker_doesnt_exit() {
   );
 }
 
-#[cfg(unix)]
 #[tokio::test]
 async fn test_permission_broker() {
   use std::io::BufRead;
   use std::io::BufReader;
 
   let context = TestContext::default();
-  let socket_path = context.temp_dir().path().join("broker.sock");
+  let socket_path = if cfg!(windows) {
+    PathRef::new(r"\\.\pipe\deno-permission-broker")
+  } else {
+    context.temp_dir().path().join("broker.sock")
+  };
 
   let mut broker = context
     .new_command()
@@ -3551,6 +3557,7 @@ async fn test_permission_broker() {
       Ok(0) => break, // EOF
       Ok(_) => {
         if line.starts_with("Permission broker listening on") {
+          eprintln!("{}", line);
           break;
         }
       }