echolot/server/internal/control/control.go

// SPDX-FileCopyrightText: 2026 Echolot contributors
// SPDX-License-Identifier: GPL-3.0-or-later

// Package control implements the control plane (spec §2): enrollment,
// profile, sessions. HTTPS with a possibly self-signed cert — clients trust
// the SPKI pin from enrollment, not a CA (spec §1).
package control

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"crypto/tls"
	"crypto/x509"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net"
	"net/http"
	"net/netip"
	"net/url"
	"strconv"
	"strings"
	"time"

	"echo-lot.app/server/internal/compat"
	"echo-lot.app/server/internal/dataplane"
	"echo-lot.app/server/internal/oidc"
	"echo-lot.app/server/internal/ratelimit"
	"echo-lot.app/server/internal/runs"
	"echo-lot.app/server/internal/session"
	"echo-lot.app/server/internal/store"
)

// Version is stamped by the build (-ldflags "-X ...").
var Version = "dev"

type Server struct {
	Store    *store.Store
	Sessions *session.Manager
	Name     string
	// Targets/capabilities for the profile response. The server offers only
	// what it actually implements; the registry grows with the code.
	UDPPort  int
	TCPPort  int
	StunPort int
	// SPKI pin of the serving cert, for the profile's pins[] field.
	PinB64 string
	// CertChain is the served leaf-first DER chain, for GET /v1/tls-reference.
	CertChain [][]byte
	// Capabilities as computed at startup from what is actually wired up.
	Capabilities []string
	// TCPRecent returns recent TCP-echo connections for a source IP (may be nil).
	TCPRecent func(ip string) any
	// DelayedEcho schedules/sends a DELAYED_ECHO for a session (may be nil).
	DelayedEcho func(sess *session.Session, actionID string) error
	// Granted server->client sends (spec §5). Both consume an asymmetric grant.
	// DownTrain's dscp is -1 for "leave the socket's default marking alone".
	DownTrain func(sess *session.Session, g *session.Grant, count, sizeBytes, intervalUs, dscp int) (int, error)
	BigSend   func(sess *session.Session, g *session.Grant, sizes []int, df bool) ([]dataplane.BigSendResult, error)
	// OIDC verifies ID tokens presented by the *app* (may be nil).
	OIDC *oidc.Verifier
	// AdminOIDC verifies tokens from the admin UI's own client. Separate because an IdP may
	// give each application its own issuer — Authentik derives it from the application slug —
	// and a verifier pins exactly one issuer and the clients belonging to it.
	AdminOIDC *oidc.Verifier
	// Runs stores uploaded measurement documents (may be nil: uploads unsupported).
	Runs *runs.Store
	// FragSend emits one datagram as hand-built IP fragments in a chosen order (may be nil:
	// needs a raw socket, so it is unavailable to an unprivileged server).
	FragSend func(sess *session.Session, g *session.Grant, sizeBytes int, mode dataplane.FragMode, fragSize int) (dataplane.FragResult, error)
	// DownThroughput sends paced traffic toward the client for a bounded time (may be nil).
	DownThroughput func(sess *session.Session, g *session.Grant, durationMs, kbps, sizeBytes int) (dataplane.ThroughputResult, error)
	// EgressMTU reports the server's own measured egress path MTU (0 = unknown). With DF set
	// we cannot emit a datagram larger than this, so requested sizes above it are refused up
	// front and reported as such — the client must not read that as a downstream path limit.
	EgressMTU func() int
	// CanaryQueries returns logged canary lookups for a session prefix (may be nil).
	CanaryQueries func(sessionPrefix string) any
	// CanaryZone is surfaced in the profile so the app knows what to query.
	CanaryZone string
	// The addresses this server can be measured on. The "_alt" pair is the second address
	// RFC 5780 behaviour discovery redirects to, and the one reserved from services so that
	// nothing answering there is itself a measurement.
	IP4, IP6, IP4Alt, IP6Alt string
	// ProvenGood reports the server's self-test signal (may be nil). Surfaced
	// in the profile so a client can trust — or skip — MTU tests: if the
	// server's own egress isn't full-MTU, client MTU results measure the
	// server, not the client.
	ProvenGood func() (mtuOK, sysctlOK bool)

	// PublicControlURL is where clients reach this server, for the enrollment link (§2.1).
	// Empty means "derive from the address we are listening on", which is right for a plain
	// deployment and wrong behind a proxy or a name — hence the override.
	PublicControlURL string
	// AppRange is the app-version window this server will serve. Zero value means the built-in
	// default (see DefaultAppRange).
	AppRange compat.Range

	// Spec §2.5 token buckets, keyed per credential and per source IP inside the limiter.
	// Nil disables a ceiling (config value 0).
	RateSessions *ratelimit.Limiter // POST /v1/sessions
	RateActions  *ratelimit.Limiter // POST /v1/sessions/{id}/actions
}

// AppVersionHeader is how a client states its version. A client too old to send it is treated as
// unknown rather than refused: the check exists to turn confusing failures into clear ones, and
// refusing something we cannot identify achieves the opposite.
const AppVersionHeader = "X-Echolot-App-Version"

// ProtocolVersion is the wire contract (probe-protocol.md) this build implements. It is what the
// version window is really about; the release version is only a proxy for it.
//
// 1.0.1: upstream trains (§3.2 types 0x03/0x04/0x05) and the action-id bytes at payload[8:16]
// of granted packets. Patch, not minor: both are additive — a client that never sends
// TRAIN_REPORT_REQ and never reads granted payloads (today's client reads only header fields)
// sees no difference, so the fleet must not be split over it (§8.1).
const ProtocolVersion = "1.0.1"

