Add the native Go client and finish M3 hardening

The new cmd/tunnel client keeps the API URL and tunnel identity in
~/.config/tunnel/config.json (0600), provisions the key/CSR and
certificate, terminates TLS, proxies to local routes, reloads routes
without a restart, and runs as a background service. `tunnel import`
takes over an existing official CLI profile, preserving its tunnel ID,
token, key and certificate.

Server hardening: per-IP and per-tunnel rate limits for creates and
binds, a CA cool-down after repeated ACME failures, a startup sweep for
stale _acme-challenge records, admin metrics, a 60-second half-close
timeout, and separate prioritized control / ordered data bridge queues
(with `end` ordered after the final bytes). The dashboard shows the new
metrics.
This commit is contained in:
2026-10-26 00:32:31 +08:00
parent cba7d2366b
commit 65a7e4e4c5
27 changed files with 3745 additions and 56 deletions
+35 -8
View File
@@ -16,6 +16,12 @@ what we take vs. skip). Never vendor it into the build.
## Layout ## Layout
- `cmd/tunneld` — flags/env wiring, listeners, signal handling - `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 - `internal/server/proto.go` — wire types (pinned JSON key order), IDs/tokens
(same generation as the reference: 12-char base32 slugs, `rly_` tokens, (same generation as the reference: 12-char base32 slugs, `rly_` tokens,
hex SHA-256 token hashes), route validation, data frames 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 - Frontend: `cd web && bun install && bun run check && bun run build` — the
dashboard test skips when `internal/dashboard/dist` has no build (only the dashboard test skips when `internal/dashboard/dist` has no build (only the
committed `.gitkeep`) committed `.gitkeep`)
- Go client: `go build -o .local/tunnel ./cmd/tunnel`, then
`.local/tunnel config set-api <url>` and `.local/tunnel route add <port> --name <name>`;
`up`/`down` manage the background service, `status` shows local + server
state. `import --profile <name> --api <url>` takes over an official CLI
profile (same tunnel/key/cert).
- Dev loop: `go build -o .local/tunneld ./cmd/tunneld && .local/tunneld -domain localhost`, - 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` 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 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 the service (`opentunnel down` / `up`) to fetch it immediately. An issuance
killed mid-flight can leave `_acme-challenge` TXT records behind; delete them killed mid-flight can leave `_acme-challenge` TXT records behind; delete them
with `dns/deleteByNameType` (production and test tunnels share the zone). 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**, - **macOS system curl (LibreSSL) does not send SNI for `*.localhost` names**,
so `curl https://api.<id>.localhost:8443/` gets closed (the relay requires so `curl https://api.<id>.localhost:8443/` gets closed (the relay requires
SNI). Test with `openssl s_client -servername ...` or Homebrew curl. SNI). Test with `openssl s_client -servername ...` or Homebrew curl.
- JSON key order of API bodies is contractual — always `json.Marshal` structs - JSON key order of API bodies is contractual — always `json.Marshal` structs
with declared field order, never maps. with declared field order, never maps.
- Bridge writes go through one ordered `outbox` channel per session. Control - Bridge writes use separate prioritized **control** and ordered **data**
messages therefore can queue behind data — a known M1 deviation from the queues per session. `open`, `reset`, pings and pongs cut ahead of queued
reference's deadlock-avoidance; a stalled visitor self-heals via the visitor data; `end` travels in the data queue so it cannot overtake the final
client's idle timeout. Revisit with per-connection flow control. 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 - The bridge WebSocket must use `coder/websocket` with compression disabled
and the `opentunnel` subprotocol; read limits are 64 KiB before attach and and the `opentunnel` subprotocol; read limits are 64 KiB before attach and
4+32 KiB after. 4+32 KiB after.
- No third-party deps beyond `github.com/coder/websocket`; keep it that way - No third-party deps beyond `github.com/coder/websocket` and
unless a milestone calls for it. `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); - 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 M3 rate limits/admin/hardening (done); M4 Go client (done). Remaining optional
into M1 fixes, or M3 scope into M2 fixes. 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).
+70 -13
View File
@@ -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 `docs/protocol.md`, so the official `opentunnel` CLI and SDK work against it
unchanged. unchanged.
Status: **M2** — core data path works end-to-end with the official CLI, and Status: **M4** — the data path, real ACME DNS-01 certificates with renewal,
tunnels get publicly trusted certificates through ACME DNS-01 (Let's Encrypt rate limits/hardening, and a native Go client are implemented. The official
+ Porkbun) with automatic renewal. The built-in local CA remains the default `opentunnel` CLI still works unchanged.
when no DNS provider is configured.
## Quick start (local development) ## Quick start (local development)
@@ -40,6 +39,51 @@ printf 'GET / HTTP/1.1\r\nHost: api.<id>.localhost\r\nConnection: close\r\n\r\n'
-CAfile ~/.local/share/tunneld/ca.pem -quiet -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 <url>` | store the server URL in the JSON config |
| `tunnel config show` | show API, tunnel, certificate and routes (never the token) |
| `tunnel route add <port\|host:port> [--name NAME] [--proxy-protocol v1\|v2]` | add/update a route; random name if omitted |
| `tunnel route remove <name>` / `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 ## Configuration
Every flag also reads an environment variable (`tunneld -h`): 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-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) | | `-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) | | `-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 ## 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 tunnel automatically, keeping the old certificate in service until the new one
is ready. 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 For testing against Let's Encrypt staging (untrusted certificates, separate
rate limits), set rate limits), set
`ACME_DIRECTORY_URL=https://acme-staging-v02.api.letsencrypt.org/directory`. `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 publicly trusted Let's Encrypt certificate and no client or visitor trust
setup is needed. setup is needed.
## What works (M1 + M2) ## What works (M1–M4)
- The full HTTP API contract (paths, status codes, bodies, key order) as - The full HTTP API contract (paths, status codes, bodies, key order) as
recorded in the reference's `worker-contract.json` recorded in the reference's `worker-contract.json`
@@ -179,24 +232,28 @@ setup is needed.
- Per-tunnel Let's Encrypt certificates (`<id>.<domain>` + optional wildcard) - Per-tunnel Let's Encrypt certificates (`<id>.<domain>` + optional wildcard)
that keep the client's key, with automatic renewal 30 days before expiry that keep the client's key, with automatic renewal 30 days before expiry
and retries on failure 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, - 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 - SNI routing on the TLS listener: the domain is terminated (API), tunnel
hostnames are passed through still encrypted hostnames are passed through still encrypted
- Admin dashboard at `/` (bearer `ADMIN_TOKEN`): tunnels, bridges, cert - 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 - Persistence: JSON store with the CA, ACME account, server certificate, and
tunnels 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 - Graceful shutdown: `drain` + close to every bridge
## Not yet (later milestones) ## Not yet (later)
- Issuance rate limits, CA cool-down, stale-DNS-record sweeps - Additional admin endpoints (export/import, token rotation)
- Metrics, more admin endpoints
- Multi-region mesh, inbound PROXY protocol on the TLS listener - Multi-region mesh, inbound PROXY protocol on the TLS listener
- 60 s half-close timeout, per-connection flow control, control messages - Per-connection flow control on the wire (the protocol has none; both sides
cutting ahead of queued data (single ordered outbound queue today) bound their queues)
- Go client/CLI — for now use the official `opentunnel` CLI - Replacing the official CLI on every platform; the Go client is macOS/Linux
(it uses POSIX process/session APIs for `up`/`down`)
## Development ## Development
+581
View File
@@ -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] <command> [options]
Commands:
config set-api <url> Store the server API URL in the config
config show Show the config (never the token)
route add <port|host:port> [--name NAME] [--proxy-protocol v1|v2]
route remove <name>
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 <url> | 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 <url>")
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 <url> | 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 <port|host:port> [--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 <name>")
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
}
+20
View File
@@ -69,6 +69,9 @@ func main() {
porkbunSecret := flag.String("porkbun-secret-api-key", envOr("PORKBUN_SECRET_API_KEY", ""), "Porkbun secret API key") 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)") 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") 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() flag.Parse()
if *caFile == "" { if *caFile == "" {
@@ -117,6 +120,16 @@ func main() {
case *dnsProvider != "": case *dnsProvider != "":
log.Fatalf("unknown DNS_PROVIDER %q (want porkbun or challtestsrv)", *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 { if dns != nil {
acme, err := server.NewACMEIssuer(store, server.ACMEOptions{ acme, err := server.NewACMEIssuer(store, server.ACMEOptions{
DirectoryURL: *acmeDirectory, DirectoryURL: *acmeDirectory,
@@ -144,12 +157,19 @@ func main() {
Version: version, Version: version,
Issuer: issuer, Issuer: issuer,
IssuerName: issuerName, IssuerName: issuerName,
CreateLimiter: server.NewRateLimiter(*createLimit, time.Hour),
BindLimiter: server.NewRateLimiter(*bindLimit, time.Hour),
BindTunnelLimiter: server.NewRateLimiter(*bindTunnelLimit, time.Hour),
} }
if *adminToken == "" { if *adminToken == "" {
log.Printf("admin dashboard: disabled (set ADMIN_TOKEN to enable it at /)") log.Printf("admin dashboard: disabled (set ADMIN_TOKEN to enable it at /)")
} else { } else {
log.Printf("admin dashboard: enabled at /") 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()) ctx, cancel := context.WithCancel(context.Background())
defer cancel() defer cancel()
+271
View File
@@ -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
}
+648
View File
@@ -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)
}
+211
View File
@@ -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])
}
}
+260
View File
@@ -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"),
}
}
+200
View File
@@ -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)
}
+248
View File
@@ -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)
}
+92
View File
@@ -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
}
+416
View File
@@ -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
}
+32
View File
@@ -60,10 +60,24 @@ type adminOverview struct {
Server adminServer `json:"server"` Server adminServer `json:"server"`
CA adminCA `json:"ca"` CA adminCA `json:"ca"`
Totals adminTotals `json:"totals"` Totals adminTotals `json:"totals"`
Metrics adminMetrics `json:"metrics"`
Tunnels []adminTunnel `json:"tunnels"` Tunnels []adminTunnel `json:"tunnels"`
Bridges []adminBridge `json:"bridges"` 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. // adminAuthorized enforces the bearer ADMIN_TOKEN; 503 when unset.
func (a *API) adminAuthorized(w http.ResponseWriter, r *http.Request) bool { func (a *API) adminAuthorized(w http.ResponseWriter, r *http.Request) bool {
if a.AdminToken == "" { if a.AdminToken == "" {
@@ -145,10 +159,28 @@ func (a *API) adminOverviewHandler(w http.ResponseWriter, r *http.Request) {
entry.CertificateState = certState(t.Certificate).Type entry.CertificateState = certState(t.Certificate).Type
entry.CertificateIssuer = t.Certificate.Issuer entry.CertificateIssuer = t.Certificate.Issuer
entry.CertificateError = t.Certificate.Reason 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.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) writeJSON(w, http.StatusOK, overview)
} }
+45
View File
@@ -10,6 +10,7 @@ import (
"net/http" "net/http"
"strings" "strings"
"sync" "sync"
"sync/atomic"
"time" "time"
) )
@@ -27,8 +28,20 @@ type API struct {
Issuer Issuer // certificate issuer; defaults to the local CA Issuer Issuer // certificate issuer; defaults to the local CA
IssuerName string // "local" or the ACME directory URL 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 issueMu sync.Mutex
issuing map[string]bool 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) { 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 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) tunnel, token, err := a.Store.createTunnel(a.Domain)
if err != nil { if err != nil {
writeJSON(w, http.StatusServiceUnavailable, errInvalidRequest{ writeJSON(w, http.StatusServiceUnavailable, errInvalidRequest{
@@ -120,6 +141,7 @@ func (a *API) createTunnel(w http.ResponseWriter, r *http.Request) {
}) })
return return
} }
a.createCount.Add(1)
writeJSON(w, http.StatusCreated, createResponse{ writeJSON(w, http.StatusCreated, createResponse{
Tunnel: TunnelInfo{ID: tunnel.ID, Hostname: tunnel.Hostname, State: "offline"}, Tunnel: TunnelInfo{ID: tunnel.ID, Hostname: tunnel.Hostname, State: "offline"},
Token: token, Token: token,
@@ -199,6 +221,17 @@ func (a *API) bindCertificate(w http.ResponseWriter, r *http.Request, id string)
}) })
return 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 { if _, err := parseCSR(csr, tunnel.Hostname); err != nil {
var hostErr errCSRHostname var hostErr errCSRHostname
if errors.As(err, &hostErr) { 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). // issuer returns the configured certificate issuer (the local CA by default).
func (a *API) issuer() Issuer { func (a *API) issuer() Issuer {
if a.Issuer != nil { 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) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
defer cancel() defer cancel()
log.Printf("issuing certificate for %s via %s", hostname, a.issuerName()) log.Printf("issuing certificate for %s via %s", hostname, a.issuerName())
a.issueAttempts.Add(1)
result, err := a.issuer().Issue(ctx, csrPEM, hostname) result, err := a.issuer().Issue(ctx, csrPEM, hostname)
// The tunnel may have been deleted, or bound a different CSR, while the // 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()} cert := storedCert{ID: certID, CSR: csrPEM, Issuer: a.issuerName()}
if err != nil { if err != nil {
a.issueFailed.Add(1)
cert.State = "failed" cert.State = "failed"
cert.Reason = err.Error() cert.Reason = err.Error()
if certReady(current.Certificate) { if certReady(current.Certificate) {
@@ -322,6 +366,7 @@ func (a *API) runIssue(tunnelID, hostname, csrPEM, certID string) {
cert.Certificate = result.Certificate cert.Certificate = result.Certificate
cert.Chain = result.Chain cert.Chain = result.Chain
cert.Expiry = expiryString(result.Expiry) cert.Expiry = expiryString(result.Expiry)
a.issueReady.Add(1)
log.Printf("certificate for %s ready (expires %s)", hostname, cert.Expiry) log.Printf("certificate for %s ready (expires %s)", hostname, cert.Expiry)
} }
if err := a.Store.bindCertificate(tunnelID, cert); err != nil { if err := a.Store.bindCertificate(tunnelID, cert); err != nil {
+106 -19
View File
@@ -166,7 +166,8 @@ type Bridge struct {
clientVersion string clientVersion string
startedAt time.Time startedAt time.Time
outbox chan outbound control chan outbound
dataQ chan outbound
done chan struct{} done chan struct{}
closeOnce sync.Once closeOnce sync.Once
@@ -244,7 +245,8 @@ func (h *Hub) attach(w http.ResponseWriter, r *http.Request, tunnel storedTunnel
ws: ws, ws: ws,
clientVersion: attach.Client.Version, clientVersion: attach.Client.Version,
startedAt: time.Now(), startedAt: time.Now(),
outbox: make(chan outbound, 256), control: make(chan outbound, 64),
dataQ: make(chan outbound, 256),
done: make(chan struct{}), done: make(chan struct{}),
conns: map[uint32]*visitorConn{}, 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{ open, _ := json.Marshal(openMessage{
Type: "open", Conn: id, Peer: peer, SNI: sni, ALPN: alpn, 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) b.removeConn(id, true)
return return
} }
for len(hello) > 0 { for len(hello) > 0 {
n := min(len(hello), maxPayload) 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) b.removeConn(id, true)
return return
} }
@@ -317,7 +319,7 @@ func (b *Bridge) pump(v *visitorConn) {
n, err := v.conn.Read(buf) n, err := v.conn.Read(buf)
v.touch() v.touch()
if n > 0 { if n > 0 {
if !b.send(websocket.MessageBinary, buildFrame(v.id, buf[:n])) { if !b.sendData(websocket.MessageBinary, buildFrame(v.id, buf[:n])) {
return return
} }
} }
@@ -332,15 +334,17 @@ func (b *Bridge) pump(v *visitorConn) {
} }
if err == io.EOF { if err == io.EOF {
end, _ := json.Marshal(endMessage{Type: "end", Conn: v.id}) end, _ := json.Marshal(endMessage{Type: "end", Conn: v.id})
b.send(websocket.MessageText, end) b.sendData(websocket.MessageText, end)
} else { } else {
reset, _ := json.Marshal(resetMessage{Type: "reset", Conn: v.id, Code: ResetIOError}) 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) b.removeConn(v.id, true)
return return
} }
if bothEnded { if bothEnded {
b.removeConn(v.id, false) b.removeConn(v.id, false)
} else {
b.scheduleHalfClose(v.id, v)
} }
return return
} }
@@ -387,7 +391,7 @@ func (b *Bridge) handleControl(raw []byte) {
switch msg.Type { switch msg.Type {
case "ping": case "ping":
pong, _ := json.Marshal(pingPongMessage{Type: "pong", TimeSent: msg.TimeSent}) pong, _ := json.Marshal(pingPongMessage{Type: "pong", TimeSent: msg.TimeSent})
b.send(websocket.MessageText, pong) b.sendControl(websocket.MessageText, pong)
case "end": case "end":
if msg.Conn != nil { if msg.Conn != nil {
b.endFromClient(*msg.Conn) b.endFromClient(*msg.Conn)
@@ -416,19 +420,48 @@ func (b *Bridge) endFromClient(connID uint32) {
} }
if bothEnded { if bothEnded {
b.removeConn(connID, false) 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() { func (b *Bridge) writeLoop() {
ctx := context.Background() 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 { for {
select { select {
case <-b.done: case <-b.done:
return return
case msg := <-b.outbox: default:
if err := writeWS(ctx, b.ws, msg.typ, msg.data); err != nil { }
b.close(websocket.StatusInternalError, "write failed") 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 return
} }
} }
@@ -459,28 +492,52 @@ func (b *Bridge) sweeper() {
b.mu.Unlock() b.mu.Unlock()
for _, id := range stale { for _, id := range stale {
reset, _ := json.Marshal(resetMessage{Type: "reset", Conn: id, Code: ResetIdleTimeout}) reset, _ := json.Marshal(resetMessage{Type: "reset", Conn: id, Code: ResetIdleTimeout})
b.send(websocket.MessageText, reset) b.sendControl(websocket.MessageText, reset)
b.removeConn(id, true) b.removeConn(id, true)
} }
} }
} }
} }
// send queues a frame, applying backpressure when the client is slow. // sendControl queues a control frame. It never blocks behind visitor data; a
// It returns false when the bridge is gone. // stalled client whose control queue is full is closed.
func (b *Bridge) send(typ websocket.MessageType, data []byte) bool { func (b *Bridge) sendControl(typ websocket.MessageType, data []byte) bool {
select { select {
case <-b.done: case <-b.done:
return false return false
case b.outbox <- outbound{typ, data}: default:
}
select {
case b.control <- outbound{typ, data}:
return true 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) { func (b *Bridge) trySend(typ websocket.MessageType, data []byte) {
select { select {
case b.outbox <- outbound{typ, data}: case b.control <- outbound{typ, data}:
default: default:
} }
} }
@@ -490,6 +547,10 @@ func (b *Bridge) removeConn(id uint32, rst bool) {
b.mu.Lock() b.mu.Lock()
v := b.conns[id] v := b.conns[id]
delete(b.conns, id) delete(b.conns, id)
if v != nil && v.halfClose != nil {
v.halfClose.Stop()
v.halfClose = nil
}
b.mu.Unlock() b.mu.Unlock()
if v == nil { if v == nil {
return return
@@ -502,6 +563,31 @@ func (b *Bridge) removeConn(id uint32, rst bool) {
v.conn.Close() 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() { func (b *Bridge) touch() {
b.lastActivity.Store(time.Now().UnixMilli()) b.lastActivity.Store(time.Now().UnixMilli())
} }
@@ -531,6 +617,7 @@ type visitorConn struct {
visitorEnded bool // pump saw EOF: no more visitor data visitorEnded bool // pump saw EOF: no more visitor data
clientEnded bool // client sent end: target will send no more clientEnded bool // client sent end: target will send no more
halfClose *time.Timer
lastActivity atomic.Int64 lastActivity atomic.Int64
} }
+94
View File
@@ -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
}
+64
View File
@@ -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)
}
}
}
+7
View File
@@ -14,6 +14,13 @@ type DNSProvider interface {
DeleteTXT(ctx context.Context, name string) error 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 // challengeRecordName is the TXT record name a hostname's DNS-01 challenge
// lives at; the wildcard of a hostname shares the same record. // lives at; the wildcard of a hostname shares the same record.
func challengeRecordName(hostname string) string { func challengeRecordName(hostname string) string {
+61 -11
View File
@@ -7,6 +7,7 @@ import (
"fmt" "fmt"
"io" "io"
"net/http" "net/http"
"strings"
"time" "time"
) )
@@ -36,38 +37,38 @@ func (p *PorkbunDNS) client() *http.Client {
return &http.Client{Timeout: 20 * time.Second} 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["apikey"] = p.APIKey
body["secretapikey"] = p.SecretKey body["secretapikey"] = p.SecretKey
raw, err := json.Marshal(body) raw, err := json.Marshal(body)
if err != nil { if err != nil {
return err return nil, err
} }
req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.base()+path, bytes.NewReader(raw)) req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.base()+path, bytes.NewReader(raw))
if err != nil { if err != nil {
return err return nil, err
} }
req.Header.Set("Content-Type", "application/json") req.Header.Set("Content-Type", "application/json")
resp, err := p.client().Do(req) resp, err := p.client().Do(req)
if err != nil { if err != nil {
return err return nil, err
} }
defer resp.Body.Close() 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 { 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 { var result struct {
Status string `json:"status"` Status string `json:"status"`
Message string `json:"message"` Message string `json:"message"`
} }
if err := json.Unmarshal(data, &result); err != nil { 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" { 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 { 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 { if err != nil {
return err 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", "name": relative, "type": "TXT", "content": value, "ttl": "600",
}) })
return err
} }
func (p *PorkbunDNS) DeleteTXT(ctx context.Context, name string) error { 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 { if err != nil {
return err 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
} }
+43
View File
@@ -72,3 +72,46 @@ func TestPorkbunDNS(t *testing.T) {
t.Errorf("expected the API error, got %v", err) 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])
}
}
}
+18 -2
View File
@@ -72,6 +72,7 @@ type acmeIssuer struct {
store *Store store *Store
opts ACMEOptions opts ACMEOptions
sem chan struct{} // bounds concurrent ACME orders sem chan struct{} // bounds concurrent ACME orders
cool cooldown
accountMu sync.Mutex // serializes first-use account registration accountMu sync.Mutex // serializes first-use account registration
} }
@@ -85,7 +86,7 @@ func NewACMEIssuer(store *Store, opts ACMEOptions) (Issuer, error) {
if opts.DNS == nil { if opts.DNS == nil {
return nil, errors.New("acme: a DNS provider is required") 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 } 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 // 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 // CSR asks for it, its wildcard) using one DNS-01 challenge record shared by
// both, then returns the certificate for the client's key. // 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) csr, err := parseCSR(csrPEM, hostname)
if err != nil { if err != nil {
return IssuedCert{}, err 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 { select {
case a.sem <- struct{}{}: case a.sem <- struct{}{}:
case <-ctx.Done(): case <-ctx.Done():
return IssuedCert{}, ctx.Err() return IssuedCert{}, ctx.Err()
} }
defer func() { <-a.sem }() 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) client, err := a.client(ctx)
if err != nil { if err != nil {
+99
View File
@@ -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)
}
+83
View File
@@ -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")
}
}
+6
View File
@@ -97,6 +97,12 @@ type errUnauthorized struct {
Message string `json:"message"` 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 { type errTaggedID struct {
Tag string `json:"_tag"` Tag string `json:"_tag"`
TunnelID string `json:"tunnelID"` TunnelID string `json:"tunnelID"`
+1 -1
View File
@@ -69,7 +69,7 @@
{#if !token} {#if !token}
<p class="dim">Enter the ADMIN_TOKEN to view the dashboard.</p> <p class="dim">Enter the ADMIN_TOKEN to view the dashboard.</p>
{:else if overview} {:else if overview}
<SummaryCards server={overview.server} ca={overview.ca} totals={overview.totals} /> <SummaryCards server={overview.server} ca={overview.ca} totals={overview.totals} metrics={overview.metrics} />
<h2>Tunnels</h2> <h2>Tunnels</h2>
<TunnelsTable tunnels={overview.tunnels} /> <TunnelsTable tunnels={overview.tunnels} />
<h2>Bridges</h2> <h2>Bridges</h2>
+20 -2
View File
@@ -1,8 +1,8 @@
<script lang="ts"> <script lang="ts">
import type { CaInfo, ServerInfo, Totals } from "./api"; import type { CaInfo, Metrics, ServerInfo, Totals } from "./api";
import { fmtDate, fmtUptime } from "./format"; import { fmtDate, fmtUptime } from "./format";
let { server, ca, totals }: { server: ServerInfo; ca: CaInfo; totals: Totals } = $props(); let { server, ca, totals, metrics }: { server: ServerInfo; ca: CaInfo; totals: Totals; metrics: Metrics } = $props();
</script> </script>
<div class="cards"> <div class="cards">
@@ -10,6 +10,19 @@
<div class="card"><div class="label">Online</div><div class="value">{totals.online}</div></div> <div class="card"><div class="label">Online</div><div class="value">{totals.online}</div></div>
<div class="card"><div class="label">Bridges</div><div class="value">{totals.bridges}</div></div> <div class="card"><div class="label">Bridges</div><div class="value">{totals.bridges}</div></div>
<div class="card"><div class="label">Visitors</div><div class="value">{totals.visitors}</div></div> <div class="card"><div class="label">Visitors</div><div class="value">{totals.visitors}</div></div>
<div class="card">
<div class="label">Certificates</div>
<div class="value">{metrics.certificates_ready}</div>
<div class="detail">
{metrics.certificates_issuing} issuing · {metrics.certificates_failed} failed
{#if metrics.certificates_retrying > 0}· {metrics.certificates_retrying} retrying{/if}
</div>
</div>
<div class="card">
<div class="label">Issuances</div>
<div class="value">{metrics.issuance_ready}</div>
<div class="detail">{metrics.issuance_attempts} attempts · {metrics.rate_limited} limited</div>
</div>
<div class="card"><div class="label">Local CA expires</div><div class="value small">{fmtDate(ca.expires)}</div></div> <div class="card"><div class="label">Local CA expires</div><div class="value small">{fmtDate(ca.expires)}</div></div>
<div class="card"><div class="label">Uptime</div><div class="value small">{fmtUptime(server.uptime_seconds)}</div></div> <div class="card"><div class="label">Uptime</div><div class="value small">{fmtUptime(server.uptime_seconds)}</div></div>
</div> </div>
@@ -25,4 +38,9 @@
font-size: 14px; font-size: 14px;
margin-top: 8px; margin-top: 8px;
} }
.detail {
margin-top: 6px;
color: #8b949e;
font-size: 11px;
}
</style> </style>
+14
View File
@@ -19,6 +19,19 @@ export interface Totals {
visitors: number; 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 { export interface Tunnel {
id: string; id: string;
hostname: string; hostname: string;
@@ -46,6 +59,7 @@ export interface Overview {
server: ServerInfo; server: ServerInfo;
ca: CaInfo; ca: CaInfo;
totals: Totals; totals: Totals;
metrics: Metrics;
tunnels: Tunnel[] | null; tunnels: Tunnel[] | null;
bridges: Bridge[] | null; bridges: Bridge[] | null;
} }