fix(api): 任务同步与SSE入口补齐项目成员鉴权

Co-authored-by: Cursor <cursoragent@cursor.com>

internal/server/handlers.go

@@ -197,6 +197,18 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) {
 	}
 	req.UserID = user.ID
 
+	if req.ProjectID > 0 {
+		if !s.isSystemAdminUser(r.Context(), user) {
+			if err := s.authorizer.RequireProjectAccess(r.Context(), req.ProjectID, user.ID); err != nil {
+				writeJSONResponse(w, http.StatusForbidden, map[string]string{
+					"error_code": "ERR_FORBIDDEN",
+					"message":    "无权访问该项目",
+				})
+				return
+			}
+		}
+	}
+
 	// 0. 如果关联了项目，在 projectworkflow 服务中记录运行实例
 	var runRecord *projectworkflow.WorkflowRun
 	if s.pwSvc != nil && req.ProjectID > 0 {

internal/server/sse.go

@@ -307,11 +307,30 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) {
 		req.Backend = q.Get("backend")
 		req.WorkDir = q.Get("work_dir")
 		req.AutoRepair, _ = strconv.ParseBool(q.Get("auto_repair"))
+		if pid := q.Get("project_id"); pid != "" {
+			req.ProjectID, _ = strconv.ParseInt(pid, 10, 64)
+		}
 		if inc := q.Get("include_details"); inc != "" {
 			includeDetails, _ = strconv.ParseBool(inc)
 		}
 	}
 
+	user, ok := s.requirePrincipal(w, r)
+	if !ok {
+		return
+	}
+	if req.ProjectID > 0 {
+		if !s.isSystemAdminUser(r.Context(), user) {
+			if err := s.authorizer.RequireProjectAccess(r.Context(), req.ProjectID, user.ID); err != nil {
+				writeJSONResponse(w, http.StatusForbidden, map[string]string{
+					"error_code": "ERR_FORBIDDEN",
+					"message":    "无权访问该项目",
+				})
+				return
+			}
+		}
+	}
+
 	// 请求追踪 ID：与 chat stream 对齐，注入 ctx 并回写响应头
 	requestID := fmt.Sprintf("req_%d", time.Now().UnixNano())
 	w.Header().Set("X-Request-Id", requestID)
@@ -322,11 +341,6 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("X-Accel-Buffering", "no") // 禁用 Nginx 等反代缓冲
 	flusher.Flush()
 
-	user, ok := s.requirePrincipal(w, r)
-	if !ok {
-		return
-	}
-
 	if req.TaskKey == "" && req.Stage == "" {
 		req.TaskKey = "intake"
 	}

internal/server/task_run_authz_pg_test.go

