TZGyn
ebae420ab1
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.
519 lines
16 KiB
Go
519 lines
16 KiB
Go
package server
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"log"
|
|
"mime"
|
|
"net/http"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
// API is the HTTP control plane. Its answers match the recorded contract in
|
|
// worker-contract.json: paths, status codes, bodies, and JSON key order.
|
|
type API struct {
|
|
Store *Store
|
|
Hub *Hub
|
|
Domain string
|
|
|
|
AdminToken string // bearer token for /api/admin/*; empty disables it
|
|
StartedAt time.Time // for the uptime shown on the dashboard
|
|
Version string
|
|
|
|
Issuer Issuer // certificate issuer; defaults to the local CA
|
|
IssuerName string // "local" or the ACME directory URL
|
|
|
|
// Limits are optional; a nil limiter allows everything.
|
|
CreateLimiter *RateLimiter
|
|
BindLimiter *RateLimiter
|
|
BindTunnelLimiter *RateLimiter
|
|
|
|
issueMu sync.Mutex
|
|
issuing map[string]bool
|
|
|
|
createCount atomic.Uint64
|
|
bindCount atomic.Uint64
|
|
rateLimited atomic.Uint64
|
|
issueAttempts atomic.Uint64
|
|
issueReady atomic.Uint64
|
|
issueFailed atomic.Uint64
|
|
}
|
|
|
|
func (a *API) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
// The reference tolerates duplicate slashes and a trailing slash.
|
|
path := "/" + strings.Join(strings.FieldsFunc(r.URL.Path, func(c rune) bool { return c == '/' }), "/")
|
|
|
|
switch {
|
|
case path == "/health" && r.Method == http.MethodGet:
|
|
w.Header().Set("Content-Type", "application/json;charset=utf-8")
|
|
w.Write([]byte(`{"ok":true}`))
|
|
case path == "/openapi.json" && r.Method == http.MethodGet:
|
|
writeJSON(w, http.StatusOK, json.RawMessage(openapiStub))
|
|
case (path == "/" || path == "/admin" || strings.HasPrefix(path, "/assets/")) && r.Method == http.MethodGet:
|
|
a.dashboard(w, r)
|
|
case path == "/api/admin/overview" && r.Method == http.MethodGet:
|
|
a.adminOverviewHandler(w, r)
|
|
case path == "/api/tunnel" && r.Method == http.MethodPost:
|
|
a.createTunnel(w, r)
|
|
default:
|
|
a.tunnelRoutes(w, r, strings.Split(strings.TrimPrefix(path, "/"), "/"))
|
|
}
|
|
}
|
|
|
|
// tunnelRoutes routes /api/tunnel/{id}[/certificate|connect].
|
|
func (a *API) tunnelRoutes(w http.ResponseWriter, r *http.Request, parts []string) {
|
|
if len(parts) < 2 || parts[0] != "api" || parts[1] != "tunnel" {
|
|
notFoundEmpty(w)
|
|
return
|
|
}
|
|
switch len(parts) {
|
|
case 3:
|
|
id := parts[2]
|
|
switch r.Method {
|
|
case http.MethodGet, http.MethodHead:
|
|
a.getTunnel(w, r, id)
|
|
case http.MethodDelete:
|
|
a.deleteTunnel(w, r, id)
|
|
default:
|
|
notFoundEmpty(w)
|
|
}
|
|
case 4:
|
|
id := parts[2]
|
|
switch parts[3] {
|
|
case "certificate":
|
|
switch r.Method {
|
|
case http.MethodPost:
|
|
a.bindCertificate(w, r, id)
|
|
case http.MethodGet, http.MethodHead:
|
|
a.getCertificate(w, r, id)
|
|
default:
|
|
notFoundEmpty(w)
|
|
}
|
|
case "connect":
|
|
if r.Method != http.MethodGet {
|
|
notFoundEmpty(w)
|
|
return
|
|
}
|
|
a.connect(w, r, id)
|
|
default:
|
|
notFoundEmpty(w)
|
|
}
|
|
default:
|
|
notFoundEmpty(w)
|
|
}
|
|
}
|
|
|
|
// ---- POST /api/tunnel
|
|
|
|
func (a *API) createTunnel(w http.ResponseWriter, r *http.Request) {
|
|
body, ok := readJSONBody(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
if name, present := body["name"]; present && name != nil {
|
|
if _, isString := name.(string); !isString {
|
|
w.WriteHeader(http.StatusBadRequest) // wrong type: 400, empty body
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusBadRequest, errInvalidRequest{
|
|
Tag: "InvalidRequestError",
|
|
Message: "Custom tunnel names are not supported; omit name to get a random hostname",
|
|
})
|
|
return
|
|
}
|
|
if ok, retry := a.CreateLimiter.Allow(clientIP(r)); !ok {
|
|
a.rateLimited.Add(1)
|
|
w.Header().Set("Retry-After", retryAfterSeconds(retry))
|
|
writeJSON(w, http.StatusTooManyRequests, errRateLimited{
|
|
Tag: "RateLimitedError", Message: "Too many tunnels created from this address; try again later",
|
|
})
|
|
return
|
|
}
|
|
tunnel, token, err := a.Store.createTunnel(a.Domain)
|
|
if err != nil {
|
|
writeJSON(w, http.StatusServiceUnavailable, errInvalidRequest{
|
|
Tag: "ServiceUnavailableError", Message: "storage unavailable",
|
|
})
|
|
return
|
|
}
|
|
a.createCount.Add(1)
|
|
writeJSON(w, http.StatusCreated, createResponse{
|
|
Tunnel: TunnelInfo{ID: tunnel.ID, Hostname: tunnel.Hostname, State: "offline"},
|
|
Token: token,
|
|
})
|
|
}
|
|
|
|
// ---- GET /api/tunnel/{id}
|
|
|
|
func (a *API) getTunnel(w http.ResponseWriter, r *http.Request, id string) {
|
|
tunnel, ok := a.lookupTunnel(w, id)
|
|
if !ok || !a.authorized(w, r, tunnel) {
|
|
return
|
|
}
|
|
info := TunnelInfo{ID: tunnel.ID, Hostname: tunnel.Hostname, State: "offline"}
|
|
if a.Hub.online(tunnel.ID) {
|
|
info.State = "online"
|
|
}
|
|
if tunnel.Certificate != nil {
|
|
info.CertificateID = tunnel.Certificate.ID
|
|
}
|
|
writeJSON(w, http.StatusOK, info)
|
|
}
|
|
|
|
// ---- DELETE /api/tunnel/{id}
|
|
|
|
func (a *API) deleteTunnel(w http.ResponseWriter, r *http.Request, id string) {
|
|
tunnel, ok := a.lookupTunnel(w, id)
|
|
if !ok || !a.authorized(w, r, tunnel) {
|
|
return
|
|
}
|
|
a.Store.remove(id)
|
|
a.Hub.closeTunnel(id)
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}
|
|
|
|
// ---- POST /api/tunnel/{id}/certificate
|
|
|
|
func (a *API) bindCertificate(w http.ResponseWriter, r *http.Request, id string) {
|
|
body, ok := readJSONBody(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
tunnel, ok := a.lookupTunnel(w, id)
|
|
if !ok || !a.authorized(w, r, tunnel) {
|
|
return
|
|
}
|
|
csrValue, present := body["csr"]
|
|
if !present {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
return
|
|
}
|
|
csr, isString := csrValue.(string)
|
|
if !isString {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
return
|
|
}
|
|
// Binding the CSR the tunnel already has answers with its current state;
|
|
// a different CSR re-issues. A failed issuance for the same CSR retries.
|
|
cert := tunnel.Certificate
|
|
if cert != nil && cert.CSR == csr {
|
|
if certReady(cert) {
|
|
a.writeCertificate(w, http.StatusAccepted, cert)
|
|
return
|
|
}
|
|
if cert.State != "failed" {
|
|
// Same CSR, issuance incomplete: resume it (for example after a
|
|
// restart) and answer with its current state.
|
|
a.startIssue(tunnel.ID, tunnel.Hostname, csr, cert.ID)
|
|
a.writeCertificate(w, http.StatusAccepted, cert)
|
|
return
|
|
}
|
|
}
|
|
if (cert != nil && cert.State == "issuing") || a.isIssuing(id) {
|
|
writeJSON(w, http.StatusConflict, errTaggedID{
|
|
Tag: "CertificateInProgressError", TunnelID: id,
|
|
Message: "Certificate issuance is already in progress",
|
|
})
|
|
return
|
|
}
|
|
// A bind that reaches here creates a new certificate (or retries a failed
|
|
// one), so it counts against the issuance limits.
|
|
if ok, retry := a.BindLimiter.Allow(clientIP(r)); !ok {
|
|
a.rateLimit(w, retry)
|
|
return
|
|
}
|
|
if ok, retry := a.BindTunnelLimiter.Allow(id); !ok {
|
|
a.rateLimit(w, retry)
|
|
return
|
|
}
|
|
a.bindCount.Add(1)
|
|
if _, err := parseCSR(csr, tunnel.Hostname); err != nil {
|
|
var hostErr errCSRHostname
|
|
if errors.As(err, &hostErr) {
|
|
writeJSON(w, http.StatusBadRequest, errInvalidHostname{
|
|
Tag: "InvalidHostnameError",
|
|
Provided: hostErr.provided,
|
|
Expected: hostErr.expected,
|
|
Message: "CSR hostname does not match tunnel hostname",
|
|
})
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusBadRequest, errInvalidRequest{
|
|
Tag: "InvalidRequestError", Message: "Failed to parse CSR",
|
|
})
|
|
return
|
|
}
|
|
fresh := storedCert{ID: certID(), CSR: csr, State: "issuing", Issuer: a.issuerName()}
|
|
if err := a.Store.bindCertificate(id, fresh); err != nil {
|
|
notFoundEmpty(w)
|
|
return
|
|
}
|
|
a.startIssue(tunnel.ID, tunnel.Hostname, csr, fresh.ID)
|
|
writeJSON(w, http.StatusAccepted, CertificateInfo{ID: fresh.ID, State: CertState{Type: "issuing"}})
|
|
}
|
|
|
|
// ---- GET /api/tunnel/{id}/certificate
|
|
|
|
func (a *API) getCertificate(w http.ResponseWriter, r *http.Request, id string) {
|
|
tunnel, ok := a.lookupTunnel(w, id)
|
|
if !ok || !a.authorized(w, r, tunnel) {
|
|
return
|
|
}
|
|
if tunnel.Certificate == nil {
|
|
writeJSON(w, http.StatusNotFound, errTaggedID{
|
|
Tag: "CertificateNotFoundError", TunnelID: id, Message: "Certificate not found",
|
|
})
|
|
return
|
|
}
|
|
a.writeCertificate(w, http.StatusOK, tunnel.Certificate)
|
|
}
|
|
|
|
func (a *API) writeCertificate(w http.ResponseWriter, status int, cert *storedCert) {
|
|
writeJSON(w, status, CertificateInfo{ID: cert.ID, State: certState(cert)})
|
|
}
|
|
|
|
// certState maps a stored certificate to the wire state.
|
|
func certState(cert *storedCert) CertState {
|
|
switch {
|
|
case cert == nil:
|
|
return CertState{Type: "failed", Reason: "no certificate"}
|
|
case certReady(cert):
|
|
return CertState{Type: "ready", Certificate: cert.Certificate, Chain: cert.Chain, Expiry: cert.Expiry}
|
|
case cert.State == "failed":
|
|
return CertState{Type: "failed", Reason: cert.Reason}
|
|
default:
|
|
return CertState{Type: "issuing"}
|
|
}
|
|
}
|
|
|
|
// rateLimit answers a 429 with the contract's error tag and a Retry-After.
|
|
func (a *API) rateLimit(w http.ResponseWriter, retryAfter time.Duration) {
|
|
a.rateLimited.Add(1)
|
|
w.Header().Set("Retry-After", retryAfterSeconds(retryAfter))
|
|
writeJSON(w, http.StatusTooManyRequests, errRateLimited{
|
|
Tag: "RateLimitedError", Message: "Rate limit exceeded; try again later",
|
|
})
|
|
}
|
|
|
|
// issuer returns the configured certificate issuer (the local CA by default).
|
|
func (a *API) issuer() Issuer {
|
|
if a.Issuer != nil {
|
|
return a.Issuer
|
|
}
|
|
return NewLocalIssuer(a.Store)
|
|
}
|
|
|
|
// issuerName identifies the current issuer, so certificates from a previous
|
|
// issuer (say the local CA) are re-issued once a real one is configured.
|
|
func (a *API) issuerName() string {
|
|
if a.IssuerName != "" {
|
|
return a.IssuerName
|
|
}
|
|
return "local"
|
|
}
|
|
|
|
// runIssue issues one certificate and records the outcome. At most one
|
|
// issuance runs per tunnel.
|
|
func (a *API) runIssue(tunnelID, hostname, csrPEM, certID string) {
|
|
if !a.beginIssue(tunnelID) {
|
|
return
|
|
}
|
|
defer a.endIssue(tunnelID)
|
|
|
|
// Drop an issuance whose bind was replaced before this run started (a
|
|
// renewal racing a new bind).
|
|
current, ok := a.Store.get(tunnelID)
|
|
if !ok || current.Certificate == nil || current.Certificate.ID != certID || current.Certificate.CSR != csrPEM {
|
|
log.Printf("skipping certificate issuance for %s: certificate %s is no longer current", hostname, certID)
|
|
return
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
|
|
defer cancel()
|
|
log.Printf("issuing certificate for %s via %s", hostname, a.issuerName())
|
|
a.issueAttempts.Add(1)
|
|
result, err := a.issuer().Issue(ctx, csrPEM, hostname)
|
|
|
|
// The tunnel may have been deleted, or bound a different CSR, while the
|
|
// order ran. Storing this result would clobber the newer certificate.
|
|
current, ok = a.Store.get(tunnelID)
|
|
if !ok || current.Certificate == nil || current.Certificate.ID != certID || current.Certificate.CSR != csrPEM {
|
|
log.Printf("discarding certificate outcome for %s: certificate %s is no longer current", hostname, certID)
|
|
return
|
|
}
|
|
|
|
cert := storedCert{ID: certID, CSR: csrPEM, Issuer: a.issuerName()}
|
|
if err != nil {
|
|
a.issueFailed.Add(1)
|
|
cert.State = "failed"
|
|
cert.Reason = err.Error()
|
|
if certReady(current.Certificate) {
|
|
// A renewal failed but the old certificate is still valid and
|
|
// serving: keep it, and let the renewal loop retry.
|
|
kept := *current.Certificate
|
|
kept.State = "failed"
|
|
kept.Reason = cert.Reason
|
|
cert = kept
|
|
}
|
|
log.Printf("certificate for %s failed: %v", hostname, err)
|
|
} else {
|
|
cert.State = "ready"
|
|
cert.Certificate = result.Certificate
|
|
cert.Chain = result.Chain
|
|
cert.Expiry = expiryString(result.Expiry)
|
|
a.issueReady.Add(1)
|
|
log.Printf("certificate for %s ready (expires %s)", hostname, cert.Expiry)
|
|
}
|
|
if err := a.Store.bindCertificate(tunnelID, cert); err != nil {
|
|
log.Printf("storing certificate for %s: %v", hostname, err)
|
|
}
|
|
}
|
|
|
|
func (a *API) beginIssue(tunnelID string) bool {
|
|
a.issueMu.Lock()
|
|
defer a.issueMu.Unlock()
|
|
if a.issuing == nil {
|
|
a.issuing = map[string]bool{}
|
|
}
|
|
if a.issuing[tunnelID] {
|
|
return false
|
|
}
|
|
a.issuing[tunnelID] = true
|
|
return true
|
|
}
|
|
|
|
// startIssue runs issuance in the background for ACME, and synchronously for
|
|
// the local CA so the next GET already answers ready.
|
|
func (a *API) startIssue(tunnelID, hostname, csrPEM, certID string) {
|
|
if a.issuer().Instant() {
|
|
a.runIssue(tunnelID, hostname, csrPEM, certID)
|
|
return
|
|
}
|
|
go a.runIssue(tunnelID, hostname, csrPEM, certID)
|
|
}
|
|
|
|
func (a *API) isIssuing(tunnelID string) bool {
|
|
a.issueMu.Lock()
|
|
defer a.issueMu.Unlock()
|
|
return a.issuing[tunnelID]
|
|
}
|
|
|
|
func (a *API) endIssue(tunnelID string) {
|
|
a.issueMu.Lock()
|
|
defer a.issueMu.Unlock()
|
|
delete(a.issuing, tunnelID)
|
|
}
|
|
|
|
// ---- GET /api/tunnel/{id}/connect
|
|
|
|
func (a *API) connect(w http.ResponseWriter, r *http.Request, id string) {
|
|
tunnel, ok := a.Store.get(id)
|
|
if !ok {
|
|
w.WriteHeader(http.StatusNotFound)
|
|
w.Write([]byte("Tunnel not found"))
|
|
return
|
|
}
|
|
if !upgradeRequested(r) {
|
|
w.WriteHeader(http.StatusUpgradeRequired)
|
|
w.Write([]byte("Expected WebSocket upgrade"))
|
|
return
|
|
}
|
|
if !subprotocolOffered(r) {
|
|
w.WriteHeader(http.StatusUpgradeRequired)
|
|
w.Write([]byte("Expected opentunnel WebSocket subprotocol"))
|
|
return
|
|
}
|
|
a.Hub.attach(w, r, tunnel)
|
|
}
|
|
|
|
// ---- helpers
|
|
|
|
// lookupTunnel answers 404 TunnelNotFoundError when id is unknown. Existence
|
|
// is checked before authorization, as the contract records.
|
|
func (a *API) lookupTunnel(w http.ResponseWriter, id string) (storedTunnel, bool) {
|
|
tunnel, ok := a.Store.get(id)
|
|
if !ok {
|
|
writeJSON(w, http.StatusNotFound, errTaggedID{
|
|
Tag: "TunnelNotFoundError", TunnelID: id, Message: "Tunnel not found",
|
|
})
|
|
return tunnel, false
|
|
}
|
|
return tunnel, true
|
|
}
|
|
|
|
// authorized answers 401 UnauthorizedError unless the request carries the
|
|
// tunnel's bearer token (the scheme is case-insensitive).
|
|
func (a *API) authorized(w http.ResponseWriter, r *http.Request, tunnel storedTunnel) bool {
|
|
scheme, token, _ := strings.Cut(r.Header.Get("Authorization"), " ")
|
|
if strings.EqualFold(scheme, "bearer") && token != "" && tokenMatches(token, tunnel.TokenHash) {
|
|
return true
|
|
}
|
|
writeJSON(w, http.StatusUnauthorized, errUnauthorized{
|
|
Tag: "UnauthorizedError", Message: "Invalid bearer token",
|
|
})
|
|
return false
|
|
}
|
|
|
|
// readJSONBody enforces the contract's body rules: a non-JSON content type is
|
|
// 415, a missing, empty, or malformed body is 400 with an empty body.
|
|
func readJSONBody(w http.ResponseWriter, r *http.Request) (map[string]any, bool) {
|
|
if ct := r.Header.Get("Content-Type"); ct != "" {
|
|
mediaType, _, err := mime.ParseMediaType(ct)
|
|
if err != nil || !strings.EqualFold(mediaType, "application/json") {
|
|
w.Header().Set("Content-Type", "text/plain")
|
|
w.WriteHeader(http.StatusUnsupportedMediaType)
|
|
w.Write([]byte("Unsupported content-type: " + ct))
|
|
return nil, false
|
|
}
|
|
}
|
|
raw, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 1<<20))
|
|
if err != nil || len(raw) == 0 {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
return nil, false
|
|
}
|
|
var body map[string]any
|
|
if err := json.Unmarshal(raw, &body); err != nil {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
return nil, false
|
|
}
|
|
return body, true
|
|
}
|
|
|
|
func writeJSON(w http.ResponseWriter, status int, body any) {
|
|
raw, err := json.Marshal(body)
|
|
if err != nil {
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(status)
|
|
w.Write(raw)
|
|
}
|
|
|
|
// notFoundEmpty is the 404 with no body the contract records for unknown
|
|
// paths and methods.
|
|
func notFoundEmpty(w http.ResponseWriter) {
|
|
w.WriteHeader(http.StatusNotFound)
|
|
}
|
|
|
|
const openapiStub = `{"openapi":"3.1.0","info":{"title":"OpenTunnel API","version":"0.1.0","description":"Create and manage blind TLS tunnels."},"paths":{"/api/tunnel":{"post":{"responses":{"201":{"description":"Success"}}}},"/api/tunnel/{id}":{"get":{"responses":{"200":{"description":"Tunnel/Info"}}},"delete":{"responses":{"204":{"description":"Deleted"}}}},"/api/tunnel/{id}/certificate":{"post":{"responses":{"202":{"description":"Accepted"}}},"get":{"responses":{"200":{"description":"Certificate"}}}},"/api/tunnel/{id}/connect":{"get":{"responses":{"101":{"description":"WebSocket upgrade"}}}}}}`
|
|
|
|
// redirectHandler is HTTP_MODE=redirect: /health answers, everything else
|
|
// permanently redirects to the API domain.
|
|
func RedirectHandler(domain string) http.Handler {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.URL.Path == "/health" {
|
|
w.Header().Set("Content-Type", "application/json;charset=utf-8")
|
|
w.Write([]byte(`{"ok":true}`))
|
|
return
|
|
}
|
|
target := "https://" + domain + r.URL.RequestURI()
|
|
http.Redirect(w, r, target, http.StatusMovedPermanently)
|
|
})
|
|
}
|