diff --git a/AGENTS.md b/AGENTS.md index 1952df0..0389b11 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,6 +16,12 @@ what we take vs. skip). Never vendor it into the build. ## Layout - `cmd/tunneld` — flags/env wiring, listeners, signal handling +- `cmd/tunnel` — the native Go client/CLI: JSON config, provisioning, routes, + background service +- `internal/client` — client implementation: config/API, key/CSR/certificate + storage, bridge WebSocket, TLS termination and local proxying +- `internal/server/limits.go`, `cooldown.go` — per-IP/per-tunnel rate limits + and the ACME CA cool-down - `internal/server/proto.go` — wire types (pinned JSON key order), IDs/tokens (same generation as the reference: 12-char base32 slugs, `rly_` tokens, hex SHA-256 token hashes), route validation, data frames @@ -57,6 +63,11 @@ what we take vs. skip). Never vendor it into the build. - Frontend: `cd web && bun install && bun run check && bun run build` — the dashboard test skips when `internal/dashboard/dist` has no build (only the committed `.gitkeep`) +- Go client: `go build -o .local/tunnel ./cmd/tunnel`, then + `.local/tunnel config set-api ` and `.local/tunnel route add --name `; + `up`/`down` manage the background service, `status` shows local + server + state. `import --profile --api ` takes over an official CLI + profile (same tunnel/key/cert). - Dev loop: `go build -o .local/tunneld ./cmd/tunneld && .local/tunneld -domain localhost`, then `OPENTUNNEL_API=http://localhost:8080 opentunnel route add 4096 --name api` with the official CLI. `go build` drops a binary next to the source — it is @@ -135,20 +146,36 @@ what we take vs. skip). Never vendor it into the build. the service (`opentunnel down` / `up`) to fetch it immediately. An issuance killed mid-flight can leave `_acme-challenge` TXT records behind; delete them with `dns/deleteByNameType` (production and test tunnels share the zone). + The server sweeps leftover `_acme-challenge.*` records at startup, which is + safe only for a single server instance sharing the zone. Five consecutive + CA-side failures pause new orders for 1 minute (doubling to 1 hour); DNS and + validation errors do not trigger it. Rate limits default to 60 creates/hour + per client IP and 30 binds/hour per IP / 10 per tunnel (`0` disables); the + API is behind Traefik, so `X-Forwarded-For` is trusted for this. - **macOS system curl (LibreSSL) does not send SNI for `*.localhost` names**, so `curl https://api..localhost:8443/` gets closed (the relay requires SNI). Test with `openssl s_client -servername ...` or Homebrew curl. - JSON key order of API bodies is contractual — always `json.Marshal` structs with declared field order, never maps. -- Bridge writes go through one ordered `outbox` channel per session. Control - messages therefore can queue behind data — a known M1 deviation from the - reference's deadlock-avoidance; a stalled visitor self-heals via the - client's idle timeout. Revisit with per-connection flow control. +- Bridge writes use separate prioritized **control** and ordered **data** + queues per session. `open`, `reset`, pings and pongs cut ahead of queued + visitor data; `end` travels in the data queue so it cannot overtake the final + response bytes. Visitor data backpressures the socket when the client is + slow. Do not move `end` back to the control queue (that truncates responses). +- The native Go client's config is `~/.config/tunnel/config.json` (0600; it + contains the tunnel token), with `private-key.pem`, `csr.pem`, + `certificate.pem` and `chain.pem` next to it. `--config` selects another + tunnel. `end` in its own bridge client is likewise ordered with data. +- Both clients claim routes on the same tunnel: stop the official + `opentunnel` service before starting `tunnel`, or the second attach gets + `route_conflict`. - The bridge WebSocket must use `coder/websocket` with compression disabled and the `opentunnel` subprotocol; read limits are 64 KiB before attach and 4+32 KiB after. -- No third-party deps beyond `github.com/coder/websocket`; keep it that way - unless a milestone calls for it. +- No third-party deps beyond `github.com/coder/websocket` and + `golang.org/x/crypto` (ACME); keep it that way unless a milestone calls for it. - Milestones: M1 core data path (done); M2 real ACME DNS-01 + renewals (done); - M3 rate limits/admin/hardening; M4 maybe a Go client. Don't pull M2+ scope - into M1 fixes, or M3 scope into M2 fixes. + M3 rate limits/admin/hardening (done); M4 Go client (done). Remaining optional + work: export/import admin endpoints, token rotation, multi-region mesh, + inbound PROXY protocol, per-connection wire flow control (the protocol has + none; queues are bounded). diff --git a/README.md b/README.md index a7ef0b9..868cd77 100644 --- a/README.md +++ b/README.md @@ -6,10 +6,9 @@ plaintext or private keys. It speaks the exact wire protocol from `docs/protocol.md`, so the official `opentunnel` CLI and SDK work against it unchanged. -Status: **M2** — core data path works end-to-end with the official CLI, and -tunnels get publicly trusted certificates through ACME DNS-01 (Let's Encrypt -+ Porkbun) with automatic renewal. The built-in local CA remains the default -when no DNS provider is configured. +Status: **M4** — the data path, real ACME DNS-01 certificates with renewal, +rate limits/hardening, and a native Go client are implemented. The official +`opentunnel` CLI still works unchanged. ## Quick start (local development) @@ -40,6 +39,51 @@ printf 'GET / HTTP/1.1\r\nHost: api..localhost\r\nConnection: close\r\n\r\n' -CAfile ~/.local/share/tunneld/ca.pem -quiet ``` +## Go client + +The repo includes a native Go client (`cmd/tunnel`) with a JSON config, so the +API URL is stored once instead of typed on every command: + +```bash +go build -o .local/tunnel ./cmd/tunnel + +.local/tunnel config set-api https://tunnel.tzgyn.dev +.local/tunnel route add 4096 --name api +.local/tunnel status +``` + +The first `route add` creates a tunnel, generates the P-256 key/CSR, waits for +the certificate, starts the background service, and prints the public URL. +Routes reload without restarting the service. + +| Command | | +| --- | --- | +| `tunnel config set-api ` | store the server URL in the JSON config | +| `tunnel config show` | show API, tunnel, certificate and routes (never the token) | +| `tunnel route add [--name NAME] [--proxy-protocol v1\|v2]` | add/update a route; random name if omitted | +| `tunnel route remove ` / `route list` | manage routes | +| `tunnel up` / `down` | start/stop the background service | +| `tunnel serve` | run in the foreground | +| `tunnel status` | local state, daemon, server state and certificate state | +| `tunnel logs` | tail the daemon log | +| `tunnel import --profile NAME --api URL` | take over an official CLI profile | + +The config lives at `~/.config/tunnel/config.json` (0600; it also stores the +tunnel token); the private key, CSR, certificate and chain sit next to it. Use +`--config FILE` for a second independent tunnel. + +### Taking over the official CLI profile + +```bash +tunnel import --profile coolify --api https://tunnel.tzgyn.dev +opentunnel --profile coolify down # release the routes +tunnel up +``` + +`import` copies the tunnel ID, token, private key and certificate, and converts +the official TOML routes to JSON. Both clients cannot claim the same routes at +the same time — stop one before starting the other. + ## Configuration Every flag also reads an environment variable (`tunneld -h`): @@ -62,6 +106,9 @@ Every flag also reads an environment variable (`tunneld -h`): | `-acme-propagation-seconds` | `ACME_PROPAGATION_TIMEOUT_SECONDS` | `60` | how long to wait for challenge TXT records; `0` skips the check | | `-acme-ca-bundle` | `ACME_CA_BUNDLE` | *(empty)* | extra roots for the ACME server itself (Pebble) | | `-challtestsrv-url` | `CHALLTESTSRV_URL` | `http://127.0.0.1:8055` | pebble-challtestsrv management URL (dev only) | +| `-create-limit-per-hour` | `TUNNEL_CREATE_LIMIT_PER_HOUR` | `60` | tunnels one client address may create per hour; `0` disables | +| `-bind-limit-per-hour` | `CERT_BIND_LIMIT_PER_HOUR` | `30` | new certificate binds one client address may request per hour | +| `-bind-limit-per-tunnel-per-hour` | `CERT_BIND_LIMIT_PER_TUNNEL_PER_HOUR` | `10` | new certificate binds one tunnel may request per hour | ## Certificates @@ -103,6 +150,12 @@ Switching a running server from the local CA to ACME re-issues every existing tunnel automatically, keeping the old certificate in service until the new one is ready. +At startup the server removes leftover `_acme-challenge.*` TXT records from the +zone (single-node assumption), and repeated CA-side failures trigger a +cool-down from one minute up to an hour instead of hammering Let's Encrypt. +New tunnel creation and CSR binds are rate-limited per client address and per +tunnel; the limits can be tuned with the settings above. + For testing against Let's Encrypt staging (untrusted certificates, separate rate limits), set `ACME_DIRECTORY_URL=https://acme-staging-v02.api.letsencrypt.org/directory`. @@ -169,7 +222,7 @@ visitors must trust that CA. With Porkbun configured, every tunnel gets a publicly trusted Let's Encrypt certificate and no client or visitor trust setup is needed. -## What works (M1 + M2) +## What works (M1–M4) - The full HTTP API contract (paths, status codes, bodies, key order) as recorded in the reference's `worker-contract.json` @@ -179,24 +232,28 @@ setup is needed. - Per-tunnel Let's Encrypt certificates (`.` + optional wildcard) that keep the client's key, with automatic renewal 30 days before expiry and retries on failure +- Rate limits, CA cool-down, stale challenge-record sweep, and admin metrics - The bridge WebSocket: attach/attached, route claiming with conflicts, - heartbeats, idle retirement, `open`/`end`/`reset`, 32 KiB data frames + heartbeats, idle retirement, `open`/`end`/`reset`, 32 KiB data frames, + prioritized control messages and ordered data - SNI routing on the TLS listener: the domain is terminated (API), tunnel hostnames are passed through still encrypted - Admin dashboard at `/` (bearer `ADMIN_TOKEN`): tunnels, bridges, cert - state/expiry, issuer, CA expiry + state/expiry, issuer, CA expiry, metrics - Persistence: JSON store with the CA, ACME account, server certificate, and tunnels +- Native Go client (`cmd/tunnel`): JSON config, routes with live reload, + PROXY protocol, background service, and import of the official CLI profile - Graceful shutdown: `drain` + close to every bridge -## Not yet (later milestones) +## Not yet (later) -- Issuance rate limits, CA cool-down, stale-DNS-record sweeps -- Metrics, more admin endpoints +- Additional admin endpoints (export/import, token rotation) - Multi-region mesh, inbound PROXY protocol on the TLS listener -- 60 s half-close timeout, per-connection flow control, control messages - cutting ahead of queued data (single ordered outbound queue today) -- Go client/CLI — for now use the official `opentunnel` CLI +- Per-connection flow control on the wire (the protocol has none; both sides + bound their queues) +- Replacing the official CLI on every platform; the Go client is macOS/Linux + (it uses POSIX process/session APIs for `up`/`down`) ## Development diff --git a/cmd/tunnel/main.go b/cmd/tunnel/main.go new file mode 100644 index 0000000..d24198a --- /dev/null +++ b/cmd/tunnel/main.go @@ -0,0 +1,581 @@ +package main + +import ( + "context" + "crypto/rand" + "encoding/base32" + "errors" + "flag" + "fmt" + "io" + "log" + "os" + "os/exec" + "os/signal" + "sort" + "strconv" + "strings" + "syscall" + "time" + + "gitea.tzgyn.dev/TZGyn/tunnel/internal/client" +) + +var version = "dev" + +func main() { + os.Exit(run(os.Args[1:])) +} + +func run(args []string) int { + configPath := client.DefaultPath() + if env := os.Getenv("TUNNEL_CONFIG"); env != "" { + configPath = env + } + fs := flag.NewFlagSet("tunnel", flag.ContinueOnError) + fs.SetOutput(os.Stderr) + fs.StringVar(&configPath, "config", configPath, "config file (default ~/.config/tunnel/config.json)") + fs.Usage = func() { usage(fs.Output()) } + if err := fs.Parse(args); err != nil { + return 2 + } + rest := fs.Args() + if len(rest) == 0 { + usage(os.Stderr) + return 2 + } + command, rest := rest[0], rest[1:] + switch command { + case "config": + return runConfig(configPath, rest) + case "route": + return runRoute(configPath, rest) + case "import": + return runImport(configPath, rest) + case "up": + return runUp(configPath) + case "down": + return runDown(configPath) + case "serve": + return runServe(configPath) + case "status": + return runStatus(configPath) + case "logs": + return runLogs(configPath) + case "version": + fmt.Println("tunnel", version) + return 0 + case "help", "-h", "--help": + usage(os.Stdout) + return 0 + default: + fmt.Fprintf(os.Stderr, "unknown command %q\n\n", command) + usage(os.Stderr) + return 2 + } +} + +func usage(w io.Writer) { + fmt.Fprint(w, `tunnel - OpenTunnel client for a self-hosted server + +Usage: + tunnel [--config FILE] [options] + +Commands: + config set-api Store the server API URL in the config + config show Show the config (never the token) + route add [--name NAME] [--proxy-protocol v1|v2] + route remove + route list + import --profile NAME --api URL + Take over an existing official opentunnel profile + up Provision if needed and run in the background + down Stop the background service + serve Run in the foreground (debugging) + status Show the tunnel, certificate, routes and daemon + logs Print the end of the daemon log + +Examples: + tunnel config set-api https://tunnel.tzgyn.dev + tunnel route add 4096 --name api + tunnel up +`) +} + +func runConfig(configPath string, args []string) int { + if len(args) == 0 { + fmt.Fprintln(os.Stderr, "usage: tunnel config set-api | tunnel config show") + return 2 + } + switch args[0] { + case "set-api", "set": + if len(args) != 2 { + fmt.Fprintln(os.Stderr, "usage: tunnel config set-api ") + return 2 + } + api, err := client.NormalizeAPI(args[1]) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + cfg, err := client.Load(configPath) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + old := cfg.API + cfg.API = api + if err := cfg.Save(configPath); err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + fmt.Printf("API set to %s\n", api) + if old != "" && old != api && cfg.TunnelID != "" { + fmt.Println("Note: the saved tunnel identity was kept; if that tunnel does not exist on this server, remove the config directory and start again.") + } + return 0 + case "show": + cfg, err := client.Load(configPath) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + printConfig(configPath, cfg) + return 0 + default: + fmt.Fprintln(os.Stderr, "usage: tunnel config set-api | tunnel config show") + return 2 + } +} + +func printConfig(configPath string, cfg *client.Config) { + fmt.Printf("Config: %s\n", configPath) + fmt.Printf("API: %s\n", valueOr(cfg.API, "(not set)")) + fmt.Printf("Tunnel: %s\n", valueOr(cfg.TunnelID, "(not created)")) + fmt.Printf("Hostname: %s\n", valueOr(cfg.Hostname, "(not created)")) + fmt.Printf("Cert: %s\n", valueOr(cfg.CertificateExpiry, "(not issued)")) + if len(cfg.Routes) == 0 { + fmt.Println("Routes: (none)") + return + } + fmt.Println("Routes:") + for _, name := range sortedRouteNames(cfg) { + route := cfg.Routes[name] + fmt.Printf(" %-20s -> %-24s %s\n", name, route.Target, cfg.RouteURL(name)) + } +} + +func runRoute(configPath string, args []string) int { + if len(args) == 0 { + fmt.Fprintln(os.Stderr, "usage: tunnel route add|remove|list") + return 2 + } + switch args[0] { + case "add": + var target, name, proxy string + for i := 1; i < len(args); i++ { + arg := args[i] + switch { + case arg == "--name": + if i+1 >= len(args) { + fmt.Fprintln(os.Stderr, "--name needs a value") + return 2 + } + i++ + name = args[i] + case strings.HasPrefix(arg, "--name="): + name = strings.TrimPrefix(arg, "--name=") + case arg == "--proxy-protocol": + if i+1 >= len(args) { + fmt.Fprintln(os.Stderr, "--proxy-protocol needs v1 or v2") + return 2 + } + i++ + proxy = args[i] + case strings.HasPrefix(arg, "--proxy-protocol="): + proxy = strings.TrimPrefix(arg, "--proxy-protocol=") + case strings.HasPrefix(arg, "-"): + fmt.Fprintf(os.Stderr, "unknown option %s\n", arg) + return 2 + case target == "": + target = arg + default: + fmt.Fprintf(os.Stderr, "unexpected argument %q\n", arg) + return 2 + } + } + if target == "" { + fmt.Fprintln(os.Stderr, "usage: tunnel route add [--name NAME]") + return 2 + } + normalized, err := client.NormalizeTarget(target) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + if name == "" { + name = randomRouteName() + } + if !client.ValidRouteName(name) { + fmt.Fprintf(os.Stderr, "invalid route name %q\n", name) + return 1 + } + cfg, err := client.Load(configPath) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + if cfg.API == "" { + fmt.Fprintln(os.Stderr, "no API configured; run: tunnel config set-api https://your-server") + return 1 + } + svc, err := client.NewService(configPath, version, nil) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) + err = svc.Ensure(ctx) + cancel() + if err != nil { + fmt.Fprintf(os.Stderr, "provision tunnel: %v\n", err) + return 1 + } + cfg = svc.Config + if err := cfg.AddRoute(name, normalized, proxy); err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + if err := cfg.Save(configPath); err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + fmt.Println("Tunnel ready.") + fmt.Printf("Route %s -> %s\n", name, cfg.RouteURL(name)) + if err := startDaemon(configPath); err != nil { + fmt.Fprintf(os.Stderr, "service: %v\n", err) + return 1 + } + return 0 + case "remove": + if len(args) != 2 { + fmt.Fprintln(os.Stderr, "usage: tunnel route remove ") + return 2 + } + cfg, err := client.Load(configPath) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + if !cfg.RemoveRoute(args[1]) { + fmt.Fprintf(os.Stderr, "no route named %q\n", args[1]) + return 1 + } + if err := cfg.Save(configPath); err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + fmt.Printf("Removed route %s (the service reloads within a few seconds).\n", args[1]) + return 0 + case "list": + cfg, err := client.Load(configPath) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + if len(cfg.Routes) == 0 { + fmt.Println("No routes.") + return 0 + } + for _, name := range sortedRouteNames(cfg) { + route := cfg.Routes[name] + fmt.Printf("%-20s -> %-24s %s\n", name, route.Target, cfg.RouteURL(name)) + } + return 0 + default: + fmt.Fprintln(os.Stderr, "usage: tunnel route add|remove|list") + return 2 + } +} + +func runImport(configPath string, args []string) int { + fs := flag.NewFlagSet("import", flag.ContinueOnError) + fs.SetOutput(os.Stderr) + var profile, api string + var force bool + fs.StringVar(&profile, "profile", "default", "official opentunnel profile name") + fs.StringVar(&api, "api", "", "server API URL") + fs.BoolVar(&force, "force", false, "overwrite an existing config") + if err := fs.Parse(args); err != nil { + return 2 + } + result, err := client.Import(client.ImportOptions{ + Profile: profile, + API: api, + ConfigPath: configPath, + Force: force, + }) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + sort.Strings(result.Routes) + fmt.Printf("Imported %s (%s) from %s\n", result.TunnelID, result.Hostname, result.Source) + fmt.Printf("Config: %s\n", result.ConfigPath) + if len(result.Routes) == 0 { + fmt.Println("Routes: (none)") + } else { + fmt.Printf("Routes: %s\n", strings.Join(result.Routes, ", ")) + } + fmt.Println() + fmt.Println("Stop the official service before starting this one:") + fmt.Printf(" opentunnel --profile %s down\n", profile) + fmt.Println("Then run:") + fmt.Printf(" tunnel --config %s up\n", result.ConfigPath) + return 0 +} + +func runUp(configPath string) int { + cfg, err := client.Load(configPath) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + if cfg.API == "" { + fmt.Fprintln(os.Stderr, "no API configured; run: tunnel config set-api https://your-server") + return 1 + } + if err := startDaemon(configPath); err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + return 0 +} + +func startDaemon(configPath string) error { + if pid, err := readPID(client.ConfigPaths(configPath).PID); err == nil && processAlive(pid) { + fmt.Printf("Already running (pid %d).\n", pid) + return nil + } + exe, err := os.Executable() + if err != nil { + return err + } + paths := client.ConfigPaths(configPath) + if err := os.MkdirAll(paths.Dir, 0o700); err != nil { + return err + } + logFile, err := os.OpenFile(paths.Log, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600) + if err != nil { + return err + } + cmd := exec.Command(exe, "--config", configPath, "serve") + cmd.Stdout = logFile + cmd.Stderr = logFile + cmd.Stdin = nil + cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true} + if err := cmd.Start(); err != nil { + logFile.Close() + return err + } + logFile.Close() + pid := cmd.Process.Pid + _ = cmd.Process.Release() + + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + pidOnDisk, err := readPID(paths.PID) + if err == nil && pidOnDisk == pid && processAlive(pid) { + fmt.Printf("Started (pid %d). Log: %s\n", pid, paths.Log) + return nil + } + if !processAlive(pid) { + break + } + time.Sleep(100 * time.Millisecond) + } + if processAlive(pid) { + fmt.Printf("Started (pid %d). Log: %s\n", pid, paths.Log) + return nil + } + tailLog(paths.Log, 20) + return errors.New("service exited during startup; see the log above") +} + +func runDown(configPath string) int { + paths := client.ConfigPaths(configPath) + pid, err := readPID(paths.PID) + if err != nil { + fmt.Println("Not running.") + return 0 + } + if !processAlive(pid) { + _ = os.Remove(paths.PID) + fmt.Println("Not running (removed stale pid file).") + return 0 + } + if err := syscall.Kill(pid, syscall.SIGTERM); err != nil { + fmt.Fprintf(os.Stderr, "stop pid %d: %v\n", pid, err) + return 1 + } + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if !processAlive(pid) { + _ = os.Remove(paths.PID) + fmt.Printf("Stopped (pid %d).\n", pid) + return 0 + } + time.Sleep(100 * time.Millisecond) + } + fmt.Fprintf(os.Stderr, "pid %d did not stop within 5s\n", pid) + return 1 +} + +func runServe(configPath string) int { + paths := client.ConfigPaths(configPath) + if pid, err := readPID(paths.PID); err == nil && pid != os.Getpid() && processAlive(pid) { + fmt.Fprintf(os.Stderr, "already running (pid %d)\n", pid) + return 1 + } + svc, err := client.NewService(configPath, version, log.New(os.Stderr, "", log.LstdFlags)) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + if err := writePID(paths.PID); err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + defer func() { + if pid, err := readPID(paths.PID); err == nil && pid == os.Getpid() { + _ = os.Remove(paths.PID) + } + }() + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + if err := svc.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { + svc.Log.Printf("service stopped: %v", err) + return 1 + } + return 0 +} + +func runStatus(configPath string) int { + cfg, err := client.Load(configPath) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + printConfig(configPath, cfg) + paths := client.ConfigPaths(configPath) + if pid, err := readPID(paths.PID); err == nil && processAlive(pid) { + fmt.Printf("Service: running (pid %d)\n", pid) + } else { + fmt.Println("Service: stopped") + } + if cfg.API == "" || cfg.TunnelID == "" || cfg.Token == "" { + return 0 + } + svc, err := client.NewService(configPath, version, nil) + if err != nil { + return 1 + } + ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) + defer cancel() + info, err := svc.API.GetTunnel(ctx, cfg.TunnelID, cfg.Token) + if err != nil { + fmt.Printf("Server: error: %v\n", err) + return 0 + } + fmt.Printf("Server: %s\n", info.State) + cert, err := svc.API.GetCertificate(ctx, cfg.TunnelID, cfg.Token) + if err == nil { + line := cert.State.Type + if cert.State.Type == "ready" { + line += " (" + cert.State.Expiry + ")" + } + if cert.State.Reason != "" { + line += ": " + cert.State.Reason + } + fmt.Printf("Cert state: %s\n", line) + } + return 0 +} + +func runLogs(configPath string) int { + paths := client.ConfigPaths(configPath) + if _, err := os.Stat(paths.Log); err != nil { + fmt.Println("No log yet:", paths.Log) + return 0 + } + fmt.Println("Log:", paths.Log) + tailLog(paths.Log, 100) + return 0 +} + +func sortedRouteNames(cfg *client.Config) []string { + names := make([]string, 0, len(cfg.Routes)) + for name := range cfg.Routes { + names = append(names, name) + } + sort.Strings(names) + return names +} + +func writePID(path string) error { + return os.WriteFile(path, []byte(strconv.Itoa(os.Getpid())+"\n"), 0o600) +} + +func readPID(path string) (int, error) { + data, err := os.ReadFile(path) + if err != nil { + return 0, err + } + pid, err := strconv.Atoi(strings.TrimSpace(string(data))) + if err != nil || pid <= 0 { + return 0, errors.New("invalid pid file") + } + return pid, nil +} + +func processAlive(pid int) bool { + err := syscall.Kill(pid, 0) + return err == nil || errors.Is(err, syscall.EPERM) +} + +func tailLog(path string, lines int) { + file, err := os.Open(path) + if err != nil { + return + } + defer file.Close() + data, err := io.ReadAll(io.LimitReader(file, 1<<20)) + if err != nil { + return + } + all := strings.Split(strings.TrimRight(string(data), "\n"), "\n") + if len(all) > lines { + all = all[len(all)-lines:] + } + for _, line := range all { + if line != "" { + fmt.Println(line) + } + } +} + +func randomRouteName() string { + buf := make([]byte, 10) + if _, err := rand.Read(buf); err != nil { + return "route" + } + return strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(buf)) +} + +func valueOr(value, fallback string) string { + if value == "" { + return fallback + } + return value +} diff --git a/cmd/tunneld/main.go b/cmd/tunneld/main.go index ebc11c3..059abac 100644 --- a/cmd/tunneld/main.go +++ b/cmd/tunneld/main.go @@ -69,6 +69,9 @@ func main() { porkbunSecret := flag.String("porkbun-secret-api-key", envOr("PORKBUN_SECRET_API_KEY", ""), "Porkbun secret API key") dnsZone := flag.String("dns-zone", envOr("DNS_ZONE", ""), "zone hosting the challenge records (default: derived from the domain)") challtestsrvURL := flag.String("challtestsrv-url", envOr("CHALLTESTSRV_URL", ""), "pebble-challtestsrv management URL for DNS_PROVIDER=challtestsrv") + createLimit := flag.Int("create-limit-per-hour", envInt("TUNNEL_CREATE_LIMIT_PER_HOUR", 60), "tunnels one client address may create per hour; 0 disables") + bindLimit := flag.Int("bind-limit-per-hour", envInt("CERT_BIND_LIMIT_PER_HOUR", 30), "new certificate binds one client address may request per hour; 0 disables") + bindTunnelLimit := flag.Int("bind-limit-per-tunnel-per-hour", envInt("CERT_BIND_LIMIT_PER_TUNNEL_PER_HOUR", 10), "new certificate binds one tunnel may request per hour; 0 disables") flag.Parse() if *caFile == "" { @@ -117,6 +120,16 @@ func main() { case *dnsProvider != "": log.Fatalf("unknown DNS_PROVIDER %q (want porkbun or challtestsrv)", *dnsProvider) } + if sweeper, ok := dns.(server.ChallengeSweeper); ok { + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + count, err := sweeper.SweepChallenges(ctx) + cancel() + if err != nil { + log.Printf("dns: stale challenge sweep removed %d record(s), then failed: %v", count, err) + } else if count > 0 { + log.Printf("dns: removed %d stale challenge record(s)", count) + } + } if dns != nil { acme, err := server.NewACMEIssuer(store, server.ACMEOptions{ DirectoryURL: *acmeDirectory, @@ -144,12 +157,19 @@ func main() { Version: version, Issuer: issuer, IssuerName: issuerName, + + CreateLimiter: server.NewRateLimiter(*createLimit, time.Hour), + BindLimiter: server.NewRateLimiter(*bindLimit, time.Hour), + BindTunnelLimiter: server.NewRateLimiter(*bindTunnelLimit, time.Hour), } if *adminToken == "" { log.Printf("admin dashboard: disabled (set ADMIN_TOKEN to enable it at /)") } else { log.Printf("admin dashboard: enabled at /") } + if *createLimit > 0 || *bindLimit > 0 || *bindTunnelLimit > 0 { + log.Printf("limits: %d creates/hour/IP, %d binds/hour/IP, %d binds/hour/tunnel", *createLimit, *bindLimit, *bindTunnelLimit) + } ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/internal/client/api.go b/internal/client/api.go new file mode 100644 index 0000000..16d25f2 --- /dev/null +++ b/internal/client/api.go @@ -0,0 +1,271 @@ +package client + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "math/rand" + "net/http" + "net/url" + "strconv" + "strings" + "time" +) + +// TunnelInfo is the API's tunnel record. +type TunnelInfo struct { + ID string `json:"id"` + Hostname string `json:"hostname"` + State string `json:"state"` + CertificateID string `json:"certificateID,omitempty"` +} + +// CertState is the API's certificate state union. +type CertState struct { + Type string `json:"type"` + Token string `json:"token,omitempty"` + Key string `json:"key,omitempty"` + Certificate string `json:"certificate,omitempty"` + Chain string `json:"chain,omitempty"` + Expiry string `json:"expiry,omitempty"` + Reason string `json:"reason,omitempty"` +} + +// CertificateInfo is the API's certificate envelope. +type CertificateInfo struct { + ID string `json:"id"` + State CertState `json:"state"` +} + +// APIError is a non-2xx API answer, including the contract's retry hints. +type APIError struct { + Status int + Tag string + Message string + RetryAfter time.Duration +} + +func (e *APIError) Error() string { + if e.Tag != "" { + return fmt.Sprintf("%s: %s", e.Tag, e.Message) + } + return fmt.Sprintf("api returned HTTP %d: %s", e.Status, e.Message) +} + +// APIClient talks to the tunnel server's HTTP API. +type APIClient struct { + Base string + HTTP *http.Client +} + +func NewAPIClient(base string) *APIClient { + return &APIClient{ + Base: strings.TrimRight(base, "/"), + HTTP: &http.Client{Timeout: 30 * time.Second}, + } +} + +type repeat int + +const ( + repeatIdempotent repeat = iota // GET, same-CSR bind: retry network errors and 5xx + repeatOnce // create/delete: only when the request cannot have arrived +) + +func (c *APIClient) do(ctx context.Context, method, path string, body any, token string, repeat repeat) ([]byte, error) { + var payload []byte + if body != nil { + var err error + payload, err = json.Marshal(body) + if err != nil { + return nil, err + } + } + for attempt := 0; ; attempt++ { + req, err := http.NewRequestWithContext(ctx, method, c.Base+path, bytes.NewReader(payload)) + if err != nil { + return nil, err + } + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } + resp, err := c.HTTP.Do(req) + if err != nil { + if attempt < 6 && repeat == repeatIdempotent && ctx.Err() == nil { + if !sleep(ctx, backoff(attempt, 0)) { + return nil, ctx.Err() + } + continue + } + return nil, err + } + data, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + resp.Body.Close() + if readErr != nil { + return nil, readErr + } + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return data, nil + } + apiErr := parseAPIError(resp, data) + if attempt < 6 && transient(resp.StatusCode, repeat) && ctx.Err() == nil { + if !sleep(ctx, backoff(attempt, apiErr.RetryAfter)) { + return nil, ctx.Err() + } + continue + } + return nil, apiErr + } +} + +func parseAPIError(resp *http.Response, data []byte) *APIError { + e := &APIError{Status: resp.StatusCode} + var body struct { + Tag string `json:"_tag"` + Message string `json:"message"` + } + if json.Unmarshal(data, &body) == nil { + e.Tag, e.Message = body.Tag, body.Message + } + if e.Message == "" { + e.Message = strings.TrimSpace(string(data)) + if e.Message == "" { + e.Message = http.StatusText(resp.StatusCode) + } + } + if after := resp.Header.Get("Retry-After"); after != "" { + if seconds, err := strconv.Atoi(after); err == nil { + e.RetryAfter = time.Duration(seconds) * time.Second + } + } + return e +} + +func transient(status int, repeat repeat) bool { + if status == 429 { + return true + } + if repeat == repeatIdempotent { + return status >= 500 && status <= 599 + } + switch status { + case 503, 521, 522, 523, 525: + return true + } + return false +} + +func backoff(attempt int, retryAfter time.Duration) time.Duration { + if retryAfter > 0 { + if retryAfter > 8*time.Second { + retryAfter = 8 * time.Second + } + return retryAfter + } + wait := 250 * time.Millisecond << attempt + if wait > 8*time.Second { + wait = 8 * time.Second + } + // Jitter between half and all of the delay. + half := int64(wait / 2) + return time.Duration(half + rand.Int63n(half+1)) +} + +func sleep(ctx context.Context, d time.Duration) bool { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} + +// CreateTunnel creates a new tunnel and returns its one-time token. +func (c *APIClient) CreateTunnel(ctx context.Context) (TunnelInfo, string, error) { + data, err := c.do(ctx, http.MethodPost, "/api/tunnel", struct{}{}, "", repeatOnce) + if err != nil { + return TunnelInfo{}, "", err + } + var out struct { + Tunnel TunnelInfo `json:"tunnel"` + Token string `json:"token"` + } + if err := json.Unmarshal(data, &out); err != nil { + return TunnelInfo{}, "", fmt.Errorf("create tunnel: %w", err) + } + if out.Tunnel.ID == "" || out.Token == "" { + return TunnelInfo{}, "", fmt.Errorf("create tunnel: incomplete response") + } + return out.Tunnel, out.Token, nil +} + +// GetTunnel reads a tunnel. +func (c *APIClient) GetTunnel(ctx context.Context, id, token string) (TunnelInfo, error) { + data, err := c.do(ctx, http.MethodGet, "/api/tunnel/"+url.PathEscape(id), nil, token, repeatIdempotent) + if err != nil { + return TunnelInfo{}, err + } + var out TunnelInfo + if err := json.Unmarshal(data, &out); err != nil { + return TunnelInfo{}, fmt.Errorf("get tunnel: %w", err) + } + return out, nil +} + +// BindCertificate submits the tunnel's CSR. Re-binding the same CSR is +// idempotent and is retried. +func (c *APIClient) BindCertificate(ctx context.Context, id, token, csr string) (CertificateInfo, error) { + data, err := c.do(ctx, http.MethodPost, "/api/tunnel/"+url.PathEscape(id)+"/certificate", + map[string]string{"csr": csr}, token, repeatIdempotent) + if err != nil { + return CertificateInfo{}, err + } + return decodeCertificate(data) +} + +// GetCertificate reads the current certificate state. +func (c *APIClient) GetCertificate(ctx context.Context, id, token string) (CertificateInfo, error) { + data, err := c.do(ctx, http.MethodGet, "/api/tunnel/"+url.PathEscape(id)+"/certificate", nil, token, repeatIdempotent) + if err != nil { + return CertificateInfo{}, err + } + return decodeCertificate(data) +} + +func decodeCertificate(data []byte) (CertificateInfo, error) { + var out CertificateInfo + if err := json.Unmarshal(data, &out); err != nil { + return CertificateInfo{}, fmt.Errorf("certificate response: %w", err) + } + return out, nil +} + +// DeleteTunnel deletes a tunnel. +func (c *APIClient) DeleteTunnel(ctx context.Context, id, token string) error { + _, err := c.do(ctx, http.MethodDelete, "/api/tunnel/"+url.PathEscape(id), nil, token, repeatOnce) + return err +} + +// WebSocketURL returns the bridge URL for a tunnel. +func (c *APIClient) WebSocketURL(id string) (string, error) { + u, err := url.Parse(c.Base) + if err != nil { + return "", err + } + if u.Scheme == "https" { + u.Scheme = "wss" + } else { + u.Scheme = "ws" + } + u.Path = "/api/tunnel/" + url.PathEscape(id) + "/connect" + u.RawQuery = "" + u.Fragment = "" + return u.String(), nil +} diff --git a/internal/client/bridge.go b/internal/client/bridge.go new file mode 100644 index 0000000..eae958b --- /dev/null +++ b/internal/client/bridge.go @@ -0,0 +1,648 @@ +package client + +import ( + "context" + "crypto/tls" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net" + "net/http" + "sort" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/coder/websocket" +) + +const ( + bridgeSubprotocol = "opentunnel" + maxPayload = 32768 + maxConnsPerBridge = 256 +) + +var ( + errFatalAttach = errors.New("fatal attach error") + errBackpressure = errors.New("bridge data queue full") +) + +type attachOut struct { + Type string `json:"type"` + Token string `json:"token"` + Transport string `json:"transport"` + Routes []string `json:"routes"` + Client struct { + Version string `json:"version"` + MaxConns int `json:"max_conns"` + } `json:"client"` +} + +type serverControl struct { + Type string `json:"type"` + Session string `json:"session,omitempty"` + Routes []string `json:"routes,omitempty"` + HeartbeatMs int `json:"heartbeat_ms,omitempty"` + IdleTimeoutMs int `json:"idle_timeout_ms,omitempty"` + Conn *uint32 `json:"conn,omitempty"` + Peer string `json:"peer,omitempty"` + SNI string `json:"sni,omitempty"` + ALPN string `json:"alpn,omitempty"` + Code string `json:"code,omitempty"` + TimeSent int64 `json:"time_sent,omitempty"` + Reason string `json:"reason,omitempty"` +} + +type outbound struct { + typ websocket.MessageType + data []byte +} + +type session struct { + svc *Service + cfg *Config + log *log.Logger + ws *websocket.Conn + routes map[string]Route + heartbeat time.Duration + idle time.Duration + + control chan outbound + data chan outbound + done chan struct{} + + closeOnce sync.Once + mu sync.Mutex + conns map[uint32]*bridgeConn + lastRecv atomic.Int64 +} + +func (s *Service) connect(ctx context.Context, cfg *Config) (*session, error) { + wsURL, err := s.API.WebSocketURL(cfg.TunnelID) + if err != nil { + return nil, err + } + dialCtx, cancel := context.WithTimeout(ctx, 15*time.Second) + ws, _, err := websocket.Dial(dialCtx, wsURL, &websocket.DialOptions{ + HTTPHeader: http.Header{"Authorization": []string{"Bearer " + cfg.Token}}, + Subprotocols: []string{bridgeSubprotocol}, + CompressionMode: websocket.CompressionDisabled, + }) + cancel() + if err != nil { + return nil, fmt.Errorf("dial bridge: %w", err) + } + ws.SetReadLimit(64 * 1024) + + routes := make([]string, 0, len(cfg.Routes)) + targets := make(map[string]Route, len(cfg.Routes)) + for name, route := range cfg.Routes { + routes = append(routes, name) + targets[name] = route + } + sort.Strings(routes) + if len(routes) > 128 { + ws.Close(websocket.StatusPolicyViolation, "too many routes") + return nil, fmt.Errorf("too many routes (%d, max 128)", len(routes)) + } + + var attach attachOut + attach.Type = "attach" + attach.Token = cfg.Token + attach.Transport = "ws" + attach.Routes = routes + attach.Client.Version = s.Version + attach.Client.MaxConns = maxConnsPerBridge + if err := writeJSON(ctx, ws, attach); err != nil { + ws.Close(websocket.StatusInternalError, "attach write") + return nil, err + } + + readCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + typ, raw, err := ws.Read(readCtx) + cancel() + if err != nil { + ws.Close(websocket.StatusInternalError, "attach read") + return nil, fmt.Errorf("attach: %w", err) + } + if typ != websocket.MessageText { + ws.Close(websocket.StatusProtocolError, "expected attach answer") + return nil, errors.New("attach: unexpected binary answer") + } + var answer serverControl + if err := json.Unmarshal(raw, &answer); err != nil { + ws.Close(websocket.StatusProtocolError, "bad attach answer") + return nil, fmt.Errorf("attach answer: %w", err) + } + switch answer.Type { + case "attached": + case "attach_error": + ws.Close(websocket.StatusPolicyViolation, answer.Code) + if answer.Code == "bad_token" || answer.Code == "cert_not_ready" { + return nil, fmt.Errorf("%w: %s", errFatalAttach, answer.Code) + } + return nil, fmt.Errorf("attach: %s", answer.Code) + default: + ws.Close(websocket.StatusProtocolError, "expected attached") + return nil, fmt.Errorf("attach: unexpected message %q", answer.Type) + } + ws.SetReadLimit(4 + maxPayload) + + heartbeat := time.Duration(answer.HeartbeatMs) * time.Millisecond + if heartbeat <= 0 { + heartbeat = 15 * time.Second + } + idle := time.Duration(answer.IdleTimeoutMs) * time.Millisecond + if idle <= 0 { + idle = 45 * time.Second + } + sess := &session{ + svc: s, + cfg: cfg, + log: s.Log, + ws: ws, + routes: targets, + heartbeat: heartbeat, + idle: idle, + control: make(chan outbound, 64), + data: make(chan outbound, 256), + done: make(chan struct{}), + conns: map[uint32]*bridgeConn{}, + } + sess.touch() + go sess.writeLoop() + go sess.heartbeatLoop() + go sess.readLoop() + s.Log.Printf("attached: session %s, %d route(s), heartbeat %s", answer.Session, len(routes), heartbeat) + return sess, nil +} + +func writeJSON(ctx context.Context, ws *websocket.Conn, value any) error { + data, err := json.Marshal(value) + if err != nil { + return err + } + writeCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + return ws.Write(writeCtx, websocket.MessageText, data) +} + +func (s *session) touch() { s.lastRecv.Store(time.Now().UnixMilli()) } + +func (s *session) readLoop() { + ctx := context.Background() + for { + typ, raw, err := s.ws.Read(ctx) + if err != nil { + s.close("read: " + err.Error()) + return + } + s.touch() + if typ == websocket.MessageText { + s.handleControl(raw) + continue + } + if len(raw) < 4 { + continue + } + id := binary.BigEndian.Uint32(raw[:4]) + s.mu.Lock() + bc := s.conns[id] + s.mu.Unlock() + if bc == nil { + continue + } + payload := raw[4:] + select { + case bc.incoming <- payload: + case <-bc.done: + default: + // The local side is not draining; drop this connection rather + // than blocking the bridge read loop. + s.resetConn(id, "upstream_io_error") + } + } +} + +func (s *session) handleControl(raw []byte) { + var msg serverControl + if err := json.Unmarshal(raw, &msg); err != nil { + return + } + switch msg.Type { + case "ping": + s.sendControl(serverControl{Type: "pong", TimeSent: msg.TimeSent}) + case "pong": + // Any received message already refreshed lastRecv. + case "open": + if msg.Conn != nil { + s.handleOpen(*msg.Conn, msg.Peer, msg.SNI, msg.ALPN) + } + case "end": + if msg.Conn != nil { + s.endFromServer(*msg.Conn) + } + case "reset": + if msg.Conn != nil { + s.resetFromServer(*msg.Conn) + } + case "drain": + s.log.Printf("server drained the bridge: %s", msg.Reason) + s.close("server drain") + case "attached", "attach_error": + // Handled during connect. + } +} + +func (s *session) handleOpen(id uint32, peer, sni, alpn string) { + route := routeForSNI(s.cfg.Hostname, sni) + if route == "" { + s.sendReset(id, "unknown_route") + return + } + target, ok := s.routes[route] + if !ok { + s.sendReset(id, "unknown_route") + return + } + s.mu.Lock() + if len(s.conns) >= maxConnsPerBridge { + s.mu.Unlock() + s.sendReset(id, "too_many_connections") + return + } + bc := newBridgeConn(id, s) + bc.peer, bc.sni, bc.alpn = peer, sni, alpn + s.conns[id] = bc + s.mu.Unlock() + go s.serveConn(bc, target, peer, sni, alpn) +} + +func (s *session) serveConn(bc *bridgeConn, route Route, peer, sni, alpn string) { + defer func() { + s.mu.Lock() + if s.conns[bc.id] == bc { + delete(s.conns, bc.id) + } + s.mu.Unlock() + bc.fullClose() + }() + + tlsConn := tls.Server(bc, s.svc.TLSConfig()) + handshakeCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + err := tlsConn.HandshakeContext(handshakeCtx) + cancel() + if err != nil { + s.log.Printf("tls handshake for %s: %v", sni, err) + s.sendReset(bc.id, "upstream_io_error") + return + } + + target, err := net.DialTimeout("tcp", route.Target, 10*time.Second) + if err != nil { + s.log.Printf("connect %s: %v", route.Target, err) + s.sendReset(bc.id, "upstream_connect_failed") + return + } + defer target.Close() + if route.ProxyProtocol != "" { + if err := writeProxyHeader(target, route.ProxyProtocol, peer, sni); err != nil { + s.log.Printf("proxy protocol for %s: %v", sni, err) + s.sendReset(bc.id, "upstream_io_error") + return + } + } + + // visitor -> target. On TLS EOF, half-close the target. + visitorDone := make(chan error, 1) + go func() { + _, err := io.Copy(target, tlsConn) + if tcp, ok := target.(*net.TCPConn); ok { + tcp.CloseWrite() + } + visitorDone <- err + }() + // target -> visitor. When the target ends, send TLS close_notify and an + // `end` control so the server half-closes the visitor. + targetDone := make(chan error, 1) + go func() { + _, err := io.Copy(tlsConn, target) + _ = tlsConn.CloseWrite() + s.sendEnd(bc.id) + targetDone <- err + }() + + var firstErr error + for done := 0; done < 2; { + select { + case err := <-visitorDone: + done++ + if err != nil && !isClosedError(err) { + firstErr = err + } + visitorDone = nil + case err := <-targetDone: + done++ + if err != nil && !isClosedError(err) { + firstErr = err + } + targetDone = nil + case <-bc.done: + s.sendReset(bc.id, "io_error") + return + case <-time.After(60 * time.Second): + s.sendReset(bc.id, "idle_timeout") + return + } + } + if firstErr != nil { + s.sendReset(bc.id, "upstream_io_error") + } +} + +func (s *session) endFromServer(id uint32) { + s.mu.Lock() + bc := s.conns[id] + s.mu.Unlock() + if bc != nil { + bc.remoteEnd() + } +} + +func (s *session) resetFromServer(id uint32) { + s.mu.Lock() + bc := s.conns[id] + delete(s.conns, id) + s.mu.Unlock() + if bc != nil { + bc.remoteReset() + } +} + +func (s *session) resetConn(id uint32, code string) { + s.mu.Lock() + bc := s.conns[id] + delete(s.conns, id) + s.mu.Unlock() + s.sendReset(id, code) + if bc != nil { + bc.remoteReset() + } +} + +func (s *session) sendReset(id uint32, code string) { + s.sendControl(serverControl{Type: "reset", Conn: &id, Code: code}) +} + +func (s *session) sendControl(msg serverControl) { + data, err := json.Marshal(msg) + if err != nil { + return + } + select { + case s.control <- outbound{typ: websocket.MessageText, data: data}: + case <-s.done: + default: + s.close("control queue full") + } +} + +func (s *session) sendData(frame []byte) error { + select { + case <-s.done: + return io.ErrClosedPipe + default: + } + select { + case s.data <- outbound{typ: websocket.MessageBinary, data: frame}: + return nil + case <-s.done: + return io.ErrClosedPipe + default: + return errBackpressure + } +} + +// sendEnd queues a half-close after any already-queued data frames. It must +// not use the priority control queue: an `end` overtaking the final response +// bytes truncates the visitor's connection. +func (s *session) sendEnd(id uint32) { + data, err := json.Marshal(serverControl{Type: "end", Conn: &id}) + if err != nil { + return + } + select { + case s.data <- outbound{typ: websocket.MessageText, data: data}: + case <-s.done: + default: + s.close("data queue full") + } +} + +func (s *session) writeLoop() { + for { + // Give control messages priority over queued visitor data. + select { + case <-s.done: + return + default: + } + select { + case <-s.done: + return + case msg := <-s.control: + if err := s.write(msg); err != nil { + s.close("write: " + err.Error()) + return + } + continue + default: + } + select { + case <-s.done: + return + case msg := <-s.control: + if err := s.write(msg); err != nil { + s.close("write: " + err.Error()) + return + } + case frame := <-s.data: + if err := s.write(frame); err != nil { + s.close("write: " + err.Error()) + return + } + } + } +} + +func (s *session) write(msg outbound) error { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + return s.ws.Write(ctx, msg.typ, msg.data) +} + +func (s *session) heartbeatLoop() { + ticker := time.NewTicker(s.heartbeat) + defer ticker.Stop() + for { + select { + case <-s.done: + return + case now := <-ticker.C: + if now.UnixMilli()-s.lastRecv.Load() > s.idle.Milliseconds() { + s.close("idle timeout") + return + } + s.sendControl(serverControl{Type: "ping", TimeSent: now.UnixMilli()}) + } + } +} + +func (s *session) close(reason string) { + s.closeOnce.Do(func() { + close(s.done) + s.ws.Close(websocket.StatusGoingAway, reason) + s.mu.Lock() + conns := make([]*bridgeConn, 0, len(s.conns)) + for _, bc := range s.conns { + conns = append(conns, bc) + } + s.conns = map[uint32]*bridgeConn{} + s.mu.Unlock() + for _, bc := range conns { + bc.remoteReset() + } + }) +} + +// bridgeConn is a net.Conn fed by bridge frames. +type bridgeConn struct { + id uint32 + session *session + incoming chan []byte + buf []byte + peer string + sni string + alpn string + + mu sync.Mutex + readErr error + readEnd chan struct{} + done chan struct{} + once sync.Once +} + +func newBridgeConn(id uint32, s *session) *bridgeConn { + return &bridgeConn{ + id: id, + session: s, + incoming: make(chan []byte, 64), + readEnd: make(chan struct{}), + done: make(chan struct{}), + } +} + +func (c *bridgeConn) Read(p []byte) (int, error) { + for len(c.buf) == 0 { + select { + case buf := <-c.incoming: + c.buf = buf + case <-c.readEnd: + c.mu.Lock() + err := c.readErr + c.mu.Unlock() + if err != nil { + return 0, err + } + return 0, io.EOF + } + } + n := copy(p, c.buf) + c.buf = c.buf[n:] + return n, nil +} + +func (c *bridgeConn) Write(p []byte) (int, error) { + select { + case <-c.done: + return 0, io.ErrClosedPipe + default: + } + frame := make([]byte, 4+len(p)) + binary.BigEndian.PutUint32(frame, c.id) + copy(frame[4:], p) + if err := c.session.sendData(frame); err != nil { + c.session.sendReset(c.id, "upstream_io_error") + return 0, err + } + return len(p), nil +} + +func (c *bridgeConn) Close() error { + c.remoteReset() + return nil +} + +func (c *bridgeConn) LocalAddr() net.Addr { return bridgeAddr("bridge") } +func (c *bridgeConn) RemoteAddr() net.Addr { return bridgeAddr(c.peer) } +func (c *bridgeConn) SetDeadline(time.Time) error { return nil } +func (c *bridgeConn) SetReadDeadline(time.Time) error { return nil } +func (c *bridgeConn) SetWriteDeadline(time.Time) error { return nil } + +func (c *bridgeConn) remoteEnd() { + c.mu.Lock() + select { + case <-c.readEnd: + default: + close(c.readEnd) + } + c.mu.Unlock() +} + +func (c *bridgeConn) remoteReset() { + c.mu.Lock() + if c.readErr == nil { + c.readErr = io.ErrClosedPipe + } + select { + case <-c.readEnd: + default: + close(c.readEnd) + } + c.mu.Unlock() + c.once.Do(func() { close(c.done) }) +} + +func (c *bridgeConn) fullClose() { + c.remoteReset() +} + +type bridgeAddr string + +func (a bridgeAddr) Network() string { return "bridge" } +func (a bridgeAddr) String() string { return string(a) } + +func routeForSNI(hostname, sni string) string { + sni = strings.ToLower(strings.TrimSuffix(sni, ".")) + hostname = strings.ToLower(hostname) + if sni == hostname { + return "@" + } + suffix := "." + hostname + if !strings.HasSuffix(sni, suffix) { + return "" + } + prefix := strings.TrimSuffix(sni, suffix) + if strings.Contains(prefix, ".") || !ValidRouteName(prefix) { + return "" + } + return prefix +} + +func isClosedError(err error) bool { + return err == nil || errors.Is(err, io.EOF) || errors.Is(err, io.ErrClosedPipe) || errors.Is(err, net.ErrClosed) +} diff --git a/internal/client/client_test.go b/internal/client/client_test.go new file mode 100644 index 0000000..c955b28 --- /dev/null +++ b/internal/client/client_test.go @@ -0,0 +1,211 @@ +package client + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func TestNormalizeAPI(t *testing.T) { + cases := map[string]string{ + "tunnel.tzgyn.dev": "https://tunnel.tzgyn.dev", + "https://tunnel.tzgyn.dev/": "https://tunnel.tzgyn.dev", + "http://127.0.0.1:8080": "http://127.0.0.1:8080", + } + for input, want := range cases { + got, err := NormalizeAPI(input) + if err != nil || got != want { + t.Errorf("NormalizeAPI(%q) = %q, %v; want %q", input, got, err, want) + } + } + if _, err := NormalizeAPI("ftp://example.com"); err == nil { + t.Error("ftp should be rejected") + } +} + +func TestNormalizeTarget(t *testing.T) { + cases := map[string]string{ + "3000": "127.0.0.1:3000", + ":3000": "127.0.0.1:3000", + "127.0.0.1:4096": "127.0.0.1:4096", + "localhost:8080": "localhost:8080", + "[::1]:3000": "[::1]:3000", + } + for input, want := range cases { + got, err := NormalizeTarget(input) + if err != nil || got != want { + t.Errorf("NormalizeTarget(%q) = %q, %v; want %q", input, got, err, want) + } + } + for _, bad := range []string{"", "http://x", "localhost", "127.0.0.1:0", "127.0.0.1:99999"} { + if _, err := NormalizeTarget(bad); err == nil { + t.Errorf("NormalizeTarget(%q) should fail", bad) + } + } +} + +func TestConfigRoundTrip(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "config.json") + cfg := &Config{API: "https://tunnel.example", Routes: map[string]Route{}} + if err := cfg.AddRoute("api", "4096", ""); err != nil { + t.Fatal(err) + } + if err := cfg.Save(path); err != nil { + t.Fatal(err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm() != 0o600 { + t.Errorf("mode = %o, want 600", info.Mode().Perm()) + } + loaded, err := Load(path) + if err != nil { + t.Fatal(err) + } + if loaded.API != cfg.API || loaded.Routes["api"].Target != "127.0.0.1:4096" { + t.Errorf("round trip: %+v", loaded) + } +} + +func TestEnsureIdentityReusesCSR(t *testing.T) { + paths := ConfigPaths(filepath.Join(t.TempDir(), "config.json")) + key1, csr1, err := EnsureIdentity(paths, "abc.tunnel.example") + if err != nil { + t.Fatal(err) + } + key2, csr2, err := EnsureIdentity(paths, "abc.tunnel.example") + if err != nil { + t.Fatal(err) + } + if string(key1) != string(key2) || string(csr1) != string(csr2) { + t.Error("identity was regenerated on the second call") + } + if !csrMatches(csr1, "abc.tunnel.example") { + t.Error("CSR does not match the hostname") + } +} + +func TestRouteForSNI(t *testing.T) { + host := "abc.tunnel.example" + cases := map[string]string{ + host: "@", + "api." + host: "api", + "API." + host: "api", + "api." + host + ".": "api", + "a.b." + host: "", + "other.example": "", + "api.abc.tunnel.example.x": "", + } + for sni, want := range cases { + if got := routeForSNI(host, sni); got != want { + t.Errorf("routeForSNI(%q) = %q, want %q", sni, got, want) + } + } +} + +func TestReadOfficialRoutes(t *testing.T) { + path := filepath.Join(t.TempDir(), "coolify.toml") + body := `[routes] +api = "127.0.0.1:4096" +web = { target = "localhost:3000", proxy_protocol = "v2" } +` + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatal(err) + } + routes, err := readOfficialRoutes(path) + if err != nil { + t.Fatal(err) + } + if routes["api"].Target != "127.0.0.1:4096" { + t.Errorf("api = %+v", routes["api"]) + } + if routes["web"].Target != "localhost:3000" || routes["web"].ProxyProtocol != "v2" { + t.Errorf("web = %+v", routes["web"]) + } +} + +func TestImportOfficialProfile(t *testing.T) { + root := t.TempDir() + dataHome := filepath.Join(root, "data") + configHome := filepath.Join(root, "config") + t.Setenv("XDG_DATA_HOME", dataHome) + t.Setenv("XDG_CONFIG_HOME", configHome) + + sourceDir := filepath.Join(dataHome, "opentunnel", "coolify") + if err := os.MkdirAll(sourceDir, 0o700); err != nil { + t.Fatal(err) + } + officialPaths := ConfigPaths(filepath.Join(root, "official", "config.json")) + keyPEM, _, err := EnsureIdentity(officialPaths, "abc.tunnel.example") + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(sourceDir, "tunnel.json"), []byte(`{"id":"abc","hostname":"abc.tunnel.example","certificateExpiry":"2027-01-01T00:00:00Z"}`), 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(sourceDir, "token"), []byte("tok\n"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(sourceDir, "private-key.pem"), keyPEM, 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(sourceDir, "certificate.pem"), []byte("dummy"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(sourceDir, "chain.pem"), []byte("dummy"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.MkdirAll(filepath.Join(configHome, "opentunnel"), 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(configHome, "opentunnel", "coolify.toml"), []byte("[routes]\napi = \"127.0.0.1:4096\"\n"), 0o600); err != nil { + t.Fatal(err) + } + + target := filepath.Join(root, "client", "config.json") + result, err := Import(ImportOptions{Profile: "coolify", API: "https://tunnel.example", ConfigPath: target}) + if err != nil { + t.Fatal(err) + } + if result.TunnelID != "abc" || result.Hostname != "abc.tunnel.example" { + t.Errorf("result = %+v", result) + } + cfg, err := Load(target) + if err != nil { + t.Fatal(err) + } + if cfg.Token != "tok" || cfg.API != "https://tunnel.example" || cfg.Routes["api"].Target != "127.0.0.1:4096" { + t.Errorf("config = %+v", cfg) + } + if _, err := os.Stat(ConfigPaths(target).Key); err != nil { + t.Errorf("key not copied: %v", err) + } +} + +func TestProxyHeaders(t *testing.T) { + if got := string(proxyV1("203.0.113.9:51234")); got != "PROXY TCP4 203.0.113.9 0.0.0.0 51234 443\r\n" { + t.Errorf("v1 = %q", got) + } + if got := string(proxyV1("nonsense")); got != "PROXY UNKNOWN\r\n" { + t.Errorf("v1 unknown = %q", got) + } + v2 := proxyV2("203.0.113.9:51234", "api.example") + sig := "\r\n\r\n\x00\r\nQUIT\n" + if !strings.HasPrefix(string(v2), sig) { + t.Fatalf("v2 signature missing") + } + if v2[12] != 0x21 || v2[13] != 0x11 { + t.Errorf("v2 version/family = %x %x", v2[12], v2[13]) + } + if !strings.HasSuffix(string(v2), "api.example") { + t.Errorf("v2 TLV missing") + } + unknown := proxyV2("nonsense", "") + if unknown[12] != 0x20 { + t.Errorf("v2 local command = %x", unknown[12]) + } +} diff --git a/internal/client/config.go b/internal/client/config.go new file mode 100644 index 0000000..0d5bd0a --- /dev/null +++ b/internal/client/config.go @@ -0,0 +1,260 @@ +// Package client implements the OpenTunnel client side: provisioning a +// tunnel, fetching and renewing its certificate, and carrying visitor +// connections over the bridge. +package client + +import ( + "encoding/json" + "errors" + "fmt" + "net" + "net/url" + "os" + "path/filepath" + "regexp" + "strconv" + "strings" +) + +// Route is one local target for a tunnel route. +type Route struct { + Target string `json:"target"` + ProxyProtocol string `json:"proxy_protocol,omitempty"` // "", "v1", or "v2" +} + +// Config is the single JSON file the CLI owns. It holds the API setting so +// commands no longer need OPENTUNNEL_API. The file is 0600 because it also +// holds the tunnel token. +type Config struct { + API string `json:"api"` + TunnelID string `json:"tunnel_id,omitempty"` + Hostname string `json:"hostname,omitempty"` + Token string `json:"token,omitempty"` + CertificateExpiry string `json:"certificate_expiry,omitempty"` + Routes map[string]Route `json:"routes,omitempty"` +} + +// DefaultPath is ~/.config/tunnel/config.json (XDG_CONFIG_HOME is honored). +func DefaultPath() string { + if dir := os.Getenv("XDG_CONFIG_HOME"); dir != "" { + return filepath.Join(dir, "tunnel", "config.json") + } + home, err := os.UserHomeDir() + if err != nil { + return "config.json" + } + return filepath.Join(home, ".config", "tunnel", "config.json") +} + +// Load reads the config, returning an empty one when the file does not exist. +func Load(path string) (*Config, error) { + cfg := &Config{Routes: map[string]Route{}} + data, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return cfg, nil + } + if err != nil { + return nil, err + } + if err := json.Unmarshal(data, cfg); err != nil { + return nil, fmt.Errorf("config %s: %w", path, err) + } + if cfg.Routes == nil { + cfg.Routes = map[string]Route{} + } + if err := cfg.Normalize(); err != nil { + return nil, fmt.Errorf("config %s: %w", path, err) + } + return cfg, nil +} + +// Save writes the config atomically as 0600. +func (c *Config) Save(path string) error { + if err := c.Normalize(); err != nil { + return err + } + dir := filepath.Dir(path) + if err := os.MkdirAll(dir, 0o700); err != nil { + return err + } + if err := os.Chmod(dir, 0o700); err != nil { + return err + } + data, err := json.MarshalIndent(c, "", " ") + if err != nil { + return err + } + data = append(data, '\n') + tmp, err := os.CreateTemp(dir, ".config-*.tmp") + if err != nil { + return err + } + name := tmp.Name() + defer os.Remove(name) + if err := tmp.Chmod(0o600); err != nil { + tmp.Close() + return err + } + if _, err := tmp.Write(data); err != nil { + tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + return os.Rename(name, path) +} + +// Normalize trims and validates the API and every route. +func (c *Config) Normalize() error { + if c.API != "" { + api, err := NormalizeAPI(c.API) + if err != nil { + return err + } + c.API = api + } + if c.Routes == nil { + c.Routes = map[string]Route{} + } + for name, route := range c.Routes { + if !ValidRouteName(name) { + return fmt.Errorf("invalid route name %q", name) + } + target, err := NormalizeTarget(route.Target) + if err != nil { + return fmt.Errorf("route %q: %w", name, err) + } + route.Target = target + switch route.ProxyProtocol { + case "", "v1", "v2": + default: + return fmt.Errorf("route %q: invalid proxy_protocol %q", name, route.ProxyProtocol) + } + c.Routes[name] = route + } + return nil +} + +// NormalizeAPI accepts URL, host, or host:port and returns a scheme-qualified +// URL without a trailing slash. +func NormalizeAPI(raw string) (string, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return "", errors.New("api is required") + } + if !strings.Contains(raw, "://") { + raw = "https://" + raw + } + u, err := url.Parse(raw) + if err != nil || u.Host == "" { + return "", fmt.Errorf("invalid api URL %q", raw) + } + if u.Scheme != "http" && u.Scheme != "https" { + return "", fmt.Errorf("api scheme must be http or https, got %q", u.Scheme) + } + u.Path = strings.TrimRight(u.Path, "/") + u.Fragment = "" + u.RawQuery = "" + return u.String(), nil +} + +var routeNameRE = regexp.MustCompile(`^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$`) + +// ValidRouteName reports whether name is "@" or one lowercase DNS label. +func ValidRouteName(name string) bool { + return name == "@" || routeNameRE.MatchString(name) +} + +// NormalizeTarget turns a bare port into 127.0.0.1:port and validates the +// resulting host:port. +func NormalizeTarget(raw string) (string, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return "", errors.New("target is required") + } + if _, err := strconv.Atoi(raw); err == nil { + raw = "127.0.0.1:" + raw + } else if strings.HasPrefix(raw, ":") { + raw = "127.0.0.1" + raw + } + host, port, err := net.SplitHostPort(raw) + if err != nil { + return "", fmt.Errorf("target %q is not host:port or a port", raw) + } + if host == "" { + host = "127.0.0.1" + } + if n, err := strconv.Atoi(port); err != nil || n < 1 || n > 65535 { + return "", fmt.Errorf("target %q has an invalid port", raw) + } + return net.JoinHostPort(host, port), nil +} + +// AddRoute validates and stores a route. proxy may be empty, "v1", or "v2". +func (c *Config) AddRoute(name, target, proxy string) error { + if !ValidRouteName(name) { + return fmt.Errorf("invalid route name %q (use @ or one lowercase DNS label)", name) + } + normalized, err := NormalizeTarget(target) + if err != nil { + return err + } + switch proxy { + case "", "v1", "v2": + default: + return fmt.Errorf("invalid proxy protocol %q", proxy) + } + if c.Routes == nil { + c.Routes = map[string]Route{} + } + c.Routes[name] = Route{Target: normalized, ProxyProtocol: proxy} + return nil +} + +// RemoveRoute deletes a route, reporting whether it existed. +func (c *Config) RemoveRoute(name string) bool { + if _, ok := c.Routes[name]; !ok { + return false + } + delete(c.Routes, name) + return true +} + +// RouteURL is the public URL for a route on this tunnel. +func (c *Config) RouteURL(name string) string { + if c.Hostname == "" { + return "" + } + if name == "@" { + return "https://" + c.Hostname + } + return "https://" + name + "." + c.Hostname +} + +// Paths are the files stored next to the config. +type Paths struct { + Config string + Dir string + Key string + CSR string + Cert string + Chain string + PID string + Log string +} + +// ConfigPaths returns the standard file layout for a config path. +func ConfigPaths(configPath string) Paths { + dir := filepath.Dir(configPath) + return Paths{ + Config: configPath, + Dir: dir, + Key: filepath.Join(dir, "private-key.pem"), + CSR: filepath.Join(dir, "csr.pem"), + Cert: filepath.Join(dir, "certificate.pem"), + Chain: filepath.Join(dir, "chain.pem"), + PID: filepath.Join(dir, "tunnel.pid"), + Log: filepath.Join(dir, "tunnel.log"), + } +} diff --git a/internal/client/identity.go b/internal/client/identity.go new file mode 100644 index 0000000..a3049bc --- /dev/null +++ b/internal/client/identity.go @@ -0,0 +1,200 @@ +package client + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "errors" + "fmt" + "os" + "path/filepath" + "time" +) + +// ErrCertificateExpired means a cached certificate can no longer be served. +var ErrCertificateExpired = errors.New("certificate expired") + +// EnsureIdentity loads (or creates) the tunnel's private key and CSR. The CSR +// is reused byte-for-byte when it matches hostname, because binding a different +// CSR makes the server issue a new certificate. +func EnsureIdentity(paths Paths, hostname string) (keyPEM, csrPEM []byte, err error) { + if err := os.MkdirAll(paths.Dir, 0o700); err != nil { + return nil, nil, err + } + keyPEM, err = os.ReadFile(paths.Key) + if err != nil && !errors.Is(err, os.ErrNotExist) { + return nil, nil, err + } + var key *ecdsa.PrivateKey + if len(keyPEM) > 0 { + key, err = parseKey(keyPEM) + if err != nil { + return nil, nil, fmt.Errorf("private key: %w", err) + } + } + if key == nil { + key, err = ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return nil, nil, err + } + der, err := x509.MarshalPKCS8PrivateKey(key) + if err != nil { + return nil, nil, err + } + keyPEM = pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: der}) + if err := writeAtomic(paths.Key, keyPEM, 0o600); err != nil { + return nil, nil, err + } + } + + csrPEM, err = os.ReadFile(paths.CSR) + if err == nil && csrMatches(csrPEM, hostname) { + return keyPEM, csrPEM, nil + } + if err != nil && !errors.Is(err, os.ErrNotExist) { + return nil, nil, err + } + csrPEM, err = generateCSR(key, hostname) + if err != nil { + return nil, nil, err + } + if err := writeAtomic(paths.CSR, csrPEM, 0o600); err != nil { + return nil, nil, err + } + return keyPEM, csrPEM, nil +} + +func parseKey(keyPEM []byte) (*ecdsa.PrivateKey, error) { + block, _ := pem.Decode(keyPEM) + if block == nil { + return nil, errors.New("not a PEM private key") + } + parsed, err := x509.ParsePKCS8PrivateKey(block.Bytes) + if err != nil { + return nil, err + } + key, ok := parsed.(*ecdsa.PrivateKey) + if !ok { + return nil, errors.New("not an ECDSA private key") + } + if key.Curve != elliptic.P256() { + return nil, errors.New("not a P-256 private key") + } + return key, nil +} + +func generateCSR(key *ecdsa.PrivateKey, hostname string) ([]byte, error) { + request := &x509.CertificateRequest{ + Subject: pkix.Name{CommonName: hostname}, + DNSNames: []string{hostname, "*." + hostname}, + } + der, err := x509.CreateCertificateRequest(rand.Reader, request, key) + if err != nil { + return nil, err + } + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE REQUEST", Bytes: der}), nil +} + +func csrMatches(csrPEM []byte, hostname string) bool { + block, _ := pem.Decode(csrPEM) + if block == nil || block.Type != "CERTIFICATE REQUEST" { + return false + } + csr, err := x509.ParseCertificateRequest(block.Bytes) + if err != nil || csr.CheckSignature() != nil { + return false + } + if csr.Subject.CommonName != hostname { + return false + } + have := map[string]bool{} + for _, name := range csr.DNSNames { + have[name] = true + } + return have[hostname] && have["*."+hostname] +} + +// LoadCertificate loads cert.pem/chain.pem/private-key.pem and checks that the +// certificate still covers hostname and has not expired. +func LoadCertificate(paths Paths, hostname string) (tls.Certificate, time.Time, error) { + keyPEM, err := os.ReadFile(paths.Key) + if err != nil { + return tls.Certificate{}, time.Time{}, err + } + certPEM, err := os.ReadFile(paths.Cert) + if err != nil { + return tls.Certificate{}, time.Time{}, err + } + chainPEM, err := os.ReadFile(paths.Chain) + if err != nil && !errors.Is(err, os.ErrNotExist) { + return tls.Certificate{}, time.Time{}, err + } + return parseCertificate(append(certPEM, chainPEM...), keyPEM, hostname) +} + +// SaveCertificate validates and stores a certificate from the API. +func SaveCertificate(paths Paths, certPEM, chainPEM, keyPEM []byte, hostname string) (tls.Certificate, time.Time, error) { + fullchain := append(append([]byte{}, certPEM...), chainPEM...) + cert, expiry, err := parseCertificate(fullchain, keyPEM, hostname) + if err != nil { + return tls.Certificate{}, time.Time{}, err + } + if err := writeAtomic(paths.Cert, certPEM, 0o644); err != nil { + return tls.Certificate{}, time.Time{}, err + } + if err := writeAtomic(paths.Chain, chainPEM, 0o644); err != nil { + return tls.Certificate{}, time.Time{}, err + } + return cert, expiry, nil +} + +func parseCertificate(fullchainPEM, keyPEM []byte, hostname string) (tls.Certificate, time.Time, error) { + cert, err := tls.X509KeyPair(fullchainPEM, keyPEM) + if err != nil { + return tls.Certificate{}, time.Time{}, fmt.Errorf("certificate/key: %w", err) + } + leaf, err := x509.ParseCertificate(cert.Certificate[0]) + if err != nil { + return tls.Certificate{}, time.Time{}, err + } + if err := leaf.VerifyHostname(hostname); err != nil { + return tls.Certificate{}, time.Time{}, fmt.Errorf("certificate does not cover %s: %w", hostname, err) + } + if time.Now().After(leaf.NotAfter) { + return tls.Certificate{}, time.Time{}, ErrCertificateExpired + } + return cert, leaf.NotAfter, nil +} + +// CertificateExpiry parses the API's RFC 3339 expiry string. +func CertificateExpiry(value string) (time.Time, error) { + if value == "" { + return time.Time{}, nil + } + return time.Parse(time.RFC3339, value) +} + +func writeAtomic(path string, data []byte, mode os.FileMode) error { + tmp, err := os.CreateTemp(filepath.Dir(path), ".tunnel-*.tmp") + if err != nil { + return err + } + name := tmp.Name() + defer os.Remove(name) + if err := tmp.Chmod(mode); err != nil { + tmp.Close() + return err + } + if _, err := tmp.Write(data); err != nil { + tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + return os.Rename(name, path) +} diff --git a/internal/client/import.go b/internal/client/import.go new file mode 100644 index 0000000..ea7489e --- /dev/null +++ b/internal/client/import.go @@ -0,0 +1,248 @@ +package client + +import ( + "bufio" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "strconv" + "strings" +) + +// ImportOptions describes importing an official opentunnel CLI profile. +type ImportOptions struct { + Profile string + API string + ConfigPath string + Force bool +} + +// ImportResult summarizes a successful import. +type ImportResult struct { + TunnelID string + Hostname string + Routes []string + ConfigPath string + Source string +} + +type officialTunnel struct { + ID string `json:"id"` + Hostname string `json:"hostname"` + CertificateExpiry string `json:"certificateExpiry"` +} + +// Import copies the identity and routes of the official Rust CLI into this +// client's JSON config. The tunnel ID, token, private key and certificate are +// preserved, so the server sees the same tunnel. +func Import(opts ImportOptions) (ImportResult, error) { + if opts.Profile == "" { + opts.Profile = "default" + } + if !routeNameRE.MatchString(opts.Profile) { + return ImportResult{}, fmt.Errorf("invalid profile name %q", opts.Profile) + } + if opts.API == "" { + return ImportResult{}, errors.New("--api is required") + } + api, err := NormalizeAPI(opts.API) + if err != nil { + return ImportResult{}, err + } + if opts.ConfigPath == "" { + opts.ConfigPath = DefaultPath() + } + if _, err := os.Stat(opts.ConfigPath); err == nil && !opts.Force { + return ImportResult{}, fmt.Errorf("%s already exists; use --force to overwrite", opts.ConfigPath) + } + + dataDir := xdgDir("XDG_DATA_HOME", ".local/share", "opentunnel", opts.Profile) + configDir := xdgDir("XDG_CONFIG_HOME", ".config", "opentunnel") + routesPath := filepath.Join(configDir, opts.Profile+".toml") + + rawTunnel, err := os.ReadFile(filepath.Join(dataDir, "tunnel.json")) + if err != nil { + return ImportResult{}, fmt.Errorf("official tunnel.json: %w", err) + } + var official officialTunnel + if err := json.Unmarshal(rawTunnel, &official); err != nil { + return ImportResult{}, fmt.Errorf("official tunnel.json: %w", err) + } + if official.ID == "" || official.Hostname == "" { + return ImportResult{}, errors.New("official tunnel.json has no id/hostname") + } + token, err := os.ReadFile(filepath.Join(dataDir, "token")) + if err != nil { + return ImportResult{}, fmt.Errorf("official token: %w", err) + } + keyPEM, err := os.ReadFile(filepath.Join(dataDir, "private-key.pem")) + if err != nil { + return ImportResult{}, fmt.Errorf("official private key: %w", err) + } + if _, err := parseKey(keyPEM); err != nil { + return ImportResult{}, fmt.Errorf("official private key: %w", err) + } + certPEM, err := os.ReadFile(filepath.Join(dataDir, "certificate.pem")) + if err != nil { + return ImportResult{}, fmt.Errorf("official certificate: %w", err) + } + chainPEM, err := os.ReadFile(filepath.Join(dataDir, "chain.pem")) + if err != nil && !errors.Is(err, os.ErrNotExist) { + return ImportResult{}, fmt.Errorf("official chain: %w", err) + } + + paths := ConfigPaths(opts.ConfigPath) + if err := os.MkdirAll(paths.Dir, 0o700); err != nil { + return ImportResult{}, err + } + if err := writeAtomic(paths.Key, keyPEM, 0o600); err != nil { + return ImportResult{}, err + } + if err := writeAtomic(paths.Cert, certPEM, 0o644); err != nil { + return ImportResult{}, err + } + if err := writeAtomic(paths.Chain, chainPEM, 0o644); err != nil { + return ImportResult{}, err + } + if _, _, err := EnsureIdentity(paths, official.Hostname); err != nil { + return ImportResult{}, fmt.Errorf("local CSR: %w", err) + } + + cfg := &Config{ + API: api, + TunnelID: official.ID, + Hostname: official.Hostname, + Token: strings.TrimSpace(string(token)), + CertificateExpiry: official.CertificateExpiry, + Routes: map[string]Route{}, + } + routes, err := readOfficialRoutes(routesPath) + if err != nil { + return ImportResult{}, err + } + for name, route := range routes { + if err := cfg.AddRoute(name, route.Target, route.ProxyProtocol); err != nil { + return ImportResult{}, err + } + } + if cfg.Token == "" { + return ImportResult{}, errors.New("official token file is empty") + } + if err := cfg.Save(opts.ConfigPath); err != nil { + return ImportResult{}, err + } + names := make([]string, 0, len(cfg.Routes)) + for name := range cfg.Routes { + names = append(names, name) + } + return ImportResult{ + TunnelID: cfg.TunnelID, + Hostname: cfg.Hostname, + Routes: names, + ConfigPath: opts.ConfigPath, + Source: dataDir, + }, nil +} + +func xdgDir(env, fallback string, parts ...string) string { + base := os.Getenv(env) + if base == "" { + home, err := os.UserHomeDir() + if err != nil { + base = "." + } else { + base = filepath.Join(home, fallback) + } + } + return filepath.Join(append([]string{base}, parts...)...) +} + +// readOfficialRoutes parses the subset of TOML the official CLI writes: +// +// [routes] +// web = "127.0.0.1:3000" +// api = { target = "127.0.0.1:4000", proxy_protocol = "v2" } +func readOfficialRoutes(path string) (map[string]Route, error) { + out := map[string]Route{} + file, err := os.Open(path) + if errors.Is(err, os.ErrNotExist) { + return out, nil + } + if err != nil { + return nil, err + } + defer file.Close() + + inRoutes := false + scanner := bufio.NewScanner(file) + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + if strings.HasPrefix(line, "[") && strings.HasSuffix(line, "]") { + inRoutes = line == "[routes]" + continue + } + if !inRoutes { + continue + } + name, value, ok := strings.Cut(line, "=") + if !ok { + return nil, fmt.Errorf("%s: bad route line %q", path, line) + } + name = strings.TrimSpace(name) + if unquoted, err := strconv.Unquote(name); err == nil { + name = unquoted + } + value = strings.TrimSpace(value) + route, err := parseRouteValue(value) + if err != nil { + return nil, fmt.Errorf("%s: route %s: %w", path, name, err) + } + out[name] = route + } + if err := scanner.Err(); err != nil { + return nil, err + } + return out, nil +} + +func parseRouteValue(value string) (Route, error) { + if strings.HasPrefix(value, "\"") { + target, err := strconv.Unquote(value) + if err != nil { + return Route{}, err + } + return Route{Target: target}, nil + } + if strings.HasPrefix(value, "{") && strings.HasSuffix(value, "}") { + route := Route{} + body := strings.TrimSpace(value[1 : len(value)-1]) + for _, field := range strings.Split(body, ",") { + key, val, ok := strings.Cut(strings.TrimSpace(field), "=") + if !ok { + continue + } + key = strings.TrimSpace(key) + val = strings.TrimSpace(val) + unquoted, err := strconv.Unquote(val) + if err != nil { + return Route{}, err + } + switch key { + case "target": + route.Target = unquoted + case "proxy_protocol": + route.ProxyProtocol = unquoted + } + } + if route.Target == "" { + return Route{}, errors.New("inline route has no target") + } + return route, nil + } + return Route{}, fmt.Errorf("unsupported route value %q", value) +} diff --git a/internal/client/proxyprotocol.go b/internal/client/proxyprotocol.go new file mode 100644 index 0000000..20609bb --- /dev/null +++ b/internal/client/proxyprotocol.go @@ -0,0 +1,92 @@ +package client + +import ( + "fmt" + "io" + "net" + "strconv" +) + +// writeProxyHeader writes the optional PROXY protocol header to a local +// target. The destination is the unspecified address of the source family on +// port 443, matching the reference clients (the real public address is not +// known to the client). +func writeProxyHeader(w io.Writer, version, peer, sni string) error { + switch version { + case "v1": + _, err := w.Write(proxyV1(peer)) + return err + case "v2": + _, err := w.Write(proxyV2(peer, sni)) + return err + default: + return fmt.Errorf("unknown proxy protocol %q", version) + } +} + +func proxyV1(peer string) []byte { + ip, port, ok := parsePeer(peer) + if !ok { + return []byte("PROXY UNKNOWN\r\n") + } + if v4 := ip.To4(); v4 != nil { + return []byte(fmt.Sprintf("PROXY TCP4 %s 0.0.0.0 %d 443\r\n", v4, port)) + } + return []byte(fmt.Sprintf("PROXY TCP6 %s :: %d 443\r\n", ip, port)) +} + +func proxyV2(peer, sni string) []byte { + signature := []byte("\r\n\r\n\x00\r\nQUIT\n") + ip, port, ok := parsePeer(peer) + + command := byte(0x21) // PROXY + family := byte(0x00) // UNSPEC + var address []byte + if ok { + if v4 := ip.To4(); v4 != nil { + family = 0x11 // TCP4 + address = append(address, v4...) + address = append(address, 0, 0, 0, 0) + } else { + family = 0x21 // TCP6 + address = append(address, ip.To16()...) + address = append(address, make([]byte, 16)...) + } + address = append(address, byte(port>>8), byte(port), 0x01, 0xbb) // destination port 443 + } else { + command = 0x20 // LOCAL + } + + var tlv []byte + if sni != "" && len(sni) <= 255 { + tlv = []byte{0x02, 0x00, byte(len(sni))} + tlv = append(tlv, sni...) + } + length := len(address) + len(tlv) + + header := make([]byte, 0, len(signature)+4+length) + header = append(header, signature...) + header = append(header, command, family, byte(length>>8), byte(length)) + header = append(header, address...) + header = append(header, tlv...) + return header +} + +func parsePeer(peer string) (net.IP, int, bool) { + host, portText, err := net.SplitHostPort(peer) + if err != nil { + host, portText = peer, "0" + } + ip := net.ParseIP(host) + if ip == nil { + return nil, 0, false + } + if v4 := ip.To4(); v4 != nil { + ip = v4 + } + port, _ := strconv.Atoi(portText) + if port < 0 || port > 65535 { + port = 0 + } + return ip, port, true +} diff --git a/internal/client/service.go b/internal/client/service.go new file mode 100644 index 0000000..d23478d --- /dev/null +++ b/internal/client/service.go @@ -0,0 +1,416 @@ +package client + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "log" + "os" + "sync" + "sync/atomic" + "time" +) + +// Service is one provisioning + serving lifecycle for a config file. +type Service struct { + ConfigPath string + Paths Paths + Config *Config + API *APIClient + Log *log.Logger + Version string + + cert atomic.Pointer[tls.Certificate] + + mu sync.RWMutex // guards Config and API during the renewal poller +} + +// NewService loads a config (which may not exist yet). +func NewService(configPath, version string, logger *log.Logger) (*Service, error) { + cfg, err := Load(configPath) + if err != nil { + return nil, err + } + if logger == nil { + logger = log.New(os.Stderr, "", log.LstdFlags) + } + s := &Service{ + ConfigPath: configPath, + Paths: ConfigPaths(configPath), + Config: cfg, + Log: logger, + Version: version, + } + if cfg.API != "" { + s.API = NewAPIClient(cfg.API) + } + return s, nil +} + +// Ensure creates the tunnel if needed, makes the local key/CSR, and makes +// sure a usable certificate is cached. +func (s *Service) Ensure(ctx context.Context) error { + cfg, err := Load(s.ConfigPath) + if err != nil { + return err + } + if cfg.API == "" { + return errors.New("no API configured; run: tunnel config set-api https://your-server") + } + s.Paths = ConfigPaths(s.ConfigPath) + s.mu.Lock() + s.Config = cfg + s.API = NewAPIClient(cfg.API) + s.mu.Unlock() + + if cfg.TunnelID == "" || cfg.Token == "" { + ctx, cancel := context.WithTimeout(ctx, 60*time.Second) + tunnel, token, err := s.API.CreateTunnel(ctx) + cancel() + if err != nil { + return fmt.Errorf("create tunnel: %w", err) + } + cfg.TunnelID, cfg.Hostname, cfg.Token = tunnel.ID, tunnel.Hostname, token + // Persist the identity before binding, so an interrupted provision + // can resume with the same token and hostname. + if err := cfg.Save(s.ConfigPath); err != nil { + return err + } + s.Log.Printf("created tunnel %s (%s)", cfg.TunnelID, cfg.Hostname) + } + + info, err := s.API.GetTunnel(ctx, cfg.TunnelID, cfg.Token) + if err != nil { + var apiErr *APIError + if errors.As(err, &apiErr) && apiErr.Status == 404 { + return fmt.Errorf("tunnel %s no longer exists on the server", cfg.TunnelID) + } + if errors.As(err, &apiErr) && apiErr.Status == 401 { + return fmt.Errorf("the server rejected the saved token for tunnel %s", cfg.TunnelID) + } + return fmt.Errorf("get tunnel: %w", err) + } + if info.Hostname != "" && info.Hostname != cfg.Hostname { + cfg.Hostname = info.Hostname + if err := cfg.Save(s.ConfigPath); err != nil { + return err + } + } + if cfg.Hostname == "" { + return errors.New("server did not return a tunnel hostname") + } + + keyPEM, csrPEM, err := EnsureIdentity(s.Paths, cfg.Hostname) + if err != nil { + return err + } + return s.ensureCertificate(ctx, keyPEM, csrPEM) +} + +func (s *Service) ensureCertificate(ctx context.Context, keyPEM, csrPEM []byte) error { + cfg := s.Config + localCert, localExpiry, localErr := LoadCertificate(s.Paths, cfg.Hostname) + + info, err := s.API.GetCertificate(ctx, cfg.TunnelID, cfg.Token) + if err != nil { + var apiErr *APIError + if errors.As(err, &apiErr) && apiErr.Status == 404 { + return s.bindAndWait(ctx, keyPEM, csrPEM) + } + if localErr == nil { + s.Log.Printf("certificate check failed (%v); using saved certificate expiring %s", err, localExpiry.Format(time.RFC3339)) + s.cert.Store(&localCert) + return nil + } + return fmt.Errorf("get certificate: %w", err) + } + + switch info.State.Type { + case "ready": + return s.useReadyCertificate(keyPEM, localCert, localExpiry, localErr, info) + case "failed": + return s.bindAndWait(ctx, keyPEM, csrPEM) + case "issuing": + return s.waitForCertificate(ctx, keyPEM) + default: + return s.bindAndWait(ctx, keyPEM, csrPEM) + } +} + +func (s *Service) useReadyCertificate(keyPEM []byte, localCert tls.Certificate, localExpiry time.Time, localErr error, info CertificateInfo) error { + serverExpiry, err := CertificateExpiry(info.State.Expiry) + if err != nil { + return fmt.Errorf("server certificate expiry: %w", err) + } + if localErr == nil && !localExpiry.Before(serverExpiry) && sameExpiry(s.Config.CertificateExpiry, serverExpiry) { + s.cert.Store(&localCert) + return nil + } + cert, expiry, err := SaveCertificate(s.Paths, []byte(info.State.Certificate), []byte(info.State.Chain), keyPEM, s.Config.Hostname) + if err != nil { + return fmt.Errorf("save certificate: %w", err) + } + s.cert.Store(&cert) + s.setExpiry(expiry) + s.Log.Printf("certificate ready (expires %s)", expiry.Format(time.RFC3339)) + return nil +} + +func (s *Service) bindAndWait(ctx context.Context, keyPEM, csrPEM []byte) error { + s.Log.Printf("requesting certificate for %s", s.Config.Hostname) + info, err := s.API.BindCertificate(ctx, s.Config.TunnelID, s.Config.Token, string(csrPEM)) + if err != nil { + var apiErr *APIError + if !errors.As(err, &apiErr) || apiErr.Status != 409 { + return fmt.Errorf("bind certificate: %w", err) + } + } + if info.State.Type == "ready" { + return s.storeReady(keyPEM, info) + } + if info.State.Type == "failed" { + return fmt.Errorf("certificate issuance failed: %s", info.State.Reason) + } + return s.waitForCertificate(ctx, keyPEM) +} + +func (s *Service) waitForCertificate(ctx context.Context, keyPEM []byte) error { + ctx, cancel := context.WithTimeout(ctx, 10*time.Minute) + defer cancel() + ticker := time.NewTicker(2 * time.Second) + defer ticker.Stop() + for { + info, err := s.API.GetCertificate(ctx, s.Config.TunnelID, s.Config.Token) + if err != nil { + return fmt.Errorf("poll certificate: %w", err) + } + switch info.State.Type { + case "ready": + return s.storeReady(keyPEM, info) + case "failed": + return fmt.Errorf("certificate issuance failed: %s", info.State.Reason) + } + select { + case <-ctx.Done(): + return fmt.Errorf("certificate not ready: %w", ctx.Err()) + case <-ticker.C: + } + } +} + +func (s *Service) storeReady(keyPEM []byte, info CertificateInfo) error { + cert, expiry, err := SaveCertificate(s.Paths, []byte(info.State.Certificate), []byte(info.State.Chain), keyPEM, s.Config.Hostname) + if err != nil { + return fmt.Errorf("save certificate: %w", err) + } + s.cert.Store(&cert) + s.setExpiry(expiry) + s.Log.Printf("certificate ready (expires %s)", expiry.Format(time.RFC3339)) + return nil +} + +func (s *Service) setExpiry(expiry time.Time) { + s.mu.Lock() + defer s.mu.Unlock() + s.Config.CertificateExpiry = expiry.UTC().Format(time.RFC3339) + if err := s.Config.Save(s.ConfigPath); err != nil { + s.Log.Printf("save config: %v", err) + } +} + +func sameExpiry(stored string, expiry time.Time) bool { + parsed, err := CertificateExpiry(stored) + return err == nil && parsed.Equal(expiry) +} + +// RefreshCertificate checks the server for a renewed certificate. A transient +// failure leaves the current certificate serving. +func (s *Service) RefreshCertificate(ctx context.Context) { + s.mu.RLock() + cfg := *s.Config + api := s.API + s.mu.RUnlock() + info, err := api.GetCertificate(ctx, cfg.TunnelID, cfg.Token) + if err != nil { + s.Log.Printf("certificate refresh check: %v", err) + return + } + if info.State.Type != "ready" { + if info.State.Type == "failed" { + s.Log.Printf("certificate renewal failed on the server: %s", info.State.Reason) + } + return + } + expiry, err := CertificateExpiry(info.State.Expiry) + if err != nil { + s.Log.Printf("certificate expiry: %v", err) + return + } + if sameExpiry(cfg.CertificateExpiry, expiry) { + return + } + keyPEM, err := os.ReadFile(s.Paths.Key) + if err != nil { + s.Log.Printf("read key: %v", err) + return + } + cert, expiry, err := SaveCertificate(s.Paths, []byte(info.State.Certificate), []byte(info.State.Chain), keyPEM, cfg.Hostname) + if err != nil { + s.Log.Printf("save renewed certificate: %v", err) + return + } + s.cert.Store(&cert) + s.setExpiry(expiry) + s.Log.Printf("certificate renewed (expires %s)", expiry.Format(time.RFC3339)) +} + +func (s *Service) refreshLoop(ctx context.Context) { + for { + interval := 12 * time.Hour + s.mu.RLock() + expiryText := s.Config.CertificateExpiry + s.mu.RUnlock() + if expiry, err := CertificateExpiry(expiryText); err == nil { + if time.Until(expiry) < 30*24*time.Hour { + interval = time.Minute + } + } + timer := time.NewTimer(interval) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + s.RefreshCertificate(ctx) + } + } +} + +// Run keeps the bridge connected, reloading routes when the config changes. +func (s *Service) Run(ctx context.Context) error { + if err := s.Ensure(ctx); err != nil { + return err + } + go s.refreshLoop(ctx) + + reload := make(chan struct{}, 1) + go watchConfig(ctx, s.ConfigPath, reload) + + delay := 250 * time.Millisecond + for { + cfg, err := s.reload(ctx) + if err != nil { + s.Log.Printf("config: %v", err) + if !wait(ctx, delay) { + return ctx.Err() + } + delay = nextBackoff(delay) + continue + } + sess, err := s.connect(ctx, cfg) + if err != nil { + s.Log.Printf("bridge: %v", err) + if errors.Is(err, errFatalAttach) { + return err + } + if !wait(ctx, delay) { + return ctx.Err() + } + delay = nextBackoff(delay) + continue + } + delay = 250 * time.Millisecond + select { + case <-ctx.Done(): + sess.close("shutdown") + return ctx.Err() + case <-reload: + s.Log.Printf("config changed; reconnecting") + sess.close("config changed") + case <-sess.done: + } + } +} + +func (s *Service) reload(ctx context.Context) (*Config, error) { + cfg, err := Load(s.ConfigPath) + if err != nil { + return nil, err + } + if cfg.API == "" { + return nil, errors.New("no API configured; run: tunnel config set-api https://your-server") + } + if cfg.TunnelID == "" || cfg.Token == "" || cfg.Hostname == "" || s.cert.Load() == nil { + if err := s.Ensure(ctx); err != nil { + return nil, err + } + s.mu.RLock() + cfg = s.Config + s.mu.RUnlock() + } + s.mu.Lock() + s.Config = cfg + s.API = NewAPIClient(cfg.API) + s.mu.Unlock() + return cfg, nil +} + +// TLSConfig serves the current certificate and supports hot renewal. +func (s *Service) TLSConfig() *tls.Config { + return &tls.Config{ + MinVersion: tls.VersionTLS12, + NextProtos: []string{"http/1.1"}, + GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) { + cert := s.cert.Load() + if cert == nil { + return nil, errors.New("no certificate loaded") + } + return cert, nil + }, + } +} + +func watchConfig(ctx context.Context, path string, changed chan<- struct{}) { + var last time.Time + if info, err := os.Stat(path); err == nil { + last = info.ModTime() + } + ticker := time.NewTicker(2 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + info, err := os.Stat(path) + if err != nil || !info.ModTime().After(last) { + continue + } + last = info.ModTime() + select { + case changed <- struct{}{}: + default: + } + } + } +} + +func wait(ctx context.Context, d time.Duration) bool { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} + +func nextBackoff(d time.Duration) time.Duration { + d *= 2 + if d > 30*time.Second { + return 30 * time.Second + } + return d +} diff --git a/internal/server/admin.go b/internal/server/admin.go index 7ef7763..a311931 100644 --- a/internal/server/admin.go +++ b/internal/server/admin.go @@ -60,10 +60,24 @@ type adminOverview struct { Server adminServer `json:"server"` CA adminCA `json:"ca"` Totals adminTotals `json:"totals"` + Metrics adminMetrics `json:"metrics"` Tunnels []adminTunnel `json:"tunnels"` Bridges []adminBridge `json:"bridges"` } +type adminMetrics struct { + CreateRequests uint64 `json:"create_requests"` + BindRequests uint64 `json:"bind_requests"` + RateLimited uint64 `json:"rate_limited"` + IssuanceAttempts uint64 `json:"issuance_attempts"` + IssuanceReady uint64 `json:"issuance_ready"` + IssuanceFailed uint64 `json:"issuance_failed"` + CertificatesReady int `json:"certificates_ready"` + CertificatesIssuing int `json:"certificates_issuing"` + CertificatesFailed int `json:"certificates_failed"` + CertificatesRetrying int `json:"certificates_retrying"` +} + // adminAuthorized enforces the bearer ADMIN_TOKEN; 503 when unset. func (a *API) adminAuthorized(w http.ResponseWriter, r *http.Request) bool { if a.AdminToken == "" { @@ -145,10 +159,28 @@ func (a *API) adminOverviewHandler(w http.ResponseWriter, r *http.Request) { entry.CertificateState = certState(t.Certificate).Type entry.CertificateIssuer = t.Certificate.Issuer entry.CertificateError = t.Certificate.Reason + switch { + case certReady(t.Certificate): + overview.Metrics.CertificatesReady++ + if t.Certificate.State == "failed" { + overview.Metrics.CertificatesRetrying++ + } + case t.Certificate.State == "failed": + overview.Metrics.CertificatesFailed++ + default: + overview.Metrics.CertificatesIssuing++ + } } overview.Tunnels = append(overview.Tunnels, entry) } + overview.Metrics.CreateRequests = a.createCount.Load() + overview.Metrics.BindRequests = a.bindCount.Load() + overview.Metrics.RateLimited = a.rateLimited.Load() + overview.Metrics.IssuanceAttempts = a.issueAttempts.Load() + overview.Metrics.IssuanceReady = a.issueReady.Load() + overview.Metrics.IssuanceFailed = a.issueFailed.Load() + writeJSON(w, http.StatusOK, overview) } diff --git a/internal/server/api.go b/internal/server/api.go index 63ed1ff..3bdf43c 100644 --- a/internal/server/api.go +++ b/internal/server/api.go @@ -10,6 +10,7 @@ import ( "net/http" "strings" "sync" + "sync/atomic" "time" ) @@ -27,8 +28,20 @@ type API struct { Issuer Issuer // certificate issuer; defaults to the local CA IssuerName string // "local" or the ACME directory URL + // Limits are optional; a nil limiter allows everything. + CreateLimiter *RateLimiter + BindLimiter *RateLimiter + BindTunnelLimiter *RateLimiter + issueMu sync.Mutex issuing map[string]bool + + createCount atomic.Uint64 + bindCount atomic.Uint64 + rateLimited atomic.Uint64 + issueAttempts atomic.Uint64 + issueReady atomic.Uint64 + issueFailed atomic.Uint64 } func (a *API) ServeHTTP(w http.ResponseWriter, r *http.Request) { @@ -113,6 +126,14 @@ func (a *API) createTunnel(w http.ResponseWriter, r *http.Request) { }) return } + if ok, retry := a.CreateLimiter.Allow(clientIP(r)); !ok { + a.rateLimited.Add(1) + w.Header().Set("Retry-After", retryAfterSeconds(retry)) + writeJSON(w, http.StatusTooManyRequests, errRateLimited{ + Tag: "RateLimitedError", Message: "Too many tunnels created from this address; try again later", + }) + return + } tunnel, token, err := a.Store.createTunnel(a.Domain) if err != nil { writeJSON(w, http.StatusServiceUnavailable, errInvalidRequest{ @@ -120,6 +141,7 @@ func (a *API) createTunnel(w http.ResponseWriter, r *http.Request) { }) return } + a.createCount.Add(1) writeJSON(w, http.StatusCreated, createResponse{ Tunnel: TunnelInfo{ID: tunnel.ID, Hostname: tunnel.Hostname, State: "offline"}, Token: token, @@ -199,6 +221,17 @@ func (a *API) bindCertificate(w http.ResponseWriter, r *http.Request, id string) }) return } + // A bind that reaches here creates a new certificate (or retries a failed + // one), so it counts against the issuance limits. + if ok, retry := a.BindLimiter.Allow(clientIP(r)); !ok { + a.rateLimit(w, retry) + return + } + if ok, retry := a.BindTunnelLimiter.Allow(id); !ok { + a.rateLimit(w, retry) + return + } + a.bindCount.Add(1) if _, err := parseCSR(csr, tunnel.Hostname); err != nil { var hostErr errCSRHostname if errors.As(err, &hostErr) { @@ -258,6 +291,15 @@ func certState(cert *storedCert) CertState { } } +// rateLimit answers a 429 with the contract's error tag and a Retry-After. +func (a *API) rateLimit(w http.ResponseWriter, retryAfter time.Duration) { + a.rateLimited.Add(1) + w.Header().Set("Retry-After", retryAfterSeconds(retryAfter)) + writeJSON(w, http.StatusTooManyRequests, errRateLimited{ + Tag: "RateLimitedError", Message: "Rate limit exceeded; try again later", + }) +} + // issuer returns the configured certificate issuer (the local CA by default). func (a *API) issuer() Issuer { if a.Issuer != nil { @@ -294,6 +336,7 @@ func (a *API) runIssue(tunnelID, hostname, csrPEM, certID string) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) defer cancel() log.Printf("issuing certificate for %s via %s", hostname, a.issuerName()) + a.issueAttempts.Add(1) result, err := a.issuer().Issue(ctx, csrPEM, hostname) // The tunnel may have been deleted, or bound a different CSR, while the @@ -306,6 +349,7 @@ func (a *API) runIssue(tunnelID, hostname, csrPEM, certID string) { cert := storedCert{ID: certID, CSR: csrPEM, Issuer: a.issuerName()} if err != nil { + a.issueFailed.Add(1) cert.State = "failed" cert.Reason = err.Error() if certReady(current.Certificate) { @@ -322,6 +366,7 @@ func (a *API) runIssue(tunnelID, hostname, csrPEM, certID string) { cert.Certificate = result.Certificate cert.Chain = result.Chain cert.Expiry = expiryString(result.Expiry) + a.issueReady.Add(1) log.Printf("certificate for %s ready (expires %s)", hostname, cert.Expiry) } if err := a.Store.bindCertificate(tunnelID, cert); err != nil { diff --git a/internal/server/bridge.go b/internal/server/bridge.go index b6776c5..31e4014 100644 --- a/internal/server/bridge.go +++ b/internal/server/bridge.go @@ -166,7 +166,8 @@ type Bridge struct { clientVersion string startedAt time.Time - outbox chan outbound + control chan outbound + dataQ chan outbound done chan struct{} closeOnce sync.Once @@ -244,7 +245,8 @@ func (h *Hub) attach(w http.ResponseWriter, r *http.Request, tunnel storedTunnel ws: ws, clientVersion: attach.Client.Version, startedAt: time.Now(), - outbox: make(chan outbound, 256), + control: make(chan outbound, 64), + dataQ: make(chan outbound, 256), done: make(chan struct{}), conns: map[uint32]*visitorConn{}, } @@ -295,13 +297,13 @@ func (b *Bridge) openVisitor(conn net.Conn, hello []byte, peer, sni, alpn string open, _ := json.Marshal(openMessage{ Type: "open", Conn: id, Peer: peer, SNI: sni, ALPN: alpn, }) - if !b.send(websocket.MessageText, open) { + if !b.sendControl(websocket.MessageText, open) { b.removeConn(id, true) return } for len(hello) > 0 { n := min(len(hello), maxPayload) - if !b.send(websocket.MessageBinary, buildFrame(id, hello[:n])) { + if !b.sendData(websocket.MessageBinary, buildFrame(id, hello[:n])) { b.removeConn(id, true) return } @@ -317,7 +319,7 @@ func (b *Bridge) pump(v *visitorConn) { n, err := v.conn.Read(buf) v.touch() if n > 0 { - if !b.send(websocket.MessageBinary, buildFrame(v.id, buf[:n])) { + if !b.sendData(websocket.MessageBinary, buildFrame(v.id, buf[:n])) { return } } @@ -332,15 +334,17 @@ func (b *Bridge) pump(v *visitorConn) { } if err == io.EOF { end, _ := json.Marshal(endMessage{Type: "end", Conn: v.id}) - b.send(websocket.MessageText, end) + b.sendData(websocket.MessageText, end) } else { reset, _ := json.Marshal(resetMessage{Type: "reset", Conn: v.id, Code: ResetIOError}) - b.send(websocket.MessageText, reset) + b.sendControl(websocket.MessageText, reset) b.removeConn(v.id, true) return } if bothEnded { b.removeConn(v.id, false) + } else { + b.scheduleHalfClose(v.id, v) } return } @@ -387,7 +391,7 @@ func (b *Bridge) handleControl(raw []byte) { switch msg.Type { case "ping": pong, _ := json.Marshal(pingPongMessage{Type: "pong", TimeSent: msg.TimeSent}) - b.send(websocket.MessageText, pong) + b.sendControl(websocket.MessageText, pong) case "end": if msg.Conn != nil { b.endFromClient(*msg.Conn) @@ -416,19 +420,48 @@ func (b *Bridge) endFromClient(connID uint32) { } if bothEnded { b.removeConn(connID, false) + } else { + b.scheduleHalfClose(connID, v) } } -// writeLoop is the only writer on the WebSocket, keeping frames ordered. +// writeLoop is the only writer on the WebSocket. Control messages get +// priority over queued visitor data; `end` messages travel in the data queue +// so they stay behind the final bytes of a response. func (b *Bridge) writeLoop() { ctx := context.Background() + write := func(msg outbound) bool { + if err := writeWS(ctx, b.ws, msg.typ, msg.data); err != nil { + b.close(websocket.StatusInternalError, "write failed") + return false + } + return true + } for { select { case <-b.done: return - case msg := <-b.outbox: - if err := writeWS(ctx, b.ws, msg.typ, msg.data); err != nil { - b.close(websocket.StatusInternalError, "write failed") + default: + } + select { + case <-b.done: + return + case msg := <-b.control: + if !write(msg) { + return + } + continue + default: + } + select { + case <-b.done: + return + case msg := <-b.control: + if !write(msg) { + return + } + case msg := <-b.dataQ: + if !write(msg) { return } } @@ -459,28 +492,52 @@ func (b *Bridge) sweeper() { b.mu.Unlock() for _, id := range stale { reset, _ := json.Marshal(resetMessage{Type: "reset", Conn: id, Code: ResetIdleTimeout}) - b.send(websocket.MessageText, reset) + b.sendControl(websocket.MessageText, reset) b.removeConn(id, true) } } } } -// send queues a frame, applying backpressure when the client is slow. -// It returns false when the bridge is gone. -func (b *Bridge) send(typ websocket.MessageType, data []byte) bool { +// sendControl queues a control frame. It never blocks behind visitor data; a +// stalled client whose control queue is full is closed. +func (b *Bridge) sendControl(typ websocket.MessageType, data []byte) bool { select { case <-b.done: return false - case b.outbox <- outbound{typ, data}: + default: + } + select { + case b.control <- outbound{typ, data}: return true + case <-b.done: + return false + default: + b.close(websocket.StatusInternalError, "control queue full") + return false } } -// trySend queues a frame only if it fits immediately. +// sendData queues visitor bytes, blocking (and thus backpressuring the visitor +// socket) when the client is slow. +func (b *Bridge) sendData(typ websocket.MessageType, data []byte) bool { + select { + case <-b.done: + return false + default: + } + select { + case b.dataQ <- outbound{typ, data}: + return true + case <-b.done: + return false + } +} + +// trySend queues a control frame only if it fits immediately. func (b *Bridge) trySend(typ websocket.MessageType, data []byte) { select { - case b.outbox <- outbound{typ, data}: + case b.control <- outbound{typ, data}: default: } } @@ -490,6 +547,10 @@ func (b *Bridge) removeConn(id uint32, rst bool) { b.mu.Lock() v := b.conns[id] delete(b.conns, id) + if v != nil && v.halfClose != nil { + v.halfClose.Stop() + v.halfClose = nil + } b.mu.Unlock() if v == nil { return @@ -502,6 +563,31 @@ func (b *Bridge) removeConn(id uint32, rst bool) { v.conn.Close() } +// halfCloseTimeout bounds how long a connection may stay half-closed before +// it is reset, so a peer that never closes its side cannot pin a connection. +const halfCloseTimeout = 60 * time.Second + +// scheduleHalfClose resets a connection whose one side ended but whose other +// side has not, after halfCloseTimeout. +func (b *Bridge) scheduleHalfClose(id uint32, v *visitorConn) { + b.mu.Lock() + defer b.mu.Unlock() + if v.halfClose != nil || (v.visitorEnded && v.clientEnded) || b.conns[id] != v { + return + } + v.halfClose = time.AfterFunc(halfCloseTimeout, func() { + b.mu.Lock() + stillKnown := b.conns[id] == v + b.mu.Unlock() + if !stillKnown { + return + } + reset, _ := json.Marshal(resetMessage{Type: "reset", Conn: id, Code: ResetIdleTimeout}) + b.sendControl(websocket.MessageText, reset) + b.removeConn(id, true) + }) +} + func (b *Bridge) touch() { b.lastActivity.Store(time.Now().UnixMilli()) } @@ -531,6 +617,7 @@ type visitorConn struct { visitorEnded bool // pump saw EOF: no more visitor data clientEnded bool // client sent end: target will send no more + halfClose *time.Timer lastActivity atomic.Int64 } diff --git a/internal/server/cooldown.go b/internal/server/cooldown.go new file mode 100644 index 0000000..4a5dc49 --- /dev/null +++ b/internal/server/cooldown.go @@ -0,0 +1,94 @@ +package server + +import ( + "errors" + "net" + "strings" + "sync" + "time" + + "golang.org/x/crypto/acme" +) + +// cooldown pauses new ACME orders after repeated CA-side failures. Five +// consecutive failures trigger a one-minute pause, doubling up to an hour; a +// success clears the state. +type cooldown struct { + mu sync.Mutex + failures int + until time.Time + now func() time.Time +} + +func newCooldown() cooldown { + return cooldown{now: time.Now} +} + +func (c *cooldown) remaining() time.Duration { + c.mu.Lock() + defer c.mu.Unlock() + if c.until.IsZero() { + return 0 + } + now := c.now() + if !now.Before(c.until) { + return 0 + } + return c.until.Sub(now) +} + +func (c *cooldown) success() { + c.mu.Lock() + c.failures = 0 + c.until = time.Time{} + c.mu.Unlock() +} + +// failure records a CA-side failure and returns the new pause, if one began. +func (c *cooldown) failure() time.Duration { + c.mu.Lock() + defer c.mu.Unlock() + c.failures++ + if c.failures < 5 { + return 0 + } + exponent := c.failures - 5 + if exponent > 6 { + exponent = 6 // cap at one hour before shifting + } + delay := time.Minute << exponent + if delay > time.Hour { + delay = time.Hour + } + c.until = c.now().Add(delay) + return delay +} + +// caFailure distinguishes failures that point at the CA (network, 5xx, rate +// limits) from local DNS/CSR problems and validation errors, which should not +// pause issuance. +func caFailure(err error) bool { + if err == nil { + return false + } + var hostErr errCSRHostname + if errors.As(err, &hostErr) { + return false + } + text := err.Error() + if strings.Contains(text, "dns:") || strings.Contains(text, "not a PEM") || strings.Contains(text, "CSR") { + return false + } + if strings.Contains(text, "authorization") || strings.Contains(text, "dns-01") { + return false + } + var netErr net.Error + if errors.As(err, &netErr) { + return true + } + var acmeErr *acme.Error + if errors.As(err, &acmeErr) { + return acmeErr.StatusCode == 429 || acmeErr.StatusCode >= 500 + } + return true +} diff --git a/internal/server/cooldown_test.go b/internal/server/cooldown_test.go new file mode 100644 index 0000000..4ed63d8 --- /dev/null +++ b/internal/server/cooldown_test.go @@ -0,0 +1,64 @@ +package server + +import ( + "errors" + "fmt" + "testing" + "time" + + "golang.org/x/crypto/acme" +) + +func TestCooldown(t *testing.T) { + now := time.Now() + c := newCooldown() + c.now = func() time.Time { return now } + + for i := 0; i < 4; i++ { + if d := c.failure(); d != 0 { + t.Fatalf("failure %d paused for %v, want no pause", i+1, d) + } + } + if d := c.failure(); d != time.Minute { + t.Fatalf("fifth failure pause = %v, want 1m", d) + } + if d := c.remaining(); d != time.Minute { + t.Fatalf("remaining = %v, want 1m", d) + } + if d := c.failure(); d != 2*time.Minute { + t.Fatalf("sixth failure pause = %v, want 2m", d) + } + for i := 0; i < 20; i++ { + c.failure() + } + if d := c.failure(); d != time.Hour { + t.Fatalf("pause cap = %v, want 1h", d) + } + + now = now.Add(2 * time.Hour) + if d := c.remaining(); d != 0 { + t.Fatalf("remaining after expiry = %v", d) + } + c.success() + if c.failures != 0 { + t.Fatalf("success left failures = %d", c.failures) + } +} + +func TestCAFailureClassification(t *testing.T) { + cases := []struct { + err error + want bool + }{ + {errors.New("dns: create _acme-challenge: nope"), false}, + {errors.New("acme: authorization for x is invalid"), false}, + {errors.New("acme: authorize order: context deadline exceeded"), true}, + {fmt.Errorf("acme: authorize order: %w", &acme.Error{StatusCode: 503}), true}, + {fmt.Errorf("acme: authorize order: %w", &acme.Error{StatusCode: 429}), true}, + } + for _, c := range cases { + if got := caFailure(c.err); got != c.want { + t.Errorf("caFailure(%v) = %v, want %v", c.err, got, c.want) + } + } +} diff --git a/internal/server/dns.go b/internal/server/dns.go index 938c2e5..3138928 100644 --- a/internal/server/dns.go +++ b/internal/server/dns.go @@ -14,6 +14,13 @@ type DNSProvider interface { DeleteTXT(ctx context.Context, name string) error } +// ChallengeSweeper is implemented by providers that can remove leftover +// challenge records at startup (for example after a process was killed +// mid-issuance). +type ChallengeSweeper interface { + SweepChallenges(ctx context.Context) (int, error) +} + // challengeRecordName is the TXT record name a hostname's DNS-01 challenge // lives at; the wildcard of a hostname shares the same record. func challengeRecordName(hostname string) string { diff --git a/internal/server/dns_porkbun.go b/internal/server/dns_porkbun.go index c814704..290bf91 100644 --- a/internal/server/dns_porkbun.go +++ b/internal/server/dns_porkbun.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "net/http" + "strings" "time" ) @@ -36,38 +37,38 @@ func (p *PorkbunDNS) client() *http.Client { return &http.Client{Timeout: 20 * time.Second} } -func (p *PorkbunDNS) post(ctx context.Context, path string, body map[string]string) error { +func (p *PorkbunDNS) post(ctx context.Context, path string, body map[string]string) ([]byte, error) { body["apikey"] = p.APIKey body["secretapikey"] = p.SecretKey raw, err := json.Marshal(body) if err != nil { - return err + return nil, err } req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.base()+path, bytes.NewReader(raw)) if err != nil { - return err + return nil, err } req.Header.Set("Content-Type", "application/json") resp, err := p.client().Do(req) if err != nil { - return err + return nil, err } defer resp.Body.Close() - data, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + data, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) if resp.StatusCode != http.StatusOK { - return fmt.Errorf("porkbun %s: HTTP %d: %s", path, resp.StatusCode, data) + return nil, fmt.Errorf("porkbun %s: HTTP %d: %s", path, resp.StatusCode, data) } var result struct { Status string `json:"status"` Message string `json:"message"` } if err := json.Unmarshal(data, &result); err != nil { - return fmt.Errorf("porkbun %s: unreadable response: %w", path, err) + return nil, fmt.Errorf("porkbun %s: unreadable response: %w", path, err) } if result.Status != "SUCCESS" { - return fmt.Errorf("porkbun %s: %s: %s", path, result.Status, result.Message) + return nil, fmt.Errorf("porkbun %s: %s: %s", path, result.Status, result.Message) } - return nil + return data, nil } func (p *PorkbunDNS) CreateTXT(ctx context.Context, name, value string) error { @@ -75,9 +76,10 @@ func (p *PorkbunDNS) CreateTXT(ctx context.Context, name, value string) error { if err != nil { return err } - return p.post(ctx, "/dns/create/"+p.Zone, map[string]string{ + _, err = p.post(ctx, "/dns/create/"+p.Zone, map[string]string{ "name": relative, "type": "TXT", "content": value, "ttl": "600", }) + return err } func (p *PorkbunDNS) DeleteTXT(ctx context.Context, name string) error { @@ -85,5 +87,53 @@ func (p *PorkbunDNS) DeleteTXT(ctx context.Context, name string) error { if err != nil { return err } - return p.post(ctx, "/dns/deleteByNameType/"+p.Zone+"/TXT/"+relative, map[string]string{}) + _, err = p.post(ctx, "/dns/deleteByNameType/"+p.Zone+"/TXT/"+relative, map[string]string{}) + return err +} + +type porkbunRecord struct { + Name string `json:"name"` + Type string `json:"type"` + Content string `json:"content"` +} + +// SweepChallenges removes every TXT record at a _acme-challenge.* name in the +// zone. It runs at startup, before this process starts any issuance, so any +// such records are leftovers from an interrupted attempt (single-node use). +func (p *PorkbunDNS) SweepChallenges(ctx context.Context) (int, error) { + data, err := p.post(ctx, "/dns/retrieve/"+p.Zone, map[string]string{}) + if err != nil { + return 0, err + } + var result struct { + Records []porkbunRecord `json:"records"` + } + if err := json.Unmarshal(data, &result); err != nil { + return 0, fmt.Errorf("porkbun retrieve: %w", err) + } + seen := map[string]struct{}{} + removed := 0 + for _, record := range result.Records { + if record.Type != "TXT" { + continue + } + name := strings.TrimSuffix(record.Name, ".") + if !strings.HasPrefix(name, "_acme-challenge.") { + continue + } + relative, err := relativeRecordName(name, p.Zone) + if err != nil { + // Only records inside this zone are ours to delete. + continue + } + if _, done := seen[relative]; done { + continue + } + seen[relative] = struct{}{} + if _, err := p.post(ctx, "/dns/deleteByNameType/"+p.Zone+"/TXT/"+relative, map[string]string{}); err != nil { + return removed, err + } + removed++ + } + return removed, nil } diff --git a/internal/server/dns_porkbun_test.go b/internal/server/dns_porkbun_test.go index c450aa3..926bd7e 100644 --- a/internal/server/dns_porkbun_test.go +++ b/internal/server/dns_porkbun_test.go @@ -72,3 +72,46 @@ func TestPorkbunDNS(t *testing.T) { t.Errorf("expected the API error, got %v", err) } } + +func TestPorkbunSweepChallenges(t *testing.T) { + var deletes []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/dns/retrieve/tzgyn.dev": + w.Write([]byte(`{"status":"SUCCESS","records":[ + {"name":"_acme-challenge.a.tunnel.tzgyn.dev","type":"TXT","content":"one"}, + {"name":"_acme-challenge.a.tunnel.tzgyn.dev","type":"TXT","content":"two"}, + {"name":"_acme-challenge.b.tunnel.tzgyn.dev","type":"TXT","content":"three"}, + {"name":"www.tzgyn.dev","type":"TXT","content":"keep"}, + {"name":"_acme-challenge.other.example.com","type":"TXT","content":"outside"} + ]}`)) + case strings.HasPrefix(r.URL.Path, "/dns/deleteByNameType/"): + deletes = append(deletes, r.URL.Path) + w.Write([]byte(`{"status":"SUCCESS"}`)) + default: + http.NotFound(w, r) + } + })) + defer srv.Close() + + p := &PorkbunDNS{APIKey: "pk", SecretKey: "sk", Zone: "tzgyn.dev", BaseURL: srv.URL, HTTP: srv.Client()} + count, err := p.SweepChallenges(context.Background()) + if err != nil { + t.Fatal(err) + } + if count != 2 { + t.Errorf("count = %d, want 2", count) + } + want := []string{ + "/dns/deleteByNameType/tzgyn.dev/TXT/_acme-challenge.a.tunnel", + "/dns/deleteByNameType/tzgyn.dev/TXT/_acme-challenge.b.tunnel", + } + if len(deletes) != len(want) { + t.Fatalf("deletes = %v, want %v", deletes, want) + } + for i := range want { + if deletes[i] != want[i] { + t.Errorf("delete[%d] = %q, want %q", i, deletes[i], want[i]) + } + } +} diff --git a/internal/server/issuer.go b/internal/server/issuer.go index a1f3577..79fea3e 100644 --- a/internal/server/issuer.go +++ b/internal/server/issuer.go @@ -72,6 +72,7 @@ type acmeIssuer struct { store *Store opts ACMEOptions sem chan struct{} // bounds concurrent ACME orders + cool cooldown accountMu sync.Mutex // serializes first-use account registration } @@ -85,7 +86,7 @@ func NewACMEIssuer(store *Store, opts ACMEOptions) (Issuer, error) { if opts.DNS == nil { return nil, errors.New("acme: a DNS provider is required") } - return &acmeIssuer{store: store, opts: opts, sem: make(chan struct{}, 2)}, nil + return &acmeIssuer{store: store, opts: opts, sem: make(chan struct{}, 2), cool: newCooldown()}, nil } func (a *acmeIssuer) Instant() bool { return false } @@ -183,18 +184,33 @@ func (a *acmeIssuer) client(ctx context.Context) (*acme.Client, error) { // Issue orders a certificate for the tunnel's names (hostname and, when the // CSR asks for it, its wildcard) using one DNS-01 challenge record shared by // both, then returns the certificate for the client's key. -func (a *acmeIssuer) Issue(ctx context.Context, csrPEM, hostname string) (IssuedCert, error) { +func (a *acmeIssuer) Issue(ctx context.Context, csrPEM, hostname string) (result IssuedCert, err error) { csr, err := parseCSR(csrPEM, hostname) if err != nil { return IssuedCert{}, err } + if delay := a.cool.remaining(); delay > 0 { + return IssuedCert{}, fmt.Errorf("acme: CA cooling down for %s after repeated failures", delay.Round(time.Second)) + } + select { case a.sem <- struct{}{}: case <-ctx.Done(): return IssuedCert{}, ctx.Err() } defer func() { <-a.sem }() + defer func() { + if err == nil { + a.cool.success() + return + } + if caFailure(err) { + if delay := a.cool.failure(); delay > 0 { + log.Printf("acme: pausing new orders for %s after repeated CA failures", delay) + } + } + }() client, err := a.client(ctx) if err != nil { diff --git a/internal/server/limits.go b/internal/server/limits.go new file mode 100644 index 0000000..7f1fbc9 --- /dev/null +++ b/internal/server/limits.go @@ -0,0 +1,99 @@ +package server + +import ( + "math" + "net" + "net/http" + "strconv" + "strings" + "sync" + "time" +) + +// RateLimiter is a fixed-window counter keyed by an arbitrary string (client +// address, tunnel ID). A nil limiter allows everything. +type RateLimiter struct { + mu sync.Mutex + limit int + window time.Duration + entries map[string]*rateEntry + now func() time.Time +} + +type rateEntry struct { + count int + reset time.Time +} + +// NewRateLimiter returns a limiter allowing limit events per window, or nil +// when limit is not positive. +func NewRateLimiter(limit int, window time.Duration) *RateLimiter { + if limit <= 0 || window <= 0 { + return nil + } + return &RateLimiter{ + limit: limit, + window: window, + entries: map[string]*rateEntry{}, + now: time.Now, + } +} + +// Allow records one event for key. When it returns false, retryAfter is the +// time left in the current window. +func (r *RateLimiter) Allow(key string) (ok bool, retryAfter time.Duration) { + if r == nil { + return true, 0 + } + r.mu.Lock() + defer r.mu.Unlock() + now := r.now() + entry := r.entries[key] + if entry == nil || !now.Before(entry.reset) { + if len(r.entries) > 4096 { + // Keep the map bounded on long-running servers; expired entries + // are safe to drop. + for k, e := range r.entries { + if !now.Before(e.reset) { + delete(r.entries, k) + } + } + } + entry = &rateEntry{reset: now.Add(r.window)} + r.entries[key] = entry + } + if entry.count >= r.limit { + return false, entry.reset.Sub(now) + } + entry.count++ + return true, 0 +} + +// clientIP returns the visitor address, honoring the reverse proxy headers +// the API sits behind. It is only meaningful for rate limiting; direct clients +// can spoof X-Forwarded-For unless the listener is reachable only through the +// proxy. +func clientIP(r *http.Request) string { + if real := strings.TrimSpace(r.Header.Get("X-Real-IP")); net.ParseIP(real) != nil { + return real + } + if forwarded := r.Header.Get("X-Forwarded-For"); forwarded != "" { + first := strings.TrimSpace(strings.Split(forwarded, ",")[0]) + if net.ParseIP(first) != nil { + return first + } + } + if host, _, err := net.SplitHostPort(r.RemoteAddr); err == nil { + return host + } + return r.RemoteAddr +} + +// retryAfterSeconds formats a Retry-After value (at least one second). +func retryAfterSeconds(d time.Duration) string { + seconds := int(math.Ceil(d.Seconds())) + if seconds < 1 { + seconds = 1 + } + return strconv.Itoa(seconds) +} diff --git a/internal/server/limits_test.go b/internal/server/limits_test.go new file mode 100644 index 0000000..5235303 --- /dev/null +++ b/internal/server/limits_test.go @@ -0,0 +1,83 @@ +package server + +import ( + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestRateLimiter(t *testing.T) { + now := time.Now() + limiter := NewRateLimiter(2, time.Hour) + limiter.now = func() time.Time { return now } + + if ok, _ := limiter.Allow("a"); !ok { + t.Fatal("first event should pass") + } + if ok, _ := limiter.Allow("a"); !ok { + t.Fatal("second event should pass") + } + ok, retry := limiter.Allow("a") + if ok || retry <= 0 || retry > time.Hour { + t.Fatalf("third event: ok=%v retry=%v", ok, retry) + } + if ok, _ := limiter.Allow("b"); !ok { + t.Fatal("another key should have its own window") + } + + now = now.Add(time.Hour + time.Second) + if ok, _ := limiter.Allow("a"); !ok { + t.Fatal("a new window should allow events again") + } +} + +func TestClientIP(t *testing.T) { + cases := []struct { + forwarded, real, remote, want string + }{ + {"203.0.113.9, 10.0.0.1", "", "10.0.0.2:1234", "203.0.113.9"}, + {"", "198.51.100.7", "10.0.0.2:1234", "198.51.100.7"}, + {"", "", "192.0.2.5:443", "192.0.2.5"}, + {"garbage", "", "192.0.2.5:443", "192.0.2.5"}, + } + for _, c := range cases { + r := httptest.NewRequest("GET", "/", nil) + r.RemoteAddr = c.remote + if c.forwarded != "" { + r.Header.Set("X-Forwarded-For", c.forwarded) + } + if c.real != "" { + r.Header.Set("X-Real-IP", c.real) + } + if got := clientIP(r); got != c.want { + t.Errorf("clientIP(%q, %q, %q) = %q, want %q", c.forwarded, c.real, c.remote, got, c.want) + } + } +} + +func TestCreateRateLimit(t *testing.T) { + store, err := OpenStore("") + if err != nil { + t.Fatal(err) + } + api := &API{ + Store: store, Hub: NewHub(store), Domain: "opentunnel.test", StartedAt: time.Now(), + CreateLimiter: NewRateLimiter(1, time.Hour), + } + srv := httptest.NewServer(api) + defer srv.Close() + + resp := request(t, "POST", srv.URL+"/api/tunnel", "", "application/json", `{}`) + if resp.StatusCode != 201 { + t.Fatalf("first create: %d", resp.StatusCode) + } + resp = request(t, "POST", srv.URL+"/api/tunnel", "", "application/json", `{}`) + body := bodyString(t, resp) + if resp.StatusCode != 429 || !strings.Contains(body, `"_tag":"RateLimitedError"`) { + t.Fatalf("second create: %d %q", resp.StatusCode, body) + } + if resp.Header.Get("Retry-After") == "" { + t.Error("429 response is missing Retry-After") + } +} diff --git a/internal/server/proto.go b/internal/server/proto.go index 7d23001..64911cd 100644 --- a/internal/server/proto.go +++ b/internal/server/proto.go @@ -97,6 +97,12 @@ type errUnauthorized struct { Message string `json:"message"` } +// errRateLimited is the 429 body used by the reference's limits. +type errRateLimited struct { + Tag string `json:"_tag"` + Message string `json:"message"` +} + type errTaggedID struct { Tag string `json:"_tag"` TunnelID string `json:"tunnelID"` diff --git a/web/src/App.svelte b/web/src/App.svelte index e851cac..59b35b6 100644 --- a/web/src/App.svelte +++ b/web/src/App.svelte @@ -69,7 +69,7 @@ {#if !token}

