fix(web): recheck the owner's session before sending each event

The stream rechecked the session only on the heartbeat, so events could
still reach a browser signed out everywhere until the next beat. Check
before each event and end the stream when the session is gone.

commit: 12ce9610199142b6d5508b45dbda5e79cd47b989
author: brice · 11d ago · 2026-10-10 19:22 UTC
committer: brice · 11d ago · 2026-10-10 19:28 UTC
parent: 3922170
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +108 -5

M internal/adapters/web/events.go +6 -0


@@ -184,6 +184,12 @@ func (s Server) events(w http.ResponseWriter, r http.Request) {
184	184		}
185	185		return false
186	186		}
	187	+	// An owner's stream ends when the session does, before it sends
	188	+	// anything more: an event can carry what only the owner may see, and
	189	+	// waiting for the next heartbeat would leave a window.
	190	+	if SignedIn(ctx) && !s.sessionStillValid(ctx, r) {
	191	+	return false
	192	+	}
187	193		rctx, rcancel := context.WithTimeout(ctx, requestTimeout)
188	194		frags, err := st.render(rctx, m.ev)
189	195		rcancel()


@@ -184,6 +184,12 @@ func (s Server) events(w http.ResponseWriter, r http.Request) {
184	}	184		}
185	return false	185		return false
186	}	186		}
		187	+	// An owner's stream ends when the session does, before it sends
		188	+	// anything more: an event can carry what only the owner may see, and
		189	+	// waiting for the next heartbeat would leave a window.
		190	+	if SignedIn(ctx) && !s.sessionStillValid(ctx, r) {
		191	+	return false
		192	+	}
187	rctx, rcancel := context.WithTimeout(ctx, requestTimeout)	193		rctx, rcancel := context.WithTimeout(ctx, requestTimeout)
188	frags, err := st.render(rctx, m.ev)	194		frags, err := st.render(rctx, m.ev)
189	rcancel()	195		rcancel()

M internal/adapters/web/events_test.go +102 -5