// SchemaVersion is the measurement-document format this server can store.
const SchemaVersion = "1.0.0"

// DefaultAppRange: everything from the first app that speaks this protocol up to — but not
// including — the next breaking series. Bounds sit at breaking boundaries so shipping a patch
// never requires touching this.
func DefaultAppRange() compat.Range {
	r, err := compat.ParseRange("0.2.0", "1.0.0")
	if err != nil {
		panic("built-in app range is malformed: " + err.Error())
	}
	return r
}

func (s *Server) appRange() compat.Range {
	if s.AppRange.Min == (compat.Version{}) && !s.AppRange.HasMax {
		return DefaultAppRange()
	}
	return s.AppRange
}

// requireCompatibleApp wraps a handler with the version window.
//
// Deliberately NOT applied to GET /v1/profile: that is where a client learns which version it
// should be. Gating it would leave a refused client with nothing to show its user but a timeout,
// which is precisely the confusion this check exists to remove.
func (s *Server) requireCompatibleApp(next http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		verdict, msg := compat.Check(r.Header.Get(AppVersionHeader), s.appRange(), "app")
		switch verdict {
		case compat.TooOld, compat.TooNew:
			slog.Info("refused incompatible app", "app_version", r.Header.Get(AppVersionHeader),
				"accepts", s.appRange().String(), "path", r.URL.Path)
			// 426 says exactly this and nothing else; the body carries the window so the app can
			// show the user the number to reach, not just that it failed.
			writeJSON(w, http.StatusUpgradeRequired, map[string]any{
				"error":            msg,
				"app_version":      r.Header.Get(AppVersionHeader),
				"accepts_app":      s.appRange().String(),
				"server_version":   Version,
				"protocol_version": ProtocolVersion,
			})
			return
		}
		next(w, r)
	}
}

func (s *Server) Handler() http.Handler {
	mux := http.NewServeMux()
	// Always reachable, whatever the version window says: this is how a client discovers the
	// window it has to satisfy.
	mux.HandleFunc("GET /v1/profile", s.profile)

	gate := s.requireCompatibleApp
	mux.HandleFunc("POST /v1/enroll", gate(s.enroll))
	// §2.5 buckets sit on the two endpoints that make the server DO things — create state,
	// send traffic. GET /v1/profile stays ungated on every axis (see above).
	mux.HandleFunc("POST /v1/sessions", gate(s.rateLimited(s.RateSessions, s.newSession)))
	mux.HandleFunc("DELETE /v1/sessions/{id}", gate(s.deleteSession))
	mux.HandleFunc("GET /v1/sessions/{id}/observations", gate(s.observations))
	mux.HandleFunc("POST /v1/sessions/{id}/actions", gate(s.rateLimited(s.RateActions, s.actions)))
	mux.HandleFunc("POST /v1/echo", gate(s.httpEcho))
	mux.HandleFunc("GET /v1/tls-reference", gate(s.tlsReference))
	mux.HandleFunc("POST /v1/runs", gate(s.uploadRun))
	mux.HandleFunc("GET /v1/runs", gate(s.listRuns))
	mux.HandleFunc("GET /v1/runs/{id}", gate(s.getRun))
	mux.HandleFunc("DELETE /v1/runs/{id}", gate(s.deleteRun))
	mux.HandleFunc("POST /v1/account/link", gate(s.linkAccount))
	mux.HandleFunc("DELETE /v1/account/link", gate(s.unlinkAccount))
	mux.HandleFunc("GET /v1/account", gate(s.accountStatus))
	return mux
}

// EchoHandler exposes just the HTTP-echo endpoint for the optional cleartext
// listener (spec §4: plaintext-path tampering test).
func (s *Server) EchoHandler() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("POST /v1/echo", s.httpEcho)
	return mux
}

// selftestSignal is the compact "server proven good" object for the profile.
// mtu_ok=false tells a client its MTU results would measure this server.
func selftestSignal(f func() (bool, bool)) map[string]any {
	if f == nil {
		return map[string]any{"mtu_ok": nil, "sysctl_ok": nil}
	}
	mtuOK, sysctlOK := f()
	return map[string]any{"mtu_ok": mtuOK, "sysctl_ok": sysctlOK}
}

// rateLimited enforces one §2.5 bucket policy on an endpoint: a token per credential AND one per
// source IP. Two keys because each closes the other's hole — keyed only by credential, one
// address cycles through credentials; keyed only by address, one credential rides many
// addresses. The refusal is 429 with Retry-After, which is the whole point of a token bucket
// over a hard drop here: a well-behaved client is told when to come back.
func (s *Server) rateLimited(l *ratelimit.Limiter, next http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		okCred, waitCred := l.Allow("cred:" + bearer(r))
		okIP, waitIP := l.Allow("ip:" + remoteIP(r))
		if !okCred || !okIP {
			wait := max(waitCred, waitIP)
			secs := int(wait/time.Second) + 1 // Retry-After is whole seconds, rounded up
			w.Header().Set("Retry-After", strconv.Itoa(secs))
			writeJSON(w, http.StatusTooManyRequests, map[string]any{
				"error": "rate limited", "retry_after_s": secs,
			})
			return
		}
		next(w, r)
	}
}

// remoteIP is the request's source address without the port, for rate-limit keys.
func remoteIP(r *http.Request) string {
	host, _, err := net.SplitHostPort(r.RemoteAddr)
	if err != nil {
		return r.RemoteAddr
	}
	return strings.Trim(host, "[]")
}

