mesh-controller/modules/route-proxy/cmd/route-proxy/ledger.go

package main

import (
	"context"
	"crypto/x509"
	"encoding/json"
	"encoding/pem"
	"log"
	"net/http"
	"reflect"
	"sort"
	"strings"
	"sync"
	"time"

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

	stdio "git.novox.be/novox/mesh-sdk/go"

	"route-proxy/internal/routestate"
)

// What this proxy says on the bus about the names it serves and the names it refuses (issue 774).
//
// **On change, never on a timer.** A route arriving or leaving, a certificate issued or failing, an upstream answer
// that differs from the last one written, a name asked for that is not served: each marks its key, and one writer
// puts the marked keys. An answer that is the same as the last one written is written again only once it is older
// than refreshAfter, so a site answering 200 all day costs one write a minute, not one per request.
//
// Until the bus is reached nothing is written; what happened meanwhile is held and written when it is.

// refreshAfter is how old a written answer, refusal or certificate failure may be before the same one is written
// again with its new time.
const refreshAfter = time.Minute

// mostAsked bounds the names held in memory as asked and not served: a scan sends any name it likes.
const mostAsked = 4096

// newPerMinute is how many names not seen before are kept by name in a minute; past it they are only counted.
const newPerMinute = 5

// kvStore is one of the module's states, as process.State is one.
type kvStore interface {
	Put(key string, value any) (uint64, error)
	Delete(key string) error
	Keys() ([]string, error)
	Get(key string) (*stdio.StateEntry, error)
}

// ledger is what this proxy keeps of what it serves and refuses, and writes to the bus.
type ledger struct {
	mu  sync.Mutex
	now func() time.Time

	node             string
	routes, unrouted kvStore

	served   map[string]*routestate.Route
	previous map[string]routestate.Route // what the bus held for this machine when it was reached
	asked    map[string]*routestate.Unrouted

	answerWritten map[string]time.Time
	askedWritten  map[string]time.Time

	dirtyRoutes map[string]bool // a name to put when served, to delete when not
	dirtyAsked  map[string]bool

	// domains are the registrable domains of the names served: a refused name under one is kept by name.
	domains                  map[string]bool
	newMinute                time.Time
	newInMinute              int
	other                    routestate.Other
	otherMinute              time.Time
	otherInMinute            int
	otherDirty               bool
	routesFailing, askFailed bool // a failed write said once, until one succeeds

	kick chan struct{}
}

func newLedger() *ledger {
	return &ledger{
		now:           time.Now,
		served:        map[string]*routestate.Route{},
		previous:      map[string]routestate.Route{},
		asked:         map[string]*routestate.Unrouted{},
		answerWritten: map[string]time.Time{},
		askedWritten:  map[string]time.Time{},
		dirtyRoutes:   map[string]bool{},
		dirtyAsked:    map[string]bool{},
		domains:       map[string]bool{},
		kick:          make(chan struct{}, 1),
	}
}

// loop writes whatever was marked, each time something was.
func (l *ledger) loop() {
	for range l.kick {
		l.flush()
	}
}

func (l *ledger) wake() {
	select {
	case l.kick <- struct{}{}:
	default:
	}
}

// attach is the bus reached, as node: what the bus held for this machine is read back, so a restart keeps a name's
// certificate and last answer, and everything held is written.
func (l *ledger) attach(node string, routes, unrouted kvStore) {
	if l == nil {
		return
	}
	before := map[string]routestate.Route{}
	if keys, err := routes.Keys(); err != nil {
		log.Printf("cannot read back what this proxy said it serves: %v", err)
	} else {
		for _, key := range keys {
			at, name := routestate.Split(key)
			if at != node {
				continue
			}
			entry, err := routes.Get(key)
			if err != nil || entry == nil {
				continue
			}
			var r routestate.Route
			if json.Unmarshal(entry.Value, &r) == nil {
				before[name] = r
			}
		}
	}
	l.mu.Lock()
	l.node, l.routes, l.unrouted = node, routes, unrouted
	for name, r := range before {
		l.previous[name] = r
		l.dirtyRoutes[name] = true // put again when served, deleted when no longer
		if now := l.served[name]; now != nil {
			keepWhatWasKnown(now, r)
		}
	}
	for name := range l.served {
		l.dirtyRoutes[name] = true
	}
	for name := range l.asked {
		l.dirtyAsked[name] = true
	}
	l.mu.Unlock()
	l.wake()
}

