amadan
signal: validate the room id before the limiter, safer 429 log peer
Fix round 2 on Task 5 re-review. - Item 1: sqid is validated (ids.go's validSqid: exactly ten characters of sqidAlphabet) before it can reach the limiter's key or a log line; else the identical not-found body, unlogged. A session with no live participation in that room (a.hub.LiveForSession) is then refused the same identical body before the limiter is ever touched, so rotating the room id in the URL can no longer mint a fresh bucket per request. rateLimiter gains a size() method and TestHandle a SignalLimiterSize() test hook to observe this directly. - Item 2: the 429 log line now carries the peer LiveForSession returned — server-derived, never the request's own claim — instead of an empty string. - Item 3: newRateLimiter's cap parameter/field renamed to maxKeys (no longer shadows the builtin). - Item 4: the no-session 500 now uses streamJSON(w, 500, nil), matching every other refusal in this file, instead of a bare WriteHeader. - Item 5: comment on TestSignalRateLimitSharesOneBucketAcrossClaimedPeerIDs noting the room half of the key is now gated by LiveForSession. - Item 6: TestAdmitByAWaitingGuestNeverLooksAtTheTargetOrItsStanding gains the admitted-guest actor case (same short-circuit one step later: Kind != member). - Item 7: the log test's DROP TABLE arm is now its own trailing t.Run, named for what it does and that it must run last. TestSignalLimits' burst loop switched from repeating leave to repeating join: a successful leave ends the participation, and from this round on a session with no live participation in the room is refused at the new gate before it ever reaches the limiter again — a burst that ended its own room membership on the first call could never prove anything past that point. join on an already-admitted peer is idempotent, so the gate and the limiter both stay reachable throughout. Every new or changed test was run against the pre-round-2 code first (reverted the LiveForSession/validSqid gate and the bare WriteHeader, keeping the new size()/SignalLimiterSize test hooks in place as instrumentation rather than behaviour under test) and confirmed to fail there; see task-5-report.md.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2
6 files changed,
+212
−31
internal/meet/app.go+7 −0internal/meet/hub/admission_test.go+23 −0internal/meet/ids.go+10 −0internal/meet/ratelimit.go+28 −17internal/meet/signal.go+52 −8internal/meettest/signal_test.go+92 −6
CI passed — run details
diff --git a/internal/meet/app.go b/internal/meet/app.go| index 70f3f40..0aeecc7 100644 |
| --- a/internal/meet/app.go |
| +++ b/internal/meet/app.go |
| @@ -115,6 +115,13 @@ func (h *TestHandle) SweepForTest(now time.Time) { |
| // without reaching into the store directly. |
| func (h *TestHandle) NextDue() time.Time { return h.a.nextDue() } |
| +// SignalLimiterSize answers how many keys the signal endpoint's rate |
| +// limiter currently holds — a black-box test's only window into |
| +// whether a request minted a bucket at all (review Task 5 item 1: a |
| +// room id rotation, or anything else refused before the limiter, must |
| +// leave this unchanged). |
| +func (h *TestHandle) SignalLimiterSize() int { return h.a.signalLimiter.size() } |
| + |
| // gormWriter is GORM's logger.Writer over the app's slog logger, so |
| // GORM's warnings land in one log stream rather than a second stdout |
| // format. rastrillo/db has the same adapter, unexported. |
diff --git a/internal/meet/hub/admission_test.go b/internal/meet/hub/admission_test.go| index 7c769af..b70744e 100644 |
| --- a/internal/meet/hub/admission_test.go |
| +++ b/internal/meet/hub/admission_test.go |
| @@ -133,6 +133,29 @@ func TestAdmitByAWaitingGuestNeverLooksAtTheTargetOrItsStanding(t *testing.T) { |
| 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) |
| } |
| + |
| + // The same short-circuit, one step later: an *admitted* guest is |
| + // past the !a.Admitted check but still fails a.Kind != KindMember, |
| + // so the target is still never reached (review Task 5 fix round 2 |
| + // item 6). |
| + if err := h.Admit(ctx, "r1", "sm", m.Peer, g.Peer); err != nil { |
| + t.Fatalf("member admitting the waiting guest: %v", err) |
| + } |
| + |
| + calls = 0 |
| + errAbsentAdmitted := h.Admit(ctx, "r1", "sg", g.Peer, "0123456789abcdef0123456789abcdef") |
| + callsForAbsentAdmitted := calls |
| + |
| + calls = 0 |
| + errPresentAdmitted := h.Admit(ctx, "r1", "sg", g.Peer, m.Peer) |
| + callsForPresentAdmitted := calls |
| + |
| + if errAbsentAdmitted != ErrNotMember || errPresentAdmitted != ErrNotMember { |
| + t.Fatalf("admitted guest's admit: absent target %v, present target %v, want ErrNotMember both times", errAbsentAdmitted, errPresentAdmitted) |
| + } |
| + if callsForAbsentAdmitted != 1 || callsForPresentAdmitted != 1 { |
| + t.Fatalf("StandingFor calls (admitted guest actor): absent target %d, present target %d, want 1 both times (the actor's own — the target is never reached)", callsForAbsentAdmitted, callsForPresentAdmitted) |
| + } |
| } |
| func TestCapAtJoinAndAtAdmit(t *testing.T) { |
diff --git a/internal/meet/ids.go b/internal/meet/ids.go| index edf570d..f7da34b 100644 |
| --- a/internal/meet/ids.go |
| +++ b/internal/meet/ids.go |
| @@ -6,6 +6,7 @@ import ( |
| "encoding/base64" |
| "encoding/hex" |
| "fmt" |
| + "strings" |
| ) |
| // sqidAlphabet is Docs' and Slides' alphabet, character for character: |
| @@ -14,6 +15,15 @@ const sqidAlphabet = "23456789abcdefghjkmnpqrstuvwxyz" |
| const sqidLength = 10 |
| const maxSqidAttempts = 5 |
| +// validSqid reports whether s is exactly the shape newSqid ever |
| +// produces: ten characters, every one of them in sqidAlphabet. A |
| +// caller that trusts a room id straight from a URL — a rate limiter's |
| +// key, a log line — must check this first: chi hands back whatever the |
| +// request line contained, unbounded (review Task 5 item 1). |
| +func validSqid(s string) bool { |
| + return len(s) == sqidLength && strings.Trim(s, sqidAlphabet) == "" |
| +} |
| + |
| func newSqid() (string, error) { |
| raw := make([]byte, sqidLength) |
| if _, err := rand.Read(raw); err != nil { |
diff --git a/internal/meet/ratelimit.go b/internal/meet/ratelimit.go| index 58a5ffa..bb039f0 100644 |
| --- a/internal/meet/ratelimit.go |
| +++ b/internal/meet/ratelimit.go |
| @@ -20,19 +20,21 @@ const maxSignalBuckets = 4096 |
| // 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 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. |
| +// Entries are process-local, like the hub itself, and bounded by |
| +// maxKeys: once the map holds that many 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 |
| + rate float64 // tokens added per second |
| + burst float64 // bucket capacity |
| + maxKeys int // 0 means unbounded — used only by tests; named to |
| + // avoid shadowing the builtin cap (review Task 5 item 3) |
| mu sync.Mutex |
| buckets map[string]*tokenBucket |
| @@ -44,8 +46,8 @@ type tokenBucket struct { |
| seenAt time.Time |
| } |
| -func newRateLimiter(rate, burst, cap int) *rateLimiter { |
| - return &rateLimiter{rate: float64(rate), burst: float64(burst), cap: cap, buckets: map[string]*tokenBucket{}} |
| +func newRateLimiter(rate, burst, maxKeys int) *rateLimiter { |
| + return &rateLimiter{rate: float64(rate), burst: float64(burst), maxKeys: maxKeys, buckets: map[string]*tokenBucket{}} |
| } |
| // allow spends one token for key, refilling by the time elapsed since |
| @@ -62,7 +64,7 @@ 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 { |
| + if l.maxKeys > 0 && len(l.buckets) >= l.maxKeys { |
| l.evictOldestLocked() |
| } |
| b = &tokenBucket{tokens: l.burst, seenAt: now} |
| @@ -90,10 +92,10 @@ 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 |
| +// bounded limiter never grows past maxKeys: 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. |
| +// right at the limit — everyday traffic never reaches it. |
| func (l *rateLimiter) evictOldestLocked() { |
| var oldestKey string |
| var oldestAt time.Time |
| @@ -107,3 +109,12 @@ func (l *rateLimiter) evictOldestLocked() { |
| delete(l.buckets, oldestKey) |
| } |
| } |
| + |
| +// size is how many keys the limiter currently holds — a test-only |
| +// window into whether a request minted a bucket at all (review Task 5 |
| +// item 1's rotation test). |
| +func (l *rateLimiter) size() int { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + return len(l.buckets) |
| +} |
diff --git a/internal/meet/signal.go b/internal/meet/signal.go| index 72ff5f4..1789603 100644 |
| --- a/internal/meet/signal.go |
| +++ b/internal/meet/signal.go |
| @@ -67,25 +67,50 @@ func (a *app) signal(w http.ResponseWriter, r *http.Request) { |
| session, ok := a.sessionKey(r) |
| if !ok { |
| a.logger.Error("meet: signal with no session subject", "path", r.URL.Path) |
| - w.WriteHeader(http.StatusInternalServerError) |
| + streamJSON(w, http.StatusInternalServerError, nil) |
| + return |
| + } |
| + |
| + // sqid is checked before it can reach anything else: the limiter's |
| + // key, a log line, or a hub lookup. chi hands back whatever the |
| + // request line contained, unbounded — a kilobyte of it would |
| + // otherwise mint a rate-limit bucket and a log line of its own on |
| + // every request (review Task 5 item 1). |
| + if !validSqid(sqid) { |
| + a.refuseBadRoom(w) |
| + return |
| + } |
| + |
| + // A session with no live participation in this exact room is |
| + // refused the identical not-found answer before the limiter is |
| + // touched at all: nothing is spent and nothing is minted for a room |
| + // this session has never opened a stream in, which is what closes |
| + // off room-id rotation as a way to mint a fresh bucket per request |
| + // (review Task 5 item 1). livePeer is the hub's own record of what |
| + // this session holds — server-derived, never the request's own |
| + // claim — so it is safe to log from here on (review item 2). |
| + livePeer, live := a.hub.LiveForSession(sqid, session) |
| + if !live { |
| + a.refuseSignal(w, sqid, "", "", http.StatusNotFound, "unbound", a.notFoundBody()) |
| 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). |
| + // the cookie the request actually presented, the room half is now |
| + // gated by LiveForSession above) 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"}) |
| + a.refuseSignal(w, sqid, livePeer, "", http.StatusTooManyRequests, "rate", map[string]any{"error": "rate"}) |
| return |
| } |
| @@ -170,7 +195,7 @@ func (a *app) signal(w http.ResponseWriter, r *http.Request) { |
| // ErrNotYours and ErrUnknownPeer share this one body on purpose: |
| // 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()}) |
| + a.refuseSignal(w, sqid, body.Peer, body.Type, http.StatusNotFound, "binding", a.notFoundBody()) |
| default: |
| // herr here is a driver error surfaced by a.hubStanding's |
| // database reads (every other outcome StandingFor or the hub can |
| @@ -182,6 +207,25 @@ func (a *app) signal(w http.ResponseWriter, r *http.Request) { |
| } |
| } |
| +// notFoundBody is the one not-found shape this endpoint ever answers: |
| +// stream.go's own precedent, so a peer that is not yours, a peer that |
| +// never existed, and a room this session holds nothing in are all one |
| +// indistinguishable body (spec §2.5). |
| +func (a *app) notFoundBody() map[string]any { |
| + return map[string]any{"error": "not-found", "boot": a.hub.Boot()} |
| +} |
| + |
| +// refuseBadRoom answers a room id that is not the shape newSqid ever |
| +// produces, with the identical not-found body every other refusal |
| +// here carries — logged without the room id itself, unlike every other |
| +// sqid this handler logs: this one has not been validated as bounded |
| +// or printable, and chi hands back whatever the request line contained |
| +// (review Task 5 item 1). |
| +func (a *app) refuseBadRoom(w http.ResponseWriter) { |
| + a.logger.Warn("meet: signal refused", "room", "invalid", "peer", "", "type", "unknown", "outcome", "bad-room") |
| + streamJSON(w, http.StatusNotFound, a.notFoundBody()) |
| +} |
| + |
| // 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 |
diff --git a/internal/meettest/signal_test.go b/internal/meettest/signal_test.go| index be4b9b1..e047781 100644 |
| --- a/internal/meettest/signal_test.go |
| +++ b/internal/meettest/signal_test.go |
| @@ -113,7 +113,10 @@ func TestSignalRateLimitIsPerSessionNotPerClaimedPeer(t *testing.T) { |
| // 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. |
| +// spends the same session+room bucket. (The room half of the key is |
| +// gated by LiveForSession before the limiter is ever reached — fix |
| +// round 2 item 1 — so this test is only about the peer half: the |
| +// session here already holds a live participation in sqid throughout.) |
| func TestSignalRateLimitSharesOneBucketAcrossClaimedPeerIDs(t *testing.T) { |
| x, sqid, _ := memberWithMeeting(t) |
| c := x.browser(t) |
| @@ -164,6 +167,71 @@ func TestSignalRefusesAnOversizedPeerAndNeverLogsIt(t *testing.T) { |
| } |
| } |
| +// fakeButValidSqid builds a room id in the exact shape ids.go's |
| +// validSqid accepts (ten characters of sqidAlphabet) without touching |
| +// crypto/rand — deterministic and cheap for a test that needs many |
| +// distinct, well-formed ids no meeting will ever have. |
| +func fakeButValidSqid(i int) string { |
| + const digits = "23456789" // a safe subset of sqidAlphabet |
| + b := make([]byte, 10) |
| + for j := 9; j >= 0; j-- { |
| + b[j] = digits[i%len(digits)] |
| + i /= len(digits) |
| + } |
| + return string(b) |
| +} |
| + |
| +// Task 5 review fix round 2 item 1: rotating the room id in the URL |
| +// must not mint a rate-limit bucket per request. A session with no live |
| +// participation in a room is refused before the limiter is ever |
| +// touched, so fifty rotated (but well-formed) room ids leave the |
| +// limiter's map exactly as it was after the one real signal. |
| +func TestSignalSqidRotationNeverMintsARateLimiterBucket(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("join: %d", resp.StatusCode) |
| + } |
| + if n := x.ctl.SignalLimiterSize(); n != 1 { |
| + t.Fatalf("limiter buckets after one signal: %d, want 1", n) |
| + } |
| + |
| + for i := 0; i < 50; i++ { |
| + fake := fakeButValidSqid(i) |
| + if resp, _ := c.signal(fake, peer, `{"type":"join"}`); resp.StatusCode != http.StatusNotFound { |
| + t.Fatalf("rotated room %q: %d, want 404", fake, resp.StatusCode) |
| + } |
| + } |
| + if n := x.ctl.SignalLimiterSize(); n != 1 { |
| + t.Fatalf("limiter buckets after 50 rotated rooms: %d, want 1 — rotation must not mint a bucket per request", n) |
| + } |
| +} |
| + |
| +// Task 5 review fix round 2 item 1: an oversized room id is refused |
| +// before it can reach the limiter's key, mint a bucket, or a log line. |
| +func TestSignalRefusesAnOversizedRoomIDAndNeverLogsIt(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") |
| + |
| + huge := strings.Repeat("a", 1024) |
| + resp, _ := c.signalRaw(huge, `{"peer":"0123456789abcdef0123456789abcdef","type":"join"}`) |
| + if resp.StatusCode != http.StatusNotFound { |
| + t.Fatalf("1 KB room id: %d, want 404", resp.StatusCode) |
| + } |
| + if n := x.ctl.SignalLimiterSize(); n != 0 { |
| + t.Fatalf("limiter buckets after an oversized room id: %d, want 0", n) |
| + } |
| + if strings.Contains(buf.String(), huge) { |
| + t.Fatal("the oversized room id 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). |
| @@ -216,9 +284,16 @@ func TestSignalLimits(t *testing.T) { |
| t.Fatalf("64 KB: %d", resp.StatusCode) |
| } |
| + // join, not leave: a successful leave ends the participation, and |
| + // from fix round 2 on a session with no live participation in this |
| + // room is refused at the LiveForSession gate before it ever reaches |
| + // the limiter (item 1) — a burst that ended its own room membership |
| + // on the first call would never reach the limiter again to prove |
| + // anything. join on an already-admitted peer is idempotent (spec |
| + // §4.3), so the participation, and the gate, stay open throughout. |
| var got429 bool |
| for i := 0; i < 100; i++ { |
| - resp, _ := c.signal(sqid, peer, `{"type":"leave"}`) |
| + resp, _ := c.signal(sqid, peer, `{"type":"join"}`) |
| if resp.StatusCode == http.StatusTooManyRequests { |
| if resp.Header.Get("Retry-After") == "" { |
| t.Fatal("429 without Retry-After") |
| @@ -438,10 +513,21 @@ func TestSignallingLogsCarryNoContent(t *testing.T) { |
| // 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"}`) |
| + // |
| + // In its own trailing t.Run, named for what it does and that it |
| + // must be last: the table it drops is gone for the rest of the |
| + // process, so nothing after this subtest may touch it (review fix |
| + // round 2 item 7 — a structural marker, not just a comment, for a |
| + // constraint a future edit could otherwise violate silently). |
| + t.Run("forced internal error (must run last: drops the meetings table)", func(t *testing.T) { |
| + if err := x.db.G.Exec("DROP TABLE meetings").Error; err != nil { |
| + t.Fatal(err) |
| + } |
| + resp, _ := c.signal(sqid, peer, `{"type":"join"}`) |
| + if resp.StatusCode != http.StatusInternalServerError { |
| + t.Fatalf("forced internal error: %d, want 500", resp.StatusCode) |
| + } |
| + }) |
| logs := buf.String() |
| for _, want := range []string{"outcome=stale-gen", "outcome=type", "outcome=internal"} { |