diff --git a/internal/server/handlers.go b/internal/server/handlers.go index e0db5232..d05355a7 100644 --- a/internal/server/handlers.go +++ b/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 { diff --git a/internal/server/sse.go b/internal/server/sse.go index 4bb43ddf..4ad7f2ad 100644 --- a/internal/server/sse.go +++ b/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" } diff --git a/internal/server/task_run_authz_pg_test.go b/internal/server/task_run_authz_pg_test.go new file mode 100644 index 00000000..df4bf484 --- /dev/null +++ b/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()) + } +}