mesh-controller/examples/route-proxy/main.go

// A reverse proxy, in the form the mesh expects one.
//
// **A route is a grant** (novox/hq ADR 0007, 08-connectivity §3). A module that must be reachable
// declares it requires `route` and contributes the name it wants; the proxy provides `route` and
// receives every consumer that asked. It is the mirror of a database grant: there the consumer
// supplies a name and receives credentials, here it supplies a target and receives a name.
//
// **It is an example, not part of the control plane.** The control plane decides and never touches
// a machine. A real deployment runs whatever proxy it likes — the contract is the file, not this
// program. What lives here is that contract, written as something that runs so it can be read
// rather than described.
//
// What it is given, written by the host from an ordinary declaration:
//
//	$ROUTES    every consumer, the name it asked for, where the mesh says that machine is, and
//	           the largest request body it will take (`max-request-body`, bytes; absent is no limit)
//
// It re-reads on change rather than being restarted, for the same reason the provisioner does:
// a route arriving or leaving is an ordinary event and must not drop the connections of every
// other workload.
package main

import (
	"bytes"
	"context"
	"crypto/sha256"
	"crypto/tls"
	"crypto/x509"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"math"
	"net"
	"net/http"
	"net/http/httputil"
	"net/url"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

	"golang.org/x/crypto/acme"
	"golang.org/x/crypto/acme/autocert"
)

// Where public certificates come from when nothing says otherwise.
//
// **Staging, deliberately** (novox/hq 04-ISSUES/004). Production issuance is rate-limited per
// domain and per account, the quota does not replenish quickly, and exhausting it removes the
// ability to issue a certificate somebody actually needs. Defaulting to production would leave
// the safe path depending on remembering to opt out of it, on exactly the work most likely to
// iterate — standing up a node, changing how names resolve.
//
// A staging certificate is trusted by no browser, so the mistake announces itself on the first
// request rather than a fortnight later at the rate limit.
const stagingDirectory = "https://acme-staging-v02.api.letsencrypt.org/directory"

// issuer is the ACME directory to ask. The lab points this at its own issuer; a node serving real
// traffic points it at production, and says so explicitly.
func issuer() string {
	if named := strings.TrimSpace(os.Getenv("ACME_DIRECTORY")); named != "" {
		return named
	}
	return stagingDirectory
}

// onlyWhatTheMeshSaid refuses to obtain a certificate for a name this proxy was not given.
//
// **The policy that stops a quota from being spent by accident.** Without it, anything that can
// reach port 443 and send a name triggers an issuance attempt for it — so a scan, or one
// misconfigured client, becomes a stream of failed orders against the account's rate limit. What
// this proxy may certify is exactly what the mesh told it to route, which is already the answer
// to what it may serve.
func onlyWhatTheMeshSaid(held *table) autocert.HostPolicy {
	return func(_ context.Context, host string) error {
		if _, known := held.find(host); known {
			return nil
		}
		return fmt.Errorf("no route for %q in this mesh, so no certificate is asked for", host)
	}
}

// what the mesh writes: the contributions file, one entry per consumer.
type given struct {
	Given []contribution `json:"given"`
}

type contribution struct {
	From string `json:"from"`
	Node string `json:"node"`
	// At is where that machine is on the private network. The mesh knows it; a proxy that had to
	// build it from the node name would be a naming convention copied into every provider.
	At     string         `json:"at"`
	Values map[string]any `json:"values"`
}

// route is one name the proxy serves: where it goes, and what it will not carry there.
type route struct {
	// Target is the workload, as a URL.
	Target string
	// MaxRequestBody is the largest request body, in bytes, this route accepts — the
	// contribution's `max-request-body`. Zero is no limit, which is what a route that said nothing
	// gets: the mesh's own proxy has never limited a body, and a default that appeared with the
	// field would have broken every route that did not ask for one.
	MaxRequestBody int64
}

// served is a route the proxy has built: the reverse proxy, and the limit it enforces in front.
type served struct {
	proxy *httputil.ReverseProxy
	limit int64
}

// table is what the proxy is currently serving, replaced whole whenever the file changes.
//
// Replaced rather than merged: the file is the whole truth about who has a route, so merging
// would keep serving a name whose module was unassigned — which is the stale-route fault
// 08-connectivity lists as open, reintroduced one level down.
type table struct {
	mu      sync.RWMutex
	to      map[string]served
	targets map[string]route
}