// keepWhatWasKnown carries a certificate and a last answer read back from the bus onto a route that knows none yet.
func keepWhatWasKnown(r *routestate.Route, before routestate.Route) {
	if r.Certificate.Outcome == "" && r.Certificate.NotAfter == nil && before.Certificate.Authority == r.Certificate.Authority {
		r.Certificate = before.Certificate
	}
	if r.LastAnswer == nil {
		r.LastAnswer = before.LastAnswer
	}
}

// serving is the table this proxy now serves: a name new or changed is written, a name gone is deleted, and a name
// that was asked for unserved and now is served leaves the unrouted state.
func (l *ledger) serving(routes map[string][]rule, public map[string]bool) {
	if l == nil {
		return
	}
	l.mu.Lock()
	defer l.mu.Unlock()
	for name, rules := range routes {
		reach := routestate.ReachInternal
		if public[name] {
			reach = routestate.ReachPublic
		}
		targets := targetsOf(rules)
		r := l.served[name]
		if r == nil {
			r = &routestate.Route{Name: name}
			if before, ok := l.previous[name]; ok {
				r.Certificate.Authority = reach
				keepWhatWasKnown(r, before)
			}
			l.served[name] = r
			l.dirtyRoutes[name] = true
		}
		if r.Reach != reach || !reflect.DeepEqual(r.Targets, targets) {
			r.Reach, r.Targets = reach, targets
			l.dirtyRoutes[name] = true
		}
		if r.Certificate.Authority != reach {
			r.Certificate = routestate.Certificate{Authority: reach}
			l.dirtyRoutes[name] = true
		}
		if _, was := l.asked[name]; was {
			delete(l.asked, name)
			l.dirtyAsked[name] = true
		}
	}
	l.domains = map[string]bool{}
	for name := range routes {
		if d := routestate.Registrable(name); d != "" {
			l.domains[d] = true
		}
	}
	for name := range l.served {
		if _, still := routes[name]; !still {
			delete(l.served, name)
			delete(l.answerWritten, name)
			l.dirtyRoutes[name] = true
		}
	}
	l.wake()
}

func targetsOf(rules []rule) []routestate.Target {
	out := make([]routestate.Target, 0, len(rules))
	for _, r := range rules {
		out = append(out, routestate.Target{Path: r.path, Module: r.from, Node: r.node, Target: r.target,
			Deny: r.policy.deny, Redirect: r.policy.redirectTo, Auth: len(r.policy.users) > 0 || r.policy.sealed != "",
			Sealed: r.policy.sealed})
	}
	return out
}

// answered is an upstream answer to a request for name: its status, or the error that stood in for one.
func (l *ledger) answered(name, path string, status int, failed error) {
	if l == nil {
		return
	}
	l.mu.Lock()
	defer l.mu.Unlock()
	r := l.served[name]
	if r == nil {
		return
	}
	now := l.now()
	why := ""
	if failed != nil {
		why = failed.Error()
	}
	before := r.LastAnswer
	r.LastAnswer = &routestate.Answer{Status: status, At: now, Error: why, Path: path}
	if before != nil && before.Status == status && before.Error == why && now.Sub(l.answerWritten[name]) < refreshAfter {
		return
	}
	l.answerWritten[name] = now
	l.dirtyRoutes[name] = true
	l.wake()
}

// refused is a name this proxy answered "no route for": in a request, or in a handshake's server name.
//
// **Only a name under a domain this proxy serves is kept by name**, and at most newPerMinute new ones a minute: a scan
// sends any name it likes, and each kept name is a key on the bus. Every other refusal is counted, and the count is
// written once a minute under the proxy's OtherName key. Nothing here scans: forgetting old names is the writer's.
// Who asked is not kept: the journal line beside it is what the machine's intrusion prevention reads.
func (l *ledger) refused(name string, internalOnly, handshake bool) {
	if l == nil {
		return
	}
	l.mu.Lock()
	defer l.mu.Unlock()
	if _, isServed := l.served[name]; isServed && !internalOnly {
		return
	}
	now := l.now()
	a := l.asked[name]
	if a == nil {
		if !routestate.Keyable(name) || !l.domains[routestate.Registrable(name)] || !l.mayAddName(now) {
			l.countOther(now)
			return
		}
		a = &routestate.Unrouted{Name: name}
		l.asked[name] = a
	}
	a.AskedAt, a.InternalOnly, a.Handshake = now, internalOnly, handshake
	a.Times++
	if a.Times > 1 && now.Sub(l.askedWritten[name]) < refreshAfter {
		return
	}
	l.askedWritten[name] = now
	l.dirtyAsked[name] = true
	l.wake()
}

