Add support for SASL and SSL (#534)

* Add Data plane security module and Kubernetes provider Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Add Auth configurations to contract Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Add contol plane security module Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Integrate security module in Broker and Sink reconciler Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Add control plane E2E tests Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Update proto schema and use PEM format Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Move data plane to PEM certificates format Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Extend E2E test by sending events Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Run E2E test multiple times to reduce flakiness Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Improve comment Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Refresh third party license list Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Update docs in proto definition Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Rename E2E test functions Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * auth.secret.name -> auth.secret.ref.name Make reference explicit. Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Add boilerplate to reconciler_test.go Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Lint and update codegen Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Change comment to Kubernetes resource reference Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Remove Nullable annotations Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Ensure TypeMeta when Tracker OnChanged is called Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Remove unused Sarama logger adapter function Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Use symlinks to testdata certs Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * KafkaSink supports SASL / SSL Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Move bootstrap servers config in one place Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Add KafkaSink E2E tests Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Refresh third party file Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Add Validation test for secret reference Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Update codegen Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Update KafkaSink CRD schema Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Test security config functions Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Test security config and scram modules Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Rename data plane roles Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Maven exclusion directly in parent pom Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com> * Refactor credentials fetching Signed-off-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com>
knative-extensions · Jan 27, 2021 · 9fd3feb · 9fd3feb
1 parent 1806e58
commit 9fd3feb
Show file tree

Hide file tree

Showing 173 changed files with 15,730 additions and 1,533 deletions.
diff --git a/control-plane/config/200-controller-cluster-role.yaml b/control-plane/config/200-controller-cluster-role.yaml
@@ -60,6 +60,15 @@ rules:
       - patch
       - watch
 
+  - apiGroups:
+      - "*"
+    resources:
+      - secrets
+    verbs:
+      - list
+      - get
+      - watch
+
   # Eventing resources and statuses we care about
   - apiGroups:
       - "eventing.knative.dev"

diff --git a/control-plane/config/sink/100-kafka-sink.yaml b/control-plane/config/sink/100-kafka-sink.yaml
@@ -82,6 +82,25 @@ spec:
                     - structured
                     - binary
                   default: structured
+                auth:
+                  description: 'Auth configurations'
+                  type: object
+                  properties:
+                    secret:
+                      description: 'Auth secret'
+                      type: object
+                      properties:
+                        ref:
+                          # TODO add format in description (?)
+                          description: |
+                            Secret reference.
+                          type: object
+                          required:
+                            - name
+                          properties:
+                            name:
+                              description: 'Secret name'
+                              type: string
             status:
               description: 'Status represents the current state of the KafkaSink. This data may be out of date.'
               type: object

diff --git a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults.go b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_defaults.go
@@ -27,7 +27,7 @@ func (ks *KafkaSink) SetDefaults(ctx context.Context) {
 func (kss *KafkaSinkSpec) SetDefaults(ctx context.Context) {
 	defaultMode := ModeStructured
 
-	if kss.ContentMode == nil {
+	if kss.ContentMode == nil || *kss.ContentMode == "" {
 		kss.ContentMode = &defaultMode
 	}
 }
diff --git a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_types.go b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_types.go
@@ -23,6 +23,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/kmeta"
 )
 
 const (
@@ -60,6 +61,7 @@ var _ apis.Defaultable = (*KafkaSink)(nil)
 var _ runtime.Object = (*KafkaSink)(nil)
 var _ duckv1.KRShaped = (*KafkaSink)(nil)
 var _ apis.Convertible = (*KafkaSink)(nil)
+var _ kmeta.OwnerRefable = (*KafkaSink)(nil)
 
 // KafkaSinkSpec defines the desired state of the Kafka Sink.
 type KafkaSinkSpec struct {
@@ -91,6 +93,24 @@ type KafkaSinkSpec struct {
 	//
 	// +optional
 	ContentMode *string `json:"contentMode,omitempty"`
+
+	// Auth configurations.
+	Auth *Auth `json:"auth,omitempty"`
+}
+
+type Auth struct {
+	// Auth Secret
+	Secret *Secret `json:"secret,omitempty"`
+}
+
+type Secret struct {
+	// Secret reference for SASL and SSL configurations.
+	Ref *SecretReference `json:"ref,omitempty"`
+}
+
+type SecretReference struct {
+	// Secret name.
+	Name string `json:"name"`
 }
 
 // KafkaSinkStatus represents the current state of the KafkaSink.
@@ -129,3 +149,7 @@ func (ks *KafkaSink) GetUntypedSpec() interface{} {
 func (ks *KafkaSink) GetStatus() *duckv1.Status {
 	return &ks.Status.Status
 }
+
+func (kss KafkaSinkSpec) HasAuthConfig() bool {
+	return kss.Auth != nil && kss.Auth.Secret != nil && kss.Auth.Secret.Ref != nil
+}
diff --git a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation.go b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation.go
@@ -62,6 +62,10 @@ func (kss *KafkaSinkSpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(apis.ErrInvalidValue(*kss.NumPartitions, "numPartitions"))
 	}
 
+	if kss.HasAuthConfig() && kss.Auth.Secret.Ref.Name == "" {
+		errs = errs.Also(apis.ErrInvalidValue("", "auth.secret.ref.name"))
+	}
+
 	return errs
 }
 

diff --git a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation_test.go b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_validation_test.go
@@ -127,6 +127,19 @@ func TestKafkaSink_Validate(t *testing.T) {
 			ctx:  context.Background(),
 			want: apis.ErrInvalidValue("-10", "spec.replicationFactor"),
 		},
+		{
+			name: "invalid secret name",
+			ks: &KafkaSink{
+				Spec: KafkaSinkSpec{
+					Topic:            "topic-name-1",
+					BootstrapServers: []string{"broker-1:9092"},
+					ContentMode:      pointer.StringPtr(ModeStructured),
+					Auth:             &Auth{Secret: &Secret{Ref: &SecretReference{}}},
+				},
+			},
+			ctx:  context.Background(),
+			want: apis.ErrInvalidValue("", "spec.auth.secret.ref.name"),
+		},
 		{
 			name: "immutable replication factor",
 			ks: &KafkaSink{

diff --git a/control-plane/pkg/apis/eventing/v1alpha1/zz_generated.deepcopy.go b/control-plane/pkg/apis/eventing/v1alpha1/zz_generated.deepcopy.go