// sessionAuth resolves {id} and requires the bearer to be the owning device.
func (s *Server) sessionAuth(w http.ResponseWriter, r *http.Request) *session.Session {
	dev := s.Store.DeviceByCredential(bearer(r))
	if dev == nil {
		writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
		return nil
	}
	sess := s.Sessions.ByID(r.PathValue("id"))
	if sess == nil || sess.Device != dev.ID {
		writeJSON(w, http.StatusNotFound, map[string]string{"error": "no such session"})
		return nil
	}
	return sess
}

// observations is spec §6 — everything the server witnessed for a session.
func (s *Server) observations(w http.ResponseWriter, r *http.Request) {
	sess := s.sessionAuth(w, r)
	if sess == nil {
		return
	}
	packetsSeen, udp, cb := sess.Observations()
	var tcp any
	if s.TCPRecent != nil {
		// Correlate by source IP: TCP echo carries no session id on the wire.
		if ds := sess.DataSource(); ds.IsValid() {
			tcp = s.TCPRecent(ds.Addr().Unmap().String())
		} else if sess.ControlSource.IsValid() {
			tcp = s.TCPRecent(sess.ControlSource.Unmap().String())
		}
	}
	var dnsCanary any
	if s.CanaryQueries != nil {
		dnsCanary = s.CanaryQueries(sess.ID[:16]) // the session's wire prefix
	}
	writeJSON(w, http.StatusOK, map[string]any{
		"udp": map[string]any{
			"packets_seen": packetsSeen,
			"packets":      udp,
			// The per-train received view (spec §6 "trains"), columnar like the wire report —
			// the flat packet list above stays for clients that predate trains.
			"trains": trainsJSON(sess.Trains()),
		},
		"tcp": tcp,
		"connect_back": cb,
		// The sender's own count, which is what makes the receiver's count mean something.
		"throughput": sess.ThroughputReports(),
		// The receiver's count for upstream runs — same idea, other direction.
		"throughput_up": upstreamJSON(sess),
		"dns_canary":    dnsCanary,
		// TODO(spec §6): http echo records
	})
}

