-
Notifications
You must be signed in to change notification settings - Fork 182
/
frame.go
443 lines (402 loc) · 9.44 KB
/
frame.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
package libproxy
import (
"encoding/binary"
"errors"
"fmt"
"io"
"net"
)
// Proto is the protocol of the flow
type Proto uint8
const (
// TCP flow
TCP Proto = 1
// UDP flow
UDP Proto = 2
// Unix domain socket flow
Unix Proto = 3
)
// Destination refers to a listening TCP or UDP service
type Destination struct {
Proto Proto
IP net.IP
Port uint16
Path string
}
func (d Destination) String() string {
switch d.Proto {
case TCP:
return fmt.Sprintf("TCP:%s:%d", d.IP.String(), d.Port)
case UDP:
return fmt.Sprintf("UDP:%s:%d", d.IP.String(), d.Port)
case Unix:
return fmt.Sprintf("Unix:%s", d.Path)
}
return "Unknown"
}
// Read header which describes TCP/UDP and destination IP:port
func unmarshalDestination(r io.Reader) (Destination, error) {
d := Destination{}
if err := binary.Read(r, binary.LittleEndian, &d.Proto); err != nil {
return d, err
}
switch d.Proto {
case TCP, UDP:
var length uint16
// IP length
if err := binary.Read(r, binary.LittleEndian, &length); err != nil {
return d, err
}
d.IP = make([]byte, length)
if err := binary.Read(r, binary.LittleEndian, &d.IP); err != nil {
return d, err
}
if err := binary.Read(r, binary.LittleEndian, &d.Port); err != nil {
return d, err
}
case Unix:
var length uint16
// String length
if err := binary.Read(r, binary.LittleEndian, &length); err != nil {
return d, err
}
path := make([]byte, length)
if err := binary.Read(r, binary.LittleEndian, &path); err != nil {
return d, err
}
d.Path = string(path)
}
return d, nil
}
func (d Destination) Write(w io.Writer) error {
if err := binary.Write(w, binary.LittleEndian, d.Proto); err != nil {
return err
}
switch d.Proto {
case TCP, UDP:
b := []byte(d.IP)
length := uint16(len(b))
if err := binary.Write(w, binary.LittleEndian, length); err != nil {
return err
}
if err := binary.Write(w, binary.LittleEndian, b); err != nil {
return err
}
if err := binary.Write(w, binary.LittleEndian, d.Port); err != nil {
return err
}
case Unix:
b := []byte(d.Path)
length := uint16(len(b))
if err := binary.Write(w, binary.LittleEndian, length); err != nil {
return err
}
if err := binary.Write(w, binary.LittleEndian, b); err != nil {
return err
}
}
return nil
}
// Size returns the marshalled size in bytes
func (d Destination) Size() int {
switch d.Proto {
case TCP, UDP:
return 1 + 2 + len(d.IP) + 2
case Unix:
return 1 + 2 + len(d.Path)
}
return 0
}
// Connection indicates whether the connection will use multiplexing or not.
type Connection int8
func (c Connection) String() string {
switch c {
case Dedicated:
return "Dedicated"
case Multiplexed:
return "Multiplexed"
default:
return "Unknown"
}
}
const (
// Dedicated means this connection will not use multiplexing
Dedicated Connection = iota + 1
// Multiplexed means this connection will contain labelled sub-connections mixed together
Multiplexed
)
// OpenFrame requests to connect to a proxy backend
type OpenFrame struct {
Connection Connection // Connection describes whether the opened connection should be dedicated or multiplexed
Destination Destination
}
func unmarshalOpen(r io.Reader) (*OpenFrame, error) {
o := &OpenFrame{}
if err := binary.Read(r, binary.LittleEndian, &o.Connection); err != nil {
return nil, err
}
d, err := unmarshalDestination(r)
if err != nil {
return nil, err
}
o.Destination = d
return o, nil
}
func (o *OpenFrame) Write(w io.Writer) error {
if err := binary.Write(w, binary.LittleEndian, o.Connection); err != nil {
return err
}
return o.Destination.Write(w)
}
// Size returns the marshalled size of the Open message
func (o *OpenFrame) Size() int {
return 1 + o.Destination.Size()
}
// CloseFrame requests to disconnect from a proxy backend
type CloseFrame struct {
}
// ShutdownFrame requests to close the write channel to a proxy backend
type ShutdownFrame struct {
}
// DataFrame is the header of a frame containing user data
type DataFrame struct {
payloadlen uint32
}
func unmarshalData(r io.Reader) (*DataFrame, error) {
d := &DataFrame{}
err := binary.Read(r, binary.LittleEndian, &d.payloadlen)
return d, err
}
func (d *DataFrame) Write(w io.Writer) error {
return binary.Write(w, binary.LittleEndian, d.payloadlen)
}
// Size returns the marshalled size of the data payload header
func (d *DataFrame) Size() int {
return 4
}
// WindowFrame is a window advertisement message
type WindowFrame struct {
seq uint64
}
func unmarshalWindow(r io.Reader) (*WindowFrame, error) {
w := &WindowFrame{}
err := binary.Read(r, binary.LittleEndian, &w.seq)
return w, err
}
func (win *WindowFrame) Write(w io.Writer) error {
return binary.Write(w, binary.LittleEndian, win.seq)
}
// Size returned the marshalled size of the Window payload
func (win *WindowFrame) Size() int {
return 8
}
// Command is the action requested by a message.
type Command int8
const (
// Open requests to open a connection to a backend service.
Open Command = iota + 1
// Close requests and then acknowledges the close of a sub-connection
Close
// Shutdown indicates that no more data will be written in this direction
Shutdown
// Data is a payload of a connection/sub-connection
Data
// Window is permission to send and consume buffer space
Window
)
// Frame is the low-level message sent to the multiplexer
type Frame struct {
Command Command // Command is the action erquested
ID uint32 // Id of the sub-connection, managed by the client
open *OpenFrame
close *CloseFrame
shutdown *ShutdownFrame
window *WindowFrame
data *DataFrame
}
func unmarshalFrame(r io.Reader) (*Frame, error) {
f := &Frame{}
var totallen uint16
if err := binary.Read(r, binary.LittleEndian, &totallen); err != nil {
return nil, err
}
if err := binary.Read(r, binary.LittleEndian, &f.Command); err != nil {
return nil, err
}
if err := binary.Read(r, binary.LittleEndian, &f.ID); err != nil {
return nil, err
}
switch f.Command {
case Open:
o, err := unmarshalOpen(r)
if err != nil {
return nil, err
}
f.open = o
case Close:
// no payload
case Shutdown:
// no payload
case Window:
w, err := unmarshalWindow(r)
if err != nil {
return nil, err
}
f.window = w
case Data:
d, err := unmarshalData(r)
if err != nil {
return nil, err
}
f.data = d
}
return f, nil
}
func (f *Frame) Write(w io.Writer) error {
frameLen := uint16(f.Size())
if err := binary.Write(w, binary.LittleEndian, frameLen); err != nil {
return err
}
if err := binary.Write(w, binary.LittleEndian, f.Command); err != nil {
return err
}
if err := binary.Write(w, binary.LittleEndian, f.ID); err != nil {
return err
}
switch f.Command {
case Open:
if err := f.open.Write(w); err != nil {
return err
}
case Close:
// no payload
case Shutdown:
// no payload
case Window:
if err := f.window.Write(w); err != nil {
return err
}
case Data:
if err := f.data.Write(w); err != nil {
return err
}
}
return nil
}
// Size returns the marshalled size of the frame
func (f *Frame) Size() int {
// include 2 for the preceeding length field
len := 2 + 1 + 4
switch f.Command {
case Open:
len = len + f.open.Size()
case Close:
// no payload
case Shutdown:
// no payload
case Window:
len = len + f.window.Size()
case Data:
len = len + f.data.Size()
}
return len
}
func (f *Frame) String() string {
switch f.Command {
case Open:
return fmt.Sprintf("%d Open %s %s", f.ID, f.open.Connection.String(), f.open.Destination.String())
case Close:
return fmt.Sprintf("%d Close", f.ID)
case Shutdown:
return fmt.Sprintf("%d Shutdown", f.ID)
case Window:
return fmt.Sprintf("%d Window %d", f.ID, f.window.seq)
case Data:
return fmt.Sprintf("%d Data length %d", f.ID, f.data.payloadlen)
default:
return "unknown"
}
}
// Window returns the payload of the frame, if it has Command = Window.
func (f *Frame) Window() (*WindowFrame, error) {
if f.Command != Window {
return nil, errors.New("Frame is not a Window()")
}
return f.window, nil
}
// Open returns the payload of the frame, if it has Command = Open
func (f *Frame) Open() (*OpenFrame, error) {
if f.Command != Open {
return nil, errors.New("Frame is not an Open()")
}
return f.open, nil
}
// Data returns the payload of the frame, if it has Command = Data
func (f *Frame) Data() (*DataFrame, error) {
if f.Command != Data {
return nil, errors.New("Frame is not Data()")
}
return f.data, nil
}
// Payload returns the payload of the frame.
func (f *Frame) Payload() interface{} {
switch f.Command {
case Open:
return f.open
case Close:
return f.close
case Shutdown:
return f.shutdown
case Window:
return f.window
case Data:
return f.data
default:
return nil
}
}
// NewWindow creates a Window message
func NewWindow(ID uint32, seq uint64) *Frame {
return &Frame{
Command: Window,
ID: ID,
window: &WindowFrame{
seq: seq,
},
}
}
// NewOpen creates an open message
func NewOpen(ID uint32, d Destination) *Frame {
return &Frame{
Command: Open,
ID: ID,
open: &OpenFrame{
Connection: Multiplexed,
Destination: d,
},
}
}
// NewData creates a data header frame
func NewData(ID, payloadlen uint32) *Frame {
return &Frame{
Command: Data,
ID: ID,
data: &DataFrame{
payloadlen: payloadlen,
},
}
}
// NewShutdown creates a shutdown frame
func NewShutdown(ID uint32) *Frame {
return &Frame{
Command: Shutdown,
ID: ID,
}
}
// NewClose creates a close frame
func NewClose(ID uint32) *Frame {
return &Frame{
Command: Close,
ID: ID,
}
}