cf66883 refactor(broker): validate open requests at decode and admission only · vev

M internal/adapters/brokeripc/client.go +0 -3


@@ -745,9 +745,6 @@ func (c *client) scopeRequest(request ports.BrokerOpenStreamRequest) (ports.Brok
745	745		if scoped.Stream == 0 {
746	746		return ports.BrokerOpenStreamRequest{}, errors.Join(ports.BrokerAdmissionInvalid, errors.New("brokeripc: open stream request has no stream identity"))
747	747		}
748		-	if err := scoped.Validate(); err != nil {
749		-	return ports.BrokerOpenStreamRequest{}, errors.Join(ports.BrokerAdmissionInvalid, err)
750		-	}
751	748		return scoped, nil
752	749		}
753	750


@@ -745,9 +745,6 @@ func (c *client) scopeRequest(request ports.BrokerOpenStreamRequest) (ports.Brok
745		if scoped.Stream == 0 {	745	if scoped.Stream == 0 {
746		return ports.BrokerOpenStreamRequest{}, errors.Join(ports.BrokerAdmissionInvalid, errors.New("brokeripc: open stream request has no stream identity"))	746	return ports.BrokerOpenStreamRequest{}, errors.Join(ports.BrokerAdmissionInvalid, errors.New("brokeripc: open stream request has no stream identity"))
747		}	747	}
748	-	if err := scoped.Validate(); err != nil {
749	-	return ports.BrokerOpenStreamRequest{}, errors.Join(ports.BrokerAdmissionInvalid, err)
750	-	}
751		return scoped, nil	748	return scoped, nil
752		}	749	}
753			750

M internal/adapters/brokeripc/session_test.go +1 -9


@@ -176,17 +176,9 @@ func TestObservationForcesExistingOnly(t *testing.T) {
176	176		require.NoError(t, err)
177	177		require.NoError(t, stream.Close())
178	178
179		-	// A spawn-capable observation is refused locally: it never reaches the
180		-	// admitted core at all.
181		-	startable := openRequest(nextStreamID(t, client))
182		-	startable.Purpose = ports.BrokerStreamObservation
183		-	startable.StartMode = ports.BrokerDaemonStartIfNeeded
184		-	_, err = client.OpenStream(ctx, startable)
185		-	require.ErrorIs(t, err, ports.BrokerAdmissionInvalid, "observation never authorizes a spawn")
186		-
187	179		core.mu.Lock()
188	180		defer core.mu.Unlock()
189		-	require.Len(t, core.opens, 1, "only the existing-only observation reached the core")
	181	+	require.Len(t, core.opens, 1, "the existing-only observation reached the core")
190	182		require.Equal(t, ports.BrokerStreamObservation, core.opens[0].Purpose)
191	183		require.Equal(t, ports.BrokerDaemonExistingOnly, core.opens[0].StartMode)
192	184		}

 	require.NoError(t, err)
 	require.NoError(t, stream.Close())
-	// A spawn-capable observation is refused locally: it never reaches the
-	// admitted core at all.
-	startable := openRequest(nextStreamID(t, client))
-	startable.Purpose = ports.BrokerStreamObservation
-	startable.StartMode = ports.BrokerDaemonStartIfNeeded
-	_, err = client.OpenStream(ctx, startable)
-	require.ErrorIs(t, err, ports.BrokerAdmissionInvalid, "observation never authorizes a spawn")
 	core.mu.Lock()
 	defer core.mu.Unlock()
-	require.Len(t, core.opens, 1, "only the existing-only observation reached the core")
+	require.Len(t, core.opens, 1, "the existing-only observation reached the core")
 	require.Equal(t, ports.BrokerStreamObservation, core.opens[0].Purpose)
 	require.Equal(t, ports.BrokerDaemonExistingOnly, core.opens[0].StartMode)
 }

M internal/adapters/brokerwire/codec.go +1 -20


