amadan

oficina / meet Public

signal: fix rate-limit keying, peer/type validation, admit ordering

Fix round 1 on Task 5 review.

- Important 1: the signal rate limiter is now keyed on session+room
  (session+"|"+sqid), never on the body's claimed peer id, and the
  check runs before the body is read at all, so an oversized flood
  still spends a token. rateLimiter gains a bounded cap with
  oldest-first eviction (newRateLimiter now takes rate, burst, cap).
- Important 2: peer must be exactly 32 lowercase hex or the request is
  400, checked before it is ever used as a log argument or a hub
  lookup key; an invalid peer is never logged. The logged "type" is
  now one of the wire's own constants or the literal "unknown",
  never arbitrary attacker text.
- Important 3: hub.Admit now settles the actor's own authority (found,
  standing applied, admitted, member) completely before it ever looks
  at the target, so a participation with no admitting authority can no
  longer distinguish an absent target (formerly 409) from a present
  one (formerly 404) by the shape of the refusal, and the target's
  StandingFor is never spent on an actor with no business asking.
- The captured-log test now uses a real guest (a distinctive address
  whose derived name is its own needle), an admit round, and a forced
  internal-error arm (dropping the meetings table, not closing the
  whole database — closing it also breaks rastrillo's own per-request
  session lookup before the request ever reaches the signal handler),
  and positively asserts the stale-gen/type/internal outcome lines
  are present as well as absent of content.
- Minors: 404 bodies now carry boot, identical for ErrNotYours and
  ErrUnknownPeer; TestSignalStaleGenAnswers409WithTheCurrentGen
  renamed to TestSignalStaleGenIsRefusedWith409; client.signalRaw
  added for the body shapes client.signal cannot express, driving a
  new table test (chat/share/bogus/malformed-JSON/missing-peer/
  malformed-admit-payload/well-formed-absent-admit-target); comments
  added on the Retry-After rounding and the unreachable ErrNotTargeted
  arm; unit tests added for ratelimit.go itself (burst, refill,
  retryAfter bounds, idle eviction, the cap).

Every new or changed test was run against the pre-fix code (reverted
signal.go/ratelimit.go/app.go for Important 1/2, admission.go for
Important 3) and confirmed to fail there before being run against the
fix; see task-5-report.md for specifics, including one test
(TestSignalRateLimitIsPerSessionNotPerClaimedPeer) that had to be
rewritten mid-round because its first version passed vacuously against
the old code too.
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 b66bca1f6cedec26b2b585371740b06d2b65b038 parent b03d50a
8 files changed, +559 −72
  • internal/meet/app.go +1 −1
  • internal/meet/hub/admission.go +46 −21
  • internal/meet/hub/admission_test.go +39 −0
  • internal/meet/ratelimit.go +47 −12
  • internal/meet/ratelimit_test.go +109 −0
  • internal/meet/signal.go +77 −21
  • internal/meettest/signal_test.go +212 −17
  • internal/meettest/sse_test.go +28 −0

CI passed — run details

