1package daemonmux
2
3import (
4 "context"
5 "testing"
6 "time"
7
8 "github.com/stretchr/testify/require"
9
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"
12 "github.com/bnema/vev/internal/protocol"
13)
14
15// attachmentRequest builds one valid local attachment Open under policy for the
16// supplied admission variant, name, target, and environment, so a listener test
17// exercises the exact metadata a real attachment carries.
18func attachmentRequest(stream int, policy ports.BrokerPolicy, admission ports.BrokerStreamAdmission, name string, target protocol.ExactSessionTarget, env []string) ports.BrokerOpenStreamRequest {
19 return ports.BrokerOpenStreamRequest{
20 Epoch: 1,
21 Purpose: ports.BrokerStreamAttachment,
22 Admission: admission,
23 Name: name,
24 Local: true,
25 Connection: ports.BrokerConnectionID{1},
26 Stream: ports.BrokerStreamID(stream),
27 Target: target,
28 Env: append([]string(nil), env...),
29 Policy: policy,
30 StartMode: ports.BrokerDaemonStartIfNeeded,
31 }
32}
33
34// admittedConnection opens one request over the harness while a concurrent
35// Accept confirms it, returning the accepted typed daemon-side connection. A
36// listener confirms Opened only as Accept delivers the stream, so the two must
37// run together.
38func admittedConnection(t *testing.T, h *listenerHarness, request ports.BrokerOpenStreamRequest) *ListenerConnection {
39 t.Helper()
40 accepted := h.acceptAsync()
41 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
42 defer cancel()
43 _, err := h.connector.Open(ctx, request)
44 require.NoError(t, err)
45 return h.awaitAccept(accepted)
46}
47
48// TestListenerStampsProvisionedOrigin proves the accepted connection's admission
49// comes from the provisioned binding, not the peer's Open: a listener
50// provisioned as local stamps Origin=Local with the accepted policy, and every
51// declared field of the Open is carried verbatim.
52func TestListenerStampsProvisionedOrigin(t *testing.T) {
53 policy := logicalTestPolicy()
54 h := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
55
56 // A local control stream carries no attachment state.
57 connection := admittedConnection(t, h, controlRequest(1))
58 admission, ok := connection.SessionAdmission()
59 require.True(t, ok)
60 require.Equal(t, ports.SessionOriginLocal, admission.Origin, "the provisioned locality is stamped")
61 require.Equal(t, policy, admission.Policy, "the accepted policy is the provisioned one")
62 require.Equal(t, ports.BrokerStreamControl, admission.Purpose)
63 require.Zero(t, admission.Admission)
64 require.Empty(t, admission.Env)
65 require.NoError(t, admission.Validate())
66
67 // An attachment stream carries its closed variant, validated name, exact
68 // target, and bounded environment.
69 target := protocol.ExactSessionTarget{LifecycleID: domain.SessionLifecycleID{7}, SessionName: "work"}
70 attach := admittedConnection(t, h, attachmentRequest(2, policy, ports.BrokerAdmissionExact, "", target, []string{"TERM=xterm-256color"}))
71 attachAdmission, ok := attach.SessionAdmission()
72 require.True(t, ok)
73 require.Equal(t, ports.SessionOriginLocal, attachAdmission.Origin)
74 require.Equal(t, ports.BrokerStreamAttachment, attachAdmission.Purpose)
75 require.Equal(t, ports.BrokerAdmissionExact, attachAdmission.Admission)
76 require.Equal(t, target, attachAdmission.Target)
77 require.Equal(t, []string{"TERM=xterm-256color"}, attachAdmission.Env)
78 require.NoError(t, attachAdmission.Validate())
79}
80
81// TestListenerRefusesLiarLocalOrigin proves the accepted origin is enforced, not
82// echoed: a listener provisioned for a remote carriage refuses a local Open (and
83// one provisioned local refuses a remote Open) exactly on its own stream, so a
84// peer can never stamp a provenance the accepting side did not provision.
85func TestListenerRefusesLiarLocalOrigin(t *testing.T) {
86 policy := logicalTestPolicy()
87
88 t.Run("remote carriage refuses a local Open", func(t *testing.T) {
89 brokerPump, daemonPump, start := newPairedPumpsUnstarted(t, DefaultMuxCeilings())
90 clock := newListenerClock(time.Now())
91 listener, err := newListener(daemonPump, listenerIdleBudget, MaxAcceptQueue, clock, remoteAcceptance(policy))
92 require.NoError(t, err)
93 t.Cleanup(func() { _ = listener.Close() })
94 start()
95 connector := mustConnector(t, brokerPump)
96
97 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
98 defer cancel()
99 _, err = connector.Open(ctx, controlRequest(1))
100 require.Error(t, err, "a local Open is refused by a remote carriage")
101
102 refused := requireStreamState(t, daemonPump, 1, StreamTerminal)
103 require.Equal(t, domain.RemoteFailureTransport, refused.FailureKind)
104 var detail ports.BrokerError
105 require.ErrorAs(t, refused.Err, &detail)
106 require.Equal(t, invalidAdmissionText, detail.Text)
107
108 // Nothing was queued for the daemon: a refused Open is never delivered.
109 listener.mu.Lock()
110 queued := len(listener.queue)
111 listener.mu.Unlock()
112 require.Zero(t, queued, "a refused Open is never queued")
113 })
114
115 t.Run("local carriage refuses a remote Open", func(t *testing.T) {
116 h := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
117 remote := controlRequest(1)
118 remote.Local = false
119 remote.Endpoint = "dev@host:22"
120 remote.Registration = testRegistration()
121
122 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
123 defer cancel()
124 _, err := h.connector.Open(ctx, remote)
125 require.Error(t, err, "a remote Open is refused by a local carriage")
126
127 refused := requireStreamState(t, h.daemonPump, 1, StreamTerminal)
128 var detail ports.BrokerError
129 require.ErrorAs(t, refused.Err, &detail)
130 require.Equal(t, invalidAdmissionText, detail.Text)
131 })
132}
133
134// TestListenerAdmissionIsolationAndDeepCopy proves every accepted connection
135// owns an independent admission: mutating one connection's returned copy never
136// changes a sibling's, and concurrent streams each receive exactly the metadata
137// their own Open declared.
138func TestListenerAdmissionIsolationAndDeepCopy(t *testing.T) {
139 policy := logicalTestPolicy()
140 h := newListenerHarness(t, listenerIdleBudget, MaxAcceptQueue)
141
142 first := admittedConnection(t, h, attachmentRequest(1, policy, ports.BrokerAdmissionCreateNamed, "alpha", protocol.ExactSessionTarget{}, []string{"TERM=one"}))
143
144 // The provider hands out a defensive copy: mutating it must not alter what
145 // the connection (or a later read) reports.
146 mutated, ok := first.SessionAdmission()
147 require.True(t, ok)
148 mutated.Env[0] = "TERM=mutated"
149 mutated.Name = "mutated"
150 mutated.Origin = ports.SessionOriginRemote
151 reread, ok := first.SessionAdmission()
152 require.True(t, ok)
153 require.Equal(t, "alpha", reread.Name)
154 require.Equal(t, ports.SessionOriginLocal, reread.Origin)
155 require.Equal(t, []string{"TERM=one"}, reread.Env)
156
157 // Two reads never share the same backing slice.
158 left, _ := first.SessionAdmission()
159 right, _ := first.SessionAdmission()
160 left.Env[0] = "TERM=left"
161 require.Equal(t, "TERM=one", right.Env[0])
162
163 // A sibling stream carries only its own declaration; the first's mutation
164 // never leaks into it.
165 second := admittedConnection(t, h, attachmentRequest(2, policy, ports.BrokerAdmissionCreateEphemeral, "", protocol.ExactSessionTarget{}, nil))
166 secondAdmission, ok := second.SessionAdmission()
167 require.True(t, ok)
168 require.Equal(t, ports.BrokerAdmissionCreateEphemeral, secondAdmission.Admission)
169 require.Empty(t, secondAdmission.Name)
170 require.Empty(t, secondAdmission.Env)
171}
172
173// TestListenerConcurrentAdmissionsKeepDistinctMetadata proves the listener's
174// per-stream admission stamp is per stream under concurrency: every admitted
175// attachment receives exactly the name, target, and environment its own Open
176// declared, and no stream sees a sibling's.
177func TestListenerConcurrentAdmissionsKeepDistinctMetadata(t *testing.T) {
178 policy := logicalTestPolicy()
179 h := newListenerHarness(t, listenerIdleBudget, 64)
180
181 const streams = 24
182 type expectation struct {
183 admission ports.SessionAdmission
184 }
185 want := make(map[ports.BrokerStreamID]expectation, streams)
186 opened := make(chan *ListenerConnection, streams)
187 openErrs := make(chan error, streams)
188 for i := 0; i < streams; i++ {
189 stream := ports.BrokerStreamID(i + 1)
190 name := ""
191 if i%2 == 0 {
192 name = []string{"alpha", "beta", "gamma"}[i%3]
193 }
194 env := []string{[]string{"TERM=first", "TERM=second", "TERM=third"}[i%3]}
195 admission := ports.BrokerAdmissionCreateNamed
196 if name == "" {
197 admission = ports.BrokerAdmissionCreateEphemeral
198 }
199 want[stream] = expectation{admission: ports.SessionAdmission{
200 Origin: ports.SessionOriginLocal, Policy: policy, Purpose: ports.BrokerStreamAttachment,
201 Admission: admission, Name: name, Env: env,
202 }}
203 request := attachmentRequest(i+1, policy, admission, name, protocol.ExactSessionTarget{}, env)
204 go func(request ports.BrokerOpenStreamRequest) {
205 // The Accept that confirms this stream must be in flight before the
206 // Open, because a listener sends Opened only as Accept delivers.
207 accepted := h.acceptAsync()
208 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
209 defer cancel()
210 _, err := h.connector.Open(ctx, request)
211 if err != nil {
212 openErrs <- err
213 return
214 }
215 opened <- h.awaitAccept(accepted)
216 }(request)
217 }
218
219 for i := 0; i < streams; i++ {
220 select {
221 case err := <-openErrs:
222 require.NoError(t, err)
223 case connection := <-opened:
224 admission, ok := connection.SessionAdmission()
225 require.True(t, ok)
226 expected, tracked := want[connection.Ref().Client]
227 require.True(t, tracked)
228 require.Equal(t, expected.admission, admission, "stream %d", connection.Ref().Client)
229 case <-time.After(p3cTestDeadline):
230 t.Fatal("concurrent admissions did not all complete")
231 }
232 }
233}