Repository: argoproj/argo (go)

Issue:
Add controller mode where only WFs from WFTemplates can be submitted
# Summary

Add a "restricted" mode to the `workflow-controller` where only `Workflows` that submit `WorkflowTempaltes` can be executed. Depends on the top-level `WorkflowTemplate` reference PR #2556 

# Motivation

After #2556 is completed, users will be able to submit Workflows that are simply a reference to a `WorkflowTemplate`.

Some users might want to take advantage of this feature to add a "restricted" controller, in which users of said controller may only submit `Workflows` that already exist in the cluster as `WorkflowTemplates` and may not submit any user-defined `Workflows`. These `WorkflowTemplates` would be maintained by the "cluster owner".

# Proposal

Allow the `workflow-controller` to start with a flag setting it as restricted, then only accept `Workflows` that contain a workflow reference as defined in #2556 

Feature idea by @jessesuen 

---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.


Pull request description:
feat: Add mode to require Workflows to use workflowTemplateRef
Closes #2867

Currently called "reference mode" and enabled as a flag to the `workflow-controller` binary.

Patch:
diff --git a/config/config.go b/config/config.go
index abde6d582..1ec28bba1 100644
--- a/config/config.go
+++ b/config/config.go
@@ -85,8 +85,11 @@ type Config struct {
 	// WorkflowDefaults are values that will apply to all Workflows from this controller, unless overridden on the Workflow-level
 	WorkflowDefaults *wfv1.Workflow `json:"workflowDefaults,omitempty"`
 
-	// PodSpecLogStrategy enable the logging of podspec on controller log.
+	// PodSpecLogStrategy enables the logging of podspec on controller log.
 	PodSpecLogStrategy PodSpecLogStrategy `json:"podSpecLogStrategy,omitempty"`
+
+	// WorkflowRestrictions restricts the controller to executing Workflows that meet certain restrictions
+	WorkflowRestrictions *WorkflowRestrictions `json:"workflowRestrictions,omitempty"`
 }
 
 // PodSpecLogStrategy contains the configuration for logging the pod spec in controller log for debugging purpose
@@ -163,6 +166,7 @@ type ConnectionPool struct {
 	MaxOpenConns    int `json:"maxOpenConns,omitempty"`
 	ConnMaxLifetime TTL `json:"connMaxLifetime,omitempty"`
 }
