From 7e1015c211f26377c480d42ef970ae9d17116999 Mon Sep 17 00:00:00 2001 From: mrambossek Date: Sat, 1 Aug 2026 10:09:13 +0200 Subject: [PATCH] =?UTF-8?q?server:=20=C2=A73.4=20asymmetric=20grants=20+?= =?UTF-8?q?=20downtrain=20and=20big=5Fsend=20actions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The grant is the keystone that makes server->client sends safe: created only by an authenticated control-plane action, bound at creation to the session's OBSERVED data-plane source (so it can never be aimed at a third party), and bounded by bytes, average rate and expiry. Sends stop the moment the budget runs out, so a buggy action cannot become a flood. Two granted actions on top of it: - downtrain: N packets at a given size/interval toward the client, with seq + send-timestamp in the payload — downstream loss/reorder/jitter, which an upstream-only train cannot measure. - big_send: one datagram per requested size, echoing the intended size in the payload — downstream MTU / black-hole evidence the client cannot produce for itself (only the far end can emit a large packet toward it). Tests cover the security properties: no grant without a verified destination, client requests clamped to server limits, byte budget stops sending exactly, expiry refuses, and the rate ceiling throttles a burst. Capabilities gain downtrain + big-send. Co-Authored-By: Claude Opus 5 --- server/cmd/echolot-server/main.go | 4 +- server/internal/control/control.go | 88 +++++++++++++++++++++-- server/internal/dataplane/granted.go | 87 ++++++++++++++++++++++ server/internal/dataplane/udp.go | 3 + server/internal/session/grant.go | 100 ++++++++++++++++++++++++++ server/internal/session/grant_test.go | 94 ++++++++++++++++++++++++ server/internal/session/session.go | 3 + 7 files changed, 372 insertions(+), 7 deletions(-) create mode 100644 server/internal/dataplane/granted.go create mode 100644 server/internal/session/grant.go create mode 100644 server/internal/session/grant_test.go diff --git a/server/cmd/echolot-server/main.go b/server/cmd/echolot-server/main.go index 029fde8..1fe0866 100644 --- a/server/cmd/echolot-server/main.go +++ b/server/cmd/echolot-server/main.go @@ -108,7 +108,7 @@ func serve(cfg *config.Config) error { TLSConfig: &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12}, } - caps := []string{"udp-probe", "delayed-echo", "connect-back", "http-echo"} + caps := []string{"udp-probe", "delayed-echo", "connect-back", "http-echo", "downtrain", "big-send"} if len(config.Addrs(cfg.TCPListen)) > 0 { caps = append(caps, "tcp-echo", "tls-echo") } @@ -118,6 +118,8 @@ func serve(cfg *config.Config) error { UDPPort: mustPort(firstAddr(cfg.UDPListen)), TCPPort: mustPort(firstAddr(cfg.TCPListen)), StunPort: mustPort(firstAddr(cfg.StunListen)), PinB64: pin, CertChain: cert.Certificate, DelayedEcho: dp.SendDelayedEcho, + DownTrain: dp.DownTrain, + BigSend: dp.BigSend, TCPRecent: func(ip string) any { return tcpSrv.RecentFor(ip) }, } diff --git a/server/internal/control/control.go b/server/internal/control/control.go index cfcef9f..ee7cd72 100644 --- a/server/internal/control/control.go +++ b/server/internal/control/control.go @@ -49,6 +49,9 @@ type Server struct { 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 func(sess *session.Session, g *session.Grant, count, sizeBytes, intervalUs int) (int, error) + BigSend func(sess *session.Session, g *session.Grant, sizes []int) ([]int, error) // 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. @@ -70,8 +73,7 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /v1/sessions/{id}/actions", s.actions) mux.HandleFunc("POST /v1/echo", s.httpEcho) mux.HandleFunc("GET /v1/tls-reference", s.tlsReference) - // TODO(spec §4): TLS-echo/JA4 (tls-echo capability, needs ClientHello capture) - // TODO(spec §5): downtrain, big_send, frag_send, throughput + // TODO(spec §5): frag_send, throughput (both build on the same grant machinery) return mux } @@ -146,10 +148,14 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) { return } var req struct { - Action string `json:"action"` - DelayS int `json:"delay_s"` - Protocol string `json:"protocol"` - Port int `json:"port"` + 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"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeJSON(w, http.StatusBadRequest, map[string]string{"error": "bad body"}) @@ -200,11 +206,81 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) { 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 + } + // 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) + slog.Info("downtrain finished", "action", actionID, "sent", sent, "bytes", g.Sent(), "err", err) + }() + writeJSON(w, http.StatusAccepted, 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}, + }) + + 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] + } + 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() { + attempted, err := s.BigSend(sess, g, sizes) + slog.Info("big_send finished", "action", actionID, "attempted", attempted, "err", err) + }() + writeJSON(w, http.StatusAccepted, map[string]any{ + "action_id": actionID, "sizes_bytes": sizes, + "grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps}, + }) + default: writeJSON(w, http.StatusBadRequest, map[string]string{"error": "unknown or unimplemented action"}) } } +// dataMinPacket is the smallest datagram that still carries a header + a little payload. +const dataMinPacket = 40 + +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() diff --git a/server/internal/dataplane/granted.go b/server/internal/dataplane/granted.go new file mode 100644 index 0000000..1e155c2 --- /dev/null +++ b/server/internal/dataplane/granted.go @@ -0,0 +1,87 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +package dataplane + +import ( + "encoding/binary" + "fmt" + "time" + + "echo-lot.app/server/internal/session" +) + +// Server-to-client sends that exceed the request size, and are therefore only legal under an +// asymmetric grant (spec §3.4). Every one of them: +// - targets the session's observed data-plane source, fixed when the grant was created; +// - stops the moment the grant's byte/rate/time budget is exhausted; +// - is HMAC-signed with the session key, so the client can tell our packets from injected ones. + +// DownTrain sends `count` DOWNTRAIN_DATA packets of `sizeBytes` spaced `intervalUs` apart. The +// client measures downstream loss, reordering and jitter from what arrives — the direction an +// upstream-only train cannot see. Returns how many packets actually went out (the grant may cut +// it short, which is itself reportable). +func (s *Server) DownTrain(sess *session.Session, g *session.Grant, count, sizeBytes, intervalUs int) (int, error) { + target := sess.DataSource() + if !target.IsValid() { + return 0, fmt.Errorf("no observed data-plane source") + } + conn := s.connFor(target) + if conn == nil { + return 0, fmt.Errorf("no data-plane socket matches target family") + } + if sizeBytes < HeaderSize+8 { + sizeBytes = HeaderSize + 8 + } + payload := make([]byte, sizeBytes-HeaderSize) + sent := 0 + for i := 0; i < count; i++ { + if !g.Allow(sizeBytes) { + break // budget or rate exhausted — stop, do not sleep it off + } + // Sequence + send timestamp in the payload head so the client can order and time them + // even when packets arrive out of order. + binary.BigEndian.PutUint32(payload[0:4], uint32(i)) + binary.BigEndian.PutUint32(payload[4:8], uint32(time.Since(s.start).Microseconds())) + s.send(conn, target, sess, TypeDownTrainData, uint32(i), payload) + sent++ + if intervalUs > 0 && i < count-1 { + time.Sleep(time.Duration(intervalUs) * time.Microsecond) + } + } + return sent, nil +} + +// BigSend transmits one datagram per requested size, largest-first metadata intact, so the client +// can see which sizes survive the *downstream* path — the mtu.pmtud_down / mtu.blackhole evidence. +// The client cannot produce this itself: only the far end can emit a large packet toward it. +func (s *Server) BigSend(sess *session.Session, g *session.Grant, sizes []int) ([]int, error) { + target := sess.DataSource() + if !target.IsValid() { + return nil, fmt.Errorf("no observed data-plane source") + } + conn := s.connFor(target) + if conn == nil { + return nil, fmt.Errorf("no data-plane socket matches target family") + } + attempted := make([]int, 0, len(sizes)) + for i, size := range sizes { + if size < HeaderSize+8 { + size = HeaderSize + 8 + } + if size > 9000 { // jumbo ceiling; beyond this the kernel will refuse anyway + size = 9000 + } + if !g.Allow(size) { + break + } + payload := make([]byte, size-HeaderSize) + // Echo the intended size into the payload so a truncated/fragmented arrival is + // still attributable to the size we meant to send. + binary.BigEndian.PutUint32(payload[0:4], uint32(size)) + s.send(conn, target, sess, TypeBigSend, uint32(i), payload) + attempted = append(attempted, size) + time.Sleep(20 * time.Millisecond) // keep bursts from being read as congestion loss + } + return attempted, nil +} diff --git a/server/internal/dataplane/udp.go b/server/internal/dataplane/udp.go index 4ab9bcb..f48782f 100644 --- a/server/internal/dataplane/udp.go +++ b/server/internal/dataplane/udp.go @@ -32,6 +32,9 @@ const ( TypeMtuProbe = 0x09 TypeMtuAck = 0x0A TypeDelayedEcho = 0x0B + // Server->client under an asymmetric grant (spec §3.4/§5). + TypeDownTrainData = 0x06 + TypeBigSend = 0x0C ) type Server struct { diff --git a/server/internal/session/grant.go b/server/internal/session/grant.go new file mode 100644 index 0000000..7aebda6 --- /dev/null +++ b/server/internal/session/grant.go @@ -0,0 +1,100 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +package session + +import ( + "sync" + "time" +) + +// Grant is the spec §3.4 "asymmetric grant": the ONLY thing that lets the server send more than +// it receives. Without one, every response is capped at the request size, which is what keeps an +// unauthenticated observer from using this server as a reflector/amplifier. +// +// A grant is created by an authenticated control-plane action (§5) for one session, and is bounded +// three ways: total bytes, send rate, and wall-clock expiry. It is consumed as the data plane +// sends; when the budget runs out the send stops, so a bug in an action cannot turn into an +// unbounded flood. +type Grant struct { + ActionID string + // Destination is fixed at creation to the session's observed data-plane source — a grant can + // never be pointed somewhere else, so it cannot be used to attack a third party. + Dest string + MaxBytes int64 + MaxKbps int + ExpiresAt time.Time + + mu sync.Mutex + sentBytes int64 + started time.Time +} + +// GrantLimits are the server-side ceilings an action may not exceed, independent of what the +// client asks for. +type GrantLimits struct { + MaxBytes int64 + MaxKbps int + MaxHold time.Duration +} + +var DefaultGrantLimits = GrantLimits{ + MaxBytes: 8 << 20, // 8 MiB per action + MaxKbps: 50_000, // matches the profile's advertised max_kbps + MaxHold: 30 * time.Second, // an action must finish inside this window +} + +// NewGrant clamps the request to the server's limits and binds it to the session's observed +// data-plane source. Returns nil when the session has no observed source yet — refusing to send +// anywhere we have not verifiably received from is the whole point. +func (s *Session) NewGrant(actionID string, wantBytes int64, wantKbps int, lim GrantLimits) *Grant { + dest := s.DataSource() + if !dest.IsValid() { + return nil + } + if wantBytes <= 0 || wantBytes > lim.MaxBytes { + wantBytes = lim.MaxBytes + } + if wantKbps <= 0 || wantKbps > lim.MaxKbps { + wantKbps = lim.MaxKbps + } + g := &Grant{ + ActionID: actionID, Dest: dest.String(), + MaxBytes: wantBytes, MaxKbps: wantKbps, + ExpiresAt: time.Now().Add(lim.MaxHold), + started: time.Now(), + } + s.mu.Lock() + s.grants = append(s.grants, g) + s.mu.Unlock() + return g +} + +// Allow reports whether n more bytes may be sent now, consuming the budget when they may. It +// enforces the byte ceiling, the expiry, and the average rate (by refusing early sends rather +// than sleeping, so callers stay in control of pacing). +func (g *Grant) Allow(n int) bool { + g.mu.Lock() + defer g.mu.Unlock() + if time.Now().After(g.ExpiresAt) { + return false + } + if g.sentBytes+int64(n) > g.MaxBytes { + return false + } + // Average-rate check: bytes allowed so far = kbps/8 * elapsed_seconds. + elapsed := time.Since(g.started).Seconds() + allowed := float64(g.MaxKbps) * 125 * elapsed // kbps -> bytes/s is kbps*1000/8 = kbps*125 + if elapsed > 0.05 && float64(g.sentBytes+int64(n)) > allowed { + return false + } + g.sentBytes += int64(n) + return true +} + +// Sent returns how many bytes this grant has consumed. +func (g *Grant) Sent() int64 { + g.mu.Lock() + defer g.mu.Unlock() + return g.sentBytes +} diff --git a/server/internal/session/grant_test.go b/server/internal/session/grant_test.go new file mode 100644 index 0000000..e472283 --- /dev/null +++ b/server/internal/session/grant_test.go @@ -0,0 +1,94 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +package session + +import ( + "net/netip" + "testing" + "time" +) + +func sessionWithSource(t *testing.T) *Session { + t.Helper() + m := NewManager(time.Minute) + s, _, err := m.New("dev", "cred", netip.MustParseAddr("127.0.0.1")) + if err != nil { + t.Fatal(err) + } + s.NoteDataSource(netip.MustParseAddrPort("127.0.0.1:5000")) + return s +} + +// Without an observed data-plane source there is nowhere verified to send, so no grant may exist. +func TestNoGrantWithoutObservedSource(t *testing.T) { + m := NewManager(time.Minute) + s, _, _ := m.New("dev", "cred", netip.MustParseAddr("127.0.0.1")) + if g := s.NewGrant("a1", 1000, 1000, DefaultGrantLimits); g != nil { + t.Fatal("granted a send with no verified destination") + } +} + +func TestGrantIsBoundToObservedSourceAndClamped(t *testing.T) { + s := sessionWithSource(t) + g := s.NewGrant("a1", 1<<40, 1<<30, DefaultGrantLimits) // absurd request + if g == nil { + t.Fatal("expected a grant") + } + if g.Dest != "127.0.0.1:5000" { + t.Fatalf("grant destination = %s, want the observed source", g.Dest) + } + if g.MaxBytes != DefaultGrantLimits.MaxBytes || g.MaxKbps != DefaultGrantLimits.MaxKbps { + t.Fatalf("client request was not clamped to server limits: %d bytes / %d kbps", + g.MaxBytes, g.MaxKbps) + } +} + +// The byte budget must actually stop sending — this is the anti-amplification guarantee. +func TestGrantStopsAtByteBudget(t *testing.T) { + s := sessionWithSource(t) + g := s.NewGrant("a1", 1000, 0, DefaultGrantLimits) + sent := 0 + for i := 0; i < 100; i++ { + if !g.Allow(100) { + break + } + sent += 100 + } + if sent != 1000 { + t.Fatalf("sent %d bytes, want exactly the 1000-byte budget", sent) + } + if g.Allow(1) { + t.Fatal("grant allowed a send after the budget was exhausted") + } + if g.Sent() != 1000 { + t.Fatalf("Sent() = %d, want 1000", g.Sent()) + } +} + +func TestExpiredGrantRefuses(t *testing.T) { + s := sessionWithSource(t) + g := s.NewGrant("a1", 10_000, 0, GrantLimits{MaxBytes: 10_000, MaxKbps: 1000, MaxHold: time.Millisecond}) + time.Sleep(5 * time.Millisecond) + if g.Allow(10) { + t.Fatal("expired grant still allowed a send") + } +} + +// The rate ceiling must throttle a burst that is well inside the byte budget. +func TestGrantEnforcesRate(t *testing.T) { + s := sessionWithSource(t) + // 8 kbps = 1000 bytes/s. A burst far beyond one second's worth must be refused. + g := s.NewGrant("a1", 1<<20, 8, DefaultGrantLimits) + time.Sleep(60 * time.Millisecond) // let the rate window open a little + sent := 0 + for i := 0; i < 1000; i++ { + if !g.Allow(100) { + break + } + sent += 100 + } + if sent > 5000 { + t.Fatalf("rate limit let %d bytes through in ~60ms at 8kbps", sent) + } +} diff --git a/server/internal/session/session.go b/server/internal/session/session.go index 3a39f24..b924b32 100644 --- a/server/internal/session/session.go +++ b/server/internal/session/session.go @@ -32,6 +32,9 @@ type Session struct { maxSeq uint32 window [16]uint64 + // Active asymmetric grants (spec §3.4/§5) — the only licence to send more than we receive. + grants []*Grant + // Observations (spec §6): per-packet UDP view + connect-back results. packetsSeen uint64 udpObs []UDPObservation // ring, newest last, cap obsCap