Table of Contents

Status

Current state: Under Discussion Accepted

Discussion thread: here

JIRA: none yet

Jira

server	ASF JIRA
serverId	5aa69414-a9e9-3523-82ec-879b028fb15b
key	KAFKA-10259

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

As part of the KIP-500 effort to remove Kafka's ZooKeeper dependency, we need a broker-side API to alter these settings.

Public Interfaces

...

describeUserScramCredentials

listScramUsers describeUserScramCredentials will list describe the currently configured SCRAM usersuser credentials. For security reasons, it will not list their passwords.

Code Block

language	java

public enum ScramMechanismTypeScramMechanism {
    UNKNOWN(0),
    HMACSCRAM_SHA_256(1),
    HMACSCRAM_SHA_512(2);

    private final byte type;

    private ScramMechanismType(byte type) {
final String mechanismName;

    public static ScramMechanism fromType(byte this.type = type;type) {
        // etc...
    }
}

    public classstatic ScramMechanism fromMechanismName(String mechanismName) {
      private final ScramMechanismType type;// etc...
    private}

 final int iterations;

 public String  public ScramMechanism(ScramMechanismType type, int iterationsmechanismName() {
        // this.type = type;
        this.iterations = iterations;etc...
    }

    public ScramMechanismTypebyte type() {
        return type;// etc...
    }

    publicprivate int iterations(ScramMechanism(byte type) {
        return iterations;// etc...
    }
}

public class ScramUserListingScramCredentialInfo {
    private final StringScramMechanism namemechanism;
    private final List<ScramMechanism>int mechanismiterations;
}

    public ScramUserListing(String name, List<ScramMechanism> mechanism)class UserScramCredentialsDescription {
    private final String  this.name = name;
    private final   this.mechanism = mechanism;
   List<ScramCredentialInfo> infos;
}

public class DescribeScramUserCredentialsOptions extends AbstractOptions<DescribeScramUserCredentialsOptions> { }

// interface Admin: Describe all public String nameusers.
default DescribeScramUserCredentialsResult describeScramUserCredentials() {
        return namedescribeScramUserCredentials(null);
    }

// interface Admin: Describe publicindicated List<ScramMechanism>users mechanism()null {
or empty implies all).
default DescribeScramUserCredentialsResult describeScramUserCredentials(List<String> users) {
  return mechanism;
 return describeScramUserCredentials(users, new }DescribeScramUserCredentialsOptions());
}

public// classinterface ListScramUsersOptions extends AbstractOptions<ListScramUsersOptions> { }

default ListScramUsersResult listScramUsers()Admin
DescribeScramUserCredentialsResult describeScramUserCredentials(List<String> users, DescribeScramUserCredentialsOptions options);

public class DescribeScramUserCredentialsResult {
    return listScramUsers(new ListScramUsersOptions());
}

ListScramUsersResult listScramUsers(ListScramUsersOptions options);

public class ListScramUsersResult {// completes successfully only if users() does and every described user does
    public KafkaFuture<Map<String, ScramUser>>UserScramCredentialsDescription>> all();
}

listScramUsers will be implemented by a new RPC.



    // completes successfully if request was authorized and the list of users described is determined
    public KafkaFuture<List<String>> users();

    // completes successfully if the individual user is described successfully
    public KafkaFuture<UserScramCredentialsDescription> description(String userName)
}

describeScramUserCredentials will be implemented by a new RPC.

Code Block

{
  "apiKey": 50,
  "type": "request",

Code Block

{ 
  "apiKey": 50,
  "type": "request",
  "name": "ListScramUsersRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
  ]
}

