[release-1.12] Add OIDC support

control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel.yaml

@@ -86,6 +86,9 @@ spec:
                           type: string
                         CACerts:
                           type: string
+                        audience:
+                          description: Audience is the OIDC audience for the deadLetterSink.
+                          type: string
                     retry:
                       description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                       type: integer
@@ -132,6 +135,9 @@ spec:
                                 type: string
                               CACerts:
                                 type: string
+                              audience:
+                                description: Audience is the OIDC audience for the deadLetterSink.
+                                type: string
                           retry:
                             description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                             type: integer
@@ -147,15 +153,28 @@ spec:
                       replyCACerts:
                         description: replyCACerts is the CA certs to trust for the reply.
                         type: string
+                      replyAudience:
+                        description: ReplyAudience is the OIDC audience for the replyUri.
+                        type: string
                       subscriberUri:
                         description: SubscriberURI is the endpoint for the subscriber
                         type: string
                       subscriberCACerts:
                         description: SubscriberCACerts is the CA certs to trust for the subscriber.
                         type: string
+                      subscriberAudience:
+                        description: SubscriberAudience is the OIDC audience for the subscriberUri.
+                        type: string
                       uid:
                         description: UID is used to understand the origin of the subscriber.
                         type: string
+                      auth:
+                        description: Auth provides the relevant information for OIDC authentication.
+                        type: object
+                        properties:
+                          serviceAccountName:
+                            description: ServiceAccountName is the name of the generated service account used for this components OIDC authentication.
+                            type: string
             status:
               description: Status represents the current state of the KafkaChannel. This data may be out of date.
               type: object
@@ -239,6 +258,9 @@ spec:
                   type: string
                 deadLetterSinkCACerts:
                   type: string
+                deadLetterSinkAudience:
+                  description: OIDC audience of the dead letter sink.
+                  type: string
                 observedGeneration:
                   description: ObservedGeneration is the 'Generation' of the Service that was last processed by the controller.
                   type: integer
@@ -262,6 +284,13 @@ spec:
                       uid:
                         description: UID is used to understand the origin of the subscriber.
                         type: string
+                      auth:
+                        description: Auth provides the relevant information for OIDC authentication.
+                        type: object
+                        properties:
+                          serviceAccountName:
+                            description: ServiceAccountName is the name of the generated service account used for this components OIDC authentication.
+                            type: string
       additionalPrinterColumns:
         - name: Ready
           type: string

control-plane/config/eventing-kafka-broker/100-source/100-kafka-source.yaml

@@ -87,9 +87,6 @@ spec:
                       description: DeadLetterSink is the sink receiving event that could not be sent to a destination.
                       type: object
                       properties:
-                        CACerts:
-                          description: CACerts are Certification Authority (CA) certificates in PEM format according to https://www.rfc-editor.org/rfc/rfc7468. If set, these CAs are appended to the set of CAs provided by the Addressable target, if any.
-                          type: string
                         ref:
                           description: Ref points to an Addressable.
                           type: object
@@ -118,6 +115,12 @@ spec:
                         uri:
                           description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
                           type: string
+                        CACerts:
+                          description: CACerts are Certification Authority (CA) certificates in PEM format according to https://www.rfc-editor.org/rfc/rfc7468. If set, these CAs are appended to the set of CAs provided by the Addressable target, if any.
+                          type: string
+                        audience:
+                          description: Audience is the OIDC audience for the deadLetterSink.
+                          type: string
                     retry:
                       description: Retry is the minimum number of retries the sender should attempt when sending an event before moving it to the dead letter sink.
                       type: integer
@@ -271,9 +274,6 @@ spec:
                   description: Sink is a reference to an object that will resolve to a uri to use as the sink.
                   type: object
                   properties:
-                    CACerts:
-                      description: CACerts are Certification Authority (CA) certificates in PEM format according to https://www.rfc-editor.org/rfc/rfc7468. If set, these CAs are appended to the set of CAs provided by the Addressable target, if any.
-                      type: string
                     ref:
                       description: Ref points to an Addressable.
                       type: object
@@ -302,6 +302,12 @@ spec:
                     uri:
                       description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
                       type: string
+                    CACerts:
+                      description: CACerts are Certification Authority (CA) certificates in PEM format according to https://www.rfc-editor.org/rfc/rfc7468. If set, these CAs are appended to the set of CAs provided by the Addressable target, if any.
+                      type: string
+                    audience:
+                      description: Audience is the OIDC audience for the sink.
+                      type: string
                 topics:
                   description: Topic topics to consume messages from
                   type: array
@@ -392,6 +398,16 @@ spec:
                 sinkUri:
                   description: SinkURI is the current active sink URI that has been configured for the Source.
                   type: string
+                sinkAudience:
+                  description: SinkAudience is the OIDC audience of the sink.
+                  type: string
+                auth:
+                  description: Auth provides the relevant information for OIDC authentication.
+                  type: object
+                  properties:
+                    serviceAccountName:
+                      description: ServiceAccountName is the name of the generated service account used for this components OIDC authentication.
+                      type: string
       subresources:
         status: {}
         scale:

control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml

@@ -140,6 +140,13 @@ rules:
       - update
       - create
       - delete