func (t *table) set(routes map[string]route) {
	made := map[string]served{}
	for name, r := range routes {
		where, err := url.Parse(r.Target)
		if err != nil {
			log.Printf("route %s points at %q, which is not a URL: %v", name, r.Target, err)
			continue
		}
		proxy := httputil.NewSingleHostReverseProxy(where)
		proxy.ErrorHandler = tooLargeOrBadGateway
		made[name] = served{proxy: proxy, limit: r.MaxRequestBody}
	}
	t.mu.Lock()
	t.to, t.targets = made, routes
	t.mu.Unlock()
}

func (t *table) find(host string) (served, bool) {
	// The port is not part of the name. A request to app.example:8080 is for app.example.
	if h, _, err := net.SplitHostPort(host); err == nil {
		host = h
	}
	t.mu.RLock()
	defer t.mu.RUnlock()
	p, ok := t.to[strings.ToLower(host)]
	return p, ok
}

// tooLargeOrBadGateway is what the proxy answers when the workload could not be reached — or when
// it was the request that stopped, because its body ran past the route's limit.
//
// A body read that hits the limit surfaces as the transport's error, which the reverse proxy
// would report as 502 — the workload's fault, in the client's eyes, for a limit the client hit.
// Named as what it was: 413, so a push that is too large is told so rather than told the registry
// is down.
func tooLargeOrBadGateway(w http.ResponseWriter, r *http.Request, err error) {
	var exceeded *http.MaxBytesError
	if errors.As(err, &exceeded) {
		http.Error(w, fmt.Sprintf("the request body for %q is larger than the %d bytes this route "+
			"accepts", r.Host, exceeded.Limit), http.StatusRequestEntityTooLarge)
		return
	}
	log.Printf("http: proxy error: %v", err)
	w.WriteHeader(http.StatusBadGateway)
}

func (t *table) names() []string {
	t.mu.RLock()
	defer t.mu.RUnlock()
	out := make([]string, 0, len(t.targets))
	for name := range t.targets {
		out = append(out, name)
	}
	sort.Strings(out)
	return out
}

func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, "mesh-route-proxy: "+err.Error())
		os.Exit(1)
	}
}

