From c89898ead481f33c3ad64cf7615685a1b6e30af6 Mon Sep 17 00:00:00 2001
From: pmahindrakar-oss <prafulla.mahindrakar@gmail.com>
Date: Tue, 12 Apr 2022 00:51:15 +0530
Subject: [PATCH] Deep check on security context (#397)

---
 pkg/manager/impl/execution_manager.go      |  9 ++++--
 pkg/manager/impl/execution_manager_test.go | 33 ++++++++++++++++++++++
 2 files changed, 40 insertions(+), 2 deletions(-)

diff --git a/pkg/manager/impl/execution_manager.go b/pkg/manager/impl/execution_manager.go
index 8267561fdb..a150fda27d 100644
--- a/pkg/manager/impl/execution_manager.go
+++ b/pkg/manager/impl/execution_manager.go
@@ -456,9 +456,14 @@ func mergeIntoExecConfig(workflowExecConfig *admin.WorkflowExecutionConfig, spec
 		workflowExecConfig.MaxParallelism = spec.GetMaxParallelism()
 		isChanged = true
 	}
+
 	if workflowExecConfig.GetSecurityContext() == nil && spec.GetSecurityContext() != nil {
-		workflowExecConfig.SecurityContext = spec.GetSecurityContext()
-		isChanged = true
+		if spec.GetSecurityContext().GetRunAs() != nil &&
+			(len(spec.GetSecurityContext().GetRunAs().GetK8SServiceAccount()) > 0 ||
+				len(spec.GetSecurityContext().GetRunAs().GetIamRole()) > 0) {
+			workflowExecConfig.SecurityContext = spec.GetSecurityContext()
+			isChanged = true
+		}
 	}
 	// Launchplan spec has label, annotation and rawOutputDataConfig initialized with empty values.
 	// Hence we do a deep check in the following conditions before assignment
diff --git a/pkg/manager/impl/execution_manager_test.go b/pkg/manager/impl/execution_manager_test.go
index 3bfc7917f1..6bc9c53dac 100644
--- a/pkg/manager/impl/execution_manager_test.go
+++ b/pkg/manager/impl/execution_manager_test.go
@@ -3814,6 +3814,39 @@ func TestGetExecutionConfigOverrides(t *testing.T) {
 		assert.Equal(t, requestOutputLocationPrefix, execConfig.RawOutputDataConfig.OutputLocationPrefix)
 		assert.Equal(t, requestLabels, execConfig.GetLabels().Values)
 	})
+	t.Run("request with empty security context", func(t *testing.T) {
+		request := &admin.ExecutionCreateRequest{
+			Project: workflowIdentifier.Project,
+			Domain:  workflowIdentifier.Domain,
+			Spec: &admin.ExecutionSpec{
+				SecurityContext: &core.SecurityContext{
+					RunAs: &core.Identity{
+						K8SServiceAccount: "",
+						IamRole:           "",
+					},
+				},
+			},
+		}
+		launchPlan := &admin.LaunchPlan{
+			Spec: &admin.LaunchPlanSpec{
+				Annotations:         &admin.Annotations{Values: launchPlanAnnotations},
+				Labels:              &admin.Labels{Values: launchPlanLabels},
+				RawOutputDataConfig: &admin.RawOutputDataConfig{OutputLocationPrefix: launchPlanOutputLocationPrefix},
+				SecurityContext: &core.SecurityContext{
+					RunAs: &core.Identity{
+						K8SServiceAccount: launchPlanK8sServiceAccount,
+					},
+				},
+				MaxParallelism: launchPlanMaxParallelism,
+			},
+		}
+		execConfig, err := executionManager.getExecutionConfig(context.TODO(), request, launchPlan)
+		assert.NoError(t, err)
+		assert.Equal(t, launchPlanMaxParallelism, execConfig.MaxParallelism)
+		assert.Equal(t, launchPlanK8sServiceAccount, execConfig.SecurityContext.RunAs.K8SServiceAccount)
+		assert.Equal(t, launchPlanOutputLocationPrefix, execConfig.RawOutputDataConfig.OutputLocationPrefix)
+		assert.Equal(t, launchPlanLabels, execConfig.GetLabels().Values)
+	})
 	t.Run("request with no config", func(t *testing.T) {
 		request := &admin.ExecutionCreateRequest{
 			Project: workflowIdentifier.Project,