moq_native/
unix.rs

1//! Unix-domain-socket qmux transport, reachable via the `unix://` URL scheme.
2//!
3//! Runs the QMux wire format over an `AF_UNIX` stream. Unlike the `tcp://`
4//! transport, the kernel reports the connecting process's credentials
5//! (`SO_PEERCRED` / `LOCAL_PEERCRED`), so a server can authenticate the peer's
6//! uid/gid/pid without a shared secret. Unix-only.
7
8use std::os::unix::fs::{FileTypeExt, PermissionsExt};
9use std::path::{Path, PathBuf};
10use std::{fs, io};
11
12use url::Url;
13
14/// The QMux wire-format version both ends speak. Fixed (not negotiated) since a
15/// raw stream has no TLS ALPN to carry it.
16const WIRE_VERSION: qmux::Version = qmux::Version::QMux01;
17
18/// Errors specific to the Unix-domain-socket qmux transport.
19#[derive(Debug, thiserror::Error)]
20#[non_exhaustive]
21pub enum Error {
22	/// The socket failed to bind, accept, connect, or chmod.
23	#[error(transparent)]
24	Io(#[from] io::Error),
25
26	/// The `unix://` URL had no socket path.
27	#[error("missing socket path in unix:// URL")]
28	MissingPath,
29
30	/// The qmux handshake failed while dialing.
31	#[error("qmux connect failed")]
32	Connect(#[source] qmux::Error),
33
34	/// The qmux handshake failed while accepting.
35	#[error("qmux accept failed")]
36	Accept(#[source] qmux::Error),
37
38	/// The bind path already exists and is not a socket, so we refuse to unlink it.
39	#[error("refusing to replace existing non-socket file at {0}")]
40	NotASocket(PathBuf),
41}
42
43type Result<T> = std::result::Result<T, Error>;
44
45/// Credentials of a connected Unix-socket peer.
46///
47/// `pid` is `None` on platforms that don't report it (e.g. some macOS versions);
48/// `uid`/`gid` are always available.
49#[derive(Clone, Copy, Debug, PartialEq, Eq)]
50pub struct PeerCred {
51	/// The peer process's effective user ID.
52	pub uid: u32,
53	/// The peer process's effective group ID.
54	pub gid: u32,
55	/// The peer process's PID, if the platform reports it.
56	pub pid: Option<i32>,
57}
58
59/// Dial a `unix://<path>` URL, advertising `protocols` for in-band ALPN
60/// negotiation. Returns a qmux session over the socket.
61///
62/// The path is taken from the URL path, so use a triple slash for an absolute
63/// path: `unix:///run/moq/internal.sock`.
64pub(crate) async fn connect(url: Url, protocols: &[&str]) -> Result<qmux::Session> {
65	let path = socket_path(&url).ok_or(Error::MissingPath)?;
66	tracing::debug!(%url, "connecting via Unix socket");
67	qmux::uds::Config::new(WIRE_VERSION)
68		.protocols(protocols.iter().copied())
69		.connect(path)
70		.await
71		.map_err(Error::Connect)
72}
73
74fn socket_path(url: &Url) -> Option<PathBuf> {
75	let path = url.path();
76	if path.is_empty() {
77		None
78	} else {
79		Some(PathBuf::from(path))
80	}
81}
82
83/// Listens for incoming qmux connections on a Unix domain socket.
84///
85/// Each accepted connection yields the session plus the peer's [`PeerCred`], so
86/// the caller can enforce a uid/gid/pid allowlist. The socket file is removed on
87/// drop.
88pub struct Listener {
89	listener: tokio::net::UnixListener,
90	path: PathBuf,
91	protocols: Vec<String>,
92}
93
94impl Listener {
95	/// Bind a Unix socket at `path`, replacing a stale socket file left by a
96	/// previous run.
97	///
98	/// Refuses to unlink the path if it exists and is not a socket, to avoid
99	/// clobbering an unrelated file.
100	pub async fn bind(path: impl AsRef<Path>) -> Result<Self> {
101		let path = path.as_ref().to_path_buf();
102
103		// A leftover socket from a crashed run would make bind() fail with
104		// EADDRINUSE, so unlink it first. Anything that isn't a socket we leave
105		// alone and error out.
106		match fs::symlink_metadata(&path) {
107			Ok(meta) if meta.file_type().is_socket() => fs::remove_file(&path)?,
108			Ok(_) => return Err(Error::NotASocket(path)),
109			Err(err) if err.kind() == io::ErrorKind::NotFound => {}
110			Err(err) => return Err(err.into()),
111		}
112
113		let listener = tokio::net::UnixListener::bind(&path)?;
114		Ok(Self {
115			listener,
116			path,
117			protocols: Vec::new(),
118		})
119	}
120
121	/// Advertise these application protocols (moq ALPNs) for in-band negotiation,
122	/// in preference order. The first server entry the client also offers wins.
123	pub fn with_protocols<I, S>(mut self, protocols: I) -> Self
124	where
125		I: IntoIterator<Item = S>,
126		S: Into<String>,
127	{
128		self.protocols = protocols.into_iter().map(Into::into).collect();
129		self
130	}
131
132	/// Set the socket file's permission bits (e.g. `0o660`).
133	pub fn set_mode(&self, mode: u32) -> Result<()> {
134		fs::set_permissions(&self.path, fs::Permissions::from_mode(mode))?;
135		Ok(())
136	}
137
138	/// The bound socket path.
139	pub fn path(&self) -> &Path {
140		&self.path
141	}
142
143	/// Accept the next connection, returning the session and the peer's credentials.
144	///
145	/// Returns `None` only if the listener itself is gone; a per-connection
146	/// failure is yielded as `Some(Err(..))` so the accept loop keeps running.
147	pub async fn accept(&self) -> Option<Result<(qmux::Session, PeerCred)>> {
148		match self.listener.accept().await {
149			Ok((stream, _addr)) => {
150				let cred = match stream.peer_cred() {
151					Ok(cred) => PeerCred {
152						uid: cred.uid(),
153						gid: cred.gid(),
154						pid: cred.pid(),
155					},
156					Err(err) => return Some(Err(err.into())),
157				};
158				let session = qmux::uds::Config::new(WIRE_VERSION)
159					.protocols(self.protocols.iter().map(String::as_str))
160					.accept(stream)
161					.await
162					.map_err(Error::Accept);
163				Some(session.map(|session| (session, cred)))
164			}
165			Err(err) => Some(Err(err.into())),
166		}
167	}
168}
169
170impl Drop for Listener {
171	fn drop(&mut self) {
172		// Best-effort: don't leave a stale socket file behind.
173		let _ = fs::remove_file(&self.path);
174	}
175}
moq_native/unix.rs

moq_native/
unix.rs