Enter the ADMIN_TOKEN to view the dashboard.

{:else if overview} - +

Tunnels

Bridges

diff --git a/web/src/lib/SummaryCards.svelte b/web/src/lib/SummaryCards.svelte index 2e893df..41d7526 100644 --- a/web/src/lib/SummaryCards.svelte +++ b/web/src/lib/SummaryCards.svelte @@ -1,8 +1,8 @@
@@ -10,6 +10,19 @@
Online
{totals.online}
Bridges
{totals.bridges}
Visitors
{totals.visitors}
+
+
Certificates
+
{metrics.certificates_ready}
+
+ {metrics.certificates_issuing} issuing · {metrics.certificates_failed} failed + {#if metrics.certificates_retrying > 0}· {metrics.certificates_retrying} retrying{/if} +
+
+
+
Issuances
+
{metrics.issuance_ready}
+
{metrics.issuance_attempts} attempts · {metrics.rate_limited} limited
+
Local CA expires
{fmtDate(ca.expires)}
Uptime
{fmtUptime(server.uptime_seconds)}
@@ -25,4 +38,9 @@ font-size: 14px; margin-top: 8px; } + .detail { + margin-top: 6px; + color: #8b949e; + font-size: 11px; + } diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 3a7d578..ffc6f0c 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -19,6 +19,19 @@ export interface Totals { visitors: number; } +export interface Metrics { + create_requests: number; + bind_requests: number; + rate_limited: number; + issuance_attempts: number; + issuance_ready: number; + issuance_failed: number; + certificates_ready: number; + certificates_issuing: number; + certificates_failed: number; + certificates_retrying: number; +} + export interface Tunnel { id: string; hostname: string; @@ -46,6 +59,7 @@ export interface Overview { server: ServerInfo; ca: CaInfo; totals: Totals; + metrics: Metrics; tunnels: Tunnel[] | null; bridges: Bridge[] | null; }