From 8118e213ae8b71fc4fa481d544f73a0e123309b0 Mon Sep 17 00:00:00 2001 From: mrambossek Date: Sun, 2 Aug 2026 13:04:54 +0200 Subject: [PATCH] server: upstream trains, observed TTL/DSCP/ECN, rate limits, action ids Types 0x03/0x04/0x05 land with a bounded columnar train buffer (head kept, truncation declared) and grant-free multi-part reports - a report row is smaller than the packet it answers, so $3.4 holds without a grant. The read loop now collects TTL/TOS cmsgs on Linux, replacing the 0xFF stubs in the observation block with what the kernel saw; downtrain gained a dscp parameter, so DSCP survival is measurable in both directions. Rate limiting ($2.5) exists now: per-credential AND per-source buckets, 429 on the control plane, silent drop on the data plane after the HMAC gate and before the replay window. UDP ceilings default above the largest legitimate run - a limit that clips a real measurement produces a confidently wrong number. Every granted packet carries its action_id at payload[8:16]; overlapping actions were unattributable before. Canary DNS logs now honor the stated 24h privacy default. /admin/enroll-tokens answers the spec's JSON shape. protocol_version 1.0.1 (additive). Co-Authored-By: Claude Opus 5 --- docs/probe-protocol.md | 7 +- server/README.md | 7 +- server/cmd/echolot-server/main.go | 20 ++- server/internal/adminui/auth.go | 39 ++++ server/internal/adminui/enrolltokens_test.go | 88 +++++++++ server/internal/adminui/pages.go | 35 ++++ server/internal/canarydns/canarydns.go | 39 +++- server/internal/canarydns/canarydns_test.go | 29 ++- server/internal/config/config.go | 17 ++ server/internal/control/control.go | 134 ++++++++++++-- server/internal/control/dscp_test.go | 27 +++ server/internal/dataplane/cmsg_linux.go | 70 ++++++++ server/internal/dataplane/cmsg_other.go | 15 ++ server/internal/dataplane/frag_linux.go | 9 +- server/internal/dataplane/granted.go | 57 ++++-- server/internal/dataplane/throughput.go | 8 +- server/internal/dataplane/tos_linux.go | 55 ++++++ server/internal/dataplane/tos_other.go | 16 ++ server/internal/dataplane/train.go | 122 +++++++++++++ server/internal/dataplane/train_test.go | 179 +++++++++++++++++++ server/internal/dataplane/udp.go | 110 ++++++++++-- server/internal/ratelimit/ratelimit.go | 100 +++++++++++ server/internal/ratelimit/ratelimit_test.go | 105 +++++++++++ server/internal/session/session.go | 3 + server/internal/session/train.go | 98 ++++++++++ server/internal/session/train_test.go | 62 +++++++ 26 files changed, 1390 insertions(+), 61 deletions(-) create mode 100644 server/internal/adminui/enrolltokens_test.go create mode 100644 server/internal/control/dscp_test.go create mode 100644 server/internal/dataplane/cmsg_linux.go create mode 100644 server/internal/dataplane/cmsg_other.go create mode 100644 server/internal/dataplane/tos_linux.go create mode 100644 server/internal/dataplane/tos_other.go create mode 100644 server/internal/dataplane/train.go create mode 100644 server/internal/dataplane/train_test.go create mode 100644 server/internal/ratelimit/ratelimit.go create mode 100644 server/internal/ratelimit/ratelimit_test.go create mode 100644 server/internal/session/train.go create mode 100644 server/internal/session/train_test.go diff --git a/docs/probe-protocol.md b/docs/probe-protocol.md index 148e091..1990aa7 100644 --- a/docs/probe-protocol.md +++ b/docs/probe-protocol.md @@ -84,7 +84,12 @@ The app re-fetches the profile at the start of every run (falling back to the ca ### 2.3 Capabilities (v1 registry) -`udp-probe`, `stun-basic`, `stun-5780`, `canary-dns`, `recursive-dns`, `connect-back`, `delayed-echo`, `big-send`, `frag-send`, `tls-echo`, `http-echo`, `throughput`, `ntp`. A server omits what it can't offer (e.g. `stun-5780` without a second IP degrades to `stun-basic`). Clients must skip, and record as `unsupported`, any test whose capability is absent. Unknown capability strings are ignored. +`udp-probe`, `stun-basic`, `stun-5780`, `canary-dns`, `recursive-dns`, `connect-back`, `delayed-echo`, `big-send`, `frag-send`, `tls-echo`, `http-echo`, `throughput`, `ntp`, plus: + +- `downtrain` — server-sent downstream trains via the §5 `downtrain` action. Upstream trains need no capability of their own: they are plain client-sent data-plane packets and ride `udp-probe`. +- `tcp-echo` — the plain-TCP echo endpoint (§4); `tls-echo` is its ALPN variant on the same port. + +A server omits what it can't offer (e.g. `stun-5780` without a second IP degrades to `stun-basic`). Clients must skip, and record as `unsupported`, any test whose capability is absent. Unknown capability strings are ignored. ### 2.4 Sessions diff --git a/server/README.md b/server/README.md index f4719aa..010bad3 100644 --- a/server/README.md +++ b/server/README.md @@ -130,8 +130,11 @@ inject one. Running your own release pipeline? Mint a keypair with ## First contact ```sh -# 1. mint an enrollment token (admin listener is loopback-only) -curl -s -X POST 'http://127.0.0.1:8444/admin/enroll-tokens?note=phone' +# 1. mint an enrollment token (admin listener is loopback-only; authenticates as the +# break-glass admin — set that once with --set-admin-password) +curl -s -u admin: -H 'Accept: application/json' \ + -X POST 'http://127.0.0.1:8444/admin/enroll-tokens?note=phone' +# → { "token": "…", "expires_in_s": 86400, "enroll_uri": "echolot://enroll?…" } # 2. device enrolls with it (normally via the echolot:// QR code) curl -sk -X POST https://:8443/v1/enroll -H 'Authorization: Bearer ' # 3. device fetches its profile diff --git a/server/cmd/echolot-server/main.go b/server/cmd/echolot-server/main.go index bdd9939..85febff 100644 --- a/server/cmd/echolot-server/main.go +++ b/server/cmd/echolot-server/main.go @@ -46,6 +46,7 @@ import ( "echo-lot.app/server/internal/control" "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/selftest" "echo-lot.app/server/internal/selfupdate" @@ -205,6 +206,16 @@ func serve(cfg *config.Config) error { sessions := session.NewManager(15 * time.Minute) dp := &dataplane.Server{Sessions: sessions} + // Spec §2.5 ceilings. Control plane answers 429; the data plane drops silently. 0 = off. + if cfg.RateUDPPps > 0 { + pps := float64(cfg.RateUDPPps) + // Burst of two seconds' worth: a 5000-packet train arrives as one burst by design. + dp.PacketRate = ratelimit.New(pps, 2*pps) + } + if cfg.RateUDPKbps > 0 { + bytesPerSec := float64(cfg.RateUDPKbps) * 125 // kbps -> bytes/s + dp.ByteRate = ratelimit.New(bytesPerSec, bytesPerSec) + } // TCP echo shares the control cert for its elt-echo TLS variant. tcpSrv := &tcpecho.Server{ TLSConfig: &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12}, @@ -305,6 +316,12 @@ func serve(cfg *config.Config) error { ctl.FragSend = dp.FragSend } ctl.DownThroughput = dp.DownThroughput + if cfg.RateSessionsPerMin > 0 { + ctl.RateSessions = ratelimit.New(float64(cfg.RateSessionsPerMin)/60, float64(cfg.RateSessionsPerMin)) + } + if cfg.RateActionsPerMin > 0 { + ctl.RateActions = ratelimit.New(float64(cfg.RateActionsPerMin)/60, float64(cfg.RateActionsPerMin)) + } ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() @@ -598,7 +615,8 @@ func serve(cfg *config.Config) error { var dnsTCP []net.Listener if dnsAddrs := config.Addrs(cfg.DNSListen); len(dnsAddrs) > 0 && cfg.CanaryZone != "" { v4, v6 := firstByFamily(dnsAddrs) - cd := canarydns.New(cfg.CanaryZone, cfg.Name, v4, v6) + cd := canarydns.New(cfg.CanaryZone, cfg.Name, v4, v6, + time.Duration(cfg.DNSLogRetentionH)*time.Hour) for _, addr := range dnsAddrs { ua, err := net.ResolveUDPAddr("udp", addr) if err != nil { diff --git a/server/internal/adminui/auth.go b/server/internal/adminui/auth.go index e1b0a54..add8cd7 100644 --- a/server/internal/adminui/auth.go +++ b/server/internal/adminui/auth.go @@ -117,6 +117,10 @@ func (s *Server) Handler() http.Handler { // affects the whole server, so those stay with the admin. mux.HandleFunc("POST /devices/{id}/revoke", s.guard(s.adminOnly(s.revokeDevice))) mux.HandleFunc("POST /enroll-tokens", s.guard(s.adminOnly(s.mintToken))) + // The spec-shaped mint endpoint (§2.1: {token, expires_in_s, enroll_uri}), for curl and + // scripts. Authenticates its own way — see apiAdmin — because guard's redirect-to-login is + // useless to a caller without a browser. + mux.HandleFunc("POST /admin/enroll-tokens", s.enrollTokensAPI) return mux } @@ -235,6 +239,41 @@ func (s *Server) loginSubmit(w http.ResponseWriter, r *http.Request) { http.Redirect(w, r, "/", http.StatusSeeOther) } +// apiAdmin authenticates a programmatic admin request: the normal session cookie, or HTTP Basic +// against the break-glass credential for callers without a cookie jar (the README's curl). +// +// The cookie path keeps CSRF, exactly like guard: a cookie is an ambient credential and this +// endpoint changes state. Basic auth is exempt — the password is supplied explicitly per +// request, so there is nothing for a cross-site form to ride on — and a wrong guess pays the +// same throttle as the login form, so this is no better a password oracle than that is. +func (s *Server) apiAdmin(w http.ResponseWriter, r *http.Request) (subject string, ok bool) { + if sess := s.session(r); sess != nil { + if !sess.Admin { + http.Error(w, "that action needs an administrator account", http.StatusForbidden) + return "", false + } + if !s.csrfOK(r, sess) { + http.Error(w, "stale form — reload the page and try again", http.StatusForbidden) + return "", false + } + return sess.Subject, true + } + if user, pass, hasBasic := r.BasicAuth(); hasBasic { + if d := s.Throttle.Delay(); d > 0 { + time.Sleep(d) + } + if cred := s.Store.LocalAdmin(); cred != nil && cred.Verify(user, pass) { + s.Throttle.Succeeded() + return "local:" + user, true + } + s.Throttle.Failed() + slog.Info("admin api auth failed", "user", user, "from", clientIP(r)) + } + w.Header().Set("WWW-Authenticate", `Basic realm="echolot-admin"`) + http.Error(w, "authentication required", http.StatusUnauthorized) + return "", false +} + // ---- OIDC ------------------------------------------------------------------------------- func (s *Server) oidcAvailable() bool { diff --git a/server/internal/adminui/enrolltokens_test.go b/server/internal/adminui/enrolltokens_test.go new file mode 100644 index 0000000..9923824 --- /dev/null +++ b/server/internal/adminui/enrolltokens_test.go @@ -0,0 +1,88 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +package adminui + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "echo-lot.app/server/internal/adminauth" +) + +// tokenFixture wires just enough of the Server for the mint endpoint: a break-glass admin and +// a stand-in EnrollLink (the real one belongs to the control server, injected the same way). +func tokenFixture(t *testing.T) *Server { + t.Helper() + s, _, _, _ := fixture(t) + secret, err := s.Store.SessionSecret() + if err != nil { + t.Fatal(err) + } + s.Sessions = adminauth.NewSessions(secret, time.Hour) + s.Throttle = adminauth.NewThrottle() + cred, err := adminauth.NewCredential("admin", "a-long-test-password") + if err != nil { + t.Fatal(err) + } + if err := s.Store.SetLocalAdmin(cred); err != nil { + t.Fatal(err) + } + s.EnrollLink = func(tok string) string { return "echolot://enroll?v=1&t=" + tok } + return s +} + +func TestEnrollTokensAPISpecShape(t *testing.T) { + h := tokenFixture(t).Handler() + + // No credentials → 401 with a challenge, never a token. + req := httptest.NewRequest("POST", "/admin/enroll-tokens?note=phone", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusUnauthorized || rec.Header().Get("WWW-Authenticate") == "" { + t.Fatalf("unauthenticated: code=%d", rec.Code) + } + + // Wrong password → still 401. + req = httptest.NewRequest("POST", "/admin/enroll-tokens", nil) + req.SetBasicAuth("admin", "wrong") + rec = httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusUnauthorized { + t.Fatalf("bad password: code=%d, want 401", rec.Code) + } + + // Basic + Accept: application/json → the §2.1 shape. + req = httptest.NewRequest("POST", "/admin/enroll-tokens?note=phone", nil) + req.SetBasicAuth("admin", "a-long-test-password") + req.Header.Set("Accept", "application/json") + rec = httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("mint: code=%d body=%s", rec.Code, rec.Body.String()) + } + var body struct { + Token string `json:"token"` + ExpiresS int `json:"expires_in_s"` + EnrollURI string `json:"enroll_uri"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + if body.Token == "" || body.ExpiresS != 86400 || !strings.HasPrefix(body.EnrollURI, "echolot://enroll?") { + t.Fatalf("spec shape violated: %+v", body) + } + + // Without Accept: the browser flow — redirect to the QR page, link in the query. + req = httptest.NewRequest("POST", "/admin/enroll-tokens", nil) + req.SetBasicAuth("admin", "a-long-test-password") + rec = httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusSeeOther || !strings.HasPrefix(rec.Header().Get("Location"), "/devices?link=") { + t.Fatalf("html flow: code=%d location=%q", rec.Code, rec.Header().Get("Location")) + } +} diff --git a/server/internal/adminui/pages.go b/server/internal/adminui/pages.go index 357404a..dbafdea 100644 --- a/server/internal/adminui/pages.go +++ b/server/internal/adminui/pages.go @@ -167,6 +167,41 @@ func (s *Server) mintToken(w http.ResponseWriter, r *http.Request, sess *adminau http.Redirect(w, r, "/devices?link="+url.QueryEscape(s.EnrollLink(tok)), http.StatusSeeOther) } +// enrollTokensAPI is POST /admin/enroll-tokens, the endpoint the spec's §2.1 example names. +// Content-negotiated: Accept: application/json gets the spec shape {token, expires_in_s, +// enroll_uri}; anything else (a browser) gets the same redirect-to-QR flow as the form above, +// so the one path serves both audiences. +func (s *Server) enrollTokensAPI(w http.ResponseWriter, r *http.Request) { + subject, ok := s.apiAdmin(w, r) + if !ok { + return + } + note := r.URL.Query().Get("note") + if note == "" { + note = "admin-api" + } + const ttl = 24 * time.Hour + tok, err := s.Store.NewEnrollToken(ttl, note) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + slog.Info("enrolment token minted", "by", subject, "note", note) + if !strings.Contains(r.Header.Get("Accept"), "application/json") { + http.Redirect(w, r, "/devices?link="+url.QueryEscape(s.EnrollLink(tok)), http.StatusSeeOther) + return + } + w.Header().Set("Content-Type", "application/json") + // The whole link, not the bare token (§2.1): the server is the only party holding URL, pin + // and token at once, and a hand-assembled pin wrong by one character fails as an inscrutable + // TLS error later rather than loudly here. + _ = json.NewEncoder(w).Encode(map[string]any{ + "token": tok, + "expires_in_s": int(ttl.Seconds()), + "enroll_uri": s.EnrollLink(tok), + }) +} + // EnrollLink is supplied by the caller so this package does not need the control server's pin. var _ = 0 diff --git a/server/internal/canarydns/canarydns.go b/server/internal/canarydns/canarydns.go index 2a91e47..98e935b 100644 --- a/server/internal/canarydns/canarydns.go +++ b/server/internal/canarydns/canarydns.go @@ -63,20 +63,24 @@ type Server struct { nsName string // this server's own name for NS/authority answers primaryV4 netip.Addr primaryV6 netip.Addr + retention time.Duration // query-log age limit; <= 0 means only the ring cap bounds it - mu sync.Mutex - log []Query // ring, newest last - retainTo time.Time + mu sync.Mutex + log []Query // ring, newest last } const logCap = 8192 // New creates a server for zone (with or without trailing dot). nsName is the // server's own hostname (for the zone's NS record); primary v4/v6 are this -// host's addresses used to answer the zone apex / NS glue. -func New(zone, nsName string, v4, v6 netip.Addr) *Server { +// host's addresses used to answer the zone apex / NS glue. retention is how +// long logged queries are kept (spec §6; the privacy default is 24 h). +func New(zone, nsName string, v4, v6 netip.Addr, retention time.Duration) *Server { z := strings.ToLower(strings.TrimSuffix(zone, ".")) + "." - return &Server{zone: z, nsName: strings.TrimSuffix(nsName, ".") + ".", primaryV4: v4, primaryV6: v6} + return &Server{ + zone: z, nsName: strings.TrimSuffix(nsName, ".") + ".", + primaryV4: v4, primaryV6: v6, retention: retention, + } } // RecentForPrefix returns logged queries whose qname contains ".." @@ -84,6 +88,7 @@ func New(zone, nsName string, v4, v6 netip.Addr) *Server { func (s *Server) RecentForPrefix(prefix string) []Query { s.mu.Lock() defer s.mu.Unlock() + s.dropExpiredLocked(time.Now().UTC()) needle := "." + strings.ToLower(prefix) + "." var out []Query for _, q := range s.log { @@ -97,12 +102,34 @@ func (s *Server) RecentForPrefix(prefix string) []Query { func (s *Server) record(q Query) { s.mu.Lock() defer s.mu.Unlock() + s.dropExpiredLocked(q.At) if len(s.log) >= logCap { s.log = s.log[1:] } s.log = append(s.log, q) } +// dropExpiredLocked enforces the retention window on the query log. +// +// The 24-hour retention was advertised as the privacy default (spec §6/§7) and then not +// enforced: the ring only bounded *count*, so on a quiet server a resolver's queries could sit +// in memory for weeks. Aged out on every write and every read — whichever comes first — so an +// idle log still forgets on schedule the moment anyone looks. Entries are appended in time +// order, so expiry is always a prefix of the slice. +func (s *Server) dropExpiredLocked(now time.Time) { + if s.retention <= 0 { + return + } + cutoff := now.Add(-s.retention) + i := 0 + for i < len(s.log) && s.log[i].At.Before(cutoff) { + i++ + } + if i > 0 { + s.log = append([]Query(nil), s.log[i:]...) // reallocate so the old backing array frees + } +} + // ServeUDP / ServeTCP run read loops; call one per bound address. func (s *Server) ServeUDP(conn *net.UDPConn) error { buf := make([]byte, 1500) diff --git a/server/internal/canarydns/canarydns_test.go b/server/internal/canarydns/canarydns_test.go index 97f5f1f..d7f6f97 100644 --- a/server/internal/canarydns/canarydns_test.go +++ b/server/internal/canarydns/canarydns_test.go @@ -8,6 +8,7 @@ import ( "net" "net/netip" "testing" + "time" ) // buildQuery makes a single-question DNS query, optionally with an EDNS OPT. @@ -65,7 +66,7 @@ func parseResponse(t *testing.T, resp []byte) (flags uint16, answers []ans) { } func newTestServer() *Server { - return New("c.echo-lot.app", "fmr", netip.MustParseAddr("192.0.2.1"), netip.MustParseAddr("2001:db8::1")) + return New("c.echo-lot.app", "fmr", netip.MustParseAddr("192.0.2.1"), netip.MustParseAddr("2001:db8::1"), 24*time.Hour) } func TestReferenceRecords(t *testing.T) { @@ -159,3 +160,29 @@ func TestOutOfZoneNXDomain(t *testing.T) { t.Fatalf("out-of-zone should be NXDOMAIN, flags=%#x", flags) } } + +func TestQueryLogRetentionForgetsOldEntries(t *testing.T) { + s := newTestServer() // 24 h retention + now := time.Now().UTC() + s.record(Query{QName: "old.sess1.c.echo-lot.app", At: now.Add(-25 * time.Hour)}) + s.record(Query{QName: "fresh.sess1.c.echo-lot.app", At: now}) + + got := s.RecentForPrefix("sess1") + if len(got) != 1 || got[0].QName != "fresh.sess1.c.echo-lot.app" { + t.Fatalf("retention not enforced: %+v", got) + } + + // Reads must age the log too: an idle server still has to forget on schedule. + s.log[0].At = now.Add(-25 * time.Hour) + if got := s.RecentForPrefix("sess1"); len(got) != 0 { + t.Fatalf("read path did not expire entries: %+v", got) + } +} + +func TestZeroRetentionKeepsEverything(t *testing.T) { + s := New("c.echo-lot.app", "fmr", netip.MustParseAddr("192.0.2.1"), netip.MustParseAddr("2001:db8::1"), 0) + s.record(Query{QName: "ancient.sess1.c.echo-lot.app", At: time.Now().UTC().Add(-1000 * time.Hour)}) + if got := s.RecentForPrefix("sess1"); len(got) != 1 { + t.Fatal("retention 0 must mean 'ring cap only', not 'keep nothing'") + } +} diff --git a/server/internal/config/config.go b/server/internal/config/config.go index 7204a2e..03d43a9 100644 --- a/server/internal/config/config.go +++ b/server/internal/config/config.go @@ -83,6 +83,18 @@ type Config struct { UploadMaxRuns int // ECHOLOT_UPLOAD_MAX_RUNS / --upload-max-runs (per device) UploadMinAnon string // ECHOLOT_UPLOAD_MIN_ANONYMIZATION / --upload-min-anonymization + // Rate limits (spec §2.5), each applied per credential and per source IP. 0 disables a + // ceiling. The UDP ceilings are deliberately above the largest legitimate run (a 200 Mbps + // upstream throughput test is ~21k pps of 1472-byte packets), so they only ever catch abuse + // — a rate limit that clips a real measurement produces a confidently wrong number. + RateSessionsPerMin int // ECHOLOT_RATE_SESSIONS_PER_MIN / --rate-sessions-per-min + RateActionsPerMin int // ECHOLOT_RATE_ACTIONS_PER_MIN / --rate-actions-per-min + RateUDPPps int // ECHOLOT_RATE_UDP_PPS / --rate-udp-pps + RateUDPKbps int // ECHOLOT_RATE_UDP_KBPS / --rate-udp-kbps + + // How long canary DNS query logs are kept, in hours (spec §6; privacy default 24). + DNSLogRetentionH int // ECHOLOT_DNS_LOG_RETENTION_H / --dns-log-retention-h + // Client compatibility window. Bounds are SemVer; an empty maximum means unbounded. The // defaults sit at breaking boundaries, so shipping a patch never requires changing them. MinAppVersion string // ECHOLOT_MIN_APP_VERSION / --min-app-version @@ -209,6 +221,11 @@ func Load(args []string) (*Config, *Actions, error) { fs.StringVar(&c.ACMEHTTPListen, "acme-http-listen", envOr("ACME_HTTP_LISTEN", ""), "port-80 listener for ACME HTTP-01 challenges and http->https redirects") fs.StringVar(&c.ACMEWebroot, "acme-webroot", envOr("ACME_WEBROOT", ""), "directory an ACME client writes challenges into (default /acme)") fs.StringVar(&c.PublicControlURL, "public-url", envOr("PUBLIC_URL", ""), "public control-plane URL for enrollment links, e.g. https://probe.example.net:8443") + fs.IntVar(&c.RateSessionsPerMin, "rate-sessions-per-min", envInt("RATE_SESSIONS_PER_MIN", 10), "per-credential and per-IP ceiling on session creation (spec §2.5); 0 disables") + fs.IntVar(&c.RateActionsPerMin, "rate-actions-per-min", envInt("RATE_ACTIONS_PER_MIN", 60), "per-credential and per-IP ceiling on §5 actions; 0 disables") + fs.IntVar(&c.RateUDPPps, "rate-udp-pps", envInt("RATE_UDP_PPS", 25_000), "per-credential and per-IP data-plane packet ceiling, packets/s, silent drop; 0 disables") + fs.IntVar(&c.RateUDPKbps, "rate-udp-kbps", envInt("RATE_UDP_KBPS", 250_000), "per-credential and per-IP data-plane byte ceiling, kbit/s, silent drop; 0 disables") + fs.IntVar(&c.DNSLogRetentionH, "dns-log-retention-h", envInt("DNS_LOG_RETENTION_H", 24), "hours canary DNS query logs are kept (spec §6 privacy default 24); 0 keeps until the ring overwrites") fs.StringVar(&c.MinAppVersion, "min-app-version", envOr("MIN_APP_VERSION", "0.2.0"), "oldest app version this server will serve (SemVer, inclusive)") fs.StringVar(&c.MaxAppVersion, "max-app-version", envOr("MAX_APP_VERSION", "1.0.0"), "first app version this server will refuse (SemVer, exclusive); empty = unbounded") fs.BoolVar(&c.Docker, "docker", envOr("DOCKER", "") == "1", "force container mode (config from env, no systemd/self-update)") diff --git a/server/internal/control/control.go b/server/internal/control/control.go index 76079a5..c0caf93 100644 --- a/server/internal/control/control.go +++ b/server/internal/control/control.go @@ -16,6 +16,7 @@ import ( "encoding/hex" "encoding/json" "errors" + "fmt" "io" "log/slog" "net" @@ -29,6 +30,7 @@ import ( "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" @@ -57,7 +59,8 @@ type Server struct { // 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) + // 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 @@ -97,6 +100,11 @@ type Server struct { // 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 @@ -106,7 +114,12 @@ 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. -const ProtocolVersion = "1.0.0" +// +// 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" @@ -164,10 +177,12 @@ func (s *Server) Handler() http.Handler { gate := s.requireCompatibleApp mux.HandleFunc("POST /v1/enroll", gate(s.enroll)) - mux.HandleFunc("POST /v1/sessions", gate(s.newSession)) + // §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.actions)) + 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)) @@ -177,7 +192,6 @@ func (s *Server) Handler() http.Handler { 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)) - // TODO(spec §5): frag_send, throughput (both build on the same grant machinery) return mux } @@ -199,6 +213,37 @@ func selftestSignal(f func() (bool, bool)) map[string]any { 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)) @@ -235,8 +280,14 @@ func (s *Server) observations(w http.ResponseWriter, r *http.Request) { 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}, - "tcp": tcp, + "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(), @@ -264,6 +315,7 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) { 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"` @@ -326,6 +378,11 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) { 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) @@ -336,13 +393,20 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) { return } go func() { - sent, err := s.DownTrain(sess, g, count, size, interval) + sent, err := s.DownTrain(sess, g, count, size, interval, dscp) slog.Info("downtrain finished", "action", actionID, "sent", sent, "bytes", g.Sent(), "err", err) }() - writeJSON(w, http.StatusAccepted, map[string]any{ + 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 { @@ -525,8 +589,54 @@ func (s *Server) maxDFPayload(sess *session.Session) int { return mtu - overhead } -// dataMinPacket is the smallest datagram that still carries a header + a little payload. -const dataMinPacket = 40 +// 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", diff --git a/server/internal/control/dscp_test.go b/server/internal/control/dscp_test.go new file mode 100644 index 0000000..9f3c9fa --- /dev/null +++ b/server/internal/control/dscp_test.go @@ -0,0 +1,27 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +package control + +import "testing" + +func TestDscpArg(t *testing.T) { + ptr := func(v int) *int { return &v } + for _, tc := range []struct { + in *int + want int + wantErr bool + }{ + {nil, -1, false}, // absent: leave the socket alone + {ptr(0), 0, false}, // explicit best-effort is not the same as absent + {ptr(46), 46, false}, // EF, the value people actually test with + {ptr(63), 63, false}, + {ptr(64), 0, true}, // one past the 6-bit field + {ptr(-1), 0, true}, + } { + got, err := dscpArg(tc.in) + if (err != nil) != tc.wantErr || got != tc.want { + t.Errorf("dscpArg(%v) = %d, err=%v; want %d, wantErr=%v", tc.in, got, err, tc.want, tc.wantErr) + } + } +} diff --git a/server/internal/dataplane/cmsg_linux.go b/server/internal/dataplane/cmsg_linux.go new file mode 100644 index 0000000..031ddc7 --- /dev/null +++ b/server/internal/dataplane/cmsg_linux.go @@ -0,0 +1,70 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +//go:build linux + +package dataplane + +import ( + "encoding/binary" + "net" + "syscall" +) + +// Per-packet TTL and TOS/traffic-class arrive as control messages, and only if asked for at +// socket setup. These fill the spec §3.3 observation-block fields that were shipped as the 0xFF +// sentinel until now — received TTL is path-length evidence, the TOS byte is DSCP/ECN survival. + +// enableRecvMeta asks the kernel to attach the cmsgs to every received datagram. Both the v4 and +// the v6 option sets are attempted on every socket: a dual-stack socket delivers v4-mapped +// traffic through the v6 fd, and the kernel refuses whichever set does not apply. Errors are +// dropped on purpose — a socket that cannot deliver metadata still serves probes, and the +// sentinel already says "not observed" for it. +func enableRecvMeta(conn *net.UDPConn) { + raw, err := conn.SyscallConn() + if err != nil { + return + } + _ = raw.Control(func(fd uintptr) { + _ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_IP, syscall.IP_RECVTTL, 1) + _ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_IP, syscall.IP_RECVTOS, 1) + _ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_IPV6, syscall.IPV6_RECVHOPLIMIT, 1) + _ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_IPV6, syscall.IPV6_RECVTCLASS, 1) + }) +} + +// parseMeta extracts TTL and the TOS byte from one datagram's control messages. Anything absent +// or unparseable keeps the sentinel — reported as unobserved, never guessed. +func parseMeta(oob []byte) pktMeta { + m := pktMeta{TTL: metaUnavailable, TOS: metaUnavailable} + if len(oob) == 0 { + return m + } + cmsgs, err := syscall.ParseSocketControlMessage(oob) + if err != nil { + return m + } + for _, c := range cmsgs { + switch { + case c.Header.Level == syscall.IPPROTO_IP && c.Header.Type == syscall.IP_TTL, + c.Header.Level == syscall.IPPROTO_IPV6 && c.Header.Type == syscall.IPV6_HOPLIMIT: + m.TTL = cmsgValue(c.Data) + case c.Header.Level == syscall.IPPROTO_IP && c.Header.Type == syscall.IP_TOS, + c.Header.Level == syscall.IPPROTO_IPV6 && c.Header.Type == syscall.IPV6_TCLASS: + m.TOS = cmsgValue(c.Data) + } + } + return m +} + +// cmsgValue reads a cmsg the kernel encodes either as a native-endian int (IP_TTL, +// IPV6_HOPLIMIT, IPV6_TCLASS) or as a single byte (IP_TOS). Both fit a byte by definition. +func cmsgValue(data []byte) uint8 { + switch { + case len(data) >= 4: + return uint8(binary.NativeEndian.Uint32(data)) + case len(data) >= 1: + return data[0] + } + return metaUnavailable +} diff --git a/server/internal/dataplane/cmsg_other.go b/server/internal/dataplane/cmsg_other.go new file mode 100644 index 0000000..7b7e74f --- /dev/null +++ b/server/internal/dataplane/cmsg_other.go @@ -0,0 +1,15 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +//go:build !linux + +package dataplane + +import "net" + +// Per-packet TTL/TOS needs Linux's IP_RECVTTL-family cmsgs. Elsewhere the observation block and +// train buffers keep the spec §3.3 sentinel (0xFF = not observed) — absent is honest, a guess +// is not. Deployment targets are Linux; this build exists so the Windows dev loop compiles. +func enableRecvMeta(_ *net.UDPConn) {} + +func parseMeta(_ []byte) pktMeta { return pktMeta{TTL: metaUnavailable, TOS: metaUnavailable} } diff --git a/server/internal/dataplane/frag_linux.go b/server/internal/dataplane/frag_linux.go index 02c45f2..78cf90d 100644 --- a/server/internal/dataplane/frag_linux.go +++ b/server/internal/dataplane/frag_linux.go @@ -104,8 +104,8 @@ func (s *Server) FragSend( return res, fmt.Errorf("session has no recorded local address") } - if sizeBytes < HeaderSize+8 { - sizeBytes = HeaderSize + 8 + if sizeBytes < HeaderSize+32 { + sizeBytes = HeaderSize + 32 } if sizeBytes > 8000 { sizeBytes = 8000 @@ -117,7 +117,10 @@ func (s *Server) FragSend( // The ELT1 packet, signed exactly as any other, then wrapped in UDP. payload := make([]byte, sizeBytes-HeaderSize) binary.BigEndian.PutUint32(payload[0:4], uint32(sizeBytes)) - copy(payload[4:], mode) + // [8:16]: the action id, as on every granted packet (spec §5 correlation); the mode string + // sits past the reserved slot. + putActionID(payload, g.ActionID) + copy(payload[16:], mode) elt := s.buildPacket(sess, TypeFragData, 0, payload) udp := buildUDP(local, target, elt) diff --git a/server/internal/dataplane/granted.go b/server/internal/dataplane/granted.go index 203c05f..331e32c 100644 --- a/server/internal/dataplane/granted.go +++ b/server/internal/dataplane/granted.go @@ -21,7 +21,11 @@ import ( // 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) { +// +// dscp ≥ 0 marks the burst (spec §5 downtrain `dscp`): downstream DSCP survival is the half the +// client cannot produce itself. Best-effort off Linux — see withTOS/TOSSupported; the action +// response has already told the client whether the marking was applied. +func (s *Server) DownTrain(sess *session.Session, g *session.Grant, count, sizeBytes, intervalUs, dscp int) (int, error) { target := sess.DataSource() if !target.IsValid() { return 0, fmt.Errorf("no observed data-plane source") @@ -30,26 +34,41 @@ func (s *Server) DownTrain(sess *session.Session, g *session.Grant, count, sizeB if conn == nil { return 0, fmt.Errorf("no data-plane socket matches target family") } - if sizeBytes < HeaderSize+8 { - sizeBytes = HeaderSize + 8 + if sizeBytes < HeaderSize+16 { + sizeBytes = HeaderSize + 16 } payload := make([]byte, sizeBytes-HeaderSize) + // Payload [8:16] carries the action id on every granted packet, so arriving traffic can be + // attributed to the action that caused it (spec §5/§9: test.params.action_id). + putActionID(payload, g.ActionID) 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) + burst := func() error { + 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 nil } - return sent, nil + if dscp >= 0 && TOSSupported { + // Same shared-socket borrow as the DF window: dfMu keeps a concurrent burst from riding + // along with — or clearing — this marking. + s.dfMu.Lock() + defer s.dfMu.Unlock() + err := withTOS(conn, dscp, burst) // sent must be read after the burst ran, not before + return sent, err + } + err := burst() + return sent, err } // BigSendResult records what happened to one requested size. `Sent` false with an EMSGSIZE-ish @@ -83,8 +102,8 @@ func (s *Server) BigSend(sess *session.Session, g *session.Grant, sizes []int, d results := make([]BigSendResult, 0, len(sizes)) burst := func() error { for i, size := range sizes { - if size < HeaderSize+8 { - size = HeaderSize + 8 + if size < HeaderSize+16 { + size = HeaderSize + 16 } if size > 9000 { // jumbo ceiling; beyond this the kernel will refuse anyway size = 9000 @@ -96,6 +115,8 @@ func (s *Server) BigSend(sess *session.Session, g *session.Grant, sizes []int, d // 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)) + // [8:16]: the action id, as on every granted packet (spec §5 correlation). + putActionID(payload, g.ActionID) err := s.sendErr(conn, target, sess, TypeBigSend, uint32(i), payload) results = append(results, BigSendResult{ SizeBytes: size, Seq: i, Sent: err == nil, Err: errString(err), diff --git a/server/internal/dataplane/throughput.go b/server/internal/dataplane/throughput.go index c1f1f26..e424e3f 100644 --- a/server/internal/dataplane/throughput.go +++ b/server/internal/dataplane/throughput.go @@ -120,7 +120,7 @@ func (s *Server) DownThroughput( // Same plan the grant was sized from, so the two cannot disagree. durationMs, kbps = ThroughputPlan(durationMs, kbps) - if sizeBytes < HeaderSize+16 { + if sizeBytes < HeaderSize+24 { sizeBytes = 1200 // a size that survives every common path unfragmented } if sizeBytes > 1472 { @@ -134,6 +134,10 @@ func (s *Server) DownThroughput( } payload := make([]byte, sizeBytes-HeaderSize) + // [8:16]: the action id, as on every granted packet (spec §5 correlation). The send + // timestamp lives past it at [16:24]; the client reads only header fields today, so + // reserving the slot costs nothing and keeps one layout rule across granted types. + putActionID(payload, g.ActionID) deadline := time.Now().Add(time.Duration(durationMs) * time.Millisecond) start := time.Now() next := start @@ -157,7 +161,7 @@ func (s *Server) DownThroughput( // Reaching here means the run is progressing normally; the clock will end it. res.LimitedBy = "duration" binary.BigEndian.PutUint32(payload[0:4], seq) - binary.BigEndian.PutUint64(payload[4:12], uint64(time.Since(s.start).Nanoseconds())) + binary.BigEndian.PutUint64(payload[16:24], uint64(time.Since(s.start).Nanoseconds())) if err := s.sendErr(conn, target, sess, TypeThroughputData, seq, payload); err != nil { // A send error mid-run is a local condition (buffer full, route gone). Stop and // report what got out rather than pretending the rest was lost on the path. diff --git a/server/internal/dataplane/tos_linux.go b/server/internal/dataplane/tos_linux.go new file mode 100644 index 0000000..e1760a2 --- /dev/null +++ b/server/internal/dataplane/tos_linux.go @@ -0,0 +1,55 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +//go:build linux + +package dataplane + +import ( + "net" + "syscall" +) + +// withTOS runs fn with the socket's TOS/traffic class set to dscp<<2 (ECN bits left zero — the +// test is about DSCP survival, and claiming ECN capability we do not use would pollute it), then +// restores what was there before. +// +// Same borrow discipline as withDF: the socket is shared by every session on that family, so the +// caller must hold Server.dfMu for the whole window or a concurrent burst rides along with — or +// clears — someone else's marking. +func withTOS(conn *net.UDPConn, dscp int, fn func() error) error { + raw, err := conn.SyscallConn() + if err != nil { + return err + } + v4 := conn.LocalAddr().(*net.UDPAddr).IP.To4() != nil + level, opt := syscall.IPPROTO_IPV6, syscall.IPV6_TCLASS + if v4 { + level, opt = syscall.IPPROTO_IP, syscall.IP_TOS + } + + var setErr error + prev := 0 + if err := raw.Control(func(fd uintptr) { + if p, e := syscall.GetsockoptInt(int(fd), level, opt); e == nil { + prev = p + } + setErr = syscall.SetsockoptInt(int(fd), level, opt, dscp<<2) + }); err != nil { + return err + } + if setErr != nil { + return setErr + } + defer func() { + _ = raw.Control(func(fd uintptr) { + _ = syscall.SetsockoptInt(int(fd), level, opt, prev) + }) + }() + return fn() +} + +// TOSSupported reports whether withTOS can actually mark packets here. Exported so the control +// plane can tell the client up front that its dscp request will not be honored, instead of the +// client measuring an unmarked burst and concluding the network stripped the marking. +const TOSSupported = true diff --git a/server/internal/dataplane/tos_other.go b/server/internal/dataplane/tos_other.go new file mode 100644 index 0000000..5c8f70a --- /dev/null +++ b/server/internal/dataplane/tos_other.go @@ -0,0 +1,16 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +//go:build !linux + +package dataplane + +import "net" + +// Setting DSCP per-burst uses IP_TOS/IPV6_TCLASS under the same fd-borrow pattern as withDF, +// which is only exercised on Linux deployments. Elsewhere the burst goes out with the default +// class and TOSSupported lets the action response say so — an unmarked burst reported as marked +// would read as "the network stripped DSCP", the exact wrong conclusion. +func withTOS(_ *net.UDPConn, _ int, fn func() error) error { return fn() } + +const TOSSupported = false diff --git a/server/internal/dataplane/train.go b/server/internal/dataplane/train.go new file mode 100644 index 0000000..147aaae --- /dev/null +++ b/server/internal/dataplane/train.go @@ -0,0 +1,122 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +package dataplane + +// Upstream trains (spec §3.2, types 0x03–0x05). The client blasts TRAIN_DATA at the server and +// the server answers nothing per-packet — a reply would double the traffic and measure the +// return path at the same time. Afterwards the client asks for the server's received view with +// TRAIN_REPORT_REQ, and gets it back columnar, split across as many TRAIN_REPORT datagrams as +// it takes to stay under a safe size. +// +// Both TRAIN_DATA and TRAIN_REPORT_REQ carry the train id in payload[0:4]; the id is the +// client's to choose, unique within the session. + +import ( + "encoding/binary" + "net" + "net/netip" + + "echo-lot.app/server/internal/session" +) + +const ( + // trainReportMaxDatagram caps one TRAIN_REPORT datagram at a size that survives every common + // path unfragmented. A report about loss must not itself be lost to MTU. + trainReportMaxDatagram = 1200 + trainReportHeader = 16 + trainReportRow = 17 // 4 seq + 8 t_rx_ns + 2 size + 1 ttl + 1 dscp + 1 ecn +) + +// recordTrain buffers one TRAIN_DATA packet into its train (session-side, bounded — see +// session/train.go). A payload too short to carry the id is unreportable and stays only in the +// flat packet log, which already recorded it. +func recordTrain(sess *session.Session, payload []byte, seq uint32, size int, tRxNs int64, meta pktMeta) { + if len(payload) < 4 { + return + } + sess.RecordTrainPacket(binary.BigEndian.Uint32(payload[0:4]), session.TrainEntry{ + Seq: seq, TRxNs: tRxNs, Size: uint16(min(size, 0xFFFF)), + TTL: meta.TTL, DSCP: meta.dscp(), ECN: meta.ecn(), + }) +} + +// trainReport answers one TRAIN_REPORT_REQ with the full columnar report. +// +// Grant-free on purpose. §3.4 caps ungranted responses at the request size, and a multi-part +// report is larger than the single REPORT_REQ that asked for it — but it cannot amplify: every +// 17-byte row accounts for one HMAC-valid TRAIN_DATA packet of at least HeaderSize+4 bytes this +// session already delivered here, so the whole report is a strict fraction of the traffic it +// describes, and it only ever goes to the session's verified source address. An unknown id gets +// a single zero-row report rather than silence — "nothing arrived" IS the measurement. +func (s *Server) trainReport(conn *net.UDPConn, raddr netip.AddrPort, sess *session.Session, payload []byte) { + if len(payload) < 4 { + return + } + id := binary.BigEndian.Uint32(payload[0:4]) + train, _ := sess.TrainView(id) + train.ID = id + for i, part := range buildTrainReport(train) { + s.send(conn, raddr, sess, TypeTrainReport, uint32(i), part) + } +} + +// buildTrainReport lays a train's received view out columnar and cuts it into datagram-sized +// payloads. Layout (mirrored by the client; all big-endian): +// +// 0 4 train_id +// 4 4 received (every packet counted, buffered or not — loss math uses this) +// 8 2 part (0-based) +// 10 2 parts +// 12 1 flags (bit0: buffer overflowed; rows beyond the cap were counted, not kept — +// the schema's evidence_truncated honesty, on the wire) +// 13 1 reserved +// 14 2 n (rows in this part) +// 16 n×4 seq, n×8 t_rx_ns, n×2 size, n×1 ttl, n×1 dscp, n×1 ecn (columns contiguous) +func buildTrainReport(t session.Train) [][]byte { + perPart := (trainReportMaxDatagram - HeaderSize - trainReportHeader) / trainReportRow + parts := (len(t.Entries) + perPart - 1) / perPart + if parts == 0 { + parts = 1 // an empty train still gets its "received: 0" answer + } + out := make([][]byte, 0, parts) + for p := 0; p < parts; p++ { + rows := t.Entries[p*perPart : min((p+1)*perPart, len(t.Entries))] + n := len(rows) + b := make([]byte, trainReportHeader+n*trainReportRow) + binary.BigEndian.PutUint32(b[0:4], t.ID) + binary.BigEndian.PutUint32(b[4:8], uint32(t.Received)) + binary.BigEndian.PutUint16(b[8:10], uint16(p)) + binary.BigEndian.PutUint16(b[10:12], uint16(parts)) + if t.Truncated { + b[12] = 1 + } + binary.BigEndian.PutUint16(b[14:16], uint16(n)) + off := trainReportHeader + for i, r := range rows { + binary.BigEndian.PutUint32(b[off+i*4:], r.Seq) + } + off += n * 4 + for i, r := range rows { + binary.BigEndian.PutUint64(b[off+i*8:], uint64(r.TRxNs)) + } + off += n * 8 + for i, r := range rows { + binary.BigEndian.PutUint16(b[off+i*2:], r.Size) + } + off += n * 2 + for i, r := range rows { + b[off+i] = r.TTL + } + off += n + for i, r := range rows { + b[off+i] = r.DSCP + } + off += n + for i, r := range rows { + b[off+i] = r.ECN + } + out = append(out, b) + } + return out +} diff --git a/server/internal/dataplane/train_test.go b/server/internal/dataplane/train_test.go new file mode 100644 index 0000000..fb0489c --- /dev/null +++ b/server/internal/dataplane/train_test.go @@ -0,0 +1,179 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +package dataplane + +import ( + "encoding/binary" + "net" + "net/netip" + "testing" + "time" + + "echo-lot.app/server/internal/session" +) + +// parseReportPart decodes one TRAIN_REPORT payload back into rows, checking the header. +func parseReportPart(t *testing.T, b []byte) (id uint32, received int, part, parts int, truncated bool, rows []session.TrainEntry) { + t.Helper() + if len(b) < trainReportHeader { + t.Fatalf("report part shorter than its header: %d", len(b)) + } + id = binary.BigEndian.Uint32(b[0:4]) + received = int(binary.BigEndian.Uint32(b[4:8])) + part = int(binary.BigEndian.Uint16(b[8:10])) + parts = int(binary.BigEndian.Uint16(b[10:12])) + truncated = b[12]&1 != 0 + n := int(binary.BigEndian.Uint16(b[14:16])) + if want := trainReportHeader + n*trainReportRow; len(b) != want { + t.Fatalf("part length %d, want %d for %d rows", len(b), want, n) + } + off := trainReportHeader + rows = make([]session.TrainEntry, n) + for i := range rows { + rows[i].Seq = binary.BigEndian.Uint32(b[off+i*4:]) + } + off += n * 4 + for i := range rows { + rows[i].TRxNs = int64(binary.BigEndian.Uint64(b[off+i*8:])) + } + off += n * 8 + for i := range rows { + rows[i].Size = binary.BigEndian.Uint16(b[off+i*2:]) + } + off += n * 2 + for i := range rows { + rows[i].TTL = b[off+i] + } + off += n + for i := range rows { + rows[i].DSCP = b[off+i] + } + off += n + for i := range rows { + rows[i].ECN = b[off+i] + } + return +} + +func TestBuildTrainReportSplitsAndRoundTrips(t *testing.T) { + const count = 250 // enough to need several parts + train := session.Train{ID: 42, Received: count, Truncated: true} + for i := 0; i < count; i++ { + train.Entries = append(train.Entries, session.TrainEntry{ + Seq: uint32(i), TRxNs: int64(i) * 1_000_000, Size: uint16(100 + i), + TTL: 64, DSCP: 46, ECN: 1, + }) + } + + parts := buildTrainReport(train) + if len(parts) < 2 { + t.Fatalf("250 rows should not fit one ≤%d-byte datagram", trainReportMaxDatagram) + } + var got []session.TrainEntry + for i, p := range parts { + if HeaderSize+len(p) > trainReportMaxDatagram { + t.Fatalf("part %d would be a %d-byte datagram, cap is %d", i, HeaderSize+len(p), trainReportMaxDatagram) + } + id, received, part, total, truncated, rows := parseReportPart(t, p) + if id != 42 || received != count || part != i || total != len(parts) || !truncated { + t.Fatalf("part %d header: id=%d received=%d part=%d/%d truncated=%v", + i, id, received, part, total, truncated) + } + got = append(got, rows...) + } + if len(got) != count { + t.Fatalf("round-tripped %d rows, want %d", len(got), count) + } + for i, r := range got { + want := train.Entries[i] + if r != want { + t.Fatalf("row %d = %+v, want %+v", i, r, want) + } + } +} + +func TestBuildTrainReportEmptyTrainStillAnswers(t *testing.T) { + parts := buildTrainReport(session.Train{ID: 9}) + if len(parts) != 1 { + t.Fatalf("empty train: %d parts, want 1 — 'nothing arrived' is the answer, not silence", len(parts)) + } + id, received, _, total, _, rows := parseReportPart(t, parts[0]) + if id != 9 || received != 0 || total != 1 || len(rows) != 0 { + t.Fatalf("empty report: id=%d received=%d parts=%d rows=%d", id, received, total, len(rows)) + } +} + +func TestTrainDataThenReportOverTheWire(t *testing.T) { + mgr, addr := startServer(t) + sess, _, err := mgr.New("dev1", "credential-ikm", netip.MustParseAddr("127.0.0.1")) + if err != nil { + t.Fatal(err) + } + client, err := net.DialUDP("udp", nil, net.UDPAddrFromAddrPort(addr)) + if err != nil { + t.Fatal(err) + } + defer client.Close() + client.SetDeadline(time.Now().Add(3 * time.Second)) + + // A short train: id 5 in payload[0:4], plus padding. + const trainID, count = 5, 4 + for i := 0; i < count; i++ { + payload := make([]byte, 60) + binary.BigEndian.PutUint32(payload[0:4], trainID) + if _, err := client.Write(craft(t, sess, TypeTrainData, uint32(i+1), payload)); err != nil { + t.Fatal(err) + } + } + // TRAIN_DATA must be silent (spec §3.2). + client.SetReadDeadline(time.Now().Add(300 * time.Millisecond)) + if _, err := client.Read(make([]byte, 1500)); err == nil { + t.Fatal("TRAIN_DATA got a response, want none") + } + + // Ask for the report. + reqPayload := make([]byte, 4) + binary.BigEndian.PutUint32(reqPayload, trainID) + client.SetDeadline(time.Now().Add(3 * time.Second)) + if _, err := client.Write(craft(t, sess, TypeTrainReportReq, 100, reqPayload)); err != nil { + t.Fatal(err) + } + buf := make([]byte, 2000) + n, err := client.Read(buf) + if err != nil { + t.Fatalf("no TRAIN_REPORT: %v", err) + } + if buf[4] != TypeTrainReport { + t.Fatalf("type = %#x, want TRAIN_REPORT", buf[4]) + } + id, received, part, parts, truncated, rows := parseReportPart(t, buf[HeaderSize:n]) + if id != trainID || received != count || part != 0 || parts != 1 || truncated { + t.Fatalf("report header: id=%d received=%d part=%d/%d truncated=%v", id, received, part, parts, truncated) + } + if len(rows) != count { + t.Fatalf("%d rows, want %d", len(rows), count) + } + for i, r := range rows { + if r.Seq != uint32(i+1) { + t.Fatalf("row %d seq = %d, want %d", i, r.Seq, i+1) + } + if r.Size != HeaderSize+60 { + t.Fatalf("row %d size = %d, want %d", i, r.Size, HeaderSize+60) + } + } + + // Unknown train id: one zero-row report, not silence. + binary.BigEndian.PutUint32(reqPayload, 999) + if _, err := client.Write(craft(t, sess, TypeTrainReportReq, 101, reqPayload)); err != nil { + t.Fatal(err) + } + n, err = client.Read(buf) + if err != nil { + t.Fatalf("no report for unknown train: %v", err) + } + id, received, _, _, _, rows = parseReportPart(t, buf[HeaderSize:n]) + if id != 999 || received != 0 || len(rows) != 0 { + t.Fatalf("unknown-train report: id=%d received=%d rows=%d", id, received, len(rows)) + } +} diff --git a/server/internal/dataplane/udp.go b/server/internal/dataplane/udp.go index e571796..6ff4855 100644 --- a/server/internal/dataplane/udp.go +++ b/server/internal/dataplane/udp.go @@ -2,15 +2,15 @@ // SPDX-License-Identifier: GPL-3.0-or-later // Package dataplane implements the binary UDP probe protocol (spec §3): -// 32-byte header, HMAC gate, anti-replay, ECHO with observation block. -// Skeleton scope: ECHO_REQ/ECHO_RESP and TIMESYNC only; trains, MTU probes -// and delayed echo land with the corresponding client tests. +// 32-byte header, HMAC gate, anti-replay, ECHO with observation block, +// upstream trains with columnar reports, and the granted server->client sends. package dataplane import ( "crypto/hmac" "crypto/sha256" "encoding/binary" + "encoding/hex" "fmt" "log/slog" "net" @@ -18,6 +18,7 @@ import ( "sync" "time" + "echo-lot.app/server/internal/ratelimit" "echo-lot.app/server/internal/session" ) @@ -25,9 +26,14 @@ const ( Magic = "ELT1" HeaderSize = 32 - TypeEchoReq = 0x01 - TypeEchoResp = 0x02 - TypeTimesyncReq = 0x07 + TypeEchoReq = 0x01 + TypeEchoResp = 0x02 + // Upstream trains (spec §3.2): DATA gets no per-packet response; REPORT_REQ fetches the + // server's received view as one or more REPORT datagrams (train.go). + TypeTrainData = 0x03 + TypeTrainReportReq = 0x04 + TypeTrainReport = 0x05 + TypeTimesyncReq = 0x07 TypeTimesyncRsp = 0x08 TypeMtuProbe = 0x09 TypeMtuAck = 0x0A @@ -47,6 +53,12 @@ const ( type Server struct { Sessions *session.Manager + // Spec §2.5 ceilings on verified traffic, silent-drop (nil = no ceiling). Charged after the + // HMAC gate so an unauthenticated flood cannot spend anyone's budget, keyed per source + // address AND per device credential so neither one hot address nor one hot credential can + // crowd out the rest. + PacketRate *ratelimit.Limiter // tokens are packets + ByteRate *ratelimit.Limiter // tokens are bytes // Epoch for server-side t_rx/t_tx: process start; observation consumers // only need differences plus the timesync exchange, not absolute time. start time.Time @@ -59,6 +71,34 @@ type Server struct { dfMu sync.Mutex } +// pktMeta is what the kernel told us about one received datagram beyond its bytes (spec §3.3: +// received TTL, DSCP, ECN). 0xFF means "not observed": non-Linux hosts and datagrams whose +// cmsg never arrived keep the sentinel rather than inventing a value. +type pktMeta struct { + TTL uint8 + TOS uint8 // the whole DSCP/ECN byte; DSCP = TOS>>2, ECN = TOS&3 +} + +const metaUnavailable = 0xFF + +func (m pktMeta) dscp() uint8 { + if m.TOS == metaUnavailable { + return metaUnavailable + } + return m.TOS >> 2 +} + +func (m pktMeta) ecn() uint8 { + if m.TOS == metaUnavailable { + return metaUnavailable + } + return m.TOS & 0x3 +} + +// oobCap fits the two cmsgs (TTL + TOS, each ≤ CMSG_SPACE(4)) with headroom for whatever else +// the kernel decides to attach. +const oobCap = 64 + // Serve runs the read loop for one socket; call once per bound address. // The socket is retained so actions (delayed echo) can pick a family-matching // sender later. @@ -69,14 +109,16 @@ func (s *Server) Serve(conn *net.UDPConn) error { } s.conns = append(s.conns, conn) s.mu.Unlock() + enableRecvMeta(conn) buf := make([]byte, 65535) + oob := make([]byte, oobCap) for { - n, raddr, err := conn.ReadFromUDPAddrPort(buf) + n, oobn, _, raddr, err := conn.ReadMsgUDPAddrPort(buf, oob) if err != nil { return err } tRx := time.Since(s.start).Nanoseconds() - s.handle(conn, raddr, buf[:n], tRx) + s.handle(conn, raddr, buf[:n], tRx, parseMeta(oob[:oobn])) } } @@ -126,7 +168,7 @@ func (s *Server) SendDelayedEcho(sess *session.Session, actionID string) error { // handle enforces spec §3.1/§3.4: unknown prefix, bad HMAC, expired session, // replayed seq → silent drop, never a response. -func (s *Server) handle(conn *net.UDPConn, raddr netip.AddrPort, pkt []byte, tRxNs int64) { +func (s *Server) handle(conn *net.UDPConn, raddr netip.AddrPort, pkt []byte, tRxNs int64, meta pktMeta) { if len(pkt) < HeaderSize || string(pkt[0:4]) != Magic { return } @@ -149,6 +191,12 @@ func (s *Server) handle(conn *net.UDPConn, raddr netip.AddrPort, pkt []byte, tRx if !hmac.Equal(mac.Sum(nil)[:4], pkt[28:32]) { return } + // Spec §2.5: over-ceiling traffic is silently dropped (probes tolerate loss by design). + // After the HMAC gate so a spoofed flood cannot drain a victim's budget; before the replay + // window so a dropped packet's seq stays usable for a resend. + if !s.allowUDP(raddr, sess.Device, len(pkt)) { + return + } if !sess.CheckSeq(seq) { return } @@ -169,9 +217,16 @@ func (s *Server) handle(conn *net.UDPConn, raddr netip.AddrPort, pkt []byte, tRx Src: raddr.String(), Size: len(pkt), Type: typ, }) + payload := pkt[HeaderSize : HeaderSize+int(payloadLen)] switch typ { case TypeEchoReq: - s.echoResp(conn, raddr, sess, pkt, seq, tRxNs) + s.echoResp(conn, raddr, sess, pkt, seq, tRxNs, meta) + case TypeTrainData: + // No response (spec §3.2): the train is upstream-only; its received view is fetched + // afterwards via TRAIN_REPORT_REQ or the observations API. + recordTrain(sess, payload, seq, len(pkt), tRxNs, meta) + case TypeTrainReportReq: + s.trainReport(conn, raddr, sess, payload) case TypeTimesyncReq: s.timesyncResp(conn, raddr, sess, pkt, seq, tRxNs) case TypeMtuProbe: @@ -181,6 +236,16 @@ func (s *Server) handle(conn *net.UDPConn, raddr netip.AddrPort, pkt []byte, tRx } } +// allowUDP charges the §2.5 packet and byte buckets, per source address and per credential. +func (s *Server) allowUDP(raddr netip.AddrPort, device string, size int) bool { + ipKey, credKey := "ip:"+raddr.Addr().String(), "cred:"+device + okA, _ := s.PacketRate.Allow(ipKey) + okC, _ := s.PacketRate.Allow(credKey) + okAB, _ := s.ByteRate.AllowN(ipKey, float64(size)) + okCB, _ := s.ByteRate.AllowN(credKey, float64(size)) + return okA && okC && okAB && okCB +} + // mtuAck replies to an MTU_PROBE with a small MTU_ACK carrying the total // datagram size the server actually received (spec §3.2). The client sends // DF-flagged probes of increasing size and binary-searches the path MTU / a @@ -198,28 +263,43 @@ func (s *Server) mtuAck(conn *net.UDPConn, raddr netip.AddrPort, sess *session.S // 8 8 t_tx_ns // 16 16 observed source IP (v4-mapped when v4) // 32 2 observed source port -// 34 1 received TTL (0xFF = not observed yet; needs recvmsg cmsgs) +// 34 1 received TTL (0xFF = not observed; cmsgs unavailable on this host) // 35 1 received DSCP/ECN byte (0xFF = not observed) // 36 4 received size -func observation(tRxNs, tTxNs int64, src netip.AddrPort, rcvd int) []byte { +func observation(tRxNs, tTxNs int64, src netip.AddrPort, rcvd int, meta pktMeta) []byte { b := make([]byte, 40) binary.BigEndian.PutUint64(b[0:8], uint64(tRxNs)) binary.BigEndian.PutUint64(b[8:16], uint64(tTxNs)) a16 := src.Addr().As16() copy(b[16:32], a16[:]) binary.BigEndian.PutUint16(b[32:34], src.Port()) - b[34], b[35] = 0xFF, 0xFF + b[34], b[35] = meta.TTL, meta.TOS binary.BigEndian.PutUint32(b[36:40], uint32(rcvd)) return b } +// putActionID writes a grant's action id into payload[8:16] — the correlation the spec promises +// (§5: "an action_id echoed in resulting data-plane packets"), consumed by the client as +// test.params.action_id (§9). Bytes [0:8] stay with the packet type; [8:16] is reserved for this +// across every granted type, so the client needs one rule, not five. +func putActionID(payload []byte, actionID string) { + if len(payload) < 16 { + return + } + raw, err := hex.DecodeString(actionID) + if err != nil || len(raw) != 8 { + return // a malformed id yields zero bytes, not a crash mid-burst + } + copy(payload[8:16], raw) +} + // echoResp mirrors the request header (type flipped), appends the observation // block, and re-HMACs with the session key. Anti-amplification: the response // is capped at the request size (spec §3.4) — the observation block replaces // padding rather than growing the datagram; if the request was smaller than // header+observation, the block is truncated to fit. -func (s *Server) echoResp(conn *net.UDPConn, raddr netip.AddrPort, sess *session.Session, req []byte, seq uint32, tRxNs int64) { - obs := observation(tRxNs, time.Since(s.start).Nanoseconds(), raddr, len(req)) +func (s *Server) echoResp(conn *net.UDPConn, raddr netip.AddrPort, sess *session.Session, req []byte, seq uint32, tRxNs int64, meta pktMeta) { + obs := observation(tRxNs, time.Since(s.start).Nanoseconds(), raddr, len(req), meta) max := len(req) if max < HeaderSize { return diff --git a/server/internal/ratelimit/ratelimit.go b/server/internal/ratelimit/ratelimit.go new file mode 100644 index 0000000..155d73e --- /dev/null +++ b/server/internal/ratelimit/ratelimit.go @@ -0,0 +1,100 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +// Package ratelimit implements the spec §2.5 token buckets: per-credential and per-source-IP +// ceilings on session creation, actions, UDP packets and bytes. One Limiter holds one policy +// (rate + burst) and lazily creates a bucket per key; callers namespace their keys ("cred:…", +// "ip:…") so a single Limiter can enforce both axes of the same rule. +package ratelimit + +import ( + "sync" + "time" +) + +// Limiter is a keyed set of token buckets sharing one rate and burst. +// +// A nil *Limiter allows everything: the ceilings are configurable down to "off" (config value 0), +// and a nil check in one place beats a sentinel policy that every call site must know about. +type Limiter struct { + rate float64 // tokens per second + burst float64 + + mu sync.Mutex + buckets map[string]*bucket + lastSweep time.Time + now func() time.Time // swappable so tests need no sleeping +} + +type bucket struct { + tokens float64 + last time.Time +} + +// New creates a limiter granting ratePerSec tokens per second per key, holding at most burst. +func New(ratePerSec, burst float64) *Limiter { + return &Limiter{ + rate: ratePerSec, + burst: burst, + buckets: map[string]*bucket{}, + now: time.Now, + } +} + +// Allow takes one token for key. See AllowN. +func (l *Limiter) Allow(key string) (bool, time.Duration) { + return l.AllowN(key, 1) +} + +// AllowN takes n tokens for key, reporting whether they were available and — when they were +// not — how long until they will be, which is what a control-plane 429 puts in Retry-After. +// A refusal consumes nothing: the caller being told to wait must not itself push the wait out. +func (l *Limiter) AllowN(key string, n float64) (bool, time.Duration) { + if l == nil { + return true, 0 + } + l.mu.Lock() + defer l.mu.Unlock() + now := l.now() + l.sweepLocked(now) + b := l.buckets[key] + if b == nil { + b = &bucket{tokens: l.burst, last: now} + l.buckets[key] = b + } + b.tokens += now.Sub(b.last).Seconds() * l.rate + if b.tokens > l.burst { + b.tokens = l.burst + } + b.last = now + if b.tokens >= n { + b.tokens -= n + return true, 0 + } + return false, time.Duration((n - b.tokens) / l.rate * float64(time.Second)) +} + +// sweepEvery bounds how often the map is walked; the walk is cheap but there is no point doing +// it per packet on the data plane's hot path. +const sweepEvery = time.Minute + +// sweepLocked drops buckets that have been idle long enough to be full again. A full bucket +// carries no state a fresh one would not, and without the sweep the map grows one entry per +// source address ever seen — an attacker-controlled key space must not be an unbounded one. +func (l *Limiter) sweepLocked(now time.Time) { + if now.Sub(l.lastSweep) < sweepEvery { + return + } + l.lastSweep = now + idle := sweepEvery + if l.rate > 0 { + if refill := time.Duration(l.burst / l.rate * float64(time.Second)); refill > idle { + idle = refill + } + } + for k, b := range l.buckets { + if now.Sub(b.last) > idle { + delete(l.buckets, k) + } + } +} diff --git a/server/internal/ratelimit/ratelimit_test.go b/server/internal/ratelimit/ratelimit_test.go new file mode 100644 index 0000000..c829d6e --- /dev/null +++ b/server/internal/ratelimit/ratelimit_test.go @@ -0,0 +1,105 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +package ratelimit + +import ( + "testing" + "time" +) + +// clockAt pins the limiter to a fake clock so refill is a function of arithmetic, not sleeping. +func clockAt(l *Limiter) *time.Time { + t := time.Unix(1000, 0) + l.now = func() time.Time { return t } + return &t +} + +func TestBurstThenRefusalThenRefill(t *testing.T) { + l := New(1, 3) // 1 token/s, burst 3 + now := clockAt(l) + + for i := 0; i < 3; i++ { + if ok, _ := l.Allow("k"); !ok { + t.Fatalf("token %d of the burst refused", i) + } + } + ok, wait := l.Allow("k") + if ok { + t.Fatal("fourth token inside the same instant should be refused") + } + if wait <= 0 || wait > time.Second { + t.Fatalf("retry-after = %v, want (0, 1s]", wait) + } + + *now = now.Add(2 * time.Second) // refills 2 tokens + if ok, _ := l.Allow("k"); !ok { + t.Fatal("refused after refill") + } + if ok, _ := l.Allow("k"); !ok { + t.Fatal("second refilled token refused") + } + if ok, _ := l.Allow("k"); ok { + t.Fatal("third token allowed but only two seconds elapsed") + } +} + +func TestRefusalConsumesNothing(t *testing.T) { + l := New(1, 1) + now := clockAt(l) + l.Allow("k") + // Hammering while empty must not push the refill out. + for i := 0; i < 10; i++ { + if ok, _ := l.Allow("k"); ok { + t.Fatal("allowed while empty") + } + } + *now = now.Add(time.Second) + if ok, _ := l.Allow("k"); !ok { + t.Fatal("the refused attempts ate the refill") + } +} + +func TestKeysAreIndependent(t *testing.T) { + l := New(1, 1) + clockAt(l) + l.Allow("a") + if ok, _ := l.Allow("b"); !ok { + t.Fatal("draining key a refused key b") + } +} + +func TestAllowNChargesBytes(t *testing.T) { + l := New(1000, 1000) // e.g. bytes/s + clockAt(l) + if ok, _ := l.AllowN("k", 900); !ok { + t.Fatal("900 of 1000 refused") + } + if ok, _ := l.AllowN("k", 200); ok { + t.Fatal("1100 of 1000 allowed") + } + if ok, _ := l.AllowN("k", 100); !ok { + t.Fatal("the refused 200 consumed the remaining 100") + } +} + +func TestNilLimiterAllowsEverything(t *testing.T) { + var l *Limiter + if ok, wait := l.AllowN("k", 1e12); !ok || wait != 0 { + t.Fatal("nil limiter must be a no-op") + } +} + +func TestSweepDropsIdleBucketsOnly(t *testing.T) { + l := New(1, 3) + now := clockAt(l) + l.Allow("idle") + *now = now.Add(2 * time.Minute) + l.Allow("busy") // triggers the sweep; "idle" refilled long ago + if _, held := l.buckets["idle"]; held { + t.Fatal("idle bucket survived the sweep") + } + if _, held := l.buckets["busy"]; !held { + t.Fatal("active bucket was swept") + } +} diff --git a/server/internal/session/session.go b/server/internal/session/session.go index b073c86..bf61b61 100644 --- a/server/internal/session/session.go +++ b/server/internal/session/session.go @@ -42,6 +42,9 @@ type Session struct { connectBack []ConnectBackResult throughput []ThroughputReport upstream UpstreamCounter + // Upstream trains (spec §3.2), buffered apart from udpObs — see train.go for why the + // flat ring must not be the only home of a 5000-packet train. + trains []*Train } const obsCap = 4096 diff --git a/server/internal/session/train.go b/server/internal/session/train.go new file mode 100644 index 0000000..79c0248 --- /dev/null +++ b/server/internal/session/train.go @@ -0,0 +1,98 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +package session + +// Upstream trains (spec §3.2): TRAIN_DATA packets get no per-packet response, so the server's +// received view is the only record of what survived the upstream path. It is kept per train — +// NOT in the flat udpObs ring, whose 4096-entry cap would silently roll the head of a 5000-packet +// train out from under the report that is about to be requested. Losing the head quietly turns +// real early-train packets into phantom loss; a bounded buffer that says when it overflowed +// (Truncated, mirroring the measurement schema's evidence_truncated) keeps the numbers honest. + +// TrainEntry is the server's received view of one train packet. TTL/DSCP/ECN use 0xFF for +// "not observed" (spec §3.3), same sentinel as the echo observation block. +type TrainEntry struct { + Seq uint32 + TRxNs int64 // server clock, session epoch + Size uint16 + TTL uint8 + DSCP uint8 + ECN uint8 +} + +// Train is the received view of one upstream train, keyed by the id the client put in the +// TRAIN_DATA payload. +type Train struct { + ID uint32 + // Received counts every packet of the train, including any the entry buffer no longer holds; + // the loss figure must come from this, not from len(Entries). + Received int + Truncated bool + Entries []TrainEntry +} + +const ( + // trainCap comfortably holds the largest train the client-side action bounds allow (5000 + // packets, matching the downtrain clamp). Overflow keeps the head and sets Truncated: the + // early packets are the ones a ring would drop, and the tail's absence is at least declared. + trainCap = 8192 + // maxTrains bounds one session's train memory (~8×8k×24 B ≈ 1.5 MiB worst case). The oldest + // train is evicted for a new one because reports are requested train-by-train, right after + // each train — an id still being sent to is always the one worth keeping. + maxTrains = 8 +) + +// RecordTrainPacket appends one received TRAIN_DATA packet to its train's buffer. +func (s *Session) RecordTrainPacket(trainID uint32, e TrainEntry) { + s.mu.Lock() + defer s.mu.Unlock() + var t *Train + for _, c := range s.trains { + if c.ID == trainID { + t = c + break + } + } + if t == nil { + if len(s.trains) >= maxTrains { + s.trains = s.trains[1:] + } + t = &Train{ID: trainID} + s.trains = append(s.trains, t) + } + t.Received++ + if len(t.Entries) >= trainCap { + t.Truncated = true + return + } + t.Entries = append(t.Entries, e) +} + +// TrainView returns a copy of one train. A missing id reports ok=false; the caller decides +// whether "never saw it" is an error or (for a report request) the answer itself. +func (s *Session) TrainView(trainID uint32) (Train, bool) { + s.mu.Lock() + defer s.mu.Unlock() + for _, t := range s.trains { + if t.ID == trainID { + cp := *t + cp.Entries = append([]TrainEntry(nil), t.Entries...) + return cp, true + } + } + return Train{}, false +} + +// Trains returns copies of every train witnessed in this session, oldest first. +func (s *Session) Trains() []Train { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]Train, 0, len(s.trains)) + for _, t := range s.trains { + cp := *t + cp.Entries = append([]TrainEntry(nil), t.Entries...) + out = append(out, cp) + } + return out +} diff --git a/server/internal/session/train_test.go b/server/internal/session/train_test.go new file mode 100644 index 0000000..24f73f8 --- /dev/null +++ b/server/internal/session/train_test.go @@ -0,0 +1,62 @@ +// SPDX-FileCopyrightText: 2026 Echolot contributors +// SPDX-License-Identifier: GPL-3.0-or-later + +package session + +import "testing" + +func TestTrainBufferKeepsHeadAndDeclaresTruncation(t *testing.T) { + s := &Session{} + over := trainCap + 100 + for i := 0; i < over; i++ { + s.RecordTrainPacket(7, TrainEntry{Seq: uint32(i), Size: 64}) + } + tr, ok := s.TrainView(7) + if !ok { + t.Fatal("train not found") + } + if tr.Received != over { + t.Fatalf("Received = %d, want %d — the count must include unbuffered packets", tr.Received, over) + } + if len(tr.Entries) != trainCap { + t.Fatalf("buffered %d entries, want the cap %d", len(tr.Entries), trainCap) + } + if !tr.Truncated { + t.Fatal("overflow must be declared, not silent") + } + // The head must survive: it is what a ring buffer would have lost. + if tr.Entries[0].Seq != 0 || tr.Entries[trainCap-1].Seq != trainCap-1 { + t.Fatalf("buffer kept seqs %d..%d, want the head 0..%d", + tr.Entries[0].Seq, tr.Entries[trainCap-1].Seq, trainCap-1) + } +} + +func TestTrainEvictionDropsOldestTrain(t *testing.T) { + s := &Session{} + for id := uint32(0); id < maxTrains+2; id++ { + s.RecordTrainPacket(id, TrainEntry{Seq: 1}) + } + if _, ok := s.TrainView(0); ok { + t.Fatal("oldest train should have been evicted") + } + if _, ok := s.TrainView(1); ok { + t.Fatal("second-oldest train should have been evicted") + } + if _, ok := s.TrainView(maxTrains + 1); !ok { + t.Fatal("newest train missing") + } + if got := len(s.Trains()); got != maxTrains { + t.Fatalf("holding %d trains, want %d", got, maxTrains) + } +} + +func TestTrainViewReturnsACopy(t *testing.T) { + s := &Session{} + s.RecordTrainPacket(3, TrainEntry{Seq: 10}) + tr, _ := s.TrainView(3) + tr.Entries[0].Seq = 99 + again, _ := s.TrainView(3) + if again.Entries[0].Seq != 10 { + t.Fatal("TrainView leaked the internal slice") + } +}