diff --git a/daemon/remote/README.md b/daemon/remote/README.md index 97fc9dbc4c47..1c02fe17fcac 100644 --- a/daemon/remote/README.md +++ b/daemon/remote/README.md @@ -62,6 +62,12 @@ Remote slot files: 3. `~/.cmux/daemon///daemon.lock` single-owner lock. 4. `~/.cmux/daemon///daemon.log` startup and crash diagnostics. +Each `serve --stdio --persistent` bridge includes a fresh `bridge_lease_id` in +its authenticated socket handshake. The persistent server tracks authenticated +bridge connections and lets the newest authenticated bridge take over the slot, +closing older connections (including half-open SSH bridges) without touching +the persistent PTY sessions. + PTY lifecycle: 1. A local attach creates or reuses a named `pty.*` session in the persistent daemon. 2. If the local surface closes, the stdio proxy disconnects and its attachment detaches, but the PTY process and bounded scrollback remain in the daemon. diff --git a/daemon/remote/cmd/cmuxd-remote/main.go b/daemon/remote/cmd/cmuxd-remote/main.go index 8de97e4e6dc3..a0b14b2a6109 100644 --- a/daemon/remote/cmd/cmuxd-remote/main.go +++ b/daemon/remote/cmd/cmuxd-remote/main.go @@ -924,7 +924,16 @@ func runPersistentStdioProxy(stdin io.Reader, stdout, stderr io.Writer, slot str if err := ensurePersistentDaemonRunning(paths, token, leasePort, stderr); err != nil { return err } - conn, err := dialPersistentDaemon(paths.socket, token) + bridgeLeaseID, err := newPersistentDaemonBridgeLeaseID() + if err != nil { + logPersistentDaemonEvent( + stderr, + "bridge_lease_generation_failed", + "error_category", persistentDaemonErrorCategory(err), + ) + return errors.New(persistentDaemonBridgeLeaseError) + } + conn, err := dialPersistentDaemonWithBridgeLease(paths.socket, token, bridgeLeaseID) if err != nil { return err } @@ -1230,6 +1239,7 @@ func servePersistentDaemonWithVerifierConfig( ) error { hub := newWebSocketPTYHub(wsPTYServerConfig{}, stderr) defer hub.closeAll() + bridgeLeases := newPersistentDaemonBridgeLeaseRegistry() var activeConnections int64 var idleSince time.Time var slotLeaseObserved bool @@ -1315,7 +1325,7 @@ func servePersistentDaemonWithVerifierConfig( atomic.AddInt64(&activeConnections, 1) go func() { defer atomic.AddInt64(&activeConnections, -1) - handlePersistentDaemonConn(conn, verifier, hub, stderr, requestShutdown) + handlePersistentDaemonConn(conn, verifier, hub, stderr, requestShutdown, bridgeLeases) }() } } @@ -1365,6 +1375,7 @@ func handlePersistentDaemonConn( hub *wsPTYHub, stderr io.Writer, requestShutdown func(), + bridgeLeases ...*persistentDaemonBridgeLeaseRegistry, ) { handlePersistentDaemonConnWithAuthTimeout( conn, @@ -1373,6 +1384,7 @@ func handlePersistentDaemonConn( stderr, persistentDaemonAuthTimeout, requestShutdown, + bridgeLeases..., ) } @@ -1383,9 +1395,16 @@ func handlePersistentDaemonConnWithAuthTimeout( stderr io.Writer, timeout time.Duration, requestShutdown func(), + bridgeLeases ...*persistentDaemonBridgeLeaseRegistry, ) { + var bridgeLeaseRegistry *persistentDaemonBridgeLeaseRegistry + if len(bridgeLeases) > 0 { + bridgeLeaseRegistry = bridgeLeases[0] + } defer conn.Close() + defer bridgeLeaseRegistry.release(conn) defer logPersistentDaemonEvent(stderr, "connection_closed") + bridgeLeaseRegistry.register(conn) logPersistentDaemonEvent(stderr, "connection_accepted") if timeout > 0 { if err := conn.SetDeadline(time.Now().Add(timeout)); err != nil { @@ -1400,7 +1419,20 @@ func handlePersistentDaemonConnWithAuthTimeout( } reader := bufio.NewReaderSize(conn, 64*1024) writer := &stdioFrameWriter{writer: bufio.NewWriter(conn)} - if err := authenticatePersistentDaemonConn(reader, writer, verifier); err != nil { + evictedBridgeConnections := 0 + if err := authenticatePersistentDaemonConnWithLease( + reader, + writer, + verifier, + func(leaseID string) error { + if bridgeLeaseRegistry == nil { + return nil + } + evicted, claimErr := bridgeLeaseRegistry.claim(conn, leaseID) + evictedBridgeConnections = evicted + return claimErr + }, + ); err != nil { logPersistentDaemonEvent( stderr, "connection_rejected", @@ -1408,6 +1440,13 @@ func handlePersistentDaemonConnWithAuthTimeout( ) return } + if evictedBridgeConnections > 0 { + logPersistentDaemonEvent( + stderr, + "bridge_lease_takeover", + "evicted_connections", strconv.Itoa(evictedBridgeConnections), + ) + } logPersistentDaemonEvent(stderr, "connection_authenticated") if timeout > 0 { if err := conn.SetDeadline(time.Time{}); err != nil { @@ -1446,12 +1485,23 @@ func persistentDaemonAuthenticationFailureReason(err error) string { "authentication method is invalid", "authentication token is invalid": return err.Error() + case "persistent daemon bridge lease rejected": + return "bridge_lease_rejected" default: return persistentDaemonErrorCategory(err) } } func authenticatePersistentDaemonConn(reader *bufio.Reader, writer *stdioFrameWriter, verifier persistentDaemonTokenVerifier) error { + return authenticatePersistentDaemonConnWithLease(reader, writer, verifier, nil) +} + +func authenticatePersistentDaemonConnWithLease( + reader *bufio.Reader, + writer *stdioFrameWriter, + verifier persistentDaemonTokenVerifier, + onBridgeLease func(string) error, +) error { line, oversized, err := readRPCFrame(reader, maxRPCFrameBytes) if err != nil || oversized { rejection := fmt.Errorf("authentication frame read failed: %w", err) @@ -1503,6 +1553,20 @@ func authenticatePersistentDaemonConn(reader *bufio.Reader, writer *stdioFrameWr }, }, errors.New("authentication token is invalid")) } + bridgeLeaseID, _ := getStringParam(req.Params, persistentDaemonBridgeLeaseParam) + bridgeLeaseID = strings.TrimSpace(bridgeLeaseID) + if bridgeLeaseID != "" && onBridgeLease != nil { + if err := onBridgeLease(bridgeLeaseID); err != nil { + return writePersistentDaemonAuthRejection(writer, rpcResponse{ + ID: req.ID, + OK: false, + Error: &rpcError{ + Code: "unauthorized", + Message: persistentDaemonBridgeLeaseError, + }, + }, errors.New("persistent daemon bridge lease rejected")) + } + } if err := writer.writeResponse(rpcResponse{ ID: req.ID, OK: true, @@ -1615,22 +1679,41 @@ func runRPCServerWithReader( } func dialPersistentDaemon(socketPath string, token string) (net.Conn, error) { - conn, err := net.DialTimeout("unix", socketPath, 2*time.Second) + return dialPersistentDaemonWithBridgeLease(socketPath, token, "") +} + +func dialPersistentDaemonWithBridgeLease(socketPath string, token string, bridgeLeaseID string) (net.Conn, error) { + dialer := net.Dialer{Timeout: 2 * time.Second} + conn, err := dialer.DialContext(context.Background(), "unix", socketPath) if err != nil { return nil, err } - if err := authenticatePersistentDaemonClient(conn, token); err != nil { + if err := authenticatePersistentDaemonClientWithBridgeLease(conn, token, bridgeLeaseID); err != nil { _ = conn.Close() return nil, err } return conn, nil } -func authenticatePersistentDaemonClient(conn net.Conn, token string) error { - return authenticatePersistentDaemonClientWithTimeout(conn, token, persistentDaemonAuthTimeout) +func authenticatePersistentDaemonClientWithTimeout(conn net.Conn, token string, timeout time.Duration) error { + return authenticatePersistentDaemonClientWithTimeoutAndBridgeLease(conn, token, timeout, "") } -func authenticatePersistentDaemonClientWithTimeout(conn net.Conn, token string, timeout time.Duration) error { +func authenticatePersistentDaemonClientWithBridgeLease(conn net.Conn, token string, bridgeLeaseID string) error { + return authenticatePersistentDaemonClientWithTimeoutAndBridgeLease( + conn, + token, + persistentDaemonAuthTimeout, + bridgeLeaseID, + ) +} + +func authenticatePersistentDaemonClientWithTimeoutAndBridgeLease( + conn net.Conn, + token string, + timeout time.Duration, + bridgeLeaseID string, +) error { if timeout > 0 { if err := conn.SetDeadline(time.Now().Add(timeout)); err != nil { return err @@ -1639,12 +1722,16 @@ func authenticatePersistentDaemonClientWithTimeout(conn net.Conn, token string, } writer := bufio.NewWriter(conn) + params := map[string]any{ + "token": token, + } + if normalizedLeaseID := strings.TrimSpace(bridgeLeaseID); normalizedLeaseID != "" { + params[persistentDaemonBridgeLeaseParam] = normalizedLeaseID + } request := rpcRequest{ ID: "auth", Method: persistentDaemonAuthMethod, - Params: map[string]any{ - "token": token, - }, + Params: params, } data, err := json.Marshal(request) if err != nil { @@ -1989,10 +2076,11 @@ func (s *rpcServer) handleProxyOpen(req rpcRequest) rpcResponse { timeoutMs = parsed } - conn, err := net.DialTimeout( + dialer := net.Dialer{Timeout: time.Duration(timeoutMs) * time.Millisecond} + conn, err := dialer.DialContext( + context.Background(), "tcp", net.JoinHostPort(host, strconv.Itoa(port)), - time.Duration(timeoutMs)*time.Millisecond, ) if err != nil { return rpcResponse{ diff --git a/daemon/remote/cmd/cmuxd-remote/persistent_bridge_lease.go b/daemon/remote/cmd/cmuxd-remote/persistent_bridge_lease.go new file mode 100644 index 000000000000..d4783a803abc --- /dev/null +++ b/daemon/remote/cmd/cmuxd-remote/persistent_bridge_lease.go @@ -0,0 +1,98 @@ +package main + +import ( + "crypto/rand" + "encoding/hex" + "errors" + "net" + "strings" + "sync" +) + +const ( + persistentDaemonBridgeLeaseParam = "bridge_lease_id" + persistentDaemonBridgeLeaseMax = 128 + persistentDaemonBridgeLeaseError = "persistent daemon connection could not be established; reconnect and try again" +) + +func newPersistentDaemonBridgeLeaseID() (string, error) { + raw := make([]byte, 16) + if _, err := rand.Read(raw); err != nil { + return "", err + } + return hex.EncodeToString(raw), nil +} + +// persistentDaemonBridgeLeaseRegistry tracks every accepted stdio connection +// and the one connection that most recently claimed the slot. A new claimed +// bridge is authoritative: it closes all older connections, which makes +// takeover work even when an older binary did not send a lease id. +type persistentDaemonBridgeLeaseRegistry struct { + mu sync.Mutex + connections map[net.Conn]struct{} + holder net.Conn +} + +func newPersistentDaemonBridgeLeaseRegistry() *persistentDaemonBridgeLeaseRegistry { + return &persistentDaemonBridgeLeaseRegistry{ + connections: make(map[net.Conn]struct{}), + } +} + +func (r *persistentDaemonBridgeLeaseRegistry) register(conn net.Conn) { + if r == nil || conn == nil { + return + } + r.mu.Lock() + if r.connections == nil { + r.connections = make(map[net.Conn]struct{}) + } + r.connections[conn] = struct{}{} + r.mu.Unlock() +} + +func (r *persistentDaemonBridgeLeaseRegistry) claim(conn net.Conn, leaseID string) (int, error) { + if r == nil || conn == nil { + return 0, errors.New("persistent daemon bridge lease registry is unavailable") + } + leaseID = strings.TrimSpace(leaseID) + if leaseID == "" { + return 0, errors.New("persistent daemon bridge lease id is required") + } + if len(leaseID) > persistentDaemonBridgeLeaseMax { + return 0, errors.New("persistent daemon bridge lease id is too long") + } + + r.mu.Lock() + if r.connections == nil { + r.connections = make(map[net.Conn]struct{}) + } + r.connections[conn] = struct{}{} + evicted := make([]net.Conn, 0, len(r.connections)-1) + for candidate := range r.connections { + if candidate == conn { + continue + } + evicted = append(evicted, candidate) + delete(r.connections, candidate) + } + r.holder = conn + r.mu.Unlock() + + for _, candidate := range evicted { + _ = candidate.Close() + } + return len(evicted), nil +} + +func (r *persistentDaemonBridgeLeaseRegistry) release(conn net.Conn) { + if r == nil || conn == nil { + return + } + r.mu.Lock() + delete(r.connections, conn) + if r.holder == conn { + r.holder = nil + } + r.mu.Unlock() +} diff --git a/daemon/remote/cmd/cmuxd-remote/persistent_bridge_lease_test.go b/daemon/remote/cmd/cmuxd-remote/persistent_bridge_lease_test.go new file mode 100644 index 000000000000..80fc0011d9c9 --- /dev/null +++ b/daemon/remote/cmd/cmuxd-remote/persistent_bridge_lease_test.go @@ -0,0 +1,150 @@ +package main + +import ( + "bufio" + "encoding/base64" + "errors" + "net" + "strings" + "testing" + "time" +) + +// A bridge whose SSH channel went half-open can remain authenticated on the +// remote Unix socket forever. The next authenticated bridge must be able to +// claim the slot and evict that stale connection before it starts forwarding. +func TestPersistentDaemonAuthenticatedBridgeLeaseTakeoverEvictsStaleHolder(t *testing.T) { + socketPath, stop := startPersistentDaemonForTest(t, "bridge-lease-token") + defer stop() + + stale, staleReader, staleWriter := openPersistentTestClientWithBridgeLease( + t, + socketPath, + "bridge-lease-token", + "stale-bridge", + ) + defer stale.Close() + staleAttach := persistentTestRPCCall(t, stale, staleReader, staleWriter, rpcRequest{ + ID: "stale-attach", + Method: "pty.attach", + Params: map[string]any{ + "session_id": "lease-preserved-session", + "attachment_id": "stale-attachment", + "client_attachment_token": "stale-attachment-token", + "cols": 80, + "rows": 24, + "command": "printf 'lease-preserved-data\\n'; sleep 60", + }, + }) + if staleAttach["ok"] != true { + t.Fatalf("stale bridge PTY attach failed: %v", staleAttach) + } + readPersistentTestEvent(t, stale, staleReader, func(frame map[string]any) bool { + return frame["event"] == "pty.ready" && frame["attachment_id"] == "stale-attachment" + }) + readPersistentTestEvent(t, stale, staleReader, func(frame map[string]any) bool { + if frame["event"] != "pty.data" || frame["attachment_id"] != "stale-attachment" { + return false + } + payload, decodeErr := base64.StdEncoding.DecodeString(frame["data_base64"].(string)) + return decodeErr == nil && strings.Contains(string(payload), "lease-preserved-data") + }) + + current, currentReader, currentWriter := openPersistentTestClientWithBridgeLease( + t, + socketPath, + "bridge-lease-token", + "replacement-bridge", + ) + defer current.Close() + + if err := stale.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatalf("set stale bridge read deadline: %v", err) + } + _, err := staleReader.ReadByte() + if err == nil { + t.Fatal("stale bridge remained connected after authenticated lease takeover") + } + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + t.Fatalf("stale bridge was not evicted before read deadline: %v", err) + } + // A Unix socket may report ECONNRESET instead of EOF when the server closes + // the connection; any non-timeout read failure is still an eviction. + reattach := persistentTestRPCCall(t, current, currentReader, currentWriter, rpcRequest{ + ID: "replacement-attach", + Method: "pty.attach", + Params: map[string]any{ + "session_id": "lease-preserved-session", + "attachment_id": "replacement-attachment", + "client_attachment_token": "replacement-attachment-token", + "cols": 100, + "rows": 30, + "require_existing": true, + }, + }) + if reattach["ok"] != true { + t.Fatalf("replacement bridge PTY reattach failed: %v", reattach) + } + result, _ := reattach["result"].(map[string]any) + if replayBytes, _ := result["replay_bytes"].(float64); replayBytes <= 0 { + t.Fatalf("replacement bridge replay_bytes = %v, want preserved output", result["replay_bytes"]) + } + readPersistentTestEvent(t, current, currentReader, func(frame map[string]any) bool { + return frame["event"] == "pty.ready" && frame["attachment_id"] == "replacement-attachment" + }) + readPersistentTestEvent(t, current, currentReader, func(frame map[string]any) bool { + if frame["event"] != "pty.data" || frame["attachment_id"] != "replacement-attachment" { + return false + } + payload, decodeErr := base64.StdEncoding.DecodeString(frame["data_base64"].(string)) + return decodeErr == nil && strings.Contains(string(payload), "lease-preserved-data") + }) +} + +func TestPersistentDaemonBridgeLeaseTakeoverEvictsLegacyAuthenticatedConnection(t *testing.T) { + socketPath, stop := startPersistentDaemonForTest(t, "legacy-bridge-token") + defer stop() + + legacy, legacyReader, _ := openPersistentTestClient(t, socketPath, "legacy-bridge-token") + defer legacy.Close() + replacement, _, _ := openPersistentTestClientWithBridgeLease( + t, + socketPath, + "legacy-bridge-token", + "replacement-bridge", + ) + defer replacement.Close() + + if err := legacy.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatalf("set legacy bridge read deadline: %v", err) + } + _, err := legacyReader.ReadByte() + if err == nil { + t.Fatal("legacy bridge remained connected after lease takeover") + } + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + t.Fatalf("legacy bridge was not evicted before read deadline: %v", err) + } +} + +func openPersistentTestClientWithBridgeLease( + t *testing.T, + socketPath string, + token string, + leaseID string, +) (net.Conn, *bufio.Reader, *bufio.Writer) { + t.Helper() + conn, err := net.Dial("unix", socketPath) + if err != nil { + t.Fatalf("dial persistent daemon: %v", err) + } + if err := authenticatePersistentDaemonClientWithBridgeLease(conn, token, leaseID); err != nil { + _ = conn.Close() + t.Fatalf("persistent daemon bridge lease auth failed: %v", err) + } + reader := bufio.NewReader(conn) + writer := bufio.NewWriter(conn) + return conn, reader, writer +} diff --git a/docs/remote-daemon-spec.md b/docs/remote-daemon-spec.md index c08a8c25917e..c43965fd00db 100644 --- a/docs/remote-daemon-spec.md +++ b/docs/remote-daemon-spec.md @@ -50,6 +50,7 @@ This is a **living implementation spec** (also called an **execution spec**): a - `DONE` bootstrap/probe failures surface actionable details. - `DONE` bootstrap installs `~/.cmux/bin/cmux` wrapper (also tries `/usr/local/bin/cmux`) so `cmux` is available in PATH on the remote. - `DONE` normal `cmux ssh` launches `cmuxd-remote serve --stdio --persistent --slot --persistent-lease-port `, where the stdio process proxies to a long-lived authenticated daemon with slot credentials under `~/.cmux/daemon///`, a short per-user socket path under `/tmp/cmuxd-remote-/`, and an exact relay-slot lease path. +- `DONE` each persistent stdio bridge claims its slot with a fresh authenticated `bridge_lease_id`; a newer bridge evicts older authenticated connections, including half-open SSH bridges, while the persistent PTY hub remains intact. - `DONE` persistent daemon slots advertise `pty.session.persistent_daemon`; cmux requires that capability before preserving a saved remote PTY session ID across app relaunch. - `DONE` clean workspace teardown verifies the relay's slot matches the workspace, sends an authenticated per-slot shutdown, waits a bounded interval for daemon ownership to release, removes relay shell state, and passively reaps disconnected, session-empty daemons whose exact previously observed relay slot lease disappears.