ALL

Solo para leer y verificar. No puedes copiar, usar, modificar ni redistribuir este código. Aviso completo en la página principal.

tunnel/proto.go

SHA-256: a5474c82760c8614c4d8f2d733bb4df7c3701c5c754dfe659333d5679d05ae95
426 líneas · 13502 bytes · Ver en GitHub

1package main
2
3import (
4 "bufio"
5 "errors"
6 "net/netip"
7 "strconv"
8 "strings"
9 "time"
10)
11
12// Protocol v2 on the wire (the contract, section 2, contracts/schemas/wire-v2.grammar.json). A line is ASCII 0x20-0x7e ended by LF, fields
13// are separated by exactly one space and none is empty. Two kinds of connection are told apart by the first line: a control connection (HELLO) and a
14// data connection (ATTACH, one per client stream).
15
16// Error words of ERR lines (policy protocol.error_codes) and the words of the node's own decisions.
17const (
18 codeBadHello = "bad_hello"
19 codeBadCommand = "bad_command"
20 codeUnsupportedVersion = "unsupported_version"
21 codeBadProof = "bad_proof"
22 codeAuth = "auth"
23 codeClockSkew = "clock_skew"
24 codeUnknownStream = "unknown_stream"
25 codeBadAttach = "bad_attach"
26 codeLicenseInactive = "license_inactive"
27 codeDeviceRevoked = "device_revoked"
28 codePaused = "paused"
29 codeBusy = "busy"
30 codeControlUnavailable = "control_unavailable"
31 codeLineTooLong = "line_too_long"
32)
33
34// Reasons of a CLOSE line (policy protocol.close_reasons) the node sends.
35const (
36 closeReplaced = "replaced"
37 closeRevoked = "revoked"
38 closeExpired = "expired"
39 closeIdle = "idle"
40 closeProtocol = "protocol"
41 closeShutdown = "shutdown"
42 closeReleased = "released"
43 closeKilled = "killed"
44 closeDrain = "drain"
45 closePaused = "paused"
46)
47
48// Classes of the source of a client stream (policy protocol.source_classes).
49const (
50 classOpenAI = "openai"
51 classHint = "hint"
52 classAnon = "anon"
53)
54
55// Direction of a line: d2n is device to node, n2d is node to device, both is CLOSE.
56const (
57 dirD2N = "d2n"
58 dirN2D = "n2d"
59 dirBoth = "both"
60)
61
62// protoError is a refused line: code is the word of the ERR the node answers with.
63type protoError struct{ code string }
64
65func (e *protoError) Error() string { return e.code }
66
67func protoFail(code string) error { return &protoError{code} }
68
69// errCode is the ERR word of an error: its own when it is a protoError, line_too_long for an over-long line, bad_command otherwise.
70func errCode(err error) string {
71 var pe *protoError
72 if errors.As(err, &pe) {
73 return pe.code
74 }
75 if errors.Is(err, bufio.ErrBufferFull) {
76 return codeLineTooLong
77 }
78 return codeBadCommand
79}
80
81// message is one parsed line: {"verb": ..., <field>: ...}. HINT gives "addresses" ([]any of strings) and OK gives "body".
82type message map[string]any
83
84func (m message) str(key string) string { s, _ := m[key].(string); return s }
85
86type fieldSpec struct {
87 name, kind string
88 err string // ERR word when the field is wrong (default bad_command)
89}
90
91type verbSpec struct {
92 dir string
93 max int // longest line, LF included; 0 = protocol.line_max_bytes
94 fields []fieldSpec
95 optional []fieldSpec
96 repeat *fieldSpec // HINT: one to hint_max_addresses of the same kind
97 json bool // OK: the rest of the line is one canonical JSON object
98}
99
100// verbs is wire-v2.grammar.json, written out. protocol_test.go runs every vector of contracts/vectors/lines.json through it.
101var verbs = map[string]verbSpec{
102 "HELLO": {dir: dirD2N, fields: []fieldSpec{{"version", "version", ""}, {"session", "jwt", codeBadHello}}},
103 "PROOF": {dir: dirD2N, fields: []fieldSpec{{"proof", "jwt", codeBadProof}}},
104 "RENEW": {dir: dirD2N, fields: []fieldSpec{{"session", "jwt", codeBadHello}}},
105 "PING": {dir: dirD2N},
106 "HINT": {dir: dirD2N, repeat: &fieldSpec{"address", "ip", ""}},
107 "CLOSE": {dir: dirBoth, optional: []fieldSpec{{"reason", "close_reason", ""}}},
108 "ATTACH": {dir: dirD2N, fields: []fieldSpec{
109 {"stream_id", "stream_id", codeBadAttach},
110 {"secret", "attach_secret", codeBadAttach},
111 }},
112 "CHALLENGE": {dir: dirN2D, fields: []fieldSpec{{"nonce", "nonce", ""}}},
113 "OK": {dir: dirN2D, json: true},
114 "ERR": {dir: dirN2D, fields: []fieldSpec{{"code", "error_code", ""}}},
115 "OPEN": {dir: dirN2D, max: policyData.Protocol.OpenLineMaxBytes, fields: []fieldSpec{
116 {"stream_id", "stream_id", ""},
117 {"secret", "attach_secret", ""},
118 {"peer_ip", "ip", ""},
119 {"class", "source_class", ""},
120 }},
121 "PONG": {dir: dirN2D},
122 "STATEMENT": {dir: dirN2D, max: policyData.Protocol.StatementLineMaxBytes, fields: []fieldSpec{{"statement", "statement", ""}}},
123 "DRAIN": {dir: dirN2D, fields: []fieldSpec{{"quiet_s", "seconds", ""}}},
124}
125
126// isJWSShape is the textual shape of a compact JWS: three non-empty base64url segments.
127func isJWSShape(s string) bool {
128 dots := 0
129 segment := 0
130 for i := 0; i < len(s); i++ {
131 c := s[i]
132 switch {
133 case c == '.':
134 if segment == 0 {
135 return false
136 }
137 dots++
138 segment = 0
139 case c >= 'A' && c <= 'Z', c >= 'a' && c <= 'z', c >= '0' && c <= '9', c == '-', c == '_':
140 segment++
141 default:
142 return false
143 }
144 }
145 return dots == 2 && segment > 0
146}
147
148// validIP is the contracts' address: IPv4 dotted quad without leading zeros or RFC 4291 text in lower case, no zone, no brackets.
149func validIP(s string) bool {
150 if s == "" || s != strings.ToLower(s) || strings.ContainsAny(s, "%[] \t\r\n/") {
151 return false
152 }
153 _, err := netip.ParseAddr(s)
154 return err == nil
155}
156
157func fieldOK(kind, token string) bool {
158 switch kind {
159 case "version":
160 return len(token) >= 1 && len(token) <= 3 && allDigits(token)
161 case "jwt":
162 return len(token) <= policyData.Protocol.JWTMaxBytes && isJWSShape(token)
163 case "statement":
164 return isJWSShape(token)
165 case "nonce":
166 return reNonce.MatchString(token)
167 case "stream_id":
168 return reStreamID.MatchString(token)
169 case "attach_secret":
170 return reSecret.MatchString(token)
171 case "ip":
172 return validIP(token)
173 case "source_class":
174 return inList(policyData.Protocol.SourceClasses, token)
175 case "close_reason":
176 return inList(policyData.Protocol.CloseReasons, token)
177 case "error_code":
178 return inList(policyData.Protocol.ErrorCodes, token)
179 case "seconds":
180 return token == "0" || (len(token) <= 6 && token[0] != '0' && allDigits(token))
181 }
182 return false
183}
184
185func allDigits(s string) bool {
186 for i := 0; i < len(s); i++ {
187 if s[i] < '0' || s[i] > '9' {
188 return false
189 }
190 }
191 return s != ""
192}
193
194// parseLine parses one protocol line (the text before the LF) travelling in dir ("d2n" or "n2d"). The error is a *protoError whose code is the word
195// the node answers with.
196func parseLine(line, dir string) (message, error) {
197 for i := 0; i < len(line); i++ {
198 if line[i] < 0x20 || line[i] > 0x7e {
199 return nil, protoFail(codeBadCommand)
200 }
201 }
202 verb, rest, hasSpace := strings.Cut(line, " ")
203 spec, known := verbs[verb]
204 limit := policyData.Protocol.LineMaxBytes
205 if known && spec.max > 0 {
206 limit = spec.max
207 }
208 if len(line)+1 > limit {
209 return nil, protoFail(codeLineTooLong)
210 }
211 if !known || (spec.dir != dir && spec.dir != dirBoth) || line == "" || (hasSpace && rest == "") {
212 return nil, protoFail(codeBadCommand)
213 }
214 out := message{"verb": verb}
215 if spec.json {
216 if !hasSpace {
217 return nil, protoFail(codeBadCommand)
218 }
219 body, err := parseJSON(rest)
220 if err != nil || checkOKBody(body) != nil {
221 return nil, protoFail(codeBadCommand)
222 }
223 out["body"] = body
224 return out, nil
225 }
226 var tokens []string
227 if hasSpace {
228 tokens = strings.Split(rest, " ")
229 }
230 for _, token := range tokens {
231 if token == "" {
232 return nil, protoFail(codeBadCommand)
233 }
234 }
235 if spec.repeat != nil {
236 if len(tokens) < 1 || len(tokens) > policyData.Protocol.HintMaxAddresses {
237 return nil, protoFail(codeBadCommand)
238 }
239 addresses := make([]any, len(tokens))
240 for i, token := range tokens {
241 if !fieldOK(spec.repeat.kind, token) {
242 return nil, protoFail(codeBadCommand)
243 }
244 addresses[i] = token
245 }
246 out["addresses"] = addresses
247 return out, nil
248 }
249 rules := append(append([]fieldSpec{}, spec.fields...), spec.optional...)
250 if len(tokens) < len(spec.fields) || len(tokens) > len(rules) {
251 return nil, protoFail(codeBadCommand)
252 }
253 for i, token := range tokens {
254 rule := rules[i]
255 if verb == "HELLO" && rule.name == "version" {
256 // the protocol version is 2: another number is unsupported_version, anything else is a malformed line
257 if token != strconv.Itoa(policyData.Protocol.Version) {
258 if fieldOK(rule.kind, token) {
259 return nil, protoFail(codeUnsupportedVersion)
260 }
261 return nil, protoFail(codeBadCommand)
262 }
263 } else if !fieldOK(rule.kind, token) {
264 code := rule.err
265 if code == "" {
266 code = codeBadCommand
267 }
268 return nil, protoFail(code)
269 }
270 out[rule.name] = token
271 }
272 return out, nil
273}
274
275// formatLine is the canonical line (without LF) of a message the node sends; it parses its own output, so a node can never write a line its
276// own grammar refuses.
277func formatLine(m message) string {
278 verb := m.str("verb")
279 spec := verbs[verb]
280 var line string
281 switch {
282 case spec.json:
283 line = verb + " " + canonicalJSON(m["body"])
284 case spec.repeat != nil:
285 parts := []string{verb}
286 for _, address := range m["addresses"].([]any) {
287 parts = append(parts, address.(string))
288 }
289 line = strings.Join(parts, " ")
290 default:
291 parts := []string{verb}
292 for _, rule := range append(append([]fieldSpec{}, spec.fields...), spec.optional...) {
293 if v, ok := m[rule.name]; ok {
294 parts = append(parts, v.(string))
295 }
296 }
297 line = strings.Join(parts, " ")
298 }
299 dir := spec.dir
300 if dir == dirBoth {
301 dir = dirD2N
302 }
303 if _, err := parseLine(line, dir); err != nil {
304 panic("tunnel: formatted a line its own grammar refuses: " + verb + ": " + err.Error())
305 }
306 return line
307}
308
309// statementLine is the STATEMENT the node forwards, or "" when the statement is not one the grammar allows (too long, not a JWS): it is then dropped,
310// never written, never a panic. The statement itself is the control plane's signed access statement; the node passes it on as it is.
311func statementLine(statement string) string {
312 line := "STATEMENT " + statement
313 if _, err := parseLine(line, dirN2D); err != nil {
314 return ""
315 }
316 return line
317}
318
319// drainLine tells a device the node is going down and how long it keeps its open streams.
320func drainLine(quiet time.Duration) string {
321 return formatLine(message{"verb": "DRAIN", "quiet_s": strconv.Itoa(int(quiet / time.Second))})
322}
323
324func errLine(code string) string { return formatLine(message{"verb": "ERR", "code": code}) }
325
326func closeLine(reason string) string {
327 if reason == "" {
328 return "CLOSE"
329 }
330 return formatLine(message{"verb": "CLOSE", "reason": reason})
331}
332
333// okBody is what an OK answers: the control plane's decision, relayed (the contract, section 2). Host, status, valid_until and
334// server_time are what the control plane answered; policy is that answer's windows, bounded by policy.json.
335type okBody struct {
336 Node string
337 Host string
338 Status string
339 ValidUntil int64
340 ServerTime int64
341 Policy okPolicy
342}
343
344type okPolicy struct {
345 PingIntervalS, DeadPeerS, StreamIdleS, StreamMaxAgeS, RenewBeforeValidUntilS int64
346}
347
348func (p okPolicy) toMap() map[string]any {
349 return map[string]any{"ping_interval_s": p.PingIntervalS, "dead_peer_s": p.DeadPeerS, "stream_idle_s": p.StreamIdleS,
350 "stream_max_age_s": p.StreamMaxAgeS, "renew_before_valid_until_s": p.RenewBeforeValidUntilS}
351}
352
353func (b okBody) line() string {
354 return formatLine(message{"verb": "OK", "body": map[string]any{"v": policyData.Protocol.Version, "node": b.Node, "host": b.Host, "status": b.Status,
355 "valid_until": b.ValidUntil, "server_time": b.ServerTime, "policy": b.Policy.toMap()}})
356}
357
358// epochOK is the contracts' epoch in seconds.
359func epochOK(v any) (int64, bool) {
360 f, ok := v.(float64)
361 if !ok || f != float64(int64(f)) || f < 1 || f > epochMax {
362 return 0, false
363 }
364 return int64(f), true
365}
366
367const epochMax = 4102444800 // 2100-01-01, common.schema.json epoch_s
368
369var okPolicyKeys = []string{"ping_interval_s", "dead_peer_s", "stream_idle_s", "stream_max_age_s", "renew_before_valid_until_s"}
370
371// checkPolicyBlock validates the five windows of an authorize answer or an OK (integers 1..86400: capped at a day only to refuse nonsense).
372func checkPolicyBlock(v any) (okPolicy, bool) {
373 m, ok := v.(map[string]any)
374 if !ok || len(m) != len(okPolicyKeys) {
375 return okPolicy{}, false
376 }
377 var got [5]int64
378 for i, key := range okPolicyKeys {
379 f, ok := m[key].(float64)
380 if !ok || f != float64(int64(f)) || f < 1 || f > 86400 {
381 return okPolicy{}, false
382 }
383 got[i] = int64(f)
384 }
385 return okPolicy{got[0], got[1], got[2], got[3], got[4]}, true
386}
387
388var reNodeID = anchored(`^[a-z0-9]{1,16}$`)
389
390// checkOKBody validates ok-body.schema.json.
391func checkOKBody(v any) error {
392 m, ok := v.(map[string]any)
393 if !ok || len(m) != 7 {
394 return errors.New("ok body shape")
395 }
396 if f, ok := m["v"].(float64); !ok || f != float64(policyData.Protocol.Version) {
397 return errors.New("ok body v")
398 }
399 if n, ok := m["node"].(string); !ok || !reNodeID.MatchString(n) {
400 return errors.New("ok body node")
401 }
402 if h, ok := m["host"].(string); !ok || !validFQDN(h) {
403 return errors.New("ok body host")
404 }
405 if s, ok := m["status"].(string); !ok || !validStatus(s) {
406 return errors.New("ok body status")
407 }
408 for _, key := range []string{"valid_until", "server_time"} {
409 if _, ok := epochOK(m[key]); !ok {
410 return errors.New("ok body " + key)
411 }
412 }
413 if _, ok := checkPolicyBlock(m["policy"]); !ok {
414 return errors.New("ok body policy")
415 }
416 return nil
417}
418
419// Statuses of an authorize answer (policy enums.authorize_status).
420const (
421 statusActive = "active"
422 statusGated = "gated"
423 statusPaused = "paused"
424)
425
426func validStatus(s string) bool { return s == statusActive || s == statusGated || s == statusPaused }