[WIP] Add TLS/SASL authentication support for Kafka functions

[aliok]

Summary

• Extend KafkaConfig with securityProtocol, tls, and sasl fields in func.yaml
• Deployer emits KAFKA_SECURITY_PROTOCOL, KAFKA_TLS_*, KAFKA_SASL_* env vars
• {{ secret:name:key }} syntax supported for sasl.user and sasl.password
• Propagate TLS/SASL env vars in both docker and host runners
• Validation: protocol enum, TLS requires SSL/SASL_SSL, SASL requires SASL_*/mechanism enum

Depends on knative-extensions/func-go#186

func.yaml example (SASL_SSL)

run:
  kafka:
    brokers: "broker:9093"
    topic: "my-topic"
    consumerGroup: "my-group"
    securityProtocol: "SASL_SSL"
    tls:
      caCert: "/etc/kafka/ca/ca.crt"
    sasl:
      mechanism: "SCRAM-SHA-512"
      user: "my-user"
      password: "{{ secret:my-user:password }}"
  volumes:
    - secret: my-cluster-ca-cert
      path: /etc/kafka/ca

Verification instructions (Kind + Strimzi)

Prerequisites

• kind, kubectl, Go 1.25+, Docker

1. Build the func CLI

Both repos have un-merged branches. Build the CLI from the kafka-tls-sasl branch:

cd ~/go/src/knative.dev/func
git checkout kafka-tls-sasl
go build -o /tmp/func-local ./cmd/func

2. Patch the scaffolding to use the func-go fork

The func-go dependency lives in the scaffolding's go.mod (embedded in the CLI), not the function's go.mod. Add a replace directive, re-tidy, regenerate the embedded filesystem, and rebuild:

cd ~/go/src/knative.dev/func/templates/go/scaffolding/instanced-cloudevents

# Add replace directive pointing to your fork branch
go mod edit -replace "knative.dev/func-go=github.com/aliok/func-go@kafka-tls-sasl"

# Tidy — needs a stub ./f module (scaffolding uses replace function => ./f)
mkdir -p f
printf 'module function\ngo 1.25.0\nrequire github.com/cloudevents/sdk-go/v2 v2.16.2' > f/go.mod
echo 'package function' > f/f.go
go mod tidy
rm -rf f

# Regenerate embedded filesystem and rebuild CLI
cd ~/go/src/knative.dev/func
go generate ./...
go build -o /tmp/func-local ./cmd/func

Note: This step is only needed while the func-go changes are un-merged. Once func-go releases a new version with TLS/SASL, the scaffolding will reference it directly and this step goes away.

3. Create a Kind cluster with Knative

kind create cluster --name kafka-tls-test

kubectl apply -f https://github.com/knative/serving/releases/latest/download/serving-crds.yaml
kubectl apply -f https://github.com/knative/serving/releases/latest/download/serving-core.yaml

kubectl apply -f https://github.com/knative/net-kourier/releases/latest/download/kourier.yaml
kubectl patch configmap/config-network \
  --namespace knative-serving \
  --type merge \
  --patch '{"data":{"ingress-class":"kourier.ingress.sigs.k8s.io"}}'

kubectl wait --for=condition=Ready pods --all -n knative-serving --timeout=120s

4. Install Strimzi with a TLS+SASL listener

kubectl create namespace kafka
kubectl apply -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka
kubectl wait --for=condition=Ready pods --all -n kafka --timeout=120s

kubectl apply -n kafka -f - <<EOF
apiVersion: kafka.strimzi.io/v1
kind: KafkaNodePool
metadata:
  name: dual-role
  labels:
    strimzi.io/cluster: my-cluster
spec:
  replicas: 1
  roles:
    - controller
    - broker
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 1Gi
        deleteClaim: true
---
apiVersion: kafka.strimzi.io/v1
kind: Kafka
metadata:
  name: my-cluster
  annotations:
    strimzi.io/node-pools: enabled
    strimzi.io/kraft: enabled
