amadan

oficina / meet Public

signal: POST with body and rate limits, dispatch, refusal logging without content

Adapts task-5-brief.md to the branch as it stands and folds in the
Task 4 review addendum, both noted in full in task-5-report.md.
Highlights: sessionKey now returns (string, bool) (Task 4's session
key change), so signal.go follows stream.go's pattern rather than the
brief's single-return snippet; the hub's own findLocked already binds
every dispatch to (room, session, peer), so signal.go does not
pre-check with hub.Find as the brief did (review ruling 2); every
hub error maps through errors.Is per review ruling 1 (404 for
ErrNotYours/ErrUnknownPeer/ErrNotAdmitted/ErrNotMember, never 403;
409 for ErrStaleGen and ErrNotWaiting; 2xx for ErrMeshFull and
ErrTooManyWaiting); every refusal answers JSON via streamJSON rather
than a bare status, with ErrNotYours and ErrUnknownPeer sharing one
body; client.signal is rewritten to marshal a map rather than splice
JSON strings, and returns (*http.Response, []byte); memberWithMeeting
now adds a second allowlisted member (paul@rslw.com) so the new
newMemberBrowser helper has someone to sign in as.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2
Paul Campbell pushed by paul@keymail.dev f2f5f3ffb11962b54f95122b59c47d2d9ff80916 parent 391c800
7 files changed, +550 −9
  • internal/meet/app.go +5 −0
  • internal/meet/ratelimit.go +74 −0
  • internal/meet/signal.go +148 −0
  • internal/meettest/harness_test.go +20 −1
  • internal/meettest/join_test.go +4 −1
  • internal/meettest/signal_test.go +273 −0
  • internal/meettest/sse_test.go +26 −7

CI passed — run details

