server: STUN, TCP echo, observations API, delayed-echo + connect-back actions

- stun: RFC 5389 binding responder + RFC 5780 attributes (OTHER-ADDRESS,
  RESPONSE-ORIGIN, CHANGE-REQUEST) on a primary/alt-port socket grid per
  address; advertises stun-5780 with >=2 same-family addrs, else
  stun-basic. Unmodified framing for tooling interop. Tested.
- tcpecho: JSON greeting with observed src + TCP_INFO MSS/options
  (Linux getsockopt; zeroed elsewhere via build tags), then byte echo.
- session: per-packet UDP observations + connect-back results, ByID lookup.
- control: GET /v1/sessions/{id}/observations, POST .../actions
  (delayed_echo → DELAYED_ECHO at the observed data-plane source;
  connect_back → dial the control-plane source, record connected/refused/
  timeout+rtt). Capabilities computed from what is actually wired.
- config/main: comma-separated STUN listeners; all planes bind explicit
  addresses; graceful shutdown of the new listeners.

Full flow smoke-tested; go test green (stun binding/change-port,
dataplane wire format).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

server/internal/control/control.go

@@ -7,14 +7,19 @@
 package control
 
 import (
+	"crypto/rand"
 	"crypto/sha256"
 	"crypto/tls"
 	"crypto/x509"
 	"encoding/base64"
+	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"log/slog"
+	"net"
 	"net/http"
 	"net/netip"
+	"strconv"
 	"strings"
 	"time"
 
@@ -29,12 +34,19 @@ type Server struct {
 	Store    *store.Store
 	Sessions *session.Manager
 	Name     string
-	// Targets/capabilities for the profile response. The skeleton offers only
+	// Targets/capabilities for the profile response. The server offers only
 	// what it actually implements; the registry grows with the code.
-	UDPPort int
-	TCPPort int
+	UDPPort  int
+	TCPPort  int
+	StunPort int
 	// SPKI pin of the serving cert, for the profile's pins[] field.
 	PinB64 string
+	// 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
 }
 
 func (s *Server) Handler() http.Handler {
@@ -43,11 +55,131 @@ func (s *Server) Handler() http.Handler {
 	mux.HandleFunc("GET /v1/profile", s.profile)
 	mux.HandleFunc("POST /v1/sessions", s.newSession)
 	mux.HandleFunc("DELETE /v1/sessions/{id}", s.deleteSession)
-	// TODO(spec §5, §6): /v1/sessions/{id}/actions, /v1/sessions/{id}/observations
-	// TODO(spec §4): POST /v1/echo, GET /v1/tls-reference
+	mux.HandleFunc("GET /v1/sessions/{id}/observations", s.observations)
+	mux.HandleFunc("POST /v1/sessions/{id}/actions", s.actions)
+	// TODO(spec §4): POST /v1/echo, GET /v1/tls-reference; TLS-echo/JA4
+	// TODO(spec §5): downtrain, big_send, frag_send, throughput
 	return mux
 }
 
+// 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())
+		}
+	}
+	writeJSON(w, http.StatusOK, map[string]any{
+		"udp":          map[string]any{"packets_seen": packetsSeen, "packets": udp},
+		"tcp":          tcp,
+		"connect_back": cb,
+		// TODO(spec §6): http echo records, dns_canary
+	})
+}
+
+// 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"`
+	}
+	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})
+	default:
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "unknown or unimplemented action"})
+	}
+}
+
+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 {
@@ -103,12 +235,13 @@ func (s *Server) profile(w http.ResponseWriter, r *http.Request) {
 		// 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": []string{"udp-probe"},
+		"capabilities": s.Capabilities,
 		"targets": []map[string]any{{
-			"id":       s.Name,
-			"ip4":      host, // TODO: explicit configured addresses, v6, second STUN addr
-			"udp_port": s.UDPPort,
-			"tcp_port": s.TCPPort,
+			"id":        s.Name,
+			"ip4":       host, // TODO: explicit configured addresses, v6, second STUN addr
+			"udp_port":  s.UDPPort,
+			"tcp_port":  s.TCPPort,
+			"stun_port": s.StunPort,
 		}},
 		"pins":      []string{"pin-sha256:" + s.PinB64},
 		"next_pins": []string{},