You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ma...@apache.org on 2022/06/21 07:21:18 UTC
[kafka] branch trunk updated: KAFKA-6945: KIP-373, allow users to create delegation token for others (#10738)
This is an automated email from the ASF dual-hosted git repository.
manikumar pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new d65d886798 KAFKA-6945: KIP-373, allow users to create delegation token for others (#10738)
d65d886798 is described below
commit d65d8867983e54a36178c3a1227b68c8a468d415
Author: Viktor Somogyi-Vass <vi...@gmail.com>
AuthorDate: Tue Jun 21 09:21:08 2022 +0200
KAFKA-6945: KIP-373, allow users to create delegation token for others (#10738)
Reviewers: Manikumar Reddy <ma...@gmail.com>
---
.../admin/CreateDelegationTokenOptions.java | 11 ++
.../kafka/clients/admin/KafkaAdminClient.java | 13 +-
.../org/apache/kafka/common/acl/AclOperation.java | 12 +-
.../requests/CreateDelegationTokenRequest.java | 3 +-
.../requests/CreateDelegationTokenResponse.java | 13 +-
.../requests/DescribeDelegationTokenRequest.java | 2 +-
.../requests/DescribeDelegationTokenResponse.java | 38 +++---
.../apache/kafka/common/resource/ResourceType.java | 7 +-
.../token/delegation/TokenInformation.java | 36 +++---
.../common/message/CreateAclsRequest.json | 3 +-
.../common/message/CreateAclsResponse.json | 3 +-
.../message/CreateDelegationTokenRequest.json | 8 +-
.../message/CreateDelegationTokenResponse.json | 8 +-
.../common/message/DeleteAclsRequest.json | 3 +-
.../common/message/DeleteAclsResponse.json | 3 +-
.../common/message/DescribeAclsRequest.json | 3 +-
.../common/message/DescribeAclsResponse.json | 3 +-
.../message/DescribeDelegationTokenRequest.json | 3 +-
.../message/DescribeDelegationTokenResponse.json | 7 +-
.../apache/kafka/common/acl/AclOperationTest.java | 4 +-
.../kafka/common/requests/RequestResponseTest.java | 9 +-
.../kafka/common/resource/ResourceTypeTest.java | 3 +-
core/src/main/scala/kafka/admin/AclCommand.scala | 9 ++
.../scala/kafka/admin/DelegationTokenCommand.scala | 10 +-
.../scala/kafka/security/authorizer/AclEntry.scala | 3 +-
.../kafka/server/DelegationTokenManager.scala | 34 ++++--
core/src/main/scala/kafka/server/KafkaApis.scala | 36 ++++--
.../DelegationTokenEndToEndAuthorizationTest.scala | 47 ++++++--
...onTokenEndToEndAuthorizationWithOwnerTest.scala | 134 +++++++++++++++++++++
.../scala/unit/kafka/admin/AclCommandTest.scala | 7 +-
.../security/authorizer/AclAuthorizerTest.scala | 4 +-
.../delegation/DelegationTokenManagerTest.scala | 79 +++++++-----
32 files changed, 435 insertions(+), 123 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenOptions.java
index 6a082d499b..693e8f52f6 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenOptions.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenOptions.java
@@ -19,6 +19,7 @@ package org.apache.kafka.clients.admin;
import java.util.LinkedList;
import java.util.List;
+import java.util.Optional;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
@@ -32,6 +33,7 @@ import org.apache.kafka.common.security.auth.KafkaPrincipal;
public class CreateDelegationTokenOptions extends AbstractOptions<CreateDelegationTokenOptions> {
private long maxLifeTimeMs = -1;
private List<KafkaPrincipal> renewers = new LinkedList<>();
+ private KafkaPrincipal owner = null;
public CreateDelegationTokenOptions renewers(List<KafkaPrincipal> renewers) {
this.renewers = renewers;
@@ -42,6 +44,15 @@ public class CreateDelegationTokenOptions extends AbstractOptions<CreateDelegati
return renewers;
}
+ public CreateDelegationTokenOptions owner(KafkaPrincipal owner) {
+ this.owner = owner;
+ return this;
+ }
+
+ public Optional<KafkaPrincipal> owner() {
+ return Optional.ofNullable(owner);
+ }
+
public CreateDelegationTokenOptions maxlifeTimeMs(long maxLifeTimeMs) {
this.maxLifeTimeMs = maxLifeTimeMs;
return this;
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index 41a0a70240..4eeafbe55f 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -3066,10 +3066,14 @@ public class KafkaAdminClient extends AdminClient {
@Override
CreateDelegationTokenRequest.Builder createRequest(int timeoutMs) {
- return new CreateDelegationTokenRequest.Builder(
- new CreateDelegationTokenRequestData()
- .setRenewers(renewers)
- .setMaxLifetimeMs(options.maxlifeTimeMs()));
+ CreateDelegationTokenRequestData data = new CreateDelegationTokenRequestData()
+ .setRenewers(renewers)
+ .setMaxLifetimeMs(options.maxlifeTimeMs());
+ if (options.owner().isPresent()) {
+ data.setOwnerPrincipalName(options.owner().get().getName());
+ data.setOwnerPrincipalType(options.owner().get().getPrincipalType());
+ }
+ return new CreateDelegationTokenRequest.Builder(data);
}
@Override
@@ -3080,6 +3084,7 @@ public class KafkaAdminClient extends AdminClient {
} else {
CreateDelegationTokenResponseData data = response.data();
TokenInformation tokenInfo = new TokenInformation(data.tokenId(), new KafkaPrincipal(data.principalType(), data.principalName()),
+ new KafkaPrincipal(data.tokenRequesterPrincipalType(), data.tokenRequesterPrincipalName()),
options.renewers(), data.issueTimestampMs(), data.maxTimestampMs(), data.expiryTimestampMs());
DelegationToken token = new DelegationToken(tokenInfo, data.hmac());
delegationTokenFuture.complete(token);
diff --git a/clients/src/main/java/org/apache/kafka/common/acl/AclOperation.java b/clients/src/main/java/org/apache/kafka/common/acl/AclOperation.java
index 671069775c..19f4edaaa7 100644
--- a/clients/src/main/java/org/apache/kafka/common/acl/AclOperation.java
+++ b/clients/src/main/java/org/apache/kafka/common/acl/AclOperation.java
@@ -106,7 +106,17 @@ public enum AclOperation {
/**
* IDEMPOTENT_WRITE operation.
*/
- IDEMPOTENT_WRITE((byte) 12);
+ IDEMPOTENT_WRITE((byte) 12),
+
+ /**
+ * CREATE_TOKENS operation.
+ */
+ CREATE_TOKENS((byte) 13),
+
+ /**
+ * DESCRIBE_TOKENS operation.
+ */
+ DESCRIBE_TOKENS((byte) 14);
// Note: we cannot have more than 30 ACL operations without modifying the format used
// to describe ACL operations in MetadataResponse.
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateDelegationTokenRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateDelegationTokenRequest.java
index 1fee1b71eb..b48f84f1fa 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/CreateDelegationTokenRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/CreateDelegationTokenRequest.java
@@ -45,7 +45,8 @@ public class CreateDelegationTokenRequest extends AbstractRequest {
@Override
public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
- return CreateDelegationTokenResponse.prepareResponse(throttleTimeMs, Errors.forException(e), KafkaPrincipal.ANONYMOUS);
+ return CreateDelegationTokenResponse.prepareResponse(version(), throttleTimeMs, Errors.forException(e),
+ KafkaPrincipal.ANONYMOUS, KafkaPrincipal.ANONYMOUS);
}
public static class Builder extends AbstractRequest.Builder<CreateDelegationTokenRequest> {
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/CreateDelegationTokenResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/CreateDelegationTokenResponse.java
index b679a30c8d..22c2e12590 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/CreateDelegationTokenResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/CreateDelegationTokenResponse.java
@@ -39,9 +39,11 @@ public class CreateDelegationTokenResponse extends AbstractResponse {
new CreateDelegationTokenResponseData(new ByteBufferAccessor(buffer), version));
}
- public static CreateDelegationTokenResponse prepareResponse(int throttleTimeMs,
+ public static CreateDelegationTokenResponse prepareResponse(int version,
+ int throttleTimeMs,
Errors error,
KafkaPrincipal owner,
+ KafkaPrincipal tokenRequester,
long issueTimestamp,
long expiryTimestamp,
long maxTimestamp,
@@ -57,11 +59,16 @@ public class CreateDelegationTokenResponse extends AbstractResponse {
.setMaxTimestampMs(maxTimestamp)
.setTokenId(tokenId)
.setHmac(hmac.array());
+ if (version > 2) {
+ data.setTokenRequesterPrincipalType(tokenRequester.getPrincipalType())
+ .setTokenRequesterPrincipalName(tokenRequester.getName());
+ }
return new CreateDelegationTokenResponse(data);
}
- public static CreateDelegationTokenResponse prepareResponse(int throttleTimeMs, Errors error, KafkaPrincipal owner) {
- return prepareResponse(throttleTimeMs, error, owner, -1, -1, -1, "", ByteBuffer.wrap(new byte[] {}));
+ public static CreateDelegationTokenResponse prepareResponse(int version, int throttleTimeMs, Errors error,
+ KafkaPrincipal owner, KafkaPrincipal requester) {
+ return prepareResponse(version, throttleTimeMs, error, owner, requester, -1, -1, -1, "", ByteBuffer.wrap(new byte[] {}));
}
@Override
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeDelegationTokenRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeDelegationTokenRequest.java
index 9bf59e844a..bd3b5fd57c 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeDelegationTokenRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeDelegationTokenRequest.java
@@ -71,7 +71,7 @@ public class DescribeDelegationTokenRequest extends AbstractRequest {
@Override
public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
- return new DescribeDelegationTokenResponse(throttleTimeMs, Errors.forException(e));
+ return new DescribeDelegationTokenResponse(version(), throttleTimeMs, Errors.forException(e));
}
public static DescribeDelegationTokenRequest parse(ByteBuffer buffer, short version) {
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/DescribeDelegationTokenResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/DescribeDelegationTokenResponse.java
index 4a2162f53a..4fd1d99652 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/DescribeDelegationTokenResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/DescribeDelegationTokenResponse.java
@@ -36,22 +36,29 @@ public class DescribeDelegationTokenResponse extends AbstractResponse {
private final DescribeDelegationTokenResponseData data;
- public DescribeDelegationTokenResponse(int throttleTimeMs, Errors error, List<DelegationToken> tokens) {
+ public DescribeDelegationTokenResponse(int version, int throttleTimeMs, Errors error, List<DelegationToken> tokens) {
super(ApiKeys.DESCRIBE_DELEGATION_TOKEN);
List<DescribedDelegationToken> describedDelegationTokenList = tokens
.stream()
- .map(dt -> new DescribedDelegationToken()
- .setTokenId(dt.tokenInfo().tokenId())
- .setPrincipalType(dt.tokenInfo().owner().getPrincipalType())
- .setPrincipalName(dt.tokenInfo().owner().getName())
- .setIssueTimestamp(dt.tokenInfo().issueTimestamp())
- .setMaxTimestamp(dt.tokenInfo().maxTimestamp())
- .setExpiryTimestamp(dt.tokenInfo().expiryTimestamp())
- .setHmac(dt.hmac())
- .setRenewers(dt.tokenInfo().renewers()
- .stream()
- .map(r -> new DescribedDelegationTokenRenewer().setPrincipalName(r.getName()).setPrincipalType(r.getPrincipalType()))
- .collect(Collectors.toList())))
+ .map(dt -> {
+ DescribedDelegationToken ddt = new DescribedDelegationToken()
+ .setTokenId(dt.tokenInfo().tokenId())
+ .setPrincipalType(dt.tokenInfo().owner().getPrincipalType())
+ .setPrincipalName(dt.tokenInfo().owner().getName())
+ .setIssueTimestamp(dt.tokenInfo().issueTimestamp())
+ .setMaxTimestamp(dt.tokenInfo().maxTimestamp())
+ .setExpiryTimestamp(dt.tokenInfo().expiryTimestamp())
+ .setHmac(dt.hmac())
+ .setRenewers(dt.tokenInfo().renewers()
+ .stream()
+ .map(r -> new DescribedDelegationTokenRenewer().setPrincipalName(r.getName()).setPrincipalType(r.getPrincipalType()))
+ .collect(Collectors.toList()));
+ if (version > 2) {
+ ddt.setTokenRequesterPrincipalType(dt.tokenInfo().tokenRequester().getPrincipalType())
+ .setTokenRequesterPrincipalName(dt.tokenInfo().tokenRequester().getName());
+ }
+ return ddt;
+ })
.collect(Collectors.toList());
this.data = new DescribeDelegationTokenResponseData()
@@ -60,8 +67,8 @@ public class DescribeDelegationTokenResponse extends AbstractResponse {
.setTokens(describedDelegationTokenList);
}
- public DescribeDelegationTokenResponse(int throttleTimeMs, Errors error) {
- this(throttleTimeMs, error, new ArrayList<>());
+ public DescribeDelegationTokenResponse(int version, int throttleTimeMs, Errors error) {
+ this(version, throttleTimeMs, error, new ArrayList<>());
}
public DescribeDelegationTokenResponse(DescribeDelegationTokenResponseData data) {
@@ -99,6 +106,7 @@ public class DescribeDelegationTokenResponse extends AbstractResponse {
.map(ddt -> new DelegationToken(new TokenInformation(
ddt.tokenId(),
new KafkaPrincipal(ddt.principalType(), ddt.principalName()),
+ new KafkaPrincipal(ddt.tokenRequesterPrincipalType(), ddt.tokenRequesterPrincipalName()),
ddt.renewers()
.stream()
.map(ddtr -> new KafkaPrincipal(ddtr.principalType(), ddtr.principalName()))
diff --git a/clients/src/main/java/org/apache/kafka/common/resource/ResourceType.java b/clients/src/main/java/org/apache/kafka/common/resource/ResourceType.java
index 2ce653fbeb..225859812d 100644
--- a/clients/src/main/java/org/apache/kafka/common/resource/ResourceType.java
+++ b/clients/src/main/java/org/apache/kafka/common/resource/ResourceType.java
@@ -63,7 +63,12 @@ public enum ResourceType {
/**
* A token ID.
*/
- DELEGATION_TOKEN((byte) 6);
+ DELEGATION_TOKEN((byte) 6),
+
+ /**
+ * A user principal
+ */
+ USER((byte) 7);
private final static HashMap<Byte, ResourceType> CODE_TO_VALUE = new HashMap<>();
diff --git a/clients/src/main/java/org/apache/kafka/common/security/token/delegation/TokenInformation.java b/clients/src/main/java/org/apache/kafka/common/security/token/delegation/TokenInformation.java
index 9903eb51b2..1236824d86 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/token/delegation/TokenInformation.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/token/delegation/TokenInformation.java
@@ -30,17 +30,24 @@ import java.util.Objects;
@InterfaceStability.Evolving
public class TokenInformation {
- private KafkaPrincipal owner;
- private Collection<KafkaPrincipal> renewers;
- private long issueTimestamp;
- private long maxTimestamp;
+ private final KafkaPrincipal owner;
+ private final KafkaPrincipal tokenRequester;
+ private final Collection<KafkaPrincipal> renewers;
+ private final long issueTimestamp;
+ private final long maxTimestamp;
private long expiryTimestamp;
- private String tokenId;
+ private final String tokenId;
- public TokenInformation(String tokenId, KafkaPrincipal owner, Collection<KafkaPrincipal> renewers,
- long issueTimestamp, long maxTimestamp, long expiryTimestamp) {
+ public TokenInformation(String tokenId, KafkaPrincipal owner,
+ Collection<KafkaPrincipal> renewers, long issueTimestamp, long maxTimestamp, long expiryTimestamp) {
+ this(tokenId, owner, owner, renewers, issueTimestamp, maxTimestamp, expiryTimestamp);
+ }
+
+ public TokenInformation(String tokenId, KafkaPrincipal owner, KafkaPrincipal tokenRequester,
+ Collection<KafkaPrincipal> renewers, long issueTimestamp, long maxTimestamp, long expiryTimestamp) {
this.tokenId = tokenId;
this.owner = owner;
+ this.tokenRequester = tokenRequester;
this.renewers = renewers;
this.issueTimestamp = issueTimestamp;
this.maxTimestamp = maxTimestamp;
@@ -51,6 +58,10 @@ public class TokenInformation {
return owner;
}
+ public KafkaPrincipal tokenRequester() {
+ return tokenRequester;
+ }
+
public String ownerAsString() {
return owner.toString();
}
@@ -88,13 +99,14 @@ public class TokenInformation {
}
public boolean ownerOrRenewer(KafkaPrincipal principal) {
- return owner.equals(principal) || renewers.contains(principal);
+ return owner.equals(principal) || tokenRequester.equals(principal) || renewers.contains(principal);
}
@Override
public String toString() {
return "TokenInformation{" +
"owner=" + owner +
+ ", tokenRequester=" + tokenRequester +
", renewers=" + renewers +
", issueTimestamp=" + issueTimestamp +
", maxTimestamp=" + maxTimestamp +
@@ -117,17 +129,13 @@ public class TokenInformation {
return issueTimestamp == that.issueTimestamp &&
maxTimestamp == that.maxTimestamp &&
Objects.equals(owner, that.owner) &&
+ Objects.equals(tokenRequester, that.tokenRequester) &&
Objects.equals(renewers, that.renewers) &&
Objects.equals(tokenId, that.tokenId);
}
@Override
public int hashCode() {
- int result = owner != null ? owner.hashCode() : 0;
- result = 31 * result + (renewers != null ? renewers.hashCode() : 0);
- result = 31 * result + Long.hashCode(issueTimestamp);
- result = 31 * result + Long.hashCode(maxTimestamp);
- result = 31 * result + (tokenId != null ? tokenId.hashCode() : 0);
- return result;
+ return Objects.hash(owner, tokenRequester, renewers, issueTimestamp, maxTimestamp, expiryTimestamp, tokenId);
}
}
diff --git a/clients/src/main/resources/common/message/CreateAclsRequest.json b/clients/src/main/resources/common/message/CreateAclsRequest.json
index 5b3bfed781..89f5cf7f43 100644
--- a/clients/src/main/resources/common/message/CreateAclsRequest.json
+++ b/clients/src/main/resources/common/message/CreateAclsRequest.json
@@ -20,7 +20,8 @@
"name": "CreateAclsRequest",
// Version 1 adds resource pattern type.
// Version 2 enables flexible versions.
- "validVersions": "0-2",
+ // Version 3 adds user resource type.
+ "validVersions": "0-3",
"flexibleVersions": "2+",
"fields": [
{ "name": "Creations", "type": "[]AclCreation", "versions": "0+",
diff --git a/clients/src/main/resources/common/message/CreateAclsResponse.json b/clients/src/main/resources/common/message/CreateAclsResponse.json
index 7b0de7e56b..da1632c03b 100644
--- a/clients/src/main/resources/common/message/CreateAclsResponse.json
+++ b/clients/src/main/resources/common/message/CreateAclsResponse.json
@@ -19,7 +19,8 @@
"name": "CreateAclsResponse",
// Starting in version 1, on quota violation, brokers send out responses before throttling.
// Version 2 enables flexible versions.
- "validVersions": "0-2",
+ // Version 3 adds user resource type.
+ "validVersions": "0-3",
"flexibleVersions": "2+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
diff --git a/clients/src/main/resources/common/message/CreateDelegationTokenRequest.json b/clients/src/main/resources/common/message/CreateDelegationTokenRequest.json
index d65d490a6e..8444f16c00 100644
--- a/clients/src/main/resources/common/message/CreateDelegationTokenRequest.json
+++ b/clients/src/main/resources/common/message/CreateDelegationTokenRequest.json
@@ -21,9 +21,15 @@
// Version 1 is the same as version 0.
//
// Version 2 is the first flexible version.
- "validVersions": "0-2",
+ //
+ // Version 3 adds owner principal
+ "validVersions": "0-3",
"flexibleVersions": "2+",
"fields": [
+ { "name": "OwnerPrincipalType", "type": "string", "versions": "3+", "nullableVersions": "3+",
+ "about": "The principal type of the owner of the token. If it's null it defaults to the token request principal." },
+ { "name": "OwnerPrincipalName", "type": "string", "versions": "3+", "nullableVersions": "3+",
+ "about": "The principal name of the owner of the token. If it's null it defaults to the token request principal." },
{ "name": "Renewers", "type": "[]CreatableRenewers", "versions": "0+",
"about": "A list of those who are allowed to renew this token before it expires.", "fields": [
{ "name": "PrincipalType", "type": "string", "versions": "0+",
diff --git a/clients/src/main/resources/common/message/CreateDelegationTokenResponse.json b/clients/src/main/resources/common/message/CreateDelegationTokenResponse.json
index 74ad905b94..bf8be5573b 100644
--- a/clients/src/main/resources/common/message/CreateDelegationTokenResponse.json
+++ b/clients/src/main/resources/common/message/CreateDelegationTokenResponse.json
@@ -20,7 +20,9 @@
// Starting in version 1, on quota violation, brokers send out responses before throttling.
//
// Version 2 is the first flexible version.
- "validVersions": "0-2",
+ //
+ // Version 3 adds token requester details
+ "validVersions": "0-3",
"flexibleVersions": "2+",
"fields": [
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
@@ -29,6 +31,10 @@
"about": "The principal type of the token owner." },
{ "name": "PrincipalName", "type": "string", "versions": "0+",
"about": "The name of the token owner." },
+ { "name": "TokenRequesterPrincipalType", "type": "string", "versions": "3+",
+ "about": "The principal type of the requester of the token." },
+ { "name": "TokenRequesterPrincipalName", "type": "string", "versions": "3+",
+ "about": "The principal type of the requester of the token." },
{ "name": "IssueTimestampMs", "type": "int64", "versions": "0+",
"about": "When this token was generated." },
{ "name": "ExpiryTimestampMs", "type": "int64", "versions": "0+",
diff --git a/clients/src/main/resources/common/message/DeleteAclsRequest.json b/clients/src/main/resources/common/message/DeleteAclsRequest.json
index fd7c1522b4..ea7106d4c8 100644
--- a/clients/src/main/resources/common/message/DeleteAclsRequest.json
+++ b/clients/src/main/resources/common/message/DeleteAclsRequest.json
@@ -20,7 +20,8 @@
"name": "DeleteAclsRequest",
// Version 1 adds the pattern type.
// Version 2 enables flexible versions.
- "validVersions": "0-2",
+ // Version 3 adds the user resource type.
+ "validVersions": "0-3",
"flexibleVersions": "2+",
"fields": [
{ "name": "Filters", "type": "[]DeleteAclsFilter", "versions": "0+",
diff --git a/clients/src/main/resources/common/message/DeleteAclsResponse.json b/clients/src/main/resources/common/message/DeleteAclsResponse.json
index 08f570283e..e00969df7a 100644
--- a/clients/src/main/resources/common/message/DeleteAclsResponse.json
+++ b/clients/src/main/resources/common/message/DeleteAclsResponse.json
@@ -20,7 +20,8 @@
// Version 1 adds the resource pattern type.
// Starting in version 1, on quota violation, brokers send out responses before throttling.
// Version 2 enables flexible versions.
- "validVersions": "0-2",
+ // Version 3 adds the user resource type.
+ "validVersions": "0-3",
"flexibleVersions": "2+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
diff --git a/clients/src/main/resources/common/message/DescribeAclsRequest.json b/clients/src/main/resources/common/message/DescribeAclsRequest.json
index 58886da654..4f0e851c72 100644
--- a/clients/src/main/resources/common/message/DescribeAclsRequest.json
+++ b/clients/src/main/resources/common/message/DescribeAclsRequest.json
@@ -20,7 +20,8 @@
"name": "DescribeAclsRequest",
// Version 1 adds resource pattern type.
// Version 2 enables flexible versions.
- "validVersions": "0-2",
+ // Version 3 adds user resource type.
+ "validVersions": "0-3",
"flexibleVersions": "2+",
"fields": [
{ "name": "ResourceTypeFilter", "type": "int8", "versions": "0+",
diff --git a/clients/src/main/resources/common/message/DescribeAclsResponse.json b/clients/src/main/resources/common/message/DescribeAclsResponse.json
index 0ae72d67c4..19de109445 100644
--- a/clients/src/main/resources/common/message/DescribeAclsResponse.json
+++ b/clients/src/main/resources/common/message/DescribeAclsResponse.json
@@ -20,7 +20,8 @@
// Version 1 adds PatternType.
// Starting in version 1, on quota violation, brokers send out responses before throttling.
// Version 2 enables flexible versions.
- "validVersions": "0-2",
+ // Version 3 adds user resource type.
+ "validVersions": "0-3",
"flexibleVersions": "2+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
diff --git a/clients/src/main/resources/common/message/DescribeDelegationTokenRequest.json b/clients/src/main/resources/common/message/DescribeDelegationTokenRequest.json
index 79c342e14e..e92c57e5a2 100644
--- a/clients/src/main/resources/common/message/DescribeDelegationTokenRequest.json
+++ b/clients/src/main/resources/common/message/DescribeDelegationTokenRequest.json
@@ -20,7 +20,8 @@
"name": "DescribeDelegationTokenRequest",
// Version 1 is the same as version 0.
// Version 2 adds flexible version support
- "validVersions": "0-2",
+ // Version 3 adds token requester into the response
+ "validVersions": "0-3",
"flexibleVersions": "2+",
"fields": [
{ "name": "Owners", "type": "[]DescribeDelegationTokenOwner", "versions": "0+", "nullableVersions": "0+",
diff --git a/clients/src/main/resources/common/message/DescribeDelegationTokenResponse.json b/clients/src/main/resources/common/message/DescribeDelegationTokenResponse.json
index 09f69ce61c..3258164caf 100644
--- a/clients/src/main/resources/common/message/DescribeDelegationTokenResponse.json
+++ b/clients/src/main/resources/common/message/DescribeDelegationTokenResponse.json
@@ -19,7 +19,8 @@
"name": "DescribeDelegationTokenResponse",
// Starting in version 1, on quota violation, brokers send out responses before throttling.
// Version 2 adds flexible version support
- "validVersions": "0-2",
+ // Version 3 adds token requester details
+ "validVersions": "0-3",
"flexibleVersions": "2+",
"fields": [
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
@@ -30,6 +31,10 @@
"about": "The token principal type." },
{ "name": "PrincipalName", "type": "string", "versions": "0+",
"about": "The token principal name." },
+ { "name": "TokenRequesterPrincipalType", "type": "string", "versions": "3+",
+ "about": "The principal type of the requester of the token." },
+ { "name": "TokenRequesterPrincipalName", "type": "string", "versions": "3+",
+ "about": "The principal type of the requester of the token." },
{ "name": "IssueTimestamp", "type": "int64", "versions": "0+",
"about": "The token issue timestamp in milliseconds." },
{ "name": "ExpiryTimestamp", "type": "int64", "versions": "0+",
diff --git a/clients/src/test/java/org/apache/kafka/common/acl/AclOperationTest.java b/clients/src/test/java/org/apache/kafka/common/acl/AclOperationTest.java
index c807e2be95..b91db6f206 100644
--- a/clients/src/test/java/org/apache/kafka/common/acl/AclOperationTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/acl/AclOperationTest.java
@@ -48,7 +48,9 @@ public class AclOperationTest {
new AclOperationTestInfo(AclOperation.CLUSTER_ACTION, 9, "cluster_action", false),
new AclOperationTestInfo(AclOperation.DESCRIBE_CONFIGS, 10, "describe_configs", false),
new AclOperationTestInfo(AclOperation.ALTER_CONFIGS, 11, "alter_configs", false),
- new AclOperationTestInfo(AclOperation.IDEMPOTENT_WRITE, 12, "idempotent_write", false)
+ new AclOperationTestInfo(AclOperation.IDEMPOTENT_WRITE, 12, "idempotent_write", false),
+ new AclOperationTestInfo(AclOperation.CREATE_TOKENS, 13, "create_tokens", false),
+ new AclOperationTestInfo(AclOperation.DESCRIBE_TOKENS, 14, "describe_tokens", false)
};
@Test
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
index 13f47a9771..b6df4c44d6 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
@@ -258,6 +258,7 @@ import static org.apache.kafka.common.protocol.ApiKeys.CREATE_TOPICS;
import static org.apache.kafka.common.protocol.ApiKeys.DELETE_ACLS;
import static org.apache.kafka.common.protocol.ApiKeys.DELETE_TOPICS;
import static org.apache.kafka.common.protocol.ApiKeys.DESCRIBE_CONFIGS;
+import static org.apache.kafka.common.protocol.ApiKeys.DESCRIBE_DELEGATION_TOKEN;
import static org.apache.kafka.common.protocol.ApiKeys.DESCRIBE_LOG_DIRS;
import static org.apache.kafka.common.protocol.ApiKeys.ELECT_LEADERS;
import static org.apache.kafka.common.protocol.ApiKeys.FETCH;
@@ -939,7 +940,7 @@ public class RequestResponseTest {
assertEquals(2, createDescribeConfigsResponse(DESCRIBE_CONFIGS.latestVersion()).errorCounts().get(Errors.NONE));
assertEquals(1, createDescribeGroupResponse().errorCounts().get(Errors.NONE));
assertEquals(2, createDescribeLogDirsResponse().errorCounts().get(Errors.NONE));
- assertEquals(1, createDescribeTokenResponse().errorCounts().get(Errors.NONE));
+ assertEquals(1, createDescribeTokenResponse(DESCRIBE_DELEGATION_TOKEN.latestVersion()).errorCounts().get(Errors.NONE));
assertEquals(2, createElectLeadersResponse().errorCounts().get(Errors.NONE));
assertEquals(1, createEndTxnResponse().errorCounts().get(Errors.NONE));
assertEquals(1, createExpireTokenResponse().errorCounts().get(Errors.NONE));
@@ -1088,7 +1089,7 @@ public class RequestResponseTest {
case CREATE_DELEGATION_TOKEN: return createCreateTokenResponse();
case RENEW_DELEGATION_TOKEN: return createRenewTokenResponse();
case EXPIRE_DELEGATION_TOKEN: return createExpireTokenResponse();
- case DESCRIBE_DELEGATION_TOKEN: return createDescribeTokenResponse();
+ case DESCRIBE_DELEGATION_TOKEN: return createDescribeTokenResponse(version);
case DELETE_GROUPS: return createDeleteGroupsResponse();
case ELECT_LEADERS: return createElectLeadersResponse();
case INCREMENTAL_ALTER_CONFIGS: return createIncrementalAlterConfigsResponse();
@@ -2980,7 +2981,7 @@ public class RequestResponseTest {
return new DescribeDelegationTokenRequest.Builder(owners).build(version);
}
- private DescribeDelegationTokenResponse createDescribeTokenResponse() {
+ private DescribeDelegationTokenResponse createDescribeTokenResponse(short version) {
List<KafkaPrincipal> renewers = new ArrayList<>();
renewers.add(SecurityUtils.parseKafkaPrincipal("User:user1"));
renewers.add(SecurityUtils.parseKafkaPrincipal("User:user2"));
@@ -2996,7 +2997,7 @@ public class RequestResponseTest {
tokenList.add(new DelegationToken(tokenInfo1, "test".getBytes()));
tokenList.add(new DelegationToken(tokenInfo2, "test".getBytes()));
- return new DescribeDelegationTokenResponse(20, Errors.NONE, tokenList);
+ return new DescribeDelegationTokenResponse(version, 20, Errors.NONE, tokenList);
}
private ElectLeadersRequest createElectLeadersRequestNullPartitions() {
diff --git a/clients/src/test/java/org/apache/kafka/common/resource/ResourceTypeTest.java b/clients/src/test/java/org/apache/kafka/common/resource/ResourceTypeTest.java
index fcde96865e..d40c8aeac6 100644
--- a/clients/src/test/java/org/apache/kafka/common/resource/ResourceTypeTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/resource/ResourceTypeTest.java
@@ -42,7 +42,8 @@ public class ResourceTypeTest {
new AclResourceTypeTestInfo(ResourceType.GROUP, 3, "group", false),
new AclResourceTypeTestInfo(ResourceType.CLUSTER, 4, "cluster", false),
new AclResourceTypeTestInfo(ResourceType.TRANSACTIONAL_ID, 5, "transactional_id", false),
- new AclResourceTypeTestInfo(ResourceType.DELEGATION_TOKEN, 6, "delegation_token", false)
+ new AclResourceTypeTestInfo(ResourceType.DELEGATION_TOKEN, 6, "delegation_token", false),
+ new AclResourceTypeTestInfo(ResourceType.USER, 7, "user", false)
};
@Test
diff --git a/core/src/main/scala/kafka/admin/AclCommand.scala b/core/src/main/scala/kafka/admin/AclCommand.scala
index cca997d408..769e99df73 100644
--- a/core/src/main/scala/kafka/admin/AclCommand.scala
+++ b/core/src/main/scala/kafka/admin/AclCommand.scala
@@ -467,6 +467,9 @@ object AclCommand extends Logging {
if (opts.options.has(opts.delegationTokenOpt))
opts.options.valuesOf(opts.delegationTokenOpt).forEach(token => resourceFilters += new ResourcePatternFilter(JResourceType.DELEGATION_TOKEN, token.trim, patternType))
+ if (opts.options.has(opts.userPrincipalOpt))
+ opts.options.valuesOf(opts.userPrincipalOpt).forEach(user => resourceFilters += new ResourcePatternFilter(JResourceType.USER, user.trim, patternType))
+
if (resourceFilters.isEmpty && dieIfNoResourceFound)
CommandLineUtils.printUsageAndDie(opts.parser, "You must provide at least one resource: --topic <topic> or --cluster or --group <group> or --delegation-token <Delegation Token ID>")
@@ -621,6 +624,12 @@ object AclCommand extends Logging {
AclCommand.AuthorizerDeprecationMessage)
.withRequiredArg().describedAs("Authorizer ZooKeeper TLS configuration").ofType(classOf[String])
+ val userPrincipalOpt = parser.accepts("user-principal", "Specifies a user principal as a resource in relation with the operation. For instance " +
+ "one could grant CreateTokens or DescribeTokens permission on a given user principal.")
+ .withRequiredArg()
+ .describedAs("user-principal")
+ .ofType(classOf[String])
+
options = parser.parse(args: _*)
def checkArgs(): Unit = {
diff --git a/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala b/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala
index 6465b143e3..78984792ce 100644
--- a/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala
+++ b/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala
@@ -78,6 +78,9 @@ object DelegationTokenCommand extends Logging {
println("Calling create token operation with renewers :" + renewerPrincipals +" , max-life-time-period :"+ maxLifeTimeMs)
val createDelegationTokenOptions = new CreateDelegationTokenOptions().maxlifeTimeMs(maxLifeTimeMs).renewers(renewerPrincipals)
+ val ownerPrincipal = getPrincipals(opts, opts.ownerPrincipalsOpt)
+ if (ownerPrincipal.isDefined)
+ createDelegationTokenOptions.owner(ownerPrincipal.get.asScala.head)
val createResult = adminClient.createDelegationToken(createDelegationTokenOptions)
val token = createResult.delegationToken().get()
println("Created delegation token with tokenId : %s".format(token.tokenInfo.tokenId)); printToken(List(token))
@@ -86,13 +89,14 @@ object DelegationTokenCommand extends Logging {
def printToken(tokens: List[DelegationToken]): Unit = {
val dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm")
- print("\n%-15s %-30s %-15s %-25s %-15s %-15s %-15s".format("TOKENID", "HMAC", "OWNER", "RENEWERS", "ISSUEDATE", "EXPIRYDATE", "MAXDATE"))
+ print("\n%-15s %-30s %-15s %-15s %-25s %-15s %-15s %-15s".format("TOKENID", "HMAC", "OWNER", "REQUESTER", "RENEWERS", "ISSUEDATE", "EXPIRYDATE", "MAXDATE"))
for (token <- tokens) {
val tokenInfo = token.tokenInfo
- print("\n%-15s %-30s %-15s %-25s %-15s %-15s %-15s".format(
+ print("\n%-15s %-30s %-15s %-15s %-25s %-15s %-15s %-15s".format(
tokenInfo.tokenId,
token.hmacAsBase64String,
tokenInfo.owner,
+ tokenInfo.tokenRequester(),
tokenInfo.renewersAsString,
dateFormat.format(tokenInfo.issueTimestamp),
dateFormat.format(tokenInfo.expiryTimestamp),
@@ -210,7 +214,7 @@ object DelegationTokenCommand extends Logging {
CommandLineUtils.checkRequiredArgs(parser, options, hmacOpt, expiryTimePeriodOpt)
// check invalid args
- CommandLineUtils.checkInvalidArgs(parser, options, createOpt, Set(hmacOpt, renewTimePeriodOpt, expiryTimePeriodOpt, ownerPrincipalsOpt))
+ CommandLineUtils.checkInvalidArgs(parser, options, createOpt, Set(hmacOpt, renewTimePeriodOpt, expiryTimePeriodOpt))
CommandLineUtils.checkInvalidArgs(parser, options, renewOpt, Set(renewPrincipalsOpt, maxLifeTimeOpt, expiryTimePeriodOpt, ownerPrincipalsOpt))
CommandLineUtils.checkInvalidArgs(parser, options, expiryOpt, Set(renewOpt, maxLifeTimeOpt, renewTimePeriodOpt, ownerPrincipalsOpt))
CommandLineUtils.checkInvalidArgs(parser, options, describeOpt, Set(renewTimePeriodOpt, maxLifeTimeOpt, hmacOpt, renewTimePeriodOpt, expiryTimePeriodOpt))
diff --git a/core/src/main/scala/kafka/security/authorizer/AclEntry.scala b/core/src/main/scala/kafka/security/authorizer/AclEntry.scala
index 2014916e7e..9e2d49fc88 100644
--- a/core/src/main/scala/kafka/security/authorizer/AclEntry.scala
+++ b/core/src/main/scala/kafka/security/authorizer/AclEntry.scala
@@ -19,7 +19,7 @@ package kafka.security.authorizer
import kafka.utils.Json
import org.apache.kafka.common.acl.{AccessControlEntry, AclOperation, AclPermissionType}
-import org.apache.kafka.common.acl.AclOperation.{READ, WRITE, CREATE, DESCRIBE, DELETE, ALTER, DESCRIBE_CONFIGS, ALTER_CONFIGS, CLUSTER_ACTION, IDEMPOTENT_WRITE}
+import org.apache.kafka.common.acl.AclOperation.{READ, WRITE, CREATE, DESCRIBE, DELETE, ALTER, DESCRIBE_CONFIGS, ALTER_CONFIGS, CLUSTER_ACTION, IDEMPOTENT_WRITE, CREATE_TOKENS, DESCRIBE_TOKENS}
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.resource.{ResourcePattern, ResourceType}
import org.apache.kafka.common.security.auth.KafkaPrincipal
@@ -103,6 +103,7 @@ object AclEntry {
case ResourceType.CLUSTER => Set(CREATE, CLUSTER_ACTION, DESCRIBE_CONFIGS, ALTER_CONFIGS, IDEMPOTENT_WRITE, ALTER, DESCRIBE)
case ResourceType.TRANSACTIONAL_ID => Set(DESCRIBE, WRITE)
case ResourceType.DELEGATION_TOKEN => Set(DESCRIBE)
+ case ResourceType.USER => Set(CREATE_TOKENS, DESCRIBE_TOKENS)
case _ => throw new IllegalArgumentException("Not a concrete resource type")
}
}
diff --git a/core/src/main/scala/kafka/server/DelegationTokenManager.scala b/core/src/main/scala/kafka/server/DelegationTokenManager.scala
index 8b96e72543..235e3a12ad 100644
--- a/core/src/main/scala/kafka/server/DelegationTokenManager.scala
+++ b/core/src/main/scala/kafka/server/DelegationTokenManager.scala
@@ -42,13 +42,14 @@ import scala.collection.mutable
object DelegationTokenManager {
val DefaultHmacAlgorithm = "HmacSHA512"
val OwnerKey ="owner"
+ val TokenRequesterKey = "tokenRequester"
val RenewersKey = "renewers"
val IssueTimestampKey = "issueTimestamp"
val MaxTimestampKey = "maxTimestamp"
val ExpiryTimestampKey = "expiryTimestamp"
val TokenIdKey = "tokenId"
val VersionKey = "version"
- val CurrentVersion = 1
+ val CurrentVersion = 3
val ErrorTimestamp = -1
/**
@@ -103,6 +104,7 @@ object DelegationTokenManager {
val tokenInfoMap = mutable.Map[String, Any]()
tokenInfoMap(VersionKey) = CurrentVersion
tokenInfoMap(OwnerKey) = Sanitizer.sanitize(tokenInfo.ownerAsString)
+ tokenInfoMap(TokenRequesterKey) = Sanitizer.sanitize(tokenInfo.tokenRequester.toString)
tokenInfoMap(RenewersKey) = tokenInfo.renewersAsString.asScala.map(e => Sanitizer.sanitize(e)).asJava
tokenInfoMap(IssueTimestampKey) = tokenInfo.issueTimestamp
tokenInfoMap(MaxTimestampKey) = tokenInfo.maxTimestamp
@@ -118,8 +120,12 @@ object DelegationTokenManager {
Json.parseBytes(bytes) match {
case Some(js) =>
val mainJs = js.asJsonObject
- require(mainJs(VersionKey).to[Int] == CurrentVersion)
+ val version = mainJs(VersionKey).to[Int]
+ require(version > 0 && version <= CurrentVersion)
val owner = SecurityUtils.parseKafkaPrincipal(Sanitizer.desanitize(mainJs(OwnerKey).to[String]))
+ var tokenRequester = owner
+ if (version >= 3)
+ tokenRequester = SecurityUtils.parseKafkaPrincipal(Sanitizer.desanitize(mainJs(TokenRequesterKey).to[String]))
val renewerStr = mainJs(RenewersKey).to[Seq[String]]
val renewers = renewerStr.map(Sanitizer.desanitize).map(SecurityUtils.parseKafkaPrincipal)
val issueTimestamp = mainJs(IssueTimestampKey).to[Long]
@@ -127,7 +133,7 @@ object DelegationTokenManager {
val maxTimestamp = mainJs(MaxTimestampKey).to[Long]
val tokenId = mainJs(TokenIdKey).to[String]
- val tokenInfo = new TokenInformation(tokenId, owner, renewers.asJava,
+ val tokenInfo = new TokenInformation(tokenId, owner, tokenRequester, renewers.asJava,
issueTimestamp, maxTimestamp, expiryTimestamp)
Some(tokenInfo)
@@ -136,17 +142,18 @@ object DelegationTokenManager {
}
}
- def filterToken(requestedPrincipal: KafkaPrincipal, owners : Option[List[KafkaPrincipal]], token: TokenInformation, authorizeToken: String => Boolean) : Boolean = {
+ def filterToken(requesterPrincipal: KafkaPrincipal, owners : Option[List[KafkaPrincipal]], token: TokenInformation,
+ authorizeToken: String => Boolean, authorizeRequester: KafkaPrincipal => Boolean) : Boolean = {
val allow =
//exclude tokens which are not requested
if (owners.isDefined && !owners.get.exists(owner => token.ownerOrRenewer(owner))) {
false
//Owners and the renewers can describe their own tokens
- } else if (token.ownerOrRenewer(requestedPrincipal)) {
+ } else if (token.ownerOrRenewer(requesterPrincipal)) {
true
// Check permission for non-owned tokens
- } else if (authorizeToken(token.tokenId)) {
+ } else if (authorizeToken(token.tokenId) || authorizeRequester(token.owner)) {
true
}
else {
@@ -259,12 +266,13 @@ class DelegationTokenManager(val config: KafkaConfig,
* @param responseCallback
*/
def createToken(owner: KafkaPrincipal,
+ tokenRequester: KafkaPrincipal,
renewers: List[KafkaPrincipal],
maxLifeTimeMs: Long,
responseCallback: CreateResponseCallback): Unit = {
if (!config.tokenAuthEnabled) {
- responseCallback(CreateTokenResult(-1, -1, -1, "", Array[Byte](), Errors.DELEGATION_TOKEN_AUTH_DISABLED))
+ responseCallback(CreateTokenResult(owner, tokenRequester, -1, -1, -1, "", Array[Byte](), Errors.DELEGATION_TOKEN_AUTH_DISABLED))
} else {
lock.synchronized {
val tokenId = CoreUtils.generateUuidAsBase64()
@@ -274,13 +282,13 @@ class DelegationTokenManager(val config: KafkaConfig,
val maxLifeTimeStamp = issueTimeStamp + maxLifeTime
val expiryTimeStamp = Math.min(maxLifeTimeStamp, issueTimeStamp + defaultTokenRenewTime)
- val tokenInfo = new TokenInformation(tokenId, owner, renewers.asJava, issueTimeStamp, maxLifeTimeStamp, expiryTimeStamp)
+ val tokenInfo = new TokenInformation(tokenId, owner, tokenRequester, renewers.asJava, issueTimeStamp, maxLifeTimeStamp, expiryTimeStamp)
val hmac = createHmac(tokenId, secretKey)
val token = new DelegationToken(tokenInfo, hmac)
updateToken(token)
info(s"Created a delegation token: $tokenId for owner: $owner")
- responseCallback(CreateTokenResult(issueTimeStamp, expiryTimeStamp, maxLifeTimeStamp, tokenId, hmac, Errors.NONE))
+ responseCallback(CreateTokenResult(owner, tokenRequester, issueTimeStamp, expiryTimeStamp, maxLifeTimeStamp, tokenId, hmac, Errors.NONE))
}
}
}
@@ -485,7 +493,9 @@ class DelegationTokenManager(val config: KafkaConfig,
}
-case class CreateTokenResult(issueTimestamp: Long,
+case class CreateTokenResult(owner: KafkaPrincipal,
+ tokenRequester: KafkaPrincipal,
+ issueTimestamp: Long,
expiryTimestamp: Long,
maxTimestamp: Long,
tokenId: String,
@@ -496,6 +506,8 @@ case class CreateTokenResult(issueTimestamp: Long,
other match {
case that: CreateTokenResult =>
error.equals(that.error) &&
+ owner.equals(that.owner) &&
+ tokenRequester.equals(that.tokenRequester) &&
tokenId.equals(that.tokenId) &&
issueTimestamp.equals(that.issueTimestamp) &&
expiryTimestamp.equals(that.expiryTimestamp) &&
@@ -506,7 +518,7 @@ case class CreateTokenResult(issueTimestamp: Long,
}
override def hashCode(): Int = {
- val fields = Seq(issueTimestamp, expiryTimestamp, maxTimestamp, tokenId, hmac, error)
+ val fields = Seq(owner, tokenRequester, issueTimestamp, expiryTimestamp, maxTimestamp, tokenId, hmac, error)
fields.map(_.hashCode()).foldLeft(0)((a, b) => 31 * a + b)
}
}
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index b128db596c..4703118ebf 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -2934,24 +2934,40 @@ class KafkaApis(val requestChannel: RequestChannel,
trace(s"Sending create token response for correlation id ${request.header.correlationId} " +
s"to client ${request.header.clientId}.")
requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
- CreateDelegationTokenResponse.prepareResponse(requestThrottleMs, createResult.error, request.context.principal, createResult.issueTimestamp,
- createResult.expiryTimestamp, createResult.maxTimestamp, createResult.tokenId, ByteBuffer.wrap(createResult.hmac)))
+ CreateDelegationTokenResponse.prepareResponse(request.context.requestVersion, requestThrottleMs, createResult.error, createResult.owner,
+ createResult.tokenRequester, createResult.issueTimestamp, createResult.expiryTimestamp, createResult.maxTimestamp, createResult.tokenId,
+ ByteBuffer.wrap(createResult.hmac)))
}
- if (!allowTokenRequests(request))
+ val ownerPrincipalName = createTokenRequest.data.ownerPrincipalName
+ val owner = if (ownerPrincipalName == null || ownerPrincipalName.isEmpty) {
+ request.context.principal
+ } else {
+ new KafkaPrincipal(createTokenRequest.data.ownerPrincipalType, ownerPrincipalName)
+ }
+ val requester = request.context.principal
+
+ if (!allowTokenRequests(request)) {
requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
- CreateDelegationTokenResponse.prepareResponse(requestThrottleMs, Errors.DELEGATION_TOKEN_REQUEST_NOT_ALLOWED, request.context.principal))
- else {
+ CreateDelegationTokenResponse.prepareResponse(request.context.requestVersion, requestThrottleMs,
+ Errors.DELEGATION_TOKEN_REQUEST_NOT_ALLOWED, owner, requester))
+ } else if (!owner.equals(requester) && !authHelper.authorize(request.context, CREATE_TOKENS, USER, owner.toString)) {
+ requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
+ CreateDelegationTokenResponse.prepareResponse(request.context.requestVersion, requestThrottleMs,
+ Errors.DELEGATION_TOKEN_AUTHORIZATION_FAILED, owner, requester))
+ } else {
val renewerList = createTokenRequest.data.renewers.asScala.toList.map(entry =>
new KafkaPrincipal(entry.principalType, entry.principalName))
if (renewerList.exists(principal => principal.getPrincipalType != KafkaPrincipal.USER_TYPE)) {
requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
- CreateDelegationTokenResponse.prepareResponse(requestThrottleMs, Errors.INVALID_PRINCIPAL_TYPE, request.context.principal))
+ CreateDelegationTokenResponse.prepareResponse(request.context.requestVersion, requestThrottleMs,
+ Errors.INVALID_PRINCIPAL_TYPE, owner, requester))
}
else {
tokenManager.createToken(
- request.context.principal,
+ owner,
+ requester,
renewerList,
createTokenRequest.data.maxLifetimeMs,
sendResponseCallback
@@ -3022,7 +3038,7 @@ class KafkaApis(val requestChannel: RequestChannel,
// the callback for sending a describe token response
def sendResponseCallback(error: Errors, tokenDetails: List[DelegationToken]): Unit = {
requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
- new DescribeDelegationTokenResponse(requestThrottleMs, error, tokenDetails.asJava))
+ new DescribeDelegationTokenResponse(request.context.requestVersion(), requestThrottleMs, error, tokenDetails.asJava))
trace("Sending describe token response for correlation id %d to client %s."
.format(request.header.correlationId, request.header.clientId))
}
@@ -3043,7 +3059,9 @@ class KafkaApis(val requestChannel: RequestChannel,
else
Some(describeTokenRequest.data.owners.asScala.map(p => new KafkaPrincipal(p.principalType(), p.principalName)).toList)
def authorizeToken(tokenId: String) = authHelper.authorize(request.context, DESCRIBE, DELEGATION_TOKEN, tokenId)
- def eligible(token: TokenInformation) = DelegationTokenManager.filterToken(requestPrincipal, owners, token, authorizeToken)
+ def authorizeRequester(owner: KafkaPrincipal) = authHelper.authorize(request.context, DESCRIBE_TOKENS, USER, owner.toString)
+ def eligible(token: TokenInformation) = DelegationTokenManager
+ .filterToken(requestPrincipal, owners, token, authorizeToken, authorizeRequester)
val tokens = tokenManager.getTokens(eligible)
sendResponseCallback(Errors.NONE, tokens)
}
diff --git a/core/src/test/scala/integration/kafka/api/DelegationTokenEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/DelegationTokenEndToEndAuthorizationTest.scala
index f025cc3471..2d63e77b11 100644
--- a/core/src/test/scala/integration/kafka/api/DelegationTokenEndToEndAuthorizationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/DelegationTokenEndToEndAuthorizationTest.scala
@@ -21,7 +21,7 @@ import java.util.Properties
import kafka.server.KafkaConfig
import kafka.utils.{JaasTestUtils, TestUtils}
import kafka.zk.ConfigEntityChangeNotificationZNode
-import org.apache.kafka.clients.admin.{Admin, AdminClientConfig, ScramCredentialInfo, UserScramCredentialAlteration, UserScramCredentialUpsertion, ScramMechanism => PublicScramMechanism}
+import org.apache.kafka.clients.admin.{Admin, AdminClientConfig, CreateDelegationTokenOptions, ScramCredentialInfo, UserScramCredentialAlteration, UserScramCredentialUpsertion, ScramMechanism => PublicScramMechanism}
import org.apache.kafka.common.config.SaslConfigs
import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
import org.apache.kafka.common.security.scram.internals.ScramMechanism
@@ -45,20 +45,27 @@ class DelegationTokenEndToEndAuthorizationTest extends EndToEndAuthorizationTest
private val clientPassword = JaasTestUtils.KafkaScramPassword
override val kafkaPrincipal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, JaasTestUtils.KafkaScramAdmin)
- private val kafkaPassword = JaasTestUtils.KafkaScramAdminPassword
+ protected val kafkaPassword = JaasTestUtils.KafkaScramAdminPassword
- private val privilegedAdminClientConfig = new Properties()
+ protected val privilegedAdminClientConfig = new Properties()
this.serverConfig.setProperty(KafkaConfig.DelegationTokenSecretKeyProp, "testKey")
+ def createDelegationTokenOptions(): CreateDelegationTokenOptions = new CreateDelegationTokenOptions()
+
+ def configureTokenAclsBeforeServersStart(): Unit = { }
+
override def configureSecurityBeforeServersStart(): Unit = {
super.configureSecurityBeforeServersStart()
+ configureTokenAclsBeforeServersStart()
zkClient.makeSurePersistentPathExists(ConfigEntityChangeNotificationZNode.path)
// Create broker admin credentials before starting brokers
createScramCredentials(zkConnect, kafkaPrincipal.getName, kafkaPassword)
}
- override def createPrivilegedAdminClient() = createScramAdminClient(kafkaClientSaslMechanism, kafkaPrincipal.getName, kafkaPassword)
+ override def createPrivilegedAdminClient(): Admin = createScramAdminClient(kafkaClientSaslMechanism, kafkaPrincipal.getName, kafkaPassword)
+
+ def createAdditionalCredentialsAfterServersStarted(): Unit = {}
override def configureSecurityAfterServersStart(): Unit = {
super.configureSecurityAfterServersStart()
@@ -67,6 +74,8 @@ class DelegationTokenEndToEndAuthorizationTest extends EndToEndAuthorizationTest
createScramCredentialsViaPrivilegedAdminClient(clientPrincipal.getName, clientPassword)
waitForUserScramCredentialToAppearOnAllBrokers(clientPrincipal.getName, kafkaClientSaslMechanism)
+ createAdditionalCredentialsAfterServersStarted()
+
//create a token with "scram-user" credentials and a privileged token with scram-admin credentials
val tokens = createDelegationTokens()
val token = tokens._1
@@ -105,12 +114,36 @@ class DelegationTokenEndToEndAuthorizationTest extends EndToEndAuthorizationTest
privilegedAdminClientConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers())
}
- private def createDelegationTokens(): (DelegationToken, DelegationToken) = {
- val adminClient = createScramAdminClient(kafkaClientSaslMechanism, clientPrincipal.getName, clientPassword)
+ def assertTokenOwner(owner: KafkaPrincipal, token: DelegationToken): Unit = {
+ assertEquals(owner, token.tokenInfo().owner())
+ }
+
+ def assertTokenRequester(requester: KafkaPrincipal, token: DelegationToken): Unit = {
+ assertEquals(requester, token.tokenInfo().tokenRequester())
+ }
+
+ def assertToken(token: DelegationToken): Unit = {
+ assertTokenOwner(clientPrincipal, token)
+ assertTokenRequester(clientPrincipal, token)
+ }
+
+ def createTokenRequesterAdminClient(): Admin = {
+ createScramAdminClient(kafkaClientSaslMechanism, clientPrincipal.getName, clientPassword)
+ }
+
+ def createDelegationTokens(): (DelegationToken, DelegationToken) = {
+ createDelegationTokens(createDelegationTokenOptions)
+ }
+
+ def createDelegationTokens(createDelegationTokenOptions: () => CreateDelegationTokenOptions, assert: Boolean = true): (DelegationToken, DelegationToken) = {
+ val adminClient = createTokenRequesterAdminClient()
try {
val privilegedAdminClient = createScramAdminClient(kafkaClientSaslMechanism, kafkaPrincipal.getName, kafkaPassword)
try {
- val token = adminClient.createDelegationToken().delegationToken().get()
+ val token = adminClient.createDelegationToken(createDelegationTokenOptions()).delegationToken().get()
+ if (assert) {
+ assertToken(token)
+ }
val privilegedToken = privilegedAdminClient.createDelegationToken().delegationToken().get()
//wait for tokens to reach all the brokers
TestUtils.waitUntilTrue(() => servers.forall(server => server.tokenCache.tokens().size() == 2),
diff --git a/core/src/test/scala/integration/kafka/api/DelegationTokenEndToEndAuthorizationWithOwnerTest.scala b/core/src/test/scala/integration/kafka/api/DelegationTokenEndToEndAuthorizationWithOwnerTest.scala
new file mode 100644
index 0000000000..3c034fc4c1
--- /dev/null
+++ b/core/src/test/scala/integration/kafka/api/DelegationTokenEndToEndAuthorizationWithOwnerTest.scala
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.api
+
+import kafka.admin.AclCommand
+import kafka.utils.JaasTestUtils
+import org.apache.kafka.clients.admin.{Admin, CreateDelegationTokenOptions, DescribeDelegationTokenOptions}
+import org.apache.kafka.common.security.auth.KafkaPrincipal
+import org.apache.kafka.common.security.token.delegation.DelegationToken
+import org.junit.jupiter.api.Assertions.{assertThrows, assertTrue}
+import org.junit.jupiter.api.Test
+
+import java.util.Collections
+import scala.concurrent.ExecutionException
+import scala.jdk.CollectionConverters._
+
+class DelegationTokenEndToEndAuthorizationWithOwnerTest extends DelegationTokenEndToEndAuthorizationTest {
+
+ def createTokenForValidUserArgs: Array[String] = Array("--authorizer-properties",
+ s"zookeeper.connect=$zkConnect",
+ s"--add",
+ s"--user-principal=$clientPrincipal",
+ s"--operation=CreateTokens",
+ s"--allow-principal=$tokenRequesterPrincipal")
+
+ // tests the naive positive case for token requesting for others
+ def describeTokenForValidUserArgs: Array[String] = Array("--authorizer-properties",
+ s"zookeeper.connect=$zkConnect",
+ s"--add",
+ s"--user-principal=$clientPrincipal",
+ s"--operation=DescribeTokens",
+ s"--allow-principal=$tokenRequesterPrincipal")
+
+ // This permission is just there so that otherClientPrincipal shows up among the resources
+ def describeTokenForAdminArgs: Array[String] = Array("--authorizer-properties",
+ s"zookeeper.connect=$zkConnect",
+ s"--add",
+ s"--user-principal=$otherClientPrincipal",
+ s"--operation=DescribeTokens",
+ s"--allow-principal=$otherClientRequesterPrincipal")
+
+ override def createDelegationTokenOptions(): CreateDelegationTokenOptions = new CreateDelegationTokenOptions().owner(clientPrincipal)
+
+ private val tokenRequesterPrincipal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, JaasTestUtils.KafkaScramUser2)
+ private val tokenRequesterPassword = JaasTestUtils.KafkaScramPassword2
+
+ private val otherClientPrincipal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "other-client-principal")
+ private val otherClientPassword = "other-client-password"
+
+ private val otherClientRequesterPrincipal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "other-client-requester-principal")
+ private val otherClientRequesterPassword = "other-client-requester-password"
+
+ private val describeTokenFailPrincipal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "describe-token-fail-principal")
+ private val describeTokenFailPassword = "describe-token-fail-password"
+
+ override def configureTokenAclsBeforeServersStart(): Unit = {
+ super.configureTokenAclsBeforeServersStart()
+ AclCommand.main(createTokenForValidUserArgs)
+ AclCommand.main(describeTokenForValidUserArgs)
+ AclCommand.main(describeTokenForAdminArgs)
+ }
+
+ override def createAdditionalCredentialsAfterServersStarted(): Unit = {
+ super.createAdditionalCredentialsAfterServersStarted()
+ createScramCredentialsViaPrivilegedAdminClient(tokenRequesterPrincipal.getName, tokenRequesterPassword)
+ createScramCredentialsViaPrivilegedAdminClient(otherClientPrincipal.getName, otherClientPassword)
+ createScramCredentialsViaPrivilegedAdminClient(otherClientRequesterPrincipal.getName, otherClientRequesterPassword)
+ createScramCredentialsViaPrivilegedAdminClient(describeTokenFailPrincipal.getName, describeTokenFailPassword)
+ }
+
+ override def assertToken(token: DelegationToken): Unit = {
+ assertTokenOwner(clientPrincipal, token)
+ assertTokenRequester(tokenRequesterPrincipal, token)
+ }
+
+ override def createTokenRequesterAdminClient(): Admin = {
+ createScramAdminClient(kafkaClientSaslMechanism, tokenRequesterPrincipal.getName, tokenRequesterPassword)
+ }
+
+ @Test
+ def testCreateTokenForOtherUserFails(): Unit = {
+ val thrown = assertThrows(classOf[ExecutionException], () => {
+ createDelegationTokens(() => new CreateDelegationTokenOptions().owner(otherClientPrincipal), assert = false)
+ })
+ assertTrue(thrown.getMessage.contains("Delegation Token authorization failed"))
+ }
+
+ @Test
+ def testDescribeTokenForOtherUserFails(): Unit = {
+ val describeTokenFailAdminClient = createScramAdminClient(kafkaClientSaslMechanism, describeTokenFailPrincipal.getName, describeTokenFailPassword)
+ val otherClientAdminClient = createScramAdminClient(kafkaClientSaslMechanism, otherClientPrincipal.getName, otherClientPassword)
+ try {
+ otherClientAdminClient.createDelegationToken().delegationToken().get()
+ val tokens = describeTokenFailAdminClient.describeDelegationToken(
+ new DescribeDelegationTokenOptions().owners(Collections.singletonList(otherClientPrincipal)))
+ .delegationTokens.get.asScala
+ assertTrue(tokens.isEmpty)
+ } finally {
+ describeTokenFailAdminClient.close()
+ otherClientAdminClient.close()
+ }
+ }
+
+ @Test
+ def testDescribeTokenForOtherUserPasses(): Unit = {
+ val adminClient = createTokenRequesterAdminClient()
+ try {
+ val tokens = adminClient.describeDelegationToken(
+ new DescribeDelegationTokenOptions().owners(Collections.singletonList(clientPrincipal)))
+ .delegationTokens.get.asScala
+ assertTrue(tokens.nonEmpty)
+ tokens.foreach(t => {
+ assertTrue(t.tokenInfo.owner.equals(clientPrincipal))
+ assertTrue(t.tokenInfo.tokenRequester.equals(tokenRequesterPrincipal))
+ })
+ } finally {
+ adminClient.close()
+ }
+ }
+}
diff --git a/core/src/test/scala/unit/kafka/admin/AclCommandTest.scala b/core/src/test/scala/unit/kafka/admin/AclCommandTest.scala
index 7cd5a18c22..b5d9692040 100644
--- a/core/src/test/scala/unit/kafka/admin/AclCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/AclCommandTest.scala
@@ -54,13 +54,15 @@ class AclCommandTest extends QuorumTestHarness with Logging {
private val GroupResources = Set(new ResourcePattern(GROUP, "testGroup-1", LITERAL), new ResourcePattern(GROUP, "testGroup-2", LITERAL))
private val TransactionalIdResources = Set(new ResourcePattern(TRANSACTIONAL_ID, "t0", LITERAL), new ResourcePattern(TRANSACTIONAL_ID, "t1", LITERAL))
private val TokenResources = Set(new ResourcePattern(DELEGATION_TOKEN, "token1", LITERAL), new ResourcePattern(DELEGATION_TOKEN, "token2", LITERAL))
+ private val UserResources = Set(new ResourcePattern(USER, "User:test-user1", LITERAL), new ResourcePattern(USER, "User:test-user2", LITERAL))
private val ResourceToCommand = Map[Set[ResourcePattern], Array[String]](
TopicResources -> Array("--topic", "test-1", "--topic", "test-2"),
Set(ClusterResource) -> Array("--cluster"),
GroupResources -> Array("--group", "testGroup-1", "--group", "testGroup-2"),
TransactionalIdResources -> Array("--transactional-id", "t0", "--transactional-id", "t1"),
- TokenResources -> Array("--delegation-token", "token1", "--delegation-token", "token2")
+ TokenResources -> Array("--delegation-token", "token1", "--delegation-token", "token2"),
+ UserResources -> Array("--user-principal", "User:test-user1", "--user-principal", "User:test-user2")
)
private val ResourceToOperations = Map[Set[ResourcePattern], (Set[AclOperation], Array[String])](
@@ -72,7 +74,8 @@ class AclCommandTest extends QuorumTestHarness with Logging {
"--operation", "AlterConfigs", "--operation", "IdempotentWrite", "--operation", "Alter", "--operation", "Describe")),
GroupResources -> (Set(READ, DESCRIBE, DELETE), Array("--operation", "Read", "--operation", "Describe", "--operation", "Delete")),
TransactionalIdResources -> (Set(DESCRIBE, WRITE), Array("--operation", "Describe", "--operation", "Write")),
- TokenResources -> (Set(DESCRIBE), Array("--operation", "Describe"))
+ TokenResources -> (Set(DESCRIBE), Array("--operation", "Describe")),
+ UserResources -> (Set(CREATE_TOKENS, DESCRIBE_TOKENS), Array("--operation", "CreateTokens", "--operation", "DescribeTokens"))
)
private def ProducerResourceToAcls(enableIdempotence: Boolean = false) = Map[Set[ResourcePattern], Set[AccessControlEntry]](
diff --git a/core/src/test/scala/unit/kafka/security/authorizer/AclAuthorizerTest.scala b/core/src/test/scala/unit/kafka/security/authorizer/AclAuthorizerTest.scala
index 4b33d3e252..ce7bca25d1 100644
--- a/core/src/test/scala/unit/kafka/security/authorizer/AclAuthorizerTest.scala
+++ b/core/src/test/scala/unit/kafka/security/authorizer/AclAuthorizerTest.scala
@@ -486,9 +486,9 @@ class AclAuthorizerTest extends QuorumTestHarness with BaseAuthorizerTest {
@Test
def testAclInheritance(): Unit = {
testImplicationsOfAllow(AclOperation.ALL, Set(READ, WRITE, CREATE, DELETE, ALTER, DESCRIBE,
- CLUSTER_ACTION, DESCRIBE_CONFIGS, ALTER_CONFIGS, IDEMPOTENT_WRITE))
+ CLUSTER_ACTION, DESCRIBE_CONFIGS, ALTER_CONFIGS, IDEMPOTENT_WRITE, CREATE_TOKENS, DESCRIBE_TOKENS))
testImplicationsOfDeny(AclOperation.ALL, Set(READ, WRITE, CREATE, DELETE, ALTER, DESCRIBE,
- CLUSTER_ACTION, DESCRIBE_CONFIGS, ALTER_CONFIGS, IDEMPOTENT_WRITE))
+ CLUSTER_ACTION, DESCRIBE_CONFIGS, ALTER_CONFIGS, IDEMPOTENT_WRITE, CREATE_TOKENS, DESCRIBE_TOKENS))
testImplicationsOfAllow(READ, Set(DESCRIBE))
testImplicationsOfAllow(WRITE, Set(DESCRIBE))
testImplicationsOfAllow(DELETE, Set(DESCRIBE))
diff --git a/core/src/test/scala/unit/kafka/security/token/delegation/DelegationTokenManagerTest.scala b/core/src/test/scala/unit/kafka/security/token/delegation/DelegationTokenManagerTest.scala
index 523b6a74b0..0cd67edfd0 100644
--- a/core/src/test/scala/unit/kafka/security/token/delegation/DelegationTokenManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/security/token/delegation/DelegationTokenManagerTest.scala
@@ -20,7 +20,6 @@ package kafka.security.token.delegation
import java.net.InetAddress
import java.nio.ByteBuffer
import java.util.{Base64, Properties}
-
import kafka.network.RequestChannel.Session
import kafka.security.authorizer.{AclAuthorizer, AuthorizerUtils}
import kafka.security.authorizer.AclEntry.WildcardHost
@@ -33,7 +32,7 @@ import org.apache.kafka.common.acl.AclPermissionType.ALLOW
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.resource.PatternType.LITERAL
import org.apache.kafka.common.resource.ResourcePattern
-import org.apache.kafka.common.resource.ResourceType.DELEGATION_TOKEN
+import org.apache.kafka.common.resource.ResourceType.{DELEGATION_TOKEN, USER}
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.security.scram.internals.ScramMechanism
import org.apache.kafka.common.security.token.delegation.internals.DelegationTokenCache
@@ -84,7 +83,7 @@ class DelegationTokenManagerTest extends QuorumTestHarness {
val config = KafkaConfig.fromProps(props)
val tokenManager = createDelegationTokenManager(config, tokenCache, time, zkClient)
- tokenManager.createToken(owner, renewer, -1, createTokenResultCallBack)
+ tokenManager.createToken(owner, owner, renewer, -1, createTokenResultCallBack)
assertEquals(Errors.DELEGATION_TOKEN_AUTH_DISABLED, createTokenResult.error)
assert(Array[Byte]() sameElements createTokenResult.hmac)
@@ -101,11 +100,11 @@ class DelegationTokenManagerTest extends QuorumTestHarness {
val tokenManager = createDelegationTokenManager(config, tokenCache, time, zkClient)
tokenManager.startup()
- tokenManager.createToken(owner, renewer, -1 , createTokenResultCallBack)
+ tokenManager.createToken(owner, owner, renewer, -1 , createTokenResultCallBack)
val issueTime = time.milliseconds
val tokenId = createTokenResult.tokenId
val password = DelegationTokenManager.createHmac(tokenId, secretKey)
- assertEquals(CreateTokenResult(issueTime, issueTime + renewTimeMsDefault, issueTime + maxLifeTimeMsDefault, tokenId, password, Errors.NONE), createTokenResult)
+ assertEquals(CreateTokenResult(owner, owner, issueTime, issueTime + renewTimeMsDefault, issueTime + maxLifeTimeMsDefault, tokenId, password, Errors.NONE), createTokenResult)
val token = tokenManager.getToken(tokenId)
assertFalse(token.isEmpty )
@@ -118,12 +117,12 @@ class DelegationTokenManagerTest extends QuorumTestHarness {
val tokenManager = createDelegationTokenManager(config, tokenCache, time, zkClient)
tokenManager.startup()
- tokenManager.createToken(owner, renewer, -1 , createTokenResultCallBack)
+ tokenManager.createToken(owner, owner, renewer, -1 , createTokenResultCallBack)
val issueTime = time.milliseconds
val maxLifeTime = issueTime + maxLifeTimeMsDefault
val tokenId = createTokenResult.tokenId
val password = DelegationTokenManager.createHmac(tokenId, secretKey)
- assertEquals(CreateTokenResult(issueTime, issueTime + renewTimeMsDefault, maxLifeTime, tokenId, password, Errors.NONE), createTokenResult)
+ assertEquals(CreateTokenResult(owner, owner, issueTime, issueTime + renewTimeMsDefault, maxLifeTime, tokenId, password, Errors.NONE), createTokenResult)
//try renewing non-existing token
tokenManager.renewToken(owner, ByteBuffer.wrap("test".getBytes), -1 , renewResponseCallback)
@@ -166,11 +165,11 @@ class DelegationTokenManagerTest extends QuorumTestHarness {
val tokenManager = createDelegationTokenManager(config, tokenCache, time, zkClient)
tokenManager.startup()
- tokenManager.createToken(owner, renewer, -1 , createTokenResultCallBack)
+ tokenManager.createToken(owner, owner, renewer, -1 , createTokenResultCallBack)
val issueTime = time.milliseconds
val tokenId = createTokenResult.tokenId
val password = DelegationTokenManager.createHmac(tokenId, secretKey)
- assertEquals(CreateTokenResult(issueTime, issueTime + renewTimeMsDefault, issueTime + maxLifeTimeMsDefault, tokenId, password, Errors.NONE), createTokenResult)
+ assertEquals(CreateTokenResult(owner, owner, issueTime, issueTime + renewTimeMsDefault, issueTime + maxLifeTimeMsDefault, tokenId, password, Errors.NONE), createTokenResult)
//try expire non-existing token
tokenManager.expireToken(owner, ByteBuffer.wrap("test".getBytes), -1 , renewResponseCallback)
@@ -201,11 +200,11 @@ class DelegationTokenManagerTest extends QuorumTestHarness {
val tokenManager = createDelegationTokenManager(config, tokenCache, time, zkClient)
tokenManager.startup()
- tokenManager.createToken(owner, renewer, -1 , createTokenResultCallBack)
+ tokenManager.createToken(owner, owner, renewer, -1 , createTokenResultCallBack)
val issueTime = time.milliseconds
val tokenId = createTokenResult.tokenId
val password = DelegationTokenManager.createHmac(tokenId, secretKey)
- assertEquals(CreateTokenResult(issueTime, issueTime + renewTimeMsDefault, issueTime + maxLifeTimeMsDefault, tokenId, password, Errors.NONE), createTokenResult)
+ assertEquals(CreateTokenResult(owner, owner, issueTime, issueTime + renewTimeMsDefault, issueTime + maxLifeTimeMsDefault, tokenId, password, Errors.NONE), createTokenResult)
// expire the token immediately
tokenManager.expireToken(owner, ByteBuffer.wrap(password), -1, renewResponseCallback)
@@ -224,10 +223,13 @@ class DelegationTokenManagerTest extends QuorumTestHarness {
val config = KafkaConfig.fromProps(props)
+ val requester1 = SecurityUtils.parseKafkaPrincipal("User:requester1")
+
val owner1 = SecurityUtils.parseKafkaPrincipal("User:owner1")
val owner2 = SecurityUtils.parseKafkaPrincipal("User:owner2")
val owner3 = SecurityUtils.parseKafkaPrincipal("User:owner3")
val owner4 = SecurityUtils.parseKafkaPrincipal("User:owner4")
+ val owner5 = SecurityUtils.parseKafkaPrincipal("User:owner5")
val renewer1 = SecurityUtils.parseKafkaPrincipal("User:renewer1")
val renewer2 = SecurityUtils.parseKafkaPrincipal("User:renewer2")
@@ -243,41 +245,47 @@ class DelegationTokenManagerTest extends QuorumTestHarness {
tokenManager.startup()
//create tokens
- tokenManager.createToken(owner1, List(renewer1, renewer2), 1 * 60 * 60 * 1000L, createTokenResultCallBack)
+ tokenManager.createToken(owner1, owner1, List(renewer1, renewer2), 1 * 60 * 60 * 1000L, createTokenResultCallBack)
- tokenManager.createToken(owner2, List(renewer3), 1 * 60 * 60 * 1000L, createTokenResultCallBack)
+ tokenManager.createToken(owner2, owner2, List(renewer3), 1 * 60 * 60 * 1000L, createTokenResultCallBack)
val tokenId2 = createTokenResult.tokenId
- tokenManager.createToken(owner3, List(renewer4), 2 * 60 * 60 * 1000L, createTokenResultCallBack)
+ tokenManager.createToken(owner3, owner3, List(renewer4), 2 * 60 * 60 * 1000L, createTokenResultCallBack)
val tokenId3 = createTokenResult.tokenId
- tokenManager.createToken(owner4, List(owner1, renewer4), 2 * 60 * 60 * 1000L, createTokenResultCallBack)
+ tokenManager.createToken(owner4, owner4, List(owner1, renewer4), 2 * 60 * 60 * 1000L, createTokenResultCallBack)
- assert(tokenManager.getAllTokenInformation.size == 4 )
+ tokenManager.createToken(requester1, owner5, List(renewer1), 1 * 60 * 60 * 1000L, createTokenResultCallBack)
+
+ assertEquals(5, tokenManager.getAllTokenInformation.size)
//get tokens non-exiting owner
var tokens = getTokens(tokenManager, aclAuthorizer, hostSession, owner1, List(SecurityUtils.parseKafkaPrincipal("User:unknown")))
- assert(tokens.size == 0)
+ assertEquals(0, tokens.size)
//get all tokens for empty owner list
tokens = getTokens(tokenManager, aclAuthorizer, hostSession, owner1, List())
- assert(tokens.size == 0)
+ assertEquals(0, tokens.size)
//get all tokens for owner1
tokens = getTokens(tokenManager, aclAuthorizer, hostSession, owner1, List(owner1))
- assert(tokens.size == 2)
+ assertEquals(2, tokens.size)
//get all tokens for owner1
tokens = getTokens(tokenManager, aclAuthorizer, hostSession, owner1, null)
- assert(tokens.size == 2)
+ assertEquals(2, tokens.size)
//get all tokens for unknown owner
tokens = getTokens(tokenManager, aclAuthorizer, hostSession, SecurityUtils.parseKafkaPrincipal("User:unknown"), null)
- assert(tokens.size == 0)
+ assertEquals(0, tokens.size)
//get all tokens for multiple owners (owner1, renewer4) and without permission for renewer4
tokens = getTokens(tokenManager, aclAuthorizer, hostSession, owner1, List(owner1, renewer4))
- assert(tokens.size == 2)
+ assertEquals(2, tokens.size)
+
+ // get tokens for owner5 with requester1
+ tokens = getTokens(tokenManager, aclAuthorizer, hostSession, requester1, List(owner5))
+ assertEquals(1, tokens.size)
def createAcl(aclBinding: AclBinding): Unit = {
val result = aclAuthorizer.createAcls(null, List(aclBinding).asJava).get(0).toCompletableFuture.get
@@ -288,22 +296,22 @@ class DelegationTokenManagerTest extends QuorumTestHarness {
createAcl(new AclBinding(new ResourcePattern(DELEGATION_TOKEN, tokenId3, LITERAL),
new AccessControlEntry(owner1.toString, WildcardHost, DESCRIBE, ALLOW)))
tokens = getTokens(tokenManager, aclAuthorizer, hostSession, owner1, List(owner1, renewer4))
- assert(tokens.size == 3)
+ assertEquals(3, tokens.size)
//get all tokens for renewer4 which is a renewer principal for some tokens
tokens = getTokens(tokenManager, aclAuthorizer, hostSession, renewer4, List(renewer4))
- assert(tokens.size == 2)
+ assertEquals(2, tokens.size)
//get all tokens for multiple owners (renewer2, renewer3) which are token renewers principals and without permissions for renewer3
tokens = getTokens(tokenManager, aclAuthorizer, hostSession, renewer2, List(renewer2, renewer3))
- assert(tokens.size == 1)
+ assertEquals(1, tokens.size)
//get all tokens for multiple owners (renewer2, renewer3) which are token renewers principals and with permissions
hostSession = Session(renewer2, InetAddress.getByName("192.168.1.1"))
createAcl(new AclBinding(new ResourcePattern(DELEGATION_TOKEN, tokenId2, LITERAL),
new AccessControlEntry(renewer2.toString, WildcardHost, DESCRIBE, ALLOW)))
tokens = getTokens(tokenManager, aclAuthorizer, hostSession, renewer2, List(renewer2, renewer3))
- assert(tokens.size == 2)
+ assertEquals(2, tokens.size)
aclAuthorizer.close()
}
@@ -315,13 +323,20 @@ class DelegationTokenManagerTest extends QuorumTestHarness {
List()
}
else {
- def authorizeToken(tokenId: String) = {
+ def authorizeToken(tokenId: String): Boolean = {
val requestContext = AuthorizerUtils.sessionToRequestContext(hostSession)
val action = new Action(AclOperation.DESCRIBE,
new ResourcePattern(DELEGATION_TOKEN, tokenId, LITERAL), 1, true, true)
aclAuthorizer.authorize(requestContext, List(action).asJava).asScala.head == AuthorizationResult.ALLOWED
}
- def eligible(token: TokenInformation) = DelegationTokenManager.filterToken(requestPrincipal, Option(requestedOwners), token, authorizeToken)
+ def authorizeRequester(owner: KafkaPrincipal): Boolean = {
+ val requestContext = AuthorizerUtils.sessionToRequestContext(hostSession)
+ val action = new Action(AclOperation.DESCRIBE_TOKENS,
+ new ResourcePattern(USER, owner.toString, LITERAL), 1, true, true)
+ aclAuthorizer.authorize(requestContext, List(action).asJava).asScala.head == AuthorizationResult.ALLOWED
+ }
+ def eligible(token: TokenInformation) = DelegationTokenManager
+ .filterToken(requestPrincipal, Option(requestedOwners), token, authorizeToken, authorizeRequester)
tokenManager.getTokens(eligible)
}
}
@@ -333,10 +348,10 @@ class DelegationTokenManagerTest extends QuorumTestHarness {
tokenManager.startup()
//create tokens
- tokenManager.createToken(owner, renewer, 1 * 60 * 60 * 1000L, createTokenResultCallBack)
- tokenManager.createToken(owner, renewer, 1 * 60 * 60 * 1000L, createTokenResultCallBack)
- tokenManager.createToken(owner, renewer, 2 * 60 * 60 * 1000L, createTokenResultCallBack)
- tokenManager.createToken(owner, renewer, 2 * 60 * 60 * 1000L, createTokenResultCallBack)
+ tokenManager.createToken(owner, owner, renewer, 1 * 60 * 60 * 1000L, createTokenResultCallBack)
+ tokenManager.createToken(owner, owner, renewer, 1 * 60 * 60 * 1000L, createTokenResultCallBack)
+ tokenManager.createToken(owner, owner, renewer, 2 * 60 * 60 * 1000L, createTokenResultCallBack)
+ tokenManager.createToken(owner, owner, renewer, 2 * 60 * 60 * 1000L, createTokenResultCallBack)
assert(tokenManager.getAllTokenInformation.size == 4 )
time.sleep(2 * 60 * 60 * 1000L)