...

This KIP lays out the protocol and API for setting SCRAM credentials via the Admin interface by transmitting
the password to the broker, which will perform the SCRAM processing and store that information in ZooKeeper.
As such by default it will initially only be possible to set SCRAM credentials over TLS connections, in order to not leak the password. A broker config is provided to allow administrators to enable password change over plain connections according to their trust in their network.
A future KIP will may allow setting SCRAM credentials over plain connections.

...

Kafka deployments often use non-TLS communication between broker and ZooKeeper. The kafka-configs.sh  script can be run on a zookeeper node (communicating with it over localhost) which avoids sending processed SCRAM information over the network. It is, however, unclear how many admins actually do this.
With this KIP the processed SCRAM information would be sent unecrypted between the broker and ZooKeeper.
While that clearly , which has all the drawbacks described in the previous section. However:

• Kafka and ZooKeeper nodes are often co-located on a secure network (this is not necessarily the case for clients and brokers)
• It is possible to deploy Kafka and ZooKeeper using TLS (for example, using more recent versions of ZooKeeper, or using a TLS tunnel)
• It is expected that the eventual removal of Kafka's dependency on ZooKeeper will remove this weakness.

Public Interfaces

• New broker configs

• New message type in the protocol
  

• New methods in the Admin/AdminClient API

Proposed Changes

Broker configs

Add new broker config sasl.scram.password.change.enabled will control whether the Admin API can be used for password change. It will take one of three possible values:

• disabled (the default) The Admin API cannot be used at all (the broker will return an error).
• enabled_over_tls The Admin API can only be used over a TLS connection.
• enabled The Admin API can be used over TLS and plain connections. This is only suitable for use where the client and broker are on the same trusted network.

Add new broker config sasl.scram.password.change.enabled.mechanisms to control the SASL mechanisms which can have passwords set via the Admin API.
This will not affect the existing kafka-configs.sh  script. The default value will be `SHA256,SHA512`.

Add new broker config sasl.scram.password.change.<mechanism>.iterations to control the number of iterations to be used
when SCRAM passwords are set. Defaults:

• sasl.scram.password.change.SHA256.iterations=4096
• sasl.scram.password.change.SHA512.iterations=4096

Add new broker config sasl.scram.password.policy.class to specify a password complexity policy.
Such classes will need to implement the Java interface:

	interface ScramPasswordPolicy {
	    /** 
	     * Return whether the given password is acceptable according to the policy.
	     * @return true if the given password is acceptable.
	     */
	    boolean accept(String password);
	    
		/**
	     * A human-readable description of the policy, which will be included in errors to the client.
	     */
        String describe();
	}

By default no password policy will be applied.

New protocol

The client sents sends a ScramPasswordRequest containing the usernames, SCRAM hash types and passwords to be set. Note that the existing possibility of having a different password for different mechanisms is retained.

	{
	  "type": "request",
	  "name": "ScramPasswordRequest",
	  "validVersions": "0",
	  "fields": [
	    { "name": "Credentials", "type": "[]ScramCredentials", "versions": "0+",
	      "about": "The SCRAM credentials to set.",
	      "fields": [
		    { "name": "Username", "type": "string", "versions": "0+",
		      "about": "The username." },
		  	        { "name": "HashType", "type": "int8", "versions": "0+",
		              "about": "The scram hash type." },
		        { "name": "Password", "type": "string", "versions": "0+",
		      "about": "The password." }
	    ]}
	  ]
	}

The broker then:

• The broker may reject the request, depending on the value of the sasl.scram.password.change.enabled config, otherwise
• The broker
• Checks the connection is TLS,
• validates each password against any configured policy,
• The broker performs the SCRAM credential processing,
• The broker asynchronously updates ZooKeeper,
• Returns The broker returns ScramPasswordResponse 

	{
	  "type": "response",
	  "name": "ScramPasswordResponse",
	  "validVersions": "0",
	  "fields": [
	    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
	      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
	    { "name": "HashResults", "type": "[]HashResult", "versions": "0+",
	      "about": "Per-user results, grouped by hash type.",
	      "fields": [
		    { "name": "HashType", "type": "int8", "versions": "0+",
		      "about": "The hash type." },
		    { "name": "UserResults", "type": "[]UserResult", "versions": "0+",
		      "about": "Per-user results for this hash type.",
		    "fields": [
		      { "name": "Username", "type": "string", "versions": "0+",
		        "about": "The username." },
		      { "name": "ErrorCode", "type": "int16", "versions": "0+",
		        "about": "The error code, or 0 if there was no error." },
		      { "name": "ErrorMessage", "type": "string", "nullableVersions": "0+", "versions": "0+",
		        "about": "The error message." }
		    ]}
	      ]}
	  ]
	}

Possible errors include:

• API_DISABLED (new code) The broker was configured with sasl.scram.password.change.enabled=disabled. The broker rejects the request and each user in the response will have error this error.
• ENCRYPTION_REQUIRED (new code) The request was made over a non-TLS connection but the broker was configured with sasl.scram.password.change.enabled=enabled_over_tls. The broker rejects the request and each user in the response will have error this error.
• USERCLUSTER_AUTHORIZATION_FAILED (new code) The authenticated user did not have permission to change the given users passwordAlter on the Cluster resource. In this KIP only super users who can alter the cluster will be allowed to set SCRAM passwords, and therefore either all the users in the response will have this or none will. In the future an authorizer might support some non-super users being able to set the password of others, or users being able to set their own passwords.
• POLICY_VIOLATION: A given users password violated the password complexity policy. The message will explain the policy.
• UNSUPPORTED_SASL_MECHANISM A given users password was to be set for an unsupported Hash.

Broker configs

Add new broker config sasl.scram.password.change.enabled.mechanisms  to control the SASL mechanisms which can have passwords set via the Admin API.
This will not affect the existing kafka-configs.sh  script (the intention being that admins can disable setting passwords via the Admin API if they consider it insecure and continue using kafka-configs.sh --zookeeper localhost:nnnn). The default value will be `SHA256,SHA512`.

Add new broker config sasl.scram.password.change.<mechanism>.iterations to control the number of iterations to be used
when SCRAM passwords are set. Defaults:

• sasl.scram.password.change.SHA256.iterations=4096
• sasl.scram.password.change.SHA512.iterations=4096

Add new broker config sasl.scram.password.policy.class to specify a password complexity policy.
Such classes will need to implement the Java interface:

	interface ScramPasswordPolicy {
	    /** 
	     * Return whether the given password is acceptable according to the policy.
	     * @return true if the given password is acceptable.
	     */
	    boolean accept(String password);
	    
		/**
	     * A human-readable description of the policy, which will be included in errors to the client.
	     */
        String describe();
	}

...

• .

Admin API

The following method will be added to Admin and its subclasses:

...