+
 type PostgreSQLConfig struct {
 	Host           string                  `json:"host"`
 	Port           string                  `json:"port"`
@@ -246,3 +250,28 @@ type MetricsConfig struct {
 	// IgnoreErrors is a flag that instructs prometheus to ignore metric emission errors
 	IgnoreErrors bool `json:"ignoreErrors,omitempty"`
 }
+
+type WorkflowRestrictions struct {
+	TemplateReferencing TemplateReferencing `json:"templateReferencing"`
+}
+
+type TemplateReferencing string
+
+const (
+	TemplateReferencingStrict TemplateReferencing = "Strict"
+	TemplateReferencingSecure TemplateReferencing = "Secure"
+)
+
+func (req *WorkflowRestrictions) MustUseReference() bool {
+	if req == nil {
+		return false
+	}
+	return req.TemplateReferencing == TemplateReferencingStrict || req.TemplateReferencing == TemplateReferencingSecure
+}
+
+func (req *WorkflowRestrictions) MustNotChangeSpec() bool {
+	if req == nil {
+		return false
+	}
+	return req.TemplateReferencing == TemplateReferencingSecure
+}
diff --git a/docs/README.md b/docs/README.md
index e10d6db12..a1e6ba165 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -59,4 +59,5 @@ Some use-case specific documentation is available:
 * [Workflow Executors](workflow-executors.md)
 * [Workflow Notifications](workflow-notifications.md)
 * [Workfow RBAC](workflow-rbac.md)
+* [Workflow Restrictions](workflow-requirements.md)
 * [Workflow Templates](workflow-templates.md)
diff --git a/docs/workflow-controller-configmap.yaml b/docs/workflow-controller-configmap.yaml
index 118224c21..5e272bd93 100644
--- a/docs/workflow-controller-configmap.yaml
+++ b/docs/workflow-controller-configmap.yaml
@@ -220,3 +220,13 @@ data:
       # be in the form <argo-server-root-url>/oauth2/callback. It must be
       # browser-accessible.
       redirectUrl: https://argo-server/oauth2/callback
+
+    # workflowRequirements restricts the Workflows that the controller will process.
+    # Current options:
+    #   referenceOnly: Only Workflows using "workflowTemplateRef" will be processed. This allows the administrator of the controller
+    #     to set a "library" of templates that may be run by its opeartor, limiting arbitrary Workflow execution.
+    #   strictReferenceOnly: Only Workflows using "workflowTemplateRef" will be processed and the controller will enforce
+    #     that the WorkflowTemplate that is referenced hasn't changed between operations. If you want to make sure the operator of the
+    #     Workflow cannot run an arbitrary Workflow, use this option.
+    workflowRequirements:
+      referenceOnly: true
diff --git a/docs/workflow-requirements.md b/docs/workflow-requirements.md
new file mode 100644
index 000000000..9aef7c9c0
--- /dev/null
+++ b/docs/workflow-requirements.md
@@ -0,0 +1,31 @@
+# Workflow Restrictions
+
+![alpha](assets/alpha.svg)
+
+> v2.9 and after
+
+## Introduction
+
+As the administrator of the controller, you may want to limit which types of Workflows your users can run. Setting workflow restrictions allows you to ensure that Workflows comply with certain requirements. 
+
+## Available Restrictions
+
+* `templateReferencing: Strict`: Only Workflows using "workflowTemplateRef" will be processed. This allows the administrator of the controller to set a "library" of templates that may be run by its opeartor, limiting arbitrary Workflow execution.
+* `templateReferencing: Secure`: Only Workflows using "workflowTemplateRef" will be processed and the controller will enforce that the WorkflowTemplate that is referenced hasn't changed between operations. If you want to make sure the operator of the Workflow cannot run an arbitrary Workflow, use this option.
+
+## Setting Workflow Restrictions
+
+Workflow Restrictions can be specified by adding them under the `workflowRestrictions` key in the [`workflow-controller-configmap`](./workflow-controller-configmap.yaml).
+
+For example, to specify that Workflows may only run with `workflowTemplateRef`
+
+```yaml
+# This file describes the config settings available in the workflow controller configmap
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: workflow-controller-configmap
+data: |
+  workflowRestrictions:
+    templateReferencing: Secure
+```
diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
index fee629b86..184dd5411 100644
--- a/workflow/controller/controller.go
+++ b/workflow/controller/controller.go
@@ -96,17 +96,7 @@ const (
 )
 
 // NewWorkflowController instantiates a new WorkflowController
-func NewWorkflowController(
-	restConfig *rest.Config,
-	kubeclientset kubernetes.Interface,
-	wfclientset wfclientset.Interface,
-	namespace string,
-	managedNamespace string,
-	executorImage,
-	executorImagePullPolicy,
-	containerRuntimeExecutor,
-	configMap string,
-) *WorkflowController {
+func NewWorkflowController(restConfig *rest.Config, kubeclientset kubernetes.Interface, wfclientset wfclientset.Interface, namespace string, managedNamespace string, executorImage, executorImagePullPolicy, containerRuntimeExecutor, configMap string) *WorkflowController {
 	eventBroadcaster := record.NewBroadcaster()
 	eventBroadcaster.StartLogging(log.Debugf)
 	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeclientset.CoreV1().Events(namespace)})
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index 5fee42038..14699c270 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -2700,20 +2700,22 @@ func (woc *wfOperationCtx) getArtifactRepositoryByRef(arRef *wfv1.ArtifactReposi
 }
 
 func (woc *wfOperationCtx) fetchWorkflowSpec() (*wfv1.WorkflowSpec, error) {
+	if woc.wf.Spec.WorkflowTemplateRef == nil {
+		return nil, fmt.Errorf("cannot fetch workflow spec without workflowTemplateRef")
+	}
+
 	var specHolder wfv1.WorkflowSpecHolder
 	var err error
 	// Logic for workflow refers Workflow template
-	if woc.wf.Spec.WorkflowTemplateRef != nil {
-		if woc.wf.Spec.WorkflowTemplateRef.ClusterScope {
-			specHolder, err = woc.controller.cwftmplInformer.Lister().Get(woc.wf.Spec.WorkflowTemplateRef.Name)
-		} else {
-			specHolder, err = woc.controller.wftmplInformer.Lister().WorkflowTemplates(woc.wf.Namespace).Get(woc.wf.Spec.WorkflowTemplateRef.Name)
-		}
-		if err != nil {
-			return nil, err
-		}
+	if woc.wf.Spec.WorkflowTemplateRef.ClusterScope {
+		specHolder, err = woc.controller.cwftmplInformer.Lister().Get(woc.wf.Spec.WorkflowTemplateRef.Name)
+	} else {
+		specHolder, err = woc.controller.wftmplInformer.Lister().WorkflowTemplates(woc.wf.Namespace).Get(woc.wf.Spec.WorkflowTemplateRef.Name)
 	}
-	return specHolder.GetWorkflowSpec().DeepCopy(), nil
+	if err != nil {
+		return nil, err
+	}
+	return specHolder.GetWorkflowSpec(), nil
 }
 
 func (woc *wfOperationCtx) loadExecutionSpec() (wfv1.TemplateReferenceHolder, wfv1.Arguments, error) {
@@ -2721,17 +2723,30 @@ func (woc *wfOperationCtx) loadExecutionSpec() (wfv1.TemplateReferenceHolder, wf
 	executionParameters := woc.wf.Spec.Arguments
 
 	if woc.wf.Spec.WorkflowTemplateRef == nil {
-		tmplRef := &wfv1.WorkflowStep{Template: woc.wfSpec.Entrypoint}
+		if woc.controller.Config.WorkflowRestrictions.MustUseReference() {
+			return nil, executionParameters, fmt.Errorf("workflows must use workflowTemplateRef to be executed when the controller is in reference mode")
+		}
+
+		tmplRef := &wfv1.WorkflowStep{Template: woc.wf.Spec.Entrypoint}
 		return tmplRef, executionParameters, nil
 	}
 
 	if woc.wf.Status.StoredWorkflowSpec == nil {
 		wftSpec, err := woc.fetchWorkflowSpec()
 		if err != nil {
-			return nil, wfv1.Arguments{}, err
+			return nil, executionParameters, err
 		}
 		woc.wf.Status.StoredWorkflowSpec = wftSpec
 		woc.updated = true
+	} else if woc.controller.Config.WorkflowRestrictions.MustNotChangeSpec() {
+		// If the controller is in reference mode, ensure that the stored spec is identical to the reference spec at every operation
+		wftSpec, err := woc.fetchWorkflowSpec()
+		if err != nil {
+			return nil, executionParameters, err
+		}
+		if woc.wf.Status.StoredWorkflowSpec.String() != wftSpec.String() {
+			return nil, executionParameters, fmt.Errorf("workflowTemplateRef reference may not change during execution when the controller is in reference mode")
+		}
 	}
 
 	woc.wfSpec = woc.wf.Status.StoredWorkflowSpec


Test patch:
diff --git a/test/test.go b/test/test.go
index df2634076..836c61743 100644
--- a/test/test.go
+++ b/test/test.go
@@ -51,6 +51,25 @@ func LoadWorkflowFromBytes(yamlBytes []byte) *wfv1.Workflow {
 	return &wf
 }
 
+// LoadTestWorkflow returns a workflow relative to the test file
+func LoadTestWorkflowTemplate(path string) *wfv1.WorkflowTemplate {
+	yamlBytes, err := ioutil.ReadFile(path)
+	if err != nil {
+		panic(err)
+	}
+	return LoadWorkflowTemplateFromBytes(yamlBytes)
+}
+
+// LoadWorkflowFromBytes returns a workflow unmarshalled from an yaml byte array
+func LoadWorkflowTemplateFromBytes(yamlBytes []byte) *wfv1.WorkflowTemplate {
+	var wf wfv1.WorkflowTemplate
+	err := yaml.Unmarshal(yamlBytes, &wf)
+	if err != nil {
+		panic(err)
+	}
+	return &wf
+}
+
 // LoadUnstructuredFromBytes returns an Unstructured unmarshalled from an yaml byte array
 func LoadUnstructuredFromBytes(yamlBytes []byte) *unstructured.Unstructured {
 	var un unstructured.Unstructured
diff --git a/workflow/controller/controller_test.go b/workflow/controller/controller_test.go
index bdb6f737b..b85bc1405 100644
--- a/workflow/controller/controller_test.go
+++ b/workflow/controller/controller_test.go
@@ -370,20 +370,14 @@ spec:
 `
 
 func TestCheckAndInitWorkflowTmplRef(t *testing.T) {
-	//_, controller := newController()
 	wf := unmarshalWF(wfWithTmplRef)
 	wftmpl := unmarshalWFTmpl(wfTmpl)
 	_, controller := newController(wf, wftmpl)
-	//_, err := controller.wfclientset.ArgoprojV1alpha1().WorkflowTemplates("default").Create(wftmpl)
-	//assert.NoError(t, err)
 	woc := wfOperationCtx{controller: controller,
 		wf: wf}
-	t.Run("WithWorkflowTmplRef", func(t *testing.T) {
-		_, _, err := woc.loadExecutionSpec()
-		assert.NoError(t, err)
-		assert.Equal(t, &wftmpl.Spec.WorkflowSpec, woc.wfSpec)
-
-	})
+	_, _, err := woc.loadExecutionSpec()
+	assert.NoError(t, err)
+	assert.Equal(t, &wftmpl.Spec.WorkflowSpec, woc.wfSpec)
 }
 
 func TestIsArchivable(t *testing.T) {
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index f6a617710..4981189ff 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -3687,3 +3687,55 @@ func TestPanicMetric(t *testing.T) {
 	}
 	assert.True(t, seen)
 }
+
+// Assert Workflows cannot be run without using workflowTemplateRef in reference mode
+func TestControllerReferenceMode(t *testing.T) {
+	wf := unmarshalWF(globalVariablePlaceholders)
+	cancel, controller := newController()
+	defer cancel()
+	controller.Config.WorkflowRestrictions = &config.WorkflowRestrictions{}
+	controller.Config.WorkflowRestrictions.TemplateReferencing = config.TemplateReferencingStrict
+	woc := newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	assert.Equal(t, wfv1.NodeError, woc.wf.Status.Phase)
+	assert.Equal(t, "workflows must use workflowTemplateRef to be executed when the controller is in reference mode", woc.wf.Status.Message)
+
+	controller.Config.WorkflowRestrictions.TemplateReferencing = config.TemplateReferencingSecure
+	woc = newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	assert.Equal(t, wfv1.NodeError, woc.wf.Status.Phase)
+	assert.Equal(t, "workflows must use workflowTemplateRef to be executed when the controller is in reference mode", woc.wf.Status.Message)
+
+	controller.Config.WorkflowRestrictions = nil
+	woc = newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
+}
+
+func TestValidReferenceMode(t *testing.T) {
+	wf := test.LoadTestWorkflow("testdata/workflow-template-ref.yaml")
+	wfTmpl := test.LoadTestWorkflowTemplate("testdata/workflow-template-submittable.yaml")
+	cancel, controller := newController(wf, wfTmpl)
+	defer cancel()
+	controller.Config.WorkflowRestrictions = &config.WorkflowRestrictions{}
+	controller.Config.WorkflowRestrictions.TemplateReferencing = config.TemplateReferencingSecure
+	woc := newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
+
+	// Change stored Workflow Spec
+	woc.wf.Status.StoredWorkflowSpec.Entrypoint = "different"
+	woc.operate()
+	assert.Equal(t, wfv1.NodeError, woc.wf.Status.Phase)
+	assert.Equal(t, "workflowTemplateRef reference may not change during execution when the controller is in reference mode", woc.wf.Status.Message)
+
+	controller.Config.WorkflowRestrictions.TemplateReferencing = config.TemplateReferencingStrict
+	woc = newWorkflowOperationCtx(wf, controller)
+	woc.operate()
+	assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
+
+	// Change stored Workflow Spec
+	woc.wf.Status.StoredWorkflowSpec.Entrypoint = "different"
+	woc.operate()
+	assert.Equal(t, wfv1.NodeRunning, woc.wf.Status.Phase)
+}
diff --git a/workflow/controller/testdata/workflow-template-ref.yaml b/workflow/controller/testdata/workflow-template-ref.yaml
new file mode 100644
index 000000000..c08109716
--- /dev/null
+++ b/workflow/controller/testdata/workflow-template-ref.yaml
@@ -0,0 +1,8 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: workflow-template-hello-world-
+  namespace: test
+spec:
+  workflowTemplateRef:
+    name: workflow-template-submittable
diff --git a/workflow/controller/testdata/workflow-template-submittable.yaml b/workflow/controller/testdata/workflow-template-submittable.yaml
new file mode 100644
index 000000000..a6b786863
--- /dev/null
+++ b/workflow/controller/testdata/workflow-template-submittable.yaml
@@ -0,0 +1,13 @@
+apiVersion: argoproj.io/v1alpha1
+kind: WorkflowTemplate
+metadata:
+  name: workflow-template-submittable
+  namespace: test
+spec:
+  entrypoint: whalesay-template
+  templates:
+    - name: whalesay-template
+      container:
+        image: docker/whalesay
+        command: [cowsay]
+        args: ['hello']