+  # To grant NamespacedBroker permissions to create OIDC tokens
+  - apiGroups:
+      - ""
+    resources:
+      - serviceaccounts/token
+    verbs:
+      - create
 
   # Scheduler permissions
   - apiGroups:

control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go

@@ -100,6 +100,10 @@ type ConsumerGroupSpec struct {
 	// More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#label-selectors
 	// +optional
 	Selector map[string]string `json:"selector,omitempty" protobuf:"bytes,2,rep,name=selector"`
+
+	// OIDCServiceAccountName is the name of service account used for this components
+	// OIDC authentication.
+	OIDCServiceAccountName *string `json:"oidcServiceAccountName,omitempty"`
 }
 
 type ConsumerGroupStatus struct {
@@ -120,6 +124,10 @@ type ConsumerGroupStatus struct {
 	// +optional
 	SubscriberCACerts *string `json:"subscriberCACerts,omitempty"`
 
+	// SubscriberAudience is the OIDC audience for the resolved URI
+	// +optional
+	SubscriberAudience *string `json:"subscriberAudience,omitempty"`
+
 	// DeliveryStatus contains a resolved URL to the dead letter sink address, and any other
 	// resolved delivery options.
 	eventingduckv1.DeliveryStatus `json:",inline"`

control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go

@@ -103,6 +103,10 @@ type ConsumerSpec struct {
 
 	// PodBind represents a reference to the pod in which the consumer should be placed.
 	PodBind *PodBind `json:"podBind"`
+
+	// OIDCServiceAccountName is the name of the generated service account
+	// used for this components OIDC authentication.
+	OIDCServiceAccountName *string `json:"oidcServiceAccountName,omitempty"`
 }
 
 type ReplyStrategy struct {
@@ -208,6 +212,10 @@ type ConsumerStatus struct {
 	// +optional
 	SubscriberCACerts *string `json:"subscriberCACerts,omitempty"`
 
+	// SubscriberAudience is the OIDC audience for the resolved URI
+	// +optional
+	SubscriberAudience *string `json:"subscriberAudience,omitempty"`
+
 	// DeliveryStatus contains a resolved URL to the dead letter sink address, and any other
 	// resolved delivery options.
 	eventingduck.DeliveryStatus `json:",inline"`

control-plane/pkg/apis/sources/v1beta1/kafka_lifecycle.go

@@ -46,6 +46,9 @@ const (
 	// KafkaConditionInitialOffsetsCommitted is True when the KafkaSource has committed the
 	// initial offset of all claims
 	KafkaConditionInitialOffsetsCommitted apis.ConditionType = "InitialOffsetsCommitted"
+
+	// KafkaConditionOIDCIdentityCreated has status True when the KafkaSource has created an OIDC identity.
+	KafkaConditionOIDCIdentityCreated apis.ConditionType = "OIDCIdentityCreated"
 )
 
 var (
@@ -54,6 +57,7 @@ var (
 		KafkaConditionDeployed,
 		KafkaConditionConnectionEstablished,
 		KafkaConditionInitialOffsetsCommitted,
+		KafkaConditionOIDCIdentityCreated,
 	)
 
 	kafkaCondSetLock = sync.RWMutex{}
@@ -91,6 +95,7 @@ func (s *KafkaSourceStatus) MarkSink(addr *duckv1.Addressable) {
 	if addr.URL != nil && !addr.URL.IsEmpty() {
 		s.SinkURI = addr.URL
 		s.SinkCACerts = addr.CACerts
+		s.SinkAudience = addr.Audience
 		KafkaSourceCondSet.Manage(s).MarkTrue(KafkaConditionSinkProvided)
 	} else {
 		KafkaSourceCondSet.Manage(s).MarkUnknown(KafkaConditionSinkProvided, "SinkEmpty", "Sink has resolved to empty.%s", "")
@@ -160,6 +165,22 @@ func (s *KafkaSourceStatus) MarkInitialOffsetNotCommitted(reason, messageFormat
 	KafkaSourceCondSet.Manage(s).MarkFalse(KafkaConditionInitialOffsetsCommitted, reason, messageFormat, messageA...)
 }
 
+func (s *KafkaSourceStatus) MarkOIDCIdentityCreatedSucceeded() {
+	KafkaSourceCondSet.Manage(s).MarkTrue(KafkaConditionOIDCIdentityCreated)
+}
+
+func (s *KafkaSourceStatus) MarkOIDCIdentityCreatedSucceededWithReason(reason, messageFormat string, messageA ...interface{}) {
+	KafkaSourceCondSet.Manage(s).MarkTrueWithReason(KafkaConditionOIDCIdentityCreated, reason, messageFormat, messageA...)
+}
+
+func (s *KafkaSourceStatus) MarkOIDCIdentityCreatedFailed(reason, messageFormat string, messageA ...interface{}) {
+	KafkaSourceCondSet.Manage(s).MarkFalse(KafkaConditionOIDCIdentityCreated, reason, messageFormat, messageA...)
+}
+
+func (s *KafkaSourceStatus) MarkOIDCIdentityCreatedUnknown(reason, messageFormat string, messageA ...interface{}) {
+	KafkaSourceCondSet.Manage(s).MarkUnknown(KafkaConditionOIDCIdentityCreated, reason, messageFormat, messageA...)
+}
+
 func (s *KafkaSourceStatus) UpdateConsumerGroupStatus(status string) {
 	s.Claims = status
 }