Merge pull request #77 from thand-io/self-approvals

added support for self approvals

Author: hughneale

.github/workflows/test-and-build.yml

@@ -208,6 +208,16 @@ jobs:
       with:
         go-version: ${{ env.GO_VERSION }}
 
+    - name: Cache Go modules
+      uses: actions/cache@v3
+      with:
+        path: |
+          ~/.cache/go-build
+          ~/go/pkg/mod
+        key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+        restore-keys: |
+          ${{ runner.os }}-go-
+
     - name: Get version from auto-tag
       id: version
       run: |

examples/workflows/thand.approvals.example.yml

@@ -36,7 +36,7 @@ workflows:
             with:
 
               approvals: 1 # number of approvals required
-
+              selfApprove: false # can the requester approve their own request
               # 1. The thand notifier simplifies multiple workflow steps down
               #    to a single call.
               #    Slack - Sends request to channel

internal/daemon/elevate.go

@@ -375,6 +375,50 @@ func (s *Server) getElevateAuthOAuth2(c *gin.Context) {
 
 func (s *Server) resumeWorkflow(c *gin.Context, workflow *models.WorkflowTask) {
 
+	// Get user context
+	if !s.Config.IsServer() {
+		s.getErrorPage(c, http.StatusBadRequest, "Cannot process elevation request")
+		return
+	}
+
+	// Get user context
+	// TODO: Validate the provider that we're using?
+	_, foundUser, err := s.getUser(c)
+
+	if err != nil {
+		logrus.WithError(err).Error("failed to get user")
+		s.getErrorPage(c, http.StatusUnauthorized, "Unauthorized: unable to get user for elevation", err)
+		return
+	}
+
+	if foundUser == nil {
+		s.getErrorPage(c, http.StatusUnauthorized, "Unauthorized: user not found for elevation")
+		return
+	}
+
+	if foundUser.User == nil {
+		s.getErrorPage(c, http.StatusUnauthorized, "Unauthorized: user information is missing for elevation")
+		return
+	}
+
+	// Lets check if the workflow has a cloudevent input to process
+	event := workflow.GetInputAsCloudEvent()
+
+	if event != nil {
+
+		// Extensions only support basic types so we need to set the user identity as a string
+		event.SetExtension(models.VarsContextUser, foundUser.User.GetIdentity())
+
+		if len(event.FieldErrors) > 0 {
+			logrus.WithField("errors", event.FieldErrors).
+				Error("failed to set user extension on cloudevent")
+			s.getErrorPage(c, http.StatusBadRequest, "Failed to set user extension on cloudevent")
+			return
+		}
+
+		workflow.SetInput(event)
+	}
+
 	// Provide no input to resume the workflow as it'll use the saved state
 	// inputs are only for signals
 	workflowTask, err := s.Workflows.ResumeWorkflow(

internal/models/workflow_ctx.go

@@ -6,10 +6,12 @@ import (
 	"maps"
 	"time"
 
+	cloudevents "github.com/cloudevents/sdk-go/v2"
 	swctx "github.com/serverlessworkflow/sdk-go/v3/impl/ctx"
 	"github.com/serverlessworkflow/sdk-go/v3/impl/utils"
 	"github.com/serverlessworkflow/sdk-go/v3/model"
 	"github.com/sirupsen/logrus"
+	"github.com/thand-io/agent/internal/common"
 )
 
 func NewWorkflowContext(workflow *Workflow) (*WorkflowTask, error) {
@@ -235,6 +237,36 @@ func (ctx *WorkflowTask) GetInputAsMap() map[string]any {
 	return map[string]any{"input": ctx.Input}
 }
 
+func (ctx *WorkflowTask) GetInputAsCloudEvent() *cloudevents.Event {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	var event cloudevents.Event
+	if err := common.ConvertInterfaceToInterface(ctx.Input, &event); err != nil {
+		logrus.WithError(err).Error("failed to unmarshal cloudevent from workflow input")
+		return nil
+	}
+
+	if len(event.ID()) == 0 {
+		logrus.Error("cloudevent validation failed: missing ID")
+		return nil
+	}
+	if event.Time().IsZero() {
+		logrus.Error("cloudevent validation failed: missing Time")
+		return nil
+	}
+	if len(event.Source()) == 0 {
+		logrus.Error("cloudevent validation failed: missing Source")
+		return nil
+	}
+	if len(event.Type()) == 0 {
+		logrus.Error("cloudevent validation failed: missing Type")
+		return nil
+	}
+
+	return &event
+}
+
 func (ctx *WorkflowTask) GetContextAsMap() map[string]any {
 	ctx.mu.Lock()
 	defer ctx.mu.Unlock()

internal/workflows/manager/manager.go

@@ -294,7 +294,8 @@ func (m *WorkflowManager) ResumeWorkflow(
 
 		// Lets signal the workflow to continue
 		err = temporalClient.SignalWorkflow(
-			ctx, result.WorkflowID, models.TemporalEmptyRunId, models.TemporalResumeSignalName, result)
+			ctx, result.WorkflowID, models.TemporalEmptyRunId,
+			models.TemporalResumeSignalName, result)
 
 		if err != nil {
 			return nil, fmt.Errorf("failed to signal workflow: %w", err)

internal/workflows/tasks/providers/thand/approvals.go

@@ -3,6 +3,8 @@ package thand
 import (
 	"errors"
 	"fmt"
+	"slices"
+	"time"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/serverlessworkflow/sdk-go/v3/model"
@@ -21,8 +23,26 @@ func (t *thandTask) executeApprovalsTask(
 	call *taskModel.ThandTask,
 	input any) (any, error) {
 
+	elevationRequest, err := workflowTask.GetContextAsElevationRequest()
+
+	if err != nil {
+		logrus.WithError(err).WithFields(logrus.Fields{
+			"taskName": taskName,
+		}).Error("Failed to get elevation request from context")
+
+		return nil, err
+	}
+
 	var notifyReq NotifyRequest
-	common.ConvertInterfaceToInterface(call.With, &notifyReq)
+	err = common.ConvertInterfaceToInterface(call.With, &notifyReq)
+
+	if err != nil {
+		logrus.WithError(err).WithFields(logrus.Fields{
+			"taskName": taskName,
+		}).Error("Failed to parse notification request")
+
+		return nil, err
+	}
 
 	if !notifyReq.IsValid() {
 		return nil, errors.New("invalid notification request")
@@ -82,6 +102,10 @@ func (t *thandTask) executeApprovalsTask(
 		return nil, err
 	}
 
+	defaultFlowState := model.FlowDirective{
+		Value: taskName, // loop back to await more approvals
+	}
+
 	// Set the context to hold all the approvals
 	/*
 		output:
@@ -97,37 +121,76 @@ func (t *thandTask) executeApprovalsTask(
 
 	workflowContext := workflowTask.GetContextAsMap()
 
-	approvals, ok := workflowContext["approvals"].([]any)
+	approvals, ok := workflowContext["approvals"].(map[string]any)
 
 	if !ok {
-		approvals = []any{}
+		approvals = map[string]any{}
 	}
 
 	var approvalData map[string]any
 
 	if approvalEvent, ok := approval.(*cloudevents.Event); ok {
 
 		approvalEvent.DataAs(&approvalData)
+		extensions := approvalEvent.Extensions()
+
+		userIdentity, userExists := extensions[models.VarsContextUser].(string)
+
+		if !userExists {
+			logrus.Warn("Approval event missing user extension")
+			return &defaultFlowState, nil
+		}
+
+		// Check if self-approval is disabled and the approver is the requester or one of the elevated identities
+		if !notifyReq.SelfApprove {
+			requesterIdentity := elevationRequest.User.GetIdentity()
+
+			// Check if approver is the requester
+			if userIdentity == requesterIdentity {
+				logrus.WithFields(logrus.Fields{
+					"taskName":          taskName,
+					"userIdentity":      userIdentity,
+					"requesterIdentity": requesterIdentity,
+				}).Warn("Self-approval is disabled; ignoring approval from requester")
+
+				// Return to the default flow state to await more approvals
+				return &defaultFlowState, nil
+			}
+
+			// Check if approver is one of the identities being elevated
+			if slices.Contains(elevationRequest.Identities, userIdentity) {
+				logrus.WithFields(logrus.Fields{
+					"taskName":     taskName,
+					"userIdentity": userIdentity,
+				}).Warn("Self-approval is disabled; ignoring approval from identity being elevated")
+
+				// Return to the default flow state to await more approvals
+				return &defaultFlowState, nil
+			}
+		}
 
 		if approved, exists := approvalData["approved"]; exists {
-			approvals = append(approvals, map[string]any{
-				"approved": approved,
-			})
+			approvals[userIdentity] = map[string]any{
+				"approved":  approved,
+				"timestamp": time.Now().UTC().Format(time.RFC3339),
+			}
 		}
 	}
 
 	workflowTask.SetContextKeyValue("approvals", approvals)
 
 	/*
 		# If anyone rejects then reject the entire request
-		# otherwise if there is more than one approval then
-		# authorize
+		# otherwise if the required number of approvals is met then authorize
+		# Approvals are stored as a map[identity]approval_data structure
 		- case1:
-			when: any($context.approvals[]; .approved == false)
+			when: any($context.approvals | to_entries[]; .value.approved == false)
 			then: denied
 		- case2:
-			when: '[$context.approvals[] | select(.approved == true)] | length >= 1'
+			when: '[$context.approvals | to_entries[] | select(.value.approved == true)] | length >= N'
 			then: authorize
+		- default:
+			then: loop back to task to await more approvals
 	*/
 
 	approvedState, foundApprovedState := call.On.GetString("approved")
@@ -138,7 +201,43 @@ func (t *thandTask) executeApprovalsTask(
 	}
 
 	// Create the switch task to handle approval or rejection
-	flowDirective, err := runner.SwitchTaskHandler(
+	flowDirective, err := t.evaluateApprovalSwitch(
+		workflowTask,
+		taskName,
+		approvals,
+		notifyReq.Approvals,
+		approvedState,
+		deniedState,
+	)
+
+	if err != nil {
+		logrus.WithError(err).WithFields(logrus.Fields{
+			"taskName": taskName,
+		}).Error("Failed to execute switch task for approval logic")
+
+		return nil, err
+	}
+
+	logrus.WithFields(logrus.Fields{
+		"taskName":      taskName,
+		"flowDirective": flowDirective.Value,
+	}).Info("Completed Thand approvals task")
+
+	return flowDirective, nil
+}
+
+// evaluateApprovalSwitch evaluates the approval logic using a switch task
+// to determine if the request should be approved, denied, or loop back for more approvals
+func (t *thandTask) evaluateApprovalSwitch(
+	workflowTask *models.WorkflowTask,
+	taskName string,
+	approvals map[string]any,
+	requiredApprovals int,
+	approvedState string,
+	deniedState string,
+) (*model.FlowDirective, error) {
+
+	return runner.SwitchTaskHandler(
 		workflowTask,
 		map[string]any{
 			"approvals": approvals,
@@ -149,7 +248,7 @@ func (t *thandTask) executeApprovalsTask(
 				{
 					"case1": model.SwitchCase{
 						When: &model.RuntimeExpression{
-							Value: "any($context.approvals[]; .approved == false)",
+							Value: "any($context.approvals | to_entries[]; .value.approved == false)",
 						},
 						Then: &model.FlowDirective{
 							Value: deniedState, // go to denied state
@@ -159,7 +258,7 @@ func (t *thandTask) executeApprovalsTask(
 				{
 					"case2": model.SwitchCase{
 						When: &model.RuntimeExpression{
-							Value: fmt.Sprintf("[$context.approvals[] | select(.approved == true)] | length >= %d", notifyReq.Approvals),
+							Value: fmt.Sprintf("[$context.approvals | to_entries[] | select(.value.approved == true)] | length >= %d", requiredApprovals),
 						},
 						Then: &model.FlowDirective{
 							Value: approvedState, // proceed to the next state
@@ -176,19 +275,4 @@ func (t *thandTask) executeApprovalsTask(
 				},
 			},
 		})
-
-	if err != nil {
-		logrus.WithError(err).WithFields(logrus.Fields{
-			"taskName": taskName,
-		}).Error("Failed to execute switch task for approval logic")
-
-		return nil, err
-	}
-
-	logrus.WithFields(logrus.Fields{
-		"taskName":      taskName,
-		"flowDirective": flowDirective.Value,
-	}).Info("Completed Thand approvals task")
-
-	return flowDirective, nil
 }