You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 14 Next »

Status

Current state: Under Discussion

Discussion thread: here

JIRA:  Unable to render Jira issues macro, execution error.

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

Motivation

The kafka-configs.sh command supports changing the SCRAM settings for users.  For example:

bin/kafka-configs --zookeeper localhost:2181 \
    --alter \
    --entity-type users \
    --entity-name alice \
    --add-config 'SCRAM-SHA-256=[iterations=8192,password=alice-secret],SCRAM-SHA-512=[password=alice-secret]'

However, this command depends on Apache ZooKeeper.  There is no broker-based API to change these settings.

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

describeScramUsers

describeScramUsers will describe the currently configured SCRAM users.  For security reasons, it will not list their passwords.

public enum ScramMechanism {
    UNKNOWN(0),
    HMAC_SHA_256(1),
    HMAC_SHA_512(2);

    byte type;

    private ScramMechanism(byte type) {
        this.type = type;
    }
}

public class ScramMechanismInfo {
    private final ScramMechanism mechanism;
    private final int iterations;
}

public class ScramUserDescription {
    private final String name;
    private final List<ScramMechanismInfo> infos;
}

public class DescribeScramUsersOptions extends AbstractOptions<DescribeScramUsersOptions> { }

// Describe all users.
default DescribeScramUsersResult describeScramUsers() {
    return describeScramUsers(null);
}

default DescribeScramUsersResult describeScramUsers(List<String> users) {
    return describeScramUsers(users, new DescribeScramUsersOptions());
}

DescribeScramUsersResult describeScramUsers(DescribeScramUsersOptions options);

public class DescribeScramUsersResult {
    public KafkaFuture<Map<String, ScramUserDescription>> all();
}

describeScramUsers will be implemented by a new RPC.

{ 
  "apiKey": 50,
  "type": "request",
  "name": "DescribeScramUsersRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Users", "type": "[]UserName", "versions": "0+", "nullableVersions": "0+",
      "about": "The users to describe, or null to describe all users.", "fields": [
      { "name": "Name", "type": "string", "versions": "0+",
        "about": "The user name." }
    ]},
  ]
}

{ 
  "apiKey": 50, 
  "type": "response",
  "name": "DescribeScramUsersResponse", 
  "validVersions": "0", 
  "flexibleVersions": "0+", 
  "fields": [ 
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The time spent waiting for quota." },
    { "name": "Error", "type": "int16", "versions": "0+",
      "about": "The message-level error code." },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
      "about": "The message-level error message." },
    { "name": "Users", "type": "[]ScramUser", "versions": "0+",
      "about": "The SCRAM users.", "fields": [
      { "name": "Name", "type": "string", "versions": "0+",
        "about": "The user name." },
      { "name": "MechanismInfos", "type": "[]ScramUserMechanismInfo", "versions": "0+",
        "about": "The mechanisms associated with the user.", "fields": [
        { "name": "Mechanism", "type": "int8", "versions": "0+",
          "about": "The SCRAM mechanism." },
        { "name": "Iterations", "type": "int32", "versions": "0+",
          "about": "The number of iterations used in the SCRAM mechanism." }
      }
    ]}
  ]
}

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 create or change SCRAM users.

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

interface ScramCredentialAlteration {
    private final String user;
}

public class ScramCredentialAddition extends ScramCredentialAlteration {
    private final ScramMechanismInfo 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 ScramCredentialDeletion extends ScramCredentialAlteration {
    private final ScramMechanism mechanism;
}

public class DeletedScramCredential {
    private final ScramMechanismInfo info;
}

public class AlterScramUsersOptions extends AbstractOptions<AlterScramUsersOptions> {}

default AlterScramUsersResult alterScramUsers(List<ScramUserAlteration> alterations) {
    return alterScramUsers(deletions, alterations, new AlterScramUsersOptions());
}

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

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

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 AlterScramUsersRequest and AlterScramUsersResponse implement the new API.

{ 
  "apiKey": 51, 
  "type": "request",
  "name": "AlterScramUsersRequest",
  "validVersions": "0", 
  "flexibleVersions": "0+", 
  "fields": [ 
    { "name": "Alterations", "type": "[]ScramUserAlteration", "versions": "0+",
      "about": "The SCRAM users to create or alter.", "fields": [
      { "name": "Name", "type": "string", "versions": "0+",
        "about": "The user name." }
      { "name": "Deletions", "type": "[]ScramCredentialDeletion", "versions": "0+",
        "about": "The credentials to delete.", "fields": [
        { "name": "Mechanism", "type": "int8", "versions": "0+",
          "about": "The SCRAM mechanism." }
      ]},
      { "name": "Additions", "type": "[]ScramCredentialAddition", "versions": "0+",
        "about": "The SCRAM credentials to add." }
        { "name": "Mechanism", "type": "int8", "versions": "0+",
          "about": "The SCRAM mechanism." },
        { "name": "Iterations", "type": "int32", "versions": "0+",
          "about": "The number of iterations, or -1 to use the server default." },
        { "name": "Salt", "type": "bytes", "versions": "0+", ",
          "about": "A random salt generated by the client." },
        { "name": "SaltedPassword", "type": "bytes", "versions": "0+", ",
          "about": "The salted password." }
      ]}
    ]}
  ]
}

{ 
  "apiKey": 51, 
  "type": "response",
  "name": "AlterScramUsersResponse",
  "validVersions": "0", 
  "flexibleVersions": "0+", 
  "fields": [ 
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The time spent waiting for quota." }, 
    { "name": "Results", "type": "[]AlterScramUsersResult", "versions": "0+",
      "about": "The results for alterations, in the same order as the request.", "fields": [
        { "name": "User", "type": "string", "versions": "0+",
          "about": "The user name." },
        { "name": "ErrorCode", "type": "int8", "versions": "0+",
          "about": "The error code." },
        { "name": "ErrorString", "type": "string", "versions": "0+", "nullableVersions": "0+",
          "about": "The error message, if any." }
    ]}  
  ]       
}

An addition 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 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:

$ 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

There is no impact on compatibility, since this adds a new API that didn't exist before.

Rejected Alternatives

Extend IncrementalAlterConfigs to handle SCRAM

Rather than creating new RPCs, we could have extended the IncrementalAlterConfigs RPC to support changing SCRAM configurations.  However, this would involve some fairly complex string manipulation for clients that wanted to use the API.  It would also be more cumbersome to list the SCRAM users.


  • No labels