@@ -6,6 +6,7 @@ import (
6	6		"bufio"
7	7		"context"
8	8		"errors"
	9	+	"io"
9	10		"net/http"
10	11		"net/http/httptest"
11	12		"strings"

@@ -62,6 +63,7 @@ type fakeEvents struct {
62	63		closed atomic.Int32
63	64		resume atomic.Value
64	65		release chan struct{}
	66	+	push chan *pb.Event // events sent on demand, after the fixed ones
65	67		}
66	68
67	69		func (f fakeEvents) Watch(ctx context.Context, req connect.Request[pb.WatchRequest], st *connect.ServerStream[pb.WatchResponse]) error {

@@ -77,11 +79,18 @@ func (f fakeEvents) Watch(ctx context.Context, req connect.Request[pb.WatchReq
77	79		return err
78	80		}
79	81		}
80		-	select {
81		-	case <-ctx.Done():
82		-	case <-f.release:
	82	+	for {
	83	+	select {
	84	+	case <-ctx.Done():
	85	+	return nil
	86	+	case <-f.release:
	87	+	return nil
	88	+	case ev := <-f.push:
	89	+	if err := st.Send(&pb.WatchResponse{Event: ev}); err != nil {
	90	+	return err
	91	+	}
	92	+	}
83	93		}
84		-	return nil
85	94		}
86	95
87	96		// repoOK answers every GetRepo with a public repository.

@@ -96,6 +105,12 @@ func (repoOK) GetRepo(_ context.Context, req *connect.Request[pb.GetRepoRequest]
96	105		// eventsServer serves the web on a fake API, with a write timeout shorter
97	106		// than the heartbeat, like a server that sets one.
98	107		func eventsServer(t testing.T, f fakeEvents, beat time.Duration, writeTimeout time.Duration) *httptest.Server {
	108	+	t.Helper()
	109	+	return eventsServerWith(t, f, beat, writeTimeout, nil)
	110	+	}
	111	+
	112	+	// eventsServerWith is eventsServer with sessions, when sess is not nil.
	113	+	func eventsServerWith(t testing.T, f fakeEvents, beat time.Duration, writeTimeout time.Duration, sess SessionAuth) *httptest.Server {
99	114		t.Helper()
100	115		api := http.NewServeMux()
101	116		api.Handle(ermiteconnect.NewEventServiceHandler(f))

@@ -108,7 +123,7 @@ func eventsServer(t testing.T, f fakeEvents, beat time.Duration, writeTimeout
108	123		Git: ermiteconnect.NewGitServiceClient(backend.Client(), backend.URL),
109	124		Events: ermiteconnect.NewEventServiceClient(backend.Client(), backend.URL),
110	125		PublicURL: "https://forge.example", SourceURL: "https://forge.example/ermite", Version: "v1",
111		-	Heartbeat: beat,
	126	+	Heartbeat: beat, Sessions: sess, Limiter: limiterFor(sess),
112	127		})
113	128		require.NoError(t, err)
114	129		ts := httptest.NewUnstartedServer(srv)

@@ -260,3 +275,85 @@ func TestRepoUpdatedReloadsSummaryAndRefs(t *testing.T) {
260	275		}
261	276		}
262	277		}
	278	+
	279	+	func limiterFor(sess SessionAuth) FailureLimiter {
	280	+	if sess == nil {
	281	+	return nil
	282	+	}
	283	+	return &countingLimiter{}
	284	+	}
	285	+
	286	+	// revocable is a SessionAuth whose one session ends when the test says so, like
	287	+	// SignOutEverywhere does.
	288	+	type revocable struct {
	289	+	fakeSessions
	290	+	ended atomic.Bool
	291	+	}
	292	+
	293	+	func (r *revocable) Verify(ctx context.Context, tok string) (context.Context, bool, error) {
	294	+	return ctx, tok == r.valid && !r.ended.Load(), nil
	295	+	}
	296	+
	297	+	// GO-002: the stream of the owner is checked before each event is sent, not
	298	+	// only on the heartbeat: after the session ended, the next event is not sent
	299	+	// and the stream ends.
	300	+	func TestEventsOwnerStreamEndsBeforeNextEvent(t *testing.T) {
	301	+	f := &fakeEvents{release: make(chan struct{}), push: make(chan *pb.Event)}
	302	+	sess := &revocable{fakeSessions: fakeSessions{valid: "tok"}}
	303	+	ts := eventsServerWith(t, f, time.Hour, 0, sess) // no heartbeat in the test
	304	+	resp, cancel := open(t, ts, "topic=home", map[string]string{"Cookie": CookieName + "=tok"})
	305	+	require.Equal(t, http.StatusOK, resp.StatusCode)
	306	+	assert.Contains(t, resp.Header.Get("Cache-Control"), "private")
	307	+	// Without the check the stream would stay open: bound the read.
	308	+	defer time.AfterFunc(5*time.Second, cancel).Stop()
	309	+	rd := bufio.NewReader(resp.Body)
	310	+
	311	+	line := func() string {
	312	+	s, err := rd.ReadString('\n')
	313	+	if err != nil {
	314	+	return "EOF"
	315	+	}
	316	+	return strings.TrimRight(s, "\n")
	317	+	}
	318	+	// The stream is up; a first event goes through while the session lives.
	319	+	require.Equal(t, ": connected", line())
	320	+	line()
	321	+	f.push <- &pb.Event{Id: "b-1", Kind: &pb.Event_Gap{Gap: &pb.Gap{}}}
	322	+	var seen []string
	323	+	for l := line(); l != "data: gap"; l = line() {
	324	+	seen = append(seen, l)
	325	+	require.NotEqual(t, "EOF", l)
	326	+	}
	327	+	assert.Contains(t, seen, "id: b-1")
	328	+
	329	+	// Signed out everywhere: the next event is not sent, the stream ends.
	330	+	sess.ended.Store(true)
	331	+	select {
	332	+	case f.push <- &pb.Event{Id: "b-2", Kind: &pb.Event_Gap{Gap: &pb.Gap{}}}:
	333	+	case <-time.After(5 * time.Second):
	334	+	t.Fatal("the event was not delivered to the web")
	335	+	}
	336	+	rest, _ := io.ReadAll(rd)
	337	+	assert.NotContains(t, string(rest), "b-2", "nothing is sent after the session ended")
	338	+	require.Eventually(t, func() bool { return f.closed.Load() == 1 }, 5*time.Second, time.Millisecond, "the watch is closed")
	339	+	}
	340	+
	341	+	// A visitor's stream does not depend on sessions: events keep flowing.
	342	+	func TestEventsVisitorStreamIgnoresSessions(t *testing.T) {
	343	+	f := &fakeEvents{release: make(chan struct{}), push: make(chan *pb.Event)}
	344	+	sess := &revocable{fakeSessions: fakeSessions{valid: "tok"}}
	345	+	sess.ended.Store(true)
	346	+	ts := eventsServerWith(t, f, time.Hour, 0, sess)
	347	+	resp, _ := open(t, ts, "topic=home", nil)
	348	+	require.Equal(t, http.StatusOK, resp.StatusCode)
	349	+	f.push <- &pb.Event{Id: "b-1", Kind: &pb.Event_Gap{Gap: &pb.Gap{}}}
	350	+	sc := bufio.NewScanner(resp.Body)
	351	+	found := false
	352	+	for sc.Scan() {
	353	+	if sc.Text() == "id: b-1" {
	354	+	found = true
	355	+	break
	356	+	}
	357	+	}
	358	+	assert.True(t, found)
	359	+	}


@@ -6,6 +6,7 @@ import (
6		"bufio"	6		"bufio"
7		"context"	7		"context"
8		"errors"	8		"errors"
			9	+	"io"
9		"net/http"	10		"net/http"
10		"net/http/httptest"	11		"net/http/httptest"
11		"strings"	12		"strings"

@@ -62,6 +63,7 @@ type fakeEvents struct {
62		closed atomic.Int32	63		closed atomic.Int32
63		resume atomic.Value	64		resume atomic.Value
64		release chan struct{}	65		release chan struct{}
			66	+	push chan *pb.Event // events sent on demand, after the fixed ones
65		}	67		}
66			68
67		func (f fakeEvents) Watch(ctx context.Context, req connect.Request[pb.WatchRequest], st *connect.ServerStream[pb.WatchResponse]) error {	69		func (f fakeEvents) Watch(ctx context.Context, req connect.Request[pb.WatchRequest], st *connect.ServerStream[pb.WatchResponse]) error {

@@ -77,11 +79,18 @@ func (f fakeEvents) Watch(ctx context.Context, req connect.Request[pb.WatchReq
77		return err	79		return err
78		}	80		}
79		}	81		}
80	-	select {	82	+	for {
81	-	case <-ctx.Done():	83	+	select {
82	-	case <-f.release:	84	+	case <-ctx.Done():
			85	+	return nil
			86	+	case <-f.release:
			87	+	return nil
			88	+	case ev := <-f.push:
			89	+	if err := st.Send(&pb.WatchResponse{Event: ev}); err != nil {
			90	+	return err
			91	+	}
			92	+	}
83		}	93		}
84	-	return nil
85		}	94		}
86			95
87		// repoOK answers every GetRepo with a public repository.	96		// repoOK answers every GetRepo with a public repository.