diff --git a/internal/meet/app.go b/internal/meet/app.go
index 11088d6..d8129f3 100644
--- a/internal/meet/app.go
+++ b/internal/meet/app.go
@@ -36,6 +36,9 @@ type app struct {
// hubSalt keys the session cookie into the hub's session id, so the
// hub holds no session material it did not have to.
hubSalt string
+ // signalLimiter is POST /m/{sqid}/signal's per-peer token bucket
+ // (spec §4.4): thirty a second, burst sixty.
+ signalLimiter *rateLimiter
// The stream's two timings, held as atomics because a test moves
// them (SetStreamTimings) while stream handlers are reading them.
@@ -241,6 +244,7 @@ func build(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *app, err
a.hub = hub.New(hub.Config{
Now: st.now, Logger: rc.Logger, StandingFor: a.hubStanding,
})
+ a.signalLimiter = newRateLimiter(30, 60)
writer, err := d.G.DB()
if err != nil {
@@ -288,6 +292,7 @@ func build(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *app, err
r.Post("/meetings/{sqid}/end", a.endMeeting)
})
r.Get("/m/{sqid}/stream", a.stream)
+ r.Post("/m/{sqid}/signal", a.signal)
})
mux := http.NewServeMux()
diff --git a/internal/meet/ratelimit.go b/internal/meet/ratelimit.go
new file mode 100644
index 0000000..b6113b4
--- /dev/null
+++ b/internal/meet/ratelimit.go
@@ -0,0 +1,74 @@
+package meet
+
+import (
+ "sync"
+ "time"
+)
+
+// rateLimiter is a per-key token bucket (spec §4.4): rate tokens are
+// added to a key's bucket per second, up to burst held, and one message
+// spends one token. It is keyed by peer id for the signalling endpoint,
+// so one noisy peer never spends another's budget.
+//
+// Entries are process-local, like the hub itself, and are evicted once
+// idle for a minute so a room that closed does not hold its peers'
+// buckets forever. The sweep runs inline, every thousandth call, rather
+// than on a timer: nothing here needs a goroutine, and the call rate
+// this guards is exactly what would drive the sweep anyway.
+type rateLimiter struct {
+ rate float64 // tokens added per second
+ burst float64 // bucket capacity
+
+ mu sync.Mutex
+ buckets map[string]*tokenBucket
+ calls uint64
+}
+
+type tokenBucket struct {
+ tokens float64
+ seenAt time.Time
+}
+
+func newRateLimiter(rate, burst int) *rateLimiter {
+ return &rateLimiter{rate: float64(rate), burst: float64(burst), buckets: map[string]*tokenBucket{}}
+}
+
+// allow spends one token for key, refilling by the time elapsed since
+// its last visit. ok is false when the bucket is empty, in which case
+// retryAfter is how long until one token is available.
+func (l *rateLimiter) allow(key string, now time.Time) (ok bool, retryAfter time.Duration) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+
+ l.calls++
+ if l.calls%1000 == 0 {
+ l.sweepLocked(now)
+ }
+
+ b, found := l.buckets[key]
+ if !found {
+ b = &tokenBucket{tokens: l.burst, seenAt: now}
+ l.buckets[key] = b
+ } else if elapsed := now.Sub(b.seenAt).Seconds(); elapsed > 0 {
+ b.tokens = min(l.burst, b.tokens+elapsed*l.rate)
+ b.seenAt = now
+ }
+
+ if b.tokens < 1 {
+ need := (1 - b.tokens) / l.rate
+ return false, time.Duration(need * float64(time.Second))
+ }
+ b.tokens--
+ return true, 0
+}
+
+// sweepLocked drops any bucket idle for a full minute: it will simply be
+// recreated at burst on its next visit, which is correct — a key that
+// has been quiet for a minute owes nothing.
+func (l *rateLimiter) sweepLocked(now time.Time) {
+ for k, b := range l.buckets {
+ if now.Sub(b.seenAt) > time.Minute {
+ delete(l.buckets, k)
+ }
+ }
+}
diff --git a/internal/meet/signal.go b/internal/meet/signal.go
new file mode 100644
index 0000000..120f658
--- /dev/null
+++ b/internal/meet/signal.go
@@ -0,0 +1,148 @@
+package meet
+
+import (
+ "encoding/json"
+ "errors"
+ "io"
+ "net/http"
+ "strconv"
+
+ "github.com/go-chi/chi/v5"
+
+ "amadan.net/oficina/meet/internal/meet/hub"
+)
+
+// maxSignalBody is spec §4.4's 64 KB cap, refused before decoding.
+const maxSignalBody = 64 * 1024
+
+// signalBody is the one shape every upstream message arrives in.
+// Payload is a json.RawMessage, which encoding/json only ever populates
+// with a syntactically valid JSON value — decoding into this struct is
+// what validates the whole body, including payload, before any of it is
+// trusted; nothing here is built from unvalidated bytes by hand.
+type signalBody struct {
+ Peer hub.PeerID `json:"peer"`
+ Type string `json:"type"`
+ To hub.PeerID `json:"to,omitempty"`
+ Seq uint64 `json:"seq,omitempty"`
+ Gen uint64 `json:"gen,omitempty"`
+ Payload json.RawMessage `json:"payload,omitempty"`
+}
+
+// signal is POST /m/{sqid}/signal: the upstream half of spec §4.1. The
+// hub is the only authority on whether the named peer belongs to this
+// room and session — every dispatch below hands it session and peer and
+// lets its own lookup enforce the binding, rather than this handler
+// pre-checking a peer id that came from the request body. `from` and
+// `name` are never taken from the body; the hub rebuilds the envelope
+// from the participation it finds.
+func (a *app) signal(w http.ResponseWriter, r *http.Request) {
+ sqid := chi.URLParam(r, "sqid")
+ session, ok := a.sessionKey(r)
+ if !ok {
+ a.logger.Error("meet: signal with no session subject", "path", r.URL.Path)
+ w.WriteHeader(http.StatusInternalServerError)
+ return
+ }
+
+ // The size cap is enforced by refusing to read past it, not by
+ // checking Content-Length: a refused body is never read past the
+ // limit (spec §4.4).
+ r.Body = http.MaxBytesReader(w, r.Body, maxSignalBody)
+ raw, err := io.ReadAll(r.Body)
+ if err != nil {
+ var tooBig *http.MaxBytesError
+ if errors.As(err, &tooBig) {
+ a.refuseSignal(w, sqid, "", "", http.StatusRequestEntityTooLarge, "too-large", map[string]any{"error": "too-large"})
+ return
+ }
+ a.refuseSignal(w, sqid, "", "", http.StatusBadRequest, "read", map[string]any{"error": "bad-request"})
+ return
+ }
+
+ var body signalBody
+ if err := json.Unmarshal(raw, &body); err != nil || body.Peer == "" || body.Type == "" {
+ a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "shape", map[string]any{"error": "bad-request"})
+ return
+ }
+
+ // Rate limiting is keyed by peer id and happens before any lookup or
+ // dispatch, so a peer that has burned its budget is refused without
+ // the hub doing any work on its behalf (spec §4.4).
+ if allowed, retry := a.signalLimiter.allow(string(body.Peer), a.store.now()); !allowed {
+ w.Header().Set("Retry-After", strconv.Itoa(int(retry.Seconds())+1))
+ a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusTooManyRequests, "rate", map[string]any{"error": "rate"})
+ return
+ }
+
+ var herr error
+ switch body.Type {
+ case hub.TypeJoin:
+ herr = a.hub.Join(r.Context(), sqid, session, body.Peer)
+ case hub.TypeAdmit:
+ var p struct {
+ Peer hub.PeerID `json:"peer"`
+ }
+ if err := json.Unmarshal(body.Payload, &p); err != nil || p.Peer == "" {
+ a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "shape", map[string]any{"error": "bad-request"})
+ return
+ }
+ herr = a.hub.Admit(r.Context(), sqid, session, body.Peer, p.Peer)
+ case hub.TypeLeave:
+ herr = a.hub.Leave(sqid, session, body.Peer)
+ case hub.TypeOffer, hub.TypeAnswer, hub.TypeCandidate:
+ herr = a.hub.Relay(sqid, session, body.Peer, hub.Message{
+ Type: body.Type, To: body.To, Seq: body.Seq, Gen: body.Gen, Payload: body.Payload,
+ })
+ case hub.TypeReset:
+ herr = a.hub.Reset(sqid, session, body.Peer, body.To)
+ case hub.TypeChat, hub.TypeShare:
+ // Slice 4.
+ a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "type", map[string]any{"error": "bad-request"})
+ return
+ default:
+ a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "type", map[string]any{"error": "bad-request"})
+ return
+ }
+
+ switch {
+ case herr == nil, errors.Is(herr, hub.ErrMeshFull), errors.Is(herr, hub.ErrTooManyWaiting):
+ // A full mesh or a full waiting queue is not a failed request:
+ // the peer has already been sent `full` and may retry with
+ // `join`, so the answer is 2xx and the event carries the news
+ // (review ruling 1).
+ w.WriteHeader(http.StatusOK)
+ case errors.Is(herr, hub.ErrStaleGen):
+ a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusConflict, "stale-gen", map[string]any{"error": "stale"})
+ case errors.Is(herr, hub.ErrNotWaiting):
+ a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusConflict, "not-waiting", map[string]any{"error": "conflict"})
+ case errors.Is(herr, hub.ErrNotTargeted):
+ a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "type", map[string]any{"error": "bad-request"})
+ case errors.Is(herr, hub.ErrNotYours), errors.Is(herr, hub.ErrUnknownPeer),
+ errors.Is(herr, hub.ErrNotAdmitted), errors.Is(herr, hub.ErrNotMember):
+ // A meeting, or a peer, that is not yours does not exist (§2.5).
+ // ErrNotYours and ErrUnknownPeer share this one body on purpose:
+ // the review ruling requires the two to be indistinguishable.
+ a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusNotFound, "binding", map[string]any{"error": "not-found"})
+ default:
+ a.logger.Error("meet: signal: dispatch", "room", sqid, "type", body.Type, "err", herr)
+ a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusInternalServerError, "internal", nil)
+ }
+}
+
+// refuseSignal is every refusal this endpoint makes: it logs the type,
+// room, peer and outcome and nothing else (spec §4.4, §0.1), then
+// answers JSON, following stream.go's streamJSON precedent rather than
+// ErrorPage — this endpoint's client is a script, not a page.
+func (a *app) refuseSignal(w http.ResponseWriter, sqid string, peer hub.PeerID, typ string, status int, outcome string, body map[string]any) {
+ a.logRefusal(sqid, peer, typ, outcome)
+ streamJSON(w, status, body)
+}
+
+// logRefusal is the only signalling log line, and it carries no content:
+// no payload, no chat text, no SDP, no candidate, no display name, no
+// title and no link token (spec §4.4, §0.1). An accepted message logs
+// nothing at all.
+func (a *app) logRefusal(sqid string, peer hub.PeerID, typ, outcome string) {
+ a.logger.Warn("meet: signal refused", "room", sqid, "peer", string(peer), "type", typ, "outcome", outcome)
+}
diff --git a/internal/meettest/harness_test.go b/internal/meettest/harness_test.go
index 2ce22b6..020cbf6 100644
--- a/internal/meettest/harness_test.go
+++ b/internal/meettest/harness_test.go
@@ -136,7 +136,15 @@ func (m *recordingMailer) count() int {
func newHarness(t *testing.T, members ...string) *harness {
t.Helper()
- logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+ return newHarnessWithLogger(t, slog.New(slog.NewTextHandler(io.Discard, nil)), members...)
+}
+
+// newHarnessWithLogger is newHarness with the caller's own logger wired
+// through both the app and its database, instead of the usual
+// io.Discard one — used to capture and assert on log lines (spec §0.1,
+// §4.4: no payload, address, title or token is ever a log argument).
+func newHarnessWithLogger(t *testing.T, logger *slog.Logger, members ...string) *harness {
+ t.Helper()
dbPath := filepath.Join(t.TempDir(), "app.db")
d, err := db.Open(dbPath, logger)
if err != nil {
@@ -195,6 +203,17 @@ func (x *harness) browser(t *testing.T) *client {
return &client{t: t, x: x, cookies: map[string]*http.Cookie{}}
}
+// newMemberBrowser is a fresh browser signed in as address, which must
+// already be one of x's AllowedAddresses — a second member sharing the
+// harness memberWithMeeting built, for tests that need two members in
+// one room.
+func newMemberBrowser(t *testing.T, x *harness, address string) *client {
+ t.Helper()
+ c := x.browser(t)
+ c.signIn(address)
+ return c
+}
+
func (c *client) do(req *http.Request) *httptest.ResponseRecorder {
for _, ck := range c.cookies {
req.AddCookie(ck)
diff --git a/internal/meettest/join_test.go b/internal/meettest/join_test.go
index e38d3f1..b752440 100644
--- a/internal/meettest/join_test.go
+++ b/internal/meettest/join_test.go
@@ -9,9 +9,12 @@ import (
"time"
)
+// memberWithMeeting's harness carries a second member, paul@rslw.com,
+// unused by most callers, for the signal tests that need two members in
+// one room (newMemberBrowser).
func memberWithMeeting(t *testing.T) (*harness, string, string) {
t.Helper()
- x := newHarness(t, "paul@example.com")
+ x := newHarness(t, "paul@example.com", "paul@rslw.com")
c := x.browser(t)
c.signIn("paul@example.com")
sqid, tok := createMeeting(t, c, "Board meeting", "")
diff --git a/internal/meettest/signal_test.go b/internal/meettest/signal_test.go
new file mode 100644
index 0000000..c225192
--- /dev/null
+++ b/internal/meettest/signal_test.go
@@ -0,0 +1,273 @@
+package meettest
+
+import (
+ "bytes"
+ "encoding/json"
+ "log/slog"
+ "net/http"
+ "net/url"
+ "strings"
+ "testing"
+ "time"
+)
+
+// Spec §4.2: every POST /signal must carry a peer the hub finds in this
+// room, bound to this session; anything else is refused with 404 —
+// never 403, and the same 404 whether the peer never existed or exists
+// through another session (§2.5's "a meeting that is not yours does not
+// exist").
+func TestSignalNeedsAPeerBoundToThisSessionAndRoom(t *testing.T) {
+ x, sqid, tok := memberWithMeeting(t)
+ c := x.browser(t)
+ c.signIn("paul@example.com")
+ s := c.stream("/m/" + sqid + "/stream")
+ defer s.close()
+ peer, _ := s.hello()
+
+ resp, unknownBody := c.signal(sqid, "0123456789abcdef0123456789abcdef", `{"type":"join"}`)
+ if resp.StatusCode != http.StatusNotFound {
+ t.Fatalf("foreign peer: %d", resp.StatusCode)
+ }
+
+ other := x.browser(t)
+ other.get("/j/" + tok)
+ other.signIn("guest@example.com")
+ resp, otherSessionBody := other.signal(sqid, peer, `{"type":"join"}`)
+ if resp.StatusCode != http.StatusNotFound {
+ t.Fatalf("peer through another session: %d", resp.StatusCode)
+ }
+ if !bytes.Equal(unknownBody, otherSessionBody) {
+ t.Fatalf("404 bodies differ: unknown %q, other session %q", unknownBody, otherSessionBody)
+ }
+
+ resp, _ = c.signal(sqid, peer, `{"type":"join"}`)
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("own join: %d", resp.StatusCode)
+ }
+}
+
+// A peer that is perfectly real, and bound to this very session, but in
+// a different room, is the same 404: lookups start from the room named
+// in the URL (spec §4.2's "cross-room delivery is impossible by
+// construction").
+func TestSignalRefusesAPeerFromAnotherMeeting(t *testing.T) {
+ x, sqid, _ := memberWithMeeting(t)
+ c := x.browser(t)
+ c.signIn("paul@example.com")
+ s := c.stream("/m/" + sqid + "/stream")
+ defer s.close()
+ peer, _ := s.hello()
+
+ elsewhere, _ := createMeeting(t, c, "Different room", "")
+ if resp, _ := c.signal(elsewhere, peer, `{"type":"join"}`); resp.StatusCode != http.StatusNotFound {
+ t.Fatalf("peer from another meeting: %d, want 404", resp.StatusCode)
+ }
+}
+
+func TestSignalLimits(t *testing.T) {
+ x, sqid, _ := memberWithMeeting(t)
+ c := x.browser(t)
+ c.signIn("paul@example.com")
+ s := c.stream("/m/" + sqid + "/stream")
+ defer s.close()
+ peer, _ := s.hello()
+ c.signal(sqid, peer, `{"type":"join"}`)
+
+ big := `{"type":"candidate","to":"` + peer + `","seq":1,"gen":1,"payload":{"x":"` + strings.Repeat("a", 70*1024) + `"}}`
+ if resp, _ := c.signal(sqid, peer, big); resp.StatusCode != http.StatusRequestEntityTooLarge {
+ t.Fatalf("64 KB: %d", resp.StatusCode)
+ }
+
+ var got429 bool
+ for i := 0; i < 100; i++ {
+ resp, _ := c.signal(sqid, peer, `{"type":"leave"}`)
+ if resp.StatusCode == http.StatusTooManyRequests {
+ if resp.Header.Get("Retry-After") == "" {
+ t.Fatal("429 without Retry-After")
+ }
+ got429 = true
+ break
+ }
+ }
+ if !got429 {
+ t.Fatal("100 messages in a burst never hit the rate limit")
+ }
+}
+
+func TestSignalStaleGenAnswers409WithTheCurrentGen(t *testing.T) {
+ x, sqid, _ := memberWithMeeting(t)
+ a := x.browser(t)
+ a.signIn("paul@example.com")
+ sa := a.stream("/m/" + sqid + "/stream")
+ defer sa.close()
+ pa, _ := sa.hello()
+ a.signal(sqid, pa, `{"type":"join"}`)
+
+ b := newMemberBrowser(t, x, "paul@rslw.com")
+ sb := b.stream("/m/" + sqid + "/stream")
+ defer sb.close()
+ pb, _ := sb.hello()
+ b.signal(sqid, pb, `{"type":"join"}`)
+
+ resp, _ := a.signal(sqid, pa, `{"type":"offer","to":"`+pb+`","seq":1,"gen":7,"payload":{}}`)
+ if resp.StatusCode != http.StatusConflict {
+ t.Fatalf("stale gen: %d", resp.StatusCode)
+ }
+}
+
+// Spec §4.3: a guest's join marks it waiting and tells admitted members
+// with knock; the guest itself learns nothing at all from a fresh join
+// (ruling 9, and hub's own TestMemberJoinsAndGuestWaits, which asserts
+// zero events on a waiting guest's sub) — it is the *resume* of a
+// waiting or unjoined participation that answers state{waiting:true}
+// (stream.go, Task 4), not the join itself. This is the guest half of
+// standingFor's hub.KindGuest mapping proven end-to-end over HTTP: a
+// guest waits, a member in the same room sees knock.
+func TestSignalGuestJoinWaitsAndKnocksTheRoom(t *testing.T) {
+ x, sqid, tok := memberWithMeeting(t)
+ member := x.browser(t)
+ member.signIn("paul@example.com")
+ ms := member.stream("/m/" + sqid + "/stream")
+ defer ms.close()
+ mpeer, _ := ms.hello()
+ if resp, _ := member.signal(sqid, mpeer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
+ t.Fatalf("member join: %d", resp.StatusCode)
+ }
+ if ev, _, ok := ms.next(2 * time.Second); !ok || ev != "roster" {
+ t.Fatalf("member after its own join: %s ok=%v, want roster", ev, ok)
+ }
+
+ guest := x.browser(t)
+ guest.get("/j/" + tok)
+ guest.signIn("guest@example.com")
+ gs := guest.stream("/m/" + sqid + "/stream")
+ defer gs.close()
+ gpeer, _ := gs.hello()
+
+ if resp, _ := guest.signal(sqid, gpeer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
+ t.Fatalf("guest join: %d", resp.StatusCode)
+ }
+ if ev, _, ok := ms.next(2 * time.Second); !ok || ev != "knock" {
+ t.Fatalf("member after the guest's join: %s ok=%v, want knock", ev, ok)
+ }
+ if ev, _, ok := gs.next(200 * time.Millisecond); ok {
+ t.Fatalf("waiting guest received %q, want nothing", ev)
+ }
+
+ // The guest really is only waiting, not admitted: resuming its
+ // stream answers state{waiting:true}, not a roster.
+ gs.close()
+ gs2 := guest.stream("/m/" + sqid + "/stream?peer=" + gpeer)
+ defer gs2.close()
+ if p, _ := gs2.hello(); p != gpeer {
+ t.Fatalf("resume peer %q, want %q", p, gpeer)
+ }
+ ev, data, ok := gs2.next(2 * time.Second)
+ if !ok || ev != "state" {
+ t.Fatalf("waiting guest's resume: %s ok=%v, want state", ev, ok)
+ }
+ var state struct{ Waiting bool }
+ json.Unmarshal(data, &state)
+ if !state.Waiting {
+ t.Fatalf("state %s, want waiting", data)
+ }
+}
+
+// Spec §4.5 and §4.6: two admitted peers, one drops its stream and
+// resumes with ?peer= and last, and gets hello, roster, then the
+// replay — in that order, and only the messages above last.
+func TestSignalAdmittedResumeReplaysAfterRoster(t *testing.T) {
+ x, sqid, _ := memberWithMeeting(t)
+ a := x.browser(t)
+ a.signIn("paul@example.com")
+ sa := a.stream("/m/" + sqid + "/stream")
+ defer sa.close()
+ pa, _ := sa.hello()
+ if resp, _ := a.signal(sqid, pa, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
+ t.Fatalf("a's join: %d", resp.StatusCode)
+ }
+ if ev, _, ok := sa.next(2 * time.Second); !ok || ev != "roster" {
+ t.Fatalf("a after its own join: %s ok=%v, want roster", ev, ok)
+ }
+
+ b := newMemberBrowser(t, x, "paul@rslw.com")
+ sb := b.stream("/m/" + sqid + "/stream")
+ pb, _ := sb.hello()
+ if resp, _ := b.signal(sqid, pb, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
+ t.Fatalf("b's join: %d", resp.StatusCode)
+ }
+ if ev, _, ok := sb.next(2 * time.Second); !ok || ev != "roster" {
+ t.Fatalf("b after its own join: %s ok=%v, want roster", ev, ok)
+ }
+ if ev, _, ok := sa.next(2 * time.Second); !ok || ev != "peer" {
+ t.Fatalf("a on b's admission: %s ok=%v, want peer", ev, ok)
+ }
+
+ // One offer delivered live, so it is out of the replay's range.
+ if resp, _ := a.signal(sqid, pa, `{"type":"offer","to":"`+pb+`","seq":1,"gen":1,"payload":{}}`); resp.StatusCode != http.StatusOK {
+ t.Fatalf("a's first offer: %d", resp.StatusCode)
+ }
+ if ev, _, ok := sb.next(2 * time.Second); !ok || ev != "offer" {
+ t.Fatalf("b receiving the first offer live: %s ok=%v", ev, ok)
+ }
+
+ sb.close()
+
+ // A second offer, sent while b is detached, lands only in the ring.
+ if resp, _ := a.signal(sqid, pa, `{"type":"offer","to":"`+pb+`","seq":2,"gen":1,"payload":{}}`); resp.StatusCode != http.StatusOK {
+ t.Fatalf("a's second offer: %d", resp.StatusCode)
+ }
+
+ last, err := json.Marshal(map[string]uint64{pa: 1})
+ if err != nil {
+ t.Fatal(err)
+ }
+ sb2 := b.stream("/m/" + sqid + "/stream?peer=" + pb + "&last=" + url.QueryEscape(string(last)))
+ defer sb2.close()
+ if p, _ := sb2.hello(); p != pb {
+ t.Fatalf("resume hello peer %q, want %q", p, pb)
+ }
+ if ev, _, ok := sb2.next(2 * time.Second); !ok || ev != "roster" {
+ t.Fatalf("first after resume hello: %s ok=%v, want roster", ev, ok)
+ }
+ ev, data, ok := sb2.next(2 * time.Second)
+ if !ok || ev != "offer" {
+ t.Fatalf("second after resume: %s ok=%v, want the replayed offer", ev, ok)
+ }
+ var replay struct{ Seq uint64 }
+ json.Unmarshal(data, &replay)
+ if replay.Seq != 2 {
+ t.Fatalf("replayed offer seq %d, want 2 (the one sent while b was detached)", replay.Seq)
+ }
+}
+
+// The §0.1 and §4.4 promise: no payload, name, title or token in any log
+// line, on accepted and refused messages and on an error path.
+func TestSignallingLogsCarryNoContent(t *testing.T) {
+ var buf bytes.Buffer
+ x := newHarnessWithLogger(t, slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})), "paul@example.com", "paul@rslw.com")
+ c := x.browser(t)
+ c.signIn("paul@example.com")
+ sqid, tok := createMeeting(t, c, "Secret title", "")
+ s := c.stream("/m/" + sqid + "/stream")
+ defer s.close()
+ peer, _ := s.hello()
+ c.signal(sqid, peer, `{"type":"join"}`)
+ b := newMemberBrowser(t, x, "paul@rslw.com")
+ sb := b.stream("/m/" + sqid + "/stream")
+ defer sb.close()
+ pb, _ := sb.hello()
+ b.signal(sqid, pb, `{"type":"join"}`)
+ c.signal(sqid, peer, `{"type":"offer","to":"`+pb+`","seq":1,"gen":1,"payload":{"sdp":"v=0 SDPSECRET a=candidate 203.0.113.9"}}`)
+ c.signal(sqid, peer, `{"type":"candidate","to":"`+pb+`","seq":2,"gen":1,"payload":{"candidate":"candidate:1 1 udp 2 203.0.113.9 5000 typ host"}}`)
+ c.signal(sqid, peer, `{"type":"offer","to":"`+pb+`","seq":3,"gen":9,"payload":{"sdp":"STALESECRET"}}`)
+ c.signal(sqid, peer, `{"type":"bogus"}`)
+ x.advance(time.Minute)
+ x.ctl.TickForTest()
+ logs := buf.String()
+ for _, needle := range []string{"Secret title", "SDPSECRET", "203.0.113.9", "STALESECRET", tok, "paul@example.com"} {
+ if strings.Contains(logs, needle) {
+ t.Errorf("log carries %q", needle)
+ }
+ }
+}
diff --git a/internal/meettest/sse_test.go b/internal/meettest/sse_test.go
index 4ecdef9..b9188ee 100644
--- a/internal/meettest/sse_test.go
+++ b/internal/meettest/sse_test.go
@@ -2,6 +2,7 @@ package meettest
import (
"bufio"
+ "bytes"
"context"
"encoding/json"
"io"
@@ -157,12 +158,27 @@ func (s *sseReader) close() {
}
// signal is POST /m/{sqid}/signal, the upstream half of §4.1: a JSON
-// body carrying the peer and the message, under CSRF. The endpoint
-// itself arrives with the signal handler; this is the client for it.
-func (c *client) signal(sqid, peer, body string) *http.Response {
+// body carrying the peer and the message, under CSRF. body is decoded
+// into a map and given "peer", rather than spliced in as a string — a
+// caller's body of "{}" or "" is not a special case this way, and every
+// call sends one well-formed JSON document. The endpoint itself arrives
+// with the signal handler; this is the client for it.
+func (c *client) signal(sqid, peer, body string) (*http.Response, []byte) {
c.t.Helper()
- payload := `{"peer":"` + peer + `",` + strings.TrimPrefix(body, "{")
- req, err := http.NewRequest(http.MethodPost, c.x.server().URL+"/m/"+sqid+"/signal", strings.NewReader(payload))
+ msg := map[string]any{}
+ if strings.TrimSpace(body) != "" {
+ if err := json.Unmarshal([]byte(body), &msg); err != nil {
+ c.t.Fatalf("signal body %q: %v", body, err)
+ }
+ }
+ msg["peer"] = peer
+ payload, err := json.Marshal(msg)
+ if err != nil {
+ c.t.Fatal(err)
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+ defer cancel()
+ req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.x.server().URL+"/m/"+sqid+"/signal", bytes.NewReader(payload))
if err != nil {
c.t.Fatal(err)
}
@@ -175,7 +191,10 @@ func (c *client) signal(sqid, peer, body string) *http.Response {
if err != nil {
c.t.Fatalf("POST signal: %v", err)
}
- io.Copy(io.Discard, resp.Body)
+ b, err := io.ReadAll(resp.Body)
resp.Body.Close()
- return resp
+ if err != nil {
+ c.t.Fatalf("POST signal: reading body: %v", err)
+ }
+ return resp, b
}