{ 
  "apiKey": 50, 
  "type": "response",
  "name": "ListScramUsersResponse", 
  "validVersions": "0", 
  "flexibleVersions": "0+", 
  "fields": [ 
    { "name": "ErrorDescribeUserScramCredentialsRequest",
  "typevalidVersions": "int160",
  "versionsflexibleVersions": "0+",
  "fields": [
   "about": "The message-level error code." },
    { "name": "ErrorMessageUsers", "type": "string[]UserName", "versions": "0+", "nullableVersions": "0+",
      "about": "The message-levelusers error message." },
    { "name": "Usersto describe, or null/empty to describe all users.", "typefields": "[]ScramUser", "versions": "0+",
      "about": "The SCRAM users.", "fields": [
[
      { "name": "Name", "type": "string", "versions": "0+",
        "about": "The user name." },
    ]}
  ]
}

{
  "nameapiKey": "Mechanisms"50,
  "type": "ScramUserMechanismresponse",
  "versionsname": "0+DescribeUserScramCredentialsResponse",
      "validVersions": "0",
  "aboutflexibleVersions": "The user name." }0+",
    "fields": [
    { "name": "TypeThrottleTimeMs", "type": "int8int32", "versions": "0+",
          "about": "The SCRAMduration mechanismin type." },
    milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "IterationsErrorCode", "type": "int32int16", "versions": "0+",
          "about": "The numbermessage-level oferror iterationscode, used in the SCRAM mechanism0 except for user authorization or infrastructure issues." },
    {  }
    ]}
  ]
}

It will require ALTER permissions on the CLUSTER resource. It will return CLUSTER_AUTHORIZATION_FAILED if the user has insufficient permissions.

It will be will be sent to the controller, and will return NOT_CONTROLLER if the receiving broker is not the controller.

AlterScramUsers

alterScramUsers will delete, create, or change SCRAM users.

Deletions are done by user name.

Alterations will create the given user if it doesn't exist, or alter it if it does.

Code Block

public class ScramUserDeletion {
    private final String user;

    ScramUserDeletion(String user) {"name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
      "about": "The message-level error message, if any." },
    { "name": "Results", "type": "[]DescribeUserScramCredentialsResult", "versions": "0+",
      "about": "The results for descriptions, one per user.", "fields": [
      {  this.user = user;"name": "User", "type": "string", "versions": "0+",
    }

    public String user() {"about": "The user name." },
      {  return user;
    }
}

public class ScramCredential {
"name": "ErrorCode", "type": "int16", "versions": "0+",
       private final ScramMechanism mechanism;
    private final byte[] salt;
    private final byte[] saltedPassword;
}

public class ScramUserAlteration {"about": "The user-level error code." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
    private  final String user;
    private final List<ScramCredential> credentials;

    public ScramCredentialAlteration(String user, List<ScramCredential> credentials) {
        this.user = user; "about": "The user-level error message, if any." },
      { "name": "CredentialInfos", "type": "[]CredentialInfo", "versions": "0+",
        this.credentials = credentials;
    }

    public String user() {"about": "The mechanism and related information associated with the user's SCRAM credentials.", "fields": [
        return user;
    }

{ "name": "Mechanism", "type": "int8", "versions": "0+",
    public List<ScramCredential> credentials() {
   "about": "The SCRAM mechanism." },
 return credentials;
    }
}

public class AlterScramUsersOptions extends AbstractOptions<AlterScramUsersOptions> {}

default AlterScramUsersResult alterScramUsers(List<ScramUserDeletion> deletions{ "name": "Iterations", "type": "int32", "versions": "0+",
          "about": "The number of iterations used in the SCRAM credential." }]}
                          List<ScramUserAlteration> alterations) {
    return alterScramUsers(deletions, alterations, new AlterScramUsersOptions());
}

AlterScramUsersResult alterScramUsers(List<ScramUserDeletion> deletions,
                                      List<ScramUserAlteration> alterations,
                                      AlterScramUsersOptions options);

public class AlterScramCredentialsResult {
    public KafkaFuture<Void> all();
    public Map<String, KafkaFuture<Void>> results();
}

The AlterScramUsersRequest and AlterScreamUsersResponse implement the new API.

]}
  ]
}

It will require DESCRIBE permissions on the CLUSTER resource. It will return CLUSTER_AUTHORIZATION_FAILED if the user has insufficient permissions.

It will return RESOURCE_NOT_FOUND if a user is requested to be described but that user does not exist/has no credentials. Note that DescribeScramUserCredentialsResult will not consider such a username to be part of its users() result list, and this RESOURCE_NOT_FOUND error by itself will not prevent the future returned by all() from completing successfully.

