diff --git a/Cargo.lock b/Cargo.lock
index c7bbf979..67f8d545 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4214,9 +4214,9 @@ dependencies = [
 
 [[package]]
 name = "tinyvec"
-version = "1.13.0"
+version = "1.13.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ba2077be5cd93c7408c849e45a4ab261b59f923f4bcdee2a724b8ed5a175a77"
+checksum = "4cf0ded5c4e56918d8f8a339e1bb67d038d3bc6d144ac407904015ba2e4cde9b"
 dependencies = [
  "tinyvec_macros",
 ]
diff --git a/apps/neshub/src/ipc_listener.rs b/apps/neshub/src/ipc_listener.rs
index e61d3fd9..c3f977cb 100644
--- a/apps/neshub/src/ipc_listener.rs
+++ b/apps/neshub/src/ipc_listener.rs
@@ -292,7 +292,24 @@ pub async fn run_stats_ipc_listener(
     tracing::info!("stats IPC listener exited");
 }
 
-pub async fn run_ticket_ipc_listener(socket_path: PathBuf, ticket: crate::NestriTicket) {
+/// Serve the address a client needs, rebuilt on every read.
+///
+/// **The ticket is built per connection and not once at startup.** An endpoint
+/// does not know all of its own addresses when it binds: a direct one is there
+/// immediately, and a relayed or hole-punched one becomes known seconds later.
+/// A ticket captured once therefore carries only the address that was available
+/// first, which is the one that works on the same network and fails from
+/// anywhere else — and whoever reads this polls precisely so that a better
+/// answer can replace it. Serving a snapshot made that polling pointless: every
+/// read returned the same local-only address forever.
+///
+/// The stream name is generated once and kept, because it identifies this
+/// session rather than describing how to reach it. Only the addresses change.
+pub async fn run_ticket_ipc_listener(
+    socket_path: PathBuf,
+    endpoint: iroh::Endpoint,
+    stream_name: String,
+) {
     if socket_path.exists() {
         let _ = std::fs::remove_file(&socket_path);
     }
@@ -319,6 +336,9 @@ pub async fn run_ticket_ipc_listener(socket_path: PathBuf, ticket: crate::Nestri
     loop {
         match listener.accept().await {
             Ok((mut stream, _)) => {
+                // Asked of the endpoint now, so an address it has learned since
+                // the last read is in this answer.
+                let ticket = crate::NestriTicket::new(endpoint.addr(), stream_name.clone());
                 if let Err(e) = stream.write_all(format!("{ticket}\n").as_bytes()).await {
                     tracing::warn!("could not write ticket to IPC: {e}");
                 }
diff --git a/apps/neshub/src/main.rs b/apps/neshub/src/main.rs
index 27636cd1..9cb96bc9 100644
--- a/apps/neshub/src/main.rs
+++ b/apps/neshub/src/main.rs
@@ -207,7 +207,10 @@ async fn main() -> Result<()> {
 
     // ── Accept mode: generate ticket, wait for desktop-app to connect ─────
     let stream_name = ticket::generate_stream_name();
-    let ticket = NestriTicket::new(endpoint_addr, stream_name);
+    // For the log line below only. What a reader of the socket gets is built
+    // per read from the endpoint itself, because the addresses this can be
+    // reached at are not all known yet.
+    let ticket = NestriTicket::new(endpoint_addr, stream_name.clone());
 
     tracing::info!("╔═══════════════╗");
     tracing::info!("║ NESTRI TICKET ║");
@@ -245,7 +248,12 @@ async fn main() -> Result<()> {
 
     let ticket_ipc = args.ticket_ipc.clone();
     tokio::spawn({
-        async move { ipc_listener::run_ticket_ipc_listener(ticket_ipc, ticket).await }
+        // The endpoint rather than a ticket made from it: the addresses it can
+        // be reached at are not all known yet, and whoever reads this socket
+        // re-reads it so that a better one can replace the first.
+        let endpoint = endpoint.clone();
+        let stream_name = stream_name.clone();
+        async move { ipc_listener::run_ticket_ipc_listener(ticket_ipc, endpoint, stream_name).await }
     });
 
     // Accept loop
diff --git a/apps/nesinit/src/filesystems.rs b/apps/nesinit/src/filesystems.rs
index 8c2768a8..04d03d5e 100644
--- a/apps/nesinit/src/filesystems.rs
+++ b/apps/nesinit/src/filesystems.rs
@@ -63,14 +63,42 @@ const EARLY: &[Early] = &[
         cost: "whatever serves this session's address cannot bind its socket, \
                so the session never gets one",
     },
+    // `/run` before anything under it, for the same reason `/proc` comes first:
+    // a directory cannot be created inside a mount that is not there, and the
+    // root it would otherwise land on is read-only.
+    Early {
+        source: "tmpfs",
+        target: "/run",
+        fstype: "tmpfs",
+        flags: NOSUID_NODEV,
+        // Octal, and without a leading zero on purpose: the kernel parses a
+        // tmpfs mode as octal either way, and this is the spelling `mount`
+        // itself documents.
+        data: "mode=755",
+        cost: "there is nowhere for a runtime socket to live, so neither the \
+               payload relay nor this session's address can be served",
+    },
+    // The relay's own directory, and it is deliberately **not** in the tree the
+    // session's shares live in.
+    //
+    // It was, and that was wrong in a way no test here would have caught: a
+    // fresh tmpfs over the share tree hides every directory the image prepared
+    // underneath it — the install, the user state, the work directory, and the
+    // mount point the log share is attached to from `fstab`. The box then has a
+    // socket and none of the places its workload expects to find its files, and
+    // the exact-path check below cannot notice, because what `fstab` mounts is
+    // a directory *inside* that tree rather than the tree itself.
+    //
+    // Owned by this process and writable by nothing else, which is what makes
+    // the socket in it unreplaceable. The workload reaches it because the
+    // directory is traversable and the socket itself is not restricted; see
+    // `payload::serve`.
     Early {
         source: "tmpfs",
         target: crate::payload::DIRECTORY,
         fstype: "tmpfs",
         flags: NOSUID_NODEV | libc::MS_NOEXEC,
-        // Only this process and the workload it starts, and they are the only
-        // two that ever have business here.
-        data: "mode=0770",
+        data: "mode=755",
         cost: "the payload relay cannot bind, so nothing reaches the workload \
                over the channel",
     },
@@ -198,6 +226,39 @@ mod tests {
         assert_eq!(EARLY[0].target, "/proc");
     }
 
+    /// A mount has to come after whatever it lives inside, or it is a
+    /// directory created on a read-only root and the mount fails.
+    #[test]
+    fn nothing_is_mounted_before_the_mount_it_lives_inside() {
+        for (i, early) in EARLY.iter().enumerate() {
+            for other in &EARLY[i + 1..] {
+                assert!(
+                    !early.target.starts_with(&format!("{}/", other.target)),
+                    "{} is mounted before {}, which contains it",
+                    early.target,
+                    other.target
+                );
+            }
+        }
+    }
+
+    /// **Nothing here may be mounted over the tree the session's shares live
+    /// in.** A fresh tmpfs there hides every directory the image prepared
+    /// underneath — the install, the user state, the work directory, and the
+    /// mount point the log share attaches to — and the exact-path check cannot
+    /// notice, because what is mounted from `fstab` is a directory inside that
+    /// tree rather than the tree itself. So a box would come up with a socket
+    /// and without any of the places its workload looks for its files.
+    #[test]
+    fn the_share_tree_is_never_mounted_over() {
+        for early in EARLY {
+            assert_ne!(
+                early.target, "/nestri",
+                "this hides the directories the image prepared for a session"
+            );
+        }
+    }
+
     /// The relay's directory is the one this cannot hardcode: it belongs to
     /// `payload`, and a rename there that missed this file would take the
     /// relay down again in exactly the way this exists to prevent.
diff --git a/apps/nesinit/src/main.rs b/apps/nesinit/src/main.rs
index 239be9b7..cb44cbed 100644
--- a/apps/nesinit/src/main.rs
+++ b/apps/nesinit/src/main.rs
@@ -114,10 +114,18 @@ async fn guest(waiters: &Waiters, workload: &mut Process) -> anyhow::Result<Outc
     // whatever serves the address may bind the moment it comes up, and nothing
     // here should be the reason a session waits to be reachable.
     let (found_tx, mut found_rx) = tokio::sync::mpsc::channel(ADDRESS_DEPTH);
-    tokio::spawn(ticket::carry(PathBuf::from(ticket::SOCKET), found_tx));
+    // Which user the carrier must not accept an address from. Empty until the
+    // descriptor names it, which is also when the workload that could abuse it
+    // is started -- so there is nothing to refuse before it is filled in.
+    let untrusted = ticket::Untrusted::unknown();
+    tokio::spawn(ticket::carry(
+        PathBuf::from(ticket::SOCKET),
+        found_tx,
+        untrusted.clone(),
+    ));
 
     let outcome = tokio::select! {
-        outcome = session::run(channel, workload, &mut ports, &mut found_rx) => outcome?,
+        outcome = session::run(channel, workload, &mut ports, &mut found_rx, &untrusted) => outcome?,
         signal = asked_to_stop() => {
             signal?;
             tracing::info!("asked to stop");
diff --git a/apps/nesinit/src/payload.rs b/apps/nesinit/src/payload.rs
index 944bcbca..cbfebe00 100644
--- a/apps/nesinit/src/payload.rs
+++ b/apps/nesinit/src/payload.rs
@@ -9,6 +9,7 @@
 // anything above it.
 
 use std::io;
+use std::os::unix::fs::PermissionsExt;
 use std::path::Path;
 
 use nesprotocol::lifecycle::{Payload, from_line, to_line};
@@ -22,10 +23,18 @@ use tokio::sync::mpsc::{Receiver, Sender};
 /// is read-only, so this is a tmpfs that `filesystems` puts there, and a
 /// rename here that did not reach the mount table would take the relay down
 /// with an `EROFS` that looks like nothing to do with a path.
-pub const DIRECTORY: &str = "/nestri";
+///
+/// **Under `/run` rather than in the tree the session's shares live in**, and
+/// that was a correction. Putting a fresh tmpfs on the share tree hid every
+/// directory the image had prepared underneath it — the install, the user
+/// state, the work directory and the log share's own mount point — so a box
+/// gained a socket and lost the places its workload was supposed to find its
+/// files. `/run` is where a runtime socket belongs, it is a tmpfs already, and
+/// nothing else is mounted inside it.
+pub const DIRECTORY: &str = "/run/nestri";
 
 /// Where the workload finds the relay.
-pub const SOCKET: &str = "/nestri/payload.sock";
+pub const SOCKET: &str = "/run/nestri/payload.sock";
 
 /// The longest envelope this will assemble before giving up on the connection.
 ///
@@ -63,6 +72,17 @@ pub async fn serve(
     let _ = std::fs::remove_file(path);
     let listener = UnixListener::bind(path)?;
 
+    // The workload does not run as this process does, and it has to be able to
+    // connect. It reaches the socket through a directory this process owns and
+    // nothing else may write to, so the permission that matters is on the
+    // socket rather than on the path: the directory is what stops anybody
+    // replacing this listener, and this is what lets the workload talk to it.
+    //
+    // Without it the workload gets `EACCES` on connect and the payload layer
+    // is dead in both directions, silently, because nothing in the guest is
+    // waiting to be told about a relay it cannot reach.
+    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o666))?;
+
     loop {
         let stream = tokio::select! {
             accepted = listener.accept() => accepted?.0,
diff --git a/apps/nesinit/src/session.rs b/apps/nesinit/src/session.rs
index 8490f499..a94e0b33 100644
--- a/apps/nesinit/src/session.rs
+++ b/apps/nesinit/src/session.rs
@@ -41,12 +41,13 @@ pub async fn run<C, W>(
     workload: &mut W,
     payload: &mut Ports,
     addresses: &mut Receiver<String>,
+    untrusted: &crate::ticket::Untrusted,
 ) -> std::io::Result<Outcome>
 where
     C: AsyncRead + AsyncWrite,
     W: Workload,
 {
-    match converse(channel, workload, payload, addresses).await {
+    match converse(channel, workload, payload, addresses, untrusted).await {
         Err(error) if channel_gone(&error) => {
             // A caller that has stopped reading has also stopped being able to
             // tell us to stop, which is the same situation as the channel
@@ -73,6 +74,7 @@ async fn converse<C, W>(
     workload: &mut W,
     payload: &mut Ports,
     addresses: &mut Receiver<String>,
+    untrusted: &crate::ticket::Untrusted,
 ) -> std::io::Result<Outcome>
 where
     C: AsyncRead + AsyncWrite,
@@ -186,6 +188,11 @@ where
                     }
                 }
 
+                // Before the workload exists, so there is no window in which
+                // it is running and something else would still be trusted to
+                // serve this session's address.
+                untrusted.is(descriptor.exec.uid);
+
                 match workload.start(&descriptor.exec) {
                     Ok(exited) => {
                         send(&mut writer, &GuestToHost::Started).await?;
@@ -354,9 +361,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
-                .await
-                .unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -385,9 +398,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports, &mut found_rx)
-                .await
-                .unwrap()
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut found_rx,
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
         });
 
         assert_eq!(
@@ -436,9 +455,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports, &mut found_rx)
-                .await
-                .unwrap()
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut found_rx,
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
         });
 
         assert_eq!(
@@ -462,9 +487,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
-                .await
-                .unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -490,9 +521,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_at_once(Exit::code(3));
-            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
-                .await
-                .unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -528,9 +565,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_at_once(Exit::signal(9));
-            run(guest, &mut workload, &mut ports, &mut nowhere())
-                .await
-                .unwrap()
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
         });
 
         assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));
@@ -561,9 +604,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
-                .await
-                .unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -589,9 +638,15 @@ mod tests {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_at_once(Exit::code(0));
             workload.mount_failure = Some(Failure::new("EACCES: /mnt/user"));
-            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
-                .await
-                .unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -626,9 +681,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports, &mut nowhere())
-                .await
-                .unwrap()
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
         });
 
         assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));
@@ -651,9 +712,15 @@ mod tests {
         let session = tokio::spawn(async move {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
-                .await
-                .unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -685,9 +752,15 @@ mod tests {
             let (mut ports, _to_workload, _from_workload) = ports();
             let mut workload = Double::exits_at_once(Exit::code(0));
             workload.start_failure = Some(Failure::new("ENOENT: /usr/bin/workload"));
-            let outcome = run(guest, &mut workload, &mut ports, &mut nowhere())
-                .await
-                .unwrap();
+            let outcome = run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap();
             (outcome, workload)
         });
 
@@ -729,9 +802,15 @@ mod tests {
                 from_workload: up_rx,
             };
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports, &mut nowhere())
-                .await
-                .unwrap()
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
         });
         let mut to_relay = down_rx;
 
@@ -780,9 +859,15 @@ mod tests {
                 from_workload: up_rx,
             };
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports, &mut nowhere())
-                .await
-                .unwrap()
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
         });
 
         assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));