// actions is spec §5 — authenticated asymmetric operations. Implemented:
// delayed_echo, connect_back. Destination is ALWAYS the session's observed
// source (data-plane source; connect_back uses the control-plane source).
func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
	sess := s.sessionAuth(w, r)
	if sess == nil {
		return
	}
	var req struct {
		Action     string `json:"action"`
		DelayS     int    `json:"delay_s"`
		Protocol   string `json:"protocol"`
		Port       int    `json:"port"`
		Count      int    `json:"count"`
		SizeBytes  int    `json:"size_bytes"`
		IntervalUs int    `json:"interval_us"`
		SizesBytes []int  `json:"sizes_bytes"`
		DSCP       *int   `json:"dscp"`
		DF         *bool  `json:"df"`
		Mode       string `json:"mode"`
		FragBytes  int    `json:"frag_bytes"`
		Direction  string `json:"direction"`
		DurationS  int    `json:"duration_s"`
		Kbps       int    `json:"kbps"`
		Streams    int    `json:"streams"`
	}
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "bad body"})
		return
	}
	actionID := randomID()
	switch req.Action {
	case "delayed_echo":
		if s.DelayedEcho == nil {
			writeJSON(w, http.StatusNotImplemented, map[string]string{"error": "delayed_echo not wired"})
			return
		}
		delay := min(max(req.DelayS, 1), 600)
		if !sess.DataSource().IsValid() {
			writeJSON(w, http.StatusConflict, map[string]string{"error": "no data-plane traffic seen yet — send an ECHO first"})
			return
		}
		time.AfterFunc(time.Duration(delay)*time.Second, func() {
			if err := s.DelayedEcho(sess, actionID); err != nil {
				slog.Debug("delayed echo failed", "err", err)
			}
		})
		writeJSON(w, http.StatusAccepted, map[string]any{"action_id": actionID, "delay_s": delay})
	case "connect_back":
		if req.Port < 1 || req.Port > 65535 || (req.Protocol != "tcp" && req.Protocol != "udp") {
			writeJSON(w, http.StatusBadRequest, map[string]string{"error": "connect_back needs protocol tcp|udp and a port"})
			return
		}
		target := net.JoinHostPort(sess.ControlSource.Unmap().String(), strconv.Itoa(req.Port))
		go func() {
			start := time.Now()
			conn, err := net.DialTimeout(req.Protocol, target, 5*time.Second)
			res := session.ConnectBackResult{ActionID: actionID, RttMs: float64(time.Since(start).Microseconds()) / 1000}
			switch {
			case err == nil:
				res.Result = "connected"
				if req.Protocol == "udp" {
					// UDP "dial" always succeeds locally; send one datagram
					// so the client actually observes something.
					_, _ = conn.Write([]byte("echolot-connect-back " + actionID))
				}
				conn.Close()
			case isTimeout(err):
				res.Result = "timeout"
			default:
				res.Result = "refused"
			}
			sess.RecordConnectBack(res)
		}()
		writeJSON(w, http.StatusAccepted, map[string]any{"action_id": actionID, "target": target})
	case "downtrain":
		if s.DownTrain == nil {
			writeJSON(w, http.StatusNotImplemented, map[string]string{"error": "downtrain not wired"})
			return
		}
		dscp, err := dscpArg(req.DSCP)
		if err != nil {
			writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
			return
		}
		// A downstream train sends far more than it receives, so it needs a grant (§3.4).
		count := clamp(req.Count, 1, 5000)
		size := clamp(req.SizeBytes, dataMinPacket, 1500)
		interval := clamp(req.IntervalUs, 0, 1_000_000)
		g := sess.NewGrant(actionID, int64(count*size), 0, session.DefaultGrantLimits)
		if g == nil {
			writeJSON(w, http.StatusConflict, noDataPlaneYet)
			return
		}
		go func() {
			sent, err := s.DownTrain(sess, g, count, size, interval, dscp)
			slog.Info("downtrain finished", "action", actionID, "sent", sent, "bytes", g.Sent(), "err", err)
		}()
		resp := map[string]any{
			"action_id": actionID, "count": count, "size_bytes": size, "interval_us": interval,
			"grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps},
		}
		if dscp >= 0 {
			resp["dscp"] = dscp
			// Told up front, not discovered: a client measuring DSCP survival on a burst the
			// server could not mark would conclude the network stripped it.
			resp["dscp_applied"] = dataplane.TOSSupported
		}
		writeJSON(w, http.StatusAccepted, resp)

	case "big_send":
		if s.BigSend == nil {
			writeJSON(w, http.StatusNotImplemented, map[string]string{"error": "big_send not wired"})
			return
		}
		sizes := req.SizesBytes
		if len(sizes) == 0 {
			sizes = []int{1200, 1400, 1472, 1500, 2000, 4000}
		}
		if len(sizes) > 32 {
			sizes = sizes[:32]
		}
		// DF on by default: an unfragmented burst is what makes the result a path-MTU
		// measurement rather than a fragment-delivery one. Callers opt out explicitly.
		df := true
		if req.DF != nil {
			df = *req.DF
		}
		// With DF we can only emit up to our own egress MTU minus IP+UDP headers. Drop the
		// rest here and say so, rather than sending nothing and letting the client blame
		// the path.
		maxDF := 0
		if df {
			maxDF = s.maxDFPayload(sess)
			if maxDF > 0 {
				kept := sizes[:0]
				for _, x := range sizes {
					if x <= maxDF {
						kept = append(kept, x)
					}
				}
				sizes = kept
			}
		}
		if len(sizes) == 0 {
			writeJSON(w, http.StatusBadRequest, map[string]any{
				"error":        "every requested size exceeds the server's own egress MTU with DF set",
				"max_df_bytes": maxDF,
			})
			return
		}
		total := 0
		for _, x := range sizes {
			total += clamp(x, dataMinPacket, 9000)
		}
		g := sess.NewGrant(actionID, int64(total), 0, session.DefaultGrantLimits)
		if g == nil {
			writeJSON(w, http.StatusConflict, noDataPlaneYet)
			return
		}
		go func() {
			results, err := s.BigSend(sess, g, sizes, df)
			slog.Info("big_send finished", "action", actionID, "results", results, "df", df, "err", err)
		}()
		writeJSON(w, http.StatusAccepted, map[string]any{
			"action_id": actionID, "sizes_bytes": sizes, "df": df, "max_df_bytes": maxDF,
			"grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps},
		})

	case "frag_send":
		if s.FragSend == nil {
			writeJSON(w, http.StatusNotImplemented, map[string]string{
				"error": "frag_send needs a raw socket, which this server does not have",
			})
			return
		}
		size := clamp(req.SizeBytes, 1600, 8000) // must exceed the path MTU or nothing fragments
		mode := dataplane.FragMode(req.Mode)
		switch mode {
		case dataplane.FragInOrder, dataplane.FragReversed, dataplane.FragFirstLast:
		default:
			mode = dataplane.FragInOrder
		}
		fragBytes := clamp(req.FragBytes, 8, 1400)
		g := sess.NewGrant(actionID, int64(size), 0, session.DefaultGrantLimits)
		if g == nil {
			writeJSON(w, http.StatusConflict, noDataPlaneYet)
			return
		}
		// Synchronous: the whole burst is a few kB and at most a few hundred milliseconds, and
		// the caller wants to know it was actually emitted before it starts listening. An
		// asynchronous send would make "nothing arrived" ambiguous between a path drop and a
		// send that never happened — the one distinction this test exists to make.
		result, err := s.FragSend(sess, g, size, mode, fragBytes)
		slog.Info("frag_send finished", "action", actionID, "mode", mode,
			"size", size, "fragments", result.Fragments, "err", err)
		if err != nil {
			writeJSON(w, http.StatusConflict, map[string]any{
				"error": err.Error(), "action_id": actionID, "result": result,
			})
			return
		}
		writeJSON(w, http.StatusAccepted, map[string]any{
			"action_id": actionID, "mode": string(mode), "size_bytes": size,
			"frag_bytes": fragBytes, "fragments": result.Fragments,
			"grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps},
		})

	case "throughput":
		if s.DownThroughput == nil {
			writeJSON(w, http.StatusNotImplemented, map[string]string{"error": "throughput not wired"})
			return
		}
		// Only the downstream direction needs the server to send. Upstream is the client
		// sending and the server counting, which needs no action at all — so asking for it here
		// is a client bug worth naming rather than silently doing the other thing.
		if req.Direction == "up" {
			// Upstream needs nothing sent from here — the client generates the traffic and the
			// server counts it. The only thing an action can usefully do is zero the counter so
			// the run measures itself rather than inheriting an earlier one.
			sess.ResetUpstream()
			writeJSON(w, http.StatusAccepted, map[string]any{
				"action_id": actionID, "direction": "up", "reset": true,
				"note": "send TYPE_THROUGHPUT_UP packets, then read observations.throughput_up",
			})
			return
		}
		if req.Direction != "" && req.Direction != "down" {
			writeJSON(w, http.StatusBadRequest, map[string]string{
				"error": "direction must be up or down",
			})
			return
		}
		// Planned once, here, so the response promises exactly what the run will do. A request
		// that would outlast the server's byte cap comes back with a shorter duration rather
		// than being truncated halfway.
		durationMs, kbps := dataplane.ThroughputPlan(
			clamp(req.DurationS, 1, 30)*1000, clamp(req.Kbps, 100, 200_000))
		size := clamp(req.SizeBytes, dataMinPacket, 1472)
		if req.SizeBytes == 0 {
			size = 1200
		}
		g := sess.NewGrant(actionID, 0, kbps, dataplane.ThroughputLimits(durationMs, kbps))
		if g == nil {
			writeJSON(w, http.StatusConflict, noDataPlaneYet)
			return
		}
		// Answered before the run so the client can start listening, then reported through the
		// observations API. Doing it the other way round would have the client miss the first
		// second of a ten-second test.
		writeJSON(w, http.StatusAccepted, map[string]any{
			"action_id": actionID, "direction": "down",
			"duration_s": durationMs / 1000, "duration_ms": durationMs,
			"requested_duration_s": clamp(req.DurationS, 1, 30),
			"kbps":                 kbps, "size_bytes": size,
			"grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps},
		})
		if f, ok := w.(http.Flusher); ok {
			f.Flush()
		}
		go func() {
			result, err := s.DownThroughput(sess, g, durationMs, kbps, size)
			slog.Info("throughput finished", "action", actionID, "packets", result.Packets,
				"bytes", result.Bytes, "kbps", result.Kbps, "limited_by", result.LimitedBy, "err", err)
			sess.RecordThroughput(actionID, result.Packets, result.Bytes, result.DurationMs,
				result.Kbps, result.LimitedBy)
		}()

	default:
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "unknown or unimplemented action"})
	}
}

