THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
| Table of Contents |
|---|
Status
Current state: Under Discussion Accepted
Discussion thread: here
JIRA:
| Jira | ||||||
|---|---|---|---|---|---|---|
|
...
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
describeScramUsers describeUserScramCredentials will describe the currently configured SCRAM usersuser credentials. For security reasons, it will not list their passwords.
| Code Block | ||
|---|---|---|
| ||
public enum ScramMechanism {
UNKNOWN(0),
HMACSCRAM_SHA_256(1),
HMACSCRAM_SHA_512(2);
private final byte type;
private final String mechanismName;
public privatestatic ScramMechanism fromType(byte type) {
// this.type = type;etc...
}
}
public class ScramMechanismInfo {
public static ScramMechanism private final ScramMechanism mechanism;
fromMechanismName(String mechanismName) {
private final int iterations;
// etc...
}
public classString ScramUserDescriptionmechanismName() {
private final String name;
private final List<ScramMechanismInfo> infos;
// etc...
}
public class DescribeScramUsersOptions extends AbstractOptions<DescribeScramUsersOptions>public { }
// Describe all users.
default DescribeScramUsersResult describeScramUsers() {
return describeScramUsers(null);
}
default DescribeScramUsersResult describeScramUsers(List<String> usersbyte type() {
// etc...
}
private ScramMechanism(byte type) {
return describeScramUsers(users, new DescribeScramUsersOptions());
}
DescribeScramUsersResult describeScramUsers(DescribeScramUsersOptions options); // etc...
}
}
public class DescribeScramUsersResultScramCredentialInfo {
publicprivate KafkaFuture<Map<String,final ScramUserDescription>>ScramMechanism all()mechanism;
} |
describeScramUsers will be implemented by a new RPC.
| Code Block |
|---|
{ private final int iterations; } public class UserScramCredentialsDescription { private final String name; private final List<ScramCredentialInfo> infos; } public class DescribeScramUserCredentialsOptions extends AbstractOptions<DescribeScramUserCredentialsOptions> { } // interface Admin: Describe all users. default DescribeScramUserCredentialsResult describeScramUserCredentials() { return describeScramUserCredentials(null); } // interface Admin: Describe indicated users (null or empty implies all). default DescribeScramUserCredentialsResult describeScramUserCredentials(List<String> users) { return describeScramUserCredentials(users, new DescribeScramUserCredentialsOptions()); } // interface Admin DescribeScramUserCredentialsResult describeScramUserCredentials(List<String> users, DescribeScramUserCredentialsOptions options); public class DescribeScramUserCredentialsResult { // completes successfully only if users() does and every described user does public KafkaFuture<Map<String, UserScramCredentialsDescription>> all(); // 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", "name": "DescribeUserScramCredentialsRequest", "validVersions": "0", "flexibleVersions"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 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": "Error", "type": "int16", "versions": "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": "MechanismInfos"50, "type": "[]ScramUserMechanismInforesponse", "versionsname": "0+DescribeUserScramCredentialsResponse", "validVersions": "0", "aboutflexibleVersions": "The mechanisms associated with the user.0+", "fields": [ { "name": "MechanismThrottleTimeMs", "type": "int8int32", "versions": "0+", "about": "The SCRAMduration mechanism." }, in 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, used0 in the SCRAM mechanismexcept for user authorization or infrastructure issues." }, { "name": "ErrorMessage", "type": "string", "versions": "0+", }"nullableVersions": "0+", ]} ] } |
It will require DESCRIBE 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
"about": "The message-level error message, if any." },
{ "name": "Results", "type": "[]DescribeUserScramCredentialsResult", "versions": "0+",
"about": "The results for descriptions, one per user.", "fields": [
{ "name": "User", "type": "string", "versions": "0+",
"about": "The user name." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The user-level error code." },
{ "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
"about": "The user-level error message, if any." },
{ "name": "CredentialInfos", "type": "[]CredentialInfo", "versions": "0+",
"about": "The mechanism and related information associated with the user's SCRAM credentials.", "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 credential." }]}
]}
]
}
|
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 credentialsalterScramUsers will create or change SCRAM users.
Alterations will create the given user credential if it doesn't exist, or alter it if it does.
| Code Block |
|---|
interface ScramCredentialAlterationpublic abstract class UserScramCredentialAlteration { private final String user; } public class ScramCredentialAdditionUserScramCredentialUpsertion extends ScramCredentialAlterationUserScramCredentialAlteration { private final ScramMechanismInfoScramCredentialInfo 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 ScramCredentialDeletionUserScramCredentialDeletion extends ScramCredentialAlterationUserScramCredentialAlteration { private final ScramMechanism mechanism; } public class DeletedScramCredential { private final ScramMechanismInfo info; } public class AlterScramUsersOptionsAlterScramUserCredentialsOptions extends AbstractOptions<AlterScramUsersOptions>AbstractOptions<AlterScramUserCredentialsOptions> {} // interface Admin default AlterScramUsersResultAlterScramUserCredentialsResult alterScramUsersalterScramUserCredentials(List<ScramUserAlteration>List<UserScramCredentialAlteration> alterations) { return alterScramUsersalterScramUserCredentials(deletions, alterations, new AlterScramUsersOptionsAlterScramUserCredentialsOptions()); } AlterScramUsersResult alterScramUsers(List<ScramUserAlteration>// interface Admin AlterScramUserCredentialsResult alterScramUserCredentials(List<UserScramCredentialAlteration> alterations, AlterScramUsersOptionsAlterScramUserCredentialsOptions options); public class AlterScramCredentialsResultAlterUserScramCredentialsResult { public Map<String, KafkaFuture<Void>> values(); public KafkaFuture<Void> all(); // completes successfully only publicif Map<String,everything KafkaFuture<Void>>in resultsvalues(); 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 AlterScramUsersRequest AlterScramUserCredentialsRequest and AlterScramUsersResponse AlterScramUserCredentialsResponse implement the new API.
| Code Block | ||
|---|---|---|
| ||
{
"apiKey": 51,
"type": "request",
"name": "AlterScramUsersRequestAlterUserScramCredentialsRequest",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "AlterationsDeletions", "type": "[]ScramUserAlterationScramCredentialDeletion", "versions": "0+",
"about": "The SCRAM userscredentials to create or alterremove.", "fields": [
{ "name": "Name", "type": "string", "versions": "0+",
"about": "The user name." },
{ "name": "DeletionsMechanism", "type": "[]ScramCredentialDeletionint8", "versions": "0+",
"about": "The credentialsSCRAM to deletemechanism.", "fields": [ }
]},
{ "name": "MechanismUpsertions", "type": "int8[]ScramCredentialUpsertion", "versions": "0+",
"about": "The SCRAM mechanism credentials to update/insert." }
]},, "fields": [
{ "name": "AdditionsName", "type": "[]ScramCredentialAdditionstring", "versions": "0+",
"about": "The SCRAM credentials to adduser name." },
{ "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 of iterations." },
{ "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": "AlterScramUsersResponseAlterUserScramCredentialsResponse",
"validVersions": "0",
"flexibleVersions": "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": "Results", "type": "[]AlterScramUsersResultAlterUserScramCredentialsResult", "versions": "0+",
"about": "The results for deletions and alterations, inone theper same order as the requestaffected user.", "fields": [
{ "name": "User", "type": "string", "versions": "0+",
"about": "The user name." },
{ "name": "ErrorCode", "type": "int8int16", "versions": "0+",
"about": "The error code." },
{ "name": "ErrorStringErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
"about": "The error message, if any." }
]}
]
} |
An addition will return INVALIDUNACCEPTABLE_REQUEST CREDENTIAL 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 usedan 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.
...