AppsCode
    KubeVault
    github-icon twitter-icon
    TRY NOW FREE

    New to KubeVault? Please start here.

    Manage Apache Kafka credentials using the KubeVault operator

    OpenBao’s kafka-database-plugin is a dynamic-credentials database plugin for Apache Kafka. The plugin provisions credentials by writing SASL/SCRAM user records via the franz-go AdminClient: each issued credential becomes a SCRAM-SHA-256 (default) or SCRAM-SHA-512 user on the cluster, and dynamically creates any configured ACLs.

    The same CRD shape is used both for the in-process kafka-database-plugin and for the hub-spoke remote-kafka-plugin; the difference is whether the Vault AppBinding referenced by SecretEngine.spec.vaultRef is marked deploymentMode: RemoteAgent (then the SecretEngine controller rewrites plugin_name to remote-kafka-plugin and attaches spoke_name).

    You need to be familiar with the following CRDs:

    Before you begin

    • Install KubeVault operator in your cluster from here.
    • Run an Apache Kafka cluster with SASL/SCRAM enabled. The plugin authenticates to the brokers using a SASL principal that has permission to write SCRAM credential records (AlterUserSCRAMs) and create/delete ACLs (CreateACLs, DeleteACLs). The broker listener must speak SASL_PLAINTEXT or SASL_SSL.
    $ kubectl create ns demo