// maxDFPayload is the largest UDP payload the server can emit toward this session without
// fragmenting: its own egress MTU less the IP and UDP headers of the session's address family.
// Returns 0 when the egress MTU is unknown, meaning "do not clamp".
func (s *Server) maxDFPayload(sess *session.Session) int {
	if s.EgressMTU == nil {
		return 0
	}
	mtu := s.EgressMTU()
	if mtu <= 0 {
		return 0
	}
	overhead := 28 // IPv4 (20) + UDP (8)
	if src := sess.DataSource(); src.IsValid() && !src.Addr().Unmap().Is4() {
		overhead = 48 // IPv6 (40) + UDP (8)
	}
	return mtu - overhead
}

// dscpArg validates the optional downtrain dscp parameter (spec §5). Absent means -1: leave the
// socket's default marking alone, which is different from asking for DSCP 0 (explicitly
// best-effort). Out-of-range values are refused rather than clamped — a clamped 46→63 would mark
// the burst with a class the client never asked for and silently change what the test measures.
func dscpArg(v *int) (int, error) {
	if v == nil {
		return -1, nil
	}
	if *v < 0 || *v > 63 {
		return 0, fmt.Errorf("dscp %d is out of range: the field is 6 bits (0..63)", *v)
	}
	return *v, nil
}

// trainsJSON renders the per-train received view columnar — one array per field, matching the
// wire report and the schema's train-evidence shape — with []int for the byte-wide columns
// because encoding/json would base64 a []uint8.
func trainsJSON(trains []session.Train) []map[string]any {
	out := make([]map[string]any, 0, len(trains))
	for _, t := range trains {
		n := len(t.Entries)
		seq := make([]uint32, n)
		trx := make([]int64, n)
		size := make([]int, n)
		ttl := make([]int, n)
		dscp := make([]int, n)
		ecn := make([]int, n)
		for i, e := range t.Entries {
			seq[i], trx[i], size[i] = e.Seq, e.TRxNs, int(e.Size)
			ttl[i], dscp[i], ecn[i] = int(e.TTL), int(e.DSCP), int(e.ECN)
		}
		out = append(out, map[string]any{
			"train_id": t.ID,
			// The loss denominator: every packet counted, whether or not its row was kept.
			"packets_received": t.Received,
			"truncated":        t.Truncated,
			"seq":              seq, "t_rx_ns": trx, "size": size,
			"ttl": ttl, "dscp": dscp, "ecn": ecn,
		})
	}
	return out
}

// dataMinPacket is the smallest granted datagram: header + 16 payload bytes, because [8:16] of
// every granted payload carries the action id. It must match what the senders raise short sizes
// to, or a minimum-size train's grant is budgeted for fewer bytes than actually leave and the
// train is cut short by its own arithmetic.
const dataMinPacket = dataplane.HeaderSize + 16

var noDataPlaneYet = map[string]string{
	"error": "no data-plane traffic seen yet — send an ECHO first so the destination is verified",
}

func clamp(v, lo, hi int) int {
	if v < lo {
		return lo
	}
	if v > hi {
		return hi
	}
	return v
}

func isTimeout(err error) bool {
	var ne net.Error
	return errors.As(err, &ne) && ne.Timeout()
}

func randomID() string {
	var b [8]byte
	_, _ = rand.Read(b[:])
	return hex.EncodeToString(b[:])
}

func bearer(r *http.Request) string {
	h := r.Header.Get("Authorization")
	if v, ok := strings.CutPrefix(h, "Bearer "); ok {
		return strings.TrimSpace(v)
	}
	return ""
}

