Implement org/user agents

server/grpc/auth_server.go

@@ -60,13 +60,13 @@ func (s *WoodpeckerAuthServer) getAgent(agentID int64, agentToken string) (*mode
 	// global agent secret auth
 	if s.agentMasterToken != "" {
 		if agentToken == s.agentMasterToken && agentID == -1 {
-			agent := new(model.Agent)
-			agent.Name = ""
-			agent.OwnerID = -1 // system agent
-			agent.Token = s.agentMasterToken
-			agent.Backend = ""
-			agent.Platform = ""
-			agent.Capacity = -1
+			agent := &model.Agent{
+				OwnerID:  model.IDNotSet,
+				OrgID:    model.IDNotSet,
+				RepoID:   model.IDNotSet,
+				Token:    s.agentMasterToken,
+				Capacity: -1,
+			}
 			err := s.store.AgentCreate(agent)
 			if err != nil {
 				log.Error().Err(err).Msg("error creating system agent")

server/grpc/rpc.go

@@ -57,8 +57,6 @@ func (s *RPC) Next(c context.Context, agentFilter rpc.Filter) (*rpc.Workflow, er
 		log.Debug().Msgf("agent connected: %s: polling", hostname)
 	}
 
-	filterFn := createFilterFunc(agentFilter)
-
 	agent, err := s.getAgentFromContext(c)
 	if err != nil {
 		return nil, err
@@ -69,6 +67,20 @@ func (s *RPC) Next(c context.Context, agentFilter rpc.Filter) (*rpc.Workflow, er
 		return nil, nil
 	}
 
+	// enforce server set agent filters ...
+	agentServerFilters, err := agent.GetServerFilters()
+	if err != nil {
+		return nil, err
+	}
+	// ... by overwrite and extend the agent labels
+	for k, v := range agentServerFilters {
+		agentFilter.Labels[k] = v
+	}
+
+	log.Trace().Msgf("Agent %s[%d] try pull task with filter labels: %v", agent.Name, agent.ID, agentFilter.Labels)
+
+	filterFn := createFilterFunc(agentFilter)
+
 	for {
 		// poll blocks until a task is available or the context is canceled / worker is kicked
 		task, err := s.queue.Poll(c, agent.ID, filterFn)
@@ -91,6 +103,15 @@ func (s *RPC) Next(c context.Context, agentFilter rpc.Filter) (*rpc.Workflow, er
 
 // Wait blocks until the workflow with the given ID is done.
 func (s *RPC) Wait(c context.Context, workflowID string) error {
+	agent, err := s.getAgentFromContext(c)
+	if err != nil {
+		return err
+	}
+
+	if err := s.checkAgentPermissionByWorkflow(c, agent, workflowID, nil, nil); err != nil {
+		return err
+	}
+
 	return s.queue.Wait(c, workflowID)
 }
 
@@ -106,11 +127,15 @@ func (s *RPC) Extend(c context.Context, workflowID string) error {
 		return err
 	}
 
-	return s.queue.Extend(c, workflowID)
+	if err := s.checkAgentPermissionByWorkflow(c, agent, workflowID, nil, nil); err != nil {
+		return err
+	}
+
+	return s.queue.Extend(c, agent.ID, workflowID)
 }
 
 // Update updates the state of a step.
-func (s *RPC) Update(_ context.Context, strWorkflowID string, state rpc.StepState) error {
+func (s *RPC) Update(c context.Context, strWorkflowID string, state rpc.StepState) error {
 	workflowID, err := strconv.ParseInt(strWorkflowID, 10, 64)
 	if err != nil {
 		return err
@@ -128,6 +153,11 @@ func (s *RPC) Update(_ context.Context, strWorkflowID string, state rpc.StepStat
 		return err
 	}
 
+	agent, err := s.getAgentFromContext(c)
+	if err != nil {
+		return err
+	}
+
 	step, err := s.store.StepByUUID(state.StepUUID)
 	if err != nil {
 		log.Error().Err(err).Msgf("cannot find step with uuid %s", state.StepUUID)
@@ -149,6 +179,11 @@ func (s *RPC) Update(_ context.Context, strWorkflowID string, state rpc.StepStat
 		return err
 	}
 
+	// check before agent can alter some state
+	if err := s.checkAgentPermissionByWorkflow(c, agent, strWorkflowID, currentPipeline, repo); err != nil {
+		return err
+	}
+
 	if err := pipeline.UpdateStepStatus(s.store, step, state); err != nil {
 		log.Error().Err(err).Msg("rpc.update: cannot update step")
 	}
@@ -192,6 +227,7 @@ func (s *RPC) Init(c context.Context, strWorkflowID string, state rpc.WorkflowSt
 	if err != nil {
 		return err
 	}
+
 	workflow.AgentID = agent.ID
 
 	currentPipeline, err := s.store.GetPipeline(workflow.PipelineID)
@@ -206,6 +242,11 @@ func (s *RPC) Init(c context.Context, strWorkflowID string, state rpc.WorkflowSt
 		return err
 	}
 
+	// check before agent can alter some state
+	if err := s.checkAgentPermissionByWorkflow(c, agent, strWorkflowID, currentPipeline, repo); err != nil {
+		return err
+	}
+
 	if currentPipeline.Status == model.StatusPending {
 		if currentPipeline, err = pipeline.UpdateToStatusRunning(s.store, *currentPipeline, state.Started); err != nil {
 			log.Error().Err(err).Msgf("init: cannot update pipeline %d state", currentPipeline.ID)
@@ -272,6 +313,16 @@ func (s *RPC) Done(c context.Context, strWorkflowID string, state rpc.WorkflowSt
 		return err
 	}
 
+	agent, err := s.getAgentFromContext(c)
+	if err != nil {
+		return err
+	}
+
+	// check before agent can alter some state
+	if err := s.checkAgentPermissionByWorkflow(c, agent, strWorkflowID, currentPipeline, repo); err != nil {
+		return err
+	}
+
 	logger := log.With().
 		Str("repo_id", fmt.Sprint(repo.ID)).
 		Str("pipeline_id", fmt.Sprint(currentPipeline.ID)).
@@ -328,10 +379,6 @@ func (s *RPC) Done(c context.Context, strWorkflowID string, state rpc.WorkflowSt
 		s.pipelineTime.WithLabelValues(repo.FullName, currentPipeline.Branch, string(workflow.State), workflow.Name).Set(float64(workflow.Finished - workflow.Started))
 	}
 
-	agent, err := s.getAgentFromContext(c)
-	if err != nil {
-		return err
-	}
 	return s.updateAgentLastWork(agent)
 }
 
@@ -348,6 +395,17 @@ func (s *RPC) Log(c context.Context, stepUUID string, rpcLogEntries []*rpc.LogEn
 		return err
 	}
 
+	currentPipeline, err := s.store.GetPipeline(step.PipelineID)
+	if err != nil {
+		log.Error().Err(err).Msgf("cannot find pipeline with id %d", step.PipelineID)
+		return err
+	}
+
+	// check before agent can alter some state
+	if err := s.checkAgentPermissionByWorkflow(c, agent, "", currentPipeline, nil); err != nil {
+		return err
+	}
+
 	err = s.updateAgentLastWork(agent)
 	if err != nil {
 		return err
@@ -441,6 +499,44 @@ func (s *RPC) ReportHealth(ctx context.Context, status string) error {
 	return s.store.AgentUpdate(agent)
 }
 
+func (s *RPC) checkAgentPermissionByWorkflow(_ context.Context, agent *model.Agent, strWorkflowID string, pipeline *model.Pipeline, repo *model.Repo) error {
+	var err error
+	if repo == nil {
+		if pipeline == nil {
+			workflowID, err := strconv.ParseInt(strWorkflowID, 10, 64)
+			if err != nil {
+				return err
+			}
+
+			workflow, err := s.store.WorkflowLoad(workflowID)
+			if err != nil {
+				log.Error().Err(err).Msgf("rpc.update: cannot find workflow with id %d", workflowID)
+				return err
+			}
+
+			pipeline, err = s.store.GetPipeline(workflow.PipelineID)
+			if err != nil {
+				log.Error().Err(err).Msgf("cannot find pipeline with id %d", workflow.PipelineID)
+				return err
+			}
+		}
+
+		repo, err = s.store.GetRepo(pipeline.RepoID)
+		if err != nil {
+			log.Error().Err(err).Msgf("cannot find repo with id %d", pipeline.RepoID)
+			return err
+		}
+	}
+
+	if agent.CanAccessRepo(repo) {
+		return nil
+	}
+
+	msg := fmt.Sprintf("agent '%d' is not allowed to interact with repo[%d] '%s'", agent.ID, repo.ID, repo.FullName)
+	log.Error().Int64("repoId", repo.ID).Msg(msg)
+	return errors.New(msg)
+}
+
 func (s *RPC) completeChildrenIfParentCompleted(completedWorkflow *model.Workflow) {
 	for _, c := range completedWorkflow.Children {
 		if c.Running() {

server/model/agent.go

@@ -14,6 +14,13 @@
 
 package model
 
+import (
+	"encoding/base32"
+	"fmt"
+
+	"github.com/gorilla/securecookie"
+)
+
 type Agent struct {
 	ID          int64  `json:"id"            xorm:"pk autoincr 'id'"`
 	Created     int64  `json:"created"       xorm:"created"`
@@ -28,13 +35,54 @@ type Agent struct {
 	Capacity    int32  `json:"capacity"      xorm:"capacity"`
 	Version     string `json:"version"       xorm:"'version'"`
 	NoSchedule  bool   `json:"no_schedule"   xorm:"no_schedule"`
+	// OrgID is counted as unset if set to -1, this is done to ensure a new(Agent) still enforce the OrgID check by default
+	OrgID int64 `json:"org_id"        xorm:"INDEX 'org_id'"`
+	// RepoID is counted as unset if set to -1, this is done to ensure a new(Agent) still enforce the OrgID check by default
+	RepoID int64 `json:"repo_id"       xorm:"INDEX 'repo_id'"`
 } //	@name Agent
 
+const (
+	IDNotSet          = -1
+	agentFilterOrgID  = "org-id"
+	agentFilterRepoID = "repo-id"
+)
+
 // TableName return database table name for xorm.
 func (Agent) TableName() string {
 	return "agents"
 }
 
 func (a *Agent) IsSystemAgent() bool {
-	return a.OwnerID == -1
+	return a.OwnerID == IDNotSet
+}
+
+func GenerateNewAgentToken() string {
+	return base32.StdEncoding.EncodeToString(securecookie.GenerateRandomKey(32))
+}
+
+func (a *Agent) GetServerFilters() (map[string]string, error) {
+	filters := make(map[string]string)
+
+	// enforce filters for user and organization agents
+	if a.OrgID != IDNotSet {
+		filters[agentFilterOrgID] = fmt.Sprintf("%d", a.OrgID)
+	} else {
+		filters[agentFilterOrgID] = "*"
+	}
+	if a.RepoID != IDNotSet {
+		filters[agentFilterRepoID] = fmt.Sprintf("%d", a.RepoID)
+	} else {
+		filters[agentFilterRepoID] = "*"
+	}
+
+	return filters, nil
+}
+
+func (a *Agent) CanAccessRepo(repo *Repo) bool {
+	if a.IsSystemAgent() {
+		return true
+	}
+
+	return a.RepoID != IDNotSet && a.RepoID == repo.ID && (a.OrgID == IDNotSet || a.OrgID == repo.OrgID) ||
+		a.OrgID != IDNotSet && a.OrgID == repo.OrgID && (a.RepoID == IDNotSet || a.RepoID == repo.ID)
 }