@@ -525,26 +525,6 @@ func unsubscribeFromWire(message *wire.Unsubscribe) (Unsubscribe, error) {
525	525		}
526	526
527	527		func openStreamToWire(m OpenStream) (*wire.OpenStream, error) {
528		-	request := ports.BrokerOpenStreamRequest{
529		-	Epoch: m.Epoch, Purpose: m.Purpose, Admission: m.Admission, Name: m.Name,
530		-	Local: m.Local,
531		-	Connection: m.Connection, Stream: m.Stream,
532		-	Endpoint: m.Endpoint, Registration: m.Registration,
533		-	Target: m.Target, Env: m.Env, Policy: m.Policy,
534		-	StartMode: m.StartMode,
535		-	}
536		-	if err := request.Validate(); err != nil {
537		-	// Distinguish bound refusals from semantic refusals.
538		-	if len(m.Env) > ports.BrokerMaxEnvEntries {
539		-	return nil, ErrTooLarge
540		-	}
541		-	for _, entry := range m.Env {
542		-	if len(entry) > ports.BrokerMaxEnvEntryBytes {
543		-	return nil, ErrTooLarge
544		-	}
545		-	}
546		-	return nil, ErrInvalidMessage
547		-	}
548	528		if err := checkEnvEntries(m.Env); err != nil {
549	529		return nil, err
550	530		}

@@ -724,6 +704,7 @@ func openStreamFromWire(message *wire.OpenStream) (OpenStream, error) {
724	704		Target: candidate.Target, Env: candidate.Env, Policy: candidate.Policy,
725	705		StartMode: candidate.StartMode,
726	706		}
	707	+	// Decode owns validation of untrusted open requests before they enter the broker.
727	708		if err := request.Validate(); err != nil {
728	709		return OpenStream{}, ErrInvalidMessage
729	710		}


@@ -525,26 +525,6 @@ func unsubscribeFromWire(message *wire.Unsubscribe) (Unsubscribe, error) {
525		}	525		}
526			526
527		func openStreamToWire(m OpenStream) (*wire.OpenStream, error) {	527		func openStreamToWire(m OpenStream) (*wire.OpenStream, error) {
528	-	request := ports.BrokerOpenStreamRequest{
529	-	Epoch: m.Epoch, Purpose: m.Purpose, Admission: m.Admission, Name: m.Name,
530	-	Local: m.Local,
531	-	Connection: m.Connection, Stream: m.Stream,
532	-	Endpoint: m.Endpoint, Registration: m.Registration,
533	-	Target: m.Target, Env: m.Env, Policy: m.Policy,
534	-	StartMode: m.StartMode,
535	-	}
536	-	if err := request.Validate(); err != nil {
537	-	// Distinguish bound refusals from semantic refusals.
538	-	if len(m.Env) > ports.BrokerMaxEnvEntries {
539	-	return nil, ErrTooLarge
540	-	}
541	-	for _, entry := range m.Env {
542	-	if len(entry) > ports.BrokerMaxEnvEntryBytes {
543	-	return nil, ErrTooLarge
544	-	}
545	-	}
546	-	return nil, ErrInvalidMessage
547	-	}
548		if err := checkEnvEntries(m.Env); err != nil {	528		if err := checkEnvEntries(m.Env); err != nil {
549		return nil, err	529		return nil, err
550		}	530		}

@@ -724,6 +704,7 @@ func openStreamFromWire(message *wire.OpenStream) (OpenStream, error) {
724		Target: candidate.Target, Env: candidate.Env, Policy: candidate.Policy,	704		Target: candidate.Target, Env: candidate.Env, Policy: candidate.Policy,
725		StartMode: candidate.StartMode,	705		StartMode: candidate.StartMode,
726		}	706		}
			707	+	// Decode owns validation of untrusted open requests before they enter the broker.
727		if err := request.Validate(); err != nil {	708		if err := request.Validate(); err != nil {
728		return OpenStream{}, ErrInvalidMessage	709		return OpenStream{}, ErrInvalidMessage
729		}	710		}

M internal/adapters/brokerwire/codec_test.go +38 -5


@@ -379,7 +379,7 @@ func TestBrokerClientVariants(t *testing.T) {
379	379		// TestBrokerOpenStreamAdmissionVariants pins the attachment-admission
380	380		// contract: exact, create-named, and create-ephemeral round-trip, control
381	381		// carries none, and wrong admission/purpose/name/target combinations are
382		-	// refused exactly like ports.BrokerOpenStreamRequest.Validate.
	382	+	// refused on decode exactly like ports.BrokerOpenStreamRequest.Validate.
383	383		func TestBrokerOpenStreamAdmissionVariants(t *testing.T) {
384	384		connection := testConnectionID(0x11)
385	385		base := func() OpenStream {

@@ -475,7 +475,24 @@ func TestBrokerOpenStreamAdmissionVariants(t *testing.T) {
475	475		}
476	476		for name, message := range cases {
477	477		t.Run(name, func(t *testing.T) {
478		-	_, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
	478	+	// Start mode is a closed wire enum and is refused even by the encoder.
	479	+	if message.StartMode == 0 {
	480	+	_, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
	481	+	require.Error(t, err)
	482	+	return
	483	+	}
	484	+	raw, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
	485	+	require.NoError(t, err)
	486	+	_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
	487	+	if name == "named with target" {
	488	+	// The encoder omits a target on creation; test the untrusted wire directly.
	489	+	envelope := &wire.BrokerClientEnvelope{}
	490	+	require.NoError(t, proto.Unmarshal(raw, envelope))
	491	+	envelope.GetOpenStream().Target = exactTargetToWire(testTarget())
	492	+	raw, err = proto.Marshal(envelope)
	493	+	require.NoError(t, err)
	494	+	_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
	495	+	}
479	496		require.ErrorIs(t, err, ErrInvalidMessage)
480	497		})
481	498		}

@@ -861,7 +878,14 @@ func TestBrokerBounds(t *testing.T) {
861	878		require.ErrorIs(t, err, ErrTooLarge)
862	879		})
863	880		t.Run("control env carries no attachment state", func(t *testing.T) {
864		-	_, err := EncodeClient(OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), Target: testTarget()}, testEnvelopeCeiling, testChunkCeiling)
	881	+	message := OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
	882	+	raw := mustEncodeClient(t, message)
	883	+	envelope := &wire.BrokerClientEnvelope{}
	884	+	require.NoError(t, proto.Unmarshal(raw, envelope))
	885	+	envelope.GetOpenStream().Target = exactTargetToWire(testTarget())
	886	+	raw, err := proto.Marshal(envelope)
	887	+	require.NoError(t, err)
	888	+	_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
865	889		require.ErrorIs(t, err, ErrInvalidMessage)
866	890		})
867	891		}