func run() error {
	path := strings.TrimSpace(os.Getenv("ROUTES"))
	if path == "" {
		return fmt.Errorf("ROUTES is not set, so this proxy does not know what it is routing")
	}
	listen := strings.TrimSpace(os.Getenv("LISTEN"))
	if listen == "" {
		listen = ":80"
	}

	held := newTable()
	read := func() {
		routes, err := routesFrom(path)
		if err != nil {
			// Kept serving what it had. A file being rewritten is momentarily unreadable, and
			// dropping every route because one read landed mid-write would turn an ordinary
			// event into an outage.
			log.Printf("cannot read %s, keeping what is already served: %v", path, err)
			return
		}
		held.set(routes)
		log.Printf("serving %d route(s): %s", len(routes), strings.Join(held.names(), ", "))
	}
	read()

	go func() {
		for range time.Tick(2 * time.Second) {
			read()
		}
	}()

	// TLS is opt-in. A proxy with no `TLS_LISTEN` serves plain HTTP exactly as before — which is
	// what an internal-only mesh wants, and what the mesh's own certificate authority already
	// covers for names inside it (novox/hq 08-connectivity). This is for names reachable from
	// outside, where the authority has to be one the world already trusts.
	secure := strings.TrimSpace(os.Getenv("TLS_LISTEN"))
	if secure == "" {
		return http.ListenAndServe(listen, handler(held))
	}

	cache := strings.TrimSpace(os.Getenv("ACME_CACHE"))
	if cache == "" {
		// Refused rather than defaulted. Without somewhere durable to keep them, every restart
		// orders new certificates — which works, silently, until the rate limit says it does not.
		return fmt.Errorf("TLS_LISTEN is set and ACME_CACHE is not: certificates need somewhere " +
			"to persist, or every restart orders them again")
	}
	client := &acme.Client{DirectoryURL: issuer()}
	// An issuer that is not one of the public ones serves its own API over TLS with a certificate
	// nothing trusts yet — the lab's, or an internal step-ca. Trusting it is a deliberate act and
	// names a file, rather than the client being told to skip verification: *skip* would also
	// apply on the day this points at a public issuer, and nothing would say so.
	var root []byte
	if bundle := strings.TrimSpace(os.Getenv("ACME_CA_BUNDLE")); bundle != "" {
		read, err := os.ReadFile(bundle)
		if err != nil {
			return fmt.Errorf("ACME_CA_BUNDLE names %s and it cannot be read: %w", bundle, err)
		}
		root = read
		// An empty bundle means the issuer's root is already in the system trust store — a public
		// authority whose root ships with the OS, pointed at by a provider that serves an empty
		// root (novox/hq ADR 0066). The mesh always writes the bundle file, so it exists and holds
		// nothing; that is the signal to fall back to the system roots, the same as if nothing had
		// named a bundle at all. A file that holds bytes but no certificate is still a
		// misconfiguration and is refused, because there the operator meant to trust something.
		if strings.TrimSpace(string(root)) != "" {
			pool := x509.NewCertPool()
			if !pool.AppendCertsFromPEM(root) {
				return fmt.Errorf("%s holds no certificate this can trust", bundle)
			}
			client.HTTPClient = &http.Client{
				Timeout:   30 * time.Second,
				Transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: pool}},
			}
		}
	}
	// Where this authority's account and certificates are kept. Per authority, not per proxy — see
	// forThisAuthority, which is what makes a re-initialised CA heal itself.
	mine := forThisAuthority(cache, issuer(), root)
	manager := &autocert.Manager{
		Cache:      autocert.DirCache(mine),
		Prompt:     autocert.AcceptTOS,
		HostPolicy: onlyWhatTheMeshSaid(held),
		Client:     client,
	}
	log.Printf("issuing from %s into %s, for whatever the mesh routes here", issuer(), mine)

	// Port 80 answers the HTTP-01 challenge and goes on proxying everything else. The challenge
	// must be answered *at the name being certified*, which is why issuance happens on the node
	// that is publicly reachable rather than wherever the workload runs.
	go func() {
		if err := http.ListenAndServe(listen, manager.HTTPHandler(handler(held))); err != nil {
			log.Printf("plain HTTP stopped: %v", err)
		}
	}()

	server := &http.Server{
		Addr:      secure,
		Handler:   handler(held),
		TLSConfig: manager.TLSConfig(),
	}
	return server.ListenAndServeTLS("", "")
}

// forThisAuthority is where one ACME authority's account and certificates are kept.
//
// **A cached ACME account belongs to the authority that issued it, and nothing in the cache says
// so** (novox/hq ADR 0066). autocert keeps its account key at one fixed name — `acme_account+key` —
// in whatever directory it is given, and reuses it for ever. That is right while the authority stays
// the same and silently wrong the moment it does not: an internal CA that is re-initialised is a new
// authority with a new root, it has never heard of the account in the cache, and every attempt to
// use it is rejected. autocert has no path back from that. Nothing is retried, nothing is
// re-registered, no order ever reaches the CA — issuance simply stops, with no error anybody sees,
// until a person deletes the directory by hand and finds out that was the answer.
//
// So the directory is named after the authority instead of being shared by all of them. The name is
// a digest of the two things that identify one: the directory URL, and the root this proxy was told
// to verify it with. Re-initialising the CA produces a new root; the mesh delivers it as a changed
// bundle; this proxy restarts on that file and lands in a directory with no account in it, so
// autocert registers afresh and orders again. **The healing is that the question "is this account
// still valid" never has to be asked** — an account is only ever found where it is still valid.
//
// It also fixes a latent one of the same shape: pointing ACME_DIRECTORY at production after testing
// against staging reused the staging account, because the cache had no idea they were different.
//
// The old directories stay on disk, unused. Left rather than deleted: they are the only copy of
// certificates that may still be valid, and this program is not the thing that should decide a
// certificate is finished with.
func forThisAuthority(cache, directory string, root []byte) string {
	sum := sha256.Sum256([]byte(directory + "\x00" + string(bytes.TrimSpace(root))))
	return filepath.Join(cache, hex.EncodeToString(sum[:])[:16])
}

// newTable is an empty routing table.
func newTable() *table {
	return &table{to: map[string]served{}, targets: map[string]route{}}
}

