tunnel/internal/server/api.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)
	})
}