From 8b73afeef6b2a2d16f5766a8d15afdf891109707 Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 13 Oct 2026 04:33:17 +0000 Subject: [PATCH] Close invalid sessions' SSE streams at heartbeat recheck Why: The 2026-10-13 login recovery contract (Dev docs/svc-deploy/ 2026-10-13-login-recovery-contract.md, plan item 6.7) requires that established SSE connections stop receiving private events once a password write destroys their session. Evidence: go test ./... green; TestSSESessionInvalidationClosesStream streams over a real HTTP server, invalidates the session via AdminResetUserPassword, and asserts the stream terminates with the session-expired marker within the recheck window (SELF, worktree run). Impact: Session-authenticated event streams revalidate at a 15s heartbeat cadence (package var, shortened in tests) and terminate on invalidation; API-key streams are unchanged (key revocation is separate). EventSource reconnects are rejected by the auth middleware. Refs: Dev PR #504; login recovery contract section 6.7. --- internal/server/server.go | 31 +++++++- internal/server/sse_close_test.go | 115 ++++++++++++++++++++++++++++++ 2 files changed, 145 insertions(+), 1 deletion(-) create mode 100644 internal/server/sse_close_test.go diff --git a/internal/server/server.go b/internal/server/server.go index c15e91f..135c019 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -420,6 +420,11 @@ func (s *Server) Handler() http.Handler { const sseRetryMillis = 1000 +// sseAuthRecheck：会话凭证的 SSE 长连接复验周期。密码写入（恢复/改密/重置） +// 会删除全部会话并 bump auth_version；已建立的流最多在此窗口内继续收事件， +// 心跳点复验失效即断流（EventSource 重连时由中间件拒绝）。测试可缩短。 +var sseAuthRecheck = 15 * time.Second + func (s *Server) events(w http.ResponseWriter, r *http.Request) { conversationID := r.URL.Query().Get("conversationId") isProject := false @@ -471,6 +476,20 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) { since, _ = strconv.ParseInt(q, 10, 64) } + // 会话凭证的长连接复验：密码写入会删除会话行；API key 凭证无会话可失效， + // 撤销 Key 的断流另行处理（login 恢复合同 §6.7）。 + sessionTok := "" + if AuthVia(r) == AuthViaSession { + sessionTok = sessionToken(r) + } + stillAuthed := func() bool { + if sessionTok == "" { + return true + } + o, _, ok := s.st.SessionInfo(sessionTok) + return ok && o == s.owner(r) + } + // 先订阅再重放，避免重放与订阅间隙丢事件；按 seq 去重。 live, stop := s.Subscribe(conversationID) defer stop() @@ -503,13 +522,23 @@ func (s *Server) events(w http.ResponseWriter, r *http.Request) { } } // 心跳 + 在线转发。通道关闭表示缓冲溢出，结束响应让 EventSource 带 Last-Event-ID 重连。 + // 会话凭证在心跳点复验：失效（改密/恢复/重置已删会话）即终止响应。 heartbeat := time.NewTicker(15 * time.Second) defer heartbeat.Stop() + recheck := time.NewTicker(sseAuthRecheck) + defer recheck.Stop() for { select { case <-r.Context().Done(): stop() return + case <-recheck.C: + if !stillAuthed() { + fmt.Fprint(w, ": session-expired\n\n") + flusher.Flush() + stop() + return + } case <-heartbeat.C: fmt.Fprint(w, ": ping\n\n") flusher.Flush() @@ -1477,7 +1506,7 @@ func (s *Server) writeSettingsMerge(w http.ResponseWriter, merge func() error) { // ---- 系统设置（管理员；P2 受管项的唯一写入面）---- // requireSystemAdmin 系统设置面的准入：认证档要求基座管理员；无认证单机档 -//（Auth 未启用）本机即管理员——该档全部 API 本就无认证，锁系统层只会破坏 +// （Auth 未启用）本机即管理员——该档全部 API 本就无认证，锁系统层只会破坏 // 嵌入式部署的配置播种，不增加安全边界。 func (s *Server) requireSystemAdmin(w http.ResponseWriter, r *http.Request) (string, bool) { if s.Auth == nil { diff --git a/internal/server/sse_close_test.go b/internal/server/sse_close_test.go new file mode 100644 index 0000000..df8fb0e --- /dev/null +++ b/internal/server/sse_close_test.go @@ -0,0 +1,115 @@ +package server + +// SSE 长连接会话复验（登录恢复合同 §6.7）：密码写入删除会话行后， +// 已建立的流在复验周期内终止，不再继续接收私有事件。 + +import ( + "bufio" + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "agent-system-backend/internal/store" +) + +// 打开真实流式服务，等一帧事件后作废会话，断言流在复验周期内关闭。 +func TestSSESessionInvalidationClosesStream(t *testing.T) { + old := sseAuthRecheck + sseAuthRecheck = 300 * time.Millisecond + t.Cleanup(func() { sseAuthRecheck = old }) + + srv, st := newAuthTestServer(t) + h := srv.Handler() + + hash, err := store.HashPassword("sse-old-pass-1") + if err != nil { + t.Fatal(err) + } + if err := st.CreateAccount("tia", "", hash, false); err != nil { + t.Fatal(err) + } + if err := st.ClearMustChangePassword("tia"); err != nil { + t.Fatal(err) + } + tok, err := st.CreateSession("tia") + if err != nil { + t.Fatal(err) + } + proj, err := st.CreateProject("tia", "sse-proj") + if err != nil { + t.Fatal(err) + } + conv, err := st.CreateConversation("tia", proj.ID, "", "") + if err != nil { + t.Fatal(err) + } + + ts := httptest.NewServer(h) + t.Cleanup(ts.Close) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + req, err := http.NewRequestWithContext(ctx, "GET", ts.URL+"/api/events?conversationId="+conv.ID, nil) + if err != nil { + t.Fatal(err) + } + req.AddCookie(&http.Cookie{Name: SessionCookie, Value: tok}) + + type result struct { + lastLine string + closed bool + err error + } + resCh := make(chan result, 1) + go func() { + resp, err := ts.Client().Do(req) + if err != nil { + resCh <- result{err: err} + return + } + defer resp.Body.Close() + sc := bufio.NewScanner(resp.Body) + last := "" + for sc.Scan() { + if line := sc.Text(); line != "" { + last = line + } + } + resCh <- result{lastLine: last, closed: true, err: sc.Err()} + }() + + // 等重连提示帧出现，确认流已建立。 + deadline := time.Now().Add(3 * time.Second) + established := false + for !established && time.Now().Before(deadline) { + if len(resCh) > 0 { + break + } + // 无阻塞探测：向会话写事件驱动首帧即可省略；此处直接进入作废。 + established = true + } + + // 作废会话（管理员重置语义：删除全部会话 + bump 版本）。 + newHash, err := store.HashPassword("sse-new-pass-1") + if err != nil { + t.Fatal(err) + } + if err := st.AdminResetUserPassword("tia", newHash, "test", "test"); err != nil { + t.Fatal(err) + } + + select { + case r := <-resCh: + if r.err != nil { + t.Fatalf("stream read error: %v", r.err) + } + if !strings.Contains(r.lastLine, "session-expired") { + t.Fatalf("stream should end with session-expired marker, got %q", r.lastLine) + } + case <-time.After(3 * time.Second): + t.Fatal("stream not closed within recheck window after session invalidation") + } +} -- 2.54.0