use std::io::{self, Read, Write};
#[cfg(feature = "ipc-async")]
use std::pin::Pin;
#[cfg(feature = "ipc-async")]
use std::task::{Context, Poll};
use interprocess::local_socket::prelude::*;
#[cfg(feature = "ipc-async")]
use interprocess::local_socket::tokio::prelude::*;
use interprocess::local_socket::{GenericNamespaced, ListenerOptions, PeerCreds, ToNsName};
use interprocess::TryClone;
#[cfg(feature = "ipc-async")]
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Endpoint(String);
impl Endpoint {
pub fn new(path: impl Into<String>) -> io::Result<Self> {
let path = path.into();
name(&path)?;
Ok(Self(path))
}
pub fn display(&self) -> &str {
&self.0
}
#[cfg(test)]
pub fn test(label: &str) -> io::Result<Self> {
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
Self::new(format!(
r"\\.\pipe\rp-ipc-{label}-{}-{nonce}",
std::process::id()
))
}
}
fn name(path: &str) -> io::Result<interprocess::local_socket::Name<'_>> {
path.strip_prefix(r"\\.\pipe\")
.unwrap_or(path)
.to_ns_name::<GenericNamespaced>()
.map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PeerIdentity {
pub pid: u32,
pub user_id: String,
}
fn peer_identity(creds: PeerCreds) -> PeerIdentity {
let pid = creds.pid().unwrap_or(0);
PeerIdentity {
pid,
user_id: if pid == 0 {
String::new()
} else {
process_user_sid(pid).unwrap_or_default()
},
}
}
fn process_user_sid(pid: u32) -> io::Result<String> {
use windows_sys::Win32::Security::{
GetLengthSid, GetTokenInformation, IsValidSid, TokenUser, TOKEN_QUERY, TOKEN_USER,
};
use windows_sys::Win32::System::Threading::{
OpenProcess, OpenProcessToken, PROCESS_QUERY_LIMITED_INFORMATION,
};
unsafe {
let process = OwnedHandle(OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid));
if process.0.is_null() {
return Err(io::Error::last_os_error());
}
let mut token = std::ptr::null_mut();
if OpenProcessToken(process.0, TOKEN_QUERY, &mut token) == 0 {
return Err(io::Error::last_os_error());
}
let token = OwnedHandle(token);
let mut required = 0;
let _ = GetTokenInformation(
token.0,
TokenUser,
std::ptr::null_mut(),
0,
&mut required,
);
if required == 0 {
return Err(io::Error::last_os_error());
}
let mut buffer = vec![0_u8; required as usize];
let queried = GetTokenInformation(
token.0,
TokenUser,
buffer.as_mut_ptr().cast(),
required,
&mut required,
);
if queried == 0 {
return Err(io::Error::last_os_error());
}
let sid = (*(buffer.as_ptr().cast::<TOKEN_USER>())).User.Sid;
if sid.is_null() || IsValidSid(sid) == 0 {
return Err(io::Error::other("invalid Windows SID"));
}
let len = GetLengthSid(sid) as usize;
if len == 0 || len > 1024 {
return Err(io::Error::other("implausible Windows SID length"));
}
let bytes = std::slice::from_raw_parts(sid.cast::<u8>(), len);
let mut out = String::with_capacity("windows-sid:".len() + len * 2);
out.push_str("windows-sid:");
for byte in bytes {
use std::fmt::Write as _;
let _ = write!(out, "{byte:02x}");
}
Ok(out)
}
}
struct OwnedHandle(windows_sys::Win32::Foundation::HANDLE);
impl Drop for OwnedHandle {
fn drop(&mut self) {
if !self.0.is_null() {
unsafe {
windows_sys::Win32::Foundation::CloseHandle(self.0);
}
}
}
}
pub fn current_user_id() -> io::Result<String> {
process_user_sid(std::process::id())
}
pub struct Stream(pub(crate) interprocess::local_socket::Stream);
impl Stream {
pub fn connect(endpoint: &Endpoint) -> io::Result<Self> {
interprocess::local_socket::Stream::connect(name(endpoint.display())?).map(Self)
}
pub fn try_clone(&self) -> io::Result<Self> {
self.0.try_clone().map(Self)
}
pub fn set_nonblocking(&self, nonblocking: bool) -> io::Result<()> {
interprocess::local_socket::traits::Stream::set_nonblocking(&self.0, nonblocking)
}
pub fn peer_identity(&self) -> io::Result<PeerIdentity> {
self.0.peer_creds().map(peer_identity)
}
}
impl Read for Stream {
fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
self.0.read(buffer)
}
}
impl Write for Stream {
fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
self.0.write(buffer)
}
fn flush(&mut self) -> io::Result<()> {
self.0.flush()
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum ListenerNonblockingMode {
#[default]
Neither,
Accept,
Stream,
Both,
}
impl From<ListenerNonblockingMode> for interprocess::local_socket::ListenerNonblockingMode {
fn from(value: ListenerNonblockingMode) -> Self {
match value {
ListenerNonblockingMode::Neither => Self::Neither,
ListenerNonblockingMode::Accept => Self::Accept,
ListenerNonblockingMode::Stream => Self::Stream,
ListenerNonblockingMode::Both => Self::Both,
}
}
}
pub struct Listener(interprocess::local_socket::Listener);
impl Listener {
pub fn bind(endpoint: &Endpoint) -> io::Result<Self> {
Self::bind_with_options(endpoint, true, ListenerNonblockingMode::Neither)
}
pub fn bind_with_options(
endpoint: &Endpoint,
reclaim_name: bool,
nonblocking: ListenerNonblockingMode,
) -> io::Result<Self> {
ListenerOptions::new()
.name(name(endpoint.display())?)
.reclaim_name(reclaim_name)
.nonblocking(nonblocking.into())
.create_sync()
.map(Self)
}
pub fn accept(&self) -> io::Result<Stream> {
self.0.accept().map(Stream)
}
pub fn set_nonblocking(&self, mode: ListenerNonblockingMode) -> io::Result<()> {
self.0.set_nonblocking(mode.into())
}
pub fn do_not_reclaim_name_on_drop(&mut self) {
self.0.do_not_reclaim_name_on_drop();
}
}
#[cfg(feature = "ipc-async")]
pub struct AsyncStream(pub(crate) interprocess::local_socket::tokio::Stream);
#[cfg(feature = "ipc-async")]
impl AsyncStream {
pub async fn connect(endpoint: &Endpoint) -> io::Result<Self> {
interprocess::local_socket::tokio::Stream::connect(name(endpoint.display())?)
.await
.map(Self)
}
pub fn peer_identity(&self) -> io::Result<PeerIdentity> {
self.0.peer_creds().map(peer_identity)
}
}
#[cfg(feature = "ipc-async")]
impl AsyncRead for AsyncStream {
fn poll_read(
mut self: Pin<&mut Self>,
context: &mut Context<'_>,
buffer: &mut ReadBuf<'_>,
) -> Poll<io::Result<()>> {
Pin::new(&mut self.0).poll_read(context, buffer)
}
}
#[cfg(feature = "ipc-async")]
impl AsyncWrite for AsyncStream {
fn poll_write(
mut self: Pin<&mut Self>,
context: &mut Context<'_>,
buffer: &[u8],
) -> Poll<io::Result<usize>> {
Pin::new(&mut self.0).poll_write(context, buffer)
}
fn poll_flush(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.0).poll_flush(context)
}
fn poll_shutdown(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(&mut self.0).poll_shutdown(context)
}
}
#[cfg(feature = "ipc-async")]
pub struct AsyncListener(interprocess::local_socket::tokio::Listener);
#[cfg(feature = "ipc-async")]
impl AsyncListener {
pub fn bind(endpoint: &Endpoint) -> io::Result<Self> {
ListenerOptions::new()
.name(name(endpoint.display())?)
.create_tokio()
.map(Self)
}
pub async fn accept(&self) -> io::Result<AsyncStream> {
self.0.accept().await.map(AsyncStream)
}
pub fn do_not_reclaim_name_on_drop(&mut self) {
self.0.do_not_reclaim_name_on_drop();
}
}