[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

[knative-prow]

@aliok: The following test failed, say /retest to rerun all failed tests or /retest-required to rerun all mandatory failed tests:

Test name	Commit	Details	Required	Rerun command
unit-tests_func_main	57b137d	link	true	/test unit-tests

Your PR dashboard.

Details

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here.

[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.

[knative-prow-robot]

PR needs rebase.

Details

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository.