Make stream correlation a first-class seam (CorrelationID)

CONTEXT.md

@@ -0,0 +1,70 @@
+# mosey
+
+Domain glossary for mosey — a remote PTY attach system with pluggable
+transports and pluggable authentication. This file is the project's ubiquitous
+language: term definitions only, no implementation detail.
+
+## Capabilities
+
+**Caps**:
+The permission bitmask a peer holds over a session — some combination of Write,
+Resize, and Forge. Deliberately does *not* include ownership; an owner is a
+structural fact that implies the full cap set.
+_Avoid_: permissions, rights, scopes, capability bits (when you mean the type).
+
+**Write / Resize / Forge**:
+The three caps. Write: may send input to the PTY. Resize: may change the PTY
+geometry. Forge: may sign further off-chain delegations rooted at this grant.
+_Avoid_: type/input (for Write), resize-perm.
+
+**Owner**:
+A peer with full control of a session — mode switching, client management, and
+implicitly every Cap. Ownership is structural (who the session belongs to), not
+a Cap bit; it travels alongside Caps rather than inside the bitmask.
+_Avoid_: admin, root, superuser.
+
+**Grant**:
+The result of resolving a peer's authority: an Owner flag paired with a Caps
+set. What an Authenticator produces and what enforcement reads.
+_Avoid_: permission set, ACL entry.
+
+## Auth
+
+**Authenticator**:
+The pluggable credential model that runs the handshake and yields an Identity.
+Three exist: PSK, workspace cert, and wallet delegation.
+_Avoid_: auth provider, credential handler.
+
+**Identity**:
+The auth-layer result of a successful handshake — a Label plus the peer's Grant
+(Owner + Caps). The zero value means unauthenticated.
+_Avoid_: principal, user, subject.
+
+**Delegation**:
+A signed statement that attenuates authority from one wallet to another for a
+session — the wallet-auth building block folded into a chain to compute Caps.
+_Avoid_: grant (reserve that for the resolved Owner+Caps result), token.
+
+## Transport
+
+**Stream**:
+One bidirectional byte channel between two peers, handed out by a Transport
+backend (libp2p mux stream, HTTP/2 stream, WebSocket, unix socket). The unit the
+auth wrap, vterm, and attach client program against — never a specific backend.
+_Avoid_: connection, channel, socket.
+
+**CorrelationID**:
+An unauthenticated, per-dialer-instance handle a Stream carries so the auth wrap
+can *link* streams — recognise that a later application stream comes from the
+same party as an earlier auth-handshake stream. Stable for the dialer's lifetime
+and identical across every stream that dialer opens (granularity: one client
+process). It only *links* streams to an Identity the handshake already
+established; it never itself *grants* identity. Empty when a backend cannot
+correlate.
+_Avoid_: session, session id, peer id, auth token, RemoteID.
+
+**RemoteID**:
+A backend-specific, human-facing identifier for the remote peer (libp2p peer id,
+remote address, TLS subject) used only for log tagging. May be empty. Explicitly
+*not* an authentication or correlation input — that job belongs to CorrelationID.
+_Avoid_: correlation id, peer key, identity.

auth/wrap.go

