From 7c5f8deeb101c9e2cf708a17b8e68fd073824267 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 12 Sep 2026 05:27:25 +0000 Subject: [PATCH] feat(abac): gate bizdev agent context by workflow stage duties Inject stage access gate into bizdev service so task execution, project context reads, and prior docs respect live workflow ABAC. Co-authored-by: Cursor --- internal/bizdev/service.go | 39 +++++++-- internal/bizdev/stage_access.go | 50 +++++++++++ internal/bizdev/stage_access_test.go | 57 +++++++++++++ internal/server/bizdev_stage_gate.go | 123 +++++++++++++++++++++++++++ internal/server/server.go | 3 + 5 files changed, 266 insertions(+), 6 deletions(-) create mode 100644 internal/bizdev/stage_access.go create mode 100644 internal/bizdev/stage_access_test.go create mode 100644 internal/server/bizdev_stage_gate.go diff --git a/internal/bizdev/service.go b/internal/bizdev/service.go index 5f19bc89..f4a72482 100644 --- a/internal/bizdev/service.go +++ b/internal/bizdev/service.go @@ -39,6 +39,7 @@ type Service struct { projectWorkflow *projectworkflow.Service pool *pgxpool.Pool schemaVersion int64 + stageGate StageAccessGate } // NewService 构造业务服务；生产/test 环境数据库不可用且未显式 ALLOW_IN_MEMORY 时返回 error。 @@ -230,6 +231,14 @@ func (s *Service) ListStages() []StageMeta { } func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRunResult, error) { + if err := s.ensureTaskStageAccess(ctx, req.ProjectID, req.UserID, req.TaskKey); err != nil { + return &TaskRunResult{ + TaskKey: req.TaskKey, + Success: false, + ErrorReason: err.Error(), + }, err + } + materials, inputManifest, err := s.ResolveMaterials(ctx, req) if err != nil { return nil, err @@ -240,7 +249,7 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun // 针对 system_reqs（智能产品经理）：若未显式提供 materials，自动抽取项目事实黑板/集体上下文为需求输入 if req.TaskKey == "system_reqs" && len(materials) == 0 && s.ctxMgr != nil && req.ProjectID > 0 { - coll, err := s.ctxMgr.GetProjectContext(ctx, req.ProjectID) + coll, err := s.getProjectContextForUser(ctx, req.ProjectID, req.UserID) if err == nil && coll != nil { var b strings.Builder customer := coll.BlackboardFacts["customer_name"] @@ -342,9 +351,11 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun if req.TaskKey == "test_manager" { upstreamDocID := fmt.Sprintf("doc_proj%d_system_reqs", req.ProjectID) if s.docGov != nil && req.ProjectID > 0 { - doc, err := s.docGov.GetDocument(ctx, upstreamDocID) - if err == nil && doc != nil && doc.CurrentState == docgov.StateApproved { - upstreamSRDoc = doc + if gateErr := s.ensureTaskStageAccess(ctx, req.ProjectID, req.UserID, "system_reqs"); gateErr == nil { + doc, err := s.docGov.GetDocument(ctx, upstreamDocID) + if err == nil && doc != nil && doc.CurrentState == docgov.StateApproved { + upstreamSRDoc = doc + } } } if upstreamSRDoc == nil { @@ -592,6 +603,11 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun } } + priorDocs := req.PriorDocs + if s.stageGate != nil && req.ProjectID > 0 && req.UserID > 0 { + priorDocs = s.stageGate.FilterPriorDocs(ctx, req.ProjectID, req.UserID, priorDocs) + } + squadReq := squad.SquadRunRequest{ TaskKey: req.TaskKey, Title: req.Title, @@ -608,7 +624,7 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun CustomSteps: req.CustomSteps, AutoRepair: req.AutoRepair, KnowledgeHits: hits, - PriorDocs: req.PriorDocs, + PriorDocs: priorDocs, UserContext: userCtxStr, MemoryBankFiles: memBank, SessionID: req.SessionID, @@ -697,6 +713,14 @@ func (s *Service) ExecuteStage(ctx context.Context, req StageRunRequest) (*Stage func (s *Service) executeStageInternal(ctx context.Context, req StageRunRequest) (*StageRunResult, error) { start := time.Now() + if err := s.ensureTaskStageAccess(ctx, req.ProjectID, req.UserID, string(req.Stage)); err != nil { + return &StageRunResult{ + Stage: req.Stage, + Success: false, + ErrorReason: err.Error(), + }, err + } + // 1. 确定工作区目录基准 stageKey := string(req.Stage) if stageKey == "" { @@ -732,7 +756,7 @@ func (s *Service) executeStageInternal(ctx context.Context, req StageRunRequest) // 若存在项目协同管理器，优先载入已沉淀的项目事实与黑板 if s.ctxMgr != nil && req.ProjectID > 0 { - if projCtx, err := s.ctxMgr.GetProjectContext(ctx, req.ProjectID); err == nil && projCtx != nil { + if projCtx, err := s.getProjectContextForUser(ctx, req.ProjectID, req.UserID); err == nil && projCtx != nil { for k, v := range projCtx.BlackboardFacts { currentFields[k] = v } @@ -761,6 +785,9 @@ func (s *Service) executeStageInternal(ctx context.Context, req StageRunRequest) } accumulatedPriorDocs := append([]squad.PriorDocItem{}, req.PriorDocs...) + if s.stageGate != nil && req.ProjectID > 0 && req.UserID > 0 { + accumulatedPriorDocs = s.stageGate.FilterPriorDocs(ctx, req.ProjectID, req.UserID, accumulatedPriorDocs) + } var executedArtifacts []*TaskRunResult // 4. 按顺序执行当前阶段的子任务序列，并将每一轮产出作为后续任务的 PriorDoc 和事实补充 diff --git a/internal/bizdev/stage_access.go b/internal/bizdev/stage_access.go new file mode 100644 index 00000000..0e915a85 --- /dev/null +++ b/internal/bizdev/stage_access.go @@ -0,0 +1,50 @@ +package bizdev + +import ( + "context" + "errors" + + "chat-agent-system/internal/ctxmgr" + "chat-agent-system/internal/squad" +) + +var ErrStageAccessDenied = errors.New("bizdev: workflow stage access denied") + +// StageAccessGate 工作流阶段 ABAC 门禁（由 server 注入 projectinstance 实现）。 +type StageAccessGate interface { + AllowTaskExecution(ctx context.Context, projectID, userID int64, taskKey string) error + AllowProjectContextRead(ctx context.Context, projectID, userID int64) error + FilterProjectContext(ctx context.Context, projectID, userID int64, pCtx *ctxmgr.ProjectCollectiveContext) *ctxmgr.ProjectCollectiveContext + FilterPriorDocs(ctx context.Context, projectID, userID int64, docs []squad.PriorDocItem) []squad.PriorDocItem +} + +func (s *Service) SetStageAccessGate(gate StageAccessGate) { + if s == nil { + return + } + s.stageGate = gate +} + +func (s *Service) getProjectContextForUser(ctx context.Context, projectID, userID int64) (*ctxmgr.ProjectCollectiveContext, error) { + if s == nil || s.ctxMgr == nil || projectID <= 0 { + return nil, nil + } + coll, err := s.ctxMgr.GetProjectContext(ctx, projectID) + if err != nil { + return nil, err + } + if s.stageGate == nil || userID <= 0 { + return coll, nil + } + if err := s.stageGate.AllowProjectContextRead(ctx, projectID, userID); err != nil { + return nil, err + } + return s.stageGate.FilterProjectContext(ctx, projectID, userID, coll), nil +} + +func (s *Service) ensureTaskStageAccess(ctx context.Context, projectID, userID int64, taskKey string) error { + if s == nil || s.stageGate == nil || projectID <= 0 || userID <= 0 { + return nil + } + return s.stageGate.AllowTaskExecution(ctx, projectID, userID, taskKey) +} diff --git a/internal/bizdev/stage_access_test.go b/internal/bizdev/stage_access_test.go new file mode 100644 index 00000000..7cb99c19 --- /dev/null +++ b/internal/bizdev/stage_access_test.go @@ -0,0 +1,57 @@ +package bizdev + +import ( + "context" + "errors" + "testing" + + "chat-agent-system/internal/ctxmgr" + "chat-agent-system/internal/squad" +) + +type mockStageGate struct { + taskErr error + ctxErr error + filterCtx *ctxmgr.ProjectCollectiveContext + filtered []squad.PriorDocItem +} + +func (m *mockStageGate) AllowTaskExecution(_ context.Context, _, _ int64, _ string) error { + return m.taskErr +} + +func (m *mockStageGate) AllowProjectContextRead(_ context.Context, _, _ int64) error { + return m.ctxErr +} + +func (m *mockStageGate) FilterProjectContext(_ context.Context, _, _ int64, pCtx *ctxmgr.ProjectCollectiveContext) *ctxmgr.ProjectCollectiveContext { + if m.filterCtx != nil { + return m.filterCtx + } + return pCtx +} + +func (m *mockStageGate) FilterPriorDocs(_ context.Context, _, _ int64, docs []squad.PriorDocItem) []squad.PriorDocItem { + if m.filtered != nil { + return m.filtered + } + return docs +} + +func TestEnsureTaskStageAccess_Denied(t *testing.T) { + svc := &Service{stageGate: &mockStageGate{taskErr: ErrStageAccessDenied}} + err := svc.ensureTaskStageAccess(context.Background(), 1, 2, "intake") + if !errors.Is(err, ErrStageAccessDenied) { + t.Fatalf("expected ErrStageAccessDenied, got %v", err) + } +} + +func TestSetStageAccessGate_NilSafe(t *testing.T) { + var svc *Service + svc.SetStageAccessGate(&mockStageGate{}) + s := &Service{} + s.SetStageAccessGate(&mockStageGate{filtered: []squad.PriorDocItem{{DocKey: "intake"}}}) + if s.stageGate == nil { + t.Fatal("expected gate to be set") + } +} diff --git a/internal/server/bizdev_stage_gate.go b/internal/server/bizdev_stage_gate.go new file mode 100644 index 00000000..f22bcf7a --- /dev/null +++ b/internal/server/bizdev_stage_gate.go @@ -0,0 +1,123 @@ +package server + +import ( + "context" + "fmt" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/ctxmgr" + "chat-agent-system/internal/projectinstance" + "chat-agent-system/internal/squad" +) + +type bizdevStageGate struct { + instSvc *projectinstance.Service + pwSvc interface { + IsProjectCreator(ctx context.Context, projectID int64, userID int64) (bool, error) + } +} + +func newBizdevStageGate(instSvc *projectinstance.Service, pwSvc interface { + IsProjectCreator(ctx context.Context, projectID int64, userID int64) (bool, error) +}) bizdev.StageAccessGate { + if instSvc == nil || !instSvc.Available() { + return nil + } + return &bizdevStageGate{instSvc: instSvc, pwSvc: pwSvc} +} + +func (g *bizdevStageGate) access(ctx context.Context, projectID, userID int64) (*projectinstance.StageAccess, error) { + if g == nil || g.instSvc == nil || !g.instSvc.Available() || projectID <= 0 || userID <= 0 { + return &projectinstance.StageAccess{}, nil + } + isCreator := false + if g.pwSvc != nil { + isCreator, _ = g.pwSvc.IsProjectCreator(ctx, projectID, userID) + } + return g.instSvc.ResolveStageAccess(ctx, projectID, userID, isCreator) +} + +func (g *bizdevStageGate) AllowTaskExecution(ctx context.Context, projectID, userID int64, taskKey string) error { + access, err := g.access(ctx, projectID, userID) + if err != nil { + return err + } + if access == nil || !access.Restricted { + return nil + } + if taskKey != "" && access.TaskKeyAllowed(taskKey) { + return nil + } + if access.StageKeyAllowed(access.ActiveStageKey) { + return nil + } + return fmt.Errorf("%w: task %q", bizdev.ErrStageAccessDenied, taskKey) +} + +func (g *bizdevStageGate) AllowProjectContextRead(ctx context.Context, projectID, userID int64) error { + access, err := g.access(ctx, projectID, userID) + if err != nil { + return err + } + if access == nil || !access.Restricted { + return nil + } + if len(access.AllowedStageKeys) == 0 { + return bizdev.ErrStageAccessDenied + } + return nil +} + +func (g *bizdevStageGate) FilterProjectContext(ctx context.Context, projectID, userID int64, pCtx *ctxmgr.ProjectCollectiveContext) *ctxmgr.ProjectCollectiveContext { + if pCtx == nil { + return pCtx + } + access, err := g.access(ctx, projectID, userID) + if err != nil || access == nil || !access.Restricted { + return pCtx + } + out := *pCtx + out.AllActivities = filterActivitiesByStage(access, pCtx.AllActivities) + out.MilestoneDecisions = filterActivitiesByStage(access, pCtx.MilestoneDecisions) + return &out +} + +func filterActivitiesByStage(access *projectinstance.StageAccess, items []ctxmgr.ActivityItem) []ctxmgr.ActivityItem { + if access == nil || !access.Restricted { + return items + } + out := make([]ctxmgr.ActivityItem, 0, len(items)) + for _, it := range items { + stageKey := "" + if it.Detail != nil { + if sk, ok := it.Detail["stage_key"].(string); ok { + stageKey = sk + } + } + if stageKey == "" || access.StageKeyAllowed(stageKey) { + out = append(out, it) + } + } + return out +} + +func (g *bizdevStageGate) FilterPriorDocs(ctx context.Context, projectID, userID int64, docs []squad.PriorDocItem) []squad.PriorDocItem { + if len(docs) == 0 { + return docs + } + access, err := g.access(ctx, projectID, userID) + if err != nil || access == nil || !access.Restricted { + return docs + } + out := make([]squad.PriorDocItem, 0, len(docs)) + for _, d := range docs { + if d.DocKey != "" && access.TaskKeyAllowed(d.DocKey) { + out = append(out, d) + continue + } + if d.Stage != "" && access.StageKeyAllowed(d.Stage) { + out = append(out, d) + } + } + return out +} diff --git a/internal/server/server.go b/internal/server/server.go index 474cc9e2..a9edf18d 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -102,6 +102,9 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { notifySvc: notification.NewService(pool), resolver: workspace.NewResolver(cfg.DataDir, pool), } + if instSvc != nil && instSvc.Available() { + svc.SetStageAccessGate(newBizdevStageGate(instSvc, pwSvc)) + } s.registerWorkflowApprovalHook() s.startNotificationReminderLoop() s.initTaskRunExecutors()