signal: chat and share fan out to every other admitted peer
signal.go dispatches hub.TypeChat/TypeShare to hub.Fanout after
validateFanout checks the two untargeted shapes (chat is exactly
{text: string} 1..2048 bytes; share is exactly {on: bool}) and rejects
a chat/share carrying "to"; hub.ErrNotFanout joins ErrNotTargeted in
the defensive 400 arm. chat_test.go's tests and the brief's sketch
disagree with the real helper shapes in two places, both adapted:
client.signal takes a JSON body string, not a map (a signalJSON test
helper marshals the map bodies the brief sketched); and
guestJoinedAndWaiting (ice_test.go) returns (*client, peer) without
the guest's own stream, which TestChatNeverReachesAWaitingGuestAndAWaitingGuestCannotSend
needs to assert on, so that helper's three steps are inlined in the
test instead of calling it.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2
3 files changed,
+220
−6
internal/meet/signal.go+60 −5internal/meettest/chat_test.go+143 −0internal/meettest/signal_test.go+17 −1
CI passed — run details
diff --git a/internal/meet/signal.go b/internal/meet/signal.go| index 973b1be..cae25ef 100644 |
| --- a/internal/meet/signal.go |
| +++ b/internal/meet/signal.go |
| @@ -1,6 +1,7 @@ |
| package meet |
| import ( |
| + "bytes" |
| "encoding/json" |
| "errors" |
| "io" |
| @@ -16,6 +17,52 @@ import ( |
| // maxSignalBody is spec §4.4's 64 KB cap, refused before decoding. |
| const maxSignalBody = 64 * 1024 |
| +// maxChatBytes is §4.4's 2 KB, counted in bytes of the text, not |
| +// characters: the limit protects the other peers' streams, which carry |
| +// bytes. |
| +const maxChatBytes = 2048 |
| + |
| +// validateFanout checks the two untargeted payload shapes without |
| +// keeping anything from them: chat is exactly {"text": string} with |
| +// 1..2048 bytes; share is exactly {"on": bool}. Returns a status and |
| +// an outcome word for the refusal log, or 0. |
| +func validateFanout(typ string, raw json.RawMessage) (int, string) { |
| + dec := json.NewDecoder(bytes.NewReader(raw)) |
| + dec.DisallowUnknownFields() |
| + switch typ { |
| + case hub.TypeChat: |
| + var p struct { |
| + Text *string `json:"text"` |
| + } |
| + if err := dec.Decode(&p); err != nil || p.Text == nil || dec.More() { |
| + return http.StatusBadRequest, "shape" |
| + } |
| + if len(*p.Text) == 0 { |
| + return http.StatusBadRequest, "shape" |
| + } |
| + if len(*p.Text) > maxChatBytes { |
| + return http.StatusRequestEntityTooLarge, "size" |
| + } |
| + case hub.TypeShare: |
| + var p struct { |
| + On *bool `json:"on"` |
| + } |
| + if err := dec.Decode(&p); err != nil || p.On == nil || dec.More() { |
| + return http.StatusBadRequest, "shape" |
| + } |
| + } |
| + return 0, "" |
| +} |
| + |
| +// fanoutError names the JSON error the two ways validateFanout can |
| +// refuse a fan-out message. |
| +func fanoutError(status int) string { |
| + if status == http.StatusRequestEntityTooLarge { |
| + return "too-large" |
| + } |
| + return "bad-request" |
| +} |
| + |
| // peerIDPattern is the exact shape a server-issued peer id takes: |
| // sixteen random bytes in hex (hub's randomID). Anything else is |
| // refused before it is ever used as a hub lookup key or a log argument |
| @@ -163,9 +210,15 @@ func (a *app) signal(w http.ResponseWriter, r *http.Request) { |
| case hub.TypeReset: |
| herr = a.hub.Reset(r.Context(), 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 |
| + if body.To != "" { |
| + a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "shape", map[string]any{"error": "bad-request"}) |
| + return |
| + } |
| + if status, outcome := validateFanout(body.Type, body.Payload); status != 0 { |
| + a.refuseSignal(w, sqid, body.Peer, body.Type, status, outcome, map[string]any{"error": fanoutError(status)}) |
| + return |
| + } |
| + herr = a.hub.Fanout(r.Context(), sqid, session, body.Peer, hub.Message{Type: body.Type, Payload: body.Payload}) |
| default: |
| a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "type", map[string]any{"error": "bad-request"}) |
| return |
| @@ -182,12 +235,14 @@ func (a *app) signal(w http.ResponseWriter, r *http.Request) { |
| 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): |
| + case errors.Is(herr, hub.ErrNotTargeted), errors.Is(herr, hub.ErrNotFanout): |
| // Defence, not a reachable path: ErrNotTargeted only ever comes |
| // from Relay's own isTargeted guard, and this dispatch only ever |
| // calls Relay for offer/answer/candidate — already targeted |
| // types — so this arm cannot fire today. Kept in case that |
| - // invariant ever drifts (review minor 10). |
| + // invariant ever drifts (review minor 10). ErrNotFanout is the |
| + // same defence for Fanout, which this dispatch only ever calls |
| + // for chat/share — already fan-out types. |
| 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): |
diff --git a/internal/meettest/chat_test.go b/internal/meettest/chat_test.go| new file mode 100644 |
| index 0000000..fe9ba87 |
| --- /dev/null |
| +++ b/internal/meettest/chat_test.go |
| @@ -0,0 +1,143 @@ |
| +package meettest |
| + |
| +import ( |
| + "encoding/json" |
| + "net/http" |
| + "strings" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// signalJSON marshals body into the JSON string client.signal expects. |
| +// body must not carry "peer": client.signal always injects that itself |
| +// from the peer argument, the way every other test in this package |
| +// calls it with a literal string. This helper only exists because a |
| +// handful of chat/share cases below need dynamic values (a repeated |
| +// string, a peer id inlined as "to") that a literal template would make |
| +// unreadable. |
| +func signalJSON(t *testing.T, body map[string]any) string { |
| + t.Helper() |
| + b, err := json.Marshal(body) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return string(b) |
| +} |
| + |
| +// Spec §4.3's fan-out: chat reaches every other admitted peer, never |
| +// the sender, and the envelope names the sender's peer id and display |
| +// name — never anything the sender's own request claimed. |
| +func TestChatReachesTheOtherAdmittedPeerWithTheSendersNameAndNotTheSender(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"}`) |
| + sa.nextOf("roster", 2*time.Second) |
| + |
| + 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"}`) |
| + sb.nextOf("roster", 2*time.Second) |
| + sa.nextOf("peer", 2*time.Second) |
| + |
| + res, _ := a.signal(sqid, pa, signalJSON(t, map[string]any{"type": "chat", "payload": map[string]any{"text": "hello there"}})) |
| + if res.StatusCode != http.StatusOK { |
| + t.Fatalf("chat: %d", res.StatusCode) |
| + } |
| + ev, data, ok := sb.nextOf("chat", 2*time.Second) |
| + if !ok || ev != "chat" { |
| + t.Fatal("Bob did not receive the chat") |
| + } |
| + s := string(data) |
| + if !strings.Contains(s, `"from":"`+pa+`"`) || !strings.Contains(s, `"name":"paul"`) || !strings.Contains(s, `"text":"hello there"`) || strings.Contains(s, `"seq"`) || strings.Contains(s, `"to"`) { |
| + t.Fatalf("envelope %s", s) |
| + } |
| + if ev, _, ok := sa.next(300 * time.Millisecond); ok && ev == "chat" { |
| + t.Fatal("the sender received its own chat") |
| + } |
| +} |
| + |
| +// Spec §2.5 and §4.3: a waiting guest is not admitted, so fan-out never |
| +// reaches it, and its own signal is the usual "a meeting that is not |
| +// yours does not exist" 404 — never a 403, whatever it tries to send. |
| +// |
| +// ice_test.go's guestJoinedAndWaiting builds the identical waiting |
| +// guest but returns only (*client, peer): its own stream is registered |
| +// for t.Cleanup and never handed back. This test needs to read from |
| +// that stream directly (to prove the guest receives nothing), so the |
| +// same three steps — join link, sign in, open stream, join — are |
| +// inlined here instead of reusing that helper. |
| +func TestChatNeverReachesAWaitingGuestAndAWaitingGuestCannotSend(t *testing.T) { |
| + x, sqid, tok := 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"}`) |
| + sa.nextOf("roster", 2*time.Second) |
| + |
| + g := x.browser(t) |
| + g.get("/j/" + tok) |
| + g.signIn("waiting-guest@example.com") |
| + gs := g.stream("/m/" + sqid + "/stream") |
| + defer gs.close() |
| + gpeer, _ := gs.hello() |
| + if resp, b := g.signal(sqid, gpeer, `{"type":"join"}`); resp.StatusCode != http.StatusOK { |
| + t.Fatalf("guest join (the knock): %d %s", resp.StatusCode, b) |
| + } |
| + sa.nextOf("knock", 2*time.Second) |
| + |
| + a.signal(sqid, pa, signalJSON(t, map[string]any{"type": "chat", "payload": map[string]any{"text": "secret"}})) |
| + if ev, data, ok := gs.next(500 * time.Millisecond); ok && (ev == "chat" || strings.Contains(string(data), "secret")) { |
| + t.Fatalf("a waiting guest received %s %s", ev, data) |
| + } |
| + res, body := g.signal(sqid, gpeer, signalJSON(t, map[string]any{"type": "chat", "payload": map[string]any{"text": "let me in"}})) |
| + if res.StatusCode != http.StatusNotFound || !strings.Contains(string(body), `"error":"not-found"`) { |
| + t.Fatalf("a waiting guest's chat must be the usual 404: %d %s", res.StatusCode, body) |
| + } |
| +} |
| + |
| +// Spec §4.4: chat text is capped at 2048 bytes, counted in bytes, not |
| +// characters; anything not exactly {text: string} (chat) or {on: bool} |
| +// (share), including a chat carrying "to", is the usual 400. |
| +func TestChatLimitsAndShapes(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"}`) |
| + sa.nextOf("roster", 2*time.Second) |
| + |
| + cases := []struct { |
| + name string |
| + body map[string]any |
| + status int |
| + }{ |
| + {"2048 bytes ok", map[string]any{"type": "chat", "payload": map[string]any{"text": strings.Repeat("a", 2048)}}, http.StatusOK}, |
| + {"2049 bytes too large", map[string]any{"type": "chat", "payload": map[string]any{"text": strings.Repeat("a", 2049)}}, http.StatusRequestEntityTooLarge}, |
| + {"multibyte counted in bytes", map[string]any{"type": "chat", "payload": map[string]any{"text": strings.Repeat("é", 1025)}}, http.StatusRequestEntityTooLarge}, |
| + {"text not a string", map[string]any{"type": "chat", "payload": map[string]any{"text": 5}}, http.StatusBadRequest}, |
| + {"extra key", map[string]any{"type": "chat", "payload": map[string]any{"text": "x", "html": "<b>"}}, http.StatusBadRequest}, |
| + {"empty text", map[string]any{"type": "chat", "payload": map[string]any{"text": ""}}, http.StatusBadRequest}, |
| + {"share on", map[string]any{"type": "share", "payload": map[string]any{"on": true}}, http.StatusOK}, |
| + {"share off", map[string]any{"type": "share", "payload": map[string]any{"on": false}}, http.StatusOK}, |
| + {"share not a bool", map[string]any{"type": "share", "payload": map[string]any{"on": "yes"}}, http.StatusBadRequest}, |
| + {"chat with a target", map[string]any{"type": "chat", "to": pa, "payload": map[string]any{"text": "x"}}, http.StatusBadRequest}, |
| + } |
| + for _, c := range cases { |
| + t.Run(c.name, func(t *testing.T) { |
| + res, _ := a.signal(sqid, pa, signalJSON(t, c.body)) |
| + if res.StatusCode != c.status { |
| + t.Fatalf("%s: %d, want %d", c.name, res.StatusCode, c.status) |
| + } |
| + }) |
| + } |
| +} |
diff --git a/internal/meettest/signal_test.go b/internal/meettest/signal_test.go| index ce7cd86..85a9987 100644 |
| --- a/internal/meettest/signal_test.go |
| +++ b/internal/meettest/signal_test.go |
| @@ -609,6 +609,21 @@ func TestSignallingLogsCarryNoContent(t *testing.T) { |
| c.signal(sqid, peer, `{"type":"offer","to":"`+gpeer+`","seq":3,"gen":9,"payload":{"sdp":"STALESECRET"}}`) // outcome=stale-gen |
| c.signal(sqid, peer, `{"type":"bogus"}`) // outcome=type |
| + // An accepted chat appends nothing to the log at all — no line, let |
| + // alone one carrying its text — and an oversize one is refused with |
| + // the usual outcome-only line, never the text that overflowed it. |
| + beforeChat := buf.Len() |
| + if resp, _ := c.signal(sqid, peer, `{"type":"chat","payload":{"text":"xyzzy-plugh-chat"}}`); resp.StatusCode != http.StatusOK { |
| + t.Fatalf("accepted chat: %d", resp.StatusCode) |
| + } |
| + if buf.Len() != beforeChat { |
| + t.Fatalf("an accepted chat appended %d bytes to the log: %q", buf.Len()-beforeChat, buf.String()[beforeChat:]) |
| + } |
| + oversizeChat := `{"type":"chat","payload":{"text":"` + strings.Repeat("a", 2049) + `"}}` |
| + if resp, _ := c.signal(sqid, peer, oversizeChat); resp.StatusCode != http.StatusRequestEntityTooLarge { |
| + t.Fatalf("oversize chat: %d", resp.StatusCode) |
| + } // outcome=size |
| + |
| x.advance(time.Minute) |
| x.ctl.TickForTest() |
| @@ -638,7 +653,7 @@ func TestSignallingLogsCarryNoContent(t *testing.T) { |
| }) |
| logs := buf.String() |
| - for _, want := range []string{"outcome=stale-gen", "outcome=type", "outcome=internal"} { |
| + for _, want := range []string{"outcome=stale-gen", "outcome=type", "outcome=size", "outcome=internal"} { |
| if !strings.Contains(logs, want) { |
| t.Errorf("log missing %q:\n%s", want, logs) |
| } |
| @@ -647,6 +662,7 @@ func TestSignallingLogsCarryNoContent(t *testing.T) { |
| forbidden := []string{ |
| "Secret title", "SDPSECRET", "203.0.113.9", "STALESECRET", tok, |
| "paul@example.com", "zaphod-xyzzy@example.com", "zaphod-xyzzy", |
| + "xyzzy-plugh-chat", |
| } |
| for _, ck := range c.cookies { |
| forbidden = append(forbidden, ck.Value) |