@@ -0,0 +1,123 @@
+package server_test
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"chat-agent-system/internal/projectworkflow"
+)
+
+func TestTaskRunAuthz_OutsiderForbidden_PG(t *testing.T) {
+	pool := openPGPoolWithMigrations(t)
+	if pool == nil {
+		return
+	}
+	defer pool.Close()
+
+	ctx := context.Background()
+	handler, _ := setupPGGovernanceServer(t, pool)
+	creatorToken := loginTestUser(t, handler, "zhanggong", "123456")
+	outsiderToken := loginTestUser(t, handler, "lizhuguan", "123456")
+
+	createBody, _ := json.Marshal(projectworkflow.CreateProjectInput{Name: "Task Run Authz Test"})
+	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) })
+
+	var wfRunsBefore, taskRunsBefore int
+	_ = pool.QueryRow(ctx, `SELECT COUNT(*) FROM workflow_runs WHERE project_id = $1`, proj.ID).Scan(&wfRunsBefore)
+	_ = pool.QueryRow(ctx, `SELECT COUNT(*) FROM task_runs WHERE project_id = $1`, proj.ID).Scan(&taskRunsBefore)
+
+	syncBody, _ := json.Marshal(map[string]interface{}{
+		"task_key":   "intake",
+		"project_id": proj.ID,
+		"backend":    "stub",
+	})
+	reqSync := httptest.NewRequest(http.MethodPost, "/api/v1/tasks/run", bytes.NewReader(syncBody))
+	withBearerAuth(reqSync, outsiderToken)
+	recSync := httptest.NewRecorder()
+	handler.ServeHTTP(recSync, reqSync)
+	if recSync.Code != http.StatusForbidden {
+		t.Fatalf("outsider sync tasks/run expected 403, got %d: %s", recSync.Code, recSync.Body.String())
+	}
+
+	streamBody, _ := json.Marshal(map[string]interface{}{
+		"task_key":   "intake",
+		"project_id": proj.ID,
+		"backend":    "stub",
+	})
+	reqStream := httptest.NewRequest(http.MethodPost, "/api/v1/tasks/run/stream", bytes.NewReader(streamBody))
+	withBearerAuth(reqStream, outsiderToken)
+	recStream := httptest.NewRecorder()
+	handler.ServeHTTP(recStream, reqStream)
+	if recStream.Code != http.StatusForbidden {
+		t.Fatalf("outsider stream tasks/run/stream expected 403, got %d: %s", recStream.Code, recStream.Body.String())
+	}
+
+	var wfRunsAfter, taskRunsAfter int
+	_ = pool.QueryRow(ctx, `SELECT COUNT(*) FROM workflow_runs WHERE project_id = $1`, proj.ID).Scan(&wfRunsAfter)
+	_ = pool.QueryRow(ctx, `SELECT COUNT(*) FROM task_runs WHERE project_id = $1`, proj.ID).Scan(&taskRunsAfter)
+	if wfRunsAfter != wfRunsBefore {
+		t.Fatalf("workflow_runs changed for forbidden sync: before=%d after=%d", wfRunsBefore, wfRunsAfter)
+	}
+	if taskRunsAfter != taskRunsBefore {
+		t.Fatalf("task_runs changed for forbidden stream: before=%d after=%d", taskRunsBefore, taskRunsAfter)
+	}
+
+	// 匿名请求
+	reqAnon := httptest.NewRequest(http.MethodPost, "/api/v1/tasks/run", bytes.NewReader(syncBody))
+	recAnon := httptest.NewRecorder()
+	handler.ServeHTTP(recAnon, reqAnon)
+	if recAnon.Code != http.StatusUnauthorized {
+		t.Fatalf("anonymous tasks/run expected 401, got %d", recAnon.Code)
+	}
+}
+
+func TestTaskRunAuthz_MemberAllowed_PG(t *testing.T) {
+	pool := openPGPoolWithMigrations(t)
+	if pool == nil {
+		return
+	}
+	defer pool.Close()
+
+	handler, _ := setupPGGovernanceServer(t, pool)
+	creatorToken := loginTestUser(t, handler, "zhanggong", "123456")
+
+	createBody, _ := json.Marshal(projectworkflow.CreateProjectInput{Name: fmt.Sprintf("Task Run Member %d", time.Now().UnixNano())})
+	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", recCreate.Code)
+	}
+	var proj projectworkflow.Project
+	_ = json.Unmarshal(recCreate.Body.Bytes(), &proj)
+	t.Cleanup(func() { cleanupProject(t, pool, proj.ID) })
+
+	syncBody, _ := json.Marshal(map[string]interface{}{
+		"task_key":   "intake",
+		"project_id": proj.ID,
+		"backend":    "stub",
+	})
+	reqSync := httptest.NewRequest(http.MethodPost, "/api/v1/tasks/run", bytes.NewReader(syncBody))
+	withBearerAuth(reqSync, creatorToken)
+	recSync := httptest.NewRecorder()
+	handler.ServeHTTP(recSync, reqSync)
+	if recSync.Code != http.StatusOK {
+		t.Fatalf("member sync tasks/run expected 200, got %d: %s", recSync.Code, recSync.Body.String())
+	}
+}