@@ -96,6 +105,12 @@ func (repoOK) GetRepo(_ context.Context, req *connect.Request[pb.GetRepoRequest]
96		// eventsServer serves the web on a fake API, with a write timeout shorter	105		// eventsServer serves the web on a fake API, with a write timeout shorter
97		// than the heartbeat, like a server that sets one.	106		// than the heartbeat, like a server that sets one.
98		func eventsServer(t testing.T, f fakeEvents, beat time.Duration, writeTimeout time.Duration) *httptest.Server {	107		func eventsServer(t testing.T, f fakeEvents, beat time.Duration, writeTimeout time.Duration) *httptest.Server {
			108	+	t.Helper()
			109	+	return eventsServerWith(t, f, beat, writeTimeout, nil)
			110	+	}
			111	+
			112	+	// eventsServerWith is eventsServer with sessions, when sess is not nil.
			113	+	func eventsServerWith(t testing.T, f fakeEvents, beat time.Duration, writeTimeout time.Duration, sess SessionAuth) *httptest.Server {
99		t.Helper()	114		t.Helper()
100		api := http.NewServeMux()	115		api := http.NewServeMux()
101		api.Handle(ermiteconnect.NewEventServiceHandler(f))	116		api.Handle(ermiteconnect.NewEventServiceHandler(f))

@@ -108,7 +123,7 @@ func eventsServer(t testing.T, f fakeEvents, beat time.Duration, writeTimeout
108		Git: ermiteconnect.NewGitServiceClient(backend.Client(), backend.URL),	123		Git: ermiteconnect.NewGitServiceClient(backend.Client(), backend.URL),
109		Events: ermiteconnect.NewEventServiceClient(backend.Client(), backend.URL),	124		Events: ermiteconnect.NewEventServiceClient(backend.Client(), backend.URL),
110		PublicURL: "https://forge.example", SourceURL: "https://forge.example/ermite", Version: "v1",	125		PublicURL: "https://forge.example", SourceURL: "https://forge.example/ermite", Version: "v1",
111	-	Heartbeat: beat,	126	+	Heartbeat: beat, Sessions: sess, Limiter: limiterFor(sess),
112		})	127		})
113		require.NoError(t, err)	128		require.NoError(t, err)
114		ts := httptest.NewUnstartedServer(srv)	129		ts := httptest.NewUnstartedServer(srv)