@@ -828,9 +913,15 @@ mod tests {
                 from_workload: up_rx,
             };
             let mut workload = Double::exits_when_stopped(Exit::code(0));
-            run(guest, &mut workload, &mut ports, &mut nowhere())
-                .await
-                .unwrap()
+            run(
+                guest,
+                &mut workload,
+                &mut ports,
+                &mut nowhere(),
+                &crate::ticket::Untrusted::unknown(),
+            )
+            .await
+            .unwrap()
         });
 
         assert!(matches!(caller.expect().await, GuestToHost::Ready { .. }));
diff --git a/apps/nesinit/src/ticket.rs b/apps/nesinit/src/ticket.rs
index 991bd8f6..82eebf18 100644
--- a/apps/nesinit/src/ticket.rs
+++ b/apps/nesinit/src/ticket.rs
@@ -22,8 +22,24 @@
 // long-lived one. Dialling also means a server that has not bound yet is an
 // error this retries, rather than a connection that has to be waited for
 // without knowing whether it is coming.
+//
+// # Who is allowed to answer
+//
+// Dialling a path means trusting whoever is behind it, and an address is the
+// capability to reach this session — so the wrong answer here does not break a
+// session, it hands one to somebody else. The workload runs arbitrary code, and
+// on a writable directory it can unlink whatever bound the socket and bind a
+// replacement; every read after that returns an address of its choosing, and
+// the client outside connects there instead.
+//
+// So the peer's credentials are checked and an answer from the workload's own
+// user is refused. That check is only as good as the workload having a user of
+// its own: run it as the same user as the process serving the address and
+// nothing can tell the two apart, which is [`Peer::indistinguishable`] and is
+// said out loud at boot rather than discovered later.
 
 use std::io;
