diff --git a/internal/server/chat_project_access_authz_test.go b/internal/server/chat_project_access_authz_test.go new file mode 100644 index 00000000..6bf81ee6 --- /dev/null +++ b/internal/server/chat_project_access_authz_test.go @@ -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")) + } + } +} diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 80756949..3947b016 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -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) diff --git a/internal/server/sse.go b/internal/server/sse.go index 66c0a661..3324df79 100644 --- a/internal/server/sse.go +++ b/internal/server/sse.go @@ -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) diff --git a/internal/server/stage_access_helpers.go b/internal/server/stage_access_helpers.go index 7a680103..940e6e19 100644 --- a/internal/server/stage_access_helpers.go +++ b/internal/server/stage_access_helpers.go @@ -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":"无权访问该阶段内容"}` }