detached f08a628

fix(protocol): harden server session channel

commit
f08a628dbb368e3133f2878e451a6c1c3da564bf
author
brice · · 2026-08-28 13:39 UTC
parent
4429457
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
11 files · +110 -44
Unified
UnifiedSplit
M internal/adapters/sessionwire/server.go +1 -1
@@ -136,7 +136,7 @@ func (c *serverConnection) Capabilities() protocol.ConnectionCapabilities {
136136 _, async := c.raw.(ports.AsyncTransport)
137137 _, synchronous := c.raw.(ports.OwnedSynchronousTransport)
138138 _, linkState := c.raw.(ports.LinkStateReporter)
139- window := uint8(8)
139+ window := uint8(protocol.MaxOutputWindow)
140140 if datagram {
141141 window = 1
142142 }
@@ -136,7 +136,7 @@ func (c *serverConnection) Capabilities() protocol.ConnectionCapabilities {
136 _, async := c.raw.(ports.AsyncTransport)136 _, async := c.raw.(ports.AsyncTransport)
137 _, synchronous := c.raw.(ports.OwnedSynchronousTransport)137 _, synchronous := c.raw.(ports.OwnedSynchronousTransport)
138 _, linkState := c.raw.(ports.LinkStateReporter)138 _, linkState := c.raw.(ports.LinkStateReporter)
139- window := uint8(8)139+ window := uint8(protocol.MaxOutputWindow)
140 if datagram {140 if datagram {
141 window = 1141 window = 1
142 }142 }
M internal/adapters/sessionwire/server_test.go +49 -17
@@ -108,33 +108,65 @@ func TestServerConnectionDecodesEveryClientMessage(t *testing.T) {
108108 func TestServerConnectionEncodesEveryServerMessage(t *testing.T) {
109109 exact := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "work"}
110110 lease := protocol.ParkedRouteLeaseID{1}
111+ welcome := protocol.Welcome{SessionID: "session"}
112+ errorMessage := protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "error"}
113+ output := protocol.Output{Epoch: 1, New: 1, Full: true, Size: domain.Size{Cols: 80, Rows: 24}}
114+ detached := protocol.Detached{Reason: protocol.ReasonDetach}
115+ pong := protocol.Pong{}
116+ sessions := protocol.Sessions{}
117+ commandResult := protocol.CommandResult{RequestID: 1, OK: true}
118+ navigation := protocol.NavigationDirective{Action: protocol.NavigationOpenHomePicker, LeaseID: lease}
119+ attachTarget := protocol.AttachTarget{Session: "work", Intent: protocol.IntentAttach}
120+ preview := protocol.RemotePreview{Version: protocol.RemotePreviewSchemaVersion, Status: protocol.RemotePreviewUnavailable}
121+ identity := protocol.CommittedRouteIdentity{Target: exact}
122+ routeAction := protocol.RouteNavigationAction{SnapshotGeneration: 1, Key: 1, Generation: 1}
123+ routeFailure := protocol.RouteNavigationFailure{Key: 1, Generation: 1, Code: protocol.RouteFailureUnavailable}
124+ routePosition := protocol.RoutePosition{Target: exact, ActiveTabID: "tab-1"}
125+ switchFailure := protocol.SamePeerSwitchFailure{RequestID: 1, Code: protocol.SamePeerSwitchUnavailable}
126+ parkedResponse := protocol.ParkedRouteResponse{RequestID: 1, Status: protocol.ParkedRouteReady}
127+
128+ outputPayload, err := wire.MarshalOutput(output)
129+ require.NoError(t, err)
130+ identityPayload, err := wire.MarshalCommittedRouteIdentity(identity)
131+ require.NoError(t, err)
132+ routeActionPayload, err := wire.MarshalRouteNavigationAction(routeAction)
133+ require.NoError(t, err)
134+ routeFailurePayload, err := wire.MarshalRouteNavigationFailure(routeFailure)
135+ require.NoError(t, err)
136+ routePositionPayload, err := wire.MarshalRoutePosition(routePosition)
137+ require.NoError(t, err)
138+ switchFailurePayload, err := wire.MarshalSamePeerSwitchFailure(switchFailure)
139+ require.NoError(t, err)
140+
111141 tests := []struct {
112142 name string
113143 message protocol.ServerMessage
114144 typeID wire.MsgType
145+ payload []byte
115146 }{
116- {name: "welcome", message: protocol.Welcome{SessionID: "session"}, typeID: wire.MsgWelcome},
117- {name: "error", message: protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "error"}, typeID: wire.MsgError},
118- {name: "output", message: protocol.Output{Epoch: 1, New: 1, Full: true, Size: domain.Size{Cols: 80, Rows: 24}}, typeID: wire.MsgOutput},
119- {name: "detached", message: protocol.Detached{Reason: protocol.ReasonDetach}, typeID: wire.MsgDetached},
120- {name: "pong", message: protocol.Pong{}, typeID: wire.MsgPong},
121- {name: "sessions", message: protocol.Sessions{}, typeID: wire.MsgSessions},
122- {name: "command result", message: protocol.CommandResult{RequestID: 1, OK: true}, typeID: wire.MsgCommandResult},
123- {name: "navigation", message: protocol.NavigationDirective{Action: protocol.NavigationOpenHomePicker, LeaseID: lease}, typeID: wire.MsgNavigationAction},
124- {name: "attach target", message: protocol.AttachTarget{Session: "work", Intent: protocol.IntentAttach}, typeID: wire.MsgAttachTarget},
125- {name: "preview", message: protocol.RemotePreview{Version: protocol.RemotePreviewSchemaVersion, Status: protocol.RemotePreviewUnavailable}, typeID: wire.MsgRemotePreviewResponse},
126- {name: "identity", message: protocol.CommittedRouteIdentity{Target: exact}, typeID: wire.MsgCommittedRouteIdentity},
127- {name: "route action", message: protocol.RouteNavigationAction{SnapshotGeneration: 1, Key: 1, Generation: 1}, typeID: wire.MsgNavigateRecentRoute},
128- {name: "route failure", message: protocol.RouteNavigationFailure{Key: 1, Generation: 1, Code: protocol.RouteFailureUnavailable}, typeID: wire.MsgRouteNavigationFailure},
129- {name: "route position", message: protocol.RoutePosition{Target: exact, ActiveTabID: "tab-1"}, typeID: wire.MsgRoutePosition},
130- {name: "switch failure", message: protocol.SamePeerSwitchFailure{RequestID: 1, Code: protocol.SamePeerSwitchUnavailable}, typeID: wire.MsgSamePeerSwitchFailure},
131- {name: "parked response", message: protocol.ParkedRouteResponse{RequestID: 1, Status: protocol.ParkedRouteReady}, typeID: wire.MsgParkedRouteResponse},
147+ {name: "welcome", message: welcome, typeID: wire.MsgWelcome, payload: wire.MarshalWelcome(welcome)},
148+ {name: "error", message: errorMessage, typeID: wire.MsgError, payload: wire.MarshalErrorMsg(errorMessage)},
149+ {name: "output", message: output, typeID: wire.MsgOutput, payload: outputPayload},
150+ {name: "detached", message: detached, typeID: wire.MsgDetached, payload: wire.MarshalDetached(detached)},
151+ {name: "pong", message: pong, typeID: wire.MsgPong, payload: wire.MarshalPong(pong)},
152+ {name: "sessions", message: sessions, typeID: wire.MsgSessions, payload: wire.MarshalSessions(sessions)},
153+ {name: "command result", message: commandResult, typeID: wire.MsgCommandResult, payload: wire.MarshalCommandResult(commandResult)},
154+ {name: "navigation", message: navigation, typeID: wire.MsgNavigationAction, payload: wire.MarshalNavigationDirective(navigation)},
155+ {name: "attach target", message: attachTarget, typeID: wire.MsgAttachTarget, payload: wire.MarshalAttachTarget(attachTarget)},
156+ {name: "preview", message: preview, typeID: wire.MsgRemotePreviewResponse, payload: wire.MarshalRemotePreview(preview)},
157+ {name: "identity", message: identity, typeID: wire.MsgCommittedRouteIdentity, payload: identityPayload},
158+ {name: "route action", message: routeAction, typeID: wire.MsgNavigateRecentRoute, payload: routeActionPayload},
159+ {name: "route failure", message: routeFailure, typeID: wire.MsgRouteNavigationFailure, payload: routeFailurePayload},
160+ {name: "route position", message: routePosition, typeID: wire.MsgRoutePosition, payload: routePositionPayload},
161+ {name: "switch failure", message: switchFailure, typeID: wire.MsgSamePeerSwitchFailure, payload: switchFailurePayload},
162+ {name: "parked response", message: parkedResponse, typeID: wire.MsgParkedRouteResponse, payload: wire.MarshalParkedRouteResponse(parkedResponse)},
132163 }
133164 for _, tt := range tests {
134165 t.Run(tt.name, func(t *testing.T) {
135166 raw := &scriptedTransport{}
136167 require.NoError(t, NewServerConnection(raw).SendServer(tt.message))
137168 require.Equal(t, tt.typeID, raw.sent.Type)
169+ require.Equal(t, tt.payload, raw.sent.Payload)
138170 })
139171 }
140172 }
@@ -218,7 +250,7 @@ func TestServerConnectionPreservesSendModesCapabilitiesAndErrors(t *testing.T) {
218250 require.ErrorIs(t, plain.SendServerSynchronous(message), ErrUnsupportedSend)
219251 require.ErrorIs(t, plain.SendOutputAsync(output), ErrUnsupportedSend)
220252 require.ErrorIs(t, plain.SendOutputSynchronous(output), ErrUnsupportedSend)
221- require.Equal(t, uint8(8), plain.Capabilities().PreferredOutputWindow)
253+ require.Equal(t, uint8(protocol.MaxOutputWindow), plain.Capabilities().PreferredOutputWindow)
222254 require.Equal(t, ports.LinkStateConnected, plain.LinkState())
223255 require.Nil(t, plain.LinkEvents())
224256 }
@@ -108,33 +108,65 @@ func TestServerConnectionDecodesEveryClientMessage(t *testing.T) {
108 func TestServerConnectionEncodesEveryServerMessage(t *testing.T) {108 func TestServerConnectionEncodesEveryServerMessage(t *testing.T) {
109 exact := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "work"}109 exact := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{1}, SessionName: "work"}
110 lease := protocol.ParkedRouteLeaseID{1}110 lease := protocol.ParkedRouteLeaseID{1}
111+ welcome := protocol.Welcome{SessionID: "session"}
112+ errorMessage := protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "error"}
113+ output := protocol.Output{Epoch: 1, New: 1, Full: true, Size: domain.Size{Cols: 80, Rows: 24}}
114+ detached := protocol.Detached{Reason: protocol.ReasonDetach}
115+ pong := protocol.Pong{}
116+ sessions := protocol.Sessions{}
117+ commandResult := protocol.CommandResult{RequestID: 1, OK: true}
118+ navigation := protocol.NavigationDirective{Action: protocol.NavigationOpenHomePicker, LeaseID: lease}
119+ attachTarget := protocol.AttachTarget{Session: "work", Intent: protocol.IntentAttach}
120+ preview := protocol.RemotePreview{Version: protocol.RemotePreviewSchemaVersion, Status: protocol.RemotePreviewUnavailable}
121+ identity := protocol.CommittedRouteIdentity{Target: exact}
122+ routeAction := protocol.RouteNavigationAction{SnapshotGeneration: 1, Key: 1, Generation: 1}
123+ routeFailure := protocol.RouteNavigationFailure{Key: 1, Generation: 1, Code: protocol.RouteFailureUnavailable}
124+ routePosition := protocol.RoutePosition{Target: exact, ActiveTabID: "tab-1"}
125+ switchFailure := protocol.SamePeerSwitchFailure{RequestID: 1, Code: protocol.SamePeerSwitchUnavailable}
126+ parkedResponse := protocol.ParkedRouteResponse{RequestID: 1, Status: protocol.ParkedRouteReady}
127+
128+ outputPayload, err := wire.MarshalOutput(output)
129+ require.NoError(t, err)
130+ identityPayload, err := wire.MarshalCommittedRouteIdentity(identity)
131+ require.NoError(t, err)
132+ routeActionPayload, err := wire.MarshalRouteNavigationAction(routeAction)
133+ require.NoError(t, err)
134+ routeFailurePayload, err := wire.MarshalRouteNavigationFailure(routeFailure)
135+ require.NoError(t, err)
136+ routePositionPayload, err := wire.MarshalRoutePosition(routePosition)
137+ require.NoError(t, err)
138+ switchFailurePayload, err := wire.MarshalSamePeerSwitchFailure(switchFailure)
139+ require.NoError(t, err)
140+
111 tests := []struct {141 tests := []struct {
112 name string142 name string
113 message protocol.ServerMessage143 message protocol.ServerMessage
114 typeID wire.MsgType144 typeID wire.MsgType
145+ payload []byte
115 }{146 }{
116- {name: "welcome", message: protocol.Welcome{SessionID: "session"}, typeID: wire.MsgWelcome},147+ {name: "welcome", message: welcome, typeID: wire.MsgWelcome, payload: wire.MarshalWelcome(welcome)},
117- {name: "error", message: protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "error"}, typeID: wire.MsgError},148+ {name: "error", message: errorMessage, typeID: wire.MsgError, payload: wire.MarshalErrorMsg(errorMessage)},
118- {name: "output", message: protocol.Output{Epoch: 1, New: 1, Full: true, Size: domain.Size{Cols: 80, Rows: 24}}, typeID: wire.MsgOutput},149+ {name: "output", message: output, typeID: wire.MsgOutput, payload: outputPayload},
119- {name: "detached", message: protocol.Detached{Reason: protocol.ReasonDetach}, typeID: wire.MsgDetached},150+ {name: "detached", message: detached, typeID: wire.MsgDetached, payload: wire.MarshalDetached(detached)},
120- {name: "pong", message: protocol.Pong{}, typeID: wire.MsgPong},151+ {name: "pong", message: pong, typeID: wire.MsgPong, payload: wire.MarshalPong(pong)},
121- {name: "sessions", message: protocol.Sessions{}, typeID: wire.MsgSessions},152+ {name: "sessions", message: sessions, typeID: wire.MsgSessions, payload: wire.MarshalSessions(sessions)},
122- {name: "command result", message: protocol.CommandResult{RequestID: 1, OK: true}, typeID: wire.MsgCommandResult},153+ {name: "command result", message: commandResult, typeID: wire.MsgCommandResult, payload: wire.MarshalCommandResult(commandResult)},
123- {name: "navigation", message: protocol.NavigationDirective{Action: protocol.NavigationOpenHomePicker, LeaseID: lease}, typeID: wire.MsgNavigationAction},154+ {name: "navigation", message: navigation, typeID: wire.MsgNavigationAction, payload: wire.MarshalNavigationDirective(navigation)},
124- {name: "attach target", message: protocol.AttachTarget{Session: "work", Intent: protocol.IntentAttach}, typeID: wire.MsgAttachTarget},155+ {name: "attach target", message: attachTarget, typeID: wire.MsgAttachTarget, payload: wire.MarshalAttachTarget(attachTarget)},
125- {name: "preview", message: protocol.RemotePreview{Version: protocol.RemotePreviewSchemaVersion, Status: protocol.RemotePreviewUnavailable}, typeID: wire.MsgRemotePreviewResponse},156+ {name: "preview", message: preview, typeID: wire.MsgRemotePreviewResponse, payload: wire.MarshalRemotePreview(preview)},
126- {name: "identity", message: protocol.CommittedRouteIdentity{Target: exact}, typeID: wire.MsgCommittedRouteIdentity},157+ {name: "identity", message: identity, typeID: wire.MsgCommittedRouteIdentity, payload: identityPayload},
127- {name: "route action", message: protocol.RouteNavigationAction{SnapshotGeneration: 1, Key: 1, Generation: 1}, typeID: wire.MsgNavigateRecentRoute},158+ {name: "route action", message: routeAction, typeID: wire.MsgNavigateRecentRoute, payload: routeActionPayload},
128- {name: "route failure", message: protocol.RouteNavigationFailure{Key: 1, Generation: 1, Code: protocol.RouteFailureUnavailable}, typeID: wire.MsgRouteNavigationFailure},159+ {name: "route failure", message: routeFailure, typeID: wire.MsgRouteNavigationFailure, payload: routeFailurePayload},
129- {name: "route position", message: protocol.RoutePosition{Target: exact, ActiveTabID: "tab-1"}, typeID: wire.MsgRoutePosition},160+ {name: "route position", message: routePosition, typeID: wire.MsgRoutePosition, payload: routePositionPayload},
130- {name: "switch failure", message: protocol.SamePeerSwitchFailure{RequestID: 1, Code: protocol.SamePeerSwitchUnavailable}, typeID: wire.MsgSamePeerSwitchFailure},161+ {name: "switch failure", message: switchFailure, typeID: wire.MsgSamePeerSwitchFailure, payload: switchFailurePayload},
131- {name: "parked response", message: protocol.ParkedRouteResponse{RequestID: 1, Status: protocol.ParkedRouteReady}, typeID: wire.MsgParkedRouteResponse},162+ {name: "parked response", message: parkedResponse, typeID: wire.MsgParkedRouteResponse, payload: wire.MarshalParkedRouteResponse(parkedResponse)},
132 }163 }
133 for _, tt := range tests {164 for _, tt := range tests {
134 t.Run(tt.name, func(t *testing.T) {165 t.Run(tt.name, func(t *testing.T) {
135 raw := &scriptedTransport{}166 raw := &scriptedTransport{}
136 require.NoError(t, NewServerConnection(raw).SendServer(tt.message))167 require.NoError(t, NewServerConnection(raw).SendServer(tt.message))
137 require.Equal(t, tt.typeID, raw.sent.Type)168 require.Equal(t, tt.typeID, raw.sent.Type)
169+ require.Equal(t, tt.payload, raw.sent.Payload)
138 })170 })
139 }171 }
140 }172 }
@@ -218,7 +250,7 @@ func TestServerConnectionPreservesSendModesCapabilitiesAndErrors(t *testing.T) {
218 require.ErrorIs(t, plain.SendServerSynchronous(message), ErrUnsupportedSend)250 require.ErrorIs(t, plain.SendServerSynchronous(message), ErrUnsupportedSend)
219 require.ErrorIs(t, plain.SendOutputAsync(output), ErrUnsupportedSend)251 require.ErrorIs(t, plain.SendOutputAsync(output), ErrUnsupportedSend)
220 require.ErrorIs(t, plain.SendOutputSynchronous(output), ErrUnsupportedSend)252 require.ErrorIs(t, plain.SendOutputSynchronous(output), ErrUnsupportedSend)
221- require.Equal(t, uint8(8), plain.Capabilities().PreferredOutputWindow)253+ require.Equal(t, uint8(protocol.MaxOutputWindow), plain.Capabilities().PreferredOutputWindow)
222 require.Equal(t, ports.LinkStateConnected, plain.LinkState())254 require.Equal(t, ports.LinkStateConnected, plain.LinkState())
223 require.Nil(t, plain.LinkEvents())255 require.Nil(t, plain.LinkEvents())
224 }256 }
M internal/app/daemon_integration_test.go +22 -0
@@ -5,6 +5,7 @@ package app
55 import (
66 "context"
77 "crypto/rand"
8+ "encoding/binary"
89 "errors"
910 "io"
1011 "log/slog"
@@ -136,6 +137,27 @@ func killAll(dir string) error {
136137 return err
137138 }
138139
140+func TestIntegration_MalformedCommandPreservesVersionAndRequestID(t *testing.T) {
141+ dir, _ := startDaemon(t)
142+ tr, err := ipc.DialContext(context.Background(), dir)
143+ require.NoError(t, err)
144+ defer func() { _ = tr.Close() }()
145+
146+ payload := make([]byte, 10)
147+ binary.BigEndian.PutUint16(payload[:2], protocol.Version+1)
148+ binary.BigEndian.PutUint64(payload[2:], 42)
149+ require.NoError(t, tr.Send(wire.Frame{Type: wire.MsgCommand, Payload: payload}))
150+
151+ frame, err := tr.Recv()
152+ require.NoError(t, err)
153+ require.Equal(t, wire.MsgCommandResult, frame.Type)
154+ result, err := wire.UnmarshalCommandResult(frame.Payload)
155+ require.NoError(t, err)
156+ require.False(t, result.OK)
157+ require.Equal(t, uint64(42), result.RequestID)
158+ require.Equal(t, protocol.ErrVersionMismatch, result.Code)
159+}
160+
139161 // awaitText decodes MsgOutput frames into a fresh VT screen and returns once
140162 // the reconstructed grid contains want.
141163 func awaitText(t *testing.T, p *pump, sz domain.Size, want string) {
@@ -5,6 +5,7 @@ package app
5 import (5 import (
6 "context"6 "context"
7 "crypto/rand"7 "crypto/rand"
8+ "encoding/binary"
8 "errors"9 "errors"
9 "io"10 "io"
10 "log/slog"11 "log/slog"
@@ -136,6 +137,27 @@ func killAll(dir string) error {
136 return err137 return err
137 }138 }
138 139
140+func TestIntegration_MalformedCommandPreservesVersionAndRequestID(t *testing.T) {
141+ dir, _ := startDaemon(t)
142+ tr, err := ipc.DialContext(context.Background(), dir)
143+ require.NoError(t, err)
144+ defer func() { _ = tr.Close() }()
145+
146+ payload := make([]byte, 10)
147+ binary.BigEndian.PutUint16(payload[:2], protocol.Version+1)
148+ binary.BigEndian.PutUint64(payload[2:], 42)
149+ require.NoError(t, tr.Send(wire.Frame{Type: wire.MsgCommand, Payload: payload}))
150+
151+ frame, err := tr.Recv()
152+ require.NoError(t, err)
153+ require.Equal(t, wire.MsgCommandResult, frame.Type)
154+ result, err := wire.UnmarshalCommandResult(frame.Payload)
155+ require.NoError(t, err)
156+ require.False(t, result.OK)
157+ require.Equal(t, uint64(42), result.RequestID)
158+ require.Equal(t, protocol.ErrVersionMismatch, result.Code)
159+}
160+
139 // awaitText decodes MsgOutput frames into a fresh VT screen and returns once161 // awaitText decodes MsgOutput frames into a fresh VT screen and returns once
140 // the reconstructed grid contains want.162 // the reconstructed grid contains want.
141 func awaitText(t *testing.T, p *pump, sz domain.Size, want string) {163 func awaitText(t *testing.T, p *pump, sz domain.Size, want string) {
M internal/protocol/session.go +3 -0
@@ -23,6 +23,9 @@ const HandshakeTimeout = 15 * time.Second
2323 // output publication in protocol version 37.
2424 const MaxOutputDataLen = (16 << 20) - 55
2525
26+// MaxOutputWindow caps output states sent before client acknowledgement.
27+const MaxOutputWindow = 8
28+
2629 // ConnectionCapabilities describes transport behavior relevant to session
2730 // flow without naming a concrete carriage.
2831 type ConnectionCapabilities struct {
@@ -23,6 +23,9 @@ const HandshakeTimeout = 15 * time.Second
23 // output publication in protocol version 37.23 // output publication in protocol version 37.
24 const MaxOutputDataLen = (16 << 20) - 5524 const MaxOutputDataLen = (16 << 20) - 55
25 25
26+// MaxOutputWindow caps output states sent before client acknowledgement.
27+const MaxOutputWindow = 8
28+
26 // ConnectionCapabilities describes transport behavior relevant to session29 // ConnectionCapabilities describes transport behavior relevant to session
27 // flow without naming a concrete carriage.30 // flow without naming a concrete carriage.
28 type ConnectionCapabilities struct {31 type ConnectionCapabilities struct {
M internal/usecase/daemon/client_frame_routing.go +7 -0
@@ -53,6 +53,13 @@ func (d *Daemon) runConnLoop(ac *attachedClient) {
5353 if err != nil {
5454 var failure *protocol.DecodeFailure
5555 if errors.As(err, &failure) {
56+ d.log.Warn("rejected client message",
57+ "category", failure.Category,
58+ "kind", failure.Kind,
59+ "type", failure.Type,
60+ "version", failure.Version,
61+ "request", failure.RequestID,
62+ )
5663 continue
5764 }
5865 for range connectionSnapshotAttempts {
@@ -53,6 +53,13 @@ func (d *Daemon) runConnLoop(ac *attachedClient) {
53 if err != nil {53 if err != nil {
54 var failure *protocol.DecodeFailure54 var failure *protocol.DecodeFailure
55 if errors.As(err, &failure) {55 if errors.As(err, &failure) {
56+ d.log.Warn("rejected client message",
57+ "category", failure.Category,
58+ "kind", failure.Kind,
59+ "type", failure.Type,
60+ "version", failure.Version,
61+ "request", failure.RequestID,
62+ )
56 continue63 continue
57 }64 }
58 for range connectionSnapshotAttempts {65 for range connectionSnapshotAttempts {
M internal/usecase/daemon/control_test.go +0 -14
@@ -194,20 +194,6 @@ func TestHandleCommandAttachedRejectionReturnsSendFailure(t *testing.T) {
194194 require.True(t, tr.closed, "the failed one-shot transport must still close")
195195 }
196196
197-func TestHandleCommandRejectsVersionBeforeDecodeOrDispatch(t *testing.T) {
198- // The version prefix is valid and mismatched, while the remainder is not a
199- // decodable request. The daemon must still return ErrVersionMismatch.
200- frame := wire.Frame{Type: wire.MsgCommand, Payload: []byte{0, byte(protocol.Version + 1)}}
201- tr, sends, _ := newConn(t, frame)
202-
203- d := newTestDaemon(t, nil, stubClock{})
204- require.NoError(t, d.handleCommandFrame(tr, frame))
205-
206- result := awaitCommandResult(t, sends)
207- require.False(t, result.OK)
208- require.Equal(t, protocol.ErrVersionMismatch, result.Code)
209-}
210-
211197 func TestHandleConnRoutesCommand(t *testing.T) {
212198 d := newTestDaemon(t, nil, stubClock{})
213199 request := protocol.CommandRequest{Version: protocol.Version, Slug: "list-sessions"}
@@ -194,20 +194,6 @@ func TestHandleCommandAttachedRejectionReturnsSendFailure(t *testing.T) {
194 require.True(t, tr.closed, "the failed one-shot transport must still close")194 require.True(t, tr.closed, "the failed one-shot transport must still close")
195 }195 }
196 196
197-func TestHandleCommandRejectsVersionBeforeDecodeOrDispatch(t *testing.T) {
198- // The version prefix is valid and mismatched, while the remainder is not a
199- // decodable request. The daemon must still return ErrVersionMismatch.
200- frame := wire.Frame{Type: wire.MsgCommand, Payload: []byte{0, byte(protocol.Version + 1)}}
201- tr, sends, _ := newConn(t, frame)
202-
203- d := newTestDaemon(t, nil, stubClock{})
204- require.NoError(t, d.handleCommandFrame(tr, frame))
205-
206- result := awaitCommandResult(t, sends)
207- require.False(t, result.OK)
208- require.Equal(t, protocol.ErrVersionMismatch, result.Code)
209-}
210-
211 func TestHandleConnRoutesCommand(t *testing.T) {197 func TestHandleConnRoutesCommand(t *testing.T) {
212 d := newTestDaemon(t, nil, stubClock{})198 d := newTestDaemon(t, nil, stubClock{})
213 request := protocol.CommandRequest{Version: protocol.Version, Slug: "list-sessions"}199 request := protocol.CommandRequest{Version: protocol.Version, Slug: "list-sessions"}
M internal/usecase/daemon/daemon.go +4 -4
@@ -54,7 +54,7 @@ const detachNotifyTimeout = time.Second
5454 // heavy output degrades to lower fps on a slow link instead of overflowing the
5555 // transport. It must stay well under the datagram carriage's 32-frame client
5656 // window so its reliable queue never fills from painting alone.
57-const maxUnackedOutputStates = 8
57+const maxUnackedOutputStates = protocol.MaxOutputWindow
5858
5959 // normalizeOutputWindow bounds the untrusted Hello value. Zero deliberately
6060 // means the legacy/default window, so malformed or absent values remain safe.
@@ -1082,7 +1082,7 @@ func (d *Daemon) handleInitialDecodeFailure(ctx context.Context, tr ports.Server
10821082 }
10831083 send(serverError(protocol.ErrInternal, "malformed hello"))
10841084 case protocol.DecodeMessageCommand:
1085- code, text := uint16(protocol.ErrInternal), "malformed command request"
1085+ code, text := protocol.ErrInternal, "malformed command request"
10861086 if failure.Version != 0 && failure.Version != protocol.Version {
10871087 code, text = protocol.ErrVersionMismatch, "protocol version mismatch"
10881088 }
@@ -1187,14 +1187,14 @@ func (d *Daemon) handleHello(tr ports.ServerConnection, hello protocol.Hello) {
11871187 // handleHelloWithContext runs the attach handshake: version check, intent
11881188 // routing, Welcome, guaranteed first paint, then the per-connection input loop.
11891189 func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut <-chan struct{}, stopHandshakeTransport, finishHandshake func(), tr ports.ServerConnection, h protocol.Hello) {
1190- if err := handshakeContextError(context.Background(), timedOut, handshakeCtx.Err()); err != nil {
1190+ if err := handshakeContextError(d.hardCtx, timedOut, handshakeCtx.Err()); err != nil {
11911191 _ = tr.Close()
11921192 return
11931193 }
11941194 sendHandshake := func(message protocol.ServerMessage) error {
11951195 err := boundedHandshakeOperation(handshakeCtx, tr, func() error { return tr.SendServer(message) })
11961196 if err != nil {
1197- return handshakeContextError(context.Background(), timedOut, err)
1197+ return handshakeContextError(d.hardCtx, timedOut, err)
11981198 }
11991199 return nil
12001200 }
@@ -54,7 +54,7 @@ const detachNotifyTimeout = time.Second
54 // heavy output degrades to lower fps on a slow link instead of overflowing the54 // heavy output degrades to lower fps on a slow link instead of overflowing the
55 // transport. It must stay well under the datagram carriage's 32-frame client55 // transport. It must stay well under the datagram carriage's 32-frame client
56 // window so its reliable queue never fills from painting alone.56 // window so its reliable queue never fills from painting alone.
57-const maxUnackedOutputStates = 857+const maxUnackedOutputStates = protocol.MaxOutputWindow
58 58
59 // normalizeOutputWindow bounds the untrusted Hello value. Zero deliberately59 // normalizeOutputWindow bounds the untrusted Hello value. Zero deliberately
60 // means the legacy/default window, so malformed or absent values remain safe.60 // means the legacy/default window, so malformed or absent values remain safe.
@@ -1082,7 +1082,7 @@ func (d *Daemon) handleInitialDecodeFailure(ctx context.Context, tr ports.Server
1082 }1082 }
1083 send(serverError(protocol.ErrInternal, "malformed hello"))1083 send(serverError(protocol.ErrInternal, "malformed hello"))
1084 case protocol.DecodeMessageCommand:1084 case protocol.DecodeMessageCommand:
1085- code, text := uint16(protocol.ErrInternal), "malformed command request"1085+ code, text := protocol.ErrInternal, "malformed command request"
1086 if failure.Version != 0 && failure.Version != protocol.Version {1086 if failure.Version != 0 && failure.Version != protocol.Version {
1087 code, text = protocol.ErrVersionMismatch, "protocol version mismatch"1087 code, text = protocol.ErrVersionMismatch, "protocol version mismatch"
1088 }1088 }
@@ -1187,14 +1187,14 @@ func (d *Daemon) handleHello(tr ports.ServerConnection, hello protocol.Hello) {
1187 // handleHelloWithContext runs the attach handshake: version check, intent1187 // handleHelloWithContext runs the attach handshake: version check, intent
1188 // routing, Welcome, guaranteed first paint, then the per-connection input loop.1188 // routing, Welcome, guaranteed first paint, then the per-connection input loop.
1189 func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut <-chan struct{}, stopHandshakeTransport, finishHandshake func(), tr ports.ServerConnection, h protocol.Hello) {1189 func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut <-chan struct{}, stopHandshakeTransport, finishHandshake func(), tr ports.ServerConnection, h protocol.Hello) {
1190- if err := handshakeContextError(context.Background(), timedOut, handshakeCtx.Err()); err != nil {1190+ if err := handshakeContextError(d.hardCtx, timedOut, handshakeCtx.Err()); err != nil {
1191 _ = tr.Close()1191 _ = tr.Close()
1192 return1192 return
1193 }1193 }
1194 sendHandshake := func(message protocol.ServerMessage) error {1194 sendHandshake := func(message protocol.ServerMessage) error {
1195 err := boundedHandshakeOperation(handshakeCtx, tr, func() error { return tr.SendServer(message) })1195 err := boundedHandshakeOperation(handshakeCtx, tr, func() error { return tr.SendServer(message) })
1196 if err != nil {1196 if err != nil {
1197- return handshakeContextError(context.Background(), timedOut, err)1197+ return handshakeContextError(d.hardCtx, timedOut, err)
1198 }1198 }
1199 return nil1199 return nil
1200 }1200 }
M internal/usecase/daemon/notify_test.go +6 -0
@@ -1,6 +1,7 @@
11 package daemon
22
33 import (
4+ "bytes"
45 "context"
56 "errors"
67 "fmt"
@@ -53,6 +54,8 @@ func TestClientNoticeMapsFixedActionsAndDismissesOnlyConnectionToast(t *testing.
5354
5455 func TestMalformedClientNoticeIsIgnored(t *testing.T) {
5556 d, sess, ac, _ := newNoticeFixture(t, &noticeClock{})
57+ var logs bytes.Buffer
58+ d.log = slog.New(slog.NewTextHandler(&logs, nil))
5659 d.attachCoordinator(sess, nil, ac, true)
5760 tr, ok := ac.tr.(*mockServerConnection)
5861 require.True(t, ok, "attached client connection must use the test wrapper")
@@ -69,6 +72,9 @@ func TestMalformedClientNoticeIsIgnored(t *testing.T) {
6972 toasts, _ := visibleToasts(ac)
7073 require.Empty(t, toasts)
7174 require.Empty(t, d.notices.history())
75+ require.Contains(t, logs.String(), "rejected client message")
76+ require.Contains(t, logs.String(), "category=1")
77+ require.Contains(t, logs.String(), "type=11")
7278 }
7379
7480 func stripNoticeTimes(notices []domain.Notification) []domain.Notification {
@@ -1,6 +1,7 @@
1 package daemon1 package daemon
2 2
3 import (3 import (
4+ "bytes"
4 "context"5 "context"
5 "errors"6 "errors"
6 "fmt"7 "fmt"
@@ -53,6 +54,8 @@ func TestClientNoticeMapsFixedActionsAndDismissesOnlyConnectionToast(t *testing.
53 54
54 func TestMalformedClientNoticeIsIgnored(t *testing.T) {55 func TestMalformedClientNoticeIsIgnored(t *testing.T) {
55 d, sess, ac, _ := newNoticeFixture(t, &noticeClock{})56 d, sess, ac, _ := newNoticeFixture(t, &noticeClock{})
57+ var logs bytes.Buffer
58+ d.log = slog.New(slog.NewTextHandler(&logs, nil))
56 d.attachCoordinator(sess, nil, ac, true)59 d.attachCoordinator(sess, nil, ac, true)
57 tr, ok := ac.tr.(*mockServerConnection)60 tr, ok := ac.tr.(*mockServerConnection)
58 require.True(t, ok, "attached client connection must use the test wrapper")61 require.True(t, ok, "attached client connection must use the test wrapper")
@@ -69,6 +72,9 @@ func TestMalformedClientNoticeIsIgnored(t *testing.T) {
69 toasts, _ := visibleToasts(ac)72 toasts, _ := visibleToasts(ac)
70 require.Empty(t, toasts)73 require.Empty(t, toasts)
71 require.Empty(t, d.notices.history())74 require.Empty(t, d.notices.history())
75+ require.Contains(t, logs.String(), "rejected client message")
76+ require.Contains(t, logs.String(), "category=1")
77+ require.Contains(t, logs.String(), "type=11")
72 }78 }
73 79
74 func stripNoticeTimes(notices []domain.Notification) []domain.Notification {80 func stripNoticeTimes(notices []domain.Notification) []domain.Notification {
M internal/usecase/daemon/picker_test.go +13 -4
@@ -1011,8 +1011,12 @@ func TestPickerSessionSwitchPublishesBeforeInFlightPaintSendCompletes(t *testing
10111011 d := newTestDaemon(t, nil, stubClock{})
10121012 enteredSend := make(chan struct{})
10131013 releaseSend := make(chan struct{})
1014- tr := newMockServerConnection(t)
1015- tr.EXPECT().Send(mock.Anything).RunAndReturn(func(wire.Frame) error {
1014+ tr := portsmocks.NewMockServerConnection(t)
1015+ tr.EXPECT().Capabilities().Return(protocol.ConnectionCapabilities{
1016+ OutputDataLimit: protocol.MaxOutputDataLen,
1017+ PreferredOutputWindow: protocol.MaxOutputWindow,
1018+ }).Maybe()
1019+ tr.EXPECT().SendOutput(mock.Anything).RunAndReturn(func(protocol.Output) error {
10161020 close(enteredSend)
10171021 <-releaseSend
10181022 return nil
@@ -1408,8 +1412,13 @@ func TestResumeStoppedAndSwitchInheritsTerminalEnv(t *testing.T) {
14081412 })).Return(floating, nil).Once()
14091413 d := newTestDaemon(t, f, stubClock{})
14101414 d.inactive["old"] = inactiveSession{name: "old", cwd: t.TempDir(), createdAt: 1, state: protocol.SessionDown}
1411- tr := newMockServerConnection(t)
1412- tr.EXPECT().Send(mock.Anything).Return(nil).Maybe()
1415+ tr := portsmocks.NewMockServerConnection(t)
1416+ tr.EXPECT().Capabilities().Return(protocol.ConnectionCapabilities{
1417+ OutputDataLimit: protocol.MaxOutputDataLen,
1418+ PreferredOutputWindow: protocol.MaxOutputWindow,
1419+ }).Maybe()
1420+ tr.EXPECT().SendOutput(mock.Anything).Return(nil).Maybe()
1421+ tr.EXPECT().SendServer(mock.Anything).Return(nil).Maybe()
14131422 tr.EXPECT().Close().Return(nil).Maybe()
14141423 sess, ac, err := d.route(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Name: "work", Size: sz, TrueColor: true}, tr)
14151424 require.NoError(t, err)
@@ -1011,8 +1011,12 @@ func TestPickerSessionSwitchPublishesBeforeInFlightPaintSendCompletes(t *testing
1011 d := newTestDaemon(t, nil, stubClock{})1011 d := newTestDaemon(t, nil, stubClock{})
1012 enteredSend := make(chan struct{})1012 enteredSend := make(chan struct{})
1013 releaseSend := make(chan struct{})1013 releaseSend := make(chan struct{})
1014- tr := newMockServerConnection(t)1014+ tr := portsmocks.NewMockServerConnection(t)
1015- tr.EXPECT().Send(mock.Anything).RunAndReturn(func(wire.Frame) error {1015+ tr.EXPECT().Capabilities().Return(protocol.ConnectionCapabilities{
1016+ OutputDataLimit: protocol.MaxOutputDataLen,
1017+ PreferredOutputWindow: protocol.MaxOutputWindow,
1018+ }).Maybe()
1019+ tr.EXPECT().SendOutput(mock.Anything).RunAndReturn(func(protocol.Output) error {
1016 close(enteredSend)1020 close(enteredSend)
1017 <-releaseSend1021 <-releaseSend
1018 return nil1022 return nil
@@ -1408,8 +1412,13 @@ func TestResumeStoppedAndSwitchInheritsTerminalEnv(t *testing.T) {
1408 })).Return(floating, nil).Once()1412 })).Return(floating, nil).Once()
1409 d := newTestDaemon(t, f, stubClock{})1413 d := newTestDaemon(t, f, stubClock{})
1410 d.inactive["old"] = inactiveSession{name: "old", cwd: t.TempDir(), createdAt: 1, state: protocol.SessionDown}1414 d.inactive["old"] = inactiveSession{name: "old", cwd: t.TempDir(), createdAt: 1, state: protocol.SessionDown}
1411- tr := newMockServerConnection(t)1415+ tr := portsmocks.NewMockServerConnection(t)
1412- tr.EXPECT().Send(mock.Anything).Return(nil).Maybe()1416+ tr.EXPECT().Capabilities().Return(protocol.ConnectionCapabilities{
1417+ OutputDataLimit: protocol.MaxOutputDataLen,
1418+ PreferredOutputWindow: protocol.MaxOutputWindow,
1419+ }).Maybe()
1420+ tr.EXPECT().SendOutput(mock.Anything).Return(nil).Maybe()
1421+ tr.EXPECT().SendServer(mock.Anything).Return(nil).Maybe()
1413 tr.EXPECT().Close().Return(nil).Maybe()1422 tr.EXPECT().Close().Return(nil).Maybe()
1414 sess, ac, err := d.route(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Name: "work", Size: sz, TrueColor: true}, tr)1423 sess, ac, err := d.route(protocol.Hello{Version: protocol.Version, Intent: protocol.IntentNew, Name: "work", Size: sz, TrueColor: true}, tr)
1415 require.NoError(t, err)1424 require.NoError(t, err)
M internal/usecase/daemon/remote_picker_test.go +3 -2
@@ -16,6 +16,7 @@ import (
1616
1717 "github.com/bnema/vev/internal/domain"
1818 "github.com/bnema/vev/internal/ports"
19+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
1920 "github.com/bnema/vev/internal/protocol"
2021 "github.com/bnema/vev/internal/protocol/wire"
2122 "github.com/bnema/vev/internal/usecase/picker"
@@ -889,8 +890,8 @@ func TestRemotePickerHandoffSendFailureKeepsPickerOpen(t *testing.T) {
889890 d.remoteCatalog.status["arch"] = remoteHostFresh
890891 d.remoteCatalog.mu.Unlock()
891892 cause := errors.New("remote attach send failed")
892- tr := newMockServerConnection(t)
893- tr.EXPECT().Send(mock.Anything).Return(cause)
893+ tr := portsmocks.NewMockServerConnection(t)
894+ tr.EXPECT().SendServer(mock.Anything).Return(cause)
894895 sess, ac, _ := addRemoteRefreshPickerOwner(t, d, "local", tr)
895896 ac.overlays.pickerMu.Lock()
896897 ac.overlays.picker = picker.New(nil, picker.SelectionConfig{})
@@ -16,6 +16,7 @@ import (
16 16
17 "github.com/bnema/vev/internal/domain"17 "github.com/bnema/vev/internal/domain"
18 "github.com/bnema/vev/internal/ports"18 "github.com/bnema/vev/internal/ports"
19+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
19 "github.com/bnema/vev/internal/protocol"20 "github.com/bnema/vev/internal/protocol"
20 "github.com/bnema/vev/internal/protocol/wire"21 "github.com/bnema/vev/internal/protocol/wire"
21 "github.com/bnema/vev/internal/usecase/picker"22 "github.com/bnema/vev/internal/usecase/picker"
@@ -889,8 +890,8 @@ func TestRemotePickerHandoffSendFailureKeepsPickerOpen(t *testing.T) {
889 d.remoteCatalog.status["arch"] = remoteHostFresh890 d.remoteCatalog.status["arch"] = remoteHostFresh
890 d.remoteCatalog.mu.Unlock()891 d.remoteCatalog.mu.Unlock()
891 cause := errors.New("remote attach send failed")892 cause := errors.New("remote attach send failed")
892- tr := newMockServerConnection(t)893+ tr := portsmocks.NewMockServerConnection(t)
893- tr.EXPECT().Send(mock.Anything).Return(cause)894+ tr.EXPECT().SendServer(mock.Anything).Return(cause)
894 sess, ac, _ := addRemoteRefreshPickerOwner(t, d, "local", tr)895 sess, ac, _ := addRemoteRefreshPickerOwner(t, d, "local", tr)
895 ac.overlays.pickerMu.Lock()896 ac.overlays.pickerMu.Lock()
896 ac.overlays.picker = picker.New(nil, picker.SelectionConfig{})897 ac.overlays.picker = picker.New(nil, picker.SelectionConfig{})
M internal/usecase/daemon/typed_connection_test.go +2 -2
@@ -235,7 +235,7 @@ func testServerCapabilities(raw ports.Transport) protocol.ConnectionCapabilities
235235 _, async := raw.(ports.AsyncTransport)
236236 _, synchronous := raw.(ports.OwnedSynchronousTransport)
237237 _, link := raw.(ports.LinkStateReporter)
238- window := uint8(8)
238+ window := uint8(protocol.MaxOutputWindow)
239239 if datagram {
240240 window = 1
241241 }
@@ -314,7 +314,7 @@ func testServerFrame(message protocol.ServerMessage) (wire.Frame, error) {
314314 func (d *Daemon) handleCommandFrame(connection ports.ServerConnection, frame wire.Frame) error {
315315 request, err := wire.UnmarshalCommandRequest(frame.Payload)
316316 if err != nil {
317- code := uint16(protocol.ErrInternal)
317+ code := protocol.ErrInternal
318318 if version, ok := wire.PeekCommandVersion(frame.Payload); ok && version != protocol.Version {
319319 code = protocol.ErrVersionMismatch
320320 }
@@ -235,7 +235,7 @@ func testServerCapabilities(raw ports.Transport) protocol.ConnectionCapabilities
235 _, async := raw.(ports.AsyncTransport)235 _, async := raw.(ports.AsyncTransport)
236 _, synchronous := raw.(ports.OwnedSynchronousTransport)236 _, synchronous := raw.(ports.OwnedSynchronousTransport)
237 _, link := raw.(ports.LinkStateReporter)237 _, link := raw.(ports.LinkStateReporter)
238- window := uint8(8)238+ window := uint8(protocol.MaxOutputWindow)
239 if datagram {239 if datagram {
240 window = 1240 window = 1
241 }241 }
@@ -314,7 +314,7 @@ func testServerFrame(message protocol.ServerMessage) (wire.Frame, error) {
314 func (d *Daemon) handleCommandFrame(connection ports.ServerConnection, frame wire.Frame) error {314 func (d *Daemon) handleCommandFrame(connection ports.ServerConnection, frame wire.Frame) error {
315 request, err := wire.UnmarshalCommandRequest(frame.Payload)315 request, err := wire.UnmarshalCommandRequest(frame.Payload)
316 if err != nil {316 if err != nil {
317- code := uint16(protocol.ErrInternal)317+ code := protocol.ErrInternal
318 if version, ok := wire.PeekCommandVersion(frame.Payload); ok && version != protocol.Version {318 if version, ok := wire.PeekCommandVersion(frame.Payload); ok && version != protocol.Version {
319 code = protocol.ErrVersionMismatch319 code = protocol.ErrVersionMismatch
320 }320 }

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close