// mayAddName says whether another new name may be kept this minute, and counts it.
func (l *ledger) mayAddName(now time.Time) bool {
	if minute := now.Truncate(time.Minute); !minute.Equal(l.newMinute) {
		l.newMinute, l.newInMinute = minute, 0
	}
	if l.newInMinute >= newPerMinute {
		return false
	}
	l.newInMinute++
	return true
}

// countOther counts a refusal not kept by name. A minute's count is written when the next minute's first one comes.
func (l *ledger) countOther(now time.Time) {
	minute := now.Truncate(time.Minute)
	if !minute.Equal(l.otherMinute) {
		if l.otherInMinute > 0 {
			l.other.Minute, l.other.LastMinute = l.otherMinute, l.otherInMinute
			l.otherDirty = true
			l.wake()
		}
		l.otherMinute, l.otherInMinute = minute, 0
	}
	l.otherInMinute++
	l.other.SinceStart++
}

// forgetOldestAsked drops the names asked longest ago from memory until mostAsked are left; the bus keeps them until
// their time to live ends. Called by the writer, never on a request.
func (l *ledger) forgetOldestAsked() {
	if len(l.asked) <= mostAsked {
		return
	}
	names := make([]string, 0, len(l.asked))
	for name := range l.asked {
		names = append(names, name)
	}
	sort.Slice(names, func(i, j int) bool { return l.asked[names[i]].AskedAt.Before(l.asked[names[j]].AskedAt) })
	for _, name := range names[:len(names)-mostAsked] {
		delete(l.asked, name)
		delete(l.askedWritten, name)
		delete(l.dirtyAsked, name)
	}
}

// certified is a certificate an authority issued for name, as autocert stored it.
func (l *ledger) certified(name string, leaf *x509.Certificate) {
	if l == nil {
		return
	}
	l.mu.Lock()
	defer l.mu.Unlock()
	r := l.served[name]
	if r == nil {
		return
	}
	now := l.now()
	notAfter := leaf.NotAfter.UTC()
	r.Certificate.Issuer, r.Certificate.NotAfter = issuerOf(leaf), &notAfter
	r.Certificate.RenewedAt, r.Certificate.At = &now, &now
	r.Certificate.Outcome, r.Certificate.Error = routestate.Issued, ""
	l.dirtyRoutes[name] = true
	l.wake()
}

// handedOut is the certificate a handshake for name was answered with: written when it is not the one recorded.
func (l *ledger) handedOut(name string, leaf *x509.Certificate) {
	if l == nil || leaf == nil {
		return
	}
	l.mu.Lock()
	defer l.mu.Unlock()
	r := l.served[name]
	if r == nil {
		return
	}
	notAfter := leaf.NotAfter.UTC()
	if r.Certificate.NotAfter != nil && r.Certificate.NotAfter.Equal(notAfter) && r.Certificate.Outcome != routestate.Failed {
		return
	}
	now := l.now()
	r.Certificate.Issuer, r.Certificate.NotAfter, r.Certificate.At = issuerOf(leaf), &notAfter, &now
	if r.Certificate.Outcome != routestate.Issued || r.Certificate.Error != "" {
		r.Certificate.Outcome, r.Certificate.Error = routestate.Cached, ""
	}
	l.dirtyRoutes[name] = true
	l.wake()
}

// certificateFailed is a handshake for name that found no certificate to answer with, and why.
func (l *ledger) certificateFailed(name string, failed error) {
	if l == nil {
		return
	}
	l.mu.Lock()
	defer l.mu.Unlock()
	r := l.served[name]
	if r == nil {
		return
	}
	now := l.now()
	same := r.Certificate.Outcome == routestate.Failed && r.Certificate.Error == failed.Error()
	if same && r.Certificate.At != nil && now.Sub(*r.Certificate.At) < refreshAfter {
		return
	}
	r.Certificate.Outcome, r.Certificate.Error, r.Certificate.At = routestate.Failed, failed.Error(), &now
	l.dirtyRoutes[name] = true
	l.wake()
}

func issuerOf(leaf *x509.Certificate) string {
	if cn := leaf.Issuer.CommonName; cn != "" {
		if len(leaf.Issuer.Organization) > 0 {
			return leaf.Issuer.Organization[0] + " " + cn
		}
		return cn
	}
	return leaf.Issuer.String()
}

// flush writes every marked key. A route's write that fails stays marked and is tried again at the next change; a
// refused name's write that fails is dropped, never tried again: the bucket refusing it is full, or the bus is away,
// and either way it is not worth a storm of retries. Each kind of failure is logged once, until a write succeeds.
func (l *ledger) flush() {
	type put struct {
		store kvStore
		key   string
		value any // nil deletes
		route bool
	}
	l.mu.Lock()
	if l.routes == nil {
		l.mu.Unlock()
		return
	}
	l.forgetOldestAsked()
	var puts []put
	for name := range l.dirtyRoutes {
		p := put{store: l.routes, key: routestate.Key(l.node, name), route: true}
		if r := l.served[name]; r != nil {
			c := *r
			c.Proxy = l.node
			p.value = c
		}
		puts = append(puts, p)
	}
	for name := range l.dirtyAsked {
		p := put{store: l.unrouted, key: routestate.Key(l.node, name)}
		if a := l.asked[name]; a != nil {
			c := *a
			c.Proxy = l.node
			p.value = c
		}
		puts = append(puts, p)
	}
	if l.otherDirty {
		o := l.other
		o.Proxy = l.node
		puts = append(puts, put{store: l.unrouted, key: routestate.Key(l.node, routestate.OtherName), value: o})
	}
	l.dirtyRoutes, l.dirtyAsked, l.otherDirty = map[string]bool{}, map[string]bool{}, false
	l.mu.Unlock()

	for _, p := range puts {
		var err error
		if p.value == nil {
			err = p.store.Delete(p.key)
		} else {
			_, err = p.store.Put(p.key, p.value)
		}
		l.mu.Lock()
		switch {
		case err == nil && p.route:
			l.routesFailing = false
		case err == nil:
			l.askFailed = false
		case p.route:
			if !l.routesFailing {
				log.Printf("cannot write %s to the bus, kept for the next change: %v", p.key, err)
			}
			l.routesFailing = true
			_, name := routestate.Split(p.key)
			l.dirtyRoutes[name] = true
		default:
			if !l.askFailed {
				log.Printf("cannot write %s to the bus, dropped with every refused name until a write succeeds: %v",
					p.key, err)
			}
			l.askFailed = true
		}
		l.mu.Unlock()
	}
}

// recordingCache is an authority's certificate cache that tells the ledger of every certificate stored in it: the
// one place autocert says it was issued one.
type recordingCache struct {
	autocert.Cache
	held *table
}

func (c recordingCache) Put(ctx context.Context, key string, data []byte) error {
	if err := c.Cache.Put(ctx, key, data); err != nil {
		return err
	}
	name := strings.TrimSuffix(key, "+rsa")
	if strings.Contains(name, "+") || !c.held.routed(name) {
		return nil // the account key, a challenge token
	}
	if leaf := leafOf(data); leaf != nil {
		c.held.ledger.certified(bareHost(name), leaf)
	}
	return nil
}

// leafOf is the first certificate in what autocert stores: the private key, then the chain, in PEM.
func leafOf(data []byte) *x509.Certificate {
	for {
		var block *pem.Block
		block, data = pem.Decode(data)
		if block == nil {
			return nil
		}
		if block.Type == "CERTIFICATE" {
			leaf, err := x509.ParseCertificate(block.Bytes)
			if err != nil {
				return nil
			}
			return leaf
		}
	}
}

// answerRecorder carries a proxied request's answer through and remembers its status and the proxy's error.
type answerRecorder struct {
	http.ResponseWriter
	status int
	err    error
}

func (a *answerRecorder) WriteHeader(status int) {
	if a.status == 0 {
		a.status = status
	}
	a.ResponseWriter.WriteHeader(status)
}

func (a *answerRecorder) Write(b []byte) (int, error) {
	if a.status == 0 {
		a.status = http.StatusOK
	}
	return a.ResponseWriter.Write(b)
}

// Unwrap lets http.ResponseController reach the connection beneath: a flush, an upgrade's hijack.
func (a *answerRecorder) Unwrap() http.ResponseWriter { return a.ResponseWriter }

// proxyFailed is the reverse proxy's answer when upstream gave none: the error kept for the ledger, and a 502.
func proxyFailed(w http.ResponseWriter, r *http.Request, err error) {
	if a, ok := w.(*answerRecorder); ok {
		a.err = err
	}
	log.Printf("http: proxy error for %s%s: %v", r.Host, r.URL.Path, err)
	w.WriteHeader(http.StatusBadGateway)
}