@@ -1077,7 +1101,9 @@ func TestBrokerValidationNegatives(t *testing.T) {
1077	1101		require.ErrorIs(t, err, ErrInvalidMessage)
1078	1102		})
1079	1103		t.Run("invalid policy refused", func(t *testing.T) {
1080		-	_, err := EncodeClient(OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Policy: ports.BrokerPolicy{}, StartMode: ports.BrokerDaemonStartIfNeeded}, testEnvelopeCeiling, testChunkCeiling)
	1104	+	raw, err := EncodeClient(OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Policy: ports.BrokerPolicy{}, StartMode: ports.BrokerDaemonStartIfNeeded}, testEnvelopeCeiling, testChunkCeiling)
	1105	+	require.NoError(t, err)
	1106	+	_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1081	1107		require.ErrorIs(t, err, ErrInvalidMessage)
1082	1108		})
1083	1109		t.Run("snapshot revision zero refused", func(t *testing.T) {

@@ -1089,7 +1115,14 @@ func TestBrokerValidationNegatives(t *testing.T) {
1089	1115		require.ErrorIs(t, err, ErrInvalidMessage)
1090	1116		})
1091	1117		t.Run("local open with registration refused", func(t *testing.T) {
1092		-	_, err := EncodeClient(OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Local: true, Registration: testRegistration(), Target: testTarget(), Policy: testPolicy()}, testEnvelopeCeiling, testChunkCeiling)
	1118	+	message := OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Local: true, Target: testTarget(), Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
	1119	+	raw := mustEncodeClient(t, message)
	1120	+	envelope := &wire.BrokerClientEnvelope{}
	1121	+	require.NoError(t, proto.Unmarshal(raw, envelope))
	1122	+	envelope.GetOpenStream().Registration = registrationToWire(testRegistration())
	1123	+	raw, err := proto.Marshal(envelope)
	1124	+	require.NoError(t, err)
	1125	+	_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
1093	1126		require.ErrorIs(t, err, ErrInvalidMessage)
1094	1127		})
1095	1128		t.Run("bad shutdown reason refused", func(t *testing.T) {

 // TestBrokerOpenStreamAdmissionVariants pins the attachment-admission
 // contract: exact, create-named, and create-ephemeral round-trip, control
 // carries none, and wrong admission/purpose/name/target combinations are
-// refused exactly like ports.BrokerOpenStreamRequest.Validate.
+// refused on decode exactly like ports.BrokerOpenStreamRequest.Validate.
 func TestBrokerOpenStreamAdmissionVariants(t *testing.T) {
 	connection := testConnectionID(0x11)
 	base := func() OpenStream {
 		}
 		for name, message := range cases {
 			t.Run(name, func(t *testing.T) {
-				_, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
+				// Start mode is a closed wire enum and is refused even by the encoder.
+				if message.StartMode == 0 {
+					_, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
+					require.Error(t, err)
+					return
+				}
+				raw, err := EncodeClient(message, testEnvelopeCeiling, testChunkCeiling)
+				require.NoError(t, err)
+				_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
+				if name == "named with target" {
+					// The encoder omits a target on creation; test the untrusted wire directly.
+					envelope := &wire.BrokerClientEnvelope{}
+					require.NoError(t, proto.Unmarshal(raw, envelope))
+					envelope.GetOpenStream().Target = exactTargetToWire(testTarget())
+					raw, err = proto.Marshal(envelope)
+					require.NoError(t, err)
+					_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
+				}
 				require.ErrorIs(t, err, ErrInvalidMessage)
 			})
 		}
 		require.ErrorIs(t, err, ErrTooLarge)
 	})
 	t.Run("control env carries no attachment state", func(t *testing.T) {
-		_, err := EncodeClient(OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), Target: testTarget()}, testEnvelopeCeiling, testChunkCeiling)
+		message := OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
+		raw := mustEncodeClient(t, message)
+		envelope := &wire.BrokerClientEnvelope{}
+		require.NoError(t, proto.Unmarshal(raw, envelope))
+		envelope.GetOpenStream().Target = exactTargetToWire(testTarget())
+		raw, err := proto.Marshal(envelope)
+		require.NoError(t, err)
+		_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
 		require.ErrorIs(t, err, ErrInvalidMessage)
 	})
 }
 		require.ErrorIs(t, err, ErrInvalidMessage)
 	})
 	t.Run("invalid policy refused", func(t *testing.T) {
-		_, err := EncodeClient(OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Policy: ports.BrokerPolicy{}, StartMode: ports.BrokerDaemonStartIfNeeded}, testEnvelopeCeiling, testChunkCeiling)
+		raw, err := EncodeClient(OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Endpoint: "dev@host:22", Registration: testRegistration(), Target: testTarget(), Policy: ports.BrokerPolicy{}, StartMode: ports.BrokerDaemonStartIfNeeded}, testEnvelopeCeiling, testChunkCeiling)
+		require.NoError(t, err)
+		_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
 		require.ErrorIs(t, err, ErrInvalidMessage)
 	})
 	t.Run("snapshot revision zero refused", func(t *testing.T) {
 		require.ErrorIs(t, err, ErrInvalidMessage)
 	})
 	t.Run("local open with registration refused", func(t *testing.T) {
-		_, err := EncodeClient(OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Local: true, Registration: testRegistration(), Target: testTarget(), Policy: testPolicy()}, testEnvelopeCeiling, testChunkCeiling)
+		message := OpenStream{Epoch: 7, Connection: connection, Stream: 3, Purpose: ports.BrokerStreamAttachment, Admission: ports.BrokerAdmissionExact, Local: true, Target: testTarget(), Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
+		raw := mustEncodeClient(t, message)
+		envelope := &wire.BrokerClientEnvelope{}
+		require.NoError(t, proto.Unmarshal(raw, envelope))
+		envelope.GetOpenStream().Registration = registrationToWire(testRegistration())
+		raw, err := proto.Marshal(envelope)
+		require.NoError(t, err)
+		_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
 		require.ErrorIs(t, err, ErrInvalidMessage)
 	})
 	t.Run("bad shutdown reason refused", func(t *testing.T) {

M internal/adapters/daemonmux/codec.go +1 -18


@@ -269,24 +269,6 @@ func openToWire(m Open) (*wire.MuxOpen, error) {
269	269		if err := m.Ref.Validate(); err != nil {
270	270		return nil, ErrInvalidMessage
271	271		}
272		-	request := ports.BrokerOpenStreamRequest{
273		-	Epoch: m.Ref.Epoch, Purpose: m.Purpose, Admission: m.Admission, Name: m.Name, Local: m.Local,
274		-	Connection: m.Ref.Connection, Stream: m.Ref.Client,
275		-	Endpoint: m.Endpoint, Registration: m.Registration,
276		-	Target: m.Target, Env: m.Env, Policy: m.Policy, StartMode: m.StartMode,
277		-	}
278		-	if err := request.Validate(); err != nil {
279		-	// Distinguish bound refusals from semantic refusals.
280		-	if len(m.Env) > ports.BrokerMaxEnvEntries {
281		-	return nil, ErrTooLarge
282		-	}
283		-	for _, entry := range m.Env {
284		-	if len(entry) > ports.BrokerMaxEnvEntryBytes {
285		-	return nil, ErrTooLarge
286		-	}
287		-	}
288		-	return nil, ErrInvalidMessage
289		-	}
290	272		if err := checkEnvEntries(m.Env); err != nil {
291	273		return nil, err
292	274		}

@@ -395,6 +377,7 @@ func openFromWire(message *wire.MuxOpen) (Open, error) {
395	377		Target: candidate.Target, Env: candidate.Env, Policy: candidate.Policy,
396	378		StartMode: candidate.StartMode,
397	379		}
	380	+	// Decode owns validation of untrusted mux opens before daemon admission.
398	381		if err := request.Validate(); err != nil {
399	382		return Open{}, ErrInvalidMessage
400	383		}


@@ -269,24 +269,6 @@ func openToWire(m Open) (*wire.MuxOpen, error) {
269		if err := m.Ref.Validate(); err != nil {	269		if err := m.Ref.Validate(); err != nil {
270		return nil, ErrInvalidMessage	270		return nil, ErrInvalidMessage
271		}	271		}
272	-	request := ports.BrokerOpenStreamRequest{
273	-	Epoch: m.Ref.Epoch, Purpose: m.Purpose, Admission: m.Admission, Name: m.Name, Local: m.Local,
274	-	Connection: m.Ref.Connection, Stream: m.Ref.Client,
275	-	Endpoint: m.Endpoint, Registration: m.Registration,
276	-	Target: m.Target, Env: m.Env, Policy: m.Policy, StartMode: m.StartMode,
277	-	}
278	-	if err := request.Validate(); err != nil {
279	-	// Distinguish bound refusals from semantic refusals.
280	-	if len(m.Env) > ports.BrokerMaxEnvEntries {
281	-	return nil, ErrTooLarge
282	-	}
283	-	for _, entry := range m.Env {
284	-	if len(entry) > ports.BrokerMaxEnvEntryBytes {
285	-	return nil, ErrTooLarge
286	-	}
287	-	}
288	-	return nil, ErrInvalidMessage
289	-	}
290		if err := checkEnvEntries(m.Env); err != nil {	272		if err := checkEnvEntries(m.Env); err != nil {
291		return nil, err	273		return nil, err
292		}	274		}

@@ -395,6 +377,7 @@ func openFromWire(message *wire.MuxOpen) (Open, error) {
395		Target: candidate.Target, Env: candidate.Env, Policy: candidate.Policy,	377		Target: candidate.Target, Env: candidate.Env, Policy: candidate.Policy,
396		StartMode: candidate.StartMode,	378		StartMode: candidate.StartMode,
397		}	379		}
			380	+	// Decode owns validation of untrusted mux opens before daemon admission.
398		if err := request.Validate(); err != nil {	381		if err := request.Validate(); err != nil {
399		return Open{}, ErrInvalidMessage	382		return Open{}, ErrInvalidMessage
400		}	383		}

M internal/adapters/daemonmux/codec_test.go +33 -9


@@ -395,23 +395,36 @@ func TestMuxValidationNegatives(t *testing.T) {
395	395
396	396		badEndpoint := testOpen()
397	397		badEndpoint.Endpoint = "not a host"
398		-	_, err = EncodeClient(badEndpoint, testEnvelopeCeiling, testChunkCeiling)
	398	+	raw, err := EncodeClient(badEndpoint, testEnvelopeCeiling, testChunkCeiling)
	399	+	require.NoError(t, err)
	400	+	_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
399	401		require.ErrorIs(t, err, ErrInvalidMessage)
400	402
401	403		local := testOpen()
402	404		local.Local = true
403		-	_, err = EncodeClient(local, testEnvelopeCeiling, testChunkCeiling)
	405	+	raw, err = EncodeClient(local, testEnvelopeCeiling, testChunkCeiling)
	406	+	require.NoError(t, err)
	407	+	_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
404	408		require.ErrorIs(t, err, ErrInvalidMessage)
405	409
406	410		controlWithTarget := testOpen()
407	411		controlWithTarget.Purpose = ports.BrokerStreamControl
408		-	_, err = EncodeClient(controlWithTarget, testEnvelopeCeiling, testChunkCeiling)
	412	+	raw, err = EncodeClient(controlWithTarget, testEnvelopeCeiling, testChunkCeiling)
	413	+	require.NoError(t, err)
	414	+	envelope := &wire.MuxClientEnvelope{}
	415	+	require.NoError(t, proto.Unmarshal(raw, envelope))
	416	+	envelope.GetOpen().Target = exactTargetToWire(testOpen().Target)
	417	+	raw, err = proto.Marshal(envelope)
	418	+	require.NoError(t, err)
	419	+	_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
409	420		require.ErrorIs(t, err, ErrInvalidMessage)
410	421
411	422		controlWithEnv := testOpen()
412	423		controlWithEnv.Purpose = ports.BrokerStreamControl
413	424		controlWithEnv.Target = protocol.ExactSessionTarget{}
414		-	_, err = EncodeClient(controlWithEnv, testEnvelopeCeiling, testChunkCeiling)
	425	+	raw, err = EncodeClient(controlWithEnv, testEnvelopeCeiling, testChunkCeiling)
	426	+	require.NoError(t, err)
	427	+	_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
415	428		require.ErrorIs(t, err, ErrInvalidMessage)
416	429
417	430		noEquals := testOpen()

@@ -421,12 +434,14 @@ func TestMuxValidationNegatives(t *testing.T) {
421	434
422	435		emptyPolicy := testOpen()
423	436		emptyPolicy.Policy = ports.BrokerPolicy{}
424		-	_, err = EncodeClient(emptyPolicy, testEnvelopeCeiling, testChunkCeiling)
	437	+	raw, err = EncodeClient(emptyPolicy, testEnvelopeCeiling, testChunkCeiling)
	438	+	require.NoError(t, err)
	439	+	_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
425	440		require.ErrorIs(t, err, ErrInvalidMessage)
426	441
427	442		// A local control open with no attachment state is legal.
428	443		valid := Open{Ref: testRef(5), Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
429		-	raw, err := EncodeClient(valid, testEnvelopeCeiling, testChunkCeiling)
	444	+	raw, err = EncodeClient(valid, testEnvelopeCeiling, testChunkCeiling)
430	445		require.NoError(t, err)
431	446		decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
432	447		require.NoError(t, err)

@@ -722,14 +737,23 @@ func TestMuxAdmissionVariants(t *testing.T) {
722	737		t.Run("exact with a creation name is refused", func(t *testing.T) {
723	738		open := testOpen()
724	739		open.Name = "work"
725		-	_, err := EncodeClient(open, testEnvelopeCeiling, testChunkCeiling)
	740	+	raw, err := EncodeClient(open, testEnvelopeCeiling, testChunkCeiling)
	741	+	require.NoError(t, err)
	742	+	_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
726	743		require.ErrorIs(t, err, ErrInvalidMessage)
727	744		})
728	745		t.Run("named creation with a target is refused", func(t *testing.T) {
729	746		named := testOpen()
730	747		named.Admission = ports.BrokerAdmissionCreateNamed
731	748		named.Name = "work"
732		-	_, err := EncodeClient(named, testEnvelopeCeiling, testChunkCeiling)
	749	+	raw, err := EncodeClient(named, testEnvelopeCeiling, testChunkCeiling)
	750	+	require.NoError(t, err)
	751	+	envelope := &wire.MuxClientEnvelope{}
	752	+	require.NoError(t, proto.Unmarshal(raw, envelope))
	753	+	envelope.GetOpen().Target = exactTargetToWire(testOpen().Target)
	754	+	raw, err = proto.Marshal(envelope)
	755	+	require.NoError(t, err)
	756	+	_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
733	757		require.ErrorIs(t, err, ErrInvalidMessage)
734	758		})
735	759		t.Run("unknown admission code is refused", func(t *testing.T) {

@@ -799,7 +823,7 @@ func TestMuxStartModeRoundTrips(t *testing.T) {
799	823
800	824		t.Run("encoding a zero start mode is refused", func(t *testing.T) {
801	825		_, err := EncodeClient(Open{Ref: testRef(6), Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy()}, testEnvelopeCeiling, testChunkCeiling)
802		-	require.ErrorIs(t, err, ErrInvalidMessage)
	826	+	require.Error(t, err) // Closed wire enum; decode refusal is tested above.
803	827		})
804	828		}
805	829

 		badEndpoint := testOpen()
 		badEndpoint.Endpoint = "not a host"
-		_, err = EncodeClient(badEndpoint, testEnvelopeCeiling, testChunkCeiling)
+		raw, err := EncodeClient(badEndpoint, testEnvelopeCeiling, testChunkCeiling)
+		require.NoError(t, err)
+		_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
 		require.ErrorIs(t, err, ErrInvalidMessage)
 		local := testOpen()
 		local.Local = true
-		_, err = EncodeClient(local, testEnvelopeCeiling, testChunkCeiling)
+		raw, err = EncodeClient(local, testEnvelopeCeiling, testChunkCeiling)
+		require.NoError(t, err)
+		_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
 		require.ErrorIs(t, err, ErrInvalidMessage)
 		controlWithTarget := testOpen()
 		controlWithTarget.Purpose = ports.BrokerStreamControl
-		_, err = EncodeClient(controlWithTarget, testEnvelopeCeiling, testChunkCeiling)
+		raw, err = EncodeClient(controlWithTarget, testEnvelopeCeiling, testChunkCeiling)
+		require.NoError(t, err)
+		envelope := &wire.MuxClientEnvelope{}
+		require.NoError(t, proto.Unmarshal(raw, envelope))
+		envelope.GetOpen().Target = exactTargetToWire(testOpen().Target)
+		raw, err = proto.Marshal(envelope)
+		require.NoError(t, err)
+		_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
 		require.ErrorIs(t, err, ErrInvalidMessage)
 		controlWithEnv := testOpen()
 		controlWithEnv.Purpose = ports.BrokerStreamControl
 		controlWithEnv.Target = protocol.ExactSessionTarget{}
-		_, err = EncodeClient(controlWithEnv, testEnvelopeCeiling, testChunkCeiling)
+		raw, err = EncodeClient(controlWithEnv, testEnvelopeCeiling, testChunkCeiling)
+		require.NoError(t, err)
+		_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
 		require.ErrorIs(t, err, ErrInvalidMessage)
 		noEquals := testOpen()
 		emptyPolicy := testOpen()
 		emptyPolicy.Policy = ports.BrokerPolicy{}
-		_, err = EncodeClient(emptyPolicy, testEnvelopeCeiling, testChunkCeiling)
+		raw, err = EncodeClient(emptyPolicy, testEnvelopeCeiling, testChunkCeiling)
+		require.NoError(t, err)
+		_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
 		require.ErrorIs(t, err, ErrInvalidMessage)
 		// A local control open with no attachment state is legal.
 		valid := Open{Ref: testRef(5), Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
-		raw, err := EncodeClient(valid, testEnvelopeCeiling, testChunkCeiling)
+		raw, err = EncodeClient(valid, testEnvelopeCeiling, testChunkCeiling)
 		require.NoError(t, err)
 		decoded, err := DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
 		require.NoError(t, err)
 	t.Run("exact with a creation name is refused", func(t *testing.T) {
 		open := testOpen()
 		open.Name = "work"
-		_, err := EncodeClient(open, testEnvelopeCeiling, testChunkCeiling)
+		raw, err := EncodeClient(open, testEnvelopeCeiling, testChunkCeiling)
+		require.NoError(t, err)
+		_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
 		require.ErrorIs(t, err, ErrInvalidMessage)
 	})
 	t.Run("named creation with a target is refused", func(t *testing.T) {
 		named := testOpen()
 		named.Admission = ports.BrokerAdmissionCreateNamed
 		named.Name = "work"
-		_, err := EncodeClient(named, testEnvelopeCeiling, testChunkCeiling)
+		raw, err := EncodeClient(named, testEnvelopeCeiling, testChunkCeiling)
+		require.NoError(t, err)
+		envelope := &wire.MuxClientEnvelope{}
+		require.NoError(t, proto.Unmarshal(raw, envelope))
+		envelope.GetOpen().Target = exactTargetToWire(testOpen().Target)
+		raw, err = proto.Marshal(envelope)
+		require.NoError(t, err)
+		_, err = DecodeClient(raw, testEnvelopeCeiling, testChunkCeiling)
 		require.ErrorIs(t, err, ErrInvalidMessage)
 	})
 	t.Run("unknown admission code is refused", func(t *testing.T) {
 	t.Run("encoding a zero start mode is refused", func(t *testing.T) {
 		_, err := EncodeClient(Open{Ref: testRef(6), Purpose: ports.BrokerStreamControl, Local: true, Policy: testPolicy()}, testEnvelopeCeiling, testChunkCeiling)
-		require.ErrorIs(t, err, ErrInvalidMessage)
+		require.Error(t, err) // Closed wire enum; decode refusal is tested above.
 	})
 }

M internal/adapters/daemonmux/connector.go +0 -3


@@ -309,9 +309,6 @@ func (c *PhysicalConnection) OpenStream(ctx context.Context, request ports.Broke
309	309		if c == nil \|\| c.logical == nil {
310	310		return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: ErrLogicalConfig}
311	311		}
312		-	if err := request.Validate(); err != nil {
313		-	return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: errors.Join(ports.BrokerAdmissionInvalid, err)}
314		-	}
315	312		if !request.Policy.Compatible(c.policy) {
316	313		return nil, ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy}
317	314		}


@@ -309,9 +309,6 @@ func (c *PhysicalConnection) OpenStream(ctx context.Context, request ports.Broke
309		if c == nil \|\| c.logical == nil {	309	if c == nil \|\| c.logical == nil {
310		return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: ErrLogicalConfig}	310	return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: ErrLogicalConfig}
311		}	311	}
312	-	if err := request.Validate(); err != nil {
313	-	return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: errors.Join(ports.BrokerAdmissionInvalid, err)}
314	-	}
315		if !request.Policy.Compatible(c.policy) {	312	if !request.Policy.Compatible(c.policy) {
316		return nil, ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy}	313	return nil, ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy}
317		}	314	}

M internal/adapters/daemonmux/logical_conn.go +0 -3


@@ -97,9 +97,6 @@ func (c *LogicalConnector) Open(ctx context.Context, request ports.BrokerOpenStr
97	97		if c == nil \|\| c.pump == nil {
98	98		return nil, ErrLogicalConfig
99	99		}
100		-	if err := request.Validate(); err != nil {
101		-	return nil, ErrLogicalConfig
102		-	}
103	100		if ctx == nil {
104	101		ctx = context.Background()
105	102		}


@@ -97,9 +97,6 @@ func (c *LogicalConnector) Open(ctx context.Context, request ports.BrokerOpenStr
97		if c == nil \|\| c.pump == nil {	97	if c == nil \|\| c.pump == nil {
98		return nil, ErrLogicalConfig	98	return nil, ErrLogicalConfig
99		}	99	}
100	-	if err := request.Validate(); err != nil {
101	-	return nil, ErrLogicalConfig
102	-	}
103		if ctx == nil {	100	if ctx == nil {
104		ctx = context.Background()	101	ctx = context.Background()
105		}	102	}

M internal/usecase/broker/pool.go +0 -3


@@ -161,9 +161,6 @@ func (p *Pool) CloseStream(id ports.BrokerConnectionID, stream ports.BrokerStrea
161	161		}
162	162
163	163		func (p *Pool) OpenStream(ctx context.Context, req ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
164		-	if err := req.Validate(); err != nil {
165		-	return nil, errors.Join(ports.BrokerAdmissionInvalid, err)
166		-	}
167	164		if req.Epoch != p.epoch {
168	165		return nil, ports.BrokerError{Code: ports.BrokerErrorStaleEpoch}
169	166		}


@@ -161,9 +161,6 @@ func (p *Pool) CloseStream(id ports.BrokerConnectionID, stream ports.BrokerStrea
161		}	161	}
162			162
163		func (p *Pool) OpenStream(ctx context.Context, req ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {	163	func (p *Pool) OpenStream(ctx context.Context, req ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
164	-	if err := req.Validate(); err != nil {
165	-	return nil, errors.Join(ports.BrokerAdmissionInvalid, err)
166	-	}
167		if req.Epoch != p.epoch {	164	if req.Epoch != p.epoch {
168		return nil, ports.BrokerError{Code: ports.BrokerErrorStaleEpoch}	165	return nil, ports.BrokerError{Code: ports.BrokerErrorStaleEpoch}
169		}	166	}

M internal/usecase/broker/pool_test.go +0 -4


@@ -248,10 +248,6 @@ func TestPoolStreamWindowAdmission(t *testing.T) {
248	248		// A duplicate inside the window is stale.
249	249		_, err = p.OpenStream(context.Background(), poolRequest(id, 1))
250	250		require.ErrorIs(t, err, ports.BrokerAdmissionStale)
251		-	// Zero is never a valid identity.
252		-	_, err = p.OpenStream(context.Background(), poolRequest(id, 0))
253		-	require.ErrorIs(t, err, ports.BrokerAdmissionInvalid)
254		-
255	251		// Push the newest admitted identity past the window, then replay a low one.
256	252		high, err := p.OpenStream(context.Background(), poolRequest(id, ports.BrokerStreamWindowSize+2))
257	253		require.NoError(t, err)


@@ -248,10 +248,6 @@ func TestPoolStreamWindowAdmission(t *testing.T) {
248		// A duplicate inside the window is stale.	248	// A duplicate inside the window is stale.
249		_, err = p.OpenStream(context.Background(), poolRequest(id, 1))	249	_, err = p.OpenStream(context.Background(), poolRequest(id, 1))
250		require.ErrorIs(t, err, ports.BrokerAdmissionStale)	250	require.ErrorIs(t, err, ports.BrokerAdmissionStale)
251	-	// Zero is never a valid identity.
252	-	_, err = p.OpenStream(context.Background(), poolRequest(id, 0))
253	-	require.ErrorIs(t, err, ports.BrokerAdmissionInvalid)
254	-
255		// Push the newest admitted identity past the window, then replay a low one.	251	// Push the newest admitted identity past the window, then replay a low one.
256		high, err := p.OpenStream(context.Background(), poolRequest(id, ports.BrokerStreamWindowSize+2))	252	high, err := p.OpenStream(context.Background(), poolRequest(id, ports.BrokerStreamWindowSize+2))
257		require.NoError(t, err)	253	require.NoError(t, err)

M internal/usecase/broker/service.go +4 -0


@@ -264,6 +264,10 @@ func (s *Service) OpenStream(ctx context.Context, request ports.BrokerOpenStream
264	264		if err := s.scope(&request); err != nil {
265	265		return nil, err
266	266		}
	267	+	// Service admission owns validation of caller input before the pool sees it.
	268	+	if err := request.Validate(); err != nil {
	269	+	return nil, errors.Join(ports.BrokerAdmissionInvalid, err)
	270	+	}
267	271		s.mu.Lock()
268	272		if s.closed {
269	273		s.mu.Unlock()


@@ -264,6 +264,10 @@ func (s *Service) OpenStream(ctx context.Context, request ports.BrokerOpenStream
264	if err := s.scope(&request); err != nil {	264		if err := s.scope(&request); err != nil {
265	return nil, err	265		return nil, err
266	}	266		}
		267	+	// Service admission owns validation of caller input before the pool sees it.
		268	+	if err := request.Validate(); err != nil {
		269	+	return nil, errors.Join(ports.BrokerAdmissionInvalid, err)
		270	+	}
267	s.mu.Lock()	271		s.mu.Lock()
268	if s.closed {	272		if s.closed {
269	s.mu.Unlock()	273		s.mu.Unlock()

M internal/usecase/broker/service_test.go +24 -0


@@ -430,6 +430,30 @@ func TestServiceNextStreamIDAllocation(t *testing.T) {
430	430		})
431	431		}
432	432
	433	+	func TestServiceOpenStreamAdmissionValidatesCallerInput(t *testing.T) {
	434	+	authority, _, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
	435	+	service, err := authority.AdmitClient(context.Background())
	436	+	require.NoError(t, err)
	437	+	t.Cleanup(func() { require.NoError(t, service.Close()) })
	438	+
	439	+	for _, tc := range []struct {
	440	+	name string
	441	+	mutate func(*ports.BrokerOpenStreamRequest)
	442	+	}{
	443	+	{"missing stream identity", func(r *ports.BrokerOpenStreamRequest) { r.Stream = 0 }},
	444	+	{"invalid purpose", func(r *ports.BrokerOpenStreamRequest) { r.Purpose = 0 }},
	445	+	{"invalid policy", func(r *ports.BrokerOpenStreamRequest) { r.Policy.Transport = "" }},
	446	+	{"observation cannot start daemon", func(r *ports.BrokerOpenStreamRequest) { r.Purpose = ports.BrokerStreamObservation }},
	447	+	} {
	448	+	t.Run(tc.name, func(t *testing.T) {
	449	+	request := poolRequest(service.ConnectionID(), 1)
	450	+	tc.mutate(&request)
	451	+	_, err := service.OpenStream(context.Background(), request)
	452	+	require.ErrorIs(t, err, ports.BrokerAdmissionInvalid)
	453	+	})
	454	+	}
	455	+	}
	456	+
433	457		func TestServiceCloseStreamScopeFencing(t *testing.T) {
434	458		authority, _, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
435	459		service, err := authority.AdmitClient(context.Background())


@@ -430,6 +430,30 @@ func TestServiceNextStreamIDAllocation(t *testing.T) {
430	})	430		})
431	}	431		}
432		432
		433	+	func TestServiceOpenStreamAdmissionValidatesCallerInput(t *testing.T) {
		434	+	authority, _, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
		435	+	service, err := authority.AdmitClient(context.Background())
		436	+	require.NoError(t, err)
		437	+	t.Cleanup(func() { require.NoError(t, service.Close()) })
		438	+
		439	+	for _, tc := range []struct {
		440	+	name string
		441	+	mutate func(*ports.BrokerOpenStreamRequest)
		442	+	}{
		443	+	{"missing stream identity", func(r *ports.BrokerOpenStreamRequest) { r.Stream = 0 }},
		444	+	{"invalid purpose", func(r *ports.BrokerOpenStreamRequest) { r.Purpose = 0 }},
		445	+	{"invalid policy", func(r *ports.BrokerOpenStreamRequest) { r.Policy.Transport = "" }},
		446	+	{"observation cannot start daemon", func(r *ports.BrokerOpenStreamRequest) { r.Purpose = ports.BrokerStreamObservation }},
		447	+	} {
		448	+	t.Run(tc.name, func(t *testing.T) {
		449	+	request := poolRequest(service.ConnectionID(), 1)
		450	+	tc.mutate(&request)
		451	+	_, err := service.OpenStream(context.Background(), request)
		452	+	require.ErrorIs(t, err, ports.BrokerAdmissionInvalid)
		453	+	})
		454	+	}
		455	+	}
		456	+
433	func TestServiceCloseStreamScopeFencing(t *testing.T) {	457		func TestServiceCloseStreamScopeFencing(t *testing.T) {
434	authority, _, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)	458		authority, _, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
435	service, err := authority.AdmitClient(context.Background())	459		service, err := authority.AdmitClient(context.Background())