It will return DUPLICATE_RESOURCE if a user is requested to be described twice. Note that DescribeScramUserCredentialsResult will consider such a username to be part of its users() result list, and this will cause the future returned by all() to complete exceptionally.

AlterScramUserCredentials

alterScramUserCredentials will create or change SCRAM user credentials.

Alterations will create the given user credential if it doesn't exist, or alter it if it does.

Code Block

public abstract class UserScramCredentialAlteration {
    private final String user;
}

public class UserScramCredentialUpsertion extends UserScramCredentialAlteration {
    private final ScramCredentialInfo info;
    private final byte[] salt;
    private final byte[] password;

    // There will be one constructor that randomly generates a salt, and one that accepts a pre-defined salt.
}

public class UserScramCredentialDeletion extends UserScramCredentialAlteration {
    private final ScramMechanism mechanism;
}

public class AlterScramUserCredentialsOptions extends AbstractOptions<AlterScramUserCredentialsOptions> {}

// interface Admin
default AlterScramUserCredentialsResult alterScramUserCredentials(List<UserScramCredentialAlteration> alterations) {
    return alterScramUserCredentials( alterations, new AlterScramUserCredentialsOptions());
}

// interface Admin
AlterScramUserCredentialsResult alterScramUserCredentials(List<UserScramCredentialAlteration> alterations

Code Block

language	java

{ 
  "apiKey": 51, 
  "type": "response",
  "name": "AlterScramUsersRequest",
  "validVersions": "0", 
  "flexibleVersions": "0+", 
  "fields": [ 
    { "name": "Deletions", "type": "[]ScramUserDeletion", "versions": "0+",
      "about": "The SCRAM users to remove.", "fields": [
                 { "name": "Name", "type": "string", "versions": "0+",
          "about": "The user name." }AlterScramUserCredentialsOptions options);

public class AlterUserScramCredentialsResult {
    public  ]}Map<String,
 KafkaFuture<Void>> values();
  {  public KafkaFuture<Void> all(); // completes successfully only if everything in values() does
}

If any of the operations associated with a single user can't be done, none of the operations will be done. On the other hand, operations could succeed for one user but fail for another, different user.

If a user doesn't exist and we add a credential for them, they will be created. If a user exists and we remove their last credential, they will be deleted.

The AlterScramUserCredentialsRequest and AlterScramUserCredentialsResponse implement the new API.

Code Block

language	java

{
  "apiKey": 51,
 "name": "Alterations", "type": "[]ScramUserAlteration", "versions": "0+",
      "about": "The SCRAM users to alter.", "fields": [
      { "name": "Name", "type": "stringrequest",
  "versionsname": "0+AlterUserScramCredentialsRequest",
    "validVersions": "0",
    "aboutflexibleVersions": "The0+",
 user name."fields": }[
      { "name": "CredentialsDeletions", "type": "ScramCredential[]ScramCredentialDeletion", "versions": "0+",
        "about": "The SCRAM credentials to configureremove." }, "fields": [
      {  { "name": "MechanismTypeName", "type": "int8string", "versions": "0+",
          "about": "The mechanismuser typename." },
        { "name": "MechanismIterationsMechanism", "type": "int32int8", "versions": "0+",
          "about": "The number of iterations, or -1 to use the server defaultSCRAM mechanism." },
    ]},
    { "name": "SaltUpsertions", "type": "bytes[]ScramCredentialUpsertion", "versions": "0+", ",
          "about": "AThe randomSCRAM saltcredentials generated by the clientto update/insert." },
 "fields": [
      { "name": "SaltedPasswordName", "type": "bytesstring", "versions": "0+", ",
          "about": "The salteduser passwordname." }
,
      ]}
    ]}  { "name": "Mechanism", "type": "int8", "versions": "0+",
  ]       
}   

{ 
  "apiKey": 51, 
  "type"about": "The SCRAM mechanism." },
      { "name": "responseIterations",
  "nametype": "AlterScramUsersResponseint32",
  "validVersionsversions": "0+",
     
   "flexibleVersionsabout": "0+", 
  "fields": [ 
The number of iterations." },
       { "name": "ResultsSalt", "type": "[]AlterScramUsersResultbytes", "versions": "0+",
        "about": "TheA resultsrandom for removals, followedsalt generated by the results for alterationsclient.", "fields": [
   },
      { "name": "ErrorCodeSaltedPassword", "type": "int8bytes", "versions": "0+",
          "about": "The errorsalted codepassword." },
    ]}
  ]
}

{
  {"apiKey": 51,
  "nametype": "ErrorStringresponse",
  "typename": "stringAlterUserScramCredentialsResponse",
  "versionsvalidVersions": "0+",
  "nullableVersionsflexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "abouttype": "The error message, if any." }
    ]}  
  ]       
}