func writeJSON(w http.ResponseWriter, code int, v any) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(code)
	enc := json.NewEncoder(w)
	// Go escapes <, > and & by default, for JSON embedded in HTML. This is an API, and the
	// escaping is actively harmful here: a refusal message reading "needs >= 0.2.0" is what
	// the user ends up seeing. Nothing we emit is ever interpolated into a page.
	enc.SetEscapeHTML(false)
	_ = enc.Encode(v)
}

// enroll redeems a single-use enrollment token for a device credential (§2.1).
func (s *Server) enroll(w http.ResponseWriter, r *http.Request) {
	tok := bearer(r)
	if tok == "" {
		writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "missing enrollment token"})
		return
	}
	var body struct {
		Name string `json:"name"`
	}
	_ = json.NewDecoder(r.Body).Decode(&body) // body is optional
	dev, err := s.Store.Redeem(tok, body.Name)
	if err != nil {
		writeJSON(w, http.StatusForbidden, map[string]string{"error": err.Error()})
		return
	}
	slog.Info("device enrolled", "device", dev.ID, "name", dev.Name)
	writeJSON(w, http.StatusCreated, map[string]string{
		"device_id": dev.ID,
		// The spec (§2.1) names this device_credential; the first implementation shipped
		// "credential". Both are sent while deployed 0.5.x clients still read the old name;
		// the client prefers the spec's. Drop "credential" once nothing reads it.
		"device_credential": dev.Credential, // returned exactly once
		"credential":        dev.Credential, // deprecated alias, see above
	})
}

// profile is spec §2.2. Only implemented capabilities are advertised.
func (s *Server) profile(w http.ResponseWriter, r *http.Request) {
	dev := s.Store.DeviceByCredential(bearer(r))
	if dev == nil {
		writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
		return
	}
	host := r.Host
	if i := strings.LastIndex(host, ":"); i > 0 {
		host = host[:i]
	}
	writeJSON(w, http.StatusOK, map[string]any{
		"profile_version": 1,
		"name":            s.Name,
		"server_version":  Version,
		// source_url makes GPL §6 compliance mechanical for operators of
		// modified builds and gives clients provenance for the measurement.
		"source_url":   "", // TODO: stamp from build metadata
		"capabilities": s.Capabilities,
		"targets": []map[string]any{s.target(host)},
		"pins":            []string{"pin-sha256:" + s.PinB64},
		"next_pins":       []string{},
		"canary_zone":     s.CanaryZone,
		"server_selftest": selftestSignal(s.ProvenGood),
		"limits":          map[string]any{"max_kbps": 50000, "max_session_s": 900},
		// The app needs the upload rules before it offers the switch: whether uploads are
		// accepted at all, and how much identifying detail it must strip first.
		"uploads": s.uploadPolicy(),
		// What a client needs to start a sign-in, without hard-coding the operator's IdP into
		// the app: where to authorize, which client id to use, and whether it is worth offering
		// sign-in at all on this server.
		"auth": s.authInfo(r.Context()),
		// What this build speaks, and which app versions it will serve. A client checks the
		// server side of the same question against its own bounds.
		"compat": map[string]any{
			"protocol_version": ProtocolVersion,
			"schema_version":   SchemaVersion,
			"app_min":          s.appRange().Min.String(),
			"app_max":          maxOrEmpty(s.appRange()),
		},
	})
}

// newSession is spec §2.4.
func (s *Server) newSession(w http.ResponseWriter, r *http.Request) {
	cred := bearer(r)
	dev := s.Store.DeviceByCredential(cred)
	if dev == nil {
		writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
		return
	}
	srcHost, _, _ := strings.Cut(r.RemoteAddr, ":")
	src, _ := netip.ParseAddr(strings.Trim(srcHost, "[]"))
	sess, salt, err := s.Sessions.New(dev.ID, cred, src)
	if err != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "session creation failed"})
		return
	}
	writeJSON(w, http.StatusCreated, map[string]any{
		"session_id": sess.ID,
		"key_salt":   base64.StdEncoding.EncodeToString(salt),
		"epoch":      sess.Epoch.Format(time.RFC3339),
		"expires_s":  int(time.Until(sess.Expires).Seconds()),
	})
}

func (s *Server) deleteSession(w http.ResponseWriter, r *http.Request) {
	if s.Store.DeviceByCredential(bearer(r)) == nil {
		writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
		return
	}
	if id := r.PathValue("id"); len(id) >= 16 {
		s.Sessions.Delete(id)
	}
	w.WriteHeader(http.StatusNoContent)
}

// SpkiPinB64 computes the RFC 7469 pin (SHA-256 over the SPKI, base64) of the
// leaf certificate — what enrollment QR codes carry and profiles re-state.
func SpkiPinB64(cert tls.Certificate) (string, error) {
	leaf := cert.Leaf
	if leaf == nil {
		parsed, err := x509.ParseCertificate(cert.Certificate[0])
		if err != nil {
			return "", err
		}
		leaf = parsed
	}
	sum := sha256.Sum256(leaf.RawSubjectPublicKeyInfo)
	return base64.StdEncoding.EncodeToString(sum[:]), nil
}

// uploadPolicy is the profile's advertisement of the operator's upload rules.
func (s *Server) uploadPolicy() map[string]any {
	if s.Runs == nil {
		return map[string]any{"mode": string(runs.ModeOff), "reason": "not configured"}
	}
	p := s.Runs.Policy()
	return map[string]any{
		"mode":                string(p.Mode),
		"max_bytes":           p.MaxBytes,
		"retention_days":      p.RetentionDays,
		"max_runs_per_device": p.MaxRunsPerDevice,
		"min_anonymization":   p.MinAnonymization,
	}
}

