1package daemonmux
2
3// Daemonmux physical preamble.
4//
5// The first daemonmux client frame on a physical connection is always
6// MuxPreambleRequest; the first daemon frame is always MuxPreambleResponse.
7// The preamble is an adapter message pair, not an envelope variant. It
8// reuses the shared session preamble magic, epoch, and exact protocol
9// version, and the shared refusal taxonomy, with daemonmux-specific roles: 5
10// names the broker client, 6 names the daemon server.
11//
12// The preamble negotiates immutable per-connection ceilings once: the
13// receive envelope ceiling (1 MiB..16 MiB), the stream chunk ceiling
14// (1..64 KiB), the concurrent stream ceiling (1..128), and the aggregate
15// buffered-bytes ceiling (1..64 MiB). The response carries the effective
16// minimum of both advertisements, the authenticated daemon identity, its
17// 16-byte incarnation, and the accepted connection policy. The effective
18// minima are the exported MuxCeilings value (limits.go) that
19// NewStreamEngineWithCeilings and NewSchedulerWithCeilings consume. Capability
20// bits must be zero: daemonmux negotiates no optional capabilities.
21//
22// Refusals carry the precise shared PreambleRejectionCode taxonomy: 1 bad
23// magic, 2 epoch mismatch, 3 version mismatch, 4 wrong role, 5 duplicate, 6
24// out of order, 7 limit refused. Code 5 means a duplicate observed inside one
25// frame: the strict scanner found a repeated singular field or oneof member
26// (wire.ErrScanDuplicate) in the serialized preamble this codec decodes. The
27// connection dispatcher separately refuses a second preamble frame on one
28// connection, also with code 5, but that stateful, cross-frame duplicate is
29// never observed here. Code 6 is reserved for the dispatcher too: out-of-order
30// application frames are a framing-order failure this stateless codec never
31// sees. A refused policy is also a code 7 refusal, because daemonmux negotiates
32// exact policy equality rather than a capability set.
33//
34// A response is either an acceptance or a refusal and never a blend of the
35// two: an acceptance carries no rejection field, and a refusal carries none of
36// the accepted-only authority fields (accepted policy, daemon identity,
37// incarnation) and a rejection code in 1..7. Both directions are enforced on
38// encode and decode so a peer can never smuggle accepted authority into a
39// refusal or a stale refusal code into an acceptance.
40
41import (
42 "errors"
43
44 "google.golang.org/protobuf/proto"
45
46 "github.com/bnema/vev/internal/ports"
47 "github.com/bnema/vev/internal/protocol"
48 "github.com/bnema/vev/internal/protocol/wire"
49)
50
51// ErrPreambleRejected reports a daemonmux preamble that fails validation:
52// bad magic, epoch, version, role, capability bits, limit advertisement,
53// policy, identity, or incarnation; a refusal code outside 1..7; an
54// acceptance that also carries a rejection field; or a refusal that also
55// carries accepted-only authority. Callers map it to a precise rejection code
56// with RejectionCodeFor.
57var ErrPreambleRejected = errors.New("daemonmux: preamble rejected")
58
59// Preamble rejection codes. They mirror the session preamble refusal
60// taxonomy carried on wire.PreambleRejectionCode.
61const (
62 // RejectionBadMagic reports a preamble magic mismatch.
63 RejectionBadMagic uint32 = 1
64 // RejectionEpochMismatch reports a protocol epoch mismatch.
65 RejectionEpochMismatch uint32 = 2
66 // RejectionVersionMismatch reports a session protocol version mismatch.
67 RejectionVersionMismatch uint32 = 3
68 // RejectionWrongRole reports a preamble role mismatch.
69 RejectionWrongRole uint32 = 4
70 // RejectionDuplicate reports a duplicate within one frame: the strict
71 // scanner found a repeated singular field or oneof member
72 // (wire.ErrScanDuplicate) in the serialized preamble. The connection
73 // dispatcher separately refuses a second preamble frame on one connection
74 // with this same code; that stateful, cross-frame duplicate is never
75 // observed by this stateless codec.
76 RejectionDuplicate uint32 = 5
77 // RejectionOutOfOrder is reserved for the connection dispatcher, which
78 // owns framing-order detection. The stateless codec never emits it, but it
79 // accepts it as a well-formed refusal code on the wire.
80 RejectionOutOfOrder uint32 = 6
81 // RejectionLimitRefused reports an unnegotiable advertisement. It also
82 // covers nonzero capability bits and a policy the daemon does not
83 // accept: the protocol negotiates neither capabilities nor a merged policy
84 // server, so either is an unnegotiable advertisement.
85 RejectionLimitRefused uint32 = 7
86)
87
88// validRejectionCode reports whether code is one of the shared preamble
89// refusal codes 1..7. The stateless codec accepts code 6 (out of order) as a
90// well-formed code on the wire even though it never emits it, because the
91// dispatcher owns framing-order detection.
92func validRejectionCode(code uint32) bool {
93 return code >= RejectionBadMagic && code <= RejectionLimitRefused
94}
95
96// MuxPreambleRequest is the validated daemonmux client preamble: the
97// requested ceilings plus the broker policy the client asks the daemon to
98// accept.
99type MuxPreambleRequest struct {
100 Ceilings MuxCeilings
101 Policy ports.BrokerPolicy
102}
103
104// MuxPreambleResponse is the validated daemonmux server preamble: the
105// effective ceilings, the authenticated daemon identity and incarnation, and
106// the accepted policy on acceptance; a precise refusal code otherwise.
107type MuxPreambleResponse struct {
108 Ceilings MuxCeilings
109 Policy ports.BrokerPolicy
110 Identity ports.BrokerDaemonIdentity
111 Incarnation ports.BrokerDaemonIncarnation
112 Accepted bool
113 Code uint32
114}
115
116// EncodePreambleRequest builds the local daemonmux client preamble offer. It
117// fails closed on a ceiling advertisement outside its window or a policy that
118// could not be accepted.
119func EncodePreambleRequest(ceilings MuxCeilings, policy ports.BrokerPolicy) (*wire.MuxPreambleRequest, error) {
120 if err := checkMuxCeilings(ceilings); err != nil {
121 return nil, err
122 }
123 if err := policy.Validate(); err != nil {
124 return nil, ErrPreambleRejected
125 }
126 return &wire.MuxPreambleRequest{
127 Magic: wire.PreambleMagic,
128 Epoch: wire.ProtocolEpoch,
129 Version: uint32(protocol.Version),
130 Role: &wire.PreambleRole{Role: MuxRoleClient},
131 MaxReceiveEnvelopeBytes: ceilings.MaxReceiveEnvelopeBytes,
132 StreamChunkLimit: ceilings.StreamChunkLimit,
133 MaxStreams: ceilings.MaxStreams,
134 MaxAggregateBytes: ceilings.MaxAggregateBytes,
135 Policy: policyToWire(policy),
136 }, nil
137}
138
139// EncodePreambleResponse builds the daemonmux server preamble answer:
140// acceptance with the effective ceilings, the authenticated daemon identity
141// and incarnation, and the accepted policy, or a typed refusal carrying the
142// precise rejection code. The answer is either an acceptance or a refusal and
143// never a blend: a refusal demands a code in 1..7 and carries no accepted
144// authority, while an acceptance demands a zero code and never carries a
145// rejection field.
146func EncodePreambleResponse(accepted bool, ceilings MuxCeilings, policy ports.BrokerPolicy, identity ports.BrokerDaemonIdentity, incarnation ports.BrokerDaemonIncarnation, code uint32) (*wire.MuxPreambleResponse, error) {
147 response := &wire.MuxPreambleResponse{
148 Magic: wire.PreambleMagic,
149 Epoch: wire.ProtocolEpoch,
150 Version: uint32(protocol.Version),
151 Role: &wire.PreambleRole{Role: MuxRoleServer},
152 MaxReceiveEnvelopeBytes: ceilings.MaxReceiveEnvelopeBytes,
153 StreamChunkLimit: ceilings.StreamChunkLimit,
154 MaxStreams: ceilings.MaxStreams,
155 MaxAggregateBytes: ceilings.MaxAggregateBytes,
156 Accepted: accepted,
157 }
158 if !accepted {
159 if !validRejectionCode(code) {
160 return nil, ErrPreambleRejected
161 }
162 response.Rejection = &wire.PreambleRejectionCode{Code: code}
163 return response, nil
164 }
165 if code != 0 {
166 // An acceptance carries no rejection field; a nonzero code is a
167 // caller bug that must fail closed rather than be dropped silently.
168 return nil, ErrPreambleRejected
169 }
170 if err := checkMuxCeilings(ceilings); err != nil {
171 return nil, err
172 }
173 if err := policy.Validate(); err != nil {
174 return nil, ErrPreambleRejected
175 }
176 if err := identity.Validate(); err != nil {
177 return nil, ErrPreambleRejected
178 }
179 if err := incarnation.Validate(); err != nil {
180 return nil, ErrPreambleRejected
181 }
182 response.AcceptedPolicy = policyToWire(policy)
183 response.DaemonIdentity = string(identity)
184 response.Incarnation = append([]byte(nil), incarnation[:]...)
185 return response, nil
186}
187
188// DecodePreambleRequest validates one raw daemonmux client preamble. Magic,
189// epoch, version, role, and zero capability bits are checked before the
190// ceiling advertisement and the requested policy; the refusal code is
191// precise.
192func DecodePreambleRequest(message *wire.MuxPreambleRequest) (MuxPreambleRequest, error) {
193 if message == nil {
194 return MuxPreambleRequest{}, ErrPreambleRejected
195 }
196 if message.GetMagic() != wire.PreambleMagic {
197 return MuxPreambleRequest{}, ErrPreambleRejected
198 }
199 if message.GetEpoch() != wire.ProtocolEpoch {
200 return MuxPreambleRequest{}, ErrPreambleRejected
201 }
202 if message.GetVersion() != uint32(protocol.Version) {
203 return MuxPreambleRequest{}, ErrPreambleRejected
204 }
205 if message.GetRole().GetRole() != MuxRoleClient {
206 return MuxPreambleRequest{}, ErrPreambleRejected
207 }
208 if message.GetCapabilityBits() != 0 {
209 return MuxPreambleRequest{}, ErrPreambleRejected
210 }
211 ceilings := MuxCeilings{
212 MaxReceiveEnvelopeBytes: message.GetMaxReceiveEnvelopeBytes(),
213 StreamChunkLimit: message.GetStreamChunkLimit(),
214 MaxStreams: message.GetMaxStreams(),
215 MaxAggregateBytes: message.GetMaxAggregateBytes(),
216 }
217 if err := checkMuxCeilings(ceilings); err != nil {
218 return MuxPreambleRequest{}, err
219 }
220 policy, err := policyFromWire(message.GetPolicy())
221 if err != nil {
222 return MuxPreambleRequest{}, ErrPreambleRejected
223 }
224 return MuxPreambleRequest{Ceilings: ceilings, Policy: policy}, nil
225}
226
227// DecodePreambleResponse validates one raw daemonmux server preamble.
228// Refusals fail with ErrPreambleRejected and carry their code; acceptance
229// additionally validates the server role, the effective ceilings, the
230// accepted policy, the authenticated daemon identity, and the 16-byte
231// incarnation. An acceptance carrying a rejection field and a refusal carrying
232// accepted-only authority are both rejected, as is any refusal code outside
233// 1..7.
234func DecodePreambleResponse(message *wire.MuxPreambleResponse) (MuxPreambleResponse, error) {
235 if message == nil {
236 return MuxPreambleResponse{}, ErrPreambleRejected
237 }
238 if message.GetMagic() != wire.PreambleMagic {
239 return MuxPreambleResponse{}, ErrPreambleRejected
240 }
241 if message.GetEpoch() != wire.ProtocolEpoch {
242 return MuxPreambleResponse{}, ErrPreambleRejected
243 }
244 if message.GetVersion() != uint32(protocol.Version) {
245 return MuxPreambleResponse{}, ErrPreambleRejected
246 }
247 if message.GetRole().GetRole() != MuxRoleServer {
248 return MuxPreambleResponse{}, ErrPreambleRejected
249 }
250 if message.GetCapabilityBits() != 0 {
251 return MuxPreambleResponse{}, ErrPreambleRejected
252 }
253 if !message.GetAccepted() {
254 code := message.GetRejection().GetCode()
255 if !validRejectionCode(code) {
256 return MuxPreambleResponse{}, ErrPreambleRejected
257 }
258 if message.GetAcceptedPolicy() != nil || message.GetDaemonIdentity() != "" || len(message.GetIncarnation()) != 0 {
259 // A refusal carries no accepted-only authority: accepted policy,
260 // daemon identity, and incarnation belong to an acceptance alone.
261 return MuxPreambleResponse{}, ErrPreambleRejected
262 }
263 return MuxPreambleResponse{Code: code}, ErrPreambleRejected
264 }
265 if message.GetRejection() != nil {
266 // An acceptance carries no rejection field.
267 return MuxPreambleResponse{}, ErrPreambleRejected
268 }
269 ceilings := MuxCeilings{
270 MaxReceiveEnvelopeBytes: message.GetMaxReceiveEnvelopeBytes(),
271 StreamChunkLimit: message.GetStreamChunkLimit(),
272 MaxStreams: message.GetMaxStreams(),
273 MaxAggregateBytes: message.GetMaxAggregateBytes(),
274 }
275 if err := checkMuxCeilings(ceilings); err != nil {
276 return MuxPreambleResponse{}, err
277 }
278 policy, err := policyFromWire(message.GetAcceptedPolicy())
279 if err != nil {
280 return MuxPreambleResponse{}, ErrPreambleRejected
281 }
282 identity := ports.BrokerDaemonIdentity(message.GetDaemonIdentity())
283 if err := identity.Validate(); err != nil {
284 return MuxPreambleResponse{}, ErrPreambleRejected
285 }
286 var incarnation ports.BrokerDaemonIncarnation
287 raw := message.GetIncarnation()
288 if len(raw) != len(incarnation) {
289 return MuxPreambleResponse{}, ErrPreambleRejected
290 }
291 copy(incarnation[:], raw)
292 if err := incarnation.Validate(); err != nil {
293 return MuxPreambleResponse{}, ErrPreambleRejected
294 }
295 return MuxPreambleResponse{
296 Ceilings: ceilings,
297 Policy: policy,
298 Identity: identity,
299 Incarnation: incarnation,
300 Accepted: true,
301 }, nil
302}
303
304// ValidatePreambleResponseAgainstOffer refuses an accepted daemonmux response
305// whose effective ceilings exceed the ceilings the local side offered. The
306// effective value is the element-wise minimum of both advertisements, so a
307// conforming daemon never grants more than the requester offered; a response
308// above the offer is a protocol violation. A refusal (which carries no
309// accepted ceilings), or an invalid advertisement on either side, is refused
310// with ErrPreambleRejected. A response strictly below the offer is accepted:
311// the daemon may lower any ceiling by its own policy.
312func ValidatePreambleResponseAgainstOffer(response MuxPreambleResponse, offered MuxCeilings) error {
313 if err := checkMuxCeilings(offered); err != nil {
314 return err
315 }
316 if !response.Accepted {
317 return ErrPreambleRejected
318 }
319 if err := checkMuxCeilings(response.Ceilings); err != nil {
320 return err
321 }
322 if response.Ceilings.MaxReceiveEnvelopeBytes > offered.MaxReceiveEnvelopeBytes ||
323 response.Ceilings.StreamChunkLimit > offered.StreamChunkLimit ||
324 response.Ceilings.MaxStreams > offered.MaxStreams ||
325 response.Ceilings.MaxAggregateBytes > offered.MaxAggregateBytes {
326 return ErrPreambleRejected
327 }
328 return nil
329}
330
331// DecodePreambleResponseBytes strictly decodes one serialized daemonmux server
332// preamble under the same 4 KiB bound, strict scan, generated unmarshal, and
333// semantic decode as DecodePreambleRequestBytes. Carriage failures return the
334// precise wire sentinel; oversize payloads and semantic failures return
335// ErrPreambleRejected.
336func DecodePreambleResponseBytes(payload []byte) (MuxPreambleResponse, error) {
337 if err := CheckPreambleSize(payload); err != nil {
338 return MuxPreambleResponse{}, err
339 }
340 message := &wire.MuxPreambleResponse{}
341 if err := wire.ScanEnvelope(message, payload); err != nil {
342 return MuxPreambleResponse{}, err
343 }
344 if err := (proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(payload, message)); err != nil {
345 return MuxPreambleResponse{}, err
346 }
347 return DecodePreambleResponse(message)
348}
349
350// RejectionCodeFor maps a preamble validation failure to its precise refusal
351// code. Limit, magic, epoch, version, role, and policy failures each carry
352// their own code; nonzero capability bits are refused as
353// RejectionLimitRefused because the protocol negotiates no capabilities; unknown scan
354// failures are limit refusals. A duplicate inside one frame
355// (wire.ErrScanDuplicate) maps to RejectionDuplicate with the in-frame meaning
356// described above, never the dispatcher's cross-frame duplicate. Code 6 is
357// never returned here: the dispatcher owns out-of-order framing.
358func RejectionCodeFor(request *wire.MuxPreambleRequest, err error) uint32 {
359 if err == nil {
360 return 0
361 }
362 if request != nil {
363 if request.GetMagic() != wire.PreambleMagic {
364 return RejectionBadMagic
365 }
366 if request.GetEpoch() != wire.ProtocolEpoch {
367 return RejectionEpochMismatch
368 }
369 if request.GetVersion() != uint32(protocol.Version) {
370 return RejectionVersionMismatch
371 }
372 if request.GetRole().GetRole() != MuxRoleClient {
373 return RejectionWrongRole
374 }
375 if request.GetCapabilityBits() != 0 {
376 // No capability is negotiable here, so any bit is a refused
377 // limit advertisement (code 7), documented explicitly.
378 return RejectionLimitRefused
379 }
380 remote := MuxCeilings{
381 MaxReceiveEnvelopeBytes: request.GetMaxReceiveEnvelopeBytes(),
382 StreamChunkLimit: request.GetStreamChunkLimit(),
383 MaxStreams: request.GetMaxStreams(),
384 MaxAggregateBytes: request.GetMaxAggregateBytes(),
385 }
386 if checkMuxCeilings(remote) != nil {
387 return RejectionLimitRefused
388 }
389 if _, err := policyFromWire(request.GetPolicy()); err != nil {
390 // A malformed or unnegotiable requested policy is a refused
391 // advertisement, not a transport failure.
392 return RejectionLimitRefused
393 }
394 }
395 if errors.Is(err, wire.ErrScanDuplicate) {
396 return RejectionDuplicate
397 }
398 return RejectionLimitRefused
399}