@@ -17,7 +17,7 @@ import (
 // requested application protocol. Server-side: call [Wrapped.Serve]
 // once to install the /mosey/auth/ listener; subsequent application
 // Handle calls are gated on a successful prior handshake from the
-// same remote peer (matched by [transport.Stream.RemoteID]).
+// same remote peer (matched by [transport.Stream.CorrelationID]).
 //
 // Streams returned to handlers carry the peer's [Identity] —
 // retrieve it via [IdentityOf]. Streams from a peer that hasn't
@@ -39,7 +39,7 @@ type Wrapped struct {
 	auth  Authenticator
 
 	mu         sync.Mutex
-	identities map[string]Identity // keyed by remote id
+	identities map[string]Identity // keyed by stream CorrelationID
 
 	localIdentityMu sync.RWMutex
 	localIdentity   Identity // most recent ClientHandshake result
@@ -55,9 +55,17 @@ func (w *Wrapped) Handle(proto string, h transport.Handler) {
 		return
 	}
 	w.inner.Handle(proto, func(s transport.Stream) {
-		remote := s.RemoteID()
+		corr := s.CorrelationID()
+		if corr == "" {
+			// The backend can't correlate this stream to a prior
+			// handshake — fail closed. An empty key would otherwise
+			// alias every uncorrelatable stream together in the
+			// identity map. Silent close, same posture as below.
+			_ = s.Close()
+			return
+		}
 		w.mu.Lock()
-		id, ok := w.identities[remote]
+		id, ok := w.identities[corr]
 		w.mu.Unlock()
 		if !ok {
 			// No prior auth for this peer. Refuse silently — same
@@ -160,8 +168,16 @@ func (w *Wrapped) Serve() {
 			}
 			return
 		}
+		corr := s.CorrelationID()
+		if corr == "" {
+			// A handshake succeeded but the backend can't give us a
+			// correlation handle to key the identity on — so no
+			// application stream could ever be matched to it. Fail
+			// closed rather than store under an empty, aliasing key.
+			return
+		}
 		w.mu.Lock()
-		w.identities[s.RemoteID()] = identity
+		w.identities[corr] = identity
 		w.mu.Unlock()
 		// Identity is now observable. Ack so the dialer's
 		// io.ReadFull in Wrapped.Dial unblocks and proceeds.

auth/wrap_test.go

@@ -0,0 +1,161 @@
+package auth
+
+import (
+	"context"
+	"io"
+	"sync"
+	"testing"
+
+	"github.com/firefly-engineering/mosey/api"
+	"github.com/firefly-engineering/mosey/transport"
+)
+
+// fakeStream is a minimal [transport.Stream] whose CorrelationID is
+// fixed by the test. Read returns EOF; Write and the rest are inert
+// except that Close is observable.
+type fakeStream struct {
+	correlation string
+	remote      string
+
+	mu     sync.Mutex
+	closed bool
+}
+
+func (s *fakeStream) Read([]byte) (int, error)    { return 0, io.EOF }
+func (s *fakeStream) Write(p []byte) (int, error) { return len(p), nil }
+func (s *fakeStream) CloseWrite() error           { return nil }
+func (s *fakeStream) RemoteID() string            { return s.remote }
+func (s *fakeStream) CorrelationID() string       { return s.correlation }
+
+func (s *fakeStream) Close() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.closed = true
+	return nil
+}
+
+func (s *fakeStream) isClosed() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.closed
+}
+
+// fakeInner is a [transport.Transport] that only records the
+// handlers registered on it, so a test can deliver a stream of its
+// choosing to the handler auth.Wrap installed.
+type fakeInner struct {
+	mu       sync.Mutex
+	handlers map[string]transport.Handler
+}
+
+func newFakeInner() *fakeInner {
+	return &fakeInner{handlers: map[string]transport.Handler{}}
+}
+
+func (f *fakeInner) Schemes() []string   { return nil }
+func (f *fakeInner) Endpoints() []string { return nil }
+func (f *fakeInner) Handle(proto string, h transport.Handler) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.handlers[proto] = h
+}
+func (f *fakeInner) Unhandle(proto string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	delete(f.handlers, proto)
+}
+func (f *fakeInner) Dial(context.Context, string, string) (transport.Stream, error) {
+	return nil, io.EOF
+}
+func (f *fakeInner) Close() error { return nil }
+
+func (f *fakeInner) deliver(proto string, s transport.Stream) {
+	f.mu.Lock()
+	h := f.handlers[proto]
+	f.mu.Unlock()
+	if h != nil {
+		h(s)
+	}
+}
+
+// fakeAuth is an [Authenticator] whose ServerHandshake always
+// succeeds with a fixed identity.
+type fakeAuth struct{ id Identity }
+
+func (a fakeAuth) Name() string { return "fake" }
+func (a fakeAuth) ClientHandshake(context.Context, io.ReadWriteCloser) (Identity, error) {
+	return a.id, nil
+}
+func (a fakeAuth) ServerHandshake(context.Context, io.ReadWriteCloser) (Identity, error) {
+	return a.id, nil
+}
+
+// TestWrap_RefusesEmptyCorrelationOnAppStream proves the fail-closed
+// guard: an application stream whose CorrelationID is empty is closed
+// and never reaches the wrapped handler, even after a successful
+// handshake stored an identity under a real key.
+func TestWrap_RefusesEmptyCorrelationOnAppStream(t *testing.T) {
+	inner := newFakeInner()
+	w := Wrap(inner, fakeAuth{id: Identity{Label: "owner"}})
+	w.Serve()
+
+	// A real handshake stores an identity under a non-empty key.
+	w.Handle("/app", func(transport.Stream) {
+		t.Error("app handler ran for an empty-correlation stream")
+	})
+	inner.deliver(api.ProtoAuth, &fakeStream{correlation: "peer-1"})
+
+	// An application stream with no correlation must be refused.
+	empty := &fakeStream{correlation: ""}
+	inner.deliver("/app", empty)
+	if !empty.isClosed() {
+		t.Error("empty-correlation app stream was not closed")
+	}
+}
+
+// TestWrap_RefusesEmptyCorrelationOnHandshake proves the symmetric
+// guard: a handshake that succeeds but yields an empty correlation
+// stores nothing, so a later stream sharing that empty key is still
+// refused rather than inheriting the identity.
+func TestWrap_RefusesEmptyCorrelationOnHandshake(t *testing.T) {
+	inner := newFakeInner()
+	w := Wrap(inner, fakeAuth{id: Identity{Label: "owner"}})
+	w.Serve()
+
+	// Handshake succeeds but the stream can't correlate — store nothing.
+	inner.deliver(api.ProtoAuth, &fakeStream{correlation: ""})
+
+	w.Handle("/app", func(transport.Stream) {
+		t.Error("app handler ran despite no identity being stored")
+	})
+	app := &fakeStream{correlation: ""}
+	inner.deliver("/app", app)
+	if !app.isClosed() {
+		t.Error("app stream was not closed after empty-correlation handshake")
+	}
+}
+
+// TestWrap_CorrelatesMatchingStream is the positive control: an app
+// stream whose CorrelationID matches a prior handshake reaches the
+// handler carrying the stored identity.
+func TestWrap_CorrelatesMatchingStream(t *testing.T) {
+	inner := newFakeInner()
+	w := Wrap(inner, fakeAuth{id: Identity{Label: "owner"}})
+	w.Serve()
+
+	inner.deliver(api.ProtoAuth, &fakeStream{correlation: "peer-1"})
+
+	var got Identity
+	ran := false
+	w.Handle("/app", func(s transport.Stream) {
+		ran = true
+		got = IdentityOf(s)
+	})
+	inner.deliver("/app", &fakeStream{correlation: "peer-1"})
+	if !ran {
+		t.Fatal("app handler did not run for a correlated stream")
+	}
+	if got.Label != "owner" {
+		t.Errorf("identity Label = %q, want owner", got.Label)
+	}
+}

docs/adr/0004-stream-correlation-seam.md

@@ -0,0 +1,70 @@
+# Stream correlation is a first-class seam, distinct from RemoteID
+
+`auth.Wrap` runs its handshake on a separate `ProtoAuth` stream, stores the
+resulting `Identity` in a map, and then must recognise later application streams
+as belonging to the same party. It keyed that map on `transport.Stream.RemoteID()`
+— a field whose own doc says "used for log tagging; not for authentication." To
+satisfy this unnamed requirement, each backend quietly overloaded `RemoteID()`
+with a correlation value: websocket mints a random 128-bit per-process token and
+smuggles it through `Sec-WebSocket-Protocol`, unix returns `uid=N:pid=M` from
+SO_PEERCRED, http2 uses `RemoteAddr`, libp2p uses the peer id. The interface
+contradicted its use, and every new backend had to reverse-engineer what auth
+needed of it.
+
+We name the requirement instead of hiding it.
+
+## Decision
+
+- **Add `CorrelationID() string` to `transport.Stream`; keep `RemoteID()` for
+  logging.** Two methods, each with one job. `CorrelationID()` carries the auth
+  contract; `RemoteID()` reverts to the honest, possibly-empty, human-facing log
+  tag its doc already claimed it was.
+
+- **Correlation granularity is the dialer instance (one client process).** Two
+  streams with equal, non-empty `CorrelationID()` are guaranteed to come from the
+  same dialer, across as many connections as it opens. We rejected *per-connection*
+  (what the review's "same peer, this connection" phrasing implied — but websocket
+  deliberately correlates across connections, and tightening it would re-auth every
+  application stream) and *per-cryptographic-identity* (circular: for 3 of 4
+  backends the whole point of the handshake is that identity isn't known until
+  after correlation has already linked the streams).
+
+- **`CorrelationID` links; the handshake grants.** The handle is an
+  *unauthenticated* linking key. It never itself confers identity — it only ties an
+  application stream to an `Identity` the handshake already established. This is the
+  contract line that keeps a linking token from being mistaken for a credential.
+
+- **Fail closed on empty, at `auth.Wrap`.** `auth.Wrap` refuses to store an
+  Identity under an empty key and refuses any application stream whose
+  `CorrelationID()` is empty. "Cannot correlate ⇒ cannot authenticate" lives once,
+  at the consumer, not scattered across four backends. Backends may still
+  pre-refuse for their own reasons.
+
+- **Unforgeability is a stated contract, per backend.** A backend may return a
+  non-empty `CorrelationID` only if that value is unforgeable or unguessable within
+  its trust domain: cryptographic (libp2p peer id), kernel-attested (unix
+  SO_PEERCRED), or ≥128-bit random (websocket token). `RemoteAddr`-grade handles
+  (http2) are the weakest rung and correlate only per-connection — acceptable, but
+  future backends get a clear bar.
+
+## Scope
+
+Interface change only. **No wire or protocol change:** the websocket
+`mosey-peer-<token>` subprotocol vehicle and the unix peercreds derivation stay
+byte-identical; only where the value surfaces in Go moves (`RemoteID()` →
+`CorrelationID()`). `auth.Wrap` rekeys its map. The backend tests that today
+assert `RemoteID` stability/distinctness (`TestBackend_RemoteIDStableAcrossDials`
+and friends) migrate to assert `CorrelationID`, since that is the property they
+were always testing; `RemoteID` gets new, looser assertions.
+
+## Consequences
+
+- The `transport.Stream` interface stops contradicting its documented contract;
+  new backends implement a named requirement instead of guessing one.
+- Correlation policy — the empty-guard, the link-not-grant rule — sits at one
+  seam (`auth.Wrap`) rather than being implied by four backends.
+- **Known limitations, unchanged by this refactor and deferred:** the `auth.Wrap`
+  identity map never evicts, so (a) a websocket/unix peer's identity persists for
+  the wrapper's lifetime and (b) unix pid-reuse can let a recycled pid inherit a
+  dead process's identity. These pre-date this change; naming correlation does not
+  alter them. Map eviction is a separate decision.

transport/http2/http2.go

@@ -287,9 +287,10 @@ func (b *Backend) serveHTTP(w http.ResponseWriter, r *http.Request) {
 	flusher.Flush()
 
 	stream := &serverStream{
-		reader: r.Body,
-		writer: writeFlusher{w: w, f: flusher},
-		remote: r.RemoteAddr,
+		reader:      r.Body,
+		writer:      writeFlusher{w: w, f: flusher},
+		remote:      r.RemoteAddr,
+		correlation: r.RemoteAddr,
 	}
 	h(stream)
 }