// uploadRun stores one measurement document for the calling device.
func (s *Server) uploadRun(w http.ResponseWriter, r *http.Request) {
	dev := s.Store.DeviceByCredential(bearer(r))
	if dev == nil {
		writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
		return
	}
	if s.Runs == nil {
		writeJSON(w, http.StatusForbidden, map[string]string{"error": runs.ErrDisabled.Error()})
		return
	}
	limit := s.Runs.Policy().MaxBytes
	if limit <= 0 {
		limit = 4 << 20
	}
	// +1 so a body exactly at the limit is distinguishable from one over it.
	body, err := io.ReadAll(io.LimitReader(r.Body, limit+1))
	if err != nil {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "read failed"})
		return
	}
	meta, err := s.Runs.Put(dev.ID, body, dev.LinkedToAccount())
	switch {
	case err == nil:
		slog.Info("run uploaded", "device", dev.ID, "run", meta.ID,
			"bytes", meta.SizeBytes, "anon", meta.Anonymization, "findings", meta.FindingCount)
		writeJSON(w, http.StatusCreated, meta)
	case errors.Is(err, runs.ErrDisabled), errors.Is(err, runs.ErrNeedAccount),
		errors.Is(err, runs.ErrNotAnonEnough):
		writeJSON(w, http.StatusForbidden, map[string]any{
			"error": err.Error(), "uploads": s.uploadPolicy(),
		})
	case errors.Is(err, runs.ErrTooLarge):
		writeJSON(w, http.StatusRequestEntityTooLarge, map[string]any{
			"error": err.Error(), "max_bytes": s.Runs.Policy().MaxBytes,
		})
	default:
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
	}
}

func (s *Server) listRuns(w http.ResponseWriter, r *http.Request) {
	dev := s.Store.DeviceByCredential(bearer(r))
	if dev == nil || s.Runs == nil {
		writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
		return
	}
	list := s.Runs.ListFor(s.visibleDevices(dev))
	if list == nil {
		list = []runs.Meta{}
	}
	writeJSON(w, http.StatusOK, map[string]any{
		"runs": list,
		// Says whose history this is, so a client can show "3 devices" rather than leaving the
		// user to wonder why runs from another phone appeared.
		"scope": map[string]any{
			"account_id": dev.AccountID,
			"devices":    len(s.visibleDevices(dev)),
		},
	})
}

func (s *Server) getRun(w http.ResponseWriter, r *http.Request) {
	dev := s.Store.DeviceByCredential(bearer(r))
	if dev == nil || s.Runs == nil {
		writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
		return
	}
	// Resolved against the caller's own devices only, so a run id from another account is not
	// found rather than being fetched from wherever it happens to live.
	owner, ok := s.Runs.OwnerOf(s.visibleDevices(dev), r.PathValue("id"))
	if !ok {
		writeJSON(w, http.StatusNotFound, map[string]string{"error": "no such run"})
		return
	}
	b, err := s.Runs.Get(owner, r.PathValue("id"))
	if err != nil {
		writeJSON(w, http.StatusNotFound, map[string]string{"error": "no such run"})
		return
	}
	w.Header().Set("Content-Type", "application/json")
	_, _ = w.Write(b)
}

