1package sessionwire
2
3import (
4 "bytes"
5 "compress/zlib"
6 "errors"
7 "io"
8 "math"
9 "sync"
10
11 "google.golang.org/protobuf/proto"
12
13 "github.com/bnema/vev/internal/domain"
14 "github.com/bnema/vev/internal/protocol"
15 "github.com/bnema/vev/internal/protocol/wire"
16)
17
18// outputCompressionThreshold preserves the small-snapshot compression threshold: small snapshots stay
19// on the canonical raw path.
20const protoOutputCompressionThreshold = 1024
21
22func helloToWire(message protocol.Hello) (*wire.Hello, error) {
23 cols, rows, pixelWidth, pixelHeight, err := geometryToWire(message.Size, message.PixelWidth, message.PixelHeight)
24 if err != nil {
25 return nil, err
26 }
27 sessionTarget, err := sessionAttachTargetToWire(message.SessionTarget)
28 if err != nil {
29 return nil, err
30 }
31 exact := exactTargetToWire(message.ExactTarget)
32 return &wire.Hello{
33 Version: uint32(message.Version),
34 Intent: uint32(message.Intent),
35 ClientId: append([]byte(nil), message.ClientID[:]...),
36 ResumeToken: message.ResumeToken,
37 Name: message.Name,
38 Cols: cols,
39 Rows: rows,
40 PixelWidth: pixelWidth,
41 PixelHeight: pixelHeight,
42 TermEnv: message.TermEnv,
43 Cwd: message.Cwd,
44 TrueColor: message.TrueColor,
45 MaxOutputInFlight: uint32(message.MaxOutputInFlight),
46 Env: append([]string(nil), message.Env...),
47 SessionTarget: sessionTarget,
48 EnvironmentPolicy: uint32(message.EnvironmentPolicy),
49 ExactTarget: exact,
50 PreferredTabId: string(message.PreferredTabID),
51 NavigationCapabilities: uint32(message.NavigationCapabilities),
52 Remote: message.Remote,
53 KittyDirectGraphics: message.KittyDirectGraphics,
54 KittyKeyboard: message.KittyKeyboard,
55 TerminalFocus: uint32(message.TerminalFocus),
56 ClientPid: message.ClientPID,
57 }, nil
58}
59
60func geometryToWire(size domain.Size, pixelWidth, pixelHeight int) (uint32, uint32, uint32, uint32, error) {
61 for _, value := range []int{size.Cols, size.Rows, pixelWidth, pixelHeight} {
62 if value < 0 || value > math.MaxUint16 {
63 return 0, 0, 0, 0, errProtoConvertRange
64 }
65 }
66 return uint32(size.Cols), uint32(size.Rows), uint32(pixelWidth), uint32(pixelHeight), nil
67}
68
69func helloFromWire(message *wire.Hello) (protocol.Hello, error) {
70 var hello protocol.Hello
71 if message == nil {
72 return hello, errProtoConvertRange
73 }
74 version, err := mustUint16(message.GetVersion())
75 if err != nil {
76 return protocol.Hello{}, err
77 }
78 hello.Version = version
79 intent, err := mustUint8(message.GetIntent())
80 if err != nil {
81 return protocol.Hello{}, err
82 }
83 hello.Intent = intent
84 if len(message.GetClientId()) != len(hello.ClientID) {
85 return protocol.Hello{}, errProtoConvertRange
86 }
87 copy(hello.ClientID[:], message.GetClientId())
88 hello.ResumeToken = message.GetResumeToken()
89 hello.Name = message.GetName()
90 cols, err := mustUint16(message.GetCols())
91 if err != nil {
92 return protocol.Hello{}, err
93 }
94 rows, err := mustUint16(message.GetRows())
95 if err != nil {
96 return protocol.Hello{}, err
97 }
98 pixelWidth, err := mustUint16(message.GetPixelWidth())
99 if err != nil {
100 return protocol.Hello{}, err
101 }
102 pixelHeight, err := mustUint16(message.GetPixelHeight())
103 if err != nil {
104 return protocol.Hello{}, err
105 }
106 hello.Size = domain.Size{Cols: int(cols), Rows: int(rows)}
107 hello.PixelWidth = int(pixelWidth)
108 hello.PixelHeight = int(pixelHeight)
109 hello.TermEnv = message.GetTermEnv()
110 hello.Cwd = message.GetCwd()
111 hello.TrueColor = message.GetTrueColor()
112 window, err := mustUint8(message.GetMaxOutputInFlight())
113 if err != nil {
114 return protocol.Hello{}, err
115 }
116 hello.MaxOutputInFlight = window
117 hello.Env = append([]string(nil), message.GetEnv()...)
118 sessionTarget, err := sessionAttachTargetFromWire(message.GetSessionTarget())
119 if err != nil {
120 return protocol.Hello{}, err
121 }
122 hello.SessionTarget = sessionTarget
123 hello.EnvironmentPolicy, err = enum8[protocol.EnvironmentPolicy](message.GetEnvironmentPolicy())
124 if err != nil {
125 return protocol.Hello{}, err
126 }
127 exact, err := exactTargetFromWire(message.GetExactTarget())
128 if err != nil {
129 return protocol.Hello{}, err
130 }
131 hello.ExactTarget = exact
132 hello.PreferredTabID = domain.TabStableID(message.GetPreferredTabId())
133 hello.NavigationCapabilities, err = enum8[protocol.NavigationCapabilities](message.GetNavigationCapabilities())
134 if err != nil {
135 return protocol.Hello{}, err
136 }
137 hello.Remote = message.GetRemote()
138 hello.KittyDirectGraphics = message.GetKittyDirectGraphics()
139 hello.KittyKeyboard = message.GetKittyKeyboard()
140 hello.ClientPID = message.GetClientPid()
141 hello.TerminalFocus, err = enum8[domain.TerminalFocus](message.GetTerminalFocus())
142 if err != nil {
143 return protocol.Hello{}, err
144 }
145 if err := protocol.ValidateHello(hello); err != nil {
146 return protocol.Hello{}, err
147 }
148 return hello, nil
149}
150
151func welcomeToWire(message protocol.Welcome) (*wire.Welcome, error) {
152 var identity *wire.CommittedRouteIdentity
153 if message.CommittedIdentity != nil {
154 converted, err := committedIdentityToWire(*message.CommittedIdentity)
155 if err != nil {
156 return nil, err
157 }
158 identity = converted
159 }
160 return &wire.Welcome{
161 SessionId: message.SessionID,
162 SessionName: message.SessionName,
163 Ephemeral: message.Ephemeral,
164 ResumeToken: message.ResumeToken,
165 Capabilities: message.Capabilities,
166 CommittedIdentity: identity,
167 }, nil
168}
169
170func welcomeFromWire(message *wire.Welcome) (protocol.Welcome, error) {
171 var welcome protocol.Welcome
172 if message == nil {
173 return welcome, errProtoConvertRange
174 }
175 welcome.SessionID = message.GetSessionId()
176 welcome.SessionName = message.GetSessionName()
177 welcome.Ephemeral = message.GetEphemeral()
178 welcome.ResumeToken = message.GetResumeToken()
179 welcome.Capabilities = message.GetCapabilities()
180 if message.GetCommittedIdentity() != nil {
181 identity, err := committedIdentityFromWire(message.GetCommittedIdentity())
182 if err != nil {
183 return protocol.Welcome{}, err
184 }
185 welcome.CommittedIdentity = &identity
186 if welcome.CommittedIdentity.Target.SessionName != welcome.SessionName || welcome.CommittedIdentity.Ephemeral != welcome.Ephemeral {
187 return protocol.Welcome{}, protocol.ErrInvalidRouteWire
188 }
189 }
190 return welcome, nil
191}
192
193func committedIdentityToWire(identity protocol.CommittedRouteIdentity) (*wire.CommittedRouteIdentity, error) {
194 if err := identity.Validate(); err != nil {
195 return nil, err
196 }
197 return &wire.CommittedRouteIdentity{Target: exactTargetToWire(&identity.Target), Ephemeral: identity.Ephemeral}, nil
198}
199
200func committedIdentityFromWire(message *wire.CommittedRouteIdentity) (protocol.CommittedRouteIdentity, error) {
201 var identity protocol.CommittedRouteIdentity
202 if message == nil {
203 return identity, protocol.ErrInvalidRouteWire
204 }
205 target, err := exactTargetFromWire(message.GetTarget())
206 if err != nil || target == nil {
207 return protocol.CommittedRouteIdentity{}, protocol.ErrInvalidRouteWire
208 }
209 identity.Target = *target
210 identity.Ephemeral = message.GetEphemeral()
211 if err := identity.Validate(); err != nil {
212 return protocol.CommittedRouteIdentity{}, err
213 }
214 return identity, nil
215}
216
217func sessionAttachTargetToWire(target *protocol.SessionAttachTarget) (*wire.SessionAttachTarget, error) {
218 if target == nil {
219 return nil, nil
220 }
221 if err := target.Validate(); err != nil {
222 return nil, err
223 }
224 message := &wire.SessionAttachTarget{SessionId: string(target.SessionID), LifecycleId: lifecycleToWire(target.LifecycleID), SessionName: target.SessionName, TabId: string(target.TabID), TabRawName: target.TabRawName, TabExpectedCount: uint32(target.TabExpectedCount), Stopped: target.Stopped}
225 if target.TabIndex != protocol.NoTabIndex {
226 message.TabIndex = proto.Int32(target.TabIndex)
227 }
228 return message, nil
229}
230
231func sessionAttachTargetFromWire(message *wire.SessionAttachTarget) (*protocol.SessionAttachTarget, error) {
232 if message == nil {
233 return nil, nil
234 }
235 if message.GetTabExpectedCount() > math.MaxUint16 {
236 return nil, protocol.ErrInvalidAttachTarget
237 }
238 lifecycle, err := lifecycleFromWire(message.GetLifecycleId())
239 if err != nil {
240 return nil, err
241 }
242 tabIndex := protocol.NoTabIndex
243 if message.TabIndex != nil {
244 tabIndex = message.GetTabIndex()
245 }
246 target := &protocol.SessionAttachTarget{SessionID: domain.SessionID(message.GetSessionId()), LifecycleID: lifecycle, SessionName: message.GetSessionName(), TabID: domain.TabStableID(message.GetTabId()), TabIndex: tabIndex, TabRawName: message.GetTabRawName(), TabExpectedCount: uint16(message.GetTabExpectedCount()), Stopped: message.GetStopped()}
247 if err := target.Validate(); err != nil {
248 return nil, err
249 }
250 return target, nil
251}
252
253func attachTargetToWire(message protocol.AttachTarget) (*wire.AttachTarget, error) {
254 if err := protocol.ValidateAttachTarget(message); err != nil {
255 return nil, err
256 }
257 sessionTarget, err := sessionAttachTargetToWire(message.SessionTarget)
258 if err != nil {
259 return nil, err
260 }
261 return &wire.AttachTarget{
262 RequestId: message.RequestID,
263 Endpoint: message.Endpoint,
264 Session: message.Session,
265 Intent: uint32(message.Intent),
266 SessionTarget: sessionTarget,
267 EnvironmentPolicy: uint32(message.EnvironmentPolicy),
268 ExactTarget: exactTargetToWire(message.ExactTarget),
269 SamePeer: message.SamePeer,
270 PreferredTabId: string(message.PreferredTabID),
271 CauseActionId: message.CauseActionID,
272 }, nil
273}
274
275func attachTargetFromWire(message *wire.AttachTarget) (protocol.AttachTarget, error) {
276 var target protocol.AttachTarget
277 if message == nil {
278 return target, protocol.ErrInvalidAttachTarget
279 }
280 target.RequestID = message.GetRequestId()
281 target.Endpoint = message.GetEndpoint()
282 target.Session = message.GetSession()
283 intent, err := mustUint8(message.GetIntent())
284 if err != nil {
285 return protocol.AttachTarget{}, protocol.ErrInvalidAttachTarget
286 }
287 target.Intent = intent
288 sessionTarget, err := sessionAttachTargetFromWire(message.GetSessionTarget())
289 if err != nil {
290 return protocol.AttachTarget{}, protocol.ErrInvalidAttachTarget
291 }
292 target.SessionTarget = sessionTarget
293 target.EnvironmentPolicy, err = enum8[protocol.EnvironmentPolicy](message.GetEnvironmentPolicy())
294 if err != nil {
295 return protocol.AttachTarget{}, protocol.ErrInvalidAttachTarget
296 }
297 exact, err := exactTargetFromWire(message.GetExactTarget())
298 if err != nil {
299 return protocol.AttachTarget{}, protocol.ErrInvalidAttachTarget
300 }
301 target.ExactTarget = exact
302 target.SamePeer = message.GetSamePeer()
303 target.PreferredTabID = domain.TabStableID(message.GetPreferredTabId())
304 target.CauseActionID = message.GetCauseActionId()
305 if err := protocol.ValidateAttachTarget(target); err != nil {
306 return protocol.AttachTarget{}, err
307 }
308 return target, nil
309}
310
311func outputToWire(message protocol.Output) (*wire.Output, error) {
312 if err := protocol.ValidateOutput(message); err != nil {
313 return nil, err
314 }
315 cols, rows, _, _, err := geometryToWire(message.Size, 0, 0)
316 if err != nil {
317 return nil, err
318 }
319 var context *wire.ViewContext
320 if message.Context != nil {
321 converted, err := viewContextToWire(*message.Context)
322 if err != nil {
323 return nil, err
324 }
325 context = converted
326 }
327 encoding, data, err := compressProtoOutput(message)
328 if err != nil {
329 return nil, err
330 }
331 return &wire.Output{
332 Epoch: message.Epoch,
333 Base: message.Base,
334 NewState: message.New,
335 Echo: message.Echo,
336 ViewRevision: message.ViewRevision,
337 Cols: cols,
338 Rows: rows,
339 Full: message.Full,
340 Context: context,
341 Encoding: encoding,
342 UncompressedLength: uint64(len(message.Data)),
343 Data: data,
344 }, nil
345}
346
347var protoOutputCompressorPool = sync.Pool{New: func() any {
348 writer, err := zlib.NewWriterLevel(io.Discard, zlib.BestSpeed)
349 if err != nil {
350 panic(err)
351 }
352 return writer
353}}
354
355// compressProtoOutput keeps the frozen policy: only full snapshots at or
356// above the threshold are zlib candidates, retained only when smaller.
357// Callers (use cases) always see uncompressed terminal data.
358func compressProtoOutput(message protocol.Output) (uint32, []byte, error) {
359 if !message.Full || len(message.Data) < protoOutputCompressionThreshold {
360 return 0, append([]byte(nil), message.Data...), nil
361 }
362 var compressed bytes.Buffer
363 writer := protoOutputCompressorPool.Get().(*zlib.Writer)
364 writer.Reset(&compressed)
365 defer func() {
366 writer.Reset(io.Discard)
367 protoOutputCompressorPool.Put(writer)
368 }()
369 if _, err := writer.Write(message.Data); err != nil {
370 return 0, nil, err
371 }
372 if err := writer.Close(); err != nil {
373 return 0, nil, err
374 }
375 if compressed.Len()+5 >= len(message.Data) {
376 return 0, append([]byte(nil), message.Data...), nil
377 }
378 return 1, compressed.Bytes(), nil
379}
380
381func outputFromWire(message *wire.Output) (protocol.Output, error) {
382 var output protocol.Output
383 if message == nil {
384 return output, protocol.ErrInvalidOutput
385 }
386 output.Epoch = message.GetEpoch()
387 output.Base = message.GetBase()
388 output.New = message.GetNewState()
389 output.Echo = message.GetEcho()
390 output.ViewRevision = message.GetViewRevision()
391 cols, err := mustUint16(message.GetCols())
392 if err != nil {
393 return protocol.Output{}, protocol.ErrInvalidOutput
394 }
395 rows, err := mustUint16(message.GetRows())
396 if err != nil {
397 return protocol.Output{}, protocol.ErrInvalidOutput
398 }
399 output.Size = domain.Size{Cols: int(cols), Rows: int(rows)}
400 output.Full = message.GetFull()
401 if message.GetContext() != nil {
402 context, err := viewContextFromWire(message.GetContext())
403 if err != nil {
404 return protocol.Output{}, protocol.ErrInvalidOutput
405 }
406 output.Context = &context
407 }
408 if err := protocol.ValidateOutput(output); err != nil {
409 return protocol.Output{}, err
410 }
411 decodedLen := message.GetUncompressedLength()
412 if decodedLen > uint64(protocol.MaxOutputDataLen) {
413 return protocol.Output{}, protocol.ErrInvalidOutput
414 }
415 switch message.GetEncoding() {
416 case 0:
417 if uint64(len(message.GetData())) != decodedLen {
418 return protocol.Output{}, protocol.ErrInvalidOutput
419 }
420 output.Data = append([]byte(nil), message.GetData()...)
421 case 1:
422 if !output.Full {
423 return protocol.Output{}, protocol.ErrInvalidOutput
424 }
425 decoded, err := decompressProtoOutput(message.GetData(), int(decodedLen))
426 if err != nil {
427 return protocol.Output{}, protocol.ErrInvalidOutput
428 }
429 output.Data = decoded
430 default:
431 return protocol.Output{}, protocol.ErrInvalidOutput
432 }
433 if err := protocol.ValidateOutput(output); err != nil {
434 return protocol.Output{}, err
435 }
436 return output, nil
437}
438
439func decompressProtoOutput(data []byte, decodedLen int) ([]byte, error) {
440 source := bytes.NewReader(data)
441 reader, err := zlib.NewReader(source)
442 if err != nil {
443 return nil, err
444 }
445 decoded := make([]byte, decodedLen)
446 if _, err := io.ReadFull(reader, decoded); err != nil {
447 _ = reader.Close()
448 return nil, err
449 }
450 var extra [1]byte
451 if n, err := reader.Read(extra[:]); n != 0 || !errors.Is(err, io.EOF) {
452 _ = reader.Close()
453 if err == nil || errors.Is(err, io.EOF) {
454 return nil, errors.New("sessionwire: compressed output exceeds declared length")
455 }
456 return nil, err
457 }
458 if err := reader.Close(); err != nil {
459 return nil, err
460 }
461 // The zlib reader stops at the stream end without consuming
462 // appended bytes, so trailing garbage must be rejected explicitly.
463 if source.Len() != 0 {
464 return nil, errors.New("sessionwire: compressed output has trailing bytes")
465 }
466 return decoded, nil
467}
468
469func viewContextToWire(context protocol.ViewContext) (*wire.ViewContext, error) {
470 if err := context.Validate(); err != nil {
471 return nil, err
472 }
473 route, err := committedIdentityToWire(context.Route)
474 if err != nil {
475 return nil, err
476 }
477 return &wire.ViewContext{
478 Publication: context.Publication,
479 Route: route,
480 TabId: string(context.TabID),
481 FocusedPaneId: string(context.FocusedPaneID),
482 }, nil
483}
484
485func viewContextFromWire(message *wire.ViewContext) (protocol.ViewContext, error) {
486 var context protocol.ViewContext
487 if message == nil {
488 return context, protocol.ErrInvalidOutput
489 }
490 context.Publication = message.GetPublication()
491 route, err := committedIdentityFromWire(message.GetRoute())
492 if err != nil {
493 return protocol.ViewContext{}, err
494 }
495 context.Route = route
496 context.TabID = domain.TabStableID(message.GetTabId())
497 context.FocusedPaneID = domain.PaneStableID(message.GetFocusedPaneId())
498 if err := context.Validate(); err != nil {
499 return protocol.ViewContext{}, err
500 }
501 return context, nil
502}
503
504func commandRequestToWire(message protocol.CommandRequest) (*wire.CommandRequest, error) {
505 if len(message.Args) > math.MaxUint16 {
506 return nil, errProtoConvertRange
507 }
508 return &wire.CommandRequest{
509 Version: uint32(message.Version),
510 RequestId: message.RequestID,
511 Attached: message.Attached,
512 Self: message.Self,
513 Slug: message.Slug,
514 Args: append([]string(nil), message.Args...),
515 TargetSession: message.TargetSession,
516 TargetTab: message.TargetTab,
517 TargetPane: message.TargetPane,
518 Json: message.JSON,
519 }, nil
520}
521
522func commandRequestFromWire(message *wire.CommandRequest) (protocol.CommandRequest, error) {
523 var request protocol.CommandRequest
524 if message == nil {
525 return request, errProtoConvertRange
526 }
527 version, err := mustUint16(message.GetVersion())
528 if err != nil {
529 return protocol.CommandRequest{}, err
530 }
531 request.Version = version
532 request.RequestID = message.GetRequestId()
533 request.Attached = message.GetAttached()
534 request.Self = message.GetSelf()
535 request.Slug = message.GetSlug()
536 request.Args = append([]string(nil), message.GetArgs()...)
537 request.TargetSession = message.GetTargetSession()
538 request.TargetTab = message.GetTargetTab()
539 request.TargetPane = message.GetTargetPane()
540 request.JSON = message.GetJson()
541 return request, nil
542}