amadan
hub: a tombstone answers only its own room and session
Review round 1 on Task 3. Important 1. h.tombstones was keyed by peer id alone, and the refusal ladder read it before the room and session checks. Peer ids are published in every roster, so for ten minutes any past room-mate could resume with someone else's id and be told why they left — removed, standing, expired, replaced, ended — and that is account state. The tombstone now records the room and session it was bound to, the check moved below the room lookup, and a mismatch falls through to the same 404-with-boot an unknown peer gets. Task 4's stream test worked around this with a peer id the hub had never heard of; it now offers the real tombstoned peer to a room it was never in and expects the 404. Minor 3. Relay and Reset refuse a self target explicitly. It was not refused at all: the pair generation for (a, a) starts at one like any other, so gen 1 was accepted and echoed. Minor 4. Gap and replay are filtered against the roster just built. The ring and the eviction marks outlive the peer that filled them, so a resume could name a peer that had left: a gap asking for a reset with nobody, or replayed signalling rebuilding the connection the roster had just told the browser to close. Targeted replay is filtered too, not only the gap and the pair the review named — the same argument covers all three, and a detached peer inside its resume window is still in the roster, so nothing legitimate is dropped. Minor 5. ErrNotTargeted is a sentinel, wrapped with the type name. Minor 2 needed nothing: Task 4's Detach identity check and its test are on the branch already.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CzvnuKhLzudvq7jZ7SvVt2
4 files changed,
+144
−18
internal/meet/hub/hub.go+11 −3internal/meet/hub/relay.go+34 −8internal/meet/hub/relay_test.go+95 −2internal/meettest/stream_test.go+4 −5
CI passed — run details
diff --git a/internal/meet/hub/hub.go b/internal/meet/hub/hub.go| index 10a7f89..6bb4ea5 100644 |
| --- a/internal/meet/hub/hub.go |
| +++ b/internal/meet/hub/hub.go |
| @@ -58,6 +58,7 @@ var ( |
| ErrNotAdmitted = errors.New("hub: peer is not admitted") |
| ErrNotMember = errors.New("hub: peer is not a member") |
| ErrNotWaiting = errors.New("hub: target is not waiting") |
| + ErrNotTargeted = errors.New("hub: not a targeted message type") |
| ErrStaleGen = errors.New("hub: stale generation") |
| ErrTooManyWaiting = errors.New("hub: eight already waiting") |
| ) |
| @@ -108,9 +109,16 @@ type room struct { |
| ended bool |
| } |
| +// tombstone remembers why a participation ended, for ten minutes, so a |
| +// stream that never received its lost event learns the reason on resume |
| +// (spec §4.5). It carries the room and session it was bound to because |
| +// the reason is account state — removed, standing, expired — and a peer |
| +// id is published in rosters and is not a credential (spec §4.2). |
| type tombstone struct { |
| - reason Reason |
| - at time.Time |
| + room string |
| + session string |
| + reason Reason |
| + at time.Time |
| } |
| type Hub struct { |
| @@ -273,7 +281,7 @@ func (h *Hub) endLocked(r *room, p *Participation, reason Reason) (wasAdmitted, |
| if h.byPerson[p.PersonID] > 0 { |
| h.byPerson[p.PersonID]-- |
| } |
| - h.tombstones[p.Peer] = tombstone{reason: reason, at: h.cfg.Now()} |
| + h.tombstones[p.Peer] = tombstone{room: r.sqid, session: p.Session, reason: reason, at: h.cfg.Now()} |
| for k := range r.gens { |
| if k.a == p.Peer || k.b == p.Peer { |
| delete(r.gens, k) |
diff --git a/internal/meet/hub/relay.go b/internal/meet/hub/relay.go| index 614decf..63bc4d6 100644 |
| --- a/internal/meet/hub/relay.go |
| +++ b/internal/meet/hub/relay.go |
| @@ -11,7 +11,7 @@ func isTargeted(t string) bool { return t == TypeOffer || t == TypeAnswer || t = |
| // binding; the payload is never read. |
| func (h *Hub) Relay(sqid, session string, from PeerID, m Message) error { |
| if !isTargeted(m.Type) { |
| - return fmt.Errorf("hub: %q is not a targeted type", m.Type) |
| + return fmt.Errorf("%w: %q", ErrNotTargeted, m.Type) |
| } |
| h.mu.Lock() |
| defer h.mu.Unlock() |
| @@ -26,6 +26,9 @@ func (h *Hub) Relay(sqid, session string, from PeerID, m Message) error { |
| if !p.Admitted { |
| return ErrNotAdmitted |
| } |
| + if m.To == p.Peer { |
| + return ErrUnknownPeer // a peer is not its own target |
| + } |
| tp, ok := r.peers[m.To] |
| if !ok || !tp.Admitted { |
| return ErrUnknownPeer |
| @@ -77,6 +80,9 @@ func (h *Hub) Reset(sqid, session string, from, to PeerID) error { |
| if !p.Admitted { |
| return ErrNotAdmitted |
| } |
| + if to == p.Peer { |
| + return ErrUnknownPeer // a peer is not its own pair |
| + } |
| tp, ok := r.peers[to] |
| if !ok || !tp.Admitted { |
| return ErrUnknownPeer |
| @@ -121,12 +127,16 @@ func (e *ResumeError) Error() string { |
| // place it is written: Resume applies it before attaching, and |
| // CanResume applies it and stops there. |
| func (h *Hub) resumeRefusalLocked(sqid, session string, peer PeerID) *ResumeError { |
| - if t, ok := h.tombstones[peer]; ok { |
| - return &ResumeError{Status: 410, Reason: t.reason} |
| - } |
| if _, ok := h.rooms[sqid]; !ok { |
| return &ResumeError{Status: 404, Boot: h.boot} |
| } |
| + // A tombstone is answered only through the room and session it was |
| + // bound to. Why a participation ended is account state, and a peer |
| + // id is published in every roster, so an ended peer's room-mate must |
| + // not be able to read the reason off it (spec §4.2's binding rule). |
| + if t, ok := h.tombstones[peer]; ok && t.room == sqid && t.session == session { |
| + return &ResumeError{Status: 410, Reason: t.reason} |
| + } |
| if _, found := h.findLocked(sqid, session, peer); !found { |
| // Unknown here, unknown in another session, unknown in another |
| // room: one answer, so nothing leaks (spec §4.2). |
| @@ -173,8 +183,18 @@ func (h *Hub) Resume(sqid, session string, peer PeerID, last map[PeerID]uint64) |
| // Replay is per sender: everything above last, unless the ring has |
| // dropped something above last, in which case that sender gets a gap |
| // and no partial replay (spec §4.5). |
| + // |
| + // Everything here is filtered against the roster that has just been |
| + // built. The ring and the eviction marks outlive the peers that |
| + // filled them, and the roster is the authority a browser reconciles |
| + // against (§4.5): it has already closed its connection to a peer |
| + // that is gone, so it must not be told to reset with it, replay its |
| + // signalling, or rebuild its pair. |
| gapped := map[PeerID]bool{} |
| for from, high := range p.evicted { |
| + if _, present := rosterGen(res.Roster, from); !present { |
| + continue |
| + } |
| if high > last[from] { |
| gapped[from] = true |
| res.Gap = append(res.Gap, from) |
| @@ -183,6 +203,9 @@ func (h *Hub) Resume(sqid, session string, peer PeerID, last map[PeerID]uint64) |
| for _, m := range p.ring { |
| switch { |
| case isTargeted(m.Type): |
| + if _, present := rosterGen(res.Roster, m.From); !present { |
| + continue |
| + } |
| if !gapped[m.From] && m.Seq > last[m.From] { |
| res.Replay = append(res.Replay, m) |
| } |
| @@ -191,7 +214,8 @@ func (h *Hub) Resume(sqid, session string, peer PeerID, last map[PeerID]uint64) |
| // admitted peer, so this never fires for a current roster; |
| // the branch is kept because the ring is the authority for |
| // what was sent, and the browser reconciles from both (§4.6). |
| - if m.Gen > h.currentGenSeenLocked(res.Roster, m.Peer) { |
| + gen, present := rosterGen(res.Roster, m.Peer) |
| + if present && m.Gen > gen { |
| res.Replay = append(res.Replay, m) |
| } |
| } |
| @@ -199,11 +223,13 @@ func (h *Hub) Resume(sqid, session string, peer PeerID, last map[PeerID]uint64) |
| return res, nil |
| } |
| -func (h *Hub) currentGenSeenLocked(ro *Roster, peer PeerID) uint64 { |
| +// rosterGen is the recipient's generation with one peer, and whether |
| +// that peer is in the roster at all. |
| +func rosterGen(ro *Roster, peer PeerID) (uint64, bool) { |
| for _, rp := range ro.Peers { |
| if rp.Peer == peer { |
| - return rp.Gen |
| + return rp.Gen, true |
| } |
| } |
| - return 0 |
| + return 0, false |
| } |
diff --git a/internal/meet/hub/relay_test.go b/internal/meet/hub/relay_test.go| index 8ba5a0b..050fbf9 100644 |
| --- a/internal/meet/hub/relay_test.go |
| +++ b/internal/meet/hub/relay_test.go |
| @@ -3,6 +3,7 @@ package hub |
| import ( |
| "context" |
| "encoding/json" |
| + "errors" |
| "testing" |
| "time" |
| ) |
| @@ -54,8 +55,28 @@ func TestRelayGoesOnlyToTheNamedAdmittedPeerWithServerBuiltEnvelope(t *testing.T |
| if err := h.Relay("r2", "sa", a.Peer, Message{Type: TypeOffer, To: b.Peer, Seq: 3, Gen: 1}); err != ErrUnknownPeer { |
| t.Fatalf("peer through another room: %v", err) |
| } |
| - if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeChat, To: b.Peer, Seq: 4, Gen: 1}); err == nil { |
| - t.Fatal("chat accepted as a targeted relay") |
| + if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeChat, To: b.Peer, Seq: 4, Gen: 1}); !errors.Is(err, ErrNotTargeted) { |
| + t.Fatalf("chat as a targeted relay: %v", err) |
| + } |
| +} |
| + |
| +// A peer is not its own target: a self-addressed relay or reset would |
| +// otherwise ring a message the sender already has and open a pair with |
| +// itself, which no roster ever names. |
| +func TestAPeerIsNotItsOwnTarget(t *testing.T) { |
| + h, _, a, _ := twoAdmitted(t) |
| + if err := h.Relay("r1", "sa", a.Peer, Message{Type: TypeOffer, To: a.Peer, Seq: 1, Gen: 1}); err != ErrUnknownPeer { |
| + t.Fatalf("self-addressed relay: %v", err) |
| + } |
| + if err := h.Reset("r1", "sa", a.Peer, a.Peer); err != ErrUnknownPeer { |
| + t.Fatalf("self-addressed reset: %v", err) |
| + } |
| + if ev := drain(a.Sub); len(ev) != 0 { |
| + t.Fatalf("a self target reached the sender: %v", types(ev)) |
| + } |
| + p, _ := h.Find("r1", "sa", a.Peer) |
| + if len(p.ring) != 0 { |
| + t.Fatalf("a self target filled the ring: %+v", p.ring) |
| } |
| } |
| @@ -168,6 +189,38 @@ func TestResumeGapWhenAnotherSendersBurstEvictedTheWholeSender(t *testing.T) { |
| } |
| } |
| +// The ring and the eviction marks outlive the peer that filled them. |
| +// A resume must not name a peer that has left: the roster the browser |
| +// reconciles against says it is gone, so a gap for it would ask for a |
| +// reset with nobody, and its replayed signalling would rebuild the |
| +// connection the roster has just told the browser to close. |
| +func TestResumeNamesNoPeerThatHasLeft(t *testing.T) { |
| + h, now, a, b := twoAdmitted(t) |
| + for i := uint64(1); i <= 40; i++ { |
| + h.Relay("r1", "sb", b.Peer, Message{Type: TypeCandidate, To: a.Peer, Seq: i, Gen: 1}) |
| + } |
| + p, _ := h.Find("r1", "sa", a.Peer) |
| + if p.evicted[b.Peer] == 0 { |
| + t.Fatal("the ring did not evict anything, so this proves nothing") |
| + } |
| + if err := h.Leave("r1", "sb", b.Peer); err != nil { |
| + t.Fatal(err) |
| + } |
| + drain(a.Sub) |
| + h.Detach("r1", a.Peer, nil) |
| + *now = now.Add(time.Second) |
| + res, err := h.Resume("r1", "sa", a.Peer, map[PeerID]uint64{b.Peer: 2}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(res.Gap) != 0 { |
| + t.Fatalf("gap names a peer that left: %v", res.Gap) |
| + } |
| + if len(res.Replay) != 0 { |
| + t.Fatalf("replay carries a departed peer's signalling: %+v", res.Replay) |
| + } |
| +} |
| + |
| // A second resume of an attached peer replaces the attachment: the old |
| // stream is told replaced, and the participation itself carries on. |
| func TestResumeReplacesAnExistingAttachment(t *testing.T) { |
| @@ -222,6 +275,46 @@ func TestResumeAnswers(t *testing.T) { |
| } |
| } |
| +// Why a participation ended is account state, and a peer id is in every |
| +// roster the room-mates hold. A tombstone is therefore answered only |
| +// through the room and session it was bound to; anyone else gets the |
| +// same 404 an unknown peer gets (spec §4.2). |
| +func TestATombstoneIsAnsweredOnlyThroughItsOwnRoomAndSession(t *testing.T) { |
| + h, _, a, b := twoAdmitted(t) |
| + if _, err := h.Open("r2", "sother", 9, "Other", KindMember); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := h.Remove(context.Background(), "r1", "sb", b.Peer, a.Peer); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + // Ann's own session, in the room she was removed from. |
| + _, err := h.Resume("r1", "sa", a.Peer, nil) |
| + re, ok := err.(*ResumeError) |
| + if !ok || re.Status != 410 || re.Reason != ReasonRemoved { |
| + t.Fatalf("her own resume: %v", err) |
| + } |
| + if re := h.CanResume("r1", "sa", a.Peer); re == nil || re.Status != 410 || re.Reason != ReasonRemoved { |
| + t.Fatalf("her own probe: %v", re) |
| + } |
| + |
| + // Bob, who has her peer id from the roster, learns nothing. |
| + for _, probe := range []struct{ what, sqid, session string }{ |
| + {"another session in the same room", "r1", "sb"}, |
| + {"another room this session is in", "r2", "sother"}, |
| + {"another room and another session", "r2", "sb"}, |
| + } { |
| + _, err := h.Resume(probe.sqid, probe.session, a.Peer, nil) |
| + re, ok := err.(*ResumeError) |
| + if !ok || re.Status != 404 || re.Boot != h.Boot() || re.Reason != "" { |
| + t.Fatalf("%s: %v", probe.what, err) |
| + } |
| + if re := h.CanResume(probe.sqid, probe.session, a.Peer); re == nil || re.Status != 404 || re.Reason != "" { |
| + t.Fatalf("%s, probed: %+v", probe.what, re) |
| + } |
| + } |
| +} |
| + |
| func TestResumeAfterWindowIsA404WithTheSameBoot(t *testing.T) { |
| h, now, a, _ := twoAdmitted(t) |
| h.Detach("r1", a.Peer, nil) |
diff --git a/internal/meettest/stream_test.go b/internal/meettest/stream_test.go| index f75384d..2c02c24 100644 |
| --- a/internal/meettest/stream_test.go |
| +++ b/internal/meettest/stream_test.go |
| @@ -261,11 +261,10 @@ func TestEndingTheMeetingClosesTheStreamAndRefusesWhatComesAfter(t *testing.T) { |
| if rec := c.get("/m/" + quiet + "/stream"); rec.Code != http.StatusNotFound { |
| t.Fatalf("fresh stream on a meeting nobody opened: %d, want 404", rec.Code) |
| } |
| - // A peer id the hub has never heard of, rather than the tombstoned |
| - // one above: a tombstone is not scoped to its room yet (the Task 3 |
| - // review's first Important), so the ended peer would be answered |
| - // 410 here, in a room it was never in. |
| - rec = c.get("/m/" + quiet + "/stream?peer=fedcba9876543210fedcba9876543210") |
| + // The peer the ending tombstoned above, offered to a room it was |
| + // never in: a tombstone answers only through its own room and |
| + // session, so this is the plain 404 an unknown peer gets. |
| + rec = c.get("/m/" + quiet + "/stream?peer=" + peer) |
| if _, gotBoot, _ := answer(t, rec); rec.Code != http.StatusNotFound || gotBoot != boot { |
| t.Fatalf("resume into a meeting nobody opened: %d boot %q", rec.Code, gotBoot) |
| } |