---
apiVersion: batch/v1
kind: Job
metadata:
  name: kafka-scram-users
  namespace: kafka
  annotations:
    kustomize.toolkit.fluxcd.io/force: enabled
spec:
  backoffLimit: 2
  template:
    metadata:
      annotations:
        sidecar.istio.io/inject: "false"
        vault.hashicorp.com/agent-inject: "true"
        vault.hashicorp.com/agent-pre-populate-only: "true"
        vault.hashicorp.com/auth-path: auth/kubernetes
        vault.hashicorp.com/role: kafka
        vault.hashicorp.com/agent-inject-secret-admin: secrets/data/kafka/bootstrap
        vault.hashicorp.com/agent-inject-template-admin: |
          {{- with secret "secrets/data/kafka/bootstrap" -}}{{ index .Data.data "interBrokerPassword" }}{{- end -}}
        vault.hashicorp.com/agent-inject-secret-pm-user: secrets/data/kafka/apps/pm
        vault.hashicorp.com/agent-inject-template-pm-user: |
          {{- with secret "secrets/data/kafka/apps/pm" -}}{{ index .Data.data "username" }}{{- end -}}
        vault.hashicorp.com/agent-inject-secret-pm-pass: secrets/data/kafka/apps/pm
        vault.hashicorp.com/agent-inject-template-pm-pass: |
          {{- with secret "secrets/data/kafka/apps/pm" -}}{{ index .Data.data "password" }}{{- end -}}
        vault.hashicorp.com/agent-inject-secret-issues-user: secrets/data/kafka/apps/issues
        vault.hashicorp.com/agent-inject-template-issues-user: |
          {{- with secret "secrets/data/kafka/apps/issues" -}}{{ index .Data.data "username" }}{{- end -}}
        vault.hashicorp.com/agent-inject-secret-issues-pass: secrets/data/kafka/apps/issues
        vault.hashicorp.com/agent-inject-template-issues-pass: |
          {{- with secret "secrets/data/kafka/apps/issues" -}}{{ index .Data.data "password" }}{{- end -}}
        vault.hashicorp.com/agent-inject-secret-flows-user: secrets/data/kafka/apps/flows
        vault.hashicorp.com/agent-inject-template-flows-user: |
          {{- with secret "secrets/data/kafka/apps/flows" -}}{{ index .Data.data "username" }}{{- end -}}
        vault.hashicorp.com/agent-inject-secret-flows-pass: secrets/data/kafka/apps/flows
        vault.hashicorp.com/agent-inject-template-flows-pass: |
          {{- with secret "secrets/data/kafka/apps/flows" -}}{{ index .Data.data "password" }}{{- end -}}
    spec:
      serviceAccountName: kafka-kafka-contour
      automountServiceAccountToken: true
      restartPolicy: Never
      imagePullSecrets:
        - name: regcred
      nodeSelector:
        dedicated: generic
      securityContext:
        runAsUser: 1001
        runAsNonRoot: true
        fsGroup: 1001
      containers:
        - name: kafka-scram-users
          image: cr.yandex/crp3ccidau046kdj8g9q/contour/kafka-nn/kafka:4.0.0-debian-12-r10
          imagePullPolicy: IfNotPresent
          env:
            - name: KAFKA_HEAP_OPTS
              value: -Xmx256m
          command:
            - /bin/bash
            - -ec
            - |
              K=/opt/bitnami/kafka/bin
              ADMIN=kafka-kafka-contour-controller-0.kafka-kafka-contour-controller-headless.kafka.svc.cluster.local:9094
              CLIENT=kafka-kafka-contour.kafka.svc.cluster.local:9092
              S=/vault/secrets
              rc=0
              printf 'security.protocol=SASL_PLAINTEXT\nsasl.mechanism=PLAIN\nsasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="inter_broker_user" password="%s";\n' "$(cat $S/admin)" > /tmp/admin.conf
              for app in pm issues flows; do
                u=$(cat $S/$app-user); p=$(cat $S/$app-pass)
                if [ -z "$u" ] || [ -z "$p" ]; then
                  echo "$app: empty username or password in kafka/apps/$app"; rc=1; continue
                fi
                if [ -f "/tmp/pw-$u" ]; then
                  if [ "$(cat /tmp/pw-$u)" != "$p" ]; then
                    echo "$app: user $u already set from another app with a different password, skipped"; rc=1
                  else
                    echo "$app: user $u already set, same password"
                  fi
                  continue
                fi
                printf '%s' "$p" > "/tmp/pw-$u"
                $K/kafka-configs.sh --bootstrap-server $ADMIN --command-config /tmp/admin.conf --alter --add-config "SCRAM-SHA-512=[iterations=4096,password=$p]" --entity-type users --entity-name "$u" > /dev/null
                echo "$app: SCRAM-SHA-512 set for user $u"
              done
              sleep 5
              for app in pm issues flows; do
                u=$(cat $S/$app-user); p=$(cat $S/$app-pass)
                printf 'security.protocol=SASL_SSL\nsasl.mechanism=SCRAM-SHA-512\nssl.truststore.type=PEM\nssl.truststore.location=/kafka-ca/ca.crt\nsasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="%s" password="%s";\n' "$u" "$p" > /tmp/$app.conf
                if $K/kafka-topics.sh --bootstrap-server $CLIENT --command-config /tmp/$app.conf --list > /dev/null; then
                  echo "$app: login as $u over SASL_SSL ok"
                else
                  echo "$app: login as $u over SASL_SSL FAILED"; rc=1
                fi
              done
              $K/kafka-configs.sh --bootstrap-server $ADMIN --command-config /tmp/admin.conf --describe --entity-type users
              rm -f /tmp/*.conf /tmp/pw-*
              exit $rc
          resources:
            requests:
              cpu: 50m
              memory: 384Mi
          volumeMounts:
            - name: kafka-ca
              mountPath: /kafka-ca
              readOnly: true
      volumes:
        - name: kafka-ca
          secret:
            secretName: kafka-kafka-contour-tls
            items:
              - key: ca.crt
                path: ca.crt