1package sessionwire
2
3// Preamble state machine (P3.1, wired at the P3.3 cutover).
4//
5// The first client frame is always PreambleRequest; the first server frame
6// is always PreambleResponse. Exact version equality is mandatory. The
7// preamble negotiates immutable per-connection ceilings once;
8// Hello.MaxOutputInFlight stays attachment-scoped and renegotiable on a
9// reused stream. One absolute deadline covers authentication, preamble,
10// Hello/Welcome, and first committed publication; layers receive the same
11// deadline and never restart it.
12
13import (
14 "context"
15 "encoding/binary"
16 "errors"
17 "io"
18
19 "google.golang.org/protobuf/proto"
20
21 "github.com/bnema/vev/internal/protocol"
22 "github.com/bnema/vev/internal/protocol/wire"
23)
24
25var (
26 ErrPreambleRejected = errors.New("sessionwire: preamble rejected")
27 ErrPreambleTimeout = errors.New("sessionwire: preamble deadline exceeded")
28 ErrPreambleOrder = errors.New("sessionwire: preamble out of order")
29)
30
31// protoCeilings are the immutable per-connection ceilings negotiated once
32// by the preamble.
33type protoCeilings struct {
34 maxReceiveEnvelopeBytes uint64
35 outputDataLimit uint64
36}
37
38// defaultProtoCeilings advertises the local receive policy.
39func defaultProtoCeilings() protoCeilings {
40 return protoCeilings{
41 maxReceiveEnvelopeBytes: wire.AbsoluteEnvelopeLimit,
42 outputDataLimit: uint64(protocol.MaxOutputDataLen),
43 }
44}
45
46// effectiveCeilings takes the minima of both sides' advertisements.
47func effectiveCeilings(local, remote protoCeilings) protoCeilings {
48 return protoCeilings{
49 maxReceiveEnvelopeBytes: min(local.maxReceiveEnvelopeBytes, remote.maxReceiveEnvelopeBytes),
50 outputDataLimit: min(local.outputDataLimit, remote.outputDataLimit),
51 }
52}
53
54func preambleRequestToWire(ceilings protoCeilings) *wire.PreambleRequest {
55 return &wire.PreambleRequest{
56 Magic: wire.PreambleMagic,
57 Epoch: wire.ProtocolEpoch,
58 Version: uint32(protocol.Version),
59 Role: &wire.PreambleRole{Role: 1},
60 MaxReceiveEnvelopeBytes: ceilings.maxReceiveEnvelopeBytes,
61 OutputDataLimit: ceilings.outputDataLimit,
62 }
63}
64
65func checkPreambleRequest(message *wire.PreambleRequest) (protoCeilings, error) {
66 if message == nil {
67 return protoCeilings{}, ErrPreambleRejected
68 }
69 if message.GetMagic() != wire.PreambleMagic {
70 return protoCeilings{}, ErrPreambleRejected
71 }
72 if message.GetEpoch() != wire.ProtocolEpoch {
73 return protoCeilings{}, ErrPreambleRejected
74 }
75 if message.GetVersion() != uint32(protocol.Version) {
76 return protoCeilings{}, ErrPreambleRejected
77 }
78 if message.GetRole().GetRole() != 1 {
79 return protoCeilings{}, ErrPreambleRejected
80 }
81 remote := protoCeilings{
82 maxReceiveEnvelopeBytes: message.GetMaxReceiveEnvelopeBytes(),
83 outputDataLimit: message.GetOutputDataLimit(),
84 }
85 if remote.maxReceiveEnvelopeBytes == 0 || remote.maxReceiveEnvelopeBytes > wire.AbsoluteEnvelopeLimit {
86 return protoCeilings{}, ErrPreambleRejected
87 }
88 if remote.outputDataLimit == 0 || remote.outputDataLimit > uint64(protocol.MaxOutputDataLen) {
89 return protoCeilings{}, ErrPreambleRejected
90 }
91 return remote, nil
92}
93
94func preambleResponseToWire(accepted bool, ceilings protoCeilings, code uint32) *wire.PreambleResponse {
95 response := &wire.PreambleResponse{
96 Magic: wire.PreambleMagic,
97 Epoch: wire.ProtocolEpoch,
98 Version: uint32(protocol.Version),
99 Role: &wire.PreambleRole{Role: 2},
100 MaxReceiveEnvelopeBytes: ceilings.maxReceiveEnvelopeBytes,
101 OutputDataLimit: ceilings.outputDataLimit,
102 Accepted: accepted,
103 }
104 if !accepted {
105 response.Rejection = &wire.PreambleRejectionCode{Code: code}
106 }
107 return response
108}
109
110func checkPreambleResponse(message *wire.PreambleResponse) (protoCeilings, error) {
111 if message == nil {
112 return protoCeilings{}, ErrPreambleRejected
113 }
114 if message.GetMagic() != wire.PreambleMagic {
115 return protoCeilings{}, ErrPreambleRejected
116 }
117 if message.GetEpoch() != wire.ProtocolEpoch {
118 return protoCeilings{}, ErrPreambleRejected
119 }
120 if message.GetVersion() != uint32(protocol.Version) {
121 return protoCeilings{}, ErrPreambleRejected
122 }
123 if message.GetRole().GetRole() != 2 {
124 return protoCeilings{}, ErrPreambleRejected
125 }
126 if !message.GetAccepted() {
127 return protoCeilings{}, ErrPreambleRejected
128 }
129 remote := protoCeilings{
130 maxReceiveEnvelopeBytes: message.GetMaxReceiveEnvelopeBytes(),
131 outputDataLimit: message.GetOutputDataLimit(),
132 }
133 if remote.maxReceiveEnvelopeBytes == 0 || remote.maxReceiveEnvelopeBytes > wire.AbsoluteEnvelopeLimit {
134 return protoCeilings{}, ErrPreambleRejected
135 }
136 if remote.outputDataLimit == 0 || remote.outputDataLimit > uint64(protocol.MaxOutputDataLen) {
137 return protoCeilings{}, ErrPreambleRejected
138 }
139 return remote, nil
140}
141
142// readPreambleFrame reads one length-delimited preamble envelope, enforcing
143// the preamble ceiling before allocation. wire.ScanEnvelope runs first so
144// unknown, repeated, concatenated, or trailing fields are rejected before
145// generated unmarshal; magic/epoch/version/role are checked afterwards.
146func readPreambleFrame(reader io.Reader, message proto.Message) error {
147 var header [4]byte
148 if _, err := io.ReadFull(reader, header[:]); err != nil {
149 return err
150 }
151 length := binary.BigEndian.Uint32(header[:])
152 if length == 0 || length > wire.PreambleLimit {
153 return wire.ErrScanLength
154 }
155 raw := make([]byte, length)
156 if _, err := io.ReadFull(reader, raw); err != nil {
157 return err
158 }
159 if err := wire.ScanEnvelope(message, raw); err != nil {
160 return err
161 }
162 return proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(raw, message)
163}
164
165// runProtoClientPreamble performs the client side over one raw envelope
166// transport: send PreambleRequest first, then accept exactly one
167// PreambleResponse. The caller owns the absolute deadline via ctx; it is
168// never restarted here.
169func runProtoClientPreamble(ctx context.Context, transport wire.Transport, ceilings protoCeilings) (protoCeilings, error) {
170 type result struct {
171 ceilings protoCeilings
172 err error
173 }
174 done := make(chan result, 1)
175 go func() {
176 raw, err := (proto.Marshal(preambleRequestToWire(ceilings)))
177 if err != nil {
178 done <- result{err: err}
179 return
180 }
181 if err := transport.Send(wire.Envelope{Payload: raw}); err != nil {
182 done <- result{err: err}
183 return
184 }
185 envelope, err := transport.Recv()
186 if err != nil {
187 done <- result{err: err}
188 return
189 }
190 if uint64(len(envelope.Payload)) > wire.PreambleLimit {
191 done <- result{err: wire.ErrScanLength}
192 return
193 }
194 var response wire.PreambleResponse
195 if err := wire.ScanEnvelope(&response, envelope.Payload); err != nil {
196 done <- result{err: err}
197 return
198 }
199 if err := (proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(envelope.Payload, &response)); err != nil {
200 done <- result{err: err}
201 return
202 }
203 remote, err := checkPreambleResponse(&response)
204 if err != nil {
205 done <- result{err: err}
206 return
207 }
208 done <- result{ceilings: effectiveCeilings(ceilings, remote)}
209 }()
210 select {
211 case <-ctx.Done():
212 return protoCeilings{}, errors.Join(ErrPreambleTimeout, ctx.Err())
213 case outcome := <-done:
214 return outcome.ceilings, outcome.err
215 }
216}
217
218// runProtoServerPreamble performs the server side: the first envelope must
219// be a valid PreambleRequest, answered with acceptance or typed refusal.
220func runProtoServerPreamble(ctx context.Context, transport wire.Transport, ceilings protoCeilings) (protoCeilings, error) {
221 type result struct {
222 ceilings protoCeilings
223 err error
224 }
225 done := make(chan result, 1)
226 go func() {
227 envelope, err := transport.Recv()
228 if err != nil {
229 done <- result{err: err}
230 return
231 }
232 if uint64(len(envelope.Payload)) > wire.PreambleLimit {
233 _ = sendPreambleResponse(transport, preambleResponseToWire(false, ceilings, 7))
234 done <- result{err: wire.ErrScanLength}
235 return
236 }
237 var request wire.PreambleRequest
238 if err := wire.ScanEnvelope(&request, envelope.Payload); err != nil {
239 _ = sendPreambleResponse(transport, preambleResponseToWire(false, ceilings, rejectionCodeFor(err)))
240 done <- result{err: err}
241 return
242 }
243 if err := (proto.UnmarshalOptions{DiscardUnknown: false}.Unmarshal(envelope.Payload, &request)); err != nil {
244 _ = sendPreambleResponse(transport, preambleResponseToWire(false, ceilings, rejectionCodeFor(err)))
245 done <- result{err: err}
246 return
247 }
248 remote, err := checkPreambleRequest(&request)
249 if err != nil {
250 _ = sendPreambleResponse(transport, preambleResponseToWire(false, ceilings, rejectionCodeFor(err)))
251 done <- result{err: err}
252 return
253 }
254 effective := effectiveCeilings(ceilings, remote)
255 if err := sendPreambleResponse(transport, preambleResponseToWire(true, effective, 0)); err != nil {
256 done <- result{err: err}
257 return
258 }
259 done <- result{ceilings: effective}
260 }()
261 select {
262 case <-ctx.Done():
263 return protoCeilings{}, errors.Join(ErrPreambleTimeout, ctx.Err())
264 case outcome := <-done:
265 return outcome.ceilings, outcome.err
266 }
267}
268
269func sendPreambleResponse(transport wire.Transport, response *wire.PreambleResponse) error {
270 raw, err := (proto.Marshal(response))
271 if err != nil {
272 return err
273 }
274 return transport.Send(wire.Envelope{Payload: raw})
275}
276
277func rejectionCodeFor(err error) uint32 {
278 if errors.Is(err, ErrPreambleRejected) {
279 return 3 // version mismatch is the common typed refusal
280 }
281 return 7
282}