// handler is the proxy itself, separated so it can be driven by a test without a listener.
func handler(held *table) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		route, known := held.find(r.Host)
		if !known {
			// **Named, not a bare 404.** A route that was withdrawn and a name that never existed
			// are different things, and a proxy that says only "not found" makes an operator go
			// and read the mesh to tell them apart. What it is serving is the answer to both.
			w.Header().Set("Content-Type", "text/plain; charset=utf-8")
			w.WriteHeader(http.StatusNotFound)
			fmt.Fprintf(w, "no route for %q in this mesh.\nserving: %s\n",
				r.Host, strings.Join(held.names(), ", "))
			return
		}
		if route.limit > 0 {
			// **The limit is the route's, enforced here rather than by the workload** — the
			// contribution says what the proxy may carry to it, which is the one thing the
			// workload cannot say for itself once a proxy stands in front. A body whose length is
			// declared and too large is refused before a byte of it is read; one whose length is
			// not declared is read up to the limit and refused at the byte past it.
			if r.ContentLength > route.limit {
				http.Error(w, fmt.Sprintf("the request body for %q is %d bytes, and this route "+
					"accepts at most %d", r.Host, r.ContentLength, route.limit),
					http.StatusRequestEntityTooLarge)
				return
			}
			r.Body = http.MaxBytesReader(w, r.Body, route.limit)
		}
		route.proxy.ServeHTTP(w, r)
	})
}

// routesFrom reads what the mesh wrote and turns it into name → route.
func routesFrom(path string) (map[string]route, error) {
	raw, err := os.ReadFile(path)
	if err != nil {
		return nil, err
	}
	var said given
	if err := json.Unmarshal(raw, &said); err != nil {
		return nil, err
	}

	out := map[string]route{}
	for _, c := range said.Given {
		name, _ := c.Values["name"].(string)
		if name == "" {
			log.Printf("%s on %s asked for a route and named nothing; skipped", c.From, c.Node)
			continue
		}
		port, ok := asPort(c.Values["port"])
		if !ok {
			log.Printf("%s on %s asked for route %q and gave no usable port; skipped",
				c.From, c.Node, name)
			continue
		}
		// A limit it cannot honour is a route it does not serve — skipped and named, like a
		// port that is not one. Serving the route with no limit instead would carry exactly what
		// the module said not to carry, and report success.
		// Absent is no limit; present is a limit or a refusal — a `null` written where a number
		// was meant is the second, not the first, and the adapter and the catalogue read it the
		// same way.
		var limit int64
		if raw, said := c.Values["max-request-body"]; said {
			var ok bool
			if limit, ok = bodyLimit(raw); !ok {
				log.Printf("%s on %s asked for route %q with a max-request-body of %v, which is "+
					"not a whole positive number of bytes; skipped", c.From, c.Node, name, raw)
				continue
			}
		}
		// Where the mesh says that machine is. Empty means it is this one — a workload beside the
		// proxy is ordinary, and reaching it over loopback is both correct and the only thing
		// that works when there is no private network.
		at := c.At
		if at == "" {
			at = "127.0.0.1"
		}
		out[strings.ToLower(name)] = route{
			Target:         fmt.Sprintf("http://%s:%d", at, port),
			MaxRequestBody: limit,
		}
	}
	return out, nil
}

// bodyLimit reads a contribution's `max-request-body`, which must be a whole positive number of
// bytes — the same rule the control plane's catalogue applies when it parses the manifest, so a
// limit that reaches here has already passed it once. Absence is the caller's to notice; a `null`
// arriving here is refused like any other non-number.
func bodyLimit(v any) (int64, bool) {
	var n float64
	switch x := v.(type) {
	case float64:
		n = x
	case int:
		n = float64(x)
	default:
		return 0, false
	}
	if n < 1 || n != math.Trunc(n) || n > math.MaxInt64 {
		return 0, false
	}
	return int64(n), true
}

// asPort accepts what JSON makes of a number, which is a float even when it was written 8080.
func asPort(v any) (int, bool) {
	switch n := v.(type) {
	case float64:
		if n < 1 || n > 65535 {
			return 0, false
		}
		return int(n), true
	case int:
		if n < 1 || n > 65535 {
			return 0, false
		}
		return n, true
	}
	return 0, false
}