weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(api): require project membership before chat and chat SSE

Browse Source 
Unrestricted stage access on projects without a live workflow must not
let arbitrary authenticated users join by project_id. Enforce member or
creator (admin per existing policy) before stage ABAC, and reject chat
streams with JSON 403 before SSE headers/flush.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-13 12:25:22 +00:00
co-authored by Cursor
parent f50692be25
commit cc394c815c
4 changed files with 205 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/server/chat_project_access_authz_test.go
+148
View File
@@ -0,0 +1,148 @@
package server_test
import (
"bytes"
"encoding/json"
"net/http"
"net/http/httptest"
"strconv"
"strings"
"testing"
"time"
"chat-agent-system/internal/projectworkflow"
)
// 与 tasks/run/stream 对齐：无-live 项目上非成员不得因 stage unrestricted 而进入对话；
// chat/stream 拒绝必须发生在 SSE headers/Flush 之前（结构化 JSON 403）。
func TestChatProjectAccess_NonMemberDeniedBeforeSSE_PG(t *testing.T) {
pool := openPGPoolWithMigrations(t)
if pool == nil {
return
}
defer pool.Close()
handler, _ := setupPGGovernanceServer(t, pool)
creatorToken := loginTestUser(t, handler, "zhanggong", "123456")
outsiderToken := loginTestUser(t, handler, "lizhuguan", "123456")
createBody, _ := json.Marshal(projectworkflow.CreateProjectInput{Name: "Chat Access NoLive " + time.Now().Format("150405.000")})
reqCreate := httptest.NewRequest(http.MethodPost, "/api/v1/projects", bytes.NewReader(createBody))
withBearerAuth(reqCreate, creatorToken)
recCreate := httptest.NewRecorder()
handler.ServeHTTP(recCreate, reqCreate)
if recCreate.Code != http.StatusCreated {
t.Fatalf("create project: %d %s", recCreate.Code, recCreate.Body.String())
}
var proj projectworkflow.Project
_ = json.Unmarshal(recCreate.Body.Bytes(), &proj)
t.Cleanup(func() { cleanupProject(t, pool, proj.ID) })
syncBody, _ := json.Marshal(map[string]any{
"session_id": "sess-chat-outsider-sync",
"message": "首轮 intake 试探",
"project_id": proj.ID,
})
reqSync := httptest.NewRequest(http.MethodPost, "/api/v1/chat", bytes.NewReader(syncBody))
withBearerAuth(reqSync, outsiderToken)
recSync := httptest.NewRecorder()
handler.ServeHTTP(recSync, reqSync)
if recSync.Code != http.StatusForbidden {
t.Fatalf("outsider sync chat expected 403, got %d: %s", recSync.Code, recSync.Body.String())
}
if !strings.Contains(recSync.Body.String(), "ERR_FORBIDDEN") {
t.Fatalf("expected ERR_FORBIDDEN json, got %s", recSync.Body.String())
}
streamBody, _ := json.Marshal(map[string]any{
"session_id": "sess-chat-outsider-stream",
"message": "首轮 intake 流式",
"project_id": proj.ID,
})
reqStream := httptest.NewRequest(http.MethodPost, "/api/v1/chat/stream", bytes.NewReader(streamBody))
withBearerAuth(reqStream, outsiderToken)
recStream := httptest.NewRecorder()
handler.ServeHTTP(recStream, reqStream)
if recStream.Code != http.StatusForbidden {
t.Fatalf("outsider chat/stream expected 403 (not 200 SSE), got %d body=%s", recStream.Code, recStream.Body.String())
}
ct := recStream.Header().Get("Content-Type")
if strings.Contains(ct, "text/event-stream") {
t.Fatalf("forbidden chat/stream must not set text/event-stream before reject, got %q", ct)
}
if !strings.Contains(recStream.Body.String(), "ERR_FORBIDDEN") {
t.Fatalf("expected structured ERR_FORBIDDEN before flush, got %s", recStream.Body.String())
}
}
func TestChatProjectAccess_CreatorAndMemberAllowed_PG(t *testing.T) {
pool := openPGPoolWithMigrations(t)
if pool == nil {
return
}
defer pool.Close()
handler, _ := setupPGGovernanceServer(t, pool)
creatorToken := loginTestUser(t, handler, "zhanggong", "123456")
memberToken := loginTestUser(t, handler, "lizhuguan", "123456")
memberID := fetchAuthUserID(t, handler, memberToken)
createBody, _ := json.Marshal(projectworkflow.CreateProjectInput{Name: "Chat Access Member " + time.Now().Format("150405.000")})
reqCreate := httptest.NewRequest(http.MethodPost, "/api/v1/projects", bytes.NewReader(createBody))
withBearerAuth(reqCreate, creatorToken)
recCreate := httptest.NewRecorder()
handler.ServeHTTP(recCreate, reqCreate)
if recCreate.Code != http.StatusCreated {
t.Fatalf("create: %d %s", recCreate.Code, recCreate.Body.String())
}
var proj projectworkflow.Project
_ = json.Unmarshal(recCreate.Body.Bytes(), &proj)
t.Cleanup(func() { cleanupProject(t, pool, proj.ID) })
addBody, _ := json.Marshal(map[string]any{"user_id": memberID, "project_role": "member"})
reqAdd := httptest.NewRequest(http.MethodPost, "/api/v1/projects/"+strconv.FormatInt(proj.ID, 10)+"/members", bytes.NewReader(addBody))
withBearerAuth(reqAdd, creatorToken)
recAdd := httptest.NewRecorder()
handler.ServeHTTP(recAdd, reqAdd)
if recAdd.Code != http.StatusCreated && recAdd.Code != http.StatusOK {
t.Fatalf("add member: %d %s", recAdd.Code, recAdd.Body.String())
}
for _, tc := range []struct {
name string
token string
}{
{"creator", creatorToken},
{"member", memberToken},
} {
body, _ := json.Marshal(map[string]any{
"session_id": "sess-chat-ok-" + tc.name,
"message": "hello",
"project_id": proj.ID,
})
req := httptest.NewRequest(http.MethodPost, "/api/v1/chat", bytes.NewReader(body))
withBearerAuth(req, tc.token)
rec := httptest.NewRecorder()
handler.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("%s sync chat want 200 got %d %s", tc.name, rec.Code, rec.Body.String())
}
sBody, _ := json.Marshal(map[string]any{
"session_id": "sess-chat-ok-stream-" + tc.name,
"message": "hello stream",
"project_id": proj.ID,
})
sReq := httptest.NewRequest(http.MethodPost, "/api/v1/chat/stream", bytes.NewReader(sBody))
withBearerAuth(sReq, tc.token)
sRec := httptest.NewRecorder()
handler.ServeHTTP(sRec, sReq)
if sRec.Code != http.StatusOK {
t.Fatalf("%s chat/stream want 200 got %d %s", tc.name, sRec.Code, sRec.Body.String())
}
if !strings.Contains(sRec.Header().Get("Content-Type"), "text/event-stream") {
t.Fatalf("%s chat/stream should be SSE after auth, ct=%q", tc.name, sRec.Header().Get("Content-Type"))
}
}
}
internal/server/handlers.go
+5
View File
@@ -173,6 +173,11 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) {
if req.ProjectID > 0 {
sess.ProjectID = req.ProjectID
}
if sess.ProjectID > 0 {
if !s.enforceChatProjectAccess(w, r, user, sess.ProjectID) {
return
}
}
s.enrichSessionWorkflowStage(r.Context(), sess, user.ID)
if sess.ProjectID > 0 && !s.allowActiveStageChatAccess(r.Context(), sess.ProjectID, user.ID) {
s.writeWorkflowAccessDenied(w)
internal/server/sse.go
+22 -13
View File
@@ -135,6 +135,9 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) {
req.WorkDir = q.Get("work_dir")
req.UserName = q.Get("user_name")
req.UserRole = q.Get("user_role")
if pid := q.Get("project_id"); pid != "" {
req.ProjectID, _ = strconv.ParseInt(pid, 10, 64)
}
if inc := q.Get("include_details"); inc != "" {
req.IncludeDetails, _ = strconv.ParseBool(inc)
}
@@ -152,19 +155,7 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) {
req.UserID = user.ID
req.WorkDir = ""
// 1. 设置 SSE 响应头 + 请求追踪 ID
requestID := fmt.Sprintf("req_%d", time.Now().UnixNano())
w.Header().Set("X-Request-Id", requestID)
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
s.writeCORS(w, r)
w.Header().Set("X-Accel-Buffering", "no") // 禁用 Nginx 等反代缓冲
flusher.Flush()
ctx := stream.WithRequestID(r.Context(), requestID)
// 2. 准备会话
// 会话与项目绑定必须在 SSE headers/Flush 之前完成授权（与 tasks/run/stream 对齐）
sess := s.getOrCreateSession(
req.SessionID,
req.ProjectName,
@@ -180,6 +171,24 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) {
if req.ProjectID > 0 {
sess.ProjectID = req.ProjectID
}
if sess.ProjectID > 0 {
if !s.enforceChatProjectAccess(w, r, user, sess.ProjectID) {
return
}
}
// 1. 设置 SSE 响应头 + 请求追踪 ID（仅授权通过后）
requestID := fmt.Sprintf("req_%d", time.Now().UnixNano())
w.Header().Set("X-Request-Id", requestID)
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
s.writeCORS(w, r)
w.Header().Set("X-Accel-Buffering", "no") // 禁用 Nginx 等反代缓冲
flusher.Flush()
ctx := stream.WithRequestID(r.Context(), requestID)
s.enrichSessionWorkflowStage(ctx, sess, req.UserID)
if sess.ProjectID > 0 && !s.allowActiveStageChatAccess(ctx, sess.ProjectID, req.UserID) {
writeWorkflowAccessDeniedSSE(w, flusher, req.SessionID)
internal/server/stage_access_helpers.go
+30
View File
@@ -5,6 +5,7 @@ import (
"encoding/json"
"net/http"
"chat-agent-system/internal/auth"
"chat-agent-system/internal/chat"
"chat-agent-system/internal/ctxmgr"
"chat-agent-system/internal/docgov"
@@ -203,6 +204,35 @@ func (s *Server) applyChatWindowStageFilter(ctx context.Context, projectID, user
s.applyChatWindowAuthFilter(authCtx, res)
}
// enforceChatProjectAccess 项目对话入口：成员/创建者；系统管理员按既有策略放行。
// 必须在 SSE write headers / Flush 之前调用，拒绝时写结构化 JSON 403（非 200 SSE）。
func (s *Server) enforceChatProjectAccess(w http.ResponseWriter, r *http.Request, user *auth.UserAccount, projectID int64) bool {
if projectID <= 0 || user == nil {
return true
}
if s.isSystemAdminUser(r.Context(), user) {
return true
}
if s.authorizer == nil {
if s.failClosedAuthz() {
writeJSONResponse(w, http.StatusForbidden, map[string]string{
"error_code": "ERR_FORBIDDEN",
"message": "无权访问该项目",
})
return false
}
return true
}
if err := s.authorizer.RequireProjectAccess(r.Context(), projectID, user.ID); err != nil {
writeJSONResponse(w, http.StatusForbidden, map[string]string{
"error_code": "ERR_FORBIDDEN",
"message": "无权访问该项目",
})
return false
}
return true
}
func workflowAccessDeniedJSON() string {
return `{"error":"WORKFLOW_ACCESS_DENIED","message":"无权访问该阶段内容"}`
}