+use std::os::fd::AsRawFd;
 use std::path::{Path, PathBuf};
 use std::time::Duration;
 
@@ -57,16 +73,66 @@ const PATIENCE: Duration = Duration::from_secs(5);
 /// buffer is the one the kernel has been told not to kill.
 const LONGEST: u64 = 8 * 1024;
 
+/// Who may not serve this session's address.
+///
+/// The workload's user, and nothing else is excluded — this is not an allow
+/// list of trusted uids, because init does not know which user an image happens
+/// to run its media components as, and inventing one here would be a second
+/// place for that to be configured wrongly.
+///
+/// Shared and filled in later, because the carrier starts before the descriptor
+/// that names the user arrives. That leaves no gap: a workload cannot serve
+/// anything before it is started, and it is started from the same descriptor,
+/// which sets this first.
+#[derive(Clone, Debug, Default)]
+pub struct Untrusted(std::sync::Arc<std::sync::atomic::AtomicU32>);
+
+/// No workload has been started, so there is no untrusted user yet.
+const NOBODY: u32 = u32::MAX;
+
+impl Untrusted {
+    pub fn unknown() -> Self {
+        Self(std::sync::Arc::new(std::sync::atomic::AtomicU32::new(
+            NOBODY,
+        )))
+    }
+
+    /// Name the user the workload runs as. Called before it is started.
+    pub fn is(&self, uid: u32) {
+        self.0.store(uid, std::sync::atomic::Ordering::Release);
+        if uid == 0 {
+            tracing::warn!(
+                "the workload runs as root, so an address it serves cannot be \
+                 told apart from a real one. Give it a user of its own."
+            );
+        }
+    }
+
+    /// The uid to refuse, or `None` when refusing anything would be wrong.
+    ///
+    /// `None` covers two cases that want the same answer for different reasons:
+    /// nothing has been started yet, so no peer can be the workload; and the
+    /// workload runs as `root`, which is every user at once — refusing root
+    /// would refuse whatever legitimately serves the address as well. The uid
+    /// being shared is the thing an operator has to fix, and `is` says so.
+    fn refuse(&self) -> Option<u32> {
+        match self.0.load(std::sync::atomic::Ordering::Acquire) {
+            NOBODY | 0 => None,
+            uid => Some(uid),
+        }
+    }
+}
+
 /// Forward every new address for as long as the session lasts.
 ///
 /// Never returns on its own. A server that is not there yet, or has gone away,
 /// is retried at the next interval — there is nothing here worth ending a
 /// running session over, and an address that stops being re-offered does not
 /// stop being correct.
