1package daemonmux
2
3// Physical preamble handshake.
4//
5// The daemonmux connection begins with exactly one physical preamble exchange
6// over the private FramedCarrier, before any application envelope. The client
7// (broker) sends one MuxPreambleRequest carrying its advertised ceilings and
8// the connection policy it needs; the server (daemon) answers with exactly one
9// MuxPreambleResponse carrying either the effective (minimum) ceilings, the
10// authoritative daemon identity and incarnation, and the accepted policy, or a
11// typed refusal. The message pair is the existing stateless preamble codec
12// (preamble.go): this file is only the stateful exchange and the authority
13// checks on top of it. Magic, epoch, exact protocol.Version, roles 5/6, zero
14// capability bits, the 4 KiB preamble bound, and valid/refusal exclusivity are
15// enforced by that codec.
16//
17// The client verifies the accepted response against the independently resolved
18// endpoint it dialed: the response identity must exactly equal
19// ports.BrokerDialTarget.ExpectedIdentity.Identity, the accepted policy must exactly equal
20// the endpoint policy, the incarnation must be nonzero, and the effective
21// ceilings must never exceed what the client offered. The response is the
22// server's authority, never a value the client supplied and got echoed back.
23//
24// The server verifies the request policy against its immutable ServerBinding
25// (binding.go) instead of echoing the requested policy: an accepted answer
26// carries the binding's identity, incarnation, and policy, and any request
27// whose policy differs in a single field is refused. The server negotiates the
28// element-wise minima of its local advertisement and the request, so a peer
29// can never grant or demand more than either side offered.
30//
31// One caller-supplied context (or its deadline) bounds the whole exchange: the
32// same value goes to the carrier's Send and Receive, no timer is created here,
33// and there is no internal retry or grace period. Every failed handshake -
34// local validation failure, carrier error, peer refusal, identity or policy
35// mismatch, or an inadmissible advertisement - closes the carrier before
36// returning, so no half-open physical connection survives a refusal. A
37// successful handshake leaves the carrier open and untouched for its owner:
38// the handshake never starts a Pump, never admits a stream, and never wraps
39// the connection in ports values. The caller starts the Pump only after this
40// function returns nil.
41//
42// Authentication premise: the handshake does not authenticate the carrier and
43// does not prove an identity in band. The carriage must already be
44// authenticated, and the expected endpoint binding is authoritative. The
45// identity, incarnation, and policy here are authorization and pooling
46// metadata on top of that premise, not a substitute for it.
47
48import (
49 "context"
50 "errors"
51 "fmt"
52
53 "google.golang.org/protobuf/proto"
54
55 "github.com/bnema/vev/internal/ports"
56 "github.com/bnema/vev/internal/protocol/wire"
57)
58
59var (
60 // ErrHandshakeRejected reports a physical preamble handshake that was
61 // refused by the peer or failed a local authority check on an otherwise
62 // well-formed response. A peer refusal is reported as a *HandshakeRefusal
63 // carrying its precise code and also matches errors.Is(ErrHandshakeRejected).
64 ErrHandshakeRejected = errors.New("daemonmux: physical handshake rejected")
65 // ErrHandshakeConfig reports a handshake called without a carrier, with an
66 // invalid endpoint or binding, or with an invalid local ceiling
67 // advertisement. It is a caller bug, not a peer refusal.
68 ErrHandshakeConfig = errors.New("daemonmux: invalid handshake configuration")
69)
70
71// HandshakeRefusal is the typed refusal a physical preamble handshake returns:
72// the precise PreambleRejectionCode the peer sent or the server sent.
73type HandshakeRefusal struct {
74 Code uint32
75}
76
77func (r *HandshakeRefusal) Error() string {
78 return fmt.Sprintf("daemonmux: physical handshake refused with code %d", r.Code)
79}
80
81// Is classifies every typed refusal as ErrHandshakeRejected so a caller can
82// test with errors.Is without losing the precise code carried on the value.
83func (r *HandshakeRefusal) Is(target error) bool { return target == ErrHandshakeRejected }
84
85// ClientHandshakeResult is the broker's validated physical handshake outcome:
86// the negotiated (effective) ceilings, and the daemon binding the server
87// proved against the independently resolved endpoint.
88type ClientHandshakeResult struct {
89 Ceilings MuxCeilings
90 Identity ports.BrokerDaemonIdentity
91 Incarnation ports.BrokerDaemonIncarnation
92 Policy ports.BrokerPolicy
93}
94
95// ServerHandshakeResult is the daemon's validated physical handshake outcome:
96// the negotiated (effective) ceilings and the binding the server enforced.
97type ServerHandshakeResult struct {
98 Ceilings MuxCeilings
99 Binding ServerBinding
100 // Policy is the exact immutable authority member admitted for this peer.
101 Policy ports.BrokerPolicy
102 // Origin is the provisioned locality of the admitted member, carried from
103 // the same provisioned entry Policy was resolved from. A downstream listener
104 // stamps it on every admitted session connection so a daemon use case can
105 // enforce the local/remote locality bound without re-deriving it.
106 Origin ports.SessionConnectionOrigin
107}
108
109// RunClientHandshake performs the broker side of the daemonmux physical
110// preamble over carrier. It offers offered and the resolved endpoint's policy,
111// then accepts exactly one response and verifies it against the endpoint:
112// identity and policy must match exactly, the incarnation must be nonzero, and
113// the effective ceilings must not exceed the offer. It returns a typed
114// *HandshakeRefusal for a peer refusal and ErrHandshakeRejected for any local
115// authority mismatch. The caller's ctx (or its deadline) bounds both the send
116// and the receive. Every failure closes carrier; success leaves it open and
117// does not start a pump.
118func RunClientHandshake(ctx context.Context, carrier FramedCarrier, endpoint ports.BrokerDialTarget, offered MuxCeilings) (ClientHandshakeResult, error) {
119 if ctx == nil {
120 ctx = context.Background()
121 }
122 if carrier == nil {
123 return ClientHandshakeResult{}, ErrHandshakeConfig
124 }
125 if err := endpoint.Validate(); err != nil {
126 closeFailedHandshake(carrier)
127 return ClientHandshakeResult{}, ErrHandshakeConfig
128 }
129 if err := offered.Validate(); err != nil {
130 closeFailedHandshake(carrier)
131 return ClientHandshakeResult{}, ErrHandshakeConfig
132 }
133
134 request, err := EncodePreambleRequest(offered, endpoint.Policy)
135 if err != nil {
136 closeFailedHandshake(carrier)
137 return ClientHandshakeResult{}, ErrHandshakeConfig
138 }
139 offer, err := proto.Marshal(request)
140 if err != nil {
141 closeFailedHandshake(carrier)
142 return ClientHandshakeResult{}, errors.Join(ErrHandshakeConfig, err)
143 }
144 if err := carrier.Send(ctx, offer); err != nil {
145 closeFailedHandshake(carrier)
146 return ClientHandshakeResult{}, err
147 }
148 payload, err := carrier.Receive(ctx)
149 if err != nil {
150 closeFailedHandshake(carrier)
151 return ClientHandshakeResult{}, err
152 }
153 response, err := DecodePreambleResponseBytes(payload)
154 if err != nil {
155 closeFailedHandshake(carrier)
156 if !response.Accepted && response.Code != 0 {
157 return ClientHandshakeResult{}, &HandshakeRefusal{Code: response.Code}
158 }
159 return ClientHandshakeResult{}, ErrHandshakeRejected
160 }
161 if err := validateClientResponse(response, endpoint, offered); err != nil {
162 closeFailedHandshake(carrier)
163 return ClientHandshakeResult{}, err
164 }
165 return ClientHandshakeResult{
166 Ceilings: EffectiveMuxCeilings(offered, response.Ceilings),
167 Identity: response.Identity,
168 Incarnation: response.Incarnation,
169 Policy: response.Policy,
170 }, nil
171}
172
173// validateClientResponse is the broker's authority check on one accepted
174// response: identity and policy exactly equal the resolved endpoint, the
175// incarnation is nonzero, and the effective ceilings never exceed the offer.
176// A non-acceptance is refused too, so a caller can never treat a refusal as an
177// accepted empty response.
178func validateClientResponse(response MuxPreambleResponse, endpoint ports.BrokerDialTarget, offered MuxCeilings) error {
179 if !response.Accepted {
180 return ErrHandshakeRejected
181 }
182 if endpoint.ExpectedIdentity.Bound && response.Identity != endpoint.ExpectedIdentity.Identity {
183 return ErrHandshakeRejected
184 }
185 if err := response.Identity.Validate(); err != nil {
186 return ErrHandshakeRejected
187 }
188 if !response.Policy.Compatible(endpoint.Policy) {
189 return ErrHandshakeRejected
190 }
191 if err := response.Incarnation.Validate(); err != nil {
192 return ErrHandshakeRejected
193 }
194 if err := ValidatePreambleResponseAgainstOffer(response, offered); err != nil {
195 return ErrHandshakeRejected
196 }
197 return nil
198}
199
200// RunServerHandshake performs the daemon side of the daemonmux physical
201// preamble over carrier. It accepts exactly one request, refuses it with the
202// precise rejection code when it is malformed or its policy differs from the
203// immutable binding, and otherwise answers with the binding's identity,
204// incarnation, and policy over the effective ceilings. The caller's ctx (or
205// its deadline) bounds both the receive and the send. Every failure sends the
206// typed refusal where possible, closes carrier, and returns a typed
207// *HandshakeRefusal; success leaves carrier open and does not start a pump.
208func RunServerHandshake(ctx context.Context, carrier FramedCarrier, binding ServerBinding, offered MuxCeilings) (ServerHandshakeResult, error) {
209 if ctx == nil {
210 ctx = context.Background()
211 }
212 if carrier == nil {
213 return ServerHandshakeResult{}, ErrHandshakeConfig
214 }
215 if err := binding.Validate(); err != nil {
216 closeFailedHandshake(carrier)
217 return ServerHandshakeResult{}, ErrHandshakeConfig
218 }
219 if err := offered.Validate(); err != nil {
220 closeFailedHandshake(carrier)
221 return ServerHandshakeResult{}, ErrHandshakeConfig
222 }
223
224 payload, err := carrier.Receive(ctx)
225 if err != nil {
226 closeFailedHandshake(carrier)
227 return ServerHandshakeResult{}, err
228 }
229 request, code, err := decodePreambleRequestWithCode(payload)
230 if err != nil {
231 _ = sendHandshakeRefusal(ctx, carrier, offered, code)
232 closeFailedHandshake(carrier)
233 return ServerHandshakeResult{}, &HandshakeRefusal{Code: code}
234 }
235 accepted, ok := binding.Accepted(request.Policy)
236 if !ok {
237 code := uint32(RejectionLimitRefused)
238 _ = sendHandshakeRefusal(ctx, carrier, offered, code)
239 closeFailedHandshake(carrier)
240 return ServerHandshakeResult{}, &HandshakeRefusal{Code: code}
241 }
242
243 effective := EffectiveMuxCeilings(offered, request.Ceilings)
244 response, err := EncodePreambleResponse(true, effective, accepted.Policy, binding.Identity(), binding.Incarnation(), 0)
245 if err != nil {
246 closeFailedHandshake(carrier)
247 return ServerHandshakeResult{}, ErrHandshakeConfig
248 }
249 answer, err := proto.Marshal(response)
250 if err != nil {
251 closeFailedHandshake(carrier)
252 return ServerHandshakeResult{}, errors.Join(ErrHandshakeConfig, err)
253 }
254 if err := carrier.Send(ctx, answer); err != nil {
255 closeFailedHandshake(carrier)
256 return ServerHandshakeResult{}, err
257 }
258 return ServerHandshakeResult{Ceilings: effective, Binding: binding, Policy: accepted.Policy, Origin: accepted.Origin}, nil
259}
260
261// decodePreambleRequestWithCode is the server's strict request decode paired
262// with the precise refusal code for its failure, so a refusal always carries
263// the shared PreambleRejectionCode taxonomy rather than a generic limit code.
264func decodePreambleRequestWithCode(payload []byte) (MuxPreambleRequest, uint32, error) {
265 if err := CheckPreambleSize(payload); err != nil {
266 return MuxPreambleRequest{}, RejectionLimitRefused, err
267 }
268 message := &wire.MuxPreambleRequest{}
269 if err := wire.ScanEnvelope(message, payload); err != nil {
270 return MuxPreambleRequest{}, RejectionCodeFor(nil, err), err
271 }
272 if err := (proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(payload, message)); err != nil {
273 return MuxPreambleRequest{}, RejectionCodeFor(message, err), err
274 }
275 request, err := DecodePreambleRequest(message)
276 if err != nil {
277 return MuxPreambleRequest{}, RejectionCodeFor(message, err), err
278 }
279 return request, 0, nil
280}
281
282// sendHandshakeRefusal best-effort sends one typed refusal. The caller closes
283// the carrier immediately afterwards regardless of the send result, so a
284// failed refusal write never leaves the connection half-open.
285func sendHandshakeRefusal(ctx context.Context, carrier FramedCarrier, offered MuxCeilings, code uint32) error {
286 response, err := EncodePreambleResponse(false, offered, ports.BrokerPolicy{}, "", ports.BrokerDaemonIncarnation{}, code)
287 if err != nil {
288 return err
289 }
290 answer, err := proto.Marshal(response)
291 if err != nil {
292 return err
293 }
294 return carrier.Send(ctx, answer)
295}
296
297// closeFailedHandshake closes the carrier on a failed handshake. The close
298// error is deliberately dropped: the handshake result is the caller's signal,
299// and a carrier that refuses to close must not replace or mask it.
300func closeFailedHandshake(carrier FramedCarrier) {
301 _ = carrier.Close()
302}