ALL

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

tunnel/main.go

SHA-256: 8a0be6c0d970cba7de4dbd9e95d1d9ae48ce21f1f41fa401a28b62e046fdb42b
213 líneas · 9781 bytes · Ver en GitHub

1package main
2
3import (
4 "errors"
5 "flag"
6 "fmt"
7 "io"
8 "log/slog"
9 "os"
10 "os/signal"
11 "strconv"
12 "syscall"
13 "time"
14)
15
16// version is overridden at build time: -ldflags "-X main.version=..."
17var version = "dev"
18
19func main() { os.Exit(run(os.Args[1:], os.Stdout, os.Stderr)) }
20
21func envStr(key, def string) string {
22 if v, ok := os.LookupEnv(key); ok && v != "" {
23 return v
24 }
25 return def
26}
27
28func envInt(key string, def int64) int64 {
29 if v := os.Getenv(key); v != "" {
30 if n, err := strconv.ParseInt(v, 10, 64); err == nil {
31 return n
32 }
33 }
34 return def
35}
36
37func envDur(key string, def time.Duration) time.Duration {
38 if v := os.Getenv(key); v != "" {
39 if d, err := time.ParseDuration(v); err == nil {
40 return d
41 }
42 }
43 return def
44}
45
46// backend is what the node asks of the outside world: the control plane's keys, the authorize call and the feed. The production build registers the real
47// ones (backend_prod.go: the HTTP client with the node's assertion, the JWKS cache); the test-node build (`-tags testnode`) registers file-driven fakes. A
48// build with neither refuses to start rather than run without verifying anyone.
49type backend struct {
50 // flags registers the backend's own flags and returns what completes cfg once the flags are parsed.
51 flags func(fs *flag.FlagSet, stdout io.Writer) func(cfg *Config) error
52}
53
54var theBackend *backend
55
56// Where the production build looks by default: its /metrics and /healthz (loopback), and the list of OpenAI connector prefixes. The test node leaves both
57// empty (many test nodes run side by side and none may call the network).
58var (
59 defaultMetricsAddr string
60 defaultOpenAIPrefixesURL string
61)
62
63var errNoBackend = errors.New("this build has no control-plane client: use the production build, or the test-node build (-tags testnode) with its fakes")
64
65// errDone is what a backend's complete returns when it did what the command line asked (printing a key) and the node must not start.
66var errDone = errors.New("done")
67
68func run(args []string, stdout, stderr io.Writer) int {
69 cfg := DefaultConfig()
70 fs := flag.NewFlagSet("all-tunnel", flag.ContinueOnError)
71 fs.SetOutput(stderr)
72
73 fs.StringVar(&cfg.HTTPSAddr, "https-addr", envStr("TUNNEL_HTTPS_ADDR", cfg.HTTPSAddr), "the public listener: devices' control and data connections and clients' TLS, told apart by SNI [TUNNEL_HTTPS_ADDR]")
74 fs.StringVar(&cfg.HTTPAddr, "http-addr", envStr("TUNNEL_HTTP_ADDR", cfg.HTTPAddr), "listener that serves only the node's own ACME webroot, '' disables [TUNNEL_HTTP_ADDR]")
75 fs.StringVar(&cfg.NodeID, "node-id", envStr("TUNNEL_NODE_ID", ""), "this node's id in the control plane, 1-16 characters of [a-z0-9] [TUNNEL_NODE_ID]")
76 fs.StringVar(&cfg.NodeHost, "node-host", envStr("TUNNEL_NODE_HOST", ""), "this node's own public name: the SNI that reaches the control handler [TUNNEL_NODE_HOST]")
77 fs.IntVar(&cfg.PublicPort, "public-port", int(envInt("TUNNEL_PUBLIC_PORT", int64(cfg.PublicPort))), "the port devices dial, only for the proof's URL [TUNNEL_PUBLIC_PORT]")
78 fs.StringVar(&cfg.ACMEWebroot, "acme-webroot", envStr("TUNNEL_ACME_WEBROOT", ""), "directory with .well-known/acme-challenge for the node's own certificate [TUNNEL_ACME_WEBROOT]")
79 certFile := fs.String("cert-file", envStr("TUNNEL_CERT_FILE", ""), "the node's public-CA certificate chain (PEM), re-read when it changes [TUNNEL_CERT_FILE]")
80 keyFile := fs.String("key-file", envStr("TUNNEL_KEY_FILE", ""), "the node certificate's private key (PEM) [TUNNEL_KEY_FILE]")
81 skewS := fs.Int64("clock-skew", envInt("TUNNEL_CLOCK_SKEW_S", int64(cfg.ClockSkew/time.Second)), "seconds a device's clock may be ahead of this node's (at most the policy's maximum) [TUNNEL_CLOCK_SKEW_S]")
82 fs.DurationVar(&cfg.FeedPoll, "feed-poll", envDur("TUNNEL_FEED_POLL", cfg.FeedPoll), "how often the node looks at the control plane's feed (at most the policy's) [TUNNEL_FEED_POLL]")
83 fs.DurationVar(&cfg.Grace, "grace", envDur("TUNNEL_GRACE", cfg.Grace), "how long a paid connection outlives its valid_until while the control plane does not answer (at most the policy's) [TUNNEL_GRACE]")
84 fs.DurationVar(&cfg.DrainQuiet, "drain-quiet", envDur("TUNNEL_DRAIN_QUIET", cfg.DrainQuiet), "on SIGTERM keep the open streams this long at most, then cut them and exit; the unit's TimeoutStopSec must be longer [TUNNEL_DRAIN_QUIET]")
85
86 fs.StringVar(&cfg.MetricsAddr, "metrics-addr", envStr("TUNNEL_METRICS_ADDR", defaultMetricsAddr), "loopback address of /metrics and /healthz (aggregate counters only), '' disables [TUNNEL_METRICS_ADDR]")
87 fs.StringVar(&cfg.OpenAIPrefixesURL, "openai-prefixes-url", envStr("TUNNEL_OPENAI_PREFIXES_URL", defaultOpenAIPrefixesURL), "where OpenAI publishes its connector prefixes (the node's openai quota class), '' disables the refresh [TUNNEL_OPENAI_PREFIXES_URL]")
88 fs.StringVar(&cfg.OpenAIPrefixesCache, "openai-prefixes-cache", envStr("TUNNEL_OPENAI_PREFIXES_CACHE", ""), "file that keeps the last good prefix list across restarts (public data) [TUNNEL_OPENAI_PREFIXES_CACHE]")
89 fs.IntVar(&cfg.MaxStreams, "max-streams", int(envInt("TUNNEL_MAX_STREAMS", int64(cfg.MaxStreams))), "max concurrent client connections [TUNNEL_MAX_STREAMS]")
90 fs.IntVar(&cfg.MaxStreamsPerHost, "max-streams-per-host", int(envInt("TUNNEL_MAX_STREAMS_PER_HOST", int64(cfg.MaxStreamsPerHost))), "max concurrent client connections per device [TUNNEL_MAX_STREAMS_PER_HOST]")
91 fs.Int64Var(&cfg.MaxStreamBytes, "max-stream-bytes", envInt("TUNNEL_MAX_STREAM_BYTES", cfg.MaxStreamBytes), "max bytes (both directions) per stream [TUNNEL_MAX_STREAM_BYTES]")
92 fs.DurationVar(&cfg.StreamIdle, "stream-idle", envDur("TUNNEL_STREAM_IDLE", cfg.StreamIdle), "close a stream after this long without bytes [TUNNEL_STREAM_IDLE]")
93 fs.DurationVar(&cfg.StreamMaxAge, "stream-max-age", envDur("TUNNEL_STREAM_MAX_AGE", cfg.StreamMaxAge), "hard cap on one stream's lifetime [TUNNEL_STREAM_MAX_AGE]")
94 fs.DurationVar(&cfg.HelloTimeout, "hello-timeout", envDur("TUNNEL_HELLO_TIMEOUT", cfg.HelloTimeout), "client must send its ClientHello within this [TUNNEL_HELLO_TIMEOUT]")
95 fs.DurationVar(&cfg.AttachTimeout, "attach-timeout", envDur("TUNNEL_ATTACH_TIMEOUT", cfg.AttachTimeout), "device must ATTACH within this after OPEN [TUNNEL_ATTACH_TIMEOUT]")
96 fs.DurationVar(&cfg.DeadPeer, "dead-peer", envDur("TUNNEL_DEAD_PEER", cfg.DeadPeer), "drop a control connection silent for this long [TUNNEL_DEAD_PEER]")
97
98 logFormat := fs.String("log-format", envStr("TUNNEL_LOG_FORMAT", "json"), "json or text [TUNNEL_LOG_FORMAT]")
99 logLevel := fs.String("log-level", envStr("TUNNEL_LOG_LEVEL", "info"), "debug, info, warn or error [TUNNEL_LOG_LEVEL]")
100 showVersion := fs.Bool("version", false, "print the version and exit")
101
102 var complete func(*Config) error
103 if theBackend != nil {
104 complete = theBackend.flags(fs, stdout)
105 }
106
107 if err := fs.Parse(args); err != nil {
108 if err == flag.ErrHelp {
109 return 0
110 }
111 return 2
112 }
113 if *showVersion {
114 fmt.Fprintln(stdout, "all-tunnel", version)
115 return 0
116 }
117
118 var lvl slog.Level
119 if err := lvl.UnmarshalText([]byte(*logLevel)); err != nil {
120 fmt.Fprintln(stderr, "error: bad -log-level")
121 return 2
122 }
123 hopts := &slog.HandlerOptions{Level: lvl}
124 if *logFormat == "text" {
125 cfg.Logger = slog.New(slog.NewTextHandler(stderr, hopts))
126 } else {
127 cfg.Logger = slog.New(slog.NewJSONHandler(stderr, hopts))
128 }
129 cfg.ClockSkew = time.Duration(*skewS) * time.Second
130
131 if complete == nil {
132 fmt.Fprintln(stderr, "error:", errNoBackend)
133 return 1
134 }
135 var reloader *certReloader
136 if *certFile != "" || *keyFile != "" {
137 var err error
138 if reloader, err = newCertReloader(*certFile, *keyFile, cfg.NodeHost, newClock(nil, nil), cfg.Logger); err != nil {
139 fmt.Fprintln(stderr, "error:", err)
140 return 1
141 }
142 cfg.CertSource = reloader.get
143 cfg.CertNotAfter = reloader.NotAfter
144 cfg.Background = append(cfg.Background, reloader.Run)
145 }
146 if err := complete(&cfg); err != nil {
147 if errors.Is(err, errDone) {
148 return 0
149 }
150 fmt.Fprintln(stderr, "error:", err)
151 return 1
152 }
153
154 srv, err := NewServer(cfg)
155 if err != nil {
156 fmt.Fprintln(stderr, "error:", err)
157 return 1
158 }
159 if err := srv.Start(); err != nil {
160 fmt.Fprintln(stderr, "error:", err)
161 return 1
162 }
163 cfg.Logger.Info("started", "version", version, "node", cfg.NodeID)
164 announce(stdout, srv)
165
166 return serveUntilSignalled(srv, cfg, reloader)
167}
168
169// serveUntilSignalled is the node's life after it started. SIGTERM drains (the open streams finish, up to the quiet period, then it exits; a second
170// SIGTERM stops waiting), SIGINT stops at once (a developer's Ctrl-C), SIGHUP reads the node's certificate again.
171func serveUntilSignalled(srv *Server, cfg Config, reloader *certReloader) int {
172 sigs := make(chan os.Signal, 8)
173 signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP)
174 defer signal.Stop(sigs)
175 drained := make(chan struct{})
176 draining := false
177 for {
178 select {
179 case <-drained:
180 return 0
181 case sig := <-sigs:
182 switch sig {
183 case syscall.SIGHUP:
184 if reloader == nil {
185 continue
186 }
187 if err := reloader.Reload(); err != nil {
188 cfg.Logger.Warn("node_cert_reload_refused", "reason", err.Error())
189 } else {
190 cfg.Logger.Info("node_cert_reloaded", "not_after", reloader.NotAfter().UTC().Format(time.RFC3339))
191 }
192 case syscall.SIGTERM:
193 if !draining {
194 draining = true
195 go func() {
196 srv.Drain(cfg.DrainQuiet)
197 close(drained)
198 }()
199 continue
200 }
201 cfg.Logger.Warn("drain_cut_short")
202 go srv.Shutdown(0)
203 default:
204 cfg.Logger.Info("shutting_down", "grace", cfg.ShutdownGrace.String())
205 srv.Shutdown(cfg.ShutdownGrace)
206 return 0
207 }
208 }
209 }
210}
211
212// announce is called once the node listens; the test-node build uses it to tell a harness the ports it got. The production build says nothing.
213var announce = func(io.Writer, *Server) {}