ALL

For reading and verification only. You may not copy, use, modify or redistribute this code. Full notice on the main page.

tunnel/controlplane.go

SHA-256: 71e410e4670e3c4b11e2cefa7c84ab5edb9faf2f7cbe0514e6bc6f67741a32a4
496 lines · 19614 bytes · View on GitHub

1package main
2
3import (
4 "bytes"
5 "context"
6 "crypto/ed25519"
7 "crypto/tls"
8 "crypto/x509"
9 "encoding/json"
10 "encoding/pem"
11 "errors"
12 "fmt"
13 "io"
14 "net"
15 "net/http"
16 "net/url"
17 "os"
18 "path/filepath"
19 "regexp"
20 "strconv"
21 "strings"
22 "sync"
23 "time"
24)
25
26// The real control-plane client (the contract, sections 3.3 and 4): what the node says to POST /v1/tunnel/authorize and GET /v1/tunnel/feed.
27//
28// The node holds exactly ONE secret: its own Ed25519 signing key. Every request carries an assertion (header x-all-node-assertion), a 60 s JWS bound to
29// the method, the URL and the SHA-256 of the exact body bytes, used once. The control plane has the public half (TUNNEL_NODE_KEYS; removing the kid
30// revokes the node). There is no shared secret, no commerce key and no bearer token in this program, and the request a device could influence is only
31// the device id and the session id inside the authorize body: never a host name.
32
33const (
34 authorizePath = "/v1/tunnel/authorize"
35 feedPath = "/v1/tunnel/feed"
36 jwksPath = "/.well-known/device-jwks.json"
37
38 maxControlBody = 256 << 10 // an answer carries at most a 64 KiB statement; anything bigger is not the control plane
39 maxJWKSBody = 64 << 10
40)
41
42var reErrorWord = regexp.MustCompile(`^[a-z_]{1,48}$`)
43
44// HTTPControlConfig is what the client needs. Everything but the key has a safe default.
45type HTTPControlConfig struct {
46 BaseURL string // https://<control plane>, nothing after the host
47 NodeID string
48 KID string // the kid of this node's public key in TUNNEL_NODE_KEYS
49 Key ed25519.PrivateKey
50 Client *http.Client // nil: a client of its own (TLS, no proxy, no redirect, bounded connections)
51 Now func() time.Time // nil: time.Now
52 Version string // for the User-Agent
53}
54
55// HTTPControl is the node's ControlPlane and FeedSource.
56type HTTPControl struct {
57 base *url.URL
58 hostHTU string // the host as the control plane's runtime writes it in a URL: the port only when it is not 443
59 nodeID string
60 kid string
61 key ed25519.PrivateKey
62 client *http.Client
63 now func() time.Time
64 agent string
65}
66
67// NewHTTPControl validates the configuration. It does not touch the network.
68func NewHTTPControl(cfg HTTPControlConfig) (*HTTPControl, error) {
69 u, err := url.Parse(cfg.BaseURL)
70 if err != nil || u.Scheme != "https" || u.Hostname() == "" || u.User != nil || u.RawQuery != "" || u.Fragment != "" || (u.Path != "" && u.Path != "/") {
71 return nil, fmt.Errorf("control plane URL %q: want https://<host>[:port] and nothing else", cfg.BaseURL)
72 }
73 if !reNodeID.MatchString(cfg.NodeID) || !reKID.MatchString(cfg.KID) {
74 return nil, fmt.Errorf("control plane client: bad node id or kid")
75 }
76 if len(cfg.Key) != ed25519.PrivateKeySize {
77 return nil, fmt.Errorf("control plane client: no node signing key")
78 }
79 c := &HTTPControl{base: u, nodeID: cfg.NodeID, kid: cfg.KID, key: cfg.Key, client: cfg.Client, now: cfg.Now, agent: "all-tunnel/" + cfg.Version}
80 c.hostHTU = strings.ToLower(u.Host)
81 if u.Port() == "443" {
82 c.hostHTU = strings.ToLower(u.Hostname())
83 }
84 if c.client == nil {
85 c.client = newControlClient()
86 } else {
87 // A client that was handed in is used as it is, except that it never follows a redirect: an assertion must not travel to a host nobody configured.
88 own := *c.client
89 own.CheckRedirect = func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }
90 c.client = &own
91 }
92 if c.now == nil {
93 c.now = time.Now
94 }
95 return c, nil
96}
97
98// newControlClient is the client of the node to its control plane: TLS from the system roots, no proxy from the environment, no redirect followed (a
99// redirect would carry an assertion to a host nobody configured) and a bounded pool (authorize may run authorize_max_concurrent at once, plus the feed).
100func newControlClient() *http.Client {
101 limit := policyData.Authorization.AuthorizeMaxConcur
102 return &http.Client{
103 Transport: &http.Transport{
104 Proxy: nil,
105 DialContext: (&net.Dialer{Timeout: 5 * time.Second, KeepAliveConfig: net.KeepAliveConfig{Enable: true}}).DialContext,
106 TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12},
107 TLSHandshakeTimeout: 5 * time.Second,
108 ForceAttemptHTTP2: true,
109 MaxIdleConnsPerHost: limit + 2,
110 MaxConnsPerHost: limit + 2,
111 IdleConnTimeout: 90 * time.Second,
112 },
113 CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
114 }
115}
116
117// compactJWS signs header and claims (canonical JSON, base64url) with key.
118func compactJWS(key ed25519.PrivateKey, header, claims map[string]any) string {
119 input := b64Encode([]byte(canonicalJSON(header))) + "." + b64Encode([]byte(canonicalJSON(claims)))
120 return input + "." + b64Encode(ed25519.Sign(key, []byte(input)))
121}
122
123// assertion is the node's proof for ONE request: the method, the exact URL and the hash of the exact body bytes, 60 s, with a jti nobody reuses.
124func (c *HTTPControl) assertion(method, requestURL string, body []byte) string {
125 iat := c.now().Unix()
126 return compactJWS(c.key,
127 map[string]any{"alg": "EdDSA", "typ": "node-assertion", "kid": c.kid},
128 map[string]any{"typ": "node-assertion", "ver": 1, "sub": c.nodeID, "aud": policyData.Identity.AudNodeAssert, "htm": method, "htu": requestURL,
129 "iat": iat, "exp": iat + int64(policyData.Authorization.NodeAssertionTTLS), "jti": newNonce(), "bh": sha256B64(body)})
130}
131
132// do sends one signed request and returns the status and the body (at most maxControlBody). A transport failure is a plain error; any answer, including
133// an error answer, is a status.
134func (c *HTTPControl) do(ctx context.Context, method, pathAndQuery string, body []byte) (int, []byte, error) {
135 requestURL := "https://" + c.hostHTU + pathAndQuery
136 req, err := http.NewRequestWithContext(ctx, method, c.base.Scheme+"://"+c.base.Host+pathAndQuery, bytes.NewReader(body))
137 if err != nil {
138 return 0, nil, err
139 }
140 req.Header.Set("x-all-node-assertion", c.assertion(method, requestURL, body))
141 req.Header.Set("accept", "application/json")
142 req.Header.Set("user-agent", c.agent)
143 if method == http.MethodPost {
144 req.Header.Set("content-type", "application/json")
145 }
146 return c.send(req, maxControlBody)
147}
148
149func (c *HTTPControl) send(req *http.Request, limit int64) (int, []byte, error) {
150 resp, err := c.client.Do(req)
151 if err != nil {
152 return 0, nil, err
153 }
154 defer resp.Body.Close()
155 data, err := io.ReadAll(io.LimitReader(resp.Body, limit+1))
156 if err != nil {
157 return 0, nil, err
158 }
159 if int64(len(data)) > limit {
160 return resp.StatusCode, nil, errors.New("control plane answer too large")
161 }
162 return resp.StatusCode, data, nil
163}
164
165// controlError turns a non-200 answer into a ControlError: the closed word of the error body when it is one (never any other text from the network),
166// bad_response when it is not.
167func controlError(status int, data []byte) *ControlError {
168 word := "bad_response"
169 if value, err := parseJSON(string(data)); err == nil {
170 if m, ok := value.(map[string]any); ok && m["ok"] == false {
171 if s, ok := m["error"].(string); ok && reErrorWord.MatchString(s) {
172 word = s
173 }
174 }
175 }
176 return &ControlError{Status: status, Code: word}
177}
178
179// Authorize asks the control plane what the device the node authenticated may do. The body carries the device id, the session's jti, the reason and
180// the node: never a host name. Anything but a well-formed 200 is an error the node maps to a closed word (errForControl).
181func (c *HTTPControl) Authorize(ctx context.Context, req AuthorizeRequest) (AuthorizeResponse, error) {
182 body := []byte(canonicalJSON(req.toMap()))
183 status, data, err := c.do(ctx, http.MethodPost, authorizePath, body)
184 if err != nil {
185 return AuthorizeResponse{}, err
186 }
187 if status != http.StatusOK {
188 return AuthorizeResponse{}, controlError(status, data)
189 }
190 resp, err := parseAuthorizeResponse(data)
191 if err != nil {
192 return AuthorizeResponse{}, &ControlError{Status: http.StatusBadGateway, Code: "bad_response"}
193 }
194 return resp, nil
195}
196
197// parseAuthorizeResponse reads a 200 answer of authorize (authorize-response.schema.json). The fields the node uses are checked strictly; a field it
198// does not know is ignored (a newer control plane may add one).
199func parseAuthorizeResponse(data []byte) (AuthorizeResponse, error) {
200 value, err := parseJSON(string(data))
201 if err != nil {
202 return AuthorizeResponse{}, err
203 }
204 m, ok := value.(map[string]any)
205 if !ok || m["ok"] != true {
206 return AuthorizeResponse{}, errors.New("authorize answer shape")
207 }
208 if v, ok := claimInt(m, "v"); !ok || v != 1 {
209 return AuthorizeResponse{}, errors.New("authorize answer version")
210 }
211 var a AuthorizeResponse
212 var okDevice, okLabel, okHost, okStatus, okPlan bool
213 a.DeviceID, okDevice = claimString(m, "device_id")
214 a.Label, okLabel = claimString(m, "label")
215 a.Host, okHost = claimString(m, "host")
216 a.Status, okStatus = claimString(m, "status")
217 a.Plan, okPlan = claimString(m, "plan")
218 validUntil, okUntil := epochOK(m["valid_until"])
219 serverTime, okTime := epochOK(m["server_time"])
220 policy, okPolicy := checkPolicyBlock(m["policy"])
221 if !okDevice || !reDeviceID.MatchString(a.DeviceID) || !okLabel || !okHost || !okStatus || !okPlan || !okUntil || !okTime || !okPolicy {
222 return AuthorizeResponse{}, errors.New("authorize answer fields")
223 }
224 a.ValidUntil, a.ServerTime, a.Policy = validUntil, serverTime, policy
225 if access, present := m["access"]; present {
226 text, ok := access.(string)
227 if !ok || statementLine(text) == "" {
228 return AuthorizeResponse{}, errors.New("authorize answer access")
229 }
230 a.Access = text
231 }
232 if nodes, present := m["nodes"]; present {
233 if _, ok := nodes.([]any); !ok {
234 return AuthorizeResponse{}, errors.New("authorize answer nodes")
235 }
236 a.Nodes = nodes
237 }
238 if wave, ok := claimInt(m, "e2e_wave"); ok && wave >= 0 && wave <= 100 {
239 a.E2EWave = int(wave)
240 }
241 a.E2EKill, _ = m["e2e_kill"].(bool)
242 a.CopyE2EE, _ = m["copy_e2ee"].(bool)
243 return a, nil
244}
245
246// ---------------------------------------------------------------------------------------------------------------------------- the feed
247
248// Types of a feed event (policy enums.feed_event_type). kill and kill_clear are global; every other event names a device and its label.
249const (
250 feedRevoke = "revoke"
251 feedPause = "pause"
252 feedResume = "resume"
253 feedLicenseEnd = "license_end"
254 feedRelease = "release"
255 feedKill = "kill"
256 feedKillClear = "kill_clear"
257)
258
259// FeedEvent is one entry of GET /v1/tunnel/feed (feed-response.schema.json).
260type FeedEvent struct {
261 Epoch int64
262 Type string
263 DeviceID string // empty for a global event
264 Label string // empty for a global event
265 At int64 // epoch seconds
266}
267
268func (e FeedEvent) global() bool { return e.Type == feedKill || e.Type == feedKillClear }
269
270// FeedAnswer is the answer of the feed: Epoch is where the node's cursor goes next. Reset says the cursor is not one the control plane can serve from
271// (older than its history, or ahead of it). Skipped counts the events of the answer the node could not read (an unknown type or a malformed field).
272type FeedAnswer struct {
273 Epoch int64
274 ServerTime int64
275 Reset bool
276 Events []FeedEvent
277 Skipped int
278}
279
280// FeedSource is the node's side of GET /v1/tunnel/feed. since is the node's cursor; wait asks the control plane to hold an empty answer until an event
281// arrives (at most policy timing.feed_poll_s).
282type FeedSource interface {
283 Feed(ctx context.Context, since int64, wait time.Duration) (FeedAnswer, error)
284}
285
286// Feed polls the feed. The URL is the one that is signed, exactly: https://<host><path>?since=<n>[&wait=<s>].
287func (c *HTTPControl) Feed(ctx context.Context, since int64, wait time.Duration) (FeedAnswer, error) {
288 query := "?since=" + strconv.FormatInt(since, 10)
289 if w := int64(wait / time.Second); w > 0 {
290 query += "&wait=" + strconv.FormatInt(min(w, int64(policyData.Timing.FeedPollS)), 10)
291 }
292 status, data, err := c.do(ctx, http.MethodGet, feedPath+query, nil)
293 if err != nil {
294 return FeedAnswer{}, err
295 }
296 if status != http.StatusOK {
297 return FeedAnswer{}, controlError(status, data)
298 }
299 answer, err := parseFeedAnswer(data)
300 if err != nil {
301 return FeedAnswer{}, &ControlError{Status: http.StatusBadGateway, Code: "bad_response"}
302 }
303 return answer, nil
304}
305
306// parseFeedAnswer reads a feed answer. The envelope is read strictly (a node must not act on an answer it cannot make sense of); an event it cannot
307// read, or whose type it does not know, is skipped and counted, because the rest of the answer is still true and a revoke must not wait for it.
308func parseFeedAnswer(data []byte) (FeedAnswer, error) {
309 value, err := parseJSON(string(data))
310 if err != nil {
311 return FeedAnswer{}, err
312 }
313 m, ok := value.(map[string]any)
314 if !ok || m["ok"] != true {
315 return FeedAnswer{}, errors.New("feed answer shape")
316 }
317 if v, ok := claimInt(m, "v"); !ok || v != 1 {
318 return FeedAnswer{}, errors.New("feed answer version")
319 }
320 epoch, okEpoch := claimInt(m, "epoch")
321 serverTime, okTime := epochOK(m["server_time"])
322 list, okList := m["events"].([]any)
323 if !okEpoch || epoch < 0 || epoch > 1<<53-1 || !okTime || !okList || len(list) > feedMaxEvents {
324 return FeedAnswer{}, errors.New("feed answer fields")
325 }
326 answer := FeedAnswer{Epoch: epoch, ServerTime: serverTime}
327 if reset, present := m["reset"]; present {
328 if answer.Reset, ok = reset.(bool); !ok {
329 return FeedAnswer{}, errors.New("feed answer reset")
330 }
331 }
332 for _, item := range list {
333 event, ok := parseFeedEvent(item)
334 if !ok {
335 answer.Skipped++
336 continue
337 }
338 answer.Events = append(answer.Events, event)
339 }
340 return answer, nil
341}
342
343const feedMaxEvents = 1000 // feed-response.schema.json: events maxItems
344
345func parseFeedEvent(item any) (FeedEvent, bool) {
346 m, ok := item.(map[string]any)
347 if !ok {
348 return FeedEvent{}, false
349 }
350 epoch, okEpoch := claimInt(m, "epoch")
351 at, okAt := epochOK(m["at"])
352 kind, okKind := claimString(m, "type")
353 if !okEpoch || epoch < 1 || epoch > 1<<53-1 || !okAt || !okKind || !inList(policyData.Enums.FeedEventType, kind) {
354 return FeedEvent{}, false
355 }
356 event := FeedEvent{Epoch: epoch, Type: kind, At: at}
357 device, label := m["device_id"], m["label"]
358 if event.global() {
359 return event, device == nil && label == nil
360 }
361 event.DeviceID, _ = device.(string)
362 event.Label, _ = label.(string)
363 return event, reDeviceID.MatchString(event.DeviceID) && reLabel.MatchString(event.Label)
364}
365
366// FetchJWKS reads the control plane's published keys. The document is public, so the request carries no assertion; it is read from the configured
367// control plane only, never from a URL an answer names, and a redirect is not followed.
368func (c *HTTPControl) FetchJWKS(ctx context.Context) ([]byte, error) {
369 req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.base.Scheme+"://"+c.base.Host+jwksPath, nil)
370 if err != nil {
371 return nil, err
372 }
373 req.Header.Set("accept", "application/json")
374 req.Header.Set("user-agent", c.agent)
375 status, data, err := c.send(req, maxJWKSBody)
376 if err != nil {
377 return nil, err
378 }
379 if status != http.StatusOK {
380 return nil, &ControlError{Status: status, Code: "jwks_unavailable"}
381 }
382 return data, nil
383}
384
385// ---------------------------------------------------------------------------------------------------------------------------- the node's key
386
387// LoadNodeKey reads the node's Ed25519 signing key: a PKCS#8 PEM file that only its owner may read (a key a group or the world can read is refused,
388// the way ssh refuses one). The one exception is the copy systemd hands the service (LoadCredential=): systemd 255 makes it 0440 root:root with an
389// ACL for the unit's dynamic user, inside $CREDENTIALS_DIRECTORY, which only that unit can open; there the group bits are the ACL mask, not a group,
390// so only world bits are refused.
391func LoadNodeKey(path string) (ed25519.PrivateKey, error) {
392 st, err := os.Stat(path)
393 if err != nil {
394 return nil, fmt.Errorf("node key: %w", err)
395 }
396 refused := os.FileMode(0o077)
397 if dir := os.Getenv("CREDENTIALS_DIRECTORY"); dir != "" && filepath.IsAbs(dir) && filepath.Dir(filepath.Clean(path)) == filepath.Clean(dir) {
398 refused = 0o007
399 }
400 if !st.Mode().IsRegular() || st.Mode().Perm()&refused != 0 {
401 return nil, fmt.Errorf("node key %s must be a regular file readable by its owner only (mode 0600)", path)
402 }
403 raw, err := os.ReadFile(path)
404 if err != nil {
405 return nil, fmt.Errorf("node key: %w", err)
406 }
407 block, _ := pem.Decode(raw)
408 if block == nil || block.Type != "PRIVATE KEY" {
409 return nil, fmt.Errorf("node key %s is not a PKCS#8 PEM private key", path)
410 }
411 parsed, err := x509.ParsePKCS8PrivateKey(block.Bytes)
412 if err != nil {
413 return nil, fmt.Errorf("node key %s: %w", path, err)
414 }
415 key, ok := parsed.(ed25519.PrivateKey)
416 if !ok {
417 return nil, fmt.Errorf("node key %s is not an Ed25519 key", path)
418 }
419 return key, nil
420}
421
422// GenerateNodeKey creates the node's key at path (mode 0600, never overwriting a file that is there) and returns it.
423func GenerateNodeKey(path string) (ed25519.PrivateKey, error) {
424 _, key, err := ed25519.GenerateKey(nil)
425 if err != nil {
426 return nil, err
427 }
428 der, err := x509.MarshalPKCS8PrivateKey(key)
429 if err != nil {
430 return nil, err
431 }
432 f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
433 if err != nil {
434 return nil, fmt.Errorf("node key: %w", err)
435 }
436 if _, err := f.Write(pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: der})); err != nil {
437 _ = f.Close()
438 _ = os.Remove(path)
439 return nil, err
440 }
441 return key, f.Close()
442}
443
444// NodeKeysEntry is the TUNNEL_NODE_KEYS document (node-keys.schema.json) that registers this node's public key with the control plane. Printing it is
445// all the node does to "publish" its key: the deployment owner sets the variable (the design plan, section 9); revoking the node is removing
446// the kid from it.
447func NodeKeysEntry(nodeID, kid string, key ed25519.PrivateKey) string {
448 pub := key.Public().(ed25519.PublicKey)
449 entry, _ := json.Marshal(map[string]any{"keys": []any{map[string]any{"kid": kid, "node": nodeID, "kty": "OKP", "crv": "Ed25519", "x": b64Encode(pub)}}})
450 return string(entry)
451}
452
453// ---------------------------------------------------------------------------------------------------------------------------- reachability
454
455// cpHealth is what the node has seen of its control plane's reachability: the moment of the last answer of any kind (an error answer such as a 402 is
456// an answer: the control plane is up) and of the last failure to get one (a transport error, a timeout, a 5xx, a 429, an answer it could not read).
457// The paid grace asks it, so that one slow answer never decides alone.
458type cpHealth struct {
459 mu sync.Mutex
460 lastGood time.Time
461 lastBad time.Time
462}
463
464func (h *cpHealth) good(now time.Time) {
465 h.mu.Lock()
466 h.lastGood = now
467 h.mu.Unlock()
468}
469
470func (h *cpHealth) bad(now time.Time) {
471 h.mu.Lock()
472 h.lastBad = now
473 h.mu.Unlock()
474}
475
476// down reports whether the last thing the node saw was a failure and nothing has answered for at least quiet.
477func (h *cpHealth) down(now time.Time, quiet time.Duration) bool {
478 h.mu.Lock()
479 defer h.mu.Unlock()
480 return h.lastBad.After(h.lastGood) && now.Sub(h.lastGood) >= quiet
481}
482
483// record files the outcome of one call to the control plane.
484func (h *cpHealth) record(now time.Time, err error) {
485 if err == nil || isRefusal(err) {
486 h.good(now)
487 return
488 }
489 h.bad(now)
490}
491
492// isRefusal reports whether err is the control plane saying no (a 4xx other than 429): it answered, so it is reachable.
493func isRefusal(err error) bool {
494 var ce *ControlError
495 return errors.As(err, &ce) && ce.Status >= 400 && ce.Status < 500 && ce.Status != http.StatusTooManyRequests
496}