-pub async fn carry(path: PathBuf, out: Sender<String>) {
+pub async fn carry(path: PathBuf, out: Sender<String>, untrusted: Untrusted) {
     let mut sent: Option<String> = None;
     loop {
-        match look(&path).await {
+        match look(&path, &untrusted).await {
             Ok(current) if Some(&current) != sent.as_ref() => {
                 // The address itself is not logged. It is a capability to reach
                 // this session, and a log inside the guest is the one place it
@@ -82,6 +148,14 @@ pub async fn carry(path: PathBuf, out: Sender<String>) {
                 sent = Some(current);
             }
             Ok(_) => {}
+            // Not the same as no address yet, and it must not be logged as
+            // though it were: this says something *is* serving an address and
+            // it is the one thing that may not. A session with no address at
+            // all is a legible failure; a session pointed somewhere else is
+            // not, so this is the line that has to be found afterwards.
+            Err(error) if error.kind() == io::ErrorKind::PermissionDenied => {
+                tracing::error!(%error, "refusing an address for this session");
+            }
             Err(error) => {
                 // Expected until whatever serves the address has bound, so it
                 // is not a warning the first several times. It stays at this
@@ -95,8 +169,8 @@ pub async fn carry(path: PathBuf, out: Sender<String>) {
 }
 
 /// One look at the socket, abandoned if it takes longer than [`PATIENCE`].
-async fn look(path: &Path) -> io::Result<String> {
-    match tokio::time::timeout(PATIENCE, read(path)).await {
+async fn look(path: &Path, untrusted: &Untrusted) -> io::Result<String> {
+    match tokio::time::timeout(PATIENCE, read(path, untrusted)).await {
         Ok(result) => result,
         Err(_) => Err(io::Error::new(
             io::ErrorKind::TimedOut,
@@ -106,8 +180,22 @@ async fn look(path: &Path) -> io::Result<String> {
 }
 
 /// One line from the socket, which is the whole protocol.
-async fn read(path: &Path) -> io::Result<String> {
+async fn read(path: &Path, untrusted: &Untrusted) -> io::Result<String> {
     let stream = UnixStream::connect(path).await?;
+
+    // Before a byte is read. The kernel answers this about the socket's peer
+    // rather than about the path, so it cannot be spoofed by whoever holds the
+    // path — which is the whole reason the check is worth anything.
+    if let Some(refuse) = untrusted.refuse()
+        && peer_uid(&stream)? == refuse
+    {
+        return Err(io::Error::new(
+            io::ErrorKind::PermissionDenied,
+            "the workload is serving this session's address, which would point \
+             a client at whatever it chose",
+        ));
+    }
+
     let mut line = String::new();
     BufReader::new(stream.take(LONGEST))
         .read_line(&mut line)
@@ -122,6 +210,35 @@ async fn read(path: &Path) -> io::Result<String> {
     Ok(line)
 }
 
+/// The uid of the process on the other end of a connected unix socket.
+///
+/// From the kernel, at connect time, and not from anything the peer says about
+/// itself. `SO_PEERCRED` records who held the other end when it connected, so a
+/// process cannot claim a uid it does not have.
+fn peer_uid(stream: &UnixStream) -> io::Result<u32> {
+    let mut credentials = libc::ucred {
+        pid: 0,
+        uid: u32::MAX,
+        gid: u32::MAX,
+    };
+    let mut length = std::mem::size_of::<libc::ucred>() as libc::socklen_t;
+    // SAFETY: a connected socket this function borrows, and an out-parameter of
+    // exactly the length being passed.
+    let rc = unsafe {
+        libc::getsockopt(
+            stream.as_raw_fd(),
+            libc::SOL_SOCKET,
+            libc::SO_PEERCRED,
+            (&raw mut credentials).cast::<libc::c_void>(),
+            &raw mut length,
+        )
+    };
+    if rc != 0 {
+        return Err(io::Error::last_os_error());
+    }
+    Ok(credentials.uid)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -160,7 +277,7 @@ mod tests {
         serve(path.clone(), vec![Some("nestri:abc".into())]);
 
         let (tx, mut rx) = mpsc::channel(4);
-        tokio::spawn(carry(path.clone(), tx));
+        tokio::spawn(carry(path.clone(), tx, Untrusted::unknown()));
 
         let first = tokio::time::timeout(Duration::from_secs(10), rx.recv())
             .await
@@ -191,7 +308,7 @@ mod tests {
         );
 
         let (tx, mut rx) = mpsc::channel(4);
-        tokio::spawn(carry(path.clone(), tx));
+        tokio::spawn(carry(path.clone(), tx, Untrusted::unknown()));
 
         let mut seen = Vec::new();
         while seen.len() < 2 {
@@ -205,13 +322,68 @@ mod tests {
         let _ = std::fs::remove_file(&path);
     }
 
+    /// The peer's user decides whether an address is trusted, and this process
+    /// is the peer in a test — so naming *it* as the workload is a real refusal
+    /// of a real connection, not a stubbed one.
+    ///
+    /// The attack this closes: the workload unlinks whatever bound the socket,
+    /// binds its own, and every read afterwards hands the client an address of
+    /// the workload's choosing.
+    #[tokio::test]
+    async fn an_address_served_by_the_workload_is_refused() {
+        let path = scratch("hostile");
+        serve(path.clone(), vec![Some("nestri:attacker".into())]);
+
+        // SAFETY: reading this process's own uid cannot fail.
+        let ours = unsafe { libc::getuid() };
+        let untrusted = Untrusted::unknown();
+        untrusted.is(ours);
+
+        let error = look(&path, &untrusted)
+            .await
+            .expect_err("an address from the workload's own user was accepted");
+        assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
+
+        // And the same socket is read happily once the workload is somebody
+        // else, which is what shows the refusal is about the peer and not about
+        // the socket.
+        let elsewhere = Untrusted::unknown();
+        elsewhere.is(ours + 1);
+        assert_eq!(look(&path, &elsewhere).await.unwrap(), "nestri:attacker");
+
+        let _ = std::fs::remove_file(&path);
+    }
+
+    /// Before a workload is started there is nothing to refuse, and refusing
+    /// anything then would mean no session ever got an address.
+    #[tokio::test]
+    async fn nothing_is_refused_before_a_workload_exists() {
+        let path = scratch("early");
+        serve(path.clone(), vec![Some("nestri:real".into())]);
+        let untrusted = Untrusted::unknown();
+        assert_eq!(untrusted.refuse(), None);
+        assert_eq!(look(&path, &untrusted).await.unwrap(), "nestri:real");
+        let _ = std::fs::remove_file(&path);
+    }
+
+    /// A workload running as root is every user at once, so refusing root would
+    /// refuse whatever legitimately serves the address too. The check stands
+    /// down and `is` warns instead — the uid being shared is the operator's to
+    /// fix and this is not the place to fail closed over it.
+    #[tokio::test]
+    async fn a_root_workload_leaves_nothing_to_tell_apart() {
+        let untrusted = Untrusted::unknown();
+        untrusted.is(0);
+        assert_eq!(untrusted.refuse(), None);
+    }
+
     /// Nothing serving the socket yet is the ordinary case at boot, not a
     /// failure: this starts before whatever binds it.
     #[tokio::test]
     async fn a_socket_that_is_not_there_yet_is_waited_out_rather_than_failed() {
         let path = scratch("late");
         let (tx, mut rx) = mpsc::channel(4);
-        tokio::spawn(carry(path.clone(), tx));
+        tokio::spawn(carry(path.clone(), tx, Untrusted::unknown()));
 
         tokio::time::sleep(EVERY * 2).await;
         serve(path.clone(), vec![Some("nestri:late".into())]);
@@ -253,7 +425,7 @@ mod tests {
         }
 
         let (tx, mut rx) = mpsc::channel(4);
-        tokio::spawn(carry(path.clone(), tx));
+        tokio::spawn(carry(path.clone(), tx, Untrusted::unknown()));
 
         let first = tokio::time::timeout(PATIENCE + EVERY * 4, rx.recv())
             .await
@@ -271,7 +443,7 @@ mod tests {
         serve(path.clone(), vec![None, Some("nestri:real".into())]);
 
         let (tx, mut rx) = mpsc::channel(4);
-        tokio::spawn(carry(path.clone(), tx));
+        tokio::spawn(carry(path.clone(), tx, Untrusted::unknown()));
 
         let first = tokio::time::timeout(Duration::from_secs(10), rx.recv())
             .await
diff --git a/build/Dockerfile b/build/Dockerfile
index 29abc082..0c9be9e3 100644
--- a/build/Dockerfile
+++ b/build/Dockerfile
@@ -145,9 +145,23 @@ RUN pacman -Syu --noconfirm --needed \
 
 # groupadd -f so this is idempotent whether or not udev's rules already
 # created these.
+#
+# **Two users, and they must stay two.** `nestri` runs the services that come
+# with this image; `nesplay` is who a workload runs as. Sharing one user between
+# them is what lets workload code impersonate a service — it can replace the
+# socket a service listens on and answer in its place, and the answer that
+# matters is the address a client is told to connect to. Init refuses an address
+# served by the workload's own user, so a single shared user does not merely
+# weaken that check, it makes every session fail it.
+#
+# The uid a workload actually runs as is chosen by whoever asks for the box, not
+# here; this account exists so that the number has a home, a shell and a name in
+# `ps`, and so the separation has somewhere to be written down.
 RUN groupadd -f audio && groupadd -f video && groupadd -f input && groupadd -f render && \
     useradd -m -u 1000 -s /bin/bash nestri && \
-    for g in audio video input render; do gpasswd -a nestri "$g" >/dev/null; done
+    for g in audio video input render; do gpasswd -a nestri "$g" >/dev/null; done && \
+    useradd -m -u 1001 -s /bin/bash nesplay && \
+    for g in audio video input render; do gpasswd -a nesplay "$g" >/dev/null; done
 
 
 # ───────────────────────────────────────────────────────────