namespace/demo created

    Vault Server

    Deploy a Vault Server using the KubeVault operator: Deploy Vault Server. The KubeVault operator will create an AppBinding wiring up Kubernetes auth.

    $ kubectl get appbinding -n demo vault -o yaml

    AppBinding for Apache Kafka

    Create an AppBinding pointing at the Kafka cluster. Unlike SQL-style engines, the Kafka engine uses seed broker addresses (host:port). The AppBinding can point to the Kafka cluster using either spec.clientConfig.url (e.g. a comma-separated list of brokers: broker1:9092,broker2:9092) or spec.clientConfig.service (referencing a Kubernetes Service). Any URI schemes (such as tcp:// or kafka://) are automatically sanitized into clean host:port pairs by the operator. The referenced Secret carries the SASL username and password of the SCRAM principal that has permission to manage user records.

    apiVersion: appcatalog.appscode.com/v1alpha1
kind: AppBinding
metadata:
  name: kafka
  namespace: demo
spec:
  clientConfig:
    url: kafka-0.kafka-broker.demo.svc:9092,kafka-1.kafka-broker.demo.svc:9092,kafka-2.kafka-broker.demo.svc:9092
  secret:
    kind: Secret
    name: kafka-cred
---
apiVersion: v1
kind: Secret
metadata:
  name: kafka-cred
  namespace: demo
type: kubernetes.io/basic-auth
stringData:
  username: admin
  password: change-me

    TLS Configuration via AppBinding: If the Kafka brokers terminate SASL on a TLS listener (SASL_SSL), configure TLS in the Kafka AppBinding via spec.clientConfig.caBundle or spec.tlsSecret (and optionally spec.clientConfig.insecureSkipTLSVerify for self-signed certificates). The KubeVault operator automatically detects TLS from the referenced AppBinding and configures TLS (use_tls, CA certificate, and verification flags) for the Kafka plugin. Do not specify useTLS or insecure on the SecretEngine object.

    Enable and Configure Kafka Secret Engine

    When a SecretEngine crd object is created, the KubeVault operator will enable a secret engine on a specified path and configure the secret engine with the given configuration.

    A sample SecretEngine for Kafka:

    apiVersion: engine.kubevault.com/v1alpha1
kind: SecretEngine
metadata:
  name: kafka-engine
  namespace: demo
spec:
  vaultRef:
    name: vault
  kafka:
    databaseRef:
      name: kafka
      namespace: demo
    pluginName: kafka-database-plugin     # optional; this is the default
    allowedRoles:
      - "*"
    mechanism: SCRAM-SHA-256              # optional; default. SCRAM-SHA-512 is also valid. PLAIN is rejected.

    Apply it and wait for STATUS=Success:

    $ kubectl apply -f kafka-engine.yaml
secretengine.engine.kubevault.com/kafka-engine created

$ kubectl get secretengines -n demo
NAME           STATUS    AGE
kafka-engine   Success   10s

    Use kubectl describe secretengine -n demo kafka-engine to inspect error events, if any.

    Create a KafkaRole

    A KafkaRole describes how the plugin should mint a dynamic credential. creationStatements is a single-element string slice holding a JSON role document. The mechanism field overrides the SecretEngine-level default on a per-role basis (SCRAM-SHA-256 or SCRAM-SHA-512). The acls array defines the permissions to grant to the generated user (e.g. for TOPIC, GROUP, or CLUSTER resources); if left empty, only the SCRAM user is created without any dynamic ACLs.

    apiVersion: engine.kubevault.com/v1alpha1
kind: KafkaRole
metadata:
  name: kafka-producer
  namespace: demo
spec:
  secretEngineRef:
    name: kafka-engine
  creationStatements:
    - '{"mechanism":"SCRAM-SHA-256","acls":[{"resource_type":"TOPIC","resource_name":"*","operation":"ALL","permission":"ALLOW"},{"resource_type":"GROUP","resource_name":"*","operation":"ALL","permission":"ALLOW"}]}'
  defaultTTL: 1h
  maxTTL: 24h

    Apply and verify:

    $ kubectl apply -f kafka-role.yaml
kafkarole.engine.kubevault.com/kafka-producer created

$ kubectl get kafkarole -n demo
NAME             STATUS    AGE
kafka-producer   Success   12s

    The role name in Vault follows the format k8s.{clusterName}.{metadata.namespace}.{metadata.name}, so you can verify directly with the Vault CLI:

    $ vault read your-database-path/roles/k8s.-.demo.kafka-producer
Key                      Value
---                      -----
creation_statements      [{"mechanism":"SCRAM-SHA-256","acls":[{"operation":"ALL","permission":"ALLOW","resource_name":"*","resource_type":"TOPIC"},{"operation":"ALL","permission":"ALLOW","resource_name":"*","resource_type":"GROUP"}]}]
db_name                  k8s.-.demo.kafka
default_ttl              1h
max_ttl                  24h

    Deleting the KafkaRole removes the role from Vault.

    Issue Kafka credentials

    Request a dynamic credential by creating a SecretAccessRequest:

    apiVersion: engine.kubevault.com/v1alpha1
kind: SecretAccessRequest
metadata:
  name: kafka-cred-rqst
  namespace: demo
spec:
  roleRef:
    kind: KafkaRole
    name: kafka-producer
  subjects:
    - kind: ServiceAccount
      name: demo-sa
      namespace: demo

    Approve it through the KubeVault CLI:

    $ kubectl vault approve secretaccessrequest kafka-cred-rqst -n demo
approved

    Once approved, the operator issues the credential, stores it in a Secret, and binds the listed subjects via a Role/RoleBinding. The plugin creates a new SCRAM user on the Kafka cluster using the configured mechanism and provisions the configured ACLs. The credential lives on the lease until you delete the SecretAccessRequest or it expires; on lease revocation the plugin removes the SCRAM user record and associated ACLs.

    $ kubectl get secretaccessrequest kafka-cred-rqst -n demo -o json | jq '.status'
{
  "lease": {
    "duration": "1h0m0s",
    "id": "your-database-path/creds/k8s.-.demo.kafka-producer/abc...",
    "renewable": true
  },
  "secret": {
    "name": "kafka-cred-rqst-xxxxxx"
  }
}

$ kubectl get secret -n demo kafka-cred-rqst-xxxxxx -o jsonpath='{.data.username}' | base64 -d
v-kubernetes-demo-XXXXXXXX

$ kubectl get secret -n demo kafka-cred-rqst-xxxxxx -o jsonpath='{.data.password}' | base64 -d
xxxxxxxxxxxxxxxxx

    Use the issued username / password as your franz-go / sarama / librdkafka client’s SASL credentials (sasl.mechanism=SCRAM-SHA-256); the credential and associated ACLs are revoked when the SecretAccessRequest is deleted.

    Further reading

    What's on this Pages

    Search
    Improve This Page