func (s *Server) deleteRun(w http.ResponseWriter, r *http.Request) {
	dev := s.Store.DeviceByCredential(bearer(r))
	if dev == nil || s.Runs == nil {
		writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
		return
	}
	owner, ok := s.Runs.OwnerOf(s.visibleDevices(dev), r.PathValue("id"))
	if !ok {
		w.WriteHeader(http.StatusNoContent) // delete is idempotent; absent is the desired state
		return
	}
	if err := s.Runs.Delete(owner, r.PathValue("id")); err != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

// maxOrEmpty renders an unbounded ceiling as "" rather than as a sentinel version, so a client
// reading the profile cannot mistake a placeholder for a real bound.
func maxOrEmpty(r compat.Range) string {
	if !r.HasMax {
		return ""
	}
	return r.Max.String()
}

// EnrollmentLink builds the §2.1 bootstrap string for a freshly minted token.
//
// The server assembles it rather than the operator, because it is the only party that knows all
// three parts at once — its own URL, its own SPKI pin, and the token. An operator copying a pin
// by hand is the step that goes wrong, and a pin wrong by one character does not fail loudly.
func (s *Server) EnrollmentLink(token string) string {
	return EnrollmentURI(s.PublicControlURL, s.PinB64, token)
}

// EnrollmentURI is the same assembly without a running server, for the mint-a-link CLI action.
// Shared rather than reimplemented: two copies of this encoding would eventually disagree, and
// the failure mode is a pin that looks right and produces an inscrutable TLS error.
func EnrollmentURI(publicURL, pinB64, token string) string {
	return "echolot://enroll?v=1" +
		"&u=" + url.QueryEscape(strings.TrimRight(publicURL, "/")) +
		"&p=" + url.QueryEscape("pin-sha256:"+pinB64) +
		"&t=" + url.QueryEscape(token)
}

// target describes where this server can be measured, so a client can say which address a result
// came from instead of "the server".
//
// The alternates matter as much as the primaries: RFC 5780 behaviour discovery needs a second
// address to redirect to, and an operator reading a report needs to know which of their addresses
// a finding refers to. [fallback] is used only when nothing was configured explicitly, so a server
// that has not been told its own addresses still answers with something usable.
func (s *Server) target(fallback string) map[string]any {
	t := map[string]any{
		"id":        s.Name,
		"udp_port":  s.UDPPort,
		"tcp_port":  s.TCPPort,
		"stun_port": s.StunPort,
	}
	ip4 := s.IP4
	if ip4 == "" {
		ip4 = fallback
	}
	for k, v := range map[string]string{
		"ip4": ip4, "ip6": s.IP6, "ip4_alt": s.IP4Alt, "ip6_alt": s.IP6Alt,
	} {
		if v != "" {
			t[k] = v
		}
	}
	return t
}

// upstreamJSON renders the upstream tally with the derived figures already computed, so every
// consumer does not have to repeat (and risk fumbling) the same arithmetic.
func upstreamJSON(sess *session.Session) map[string]any {
	u := sess.Upstream()
	return map[string]any{
		"packets": u.Packets, "bytes": u.Bytes,
		"span_ms": u.SpanMs(), "kbps": u.Kbps(),
	}
}

// authInfo advertises the sign-in configuration, so the app can present a Sign in button only
// when there is something behind it, and can drive the flow without the user typing an issuer URL.
func (s *Server) authInfo(ctx context.Context) map[string]any {
	if s.OIDC == nil || !s.OIDC.Config().Enabled() {
		return map[string]any{"enabled": false}
	}
	cfg := s.OIDC.Config()
	out := map[string]any{
		"enabled": true,
		"issuer":  cfg.Issuer,
		// The app's client, not the server's: this is what a phone should authorize as.
		"client_id": cfg.AppClientID,
		// The app is a public client on a phone: no secret can be kept, so PKCE is what
		// protects the code exchange (RFC 7636), and the redirect comes back through the
		// scheme the app already registers for enrollment links.
		"flow":         "authorization_code+pkce",
		"redirect_uri": "echolot://auth",
		"scopes":       "openid profile email",
	}
	if d, err := s.OIDC.Discover(ctx); err == nil {
		out["authorization_endpoint"] = d.AuthorizationEndpoint
		out["token_endpoint"] = d.TokenEndpoint
		out["end_session_endpoint"] = d.EndSessionEndpoint
	} else {
		// Reported rather than hidden: an unreachable IdP is the operator's problem to see, and
		// a client that knows the difference can say "sign-in is configured but the provider is
		// not answering" instead of failing obscurely.
		out["discovery_error"] = err.Error()
	}
	return out
}

// linkAccount ties the calling device to the person whose ID token it presents.
//
// The device credential proves *which device*; the ID token proves *which person*. Both are
// required, and neither substitutes for the other: enrollment admits a device to the server,
// signing in attributes it to someone.
func (s *Server) linkAccount(w http.ResponseWriter, r *http.Request) {
	dev := s.Store.DeviceByCredential(bearer(r))
	if dev == nil {
		writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
		return
	}
	if s.OIDC == nil || !s.OIDC.Config().Enabled() {
		writeJSON(w, http.StatusNotImplemented, map[string]string{
			"error": "this server has no identity provider configured, so there is nothing to sign in to",
		})
		return
	}
	var body struct {
		IDToken string `json:"id_token"`
	}
	if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.IDToken == "" {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "expected an id_token"})
		return
	}
	claims, err := s.OIDC.Verify(r.Context(), body.IDToken)
	if err != nil {
		// Deliberately terse to the client and detailed in the log: a caller probing token
		// handling should not be told which check it failed.
		slog.Info("rejected sign-in", "device", dev.ID, "err", err)
		writeJSON(w, http.StatusForbidden, map[string]string{"error": "the identity token was not accepted"})
		return
	}
	if err := s.Store.LinkAccount(dev.ID, claims.AccountID(), claims.Display()); err != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
		return
	}
	slog.Info("device linked to account", "device", dev.ID, "account", claims.AccountID(),
		"admin", s.OIDC.IsAdmin(claims))
	writeJSON(w, http.StatusOK, map[string]any{
		"account_id": claims.AccountID(), "display_name": claims.Display(),
		"admin": s.OIDC.IsAdmin(claims),
	})
}

// unlinkAccount signs out on this device. The device stays enrolled: signing out should not cost
// someone their enrollment, which an operator had to grant.
func (s *Server) unlinkAccount(w http.ResponseWriter, r *http.Request) {
	dev := s.Store.DeviceByCredential(bearer(r))
	if dev == nil {
		writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
		return
	}
	if err := s.Store.LinkAccount(dev.ID, "", ""); err != nil {
		writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
		return
	}
	w.WriteHeader(http.StatusNoContent)
}

func (s *Server) accountStatus(w http.ResponseWriter, r *http.Request) {
	dev := s.Store.DeviceByCredential(bearer(r))
	if dev == nil {
		writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{
		"signed_in":    dev.LinkedToAccount(),
		"account_id":   dev.AccountID,
		"display_name": dev.AccountName,
		"device_id":    dev.ID,
	})
}

// visibleDevices is the set of devices whose runs the caller may read.
//
// Signed in: every device on the same account, which is what an account is for. Not signed in:
// only itself — anonymous devices are not a group, and treating the absent account as a shared
// one would let any of them read all the others.
func (s *Server) visibleDevices(dev *store.Device) []string {
	if dev.LinkedToAccount() {
		if ids := s.Store.DeviceIDsForAccount(dev.AccountID); len(ids) > 0 {
			return ids
		}
	}
	return []string{dev.ID}
}