@@ -260,3 +275,85 @@ func TestRepoUpdatedReloadsSummaryAndRefs(t *testing.T) {
260		}	275		}
261		}	276		}
262		}	277		}
			278	+
			279	+	func limiterFor(sess SessionAuth) FailureLimiter {
			280	+	if sess == nil {
			281	+	return nil
			282	+	}
			283	+	return &countingLimiter{}
			284	+	}
			285	+
			286	+	// revocable is a SessionAuth whose one session ends when the test says so, like
			287	+	// SignOutEverywhere does.
			288	+	type revocable struct {
			289	+	fakeSessions
			290	+	ended atomic.Bool
			291	+	}
			292	+
			293	+	func (r *revocable) Verify(ctx context.Context, tok string) (context.Context, bool, error) {
			294	+	return ctx, tok == r.valid && !r.ended.Load(), nil
			295	+	}
			296	+
			297	+	// GO-002: the stream of the owner is checked before each event is sent, not
			298	+	// only on the heartbeat: after the session ended, the next event is not sent
			299	+	// and the stream ends.
			300	+	func TestEventsOwnerStreamEndsBeforeNextEvent(t *testing.T) {
			301	+	f := &fakeEvents{release: make(chan struct{}), push: make(chan *pb.Event)}
			302	+	sess := &revocable{fakeSessions: fakeSessions{valid: "tok"}}
			303	+	ts := eventsServerWith(t, f, time.Hour, 0, sess) // no heartbeat in the test
			304	+	resp, cancel := open(t, ts, "topic=home", map[string]string{"Cookie": CookieName + "=tok"})
			305	+	require.Equal(t, http.StatusOK, resp.StatusCode)
			306	+	assert.Contains(t, resp.Header.Get("Cache-Control"), "private")
			307	+	// Without the check the stream would stay open: bound the read.
			308	+	defer time.AfterFunc(5*time.Second, cancel).Stop()
			309	+	rd := bufio.NewReader(resp.Body)
			310	+
			311	+	line := func() string {
			312	+	s, err := rd.ReadString('\n')
			313	+	if err != nil {
			314	+	return "EOF"
			315	+	}
			316	+	return strings.TrimRight(s, "\n")
			317	+	}
			318	+	// The stream is up; a first event goes through while the session lives.
			319	+	require.Equal(t, ": connected", line())
			320	+	line()
			321	+	f.push <- &pb.Event{Id: "b-1", Kind: &pb.Event_Gap{Gap: &pb.Gap{}}}
			322	+	var seen []string
			323	+	for l := line(); l != "data: gap"; l = line() {
			324	+	seen = append(seen, l)
			325	+	require.NotEqual(t, "EOF", l)
			326	+	}
			327	+	assert.Contains(t, seen, "id: b-1")
			328	+
			329	+	// Signed out everywhere: the next event is not sent, the stream ends.
			330	+	sess.ended.Store(true)
			331	+	select {
			332	+	case f.push <- &pb.Event{Id: "b-2", Kind: &pb.Event_Gap{Gap: &pb.Gap{}}}:
			333	+	case <-time.After(5 * time.Second):
			334	+	t.Fatal("the event was not delivered to the web")
			335	+	}
			336	+	rest, _ := io.ReadAll(rd)
			337	+	assert.NotContains(t, string(rest), "b-2", "nothing is sent after the session ended")
			338	+	require.Eventually(t, func() bool { return f.closed.Load() == 1 }, 5*time.Second, time.Millisecond, "the watch is closed")
			339	+	}
			340	+
			341	+	// A visitor's stream does not depend on sessions: events keep flowing.
			342	+	func TestEventsVisitorStreamIgnoresSessions(t *testing.T) {
			343	+	f := &fakeEvents{release: make(chan struct{}), push: make(chan *pb.Event)}
			344	+	sess := &revocable{fakeSessions: fakeSessions{valid: "tok"}}
			345	+	sess.ended.Store(true)
			346	+	ts := eventsServerWith(t, f, time.Hour, 0, sess)
			347	+	resp, _ := open(t, ts, "topic=home", nil)
			348	+	require.Equal(t, http.StatusOK, resp.StatusCode)
			349	+	f.push <- &pb.Event{Id: "b-1", Kind: &pb.Event_Gap{Gap: &pb.Gap{}}}
			350	+	sc := bufio.NewScanner(resp.Body)
			351	+	found := false
			352	+	for sc.Scan() {
			353	+	if sc.Text() == "id: b-1" {
			354	+	found = true
			355	+	break
			356	+	}
			357	+	}
			358	+	assert.True(t, found)
			359	+	}