spec:
  kafka:
    version: 4.2.0
    authorization:
      type: simple
      superUsers:
        - ANONYMOUS
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: scram-sha-512
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
  entityOperator:
    topicOperator: {}
    userOperator: {}
EOF

kubectl wait kafka/my-cluster --for=condition=Ready --timeout=300s -n kafka

5. Create a KafkaUser and topic

kubectl apply -n kafka -f - <<EOF
apiVersion: kafka.strimzi.io/v1
kind: KafkaUser
metadata:
  name: my-kafka-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: scram-sha-512
  authorization:
    type: simple
    acls:
      - resource:
          type: topic
          name: test-topic
          patternType: literal
        operations: [Read, Describe]
        host: "*"
      - resource:
          type: group
          name: my-kafka-func-group
          patternType: literal
        operations: [Read]
        host: "*"
EOF

kubectl apply -n kafka -f - <<EOF
apiVersion: kafka.strimzi.io/v1
kind: KafkaTopic
metadata:
  name: test-topic
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 1
  replicas: 1
EOF

kubectl wait kafkauser/my-kafka-user --for=condition=Ready --timeout=60s -n kafka

6. Create the function

mkdir /tmp/my-kafka-tls-func && cd /tmp/my-kafka-tls-func
/tmp/func-local create -l go -t cloudevents
go mod tidy

7. Configure func.yaml

Copy secrets to the function namespace:

kubectl get secret my-cluster-cluster-ca-cert -n kafka -o json \
  | jq 'del(.metadata.namespace,.metadata.resourceVersion,.metadata.uid,.metadata.creationTimestamp,.metadata.ownerReferences)' \
  | kubectl apply -n default -f -

kubectl get secret my-kafka-user -n kafka -o json \
  | jq 'del(.metadata.namespace,.metadata.resourceVersion,.metadata.uid,.metadata.creationTimestamp,.metadata.ownerReferences)' \
  | kubectl apply -n default -f -

Edit func.yaml:

specVersion: 0.36.0
name: my-kafka-tls-func
runtime: go
created: ...
invoke: cloudevent
deploy:
  options:
    scale:
      min: 1
run:
  kafka:
    brokers: "my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9093"
    topic: "test-topic"
    consumerGroup: "my-kafka-func-group"
    securityProtocol: "SASL_SSL"
    tls:
      caCert: "/etc/kafka/ca/ca.crt"
    sasl:
      mechanism: "SCRAM-SHA-512"
      user: "my-kafka-user"
      password: "{{ secret:my-kafka-user:password }}"
  volumes:
    - secret: my-cluster-cluster-ca-cert
      path: /etc/kafka/ca

8. Deploy and verify

FUNC_REGISTRY=ttl.sh/my-kafka-tls-test /tmp/func-local deploy --build --verbose

kubectl wait pods -l serving.knative.dev/service=my-kafka-tls-func \
  --for=condition=Ready --timeout=120s

Check env vars on the pod:

kubectl get pods -l serving.knative.dev/service=my-kafka-tls-func -o json \
  | jq '.items[0].spec.containers[] | select(.name=="user-container") | .env[] | select(.name | startswith("KAFKA"))'

9. Tail logs and send a test message

# In a separate terminal
kubectl logs -l serving.knative.dev/service=my-kafka-tls-func -c user-container -f

# Send a test message
kubectl run kafka-producer -n kafka \
  --image=quay.io/strimzi/kafka:latest-kafka-4.2.0 \
  --restart=Never \
  --command -- sh -c \
  'echo "Hello from authenticated Kafka!" | bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --topic test-topic'

kubectl wait pod/kafka-producer -n kafka --for=jsonpath='{.status.phase}'=Succeeded --timeout=60s
kubectl delete pod kafka-producer -n kafka

Expected log output:

{"level":"debug","path":"/etc/kafka/ca/ca.crt","message":"loaded kafka CA certificate"}
{"level":"debug","mechanism":"SCRAM-SHA-512","user":"my-kafka-user","message":"kafka SASL configured"}
{"level":"info","message":"kafka consumer ready (partitions assigned)"}

Cleanup

kubectl delete ksvc my-kafka-tls-func
kubectl delete secret my-cluster-cluster-ca-cert my-kafka-user -n default
kubectl delete kafkauser my-kafka-user -n kafka
kubectl delete kafkatopic test-topic -n kafka
kubectl delete kafka my-cluster -n kafka
kubectl delete -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka
kubectl delete namespace kafka
kind delete cluster --name kafka-tls-test
rm -rf /tmp/my-kafka-tls-func /tmp/func-local

[knative-prow]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by: aliok
Once this PR has been reviewed and has the lgtm label, please assign jrangelramos for approval. For more information see the Code Review Process.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[Copilot]

Pull request overview

This PR extends the function-level Kafka configuration to support TLS and SASL authentication, and propagates the resulting settings into the various deploy/run paths (Kubernetes/Knative deployers and local runners).

Changes:

• Extend run.kafka schema with securityProtocol, tls, and sasl (including validation).
• Emit additional KAFKA_SECURITY_PROTOCOL, KAFKA_TLS_*, and KAFKA_SASL_* environment variables during deployment/run.
• Support {{ secret:name:key }}-style value references for Kafka SASL user/password in the k8s deployer env var generation.

Reviewed changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
pkg/knative/deployer.go	Updates Knative deploy path to use the new error-returning Kafka env injection (and track referenced resources).
pkg/k8s/deployer.go	Extends Kafka env var generation to include TLS/SASL fields and secret/configMap key refs for SASL values.
pkg/k8s/deployer_test.go	Adapts existing tests to new signature and adds coverage for TLS/SASL and secret-ref cases.
pkg/functions/runner.go	Propagates Kafka TLS/SASL env vars for the host runner (func run).
pkg/functions/function.go	Adds new Kafka config types/fields and validation rules for protocol/TLS/SASL combinations.
pkg/functions/function_test.go	Adds validation test cases for the new Kafka TLS/SASL config combinations.
pkg/docker/runner.go	Propagates Kafka TLS/SASL env vars for the Docker runner.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 8 out of 9 changed files in this pull request and generated no new comments.

Suppressed comments (4)

schema/func_yaml-schema.json:313

• schema/func_yaml-schema.json appears to be a generated artifact (see Makefile:393-403 and schema/generator/main.go:23-55). Editing it manually can easily drift from the Go struct tags in pkg/functions/function.go; it should be regenerated via make schema-generate and the regenerated output committed instead of hand-maintaining this section.

				"securityProtocol": {
					"enum": [
						"PLAINTEXT",
						"SSL",
						"SASL_PLAINTEXT",
						"SASL_SSL"
					],
					"type": "string",
					"description": "Security protocol: PLAINTEXT SSL SASL_PLAINTEXT or SASL_SSL"
				},
				"tls": {
					"$schema": "http://json-schema.org/draft-04/schema#",
					"$ref": "#/definitions/KafkaTLS",
					"description": "TLS configuration for SSL or SASL_SSL"
				},
				"sasl": {
					"$schema": "http://json-schema.org/draft-04/schema#",
					"$ref": "#/definitions/KafkaSASL",
					"description": "SASL authentication for SASL_PLAINTEXT or SASL_SSL"
				}

pkg/functions/function.go:265

• Kafka SASL template refs are validated against templateRefPattern, but the current regex doesn’t allow trailing whitespace after the closing }} while the deploy-time parser (pkg/k8s/deployer.go) effectively tolerates it via TrimSpace/Trim. This can cause valid-looking values (e.g. "{{ secret:n:k }} ") to fail validation even though they’d parse during deploy. Consider trimming before validation and allowing optional trailing whitespace so validation and parsing accept the same inputs.

