fix(tokio): check Unix allowlist before qmux negotiation

quest/m1/cut-audit/README.md

@@ -77,7 +77,6 @@ leaving JS-only redesigns of `js/net` to [rs2ts](/quest/m2/rs2ts/README.md).
 - [Rust and JS accept the same catalogs](/quest/m1/cut-audit/catalog-parity.md) - one strictness, per rendition, with a shared fixture corpus
 - [Lite datagrams obey one size limit](/quest/m1/cut-audit/datagram-size.md) - no datagram one hop accepts is dropped by the next for size
 - [Every relay runtime honors or refuses each listen setting](/quest/m1/cut-audit/uring-listen-config.md) - io_uring stops ignoring preferred addresses
-- [The Unix-socket allowlist is checked first](/quest/m1/cut-audit/uds-allowlist-first.md) - refused local users never reach the qmux parser
 - [A route's terminal state is one phase](/quest/m1/cut-audit/route-phase.md) - three resume flags collapse into one check
 - [Lite's version gates, refusals, and draft text agree](/quest/m1/cut-audit/lite-wire-nits.md) - small wire and draft mismatches fixed
 - [Small @moq/net divergences from Rust are fixed](/quest/m1/cut-audit/js-net-cleanups.md) - lite-04 probes, group consumers, accept deadlines, empty JSON

rs/moq-tokio/src/server.rs

@@ -1502,7 +1502,7 @@ fn spawn_unix_loop(
 		let mut handshakes = tokio::task::JoinSet::new();
 		loop {
 			let pending = tokio::select! {
-				pending = listener.accept_pending() => pending,
+				pending = listener.accept_pending(allow.as_ref()) => pending,
 				Some(_) = handshakes.join_next() => continue,
 			};
 			// Closed for drain: dropping the pending handshake closes its socket.
@@ -1512,17 +1512,9 @@ fn spawn_unix_loop(
 			let deadline = setup.deadline();
 			let setup = setup.clone();
 			let tx = tx.clone();
-			let allow = allow.clone();
 			handshakes.spawn(crate::panic::isolate(async move {
 				match within(deadline, pending).await {
-					Ok((session, cred)) => {
-						// Enforce the allowlist (if any) before reading SETUP bytes from the peer.
-						if let Some(allow) = &allow
-							&& !allow.permits(&cred)
-						{
-							tracing::warn!(uid = cred.uid, gid = cred.gid, pid = ?cred.pid, "unix connection rejected by allow list");
-							return;
-						}
+					Ok((session, _cred)) => {
 						stream_request(
 							session,
 							&setup,
@@ -2020,6 +2012,64 @@ mod tests {
 		let _ = task.await;
 	}
 
+	/// A refused peer must be closed without exchanging even the qmux preface.
+	/// Use our real kernel credentials and change only the configured allowlist.
+	#[cfg(all(feature = "uds", unix))]
+	#[tokio::test]
+	async fn unix_allowlist_precedes_negotiation() {
+		use tokio::io::AsyncReadExt;
+		let (peer, _other) = tokio::net::UnixStream::pair().unwrap();
+		let cred = peer.peer_cred().unwrap();
+		for allowed in [false, true] {
+			let dir = tempfile::tempdir().unwrap();
+			let path = dir.path().join("listener.sock");
+			let listener = crate::unix::Listener::bind(&path)
+				.await
+				.unwrap()
+				.with_protocols(["moq-lite-06"]);
+			let (tx, _rx) = tokio::sync::mpsc::channel(1);
+			let task = spawn_unix_loop(
+				listener,
+				Setup {
+					activity: Default::default(),
+					server: moq_net::Server::new(),
+					timeout: None,
+				},
+				Some(crate::unix::Allow {
+					uid: vec![if allowed {
+						cred.uid()
+					} else {
+						cred.uid().wrapping_add(1)
+					}],
+					..Default::default()
+				}),
+				tx,
+			);
+			if allowed {
+				let mut config = qmux::Config::new(qmux::Version::QMux01);
+				config.protocol = qmux::Protocol::Negotiate(vec!["moq-lite-06".into()]);
+				let session = qmux::uds::connect(&path, config)
+					.await
+					.expect("allowed peer negotiates");
+				assert_eq!(session.protocol(), Some("moq-lite-06"));
+			} else {
+				let mut peer = tokio::net::UnixStream::connect(&path).await.unwrap();
+				let mut byte = [0];
+				let read = peer.read(&mut byte).await;
+				// Some kernels report reset instead of EOF when the socket closes.
+				assert!(
+					matches!(read, Ok(0))
+						|| read
+							.as_ref()
+							.is_err_and(|err| err.kind() == std::io::ErrorKind::ConnectionReset),
+					"refused peer reached qmux negotiation: {read:?}, byte={byte:?}"
+				);
+			}
+			task.abort();
+			let _ = task.await;
+		}
+	}
+
 	#[cfg(all(feature = "uds", unix))]
 	#[tokio::test]
 	async fn unix_stalled_handshake_does_not_block_accept() {

rs/moq-tokio/src/unix.rs

@@ -349,12 +349,16 @@ impl Listener {
 	///
 	/// As in [`crate::tcp`], the `Option` has no `None` case left to report.
 	pub async fn accept(&self) -> Option<Result<(qmux::Session, PeerCred)>> {
-		Some(self.accept_pending().await.await)
+		Some(self.accept_pending(None).await.await)
 	}
 
 	/// Accept a socket and return its independently driven qmux handshake.
-	pub(crate) async fn accept_pending(&self) -> impl Future<Output = Result<(qmux::Session, PeerCred)>> + use<> {
+	pub(crate) async fn accept_pending(
+		&self,
+		allow: Option<&Allow>,
+	) -> impl Future<Output = Result<(qmux::Session, PeerCred)>> + use<> {
 		let stream = self.accept_socket().await;
+		let allow = allow.cloned();
 		let mut config = crate::qmux_config();
 		config.version = WIRE_VERSION;
 		config.protocol = qmux::Protocol::Negotiate(self.protocols.clone());
@@ -365,6 +369,14 @@ impl Listener {
 				gid: cred.gid(),
 				pid: cred.pid(),
 			};
+			// Qmux negotiates in-band, so enforce credentials before it can read or
+			// write any bytes, rather than after it returns a session.
+			if let Some(allow) = &allow
+				&& !allow.permits(&cred)
+			{
+				tracing::warn!(uid = cred.uid, gid = cred.gid, pid = ?cred.pid, "unix connection rejected by allow list");
+				return Err(io::Error::new(io::ErrorKind::PermissionDenied, "unix peer outside allow list").into());
+			}
 			let session = qmux::uds::accept(stream, config)
 				.await
 				.map_err(|err| Error::Accept(crate::error::message(err)))?;