diff --git a/internal/meet/app.go b/internal/meet/app.go
index 75bc5ea..70f3f40 100644
--- a/internal/meet/app.go
+++ b/internal/meet/app.go
@@ -270,7 +270,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)
+ a.signalLimiter = newRateLimiter(30, 60, maxSignalBuckets)
writer, err := d.G.DB()
if err != nil {
diff --git a/internal/meet/hub/admission.go b/internal/meet/hub/admission.go
index ee870fa..d2192cc 100644
--- a/internal/meet/hub/admission.go
+++ b/internal/meet/hub/admission.go
@@ -209,53 +209,78 @@ func (h *Hub) standing(ctx context.Context, sqid, session string, peer PeerID) (
// Admit: the actor must be admitted with current member kind; the target
// waiting with entry authority now; the cap under the lock (spec §4.3).
+//
+// The actor's own authority is settled completely — found, current
+// standing applied, admitted, member — before the target is looked at
+// at all. The other order (found first in an earlier revision) let any
+// joined participation, admitted or not, learn from the shape of the
+// refusal alone whether a candidate peer id existed in the room, and
+// spent a StandingFor call on the target even when the actor had no
+// business asking (review Important 3).
func (h *Hub) Admit(ctx context.Context, sqid, session string, actor, target PeerID) error {
actorKind, actorOK, err := h.standing(ctx, sqid, session, actor)
if err != nil {
return err
}
+
h.mu.Lock()
- var targetPerson int64
- found := false
- if r, ok := h.rooms[sqid]; ok {
- var tp *Participation
- if tp, found = r.peers[target]; found {
- targetPerson = tp.PersonID
+ a, ok := h.findLocked(sqid, session, actor)
+ if !ok {
+ h.mu.Unlock()
+ return ErrUnknownPeer
+ }
+ r := h.rooms[sqid]
+ if !actorOK {
+ lost := h.standingLostLocked(r, a)
+ h.mu.Unlock()
+ if lost {
+ return ErrNotYours
}
+ // An admitted guest whose grant lapsed keeps the room but never
+ // gains the authority to let anyone else in.
+ return ErrNotMember
}
- h.mu.Unlock()
- if !found {
+ h.setKindLocked(r, a, actorKind)
+ if !a.Admitted {
+ h.mu.Unlock()
+ return ErrNotAdmitted
+ }
+ if a.Kind != KindMember {
+ h.mu.Unlock()
+ return ErrNotMember
+ }
+ // Only now, with the actor's authority settled, is the target looked
+ // at — first for its bare existence in this room, still under this
+ // same lock, before it costs a StandingFor call at all.
+ tp, found := r.peers[target]
+ if !found || !tp.Joined || tp.Admitted {
+ h.mu.Unlock()
return ErrNotWaiting
}
+ targetPerson := tp.PersonID
+ h.mu.Unlock()
+
targetKind, targetOK, err := h.cfg.StandingFor(ctx, sqid, targetPerson)
if err != nil {
return err
}
+
h.mu.Lock()
defer h.mu.Unlock()
// Both participations are re-found: either may have ended while the
- // two standing reads were in flight.
- a, ok := h.findLocked(sqid, session, actor)
+ // target's standing read was in flight.
+ a, ok = h.findLocked(sqid, session, actor)
if !ok {
return ErrUnknownPeer
}
- r := h.rooms[sqid]
- if !actorOK {
- if h.standingLostLocked(r, a) {
- return ErrNotYours
- }
- // An admitted guest whose grant lapsed keeps the room but never
- // gains the authority to let anyone else in.
- return ErrNotMember
- }
- h.setKindLocked(r, a, actorKind)
+ r = h.rooms[sqid]
if !a.Admitted {
return ErrNotAdmitted
}
if a.Kind != KindMember {
return ErrNotMember
}
- tp, ok := r.peers[target]
+ tp, ok = r.peers[target]
if !ok || !tp.Joined || tp.Admitted {
return ErrNotWaiting
}
diff --git a/internal/meet/hub/admission_test.go b/internal/meet/hub/admission_test.go
index 1cc0b11..7c769af 100644
--- a/internal/meet/hub/admission_test.go
+++ b/internal/meet/hub/admission_test.go
@@ -96,6 +96,45 @@ func TestOnlyAnAdmittedMemberMayAdmit(t *testing.T) {
}
}
+// Task 5 review Important 3: Admit must settle the actor's own
+// authority completely before it ever looks at the target, so a
+// participation with no admitting authority cannot use the shape of the
+// refusal to learn whether a candidate peer id exists in the room, and
+// the target's StandingFor is never spent on an actor that had no
+// business asking.
+func TestAdmitByAWaitingGuestNeverLooksAtTheTargetOrItsStanding(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ var calls int
+ h := New(Config{Now: func() time.Time { return now }, StandingFor: func(ctx context.Context, room string, id int64) (Kind, bool, error) {
+ calls++
+ if id < 100 {
+ return KindMember, true, nil
+ }
+ return KindGuest, true, nil
+ }})
+ ctx := context.Background()
+ g, _ := h.Open("r1", "sg", 100, "Gus", KindGuest)
+ h.Join(ctx, "r1", "sg", g.Peer) // waiting, never admitted
+
+ calls = 0
+ errAbsent := h.Admit(ctx, "r1", "sg", g.Peer, "0123456789abcdef0123456789abcdef")
+ callsForAbsent := calls
+
+ m, _ := h.Open("r1", "sm", 1, "Mia", KindMember)
+ h.Join(ctx, "r1", "sm", m.Peer) // present, and admitted immediately as a member
+
+ calls = 0
+ errPresent := h.Admit(ctx, "r1", "sg", g.Peer, m.Peer)
+ callsForPresent := calls
+
+ if errAbsent != ErrNotAdmitted || errPresent != ErrNotAdmitted {
+ t.Fatalf("waiting guest's admit: absent target %v, present target %v, want ErrNotAdmitted both times", errAbsent, errPresent)
+ }
+ if callsForAbsent != 1 || callsForPresent != 1 {
+ t.Fatalf("StandingFor calls: absent target %d, present target %d, want 1 both times (the actor's own — the target is never reached)", callsForAbsent, callsForPresent)
+ }
+}
+
func TestCapAtJoinAndAtAdmit(t *testing.T) {
now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
h := testHub(&now)
diff --git a/internal/meet/ratelimit.go b/internal/meet/ratelimit.go
index b6113b4..58a5ffa 100644
--- a/internal/meet/ratelimit.go
+++ b/internal/meet/ratelimit.go
@@ -5,19 +5,34 @@ import (
"time"
)
+// maxSignalBuckets bounds the signalling limiter's map (review Important
+// 1): a generous multiple of what any real deployment holds at once
+// (sixteen participations to a room), so legitimate traffic never
+// evicts anything, while an attacker opening session after session
+// cannot grow the map without bound.
+const maxSignalBuckets = 4096
+
// 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.
+// spends one token. POST /m/{sqid}/signal keys it on session and room
+// (session+"|"+sqid), not on anything the request body claims: the
+// session half comes from the session cookie the request actually
+// presented, so nobody can spend another session's budget or dodge
+// their own by rotating a claimed peer id (review Important 1).
//
-// 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.
+// Entries are process-local, like the hub itself, and bounded by cap:
+// once the map holds cap keys, the least recently touched bucket is
+// evicted to make room, so an attacker opening unbounded sessions
+// cannot grow the map without bound. A bucket is also dropped once idle
+// for a minute; either way it is simply recreated at burst on its next
+// visit, which is correct — a key that has been quiet owes nothing. The
+// idle 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
+ cap int // 0 means unbounded — used only by tests
mu sync.Mutex
buckets map[string]*tokenBucket
@@ -29,8 +44,8 @@ type tokenBucket struct {
seenAt time.Time
}
-func newRateLimiter(rate, burst int) *rateLimiter {
- return &rateLimiter{rate: float64(rate), burst: float64(burst), buckets: map[string]*tokenBucket{}}
+func newRateLimiter(rate, burst, cap int) *rateLimiter {
+ return &rateLimiter{rate: float64(rate), burst: float64(burst), cap: cap, buckets: map[string]*tokenBucket{}}
}
// allow spends one token for key, refilling by the time elapsed since
@@ -47,6 +62,9 @@ func (l *rateLimiter) allow(key string, now time.Time) (ok bool, retryAfter time
b, found := l.buckets[key]
if !found {
+ if l.cap > 0 && len(l.buckets) >= l.cap {
+ l.evictOldestLocked()
+ }
b = &tokenBucket{tokens: l.burst, seenAt: now}
l.buckets[key] = b
} else if elapsed := now.Sub(b.seenAt).Seconds(); elapsed > 0 {
@@ -62,9 +80,7 @@ func (l *rateLimiter) allow(key string, now time.Time) (ok bool, retryAfter time
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.
+// sweepLocked drops any bucket idle for a full minute.
func (l *rateLimiter) sweepLocked(now time.Time) {
for k, b := range l.buckets {
if now.Sub(b.seenAt) > time.Minute {
@@ -72,3 +88,22 @@ func (l *rateLimiter) sweepLocked(now time.Time) {
}
}
}
+
+// evictOldestLocked drops the bucket least recently touched, so a
+// bounded limiter never grows past cap: a burst of distinct keys
+// displaces the quietest one rather than growing the map without bound
+// (review Important 1). O(n) in the bucket count, which only matters
+// right at the cap — everyday traffic never reaches it.
+func (l *rateLimiter) evictOldestLocked() {
+ var oldestKey string
+ var oldestAt time.Time
+ first := true
+ for k, b := range l.buckets {
+ if first || b.seenAt.Before(oldestAt) {
+ oldestKey, oldestAt, first = k, b.seenAt, false
+ }
+ }
+ if !first {
+ delete(l.buckets, oldestKey)
+ }
+}
diff --git a/internal/meet/ratelimit_test.go b/internal/meet/ratelimit_test.go
new file mode 100644
index 0000000..1d02c6f
--- /dev/null
+++ b/internal/meet/ratelimit_test.go
@@ -0,0 +1,109 @@
+package meet
+
+import (
+ "testing"
+ "time"
+)
+
+// Task 5 review minor 5: unit coverage for ratelimit.go's own behaviour,
+// independent of the signal endpoint that happens to be its only caller.
+
+func TestRateLimiterAllowsUpToBurstThenRefuses(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ l := newRateLimiter(30, 3, 0)
+ for i := 0; i < 3; i++ {
+ if ok, _ := l.allow("k", now); !ok {
+ t.Fatalf("call %d within burst was refused", i)
+ }
+ }
+ ok, retry := l.allow("k", now)
+ if ok {
+ t.Fatal("a call past the burst, with no time elapsed, was allowed")
+ }
+ if retry <= 0 {
+ t.Fatalf("retryAfter %v, want > 0", retry)
+ }
+}
+
+func TestRateLimiterRefillsAfterAnExplicitElapsedDuration(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ l := newRateLimiter(30, 1, 0)
+ if ok, _ := l.allow("k", now); !ok {
+ t.Fatal("the first call, at burst, was refused")
+ }
+ if ok, _ := l.allow("k", now); ok {
+ t.Fatal("a second call at the same instant was allowed with an empty bucket")
+ }
+ // One token at 30/s refills in 1/30 s ≈ 33 ms; 40 ms is comfortably
+ // past that.
+ later := now.Add(40 * time.Millisecond)
+ if ok, _ := l.allow("k", later); !ok {
+ t.Fatal("a call after the refill window was still refused")
+ }
+}
+
+func TestRateLimiterRetryAfterIsBoundedByOneTokensRefillTime(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ l := newRateLimiter(30, 1, 0)
+ l.allow("k", now)
+ _, retry := l.allow("k", now)
+ oneToken := time.Second / 30
+ if retry <= 0 || retry > oneToken+time.Millisecond {
+ t.Fatalf("retryAfter %v, want in (0, %v]", retry, oneToken)
+ }
+}
+
+func TestRateLimiterEvictsABucketIdleOverAMinute(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ l := newRateLimiter(30, 60, 0)
+ l.allow("k", now)
+
+ l.mu.Lock()
+ l.sweepLocked(now.Add(59 * time.Second))
+ _, stillThereAt59s := l.buckets["k"]
+ l.sweepLocked(now.Add(61 * time.Second))
+ _, stillThereAt61s := l.buckets["k"]
+ l.mu.Unlock()
+
+ if !stillThereAt59s {
+ t.Fatal("a bucket under a minute idle was swept early")
+ }
+ if stillThereAt61s {
+ t.Fatal("a bucket over a minute idle was not swept")
+ }
+}
+
+func TestRateLimiterCapEvictsTheLeastRecentlyTouchedBucket(t *testing.T) {
+ now := time.Date(2026, 9, 10, 9, 0, 0, 0, time.UTC)
+ l := newRateLimiter(30, 60, 2)
+ l.allow("a", now)
+ l.allow("b", now.Add(time.Second))
+
+ l.mu.Lock()
+ n := len(l.buckets)
+ l.mu.Unlock()
+ if n != 2 {
+ t.Fatalf("buckets after two distinct keys: %d, want 2", n)
+ }
+
+ // A third key must not grow the map past the cap: "a", the oldest,
+ // is evicted to make room.
+ l.allow("c", now.Add(2*time.Second))
+
+ l.mu.Lock()
+ _, aStillThere := l.buckets["a"]
+ _, bStillThere := l.buckets["b"]
+ _, cThere := l.buckets["c"]
+ size := len(l.buckets)
+ l.mu.Unlock()
+
+ if size > 2 {
+ t.Fatalf("bucket count %d exceeds the cap of 2", size)
+ }
+ if aStillThere {
+ t.Fatal("the least recently touched bucket ('a') was not evicted to make room")
+ }
+ if !bStillThere || !cThere {
+ t.Fatal("a bucket newer than the one evicted was lost instead")
+ }
+}
diff --git a/internal/meet/signal.go b/internal/meet/signal.go
index 120f658..72ff5f4 100644
--- a/internal/meet/signal.go
+++ b/internal/meet/signal.go
@@ -5,6 +5,7 @@ import (
"errors"
"io"
"net/http"
+ "regexp"
"strconv"
"github.com/go-chi/chi/v5"
@@ -15,6 +16,31 @@ import (
// maxSignalBody is spec §4.4's 64 KB cap, refused before decoding.
const maxSignalBody = 64 * 1024
+// 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
+// (review Important 2) — a peer field padded to tens of kilobytes never
+// reaches either.
+var peerIDPattern = regexp.MustCompile(`^[0-9a-f]{32}$`)
+
+func validPeerID(p hub.PeerID) bool { return peerIDPattern.MatchString(string(p)) }
+
+// knownSignalTypes bounds what the log's "type" argument can ever say:
+// one of the wire's own names, or the literal "unknown" — never an
+// attacker's arbitrary string, however long (review Important 2).
+var knownSignalTypes = map[string]bool{
+ hub.TypeJoin: true, hub.TypeAdmit: true, hub.TypeLeave: true,
+ hub.TypeOffer: true, hub.TypeAnswer: true, hub.TypeCandidate: true,
+ hub.TypeReset: true, hub.TypeChat: true, hub.TypeShare: true,
+}
+
+func safeSignalType(t string) string {
+ if knownSignalTypes[t] {
+ return t
+ }
+ return "unknown"
+}
+
// 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
@@ -45,6 +71,24 @@ func (a *app) signal(w http.ResponseWriter, r *http.Request) {
return
}
+ // Rate limiting is keyed on session and room, never on anything the
+ // request body claims, and happens before a single byte of the body
+ // is read: session+room is unforgeable (the session half comes from
+ // the cookie the request actually presented) and is also the one
+ // bucket a signed-in person's traffic in this room can ever spend,
+ // whatever peer id they name — and checking it first means an
+ // oversized flood spends a token too, rather than being free to
+ // retry forever below the size cap (review Important 1).
+ if allowed, retry := a.signalLimiter.allow(session+"|"+sqid, a.store.now()); !allowed {
+ // Retry-After is whole seconds per RFC 9110. At this limiter's
+ // rate, one token refills in 1/30 s = 33 ms, so retry is always
+ // sub-second and this always rounds up to "1" — coarse on
+ // purpose, not a miscalculation (review minor 11).
+ w.Header().Set("Retry-After", strconv.Itoa(int(retry.Seconds())+1))
+ a.refuseSignal(w, sqid, "", "", http.StatusTooManyRequests, "rate", map[string]any{"error": "rate"})
+ 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).
@@ -61,20 +105,17 @@ func (a *app) signal(w http.ResponseWriter, r *http.Request) {
}
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"})
+ if err := json.Unmarshal(raw, &body); err != nil || body.Type == "" || !validPeerID(body.Peer) {
+ // peer is never logged here: an invalid peer is exactly what may
+ // have failed the check, and it has not been validated as
+ // bounded, printable text (review Important 2).
+ a.refuseSignal(w, sqid, "", body.Type, http.StatusBadRequest, "shape", map[string]any{"error": "bad-request"})
return
}
+ // From here body.Peer is known to be exactly 32 lowercase hex —
+ // bounded and safe to use as a log argument and as the hub's lookup
+ // key.
var herr error
switch body.Type {
case hub.TypeJoin:
@@ -83,7 +124,7 @@ func (a *app) signal(w http.ResponseWriter, r *http.Request) {
var p struct {
Peer hub.PeerID `json:"peer"`
}
- if err := json.Unmarshal(body.Payload, &p); err != nil || p.Peer == "" {
+ if err := json.Unmarshal(body.Payload, &p); err != nil || !validPeerID(p.Peer) {
a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusBadRequest, "shape", map[string]any{"error": "bad-request"})
return
}
@@ -117,16 +158,27 @@ func (a *app) signal(w http.ResponseWriter, r *http.Request) {
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):
+ // 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).
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"})
+ // the review ruling requires the two to be indistinguishable,
+ // down to the boot id stream.go's own 404s carry.
+ a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusNotFound, "binding", map[string]any{"error": "not-found", "boot": a.hub.Boot()})
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)
+ // herr here is a driver error surfaced by a.hubStanding's
+ // database reads (every other outcome StandingFor or the hub can
+ // produce already has its own case above) — never a row value,
+ // so nothing user-supplied ever reaches this argument (review
+ // minor 9).
+ a.logger.Error("meet: signal refused", "room", sqid, "peer", string(body.Peer), "type", safeSignalType(body.Type), "outcome", "internal", "err", herr)
+ streamJSON(w, http.StatusInternalServerError, nil)
}
}
@@ -139,10 +191,14 @@ func (a *app) refuseSignal(w http.ResponseWriter, sqid string, peer hub.PeerID,
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.
+// logRefusal is the only signalling log line for a case the switch
+// above already names, 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) — and, since peer and typ may not yet be
+// validated wherever this is called with an empty peer, only ever a
+// peer already known to be exactly 32 lowercase hex and a type already
+// known to be one of the wire's own names or "unknown" (review
+// Important 2). 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)
+ a.logger.Warn("meet: signal refused", "room", sqid, "peer", string(peer), "type", safeSignalType(typ), "outcome", outcome)
}
diff --git a/internal/meettest/signal_test.go b/internal/meettest/signal_test.go
index c225192..be4b9b1 100644
--- a/internal/meettest/signal_test.go
+++ b/internal/meettest/signal_test.go
@@ -3,6 +3,7 @@ package meettest
import (
"bytes"
"encoding/json"
+ "fmt"
"log/slog"
"net/http"
"net/url"
@@ -22,12 +23,24 @@ func TestSignalNeedsAPeerBoundToThisSessionAndRoom(t *testing.T) {
c.signIn("paul@example.com")
s := c.stream("/m/" + sqid + "/stream")
defer s.close()
- peer, _ := s.hello()
+ peer, boot := s.hello()
resp, unknownBody := c.signal(sqid, "0123456789abcdef0123456789abcdef", `{"type":"join"}`)
if resp.StatusCode != http.StatusNotFound {
t.Fatalf("foreign peer: %d", resp.StatusCode)
}
+ // Minor 6: the 404 body carries boot, identical for ErrUnknownPeer
+ // and ErrNotYours, the same shape stream.go's own 404s carry.
+ var unknownAnswer struct {
+ Error string
+ Boot string
+ }
+ if err := json.Unmarshal(unknownBody, &unknownAnswer); err != nil {
+ t.Fatalf("404 body %s: %v", unknownBody, err)
+ }
+ if unknownAnswer.Error != "not-found" || unknownAnswer.Boot != boot {
+ t.Fatalf("404 body %+v, want error=not-found boot=%q", unknownAnswer, boot)
+ }
other := x.browser(t)
other.get("/j/" + tok)
@@ -64,6 +77,131 @@ func TestSignalRefusesAPeerFromAnotherMeeting(t *testing.T) {
}
}
+// Task 5 review Important 1: the limiter is keyed on session+room, never
+// on anything the request body claims, so a second session flooding the
+// endpoint spends only its own bucket.
+func TestSignalRateLimitIsPerSessionNotPerClaimedPeer(t *testing.T) {
+ x, sqid, _ := memberWithMeeting(t)
+ victim := x.browser(t)
+ victim.signIn("paul@example.com")
+ vs := victim.stream("/m/" + sqid + "/stream")
+ defer vs.close()
+ vpeer, _ := vs.hello()
+ if resp, _ := victim.signal(sqid, vpeer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
+ t.Fatalf("victim join: %d", resp.StatusCode)
+ }
+
+ // A second session in the same room, flooding while *claiming the
+ // victim's own peer id* — peer ids are published in rosters, so
+ // this is exactly what a limiter keyed on the body's claimed peer
+ // cannot tell from the victim's own traffic (review Important 1).
+ attacker := newMemberBrowser(t, x, "paul@rslw.com")
+ as := attacker.stream("/m/" + sqid + "/stream")
+ defer as.close()
+ as.hello() // the attacker's own peer, unused below on purpose
+
+ for i := 0; i < 80; i++ {
+ attacker.signal(sqid, vpeer, `{"type":"leave"}`)
+ }
+
+ // The victim's own session, still using its own real peer, must be
+ // unaffected by traffic that only ever came from another session.
+ if resp, _ := victim.signal(sqid, vpeer, `{"type":"leave"}`); resp.StatusCode == http.StatusTooManyRequests {
+ t.Fatal("another session impersonating this peer id in its own requests drained its rate-limit bucket")
+ }
+}
+
+// Task 5 review Important 1: rotating the claimed peer id within one
+// session must not evade the limit — every claimed id, real or not,
+// spends the same session+room bucket.
+func TestSignalRateLimitSharesOneBucketAcrossClaimedPeerIDs(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"}`)
+
+ var got429 bool
+ for i := 0; i < 100; i++ {
+ // A distinct, format-valid, but otherwise made-up peer id on
+ // every call: none of them is bound to anything, so if the
+ // limiter were still keyed on the claimed peer, rotating like
+ // this would never be throttled.
+ fake := fmt.Sprintf("%032x", i)
+ resp, _ := c.signal(sqid, fake, `{"type":"join"}`)
+ if resp.StatusCode == http.StatusTooManyRequests {
+ got429 = true
+ break
+ }
+ }
+ if !got429 {
+ t.Fatal("rotating the claimed peer id within one session evaded the rate limit")
+ }
+}
+
+// Task 5 review Important 2: an oversized peer field is refused, and
+// never becomes a log argument — the log's "peer" is only ever a value
+// already known to be exactly 32 lowercase hex.
+func TestSignalRefusesAnOversizedPeerAndNeverLogsIt(t *testing.T) {
+ var buf bytes.Buffer
+ x := newHarnessWithLogger(t, slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})), "paul@example.com")
+ c := x.browser(t)
+ c.signIn("paul@example.com")
+ sqid, _ := createMeeting(t, c, "x", "")
+ s := c.stream("/m/" + sqid + "/stream")
+ defer s.close()
+ s.hello()
+
+ huge := strings.Repeat("a", 60*1024)
+ resp, _ := c.signal(sqid, huge, `{"type":"join"}`)
+ if resp.StatusCode != http.StatusBadRequest {
+ t.Fatalf("60 KB peer: %d, want 400", resp.StatusCode)
+ }
+ if strings.Contains(buf.String(), huge) {
+ t.Fatal("the oversized peer string reached the log")
+ }
+}
+
+// Task 5 review minor 8: the body shapes signal.go must classify
+// correctly, including the ones client.signal cannot express (a
+// malformed document, a body with no peer at all).
+func TestSignalBodyShapeTable(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()
+ if resp, _ := c.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
+ t.Fatalf("setup join: %d", resp.StatusCode)
+ }
+ absent := fmt.Sprintf("%032x", 0) // well-formed, bound to nobody
+
+ table := []struct {
+ name string
+ body string
+ want int
+ }{
+ {"chat is slice 4", `{"peer":"` + string(peer) + `","type":"chat"}`, http.StatusBadRequest},
+ {"share is slice 4", `{"peer":"` + string(peer) + `","type":"share"}`, http.StatusBadRequest},
+ {"bogus type", `{"peer":"` + string(peer) + `","type":"bogus"}`, http.StatusBadRequest},
+ {"malformed JSON", `{`, http.StatusBadRequest},
+ {"missing peer", `{"type":"join"}`, http.StatusBadRequest},
+ {"malformed admit payload.peer", `{"peer":"` + string(peer) + `","type":"admit","payload":{"peer":"not-hex"}}`, http.StatusBadRequest},
+ {"well-formed absent admit target", `{"peer":"` + string(peer) + `","type":"admit","payload":{"peer":"` + absent + `"}}`, http.StatusConflict},
+ }
+ for _, tc := range table {
+ t.Run(tc.name, func(t *testing.T) {
+ resp, _ := c.signalRaw(sqid, tc.body)
+ if resp.StatusCode != tc.want {
+ t.Fatalf("%s: %d, want %d", tc.name, resp.StatusCode, tc.want)
+ }
+ })
+ }
+}
+
func TestSignalLimits(t *testing.T) {
x, sqid, _ := memberWithMeeting(t)
c := x.browser(t)
@@ -94,7 +232,10 @@ func TestSignalLimits(t *testing.T) {
}
}
-func TestSignalStaleGenAnswers409WithTheCurrentGen(t *testing.T) {
+// Renamed from TestSignalStaleGenAnswers409WithTheCurrentGen: the
+// ledger records the decision that no gen goes inline in the response
+// (see task-5-report.md) — the roster already carries it (spec §4.6).
+func TestSignalStaleGenIsRefusedWith409(t *testing.T) {
x, sqid, _ := memberWithMeeting(t)
a := x.browser(t)
a.signIn("paul@example.com")
@@ -241,32 +382,86 @@ func TestSignalAdmittedResumeReplaysAfterRoster(t *testing.T) {
}
}
-// 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.
+// The §0.1 and §4.4 promise: no payload, address, name, title, token or
+// session key in any log line, on accepted and refused messages and on
+// an internal-error path. Task 5 review item 4: the earlier version of
+// this test passed vacuously — it never exercised a guest's knock, an
+// admit, or the 500 arm, and its address ("paul@rslw.com") happened not
+// to leave a distinctive trace ("paul" and "example.com" both appear
+// harmlessly elsewhere). zaphod-xyzzy@example.com is chosen so its
+// derived display name (store.ensurePerson's local-part rule,
+// "zaphod-xyzzy") is its own needle.
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")
+ x := newHarnessWithLogger(t, slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})), "paul@example.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"}`)
+ if resp, _ := c.signal(sqid, peer, `{"type":"join"}`); resp.StatusCode != http.StatusOK {
+ t.Fatalf("member join: %d", resp.StatusCode)
+ }
+
+ // A real guest, via the join link, not another member: its join is
+ // the knock, and it stays out of the log entirely until admitted.
+ guest := x.browser(t)
+ guest.get("/j/" + tok)
+ guest.signIn("zaphod-xyzzy@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 (the knock): %d", resp.StatusCode)
+ }
+
+ // The admit round: previously untested by this file at all.
+ if resp, _ := c.signal(sqid, peer, `{"type":"admit","payload":{"peer":"`+gpeer+`"}}`); resp.StatusCode != http.StatusOK {
+ t.Fatalf("admit: %d", resp.StatusCode)
+ }
+
+ c.signal(sqid, peer, `{"type":"offer","to":"`+gpeer+`","seq":1,"gen":1,"payload":{"sdp":"v=0 SDPSECRET a=candidate 203.0.113.9"}}`)
+ c.signal(sqid, peer, `{"type":"candidate","to":"`+gpeer+`","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":"`+gpeer+`","seq":3,"gen":9,"payload":{"sdp":"STALESECRET"}}`) // outcome=stale-gen
+ c.signal(sqid, peer, `{"type":"bogus"}`) // outcome=type
+
x.advance(time.Minute)
x.ctl.TickForTest()
+
+ // Force the internal-error arm without breaking sign-in itself:
+ // closing the whole database also breaks rastrillo's own
+ // per-request session lookup, which refuses the request before it
+ // ever reaches the signal handler. Dropping only the meetings table
+ // leaves sessions and persons answering normally, so the request
+ // reaches hub.Join, whose standing re-derivation reads meetings and
+ // surfaces a genuine driver error — a sentinel this switch does not
+ // already name — exercising the one log line no other case in this
+ // test reaches (review minor 9).
+ if err := x.db.G.Exec("DROP TABLE meetings").Error; err != nil {
+ t.Fatal(err)
+ }
+ c.signal(sqid, peer, `{"type":"join"}`)
+
logs := buf.String()
- for _, needle := range []string{"Secret title", "SDPSECRET", "203.0.113.9", "STALESECRET", tok, "paul@example.com"} {
- if strings.Contains(logs, needle) {
+ for _, want := range []string{"outcome=stale-gen", "outcome=type", "outcome=internal"} {
+ if !strings.Contains(logs, want) {
+ t.Errorf("log missing %q:\n%s", want, logs)
+ }
+ }
+
+ forbidden := []string{
+ "Secret title", "SDPSECRET", "203.0.113.9", "STALESECRET", tok,
+ "paul@example.com", "zaphod-xyzzy@example.com", "zaphod-xyzzy",
+ }
+ for _, ck := range c.cookies {
+ forbidden = append(forbidden, ck.Value)
+ }
+ for _, ck := range guest.cookies {
+ forbidden = append(forbidden, ck.Value)
+ }
+ for _, needle := range forbidden {
+ if needle != "" && 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 3c7f0df..af43d30 100644
--- a/internal/meettest/sse_test.go
+++ b/internal/meettest/sse_test.go
@@ -246,3 +246,31 @@ func (c *client) signal(sqid, peer, body string) (*http.Response, []byte) {
}
return resp, b
}
+
+// signalRaw posts body verbatim, unlike signal, which always injects a
+// well-formed "peer" — for the shapes signal cannot express at all: a
+// malformed document, or one with no peer field.
+func (c *client) signalRaw(sqid, body string) (*http.Response, []byte) {
+ c.t.Helper()
+ 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", strings.NewReader(body))
+ if err != nil {
+ c.t.Fatal(err)
+ }
+ for _, ck := range c.cookies {
+ req.AddCookie(ck)
+ }
+ req.Header.Set("Content-Type", "application/json")
+ req.Header.Set("Origin", testOrigin)
+ resp, err := c.x.server().Client().Do(req)
+ if err != nil {
+ c.t.Fatalf("POST signal: %v", err)
+ }
+ b, err := io.ReadAll(resp.Body)
+ resp.Body.Close()
+ if err != nil {
+ c.t.Fatalf("POST signal: reading body: %v", err)
+ }
+ return resp, b
+}