var templateRefPattern = regexp.MustCompile(`^\{\{\s*(secret|configMap):[^:]+:[^:]+\s*\}\}$`)

func validateTemplateRef(field, value string) (errors []string) {
	if strings.HasPrefix(value, "{{") && !templateRefPattern.MatchString(value) {
		errors = append(errors, fmt.Sprintf("%s has invalid reference format %q, expected {{ secret:name:key }} or {{ configMap:name:key }}", field, value))
	}

pkg/k8s/deployer.go:808

• appendKafkaEnvValue only treats values as template refs when the raw string starts with "{{", so leading whitespace (e.g. " {{ secret:n:k }}") will silently be treated as a literal and won’t produce a ValueFrom SecretKeyRef/ConfigMapKeyRef. Since validation/parsing already tolerates whitespace elsewhere, it’s safer to TrimSpace before checking for {{ and to parse the inner secret|configMap:name:key by stripping {{/}} explicitly.

func appendKafkaEnvValue(envVars []corev1.EnvVar, name, value string, referencedSecrets, referencedConfigMaps *sets.Set[string]) ([]corev1.EnvVar, error) {
	if strings.HasPrefix(value, "{{") {
		if !strings.HasSuffix(strings.TrimSpace(value), "}}") {
			return nil, fmt.Errorf("invalid reference format %q, expected {{ secret:name:key }} or {{ configMap:name:key }}", value)
		}
		slices := strings.Split(strings.Trim(value, "{} "), ":")
		if len(slices) == 3 {
			valueFrom, err := createEnvVarSource(slices, referencedSecrets, referencedConfigMaps)
			if err != nil {
				return nil, err
			}
			return append(envVars, corev1.EnvVar{Name: name, ValueFrom: valueFrom}), nil
		}
		return nil, fmt.Errorf("invalid reference format %q, expected {{ secret:name:key }} or {{ configMap:name:key }}", value)
	}
	return append(envVars, corev1.EnvVar{Name: name, Value: value}), nil

pkg/functions/runner.go:344

• This adds new Kafka TLS/SASL environment variable propagation for the host runner, but there are already unit tests covering Kafka env construction (pkg/functions/runner_test.go:42-78) and they don’t assert the new variables. Adding a test case for SecurityProtocol/TLS/SASL here would prevent regressions and keep runner behavior aligned with the k8s deployer tests.

		if k.SecurityProtocol != "" && k.SecurityProtocol != "PLAINTEXT" {
			env = append(env, "KAFKA_SECURITY_PROTOCOL="+k.SecurityProtocol)
		}
		if k.TLS != nil {
			if k.TLS.CACert != "" {
				env = append(env, "KAFKA_TLS_CA_CERT="+k.TLS.CACert)
			}
			if k.TLS.ClientCert != "" {
				env = append(env, "KAFKA_TLS_CLIENT_CERT="+k.TLS.ClientCert)
			}
			if k.TLS.ClientKey != "" {
				env = append(env, "KAFKA_TLS_CLIENT_KEY="+k.TLS.ClientKey)
			}
			if k.TLS.SkipVerify {
				env = append(env, "KAFKA_TLS_SKIP_VERIFY=true")
			}
		}
		if k.SASL != nil {
			if k.SASL.Mechanism != "" {
				env = append(env, "KAFKA_SASL_MECHANISM="+k.SASL.Mechanism)
			}
			if k.SASL.User != "" {
				env = append(env, "KAFKA_SASL_USER="+k.SASL.User)
			}
			if k.SASL.Password != "" {
				env = append(env, "KAFKA_SASL_PASSWORD="+k.SASL.Password)
			}
		}

[Copilot]

Pull request overview

Copilot reviewed 8 out of 9 changed files in this pull request and generated no new comments.

Suppressed comments (7)

pkg/functions/runner.go:324

• The host runner's new TLS/SASL environment branches are untested even though buildRunnerEnv already has focused Kafka tests. Add cases covering protocol, TLS fields/skip verification, SASL credentials, and explicit PLAINTEXT precedence over an inherited protocol.

		if k.TLS != nil {
			if k.TLS.CACert != "" {
				env = append(env, "KAFKA_TLS_CA_CERT="+k.TLS.CACert)
			}
			if k.TLS.ClientCert != "" {

pkg/functions/function.go:247

• SASL protocols can currently pass validation with no sasl block, or with an empty user/password, even though the dependent runtime requires both credentials and returns an error at startup otherwise. Require the SASL block and both credential fields whenever a SASL protocol is selected.

	if kafka.SASL != nil {
		if kafka.SecurityProtocol != "SASL_PLAINTEXT" && kafka.SecurityProtocol != "SASL_SSL" {
			errors = append(errors, "run.kafka.sasl requires securityProtocol SASL_PLAINTEXT or SASL_SSL")
		}
		validMechanisms := map[string]bool{"": true, "PLAIN": true, "SCRAM-SHA-256": true, "SCRAM-SHA-512": true}

pkg/functions/runner.go:319

• An explicit securityProtocol: PLAINTEXT is not added to the host process environment. Because buildRunnerEnv starts with os.Environ(), an inherited KAFKA_SECURITY_PROTOCOL such as SASL_SSL remains effective and changes the configured function's protocol. Emit every explicitly configured protocol; only omit the empty default.

This issue also appears on line 320 of the same file.

		if k.SecurityProtocol != "" && k.SecurityProtocol != "PLAINTEXT" {
			env = append(env, "KAFKA_SECURITY_PROTOCOL="+k.SecurityProtocol)
		}

pkg/functions/function.go:241

• A configuration with only one of clientCert or clientKey passes Function.Validate, but the Kafka runtime rejects that pair before connecting. Validate that these mutual-TLS fields are either both set or both empty so deployment fails early with a configuration error.

This issue also appears on line 243 of the same file.

	if kafka.TLS != nil {
		if kafka.SecurityProtocol != "SSL" && kafka.SecurityProtocol != "SASL_SSL" {
			errors = append(errors, "run.kafka.tls requires securityProtocol SSL or SASL_SSL")
		}
	}

pkg/k8s/deployer.go:753

• An explicit securityProtocol: PLAINTEXT is omitted here, so a conflicting KAFKA_SECURITY_PROTOCOL from run.envs or the image remains effective. This also makes the dedicated Kafka field override run.envs for secure protocols but not for PLAINTEXT; emit every non-empty configured protocol.

	if kafka.SecurityProtocol != "" && kafka.SecurityProtocol != "PLAINTEXT" {
		envVars = append(envVars, corev1.EnvVar{Name: "KAFKA_SECURITY_PROTOCOL", Value: kafka.SecurityProtocol})
	}

pkg/docker/runner.go:311

• An explicit securityProtocol: PLAINTEXT is omitted here, so a conflicting KAFKA_SECURITY_PROTOCOL from run.envs or the image remains effective. This also makes dedicated Kafka configuration override run.envs for secure protocols but not for PLAINTEXT; emit every non-empty configured protocol.

		if k.SecurityProtocol != "" && k.SecurityProtocol != "PLAINTEXT" {
			c.Env = append(c.Env, "KAFKA_SECURITY_PROTOCOL="+k.SecurityProtocol)
		}

pkg/k8s/deployer_test.go:598

• This loop only checks an env var when it happens to be present, so the test still passes if either KAFKA_SASL_USER or KAFKA_SASL_PASSWORD is omitted. Look up both names explicitly and fail when either is missing before checking its SecretKeyRef.

	for _, ev := range got {
		if ev.Name == "KAFKA_SASL_USER" {
			if ev.ValueFrom == nil || ev.ValueFrom.SecretKeyRef == nil {
				t.Fatal("KAFKA_SASL_USER should have ValueFrom with SecretKeyRef")
			}