KIP-554: Add Broker-side SCRAM Config API

[arkady-emelyanov]

Hey!

Kafka 2.7 it out. Which means KIP-554 has been merged into main branch 🎉

This PR closes #1803 issue (when Kafka 2.7 is used, of course).

One thing I'm not sure about: I have divided AlterScramUserCredentials API call into two: DeleteUserScramCredentials and UpsertUserScramCredentials. I personally think this brings more cleaner interface. But, if goal is 1:1 API match, AlterUserScramCredentials could be exposed.

Protocol links:

• https://kafka.apache.org/protocol#The_Messages_DescribeUserScramCredentials
• https://kafka.apache.org/protocol#The_Messages_AlterUserScramCredentials

Local environment for testing purposes (PLAINTEXT listener on 9092, and SASL_PLAINTEXT listener on 9093):
docker-compose.yaml

version: '3'

services:
  zoo:
    image: zookeeper:3.6.2
    hostname: zoo1
    restart: unless-stopped
    environment:
      ZOO_MY_ID: 1
      ZOO_SERVERS: server.1=0.0.0.0:2888:3888;2181

  kafka:
    image: wurstmeister/kafka:2.13-2.7.0
    hostname: kafka1
    restart: unless-stopped
    ports:
      - "9092:9092"
      - "9093:9093"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
      KAFKA_ZOOKEEPER_CONNECT: zoo:2181
      KAFKA_LISTENERS: INT://:9092,EXT://:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INT:PLAINTEXT,EXT:SASL_PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: INT://127.0.0.1:9092,EXT://127.0.0.1:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: INT
      KAFKA_SASL_ENABLED_MECHANISMS: "SCRAM-SHA-256,SCRAM-SHA-512"
      KAFKA_SUPER_USERS: "User:ANONYMOUS;User:admin"
      KAFKA_OPTS: "-Djava.security.auth.login.config=/etc/jaas.conf"
    volumes:
      - ./docker-compose.jaas.conf:/etc/jaas.conf
    depends_on:
      - zoo

JAAS configuration file:
docker-compose.jaas.conf

ext.KafkaServer {
  org.apache.kafka.common.security.scram.ScramLoginModule required;
};

Sample application:
main.go

package main

import (
	"fmt"
	"github.com/Shopify/sarama"
)

/*
After running main.go, check it out:

(this command will not work)
kafkacat \
-X security.protocol=SASL_PLAINTEXT \
-X sasl.mechanisms=SCRAM-SHA-256 \
-X sasl.username='bob' \
-X sasl.password='bob-secret' \
-b 127.0.0.1:9093 \
-L

(this command will succeed):
kafkacat \
-X security.protocol=SASL_PLAINTEXT \
-X sasl.mechanisms=SCRAM-SHA-512 \
-X sasl.username='bob' \
-X sasl.password='bob-secret' \
-b 127.0.0.1:9093 \
-L


 */
func main() {
	cluster := []string{"127.0.0.1:9092"}
	config := sarama.NewConfig()
	config.Version = sarama.V2_7_0_0

	adm, err := sarama.NewClusterAdmin(cluster, config)
	if err != nil {
		panic(err)
	}

	// upsert two mechanisms: SCRAM_MECHANISM_SHA_256 and SCRAM_MECHANISM_SHA_512
	upsertList, err := adm.UpsertUserScramCredentials([]sarama.AlterUserScramCredentialsUpsert{
		{
			Name:       "bob",
			Mechanism:  sarama.SCRAM_MECHANISM_SHA_256,
			Iterations: 8192,
			Salt:       []byte("hello world"),
			Password:   []byte("bob-secret"),
		},
		{
			Name:       "bob",
			Mechanism:  sarama.SCRAM_MECHANISM_SHA_512,
			Iterations: 8192,
			Salt:       []byte("hello world"),
			Password:   []byte("bob-secret"),
		},
	})
	if err != nil {
		panic(err)
	}
	fmt.Println("> Upsert")
	for _, a := range upsertList {
		fmt.Println(a.User)
	}

	// delete mechanism: SCRAM_MECHANISM_SHA_256
	deleteList, err := adm.DeleteUserScramCredentials([]sarama.AlterUserScramCredentialsDelete{
		{
			Name: "bob",
			Mechanism:  sarama.SCRAM_MECHANISM_SHA_256,
		},
	})
	fmt.Println("> Delete")
	for _, a := range deleteList {
		fmt.Println(a.User)
	}


	describeList, err := adm.DescribeUserScramCredentials(nil)
	if err != nil {
		panic(err)
	}

	fmt.Println("> Describe")
	for _, d := range describeList {
		fmt.Println("User:", d.User)
		for _, c := range d.CredentialInfos {
			fmt.Println("Allowed SCRAM mechanism:", c.Mechanism, c.Iterations)
		}
		fmt.Println("")
	}
}

[dnwe]

Many thanks for this! The changes look good to me and great job including instructions to test the changes locally with a stood up kafka ⭐