A removal or alteration will return INVALID_REQUEST if an empty user name is passed, or an invalid number of iterations, or a duplicate user name. Note that if the number of iterations is set to -1, the server-side default will be used.

A removal will return a new error code, RESOURCE_NOT_FOUND, if it was instructed to delete a user, but that user was not found.

The RPC will require ALTER on CLUSTER. It will return CLUSTER_AUTHORIZATION_FAILED if the user has insufficient permissions. It will be will be sent to the controller, and will return NOT_CONTROLLER if the receiving broker is not the controller.

Command-Line Changes

...

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": "Results", "type": "[]AlterUserScramCredentialsResult", "versions": "0+",
      "about": "The results for deletions and alterations, one per affected user.", "fields": [
      { "name": "User", "type": "string", "versions": "0+",
        "about": "The user name." },
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
        "about": "The error code." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
        "about": "The error message, if any." }
    ]}
  ]
}

An addition will return UNACCEPTABLE_CREDENTIAL if an empty user name or an invalid number of iterations (less than the minimum required for the mechanism or more than a hard-coded max of 16,384) is passed.

DUPLICATE_RESOURCE will be returned if a user appears as both an upsertion and a deletion in the same request.

UNSUPPORTED_SASL_MECHANISM will be returned if the broker does not recognize the requested SASL mechanism.

A removal will return an error code, RESOURCE_NOT_FOUND, if it was instructed to delete a credential that did not exist.

The RPC will require ALTER on CLUSTER. It will return CLUSTER_AUTHORIZATION_FAILED if the user has insufficient permissions.

It will be will be sent to the controller and will return NOT_CONTROLLER if the receiving broker is not the controller.

Command-Line Changes

We will extend the kafka-configs.sh command to so that it is possible to set a SCRAM configuration without using --zookeeper. The command-line syntax will be unchanged, except for the fact that users will now be able to pass --bootstrap-server instead of --zookeeper.

As mentioned earlier, this API does not return secrets. Therefore, the salt, salted password, and so on will not be returned by a kafka-configs.sh --describe operation. The describe operation will return only the presence of the user plus the algorithm and number of iterations used. For example:

Code Block

$ bin/kafka-configs.sh --bootstrap-server localhost:9020 \
    --alter \
    --entity-type users \
    --entity-name alice \
    --add-config 'SCRAM-SHA-256=[iterations=8192,password=alice-secret],SCRAM-SHA-512=[password=alice-secret]'

Completed updating config for entity: user-principal 'alice'.

$ bin/kafka-configs.sh --bootstrap-server localhost:9020 --entity-type users --entity-name alice --describe

Configs for user-principal 'alice' are SCRAM-SHA-512=iterations=8192

Compatibility, Deprecation, and Migration Plan

...

Space shortcuts

Child pages

Versions Compared

Old Version 4

New Version Current

Key

Status

Public Interfaces

describeUserScramCredentials

AlterScramUsers

AlterScramUserCredentials

Command-Line Changes

Command-Line Changes

Compatibility, Deprecation, and Migration Plan

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 4

New Version Current

Key

Status

Public Interfaces

describeUserScramCredentials

AlterScramUsers

AlterScramUserCredentials

Command-Line Changes

Command-Line Changes

Compatibility, Deprecation, and Migration Plan