From 07a442fc1f7749aa7887f3a4caddaa8ef885df8b Mon Sep 17 00:00:00 2001 From: Manikumar Reddy Date: Thu, 29 Mar 2018 20:55:06 +0530 Subject: [PATCH 1/4] KAFKA-6447: Add Delegation Token Operations to KafkaAdminClient (KIP-249) --- checkstyle/suppressions.xml | 2 +- .../kafka/clients/admin/AdminClient.java | 154 ++++++++++++++++++ .../admin/CreateDelegationTokenOptions.java | 53 ++++++ .../admin/CreateDelegationTokenResult.java | 43 +++++ .../admin/DescribeDelegationTokenOptions.java | 42 +++++ .../admin/DescribeDelegationTokenResult.java | 45 +++++ .../admin/ExpireDelegationTokenOptions.java | 39 +++++ .../admin/ExpireDelegationTokenResult.java | 42 +++++ .../kafka/clients/admin/KafkaAdminClient.java | 137 ++++++++++++++++ .../admin/RenewDelegationTokenOptions.java | 39 +++++ .../admin/RenewDelegationTokenResult.java | 42 +++++ .../DescribeDelegationTokenResponse.java | 4 + .../ExpireDelegationTokenRequest.java | 4 +- .../ExpireDelegationTokenResponse.java | 4 + .../requests/RenewDelegationTokenRequest.java | 4 +- .../RenewDelegationTokenResponse.java | 4 + .../token/delegation/DelegationToken.java | 5 - .../kafka/clients/admin/MockAdminClient.java | 20 +++ .../common/requests/RequestResponseTest.java | 4 +- .../main/scala/kafka/admin/AdminClient.scala | 4 +- .../kafka/admin/DelegationTokenCommand.scala | 60 +++---- ...gationTokenEndToEndAuthorizationTest.scala | 8 +- ...legationTokenRequestsOnPlainTextTest.scala | 27 ++- .../server/DelegationTokenRequestsTest.scala | 97 ++++++----- ...nRequestsWithDisableTokenFeatureTest.scala | 32 ++-- .../unit/kafka/server/RequestQuotaTest.scala | 4 +- 26 files changed, 791 insertions(+), 128 deletions(-) create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenOptions.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenResult.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenOptions.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenResult.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenResult.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/RenewDelegationTokenOptions.java create mode 100644 clients/src/main/java/org/apache/kafka/clients/admin/RenewDelegationTokenResult.java diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index 0fec810a95bb8..2767132886dfe 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -10,7 +10,7 @@ + files="(Fetcher|Sender|SenderTest|ConsumerCoordinator|KafkaConsumer|KafkaProducer|Utils|TransactionManagerTest|KafkaAdminClient|NetworkClient|AdminClient).java"/> re */ public abstract DeleteRecordsResult deleteRecords(Map recordsToDelete, DeleteRecordsOptions options); + + /** + *

Create a Delegation Token.

+ * + *

This is a convenience method for {@link #createDelegationToken(CreateDelegationTokenOptions)} with default options. + * See the overload for more details.

+ * + * @return The CreateDelegationTokenResult. + */ + public CreateDelegationTokenResult createDelegationToken() { + return createDelegationToken(new CreateDelegationTokenOptions()); + } + + + /** + *

Create a Delegation Token.

+ * + *

This operation is supported by brokers with version 1.1.0 or higher.

+ * + *

The following exceptions can be anticipated when calling {@code get()} on the futures obtained from the + * {@link CreateDelegationTokenResult#delegationToken() delegationToken()} method of the returned {@code CreateDelegationTokenResult}

+ *
    + *
  • {@link org.apache.kafka.common.errors.UnsupportedByAuthenticationException} + * If the request sent on PLAINTEXT/1-way SSL channels or delegation token authenticated channels.
    • + *
  • {@link org.apache.kafka.common.errors.InvalidPrincipalTypeException} + * if the renewers principal type is not supported.
    • + *
  • {@link org.apache.kafka.common.errors.DelegationTokenDisabledException} + * if the delegation token feature is disabled.
    • + *
  • {@link org.apache.kafka.common.errors.TimeoutException} + * if the request was not completed in within the given {@link CreateDelegationTokenOptions#timeoutMs()}.
    • + *
+ * + * @param options The options to use when creating delegation token. + * @return The DeleteRecordsResult. + */ + public abstract CreateDelegationTokenResult createDelegationToken(CreateDelegationTokenOptions options); + + + /** + *

Renew a Delegation Token.

+ * + *

This is a convenience method for {@link #renewDelegationToken(byte[], RenewDelegationTokenOptions)} with default options. + * See the overload for more details.

+ * + * + * @param hmac HMAC of the Delegation token + * @return The RenewDelegationTokenResult. + */ + public RenewDelegationTokenResult renewDelegationToken(byte[] hmac) { + return renewDelegationToken(hmac, new RenewDelegationTokenOptions()); + } + + /** + *

Renew a Delegation Token.

+ * + *

This operation is supported by brokers with version 1.1.0 or higher.

+ * + *

The following exceptions can be anticipated when calling {@code get()} on the futures obtained from the + * {@link RenewDelegationTokenResult#expiryTimestamp() expiryTimestamp()} method of the returned {@code RenewDelegationTokenResult}

+ *
    + *
  • {@link org.apache.kafka.common.errors.UnsupportedByAuthenticationException} + * If the request sent on PLAINTEXT/1-way SSL channels or delegation token authenticated channels.
    • + *
  • {@link org.apache.kafka.common.errors.DelegationTokenDisabledException} + * if the delegation token feature is disabled.
    • + *
  • {@link org.apache.kafka.common.errors.DelegationTokenNotFoundException} + * if the delegation token is not found on server.
    • + *
  • {@link org.apache.kafka.common.errors.DelegationTokenOwnerMismatchException} + * if the authenticated user is not owner/renewer of the token.
    • + *
  • {@link org.apache.kafka.common.errors.DelegationTokenExpiredException} + * if the delegation token is expired.
    • + *
  • {@link org.apache.kafka.common.errors.TimeoutException} + * if the request was not completed in within the given {@link RenewDelegationTokenOptions#timeoutMs()}.
    • + *
+ * + * @param hmac HMAC of the Delegation token + * @param options The options to use when renewing delegation token. + * @return The RenewDelegationTokenResult. + */ + public abstract RenewDelegationTokenResult renewDelegationToken(byte[] hmac, RenewDelegationTokenOptions options); + + /** + *

Expire a Delegation Token.

+ * + *

This is a convenience method for {@link #expireDelegationToken(byte[], ExpireDelegationTokenOptions)} with default options. + * This will expire the token immediately. See the overload for more details.

+ * + * @param hmac HMAC of the Delegation token + * @return The ExpireDelegationTokenResult. + */ + public ExpireDelegationTokenResult expireDelegationToken(byte[] hmac) { + return expireDelegationToken(hmac, new ExpireDelegationTokenOptions()); + } + + /** + *

Expire a Delegation Token.

+ * + *

This operation is supported by brokers with version 1.1.0 or higher.

+ * + *

The following exceptions can be anticipated when calling {@code get()} on the futures obtained from the + * {@link ExpireDelegationTokenResult#expiryTimestamp() expiryTimestamp()} method of the returned {@code ExpireDelegationTokenResult}

+ *
    + *
  • {@link org.apache.kafka.common.errors.UnsupportedByAuthenticationException} + * If the request sent on PLAINTEXT/1-way SSL channels or delegation token authenticated channels.
    • + *
  • {@link org.apache.kafka.common.errors.DelegationTokenDisabledException} + * if the delegation token feature is disabled.
    • + *
  • {@link org.apache.kafka.common.errors.DelegationTokenNotFoundException} + * if the delegation token is not found on server.
    • + *
  • {@link org.apache.kafka.common.errors.DelegationTokenOwnerMismatchException} + * if the authenticated user is not owner/renewer of the requested token.
    • + *
  • {@link org.apache.kafka.common.errors.DelegationTokenExpiredException} + * if the delegation token is expired.
    • + *
  • {@link org.apache.kafka.common.errors.TimeoutException} + * if the request was not completed in within the given {@link ExpireDelegationTokenOptions#timeoutMs()}.
    • + *
+ * + * @param hmac HMAC of the Delegation token + * @param options The options to use when expiring delegation token. + * @return The ExpireDelegationTokenResult. + */ + public abstract ExpireDelegationTokenResult expireDelegationToken(byte[] hmac, ExpireDelegationTokenOptions options); + + /** + *

Describe the Delegation Tokens.

+ * + *

This is a convenience method for {@link #describeDelegationToken(DescribeDelegationTokenOptions)} with default options. + * This will return all the user owned tokens and tokens where user have Describe permission. See the overload for more details.

+ * + * @return The DescribeDelegationTokenResult. + */ + public DescribeDelegationTokenResult describeDelegationToken() { + return describeDelegationToken(new DescribeDelegationTokenOptions()); + } + + /** + *

Describe the Delegation Tokens.

+ * + *

This operation is supported by brokers with version 1.1.0 or higher.

+ * + *

The following exceptions can be anticipated when calling {@code get()} on the futures obtained from the + * {@link DescribeDelegationTokenResult#delegationTokens() delegationTokens()} method of the returned {@code DescribeDelegationTokenResult}

+ *
    + *
  • {@link org.apache.kafka.common.errors.UnsupportedByAuthenticationException} + * If the request sent on PLAINTEXT/1-way SSL channels or delegation token authenticated channels.
    • + *
  • {@link org.apache.kafka.common.errors.DelegationTokenDisabledException} + * if the delegation token feature is disabled.
    • + *
  • {@link org.apache.kafka.common.errors.TimeoutException} + * if the request was not completed in within the given {@link DescribeDelegationTokenOptions#timeoutMs()}.
    • + *
+ * + * @param options The options to use when describing delegation tokens. + * @return The DescribeDelegationTokenResult. + */ + public abstract DescribeDelegationTokenResult describeDelegationToken(DescribeDelegationTokenOptions options); + } 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 new file mode 100644 index 0000000000000..1b77b943800ec --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenOptions.java @@ -0,0 +1,53 @@ +/* + * 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 org.apache.kafka.clients.admin; + +import java.util.LinkedList; +import java.util.List; + +import org.apache.kafka.common.annotation.InterfaceStability; +import org.apache.kafka.common.security.auth.KafkaPrincipal; + +/** + * Options for {@link AdminClient#createDelegationToken(CreateDelegationTokenOptions)}. + * + * The API of this class is evolving, see {@link AdminClient} for details. + */ +@InterfaceStability.Evolving +public class CreateDelegationTokenOptions extends AbstractOptions { + private long maxLifeTimeMs = -1; + private List renewers = new LinkedList<>(); + + public CreateDelegationTokenOptions renewers(List renewers) { + this.renewers = renewers; + return this; + } + + public List renewers() { + return renewers; + } + + public CreateDelegationTokenOptions maxlifeTimeMs(long maxLifeTimeMs) { + this.maxLifeTimeMs = maxLifeTimeMs; + return this; + } + + public long maxlifeTimeMs() { + return maxLifeTimeMs; + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenResult.java new file mode 100644 index 0000000000000..043cbe87fef5c --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/admin/CreateDelegationTokenResult.java @@ -0,0 +1,43 @@ +/* + * 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 org.apache.kafka.clients.admin; + +import org.apache.kafka.common.KafkaFuture; +import org.apache.kafka.common.annotation.InterfaceStability; +import org.apache.kafka.common.security.token.delegation.DelegationToken; + +/** + * The result of the {@link KafkaAdminClient#createDelegationToken(CreateDelegationTokenOptions)} call. + * + * The API of this class is evolving, see {@link AdminClient} for details. + */ +@InterfaceStability.Evolving +public class CreateDelegationTokenResult { + private final KafkaFuture delegationToken; + + CreateDelegationTokenResult(KafkaFuture delegationToken) { + this.delegationToken = delegationToken; + } + + /** + * Returns a future which yields a delegation token + */ + public KafkaFuture delegationToken() { + return delegationToken; + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenOptions.java new file mode 100644 index 0000000000000..8fd98621b3776 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenOptions.java @@ -0,0 +1,42 @@ +/* + * 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 org.apache.kafka.clients.admin; + +import java.util.List; + +import org.apache.kafka.common.annotation.InterfaceStability; +import org.apache.kafka.common.security.auth.KafkaPrincipal; + +/** + * Options for {@link AdminClient#describeDelegationToken(DescribeDelegationTokenOptions)}. + * + * The API of this class is evolving, see {@link AdminClient} for details. + */ +@InterfaceStability.Evolving +public class DescribeDelegationTokenOptions extends AbstractOptions { + private List owners; + + public DescribeDelegationTokenOptions owners(List owners) { + this.owners = owners; + return this; + } + + public List owners() { + return owners; + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenResult.java new file mode 100644 index 0000000000000..7a9d4b9dd9737 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenResult.java @@ -0,0 +1,45 @@ +/* + * 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 org.apache.kafka.clients.admin; + +import java.util.List; + +import org.apache.kafka.common.KafkaFuture; +import org.apache.kafka.common.annotation.InterfaceStability; +import org.apache.kafka.common.security.token.delegation.DelegationToken; + +/** + * The result of the {@link KafkaAdminClient#describeDelegationToken(DescribeDelegationTokenOptions)} call. + * + * The API of this class is evolving, see {@link AdminClient} for details. + */ +@InterfaceStability.Evolving +public class DescribeDelegationTokenResult { + private final KafkaFuture> delegationTokens; + + DescribeDelegationTokenResult(KafkaFuture> delegationTokens) { + this.delegationTokens = delegationTokens; + } + + /** + * Returns a future which yields list of delegation tokens + */ + public KafkaFuture> delegationTokens() { + return delegationTokens; + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java new file mode 100644 index 0000000000000..971e4c3074bb0 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java @@ -0,0 +1,39 @@ +/* + * 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 org.apache.kafka.clients.admin; + +import org.apache.kafka.common.annotation.InterfaceStability; + +/** + * Options for {@link AdminClient#expireDelegationToken(byte[], ExpireDelegationTokenOptions)}. + * + * The API of this class is evolving, see {@link AdminClient} for details. + */ +@InterfaceStability.Evolving +public class ExpireDelegationTokenOptions extends AbstractOptions { + private long expiryTimePeriodMs = -1; + + public ExpireDelegationTokenOptions expiryTimePeriodMs(long expiryTimePeriodMs) { + this.expiryTimePeriodMs = expiryTimePeriodMs; + return this; + } + + public long expiryTimePeriodMs() { + return expiryTimePeriodMs; + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenResult.java new file mode 100644 index 0000000000000..41782bdcb5ce7 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenResult.java @@ -0,0 +1,42 @@ +/* + * 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 org.apache.kafka.clients.admin; + +import org.apache.kafka.common.KafkaFuture; +import org.apache.kafka.common.annotation.InterfaceStability; + +/** + * The result of the {@link KafkaAdminClient#expireDelegationToken(byte[], ExpireDelegationTokenOptions)} call. + * + * The API of this class is evolving, see {@link AdminClient} for details. + */ +@InterfaceStability.Evolving +public class ExpireDelegationTokenResult { + private final KafkaFuture expiryTimestamp; + + ExpireDelegationTokenResult(KafkaFuture expiryTimestamp) { + this.expiryTimestamp = expiryTimestamp; + } + + /** + * Returns a future which yields expiry timestamp + */ + public KafkaFuture expiryTimestamp() { + return expiryTimestamp; + } +} 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 511895354abef..3ac0e285622d9 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 @@ -69,6 +69,8 @@ import org.apache.kafka.common.requests.CreateAclsRequest.AclCreation; import org.apache.kafka.common.requests.CreateAclsResponse; import org.apache.kafka.common.requests.CreateAclsResponse.AclCreationResponse; +import org.apache.kafka.common.requests.CreateDelegationTokenRequest; +import org.apache.kafka.common.requests.CreateDelegationTokenResponse; import org.apache.kafka.common.requests.CreatePartitionsRequest; import org.apache.kafka.common.requests.CreatePartitionsResponse; import org.apache.kafka.common.requests.CreateTopicsRequest; @@ -85,12 +87,20 @@ import org.apache.kafka.common.requests.DescribeAclsResponse; import org.apache.kafka.common.requests.DescribeConfigsRequest; import org.apache.kafka.common.requests.DescribeConfigsResponse; +import org.apache.kafka.common.requests.DescribeDelegationTokenRequest; +import org.apache.kafka.common.requests.DescribeDelegationTokenResponse; import org.apache.kafka.common.requests.DescribeLogDirsRequest; import org.apache.kafka.common.requests.DescribeLogDirsResponse; +import org.apache.kafka.common.requests.ExpireDelegationTokenRequest; +import org.apache.kafka.common.requests.ExpireDelegationTokenResponse; import org.apache.kafka.common.requests.MetadataRequest; import org.apache.kafka.common.requests.MetadataResponse; +import org.apache.kafka.common.requests.RenewDelegationTokenRequest; +import org.apache.kafka.common.requests.RenewDelegationTokenResponse; import org.apache.kafka.common.requests.Resource; import org.apache.kafka.common.requests.ResourceType; +import org.apache.kafka.common.security.token.delegation.DelegationToken; +import org.apache.kafka.common.security.token.delegation.TokenInformation; import org.apache.kafka.common.utils.AppInfoParser; import org.apache.kafka.common.utils.KafkaThread; import org.apache.kafka.common.utils.LogContext; @@ -2072,4 +2082,131 @@ void handleFailure(Throwable throwable) { return new DeleteRecordsResult(new HashMap>(futures)); } + + @Override + public CreateDelegationTokenResult createDelegationToken(final CreateDelegationTokenOptions options) { + final KafkaFutureImpl delegationTokenFuture = new KafkaFutureImpl<>(); + final long now = time.milliseconds(); + runnable.call(new Call("createDelegationToken", calcDeadlineMs(now, options.timeoutMs()), + new LeastLoadedNodeProvider()) { + + @Override + AbstractRequest.Builder createRequest(int timeoutMs) { + return new CreateDelegationTokenRequest.Builder(options.renewers(), options.maxlifeTimeMs()); + } + + @Override + void handleResponse(AbstractResponse abstractResponse) { + CreateDelegationTokenResponse response = (CreateDelegationTokenResponse) abstractResponse; + if (response.hasError()) { + delegationTokenFuture.completeExceptionally(response.error().exception()); + } else { + TokenInformation tokenInfo = new TokenInformation(response.tokenId(), response.owner(), + options.renewers(), response.issueTimestamp(), response.maxTimestamp(), response.expiryTimestamp()); + DelegationToken token = new DelegationToken(tokenInfo, response.hmacBytes()); + delegationTokenFuture.complete(token); + } + } + + @Override + void handleFailure(Throwable throwable) { + delegationTokenFuture.completeExceptionally(throwable); + } + }, now); + + return new CreateDelegationTokenResult(delegationTokenFuture); + } + + @Override + public RenewDelegationTokenResult renewDelegationToken(final byte[] hmac, final RenewDelegationTokenOptions options) { + final KafkaFutureImpl expiryTimeFuture = new KafkaFutureImpl<>(); + final long now = time.milliseconds(); + runnable.call(new Call("renewDelegationToken", calcDeadlineMs(now, options.timeoutMs()), + new LeastLoadedNodeProvider()) { + + @Override + AbstractRequest.Builder createRequest(int timeoutMs) { + return new RenewDelegationTokenRequest.Builder(hmac, options.renewTimePeriodMs()); + } + + @Override + void handleResponse(AbstractResponse abstractResponse) { + RenewDelegationTokenResponse response = (RenewDelegationTokenResponse) abstractResponse; + if (response.hasError()) { + expiryTimeFuture.completeExceptionally(response.error().exception()); + } else { + expiryTimeFuture.complete(response.expiryTimestamp()); + } + } + + @Override + void handleFailure(Throwable throwable) { + expiryTimeFuture.completeExceptionally(throwable); + } + }, now); + + return new RenewDelegationTokenResult(expiryTimeFuture); + } + + @Override + public ExpireDelegationTokenResult expireDelegationToken(final byte[] hmac, final ExpireDelegationTokenOptions options) { + final KafkaFutureImpl expiryTimeFuture = new KafkaFutureImpl<>(); + final long now = time.milliseconds(); + runnable.call(new Call("expireDelegationToken", calcDeadlineMs(now, options.timeoutMs()), + new LeastLoadedNodeProvider()) { + + @Override + AbstractRequest.Builder createRequest(int timeoutMs) { + return new ExpireDelegationTokenRequest.Builder(hmac, options.expiryTimePeriodMs()); + } + + @Override + void handleResponse(AbstractResponse abstractResponse) { + ExpireDelegationTokenResponse response = (ExpireDelegationTokenResponse) abstractResponse; + if (response.hasError()) { + expiryTimeFuture.completeExceptionally(response.error().exception()); + } else { + expiryTimeFuture.complete(response.expiryTimestamp()); + } + } + + @Override + void handleFailure(Throwable throwable) { + expiryTimeFuture.completeExceptionally(throwable); + } + }, now); + + return new ExpireDelegationTokenResult(expiryTimeFuture); + } + + @Override + public DescribeDelegationTokenResult describeDelegationToken(final DescribeDelegationTokenOptions options) { + final KafkaFutureImpl> tokensFuture = new KafkaFutureImpl<>(); + final long now = time.milliseconds(); + runnable.call(new Call("describeDelegationToken", calcDeadlineMs(now, options.timeoutMs()), + new LeastLoadedNodeProvider()) { + + @Override + AbstractRequest.Builder createRequest(int timeoutMs) { + return new DescribeDelegationTokenRequest.Builder(options.owners()); + } + + @Override + void handleResponse(AbstractResponse abstractResponse) { + DescribeDelegationTokenResponse response = (DescribeDelegationTokenResponse) abstractResponse; + if (response.hasError()) { + tokensFuture.completeExceptionally(response.error().exception()); + } else { + tokensFuture.complete(response.tokens()); + } + } + + @Override + void handleFailure(Throwable throwable) { + tokensFuture.completeExceptionally(throwable); + } + }, now); + + return new DescribeDelegationTokenResult(tokensFuture); + } } diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/RenewDelegationTokenOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/RenewDelegationTokenOptions.java new file mode 100644 index 0000000000000..238dc4a3494a8 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/admin/RenewDelegationTokenOptions.java @@ -0,0 +1,39 @@ +/* + * 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 org.apache.kafka.clients.admin; + +import org.apache.kafka.common.annotation.InterfaceStability; + +/** + * Options for {@link AdminClient#renewDelegationToken(byte[], RenewDelegationTokenOptions)}. + * + * The API of this class is evolving, see {@link AdminClient} for details. + */ +@InterfaceStability.Evolving +public class RenewDelegationTokenOptions extends AbstractOptions { + private long renewTimePeriodMs = -1; + + public RenewDelegationTokenOptions renewTimePeriodMs(long renewTimePeriodMs) { + this.renewTimePeriodMs = renewTimePeriodMs; + return this; + } + + public long renewTimePeriodMs() { + return renewTimePeriodMs; + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/RenewDelegationTokenResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/RenewDelegationTokenResult.java new file mode 100644 index 0000000000000..38cdf1ae1b241 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/admin/RenewDelegationTokenResult.java @@ -0,0 +1,42 @@ +/* + * 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 org.apache.kafka.clients.admin; + +import org.apache.kafka.common.KafkaFuture; +import org.apache.kafka.common.annotation.InterfaceStability; + +/** + * The result of the {@link KafkaAdminClient#expireDelegationToken(byte[], ExpireDelegationTokenOptions)} call. + * + * The API of this class is evolving, see {@link AdminClient} for details. + */ +@InterfaceStability.Evolving +public class RenewDelegationTokenResult { + private final KafkaFuture expiryTimestamp; + + RenewDelegationTokenResult(KafkaFuture expiryTimestamp) { + this.expiryTimestamp = expiryTimestamp; + } + + /** + * Returns a future which yields expiry timestamp + */ + public KafkaFuture expiryTimestamp() { + return expiryTimestamp; + } +} 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 dba29eafe9937..7ba270a61530b 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 @@ -184,4 +184,8 @@ public Errors error() { public List tokens() { return tokens; } + + public boolean hasError() { + return this.error != Errors.NONE; + } } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ExpireDelegationTokenRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ExpireDelegationTokenRequest.java index 0d43440d32986..40f0aadc0bb85 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/ExpireDelegationTokenRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/ExpireDelegationTokenRequest.java @@ -88,9 +88,9 @@ public static class Builder extends AbstractRequest.Builder re } } + @Override + public CreateDelegationTokenResult createDelegationToken(CreateDelegationTokenOptions options) { + throw new UnsupportedOperationException("Not implemented yet"); + } + + @Override + public RenewDelegationTokenResult renewDelegationToken(byte[] hmac, RenewDelegationTokenOptions options) { + throw new UnsupportedOperationException("Not implemented yet"); + } + + @Override + public ExpireDelegationTokenResult expireDelegationToken(byte[] hmac, ExpireDelegationTokenOptions options) { + throw new UnsupportedOperationException("Not implemented yet"); + } + + @Override + public DescribeDelegationTokenResult describeDelegationToken(DescribeDelegationTokenOptions options) { + throw new UnsupportedOperationException("Not implemented yet"); + } + @Override public CreateAclsResult createAcls(Collection acls, CreateAclsOptions options) { throw new UnsupportedOperationException("Not implemented yet"); 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 bdbd10626856a..c63cecdda2813 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 @@ -1223,7 +1223,7 @@ private CreateDelegationTokenResponse createCreateTokenResponse() { } private RenewDelegationTokenRequest createRenewTokenRequest() { - return new RenewDelegationTokenRequest.Builder(ByteBuffer.wrap("test".getBytes()), System.currentTimeMillis()).build(); + return new RenewDelegationTokenRequest.Builder("test".getBytes(), System.currentTimeMillis()).build(); } private RenewDelegationTokenResponse createRenewTokenResponse() { @@ -1231,7 +1231,7 @@ private RenewDelegationTokenResponse createRenewTokenResponse() { } private ExpireDelegationTokenRequest createExpireTokenRequest() { - return new ExpireDelegationTokenRequest.Builder(ByteBuffer.wrap("test".getBytes()), System.currentTimeMillis()).build(); + return new ExpireDelegationTokenRequest.Builder("test".getBytes(), System.currentTimeMillis()).build(); } private ExpireDelegationTokenResponse createExpireTokenResponse() { diff --git a/core/src/main/scala/kafka/admin/AdminClient.scala b/core/src/main/scala/kafka/admin/AdminClient.scala index c010ba0a4ac78..50d66192ae144 100644 --- a/core/src/main/scala/kafka/admin/AdminClient.scala +++ b/core/src/main/scala/kafka/admin/AdminClient.scala @@ -351,13 +351,13 @@ class AdminClient(val time: Time, } def renewToken(hmac: ByteBuffer, renewTimePeriod: Long = -1): (Errors, Long) = { - val responseBody = sendAnyNode(ApiKeys.RENEW_DELEGATION_TOKEN, new RenewDelegationTokenRequest.Builder(hmac, renewTimePeriod)) + val responseBody = sendAnyNode(ApiKeys.RENEW_DELEGATION_TOKEN, new RenewDelegationTokenRequest.Builder(new Array[Byte](hmac.remaining), renewTimePeriod)) val response = responseBody.asInstanceOf[RenewDelegationTokenResponse] (response.error, response.expiryTimestamp) } def expireToken(hmac: ByteBuffer, expiryTimeStamp: Long = -1): (Errors, Long) = { - val responseBody = sendAnyNode(ApiKeys.EXPIRE_DELEGATION_TOKEN, new ExpireDelegationTokenRequest.Builder(hmac, expiryTimeStamp)) + val responseBody = sendAnyNode(ApiKeys.EXPIRE_DELEGATION_TOKEN, new ExpireDelegationTokenRequest.Builder(new Array[Byte](hmac.remaining), expiryTimeStamp)) val response = responseBody.asInstanceOf[ExpireDelegationTokenResponse] (response.error, response.expiryTimestamp) } diff --git a/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala b/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala index 6c5d1ce925df5..92a39473a28cf 100644 --- a/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala +++ b/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala @@ -17,12 +17,12 @@ package kafka.admin -import java.nio.ByteBuffer +import java.text.SimpleDateFormat import joptsimple._ import kafka.utils.{CommandLineUtils, Exit, Logging} import org.apache.kafka.clients.CommonClientConfigs -import org.apache.kafka.common.protocol.Errors +import org.apache.kafka.clients.admin.{CreateDelegationTokenOptions, DescribeDelegationTokenOptions, ExpireDelegationTokenOptions, RenewDelegationTokenOptions, AdminClient => JAdminClient} import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.kafka.common.security.token.delegation.DelegationToken import org.apache.kafka.common.utils.{Base64, SecurityUtils, Utils} @@ -71,19 +71,19 @@ object DelegationTokenCommand extends Logging { } } - def createToken(adminClient: AdminClient, opts: DelegationTokenCommandOptions) = { - val renewerPrincipals = getPrincipals(opts, opts.renewPrincipalsOpt) + def createToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions) = { + val renewerPrincipals = getPrincipals(opts, opts.renewPrincipalsOpt).asJava val maxLifeTimeMs = opts.options.valueOf(opts.maxLifeTimeOpt).longValue println("Calling create token operation with renewers :" + renewerPrincipals +" , max-life-time-period :"+ maxLifeTimeMs) - val response = adminClient.createToken(renewerPrincipals, maxLifeTimeMs) - response match { - case (Errors.NONE, token) => println("Created delegation token with tokenId : %s".format(token.tokenInfo.tokenId)); printToken(List(token)) - case (e, _) => throw new AdminOperationException(e.message) - } + val createDelegationTokenOptions = new CreateDelegationTokenOptions().maxlifeTimeMs(maxLifeTimeMs).renewers(renewerPrincipals) + val createResult = adminClient.createDelegationToken(createDelegationTokenOptions) + val token = createResult.delegationToken().get() + println("Created delegation token with tokenId : %s".format(token.tokenInfo.tokenId)); printToken(List(token)) } 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")) for (token <- tokens) { val tokenInfo = token.tokenInfo @@ -92,9 +92,9 @@ object DelegationTokenCommand extends Logging { token.hmacAsBase64String, tokenInfo.owner, tokenInfo.renewersAsString, - tokenInfo.issueTimestamp, - tokenInfo.expiryTimestamp, - tokenInfo.maxTimestamp)) + dateFormat.format(tokenInfo.issueTimestamp), + dateFormat.format(tokenInfo.expiryTimestamp), + dateFormat.format(tokenInfo.maxTimestamp))) println() } } @@ -106,42 +106,36 @@ object DelegationTokenCommand extends Logging { List.empty[KafkaPrincipal] } - def renewToken(adminClient: AdminClient, opts: DelegationTokenCommandOptions) = { + def renewToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions) = { val hmac = opts.options.valueOf(opts.hmacOpt) val renewTimePeriodMs = opts.options.valueOf(opts.renewTimePeriodOpt).longValue() println("Calling renew token operation with hmac :" + hmac +" , renew-time-period :"+ renewTimePeriodMs) - val response = adminClient.renewToken(ByteBuffer.wrap(Base64.decoder.decode(hmac)), renewTimePeriodMs) - response match { - case (Errors.NONE, expiryTimeStamp) => println("Completed renew operation. New expiry timestamp : %s".format(expiryTimeStamp)) - case (e, expiryTimeStamp) => throw new AdminOperationException(e.message) - } + val renewResult = adminClient.renewDelegationToken(Base64.decoder.decode(hmac), new RenewDelegationTokenOptions().renewTimePeriodMs(renewTimePeriodMs)) + val expiryTimeStamp = renewResult.expiryTimestamp().get() + println("Completed renew operation. New expiry timestamp : %s".format(expiryTimeStamp)) } - def expireToken(adminClient: AdminClient, opts: DelegationTokenCommandOptions) = { + def expireToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions) = { val hmac = opts.options.valueOf(opts.hmacOpt) val expiryTimePeriodMs = opts.options.valueOf(opts.expiryTimePeriodOpt).longValue() println("Calling expire token operation with hmac :" + hmac +" , expire-time-period : "+ expiryTimePeriodMs) - val response = adminClient.expireToken(ByteBuffer.wrap(Base64.decoder.decode(hmac)), expiryTimePeriodMs) - response match { - case (Errors.NONE, expiryTimeStamp) => println("Completed expire operation. New expiry timestamp : %s".format(expiryTimeStamp)) - case (e, expiryTimeStamp) => throw new AdminOperationException(e.message) - } + val expireResult = adminClient.expireDelegationToken(Base64.decoder.decode(hmac), new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)) + val expiryTimeStamp = expireResult.expiryTimestamp().get() + println("Completed expire operation. New expiry timestamp : %s".format(expiryTimeStamp)) } - def describeToken(adminClient: AdminClient, opts: DelegationTokenCommandOptions) = { - val ownerPrincipals = getPrincipals(opts, opts.ownerPrincipalsOpt) + def describeToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions) = { + val ownerPrincipals = getPrincipals(opts, opts.ownerPrincipalsOpt).asJava println("Calling describe token operation for owners :" + ownerPrincipals) - val response = adminClient.describeToken(ownerPrincipals) - response match { - case (Errors.NONE, tokens) => println("Total Number of tokens : %s".format(tokens.size)); printToken(tokens) - case (e, tokens) => throw new AdminOperationException(e.message) - } + val describeResult = adminClient.describeDelegationToken(new DescribeDelegationTokenOptions().owners(ownerPrincipals)) + val tokens = describeResult.delegationTokens().get().asScala.toList + println("Total Number of tokens : %s".format(tokens.size)); printToken(tokens) } - private def createAdminClient(opts: DelegationTokenCommandOptions): AdminClient = { + private def createAdminClient(opts: DelegationTokenCommandOptions): JAdminClient = { val props = Utils.loadProps(opts.options.valueOf(opts.commandConfigOpt)) props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, opts.options.valueOf(opts.bootstrapServerOpt)) - AdminClient.create(props) + JAdminClient.create(props) } class DelegationTokenCommandOptions(args: Array[String]) { diff --git a/core/src/test/scala/integration/kafka/api/DelegationTokenEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/DelegationTokenEndToEndAuthorizationTest.scala index 27a6d1127d505..56a3b8a3f392a 100644 --- a/core/src/test/scala/integration/kafka/api/DelegationTokenEndToEndAuthorizationTest.scala +++ b/core/src/test/scala/integration/kafka/api/DelegationTokenEndToEndAuthorizationTest.scala @@ -18,10 +18,9 @@ package kafka.api import java.util -import kafka.admin.AdminClient import kafka.server.KafkaConfig import kafka.utils.{JaasTestUtils, TestUtils, ZkUtils} -import org.apache.kafka.clients.admin.AdminClientConfig +import org.apache.kafka.clients.admin.{AdminClient, AdminClientConfig} import org.apache.kafka.common.config.SaslConfigs import org.apache.kafka.common.security.auth.SecurityProtocol import org.apache.kafka.common.security.scram.internal.ScramMechanism @@ -83,9 +82,8 @@ class DelegationTokenEndToEndAuthorizationTest extends EndToEndAuthorizationTest val clientLoginContext = jaasClientLoginModule(kafkaClientSaslMechanism) config.put(SaslConfigs.SASL_JAAS_CONFIG, clientLoginContext) - val adminClient = AdminClient.create(config.asScala.toMap) - val (error, token) = adminClient.createToken(List()) - + val adminClient = AdminClient.create(config) + val token = adminClient.createDelegationToken().delegationToken().get() //wait for token to reach all the brokers TestUtils.waitUntilTrue(() => servers.forall(server => !server.tokenCache.tokens().isEmpty), "Timed out waiting for token to propagate to all servers") diff --git a/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsOnPlainTextTest.scala b/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsOnPlainTextTest.scala index 4c42dd277242a..3d4be53198510 100644 --- a/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsOnPlainTextTest.scala +++ b/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsOnPlainTextTest.scala @@ -19,14 +19,13 @@ package kafka.server import java.nio.ByteBuffer import java.util -import kafka.admin.AdminClient import kafka.utils.TestUtils -import org.apache.kafka.clients.admin.AdminClientConfig -import org.apache.kafka.common.protocol.Errors -import org.junit.Assert._ +import org.apache.kafka.clients.admin.{AdminClient, AdminClientConfig} +import org.apache.kafka.common.errors.UnsupportedByAuthenticationException import org.junit.{After, Before, Test} import scala.collection.JavaConverters._ +import scala.concurrent.ExecutionException class DelegationTokenRequestsOnPlainTextTest extends BaseRequestTest { var adminClient: AdminClient = null @@ -49,21 +48,19 @@ class DelegationTokenRequestsOnPlainTextTest extends BaseRequestTest { @Test def testDelegationTokenRequests(): Unit = { - adminClient = AdminClient.create(createAdminConfig.asScala.toMap) + adminClient = AdminClient.create(createAdminConfig) - val createResponse = adminClient.createToken(List()) - assertEquals(Errors.DELEGATION_TOKEN_REQUEST_NOT_ALLOWED, createResponse._1) + val createResult = adminClient.createDelegationToken() + intercept[ExecutionException](createResult.delegationToken().get()).getCause.isInstanceOf[UnsupportedByAuthenticationException] - val describeResponse = adminClient.describeToken(List()) - assertEquals(Errors.DELEGATION_TOKEN_REQUEST_NOT_ALLOWED, describeResponse._1) + val describeResult = adminClient.describeDelegationToken() + intercept[ExecutionException](describeResult.delegationTokens().get()).getCause.isInstanceOf[UnsupportedByAuthenticationException] - //test renewing tokens - val renewResponse = adminClient.renewToken(ByteBuffer.wrap("".getBytes())) - assertEquals(Errors.DELEGATION_TOKEN_REQUEST_NOT_ALLOWED, renewResponse._1) + val renewResult = adminClient.renewDelegationToken("".getBytes()) + intercept[ExecutionException](renewResult.expiryTimestamp().get()).getCause.isInstanceOf[UnsupportedByAuthenticationException] - //test expire tokens tokens - val expireResponse = adminClient.expireToken(ByteBuffer.wrap("".getBytes())) - assertEquals(Errors.DELEGATION_TOKEN_REQUEST_NOT_ALLOWED, expireResponse._1) + val expireResult = adminClient.expireDelegationToken("".getBytes()) + intercept[ExecutionException](expireResult.expiryTimestamp().get()).getCause.isInstanceOf[UnsupportedByAuthenticationException] } diff --git a/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsTest.scala b/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsTest.scala index 55bf5fd022cc2..20b3f6d3cfc46 100644 --- a/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsTest.scala +++ b/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsTest.scala @@ -18,17 +18,17 @@ package kafka.server import java.util -import kafka.admin.AdminClient import kafka.api.{KafkaSasl, SaslSetup} import kafka.utils.{JaasTestUtils, TestUtils} -import org.apache.kafka.clients.admin.AdminClientConfig -import org.apache.kafka.common.protocol.Errors +import org.apache.kafka.clients.admin.{AdminClient, AdminClientConfig, CreateDelegationTokenOptions, DescribeDelegationTokenOptions} +import org.apache.kafka.common.errors.InvalidPrincipalTypeException import org.apache.kafka.common.security.auth.SecurityProtocol import org.apache.kafka.common.utils.SecurityUtils import org.junit.Assert._ import org.junit.{After, Before, Test} import scala.collection.JavaConverters._ +import scala.concurrent.ExecutionException class DelegationTokenRequestsTest extends BaseRequestTest with SaslSetup { override protected def securityProtocol = SecurityProtocol.SASL_PLAINTEXT @@ -46,15 +46,6 @@ class DelegationTokenRequestsTest extends BaseRequestTest with SaslSetup { super.setUp() } - def createAdminConfig():util.Map[String, Object] = { - val config = new util.HashMap[String, Object] - config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList) - val securityProps: util.Map[Object, Object] = - TestUtils.adminClientSecurityConfigs(securityProtocol, trustStoreFile, clientSaslProperties) - securityProps.asScala.foreach { case (key, value) => config.put(key.asInstanceOf[String], value) } - config - } - override def generateConfigs = { val props = TestUtils.createBrokerConfigs(numBrokers, zkConnect, enableControlledShutdown = false, enableDeleteTopic = true, @@ -64,46 +55,72 @@ class DelegationTokenRequestsTest extends BaseRequestTest with SaslSetup { props.map(KafkaConfig.fromProps) } + def createAdminConfig():util.Map[String, Object] = { + val config = new util.HashMap[String, Object] + config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList) + val securityProps: util.Map[Object, Object] = + TestUtils.adminClientSecurityConfigs(securityProtocol, trustStoreFile, clientSaslProperties) + securityProps.asScala.foreach { case (key, value) => config.put(key.asInstanceOf[String], value) } + config + } + @Test def testDelegationTokenRequests(): Unit = { - adminClient = AdminClient.create(createAdminConfig.asScala.toMap) + adminClient = AdminClient.create(createAdminConfig) - // test creating token - val renewer1 = List(SecurityUtils.parseKafkaPrincipal("User:" + JaasTestUtils.KafkaPlainUser)) - val tokenResult1 = adminClient.createToken(renewer1) - assertEquals(Errors.NONE, tokenResult1._1) - var token1 = adminClient.describeToken(null)._2.head - assertEquals(token1, tokenResult1._2) + // create token1 with renewer1 + val renewer1 = List(SecurityUtils.parseKafkaPrincipal("User:renewer1")).asJava + val createResult1 = adminClient.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer1)) + val tokenCreated = createResult1.delegationToken().get() - //test renewing tokens - val renewResponse = adminClient.renewToken(token1.hmacBuffer()) - assertEquals(Errors.NONE, renewResponse._1) + //test describe token + var describeResult = adminClient.describeDelegationToken() + var token1 = describeResult.delegationTokens().get().get(0) + assertEquals(token1, tokenCreated) - token1 = adminClient.describeToken(null)._2.head - assertEquals(renewResponse._2, token1.tokenInfo().expiryTimestamp()) + // create token2 with renewer2 + val renewer2 = List(SecurityUtils.parseKafkaPrincipal("User:renewer2")).asJava + val createResult2 = adminClient.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer2)) + val token2 = createResult2.delegationToken().get() - //test describe tokens - val renewer2 = List(SecurityUtils.parseKafkaPrincipal("User:Renewer1")) - val tokenResult2 = adminClient.createToken(renewer2) - assertEquals(Errors.NONE, tokenResult2._1) - val token2 = tokenResult2._2 + //get all tokens + var tokens = adminClient.describeDelegationToken().delegationTokens().get() + assertTrue(tokens.size() == 2) + assertEquals(Set(token1, token2), tokens.asScala.toSet) - assertTrue(adminClient.describeToken(null)._2.size == 2) + //get tokens for renewer2 + tokens = adminClient.describeDelegationToken(new DescribeDelegationTokenOptions().owners(renewer2)).delegationTokens().get() + assertTrue(tokens.size() == 1) + assertEquals(Set(token2), tokens.asScala.toSet) - //test expire tokens tokens - val expireResponse1 = adminClient.expireToken(token1.hmacBuffer()) - assertEquals(Errors.NONE, expireResponse1._1) + //test renewing tokens + val renewResult = adminClient.renewDelegationToken(token1.hmac()) + var expiryTimestamp = renewResult.expiryTimestamp().get() - val expireResponse2 = adminClient.expireToken(token2.hmacBuffer()) - assertEquals(Errors.NONE, expireResponse2._1) + describeResult = adminClient.describeDelegationToken() + val tokenId = token1.tokenInfo().tokenId(); + token1 = describeResult.delegationTokens().get().asScala.filter(dt => dt.tokenInfo().tokenId() == tokenId).head + assertEquals(expiryTimestamp, token1.tokenInfo().expiryTimestamp()) - assertTrue(adminClient.describeToken(null)._2.size == 0) + //test expire tokens + val expireResult1 = adminClient.expireDelegationToken(token1.hmac()) + expiryTimestamp = expireResult1.expiryTimestamp().get() - //create token with invalid principal type - val renewer3 = List(SecurityUtils.parseKafkaPrincipal("Group:Renewer1")) - val tokenResult3 = adminClient.createToken(renewer3) - assertEquals(Errors.INVALID_PRINCIPAL_TYPE, tokenResult3._1) + val expireResult2 = adminClient.expireDelegationToken(token2.hmac()) + expiryTimestamp = expireResult2.expiryTimestamp().get() + tokens = adminClient.describeDelegationToken().delegationTokens().get() + assertTrue(tokens.size == 0) + + //create token with invalid principal type + val renewer3 = List(SecurityUtils.parseKafkaPrincipal("Group:Renewer3")).asJava + val createResult3 = adminClient.createDelegationToken(new CreateDelegationTokenOptions().renewers(renewer3)) + intercept[ExecutionException](createResult3.delegationToken().get()).getCause.isInstanceOf[InvalidPrincipalTypeException] + + // try describing tokens for unknown owner + val unknownOwner = List(SecurityUtils.parseKafkaPrincipal("User:Unknown")).asJava + tokens = adminClient.describeDelegationToken(new DescribeDelegationTokenOptions().owners(unknownOwner)).delegationTokens().get() + assertTrue(tokens.isEmpty) } @After diff --git a/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsWithDisableTokenFeatureTest.scala b/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsWithDisableTokenFeatureTest.scala index 0561cacb8f89b..8f8842bbe97e7 100644 --- a/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsWithDisableTokenFeatureTest.scala +++ b/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsWithDisableTokenFeatureTest.scala @@ -19,17 +19,15 @@ package kafka.server import java.nio.ByteBuffer import java.util -import kafka.admin.AdminClient import kafka.api.{KafkaSasl, SaslSetup} import kafka.utils.{JaasTestUtils, TestUtils} -import org.apache.kafka.clients.admin.AdminClientConfig -import org.apache.kafka.common.protocol.Errors -import org.apache.kafka.common.utils.SecurityUtils -import org.junit.Assert._ -import org.junit.{After, Before, Test} +import org.apache.kafka.clients.admin.{AdminClient, AdminClientConfig} +import org.apache.kafka.common.errors.DelegationTokenDisabledException import org.apache.kafka.common.security.auth.SecurityProtocol +import org.junit.{After, Before, Test} import scala.collection.JavaConverters._ +import scala.concurrent.ExecutionException class DelegationTokenRequestsWithDisableTokenFeatureTest extends BaseRequestTest with SaslSetup { override protected def securityProtocol = SecurityProtocol.SASL_PLAINTEXT @@ -58,23 +56,19 @@ class DelegationTokenRequestsWithDisableTokenFeatureTest extends BaseRequestTest @Test def testDelegationTokenRequests(): Unit = { - adminClient = AdminClient.create(createAdminConfig.asScala.toMap) - - val renewer1 = List(SecurityUtils.parseKafkaPrincipal("User:" + JaasTestUtils.KafkaPlainUser)) - val createResponse = adminClient.createToken(renewer1) - assertEquals(Errors.DELEGATION_TOKEN_AUTH_DISABLED, createResponse._1) + adminClient = AdminClient.create(createAdminConfig) - val describeResponse = adminClient.describeToken(List()) - assertEquals(Errors.DELEGATION_TOKEN_AUTH_DISABLED, describeResponse._1) + val createResult = adminClient.createDelegationToken() + intercept[ExecutionException](createResult.delegationToken().get()).getCause.isInstanceOf[DelegationTokenDisabledException] - //test renewing tokens - val renewResponse = adminClient.renewToken(ByteBuffer.wrap("".getBytes())) - assertEquals(Errors.DELEGATION_TOKEN_AUTH_DISABLED, renewResponse._1) + val describeResult = adminClient.describeDelegationToken() + intercept[ExecutionException](describeResult.delegationTokens().get()).getCause.isInstanceOf[DelegationTokenDisabledException] - //test expire tokens tokens - val expireResponse = adminClient.expireToken(ByteBuffer.wrap("".getBytes())) - assertEquals(Errors.DELEGATION_TOKEN_AUTH_DISABLED, expireResponse._1) + val renewResult = adminClient.renewDelegationToken("".getBytes()) + intercept[ExecutionException](renewResult.expiryTimestamp().get()).getCause.isInstanceOf[DelegationTokenDisabledException] + val expireResult = adminClient.expireDelegationToken("".getBytes()) + intercept[ExecutionException](expireResult.expiryTimestamp().get()).getCause.isInstanceOf[DelegationTokenDisabledException] } @After diff --git a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala index 2a7d6d400d59f..ed85415eacccd 100644 --- a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala +++ b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala @@ -315,13 +315,13 @@ class RequestQuotaTest extends BaseRequestTest { new CreateDelegationTokenRequest.Builder(Collections.singletonList(SecurityUtils.parseKafkaPrincipal("User:test")), 1000) case ApiKeys.EXPIRE_DELEGATION_TOKEN => - new ExpireDelegationTokenRequest.Builder(ByteBuffer.allocate(10), 1000) + new ExpireDelegationTokenRequest.Builder("".getBytes, 1000) case ApiKeys.DESCRIBE_DELEGATION_TOKEN => new DescribeDelegationTokenRequest.Builder(Collections.singletonList(SecurityUtils.parseKafkaPrincipal("User:test"))) case ApiKeys.RENEW_DELEGATION_TOKEN => - new RenewDelegationTokenRequest.Builder(ByteBuffer.allocate(10), 1000) + new RenewDelegationTokenRequest.Builder("".getBytes, 1000) case ApiKeys.DELETE_GROUPS => new DeleteGroupsRequest.Builder(Collections.singleton("test-group")) From 156f2bb33bf519878176a23cbdf1a33298750853 Mon Sep 17 00:00:00 2001 From: Manikumar Reddy Date: Sat, 7 Apr 2018 20:30:29 +0530 Subject: [PATCH 2/4] Address review comments --- build.gradle | 1 + .../admin/DescribeDelegationTokenOptions.java | 6 +++ .../admin/ExpireDelegationTokenOptions.java | 2 +- .../kafka/common/network/ChannelBuilders.java | 2 +- .../common/network/SaslChannelBuilder.java | 2 +- .../scram/internal/ScramSaslServer.java | 2 +- .../internal/ScramServerCallbackHandler.java | 4 +- .../token/delegation/DelegationToken.java | 6 +++ .../token/delegation/TokenInformation.java | 6 +++ .../{ => internal}/DelegationTokenCache.java | 4 +- .../DelegationTokenCredentialCallback.java | 2 +- .../kafka/common/network/NioEchoServer.java | 2 +- .../scram/internal/ScramSaslServerTest.java | 2 +- .../main/scala/kafka/admin/AdminClient.scala | 29 -------------- .../kafka/admin/DelegationTokenCommand.scala | 38 +++++++++++-------- .../kafka/security/CredentialProvider.scala | 2 +- .../kafka/server/DelegationTokenManager.scala | 3 +- .../main/scala/kafka/server/KafkaServer.scala | 2 +- .../DelegationTokenManagerTest.scala | 3 +- .../server/DelegationTokenRequestsTest.scala | 6 ++- 20 files changed, 64 insertions(+), 60 deletions(-) rename clients/src/main/java/org/apache/kafka/common/security/token/delegation/{ => internal}/DelegationTokenCache.java (95%) rename clients/src/main/java/org/apache/kafka/common/security/token/delegation/{ => internal}/DelegationTokenCredentialCallback.java (94%) diff --git a/build.gradle b/build.gradle index f83698050ed5a..69f560ec38fef 100644 --- a/build.gradle +++ b/build.gradle @@ -858,6 +858,7 @@ project(':clients') { include "**/org/apache/kafka/common/config/*" include "**/org/apache/kafka/common/security/auth/*" include "**/org/apache/kafka/server/policy/*" + include "**/org/apache/kafka/common/security/token/delegation/*" } } diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenOptions.java index 8fd98621b3776..60b99354e3552 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeDelegationTokenOptions.java @@ -31,6 +31,12 @@ public class DescribeDelegationTokenOptions extends AbstractOptions { private List owners; + /** + * if owners is null, all the user owned tokens and tokens where user have Describe permission + * will be returned. + * @param owners + * @return this instance + */ public DescribeDelegationTokenOptions owners(List owners) { this.owners = owners; return this; diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java index 971e4c3074bb0..138cd4e69e477 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/ExpireDelegationTokenOptions.java @@ -26,7 +26,7 @@ */ @InterfaceStability.Evolving public class ExpireDelegationTokenOptions extends AbstractOptions { - private long expiryTimePeriodMs = -1; + private long expiryTimePeriodMs = -1L; public ExpireDelegationTokenOptions expiryTimePeriodMs(long expiryTimePeriodMs) { this.expiryTimePeriodMs = expiryTimePeriodMs; diff --git a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java index 80ccb7e1382a2..078d844e85fce 100644 --- a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java +++ b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java @@ -26,7 +26,7 @@ import org.apache.kafka.common.security.auth.KafkaPrincipalBuilder; import org.apache.kafka.common.security.authenticator.CredentialCache; import org.apache.kafka.common.security.kerberos.KerberosShortNamer; -import org.apache.kafka.common.security.token.delegation.DelegationTokenCache; +import org.apache.kafka.common.security.token.delegation.internal.DelegationTokenCache; import org.apache.kafka.common.utils.Utils; import java.util.Collections; diff --git a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java index 5502164563b2b..3985c7e97fe56 100644 --- a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java +++ b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java @@ -41,7 +41,7 @@ import org.apache.kafka.common.security.scram.internal.ScramMechanism; import org.apache.kafka.common.security.scram.internal.ScramServerCallbackHandler; import org.apache.kafka.common.security.ssl.SslFactory; -import org.apache.kafka.common.security.token.delegation.DelegationTokenCache; +import org.apache.kafka.common.security.token.delegation.internal.DelegationTokenCache; import org.apache.kafka.common.utils.Java; import org.apache.kafka.common.utils.Utils; import org.slf4j.Logger; diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/internal/ScramSaslServer.java b/clients/src/main/java/org/apache/kafka/common/security/scram/internal/ScramSaslServer.java index deee0b8fb3363..60857279f59ad 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/scram/internal/ScramSaslServer.java +++ b/clients/src/main/java/org/apache/kafka/common/security/scram/internal/ScramSaslServer.java @@ -40,7 +40,7 @@ import org.apache.kafka.common.security.scram.internal.ScramMessages.ClientFirstMessage; import org.apache.kafka.common.security.scram.internal.ScramMessages.ServerFinalMessage; import org.apache.kafka.common.security.scram.internal.ScramMessages.ServerFirstMessage; -import org.apache.kafka.common.security.token.delegation.DelegationTokenCredentialCallback; +import org.apache.kafka.common.security.token.delegation.internal.DelegationTokenCredentialCallback; import org.apache.kafka.common.utils.Utils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/clients/src/main/java/org/apache/kafka/common/security/scram/internal/ScramServerCallbackHandler.java b/clients/src/main/java/org/apache/kafka/common/security/scram/internal/ScramServerCallbackHandler.java index 377aa3d3df5c7..9a3f0dc66b764 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/scram/internal/ScramServerCallbackHandler.java +++ b/clients/src/main/java/org/apache/kafka/common/security/scram/internal/ScramServerCallbackHandler.java @@ -29,8 +29,8 @@ import org.apache.kafka.common.security.authenticator.CredentialCache; import org.apache.kafka.common.security.scram.ScramCredential; import org.apache.kafka.common.security.scram.ScramCredentialCallback; -import org.apache.kafka.common.security.token.delegation.DelegationTokenCache; -import org.apache.kafka.common.security.token.delegation.DelegationTokenCredentialCallback; +import org.apache.kafka.common.security.token.delegation.internal.DelegationTokenCache; +import org.apache.kafka.common.security.token.delegation.internal.DelegationTokenCredentialCallback; public class ScramServerCallbackHandler implements AuthenticateCallbackHandler { diff --git a/clients/src/main/java/org/apache/kafka/common/security/token/delegation/DelegationToken.java b/clients/src/main/java/org/apache/kafka/common/security/token/delegation/DelegationToken.java index 92175d7196e34..e1f97c1b72d77 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/token/delegation/DelegationToken.java +++ b/clients/src/main/java/org/apache/kafka/common/security/token/delegation/DelegationToken.java @@ -16,10 +16,16 @@ */ package org.apache.kafka.common.security.token.delegation; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.common.utils.Base64; import java.util.Arrays; +/** + * A class representing a delegation token. + * + */ +@InterfaceStability.Evolving public class DelegationToken { private TokenInformation tokenInformation; private byte[] hmac; 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 1d500d21eeff6..ffd2af3f1c231 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 @@ -16,11 +16,17 @@ */ package org.apache.kafka.common.security.token.delegation; +import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.common.security.auth.KafkaPrincipal; import java.util.ArrayList; import java.util.Collection; +/** + * A class representing a delegation token details. + * + */ +@InterfaceStability.Evolving public class TokenInformation { private KafkaPrincipal owner; diff --git a/clients/src/main/java/org/apache/kafka/common/security/token/delegation/DelegationTokenCache.java b/clients/src/main/java/org/apache/kafka/common/security/token/delegation/internal/DelegationTokenCache.java similarity index 95% rename from clients/src/main/java/org/apache/kafka/common/security/token/delegation/DelegationTokenCache.java rename to clients/src/main/java/org/apache/kafka/common/security/token/delegation/internal/DelegationTokenCache.java index adea210e67847..c05b7350b4174 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/token/delegation/DelegationTokenCache.java +++ b/clients/src/main/java/org/apache/kafka/common/security/token/delegation/internal/DelegationTokenCache.java @@ -15,12 +15,14 @@ * limitations under the License. */ -package org.apache.kafka.common.security.token.delegation; +package org.apache.kafka.common.security.token.delegation.internal; import org.apache.kafka.common.security.authenticator.CredentialCache; import org.apache.kafka.common.security.scram.ScramCredential; import org.apache.kafka.common.security.scram.internal.ScramCredentialUtils; import org.apache.kafka.common.security.scram.internal.ScramMechanism; +import org.apache.kafka.common.security.token.delegation.DelegationToken; +import org.apache.kafka.common.security.token.delegation.TokenInformation; import java.util.Collection; import java.util.HashMap; diff --git a/clients/src/main/java/org/apache/kafka/common/security/token/delegation/DelegationTokenCredentialCallback.java b/clients/src/main/java/org/apache/kafka/common/security/token/delegation/internal/DelegationTokenCredentialCallback.java similarity index 94% rename from clients/src/main/java/org/apache/kafka/common/security/token/delegation/DelegationTokenCredentialCallback.java rename to clients/src/main/java/org/apache/kafka/common/security/token/delegation/internal/DelegationTokenCredentialCallback.java index 7490a3e91b1ca..294d7b1d3a015 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/token/delegation/DelegationTokenCredentialCallback.java +++ b/clients/src/main/java/org/apache/kafka/common/security/token/delegation/internal/DelegationTokenCredentialCallback.java @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.kafka.common.security.token.delegation; +package org.apache.kafka.common.security.token.delegation.internal; import org.apache.kafka.common.security.scram.ScramCredentialCallback; diff --git a/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java b/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java index fab8e934d8e4f..68979a1e01ff9 100644 --- a/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java +++ b/clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java @@ -44,7 +44,7 @@ import java.util.List; import java.util.Map; -import org.apache.kafka.common.security.token.delegation.DelegationTokenCache; +import org.apache.kafka.common.security.token.delegation.internal.DelegationTokenCache; /** * Non-blocking EchoServer implementation that uses ChannelBuilder to create channels diff --git a/clients/src/test/java/org/apache/kafka/common/security/scram/internal/ScramSaslServerTest.java b/clients/src/test/java/org/apache/kafka/common/security/scram/internal/ScramSaslServerTest.java index 3c4b82d7921d4..f6e43f9edcff5 100644 --- a/clients/src/test/java/org/apache/kafka/common/security/scram/internal/ScramSaslServerTest.java +++ b/clients/src/test/java/org/apache/kafka/common/security/scram/internal/ScramSaslServerTest.java @@ -23,7 +23,7 @@ import org.apache.kafka.common.errors.SaslAuthenticationException; import org.apache.kafka.common.security.authenticator.CredentialCache; import org.apache.kafka.common.security.scram.ScramCredential; -import org.apache.kafka.common.security.token.delegation.DelegationTokenCache; +import org.apache.kafka.common.security.token.delegation.internal.DelegationTokenCache; import org.junit.Before; import org.junit.Test; diff --git a/core/src/main/scala/kafka/admin/AdminClient.scala b/core/src/main/scala/kafka/admin/AdminClient.scala index 50d66192ae144..bcc11fd49176c 100644 --- a/core/src/main/scala/kafka/admin/AdminClient.scala +++ b/core/src/main/scala/kafka/admin/AdminClient.scala @@ -35,8 +35,6 @@ import org.apache.kafka.common.requests.ApiVersionsResponse.ApiVersion import org.apache.kafka.common.requests.DescribeGroupsResponse.GroupMetadata import org.apache.kafka.common.requests.OffsetFetchResponse import org.apache.kafka.common.utils.LogContext -import org.apache.kafka.common.security.auth.KafkaPrincipal -import org.apache.kafka.common.security.token.delegation.{DelegationToken, TokenInformation} import org.apache.kafka.common.utils.{KafkaThread, Time, Utils} import org.apache.kafka.common.{Cluster, Node, TopicPartition} @@ -342,33 +340,6 @@ class AdminClient(val time: Time, ConsumerGroupSummary(metadata.state, metadata.protocol, Some(consumers), coordinator) } - def createToken(renewers: List[KafkaPrincipal], maxTimePeriodMs: Long = -1): (Errors, DelegationToken) = { - val responseBody = sendAnyNode(ApiKeys.CREATE_DELEGATION_TOKEN, new CreateDelegationTokenRequest.Builder(renewers.asJava, maxTimePeriodMs)) - val response = responseBody.asInstanceOf[CreateDelegationTokenResponse] - val tokenInfo = new TokenInformation(response.tokenId, response.owner, renewers.asJava, - response.issueTimestamp, response.maxTimestamp, response.expiryTimestamp) - (response.error, new DelegationToken(tokenInfo, response.hmacBytes)) - } - - def renewToken(hmac: ByteBuffer, renewTimePeriod: Long = -1): (Errors, Long) = { - val responseBody = sendAnyNode(ApiKeys.RENEW_DELEGATION_TOKEN, new RenewDelegationTokenRequest.Builder(new Array[Byte](hmac.remaining), renewTimePeriod)) - val response = responseBody.asInstanceOf[RenewDelegationTokenResponse] - (response.error, response.expiryTimestamp) - } - - def expireToken(hmac: ByteBuffer, expiryTimeStamp: Long = -1): (Errors, Long) = { - val responseBody = sendAnyNode(ApiKeys.EXPIRE_DELEGATION_TOKEN, new ExpireDelegationTokenRequest.Builder(new Array[Byte](hmac.remaining), expiryTimeStamp)) - val response = responseBody.asInstanceOf[ExpireDelegationTokenResponse] - (response.error, response.expiryTimestamp) - } - - def describeToken(owners: List[KafkaPrincipal]): (Errors, List[DelegationToken]) = { - val ownersList = if (owners == null) null else owners.asJava - val responseBody = sendAnyNode(ApiKeys.RENEW_DELEGATION_TOKEN, new DescribeDelegationTokenRequest.Builder(ownersList)) - val response = responseBody.asInstanceOf[DescribeDelegationTokenResponse] - (response.error, response.tokens().asScala.toList) - } - def deleteConsumerGroups(groups: List[String]): Map[String, Errors] = { def coordinatorLookup(group: String): Either[Node, Errors] = { diff --git a/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala b/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala index 92a39473a28cf..a4d6c3c214f0f 100644 --- a/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala +++ b/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala @@ -18,8 +18,9 @@ package kafka.admin import java.text.SimpleDateFormat +import java.util -import joptsimple._ +import joptsimple.{ArgumentAcceptingOptionSpec, OptionParser} import kafka.utils.{CommandLineUtils, Exit, Logging} import org.apache.kafka.clients.CommonClientConfigs import org.apache.kafka.clients.admin.{CreateDelegationTokenOptions, DescribeDelegationTokenOptions, ExpireDelegationTokenOptions, RenewDelegationTokenOptions, AdminClient => JAdminClient} @@ -72,7 +73,7 @@ object DelegationTokenCommand extends Logging { } def createToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions) = { - val renewerPrincipals = getPrincipals(opts, opts.renewPrincipalsOpt).asJava + val renewerPrincipals = getPrincipals(opts, opts.renewPrincipalsOpt).getOrElse(new util.LinkedList[KafkaPrincipal]()) val maxLifeTimeMs = opts.options.valueOf(opts.maxLifeTimeOpt).longValue println("Calling create token operation with renewers :" + renewerPrincipals +" , max-life-time-period :"+ maxLifeTimeMs) @@ -99,11 +100,11 @@ object DelegationTokenCommand extends Logging { } } - private def getPrincipals(opts: DelegationTokenCommandOptions, principalOptionSpec: ArgumentAcceptingOptionSpec[String]): List[KafkaPrincipal] = { + private def getPrincipals(opts: DelegationTokenCommandOptions, principalOptionSpec: ArgumentAcceptingOptionSpec[String]): Option[util.List[KafkaPrincipal]] = { if (opts.options.has(principalOptionSpec)) - opts.options.valuesOf(principalOptionSpec).asScala.map(s => SecurityUtils.parseKafkaPrincipal(s.trim)).toList + Some(opts.options.valuesOf(principalOptionSpec).asScala.map(s => SecurityUtils.parseKafkaPrincipal(s.trim)).toList.asJava) else - List.empty[KafkaPrincipal] + None } def renewToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions) = { @@ -112,7 +113,8 @@ object DelegationTokenCommand extends Logging { println("Calling renew token operation with hmac :" + hmac +" , renew-time-period :"+ renewTimePeriodMs) val renewResult = adminClient.renewDelegationToken(Base64.decoder.decode(hmac), new RenewDelegationTokenOptions().renewTimePeriodMs(renewTimePeriodMs)) val expiryTimeStamp = renewResult.expiryTimestamp().get() - println("Completed renew operation. New expiry timestamp : %s".format(expiryTimeStamp)) + val dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm") + println("Completed renew operation. New expiry date : %s".format(dateFormat.format(expiryTimeStamp))) } def expireToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions) = { @@ -121,15 +123,20 @@ object DelegationTokenCommand extends Logging { println("Calling expire token operation with hmac :" + hmac +" , expire-time-period : "+ expiryTimePeriodMs) val expireResult = adminClient.expireDelegationToken(Base64.decoder.decode(hmac), new ExpireDelegationTokenOptions().expiryTimePeriodMs(expiryTimePeriodMs)) val expiryTimeStamp = expireResult.expiryTimestamp().get() - println("Completed expire operation. New expiry timestamp : %s".format(expiryTimeStamp)) + val dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm") + println("Completed expire operation. New expiry date : %s".format(dateFormat.format(expiryTimeStamp))) } def describeToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions) = { - val ownerPrincipals = getPrincipals(opts, opts.ownerPrincipalsOpt).asJava - println("Calling describe token operation for owners :" + ownerPrincipals) - val describeResult = adminClient.describeDelegationToken(new DescribeDelegationTokenOptions().owners(ownerPrincipals)) + val ownerPrincipals = getPrincipals(opts, opts.ownerPrincipalsOpt) + if (ownerPrincipals.isEmpty) + println("Calling describe token operation for current user.") + else + println("Calling describe token operation for owners :" + ownerPrincipals.get) + + val describeResult = adminClient.describeDelegationToken(new DescribeDelegationTokenOptions().owners(ownerPrincipals.orNull)) val tokens = describeResult.delegationTokens().get().asScala.toList - println("Total Number of tokens : %s".format(tokens.size)); printToken(tokens) + println("Total number of tokens : %s".format(tokens.size)); printToken(tokens) } private def createAdminClient(opts: DelegationTokenCommandOptions): JAdminClient = { @@ -151,10 +158,11 @@ object DelegationTokenCommand extends Logging { .withRequiredArg .ofType(classOf[String]) - val createOpt = parser.accepts("create", "Create a new delegation token.") - val renewOpt = parser.accepts("renew", "Renew delegation token.") - val expiryOpt = parser.accepts("expire", "Expire delegation token.") - val describeOpt = parser.accepts("describe", "describe delegation tokens.") + val createOpt = parser.accepts("create", "Create a new delegation token. Use --renewer-principal option to pass renewers principals.") + val renewOpt = parser.accepts("renew", "Renew delegation token. Use --renew-time-period option to set renew time period.") + val expiryOpt = parser.accepts("expire", "Expire delegation token. Use --expiry-time-period option to expire the token.") + val describeOpt = parser.accepts("describe", "Describe delegation tokens for the given principals. Use --owner-principal to pass owner/renewer principals." + + " If --owner-principal option is not supplied, all the user owned tokens and tokens where user have Describe permission will be returned.") val ownerPrincipalsOpt = parser.accepts("owner-principal", "owner is a kafka principal. It is should be in principalType:name format.") .withOptionalArg() diff --git a/core/src/main/scala/kafka/security/CredentialProvider.scala b/core/src/main/scala/kafka/security/CredentialProvider.scala index 6f9c2527735e3..f20808791a211 100644 --- a/core/src/main/scala/kafka/security/CredentialProvider.scala +++ b/core/src/main/scala/kafka/security/CredentialProvider.scala @@ -24,7 +24,7 @@ import org.apache.kafka.common.security.scram.ScramCredential import org.apache.kafka.common.config.ConfigDef import org.apache.kafka.common.config.ConfigDef._ import org.apache.kafka.common.security.scram.internal.{ScramCredentialUtils, ScramMechanism} -import org.apache.kafka.common.security.token.delegation.DelegationTokenCache +import org.apache.kafka.common.security.token.delegation.internal.DelegationTokenCache class CredentialProvider(scramMechanisms: Collection[String], val tokenCache: DelegationTokenCache) { diff --git a/core/src/main/scala/kafka/server/DelegationTokenManager.scala b/core/src/main/scala/kafka/server/DelegationTokenManager.scala index 4a947a17fcf86..62a5e20322b66 100644 --- a/core/src/main/scala/kafka/server/DelegationTokenManager.scala +++ b/core/src/main/scala/kafka/server/DelegationTokenManager.scala @@ -31,7 +31,8 @@ import org.apache.kafka.common.protocol.Errors import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.kafka.common.security.scram.internal.{ScramFormatter, ScramMechanism} import org.apache.kafka.common.security.scram.ScramCredential -import org.apache.kafka.common.security.token.delegation.{DelegationToken, DelegationTokenCache, TokenInformation} +import org.apache.kafka.common.security.token.delegation.internal.DelegationTokenCache +import org.apache.kafka.common.security.token.delegation.{DelegationToken, TokenInformation} import org.apache.kafka.common.utils.{Base64, Sanitizer, SecurityUtils, Time} import scala.collection.JavaConverters._ diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala index 71056888c7365..a0d2c799e6ae5 100755 --- a/core/src/main/scala/kafka/server/KafkaServer.scala +++ b/core/src/main/scala/kafka/server/KafkaServer.scala @@ -45,7 +45,7 @@ import org.apache.kafka.common.protocol.Errors import org.apache.kafka.common.requests.{ControlledShutdownRequest, ControlledShutdownResponse} import org.apache.kafka.common.security.auth.SecurityProtocol import org.apache.kafka.common.security.scram.internal.ScramMechanism -import org.apache.kafka.common.security.token.delegation.DelegationTokenCache +import org.apache.kafka.common.security.token.delegation.internal.DelegationTokenCache import org.apache.kafka.common.security.{JaasContext, JaasUtils} import org.apache.kafka.common.utils.{AppInfoParser, LogContext, Time} import org.apache.kafka.common.{ClusterResource, Node} 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 b8388b4cb49b8..6093622a18822 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 @@ -30,7 +30,8 @@ import kafka.zk.ZooKeeperTestHarness import org.apache.kafka.common.protocol.Errors import org.apache.kafka.common.security.auth.KafkaPrincipal import org.apache.kafka.common.security.scram.internal.ScramMechanism -import org.apache.kafka.common.security.token.delegation.{DelegationToken, DelegationTokenCache, TokenInformation} +import org.apache.kafka.common.security.token.delegation.internal.DelegationTokenCache +import org.apache.kafka.common.security.token.delegation.{DelegationToken, TokenInformation} import org.apache.kafka.common.utils.{MockTime, SecurityUtils} import org.junit.Assert._ import org.junit.{After, Before, Test} diff --git a/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsTest.scala b/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsTest.scala index 20b3f6d3cfc46..17975e3f119bc 100644 --- a/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsTest.scala +++ b/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsTest.scala @@ -75,6 +75,8 @@ class DelegationTokenRequestsTest extends BaseRequestTest with SaslSetup { //test describe token var describeResult = adminClient.describeDelegationToken() + var tokens = adminClient.describeDelegationToken().delegationTokens().get() + assertTrue(tokens.size() == 1) var token1 = describeResult.delegationTokens().get().get(0) assertEquals(token1, tokenCreated) @@ -84,7 +86,7 @@ class DelegationTokenRequestsTest extends BaseRequestTest with SaslSetup { val token2 = createResult2.delegationToken().get() //get all tokens - var tokens = adminClient.describeDelegationToken().delegationTokens().get() + tokens = adminClient.describeDelegationToken().delegationTokens().get() assertTrue(tokens.size() == 2) assertEquals(Set(token1, token2), tokens.asScala.toSet) @@ -98,7 +100,7 @@ class DelegationTokenRequestsTest extends BaseRequestTest with SaslSetup { var expiryTimestamp = renewResult.expiryTimestamp().get() describeResult = adminClient.describeDelegationToken() - val tokenId = token1.tokenInfo().tokenId(); + val tokenId = token1.tokenInfo().tokenId() token1 = describeResult.delegationTokens().get().asScala.filter(dt => dt.tokenInfo().tokenId() == tokenId).head assertEquals(expiryTimestamp, token1.tokenInfo().expiryTimestamp()) From 684173ebd246be1ad914425f6b1492474a86cca5 Mon Sep 17 00:00:00 2001 From: Manikumar Reddy Date: Mon, 9 Apr 2018 23:32:14 +0530 Subject: [PATCH 3/4] Add DelegationTokenCommandTest --- .../kafka/admin/DelegationTokenCommand.scala | 12 +- .../admin/DelegationTokenCommandTest.scala | 147 ++++++++++++++++++ 2 files changed, 155 insertions(+), 4 deletions(-) create mode 100644 core/src/test/scala/unit/kafka/admin/DelegationTokenCommandTest.scala diff --git a/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala b/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala index a4d6c3c214f0f..0e6ea86034b28 100644 --- a/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala +++ b/core/src/main/scala/kafka/admin/DelegationTokenCommand.scala @@ -72,7 +72,7 @@ object DelegationTokenCommand extends Logging { } } - def createToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions) = { + def createToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions): DelegationToken = { val renewerPrincipals = getPrincipals(opts, opts.renewPrincipalsOpt).getOrElse(new util.LinkedList[KafkaPrincipal]()) val maxLifeTimeMs = opts.options.valueOf(opts.maxLifeTimeOpt).longValue @@ -81,6 +81,7 @@ object DelegationTokenCommand extends Logging { val createResult = adminClient.createDelegationToken(createDelegationTokenOptions) val token = createResult.delegationToken().get() println("Created delegation token with tokenId : %s".format(token.tokenInfo.tokenId)); printToken(List(token)) + token } def printToken(tokens: List[DelegationToken]): Unit = { @@ -107,7 +108,7 @@ object DelegationTokenCommand extends Logging { None } - def renewToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions) = { + def renewToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions): Long = { val hmac = opts.options.valueOf(opts.hmacOpt) val renewTimePeriodMs = opts.options.valueOf(opts.renewTimePeriodOpt).longValue() println("Calling renew token operation with hmac :" + hmac +" , renew-time-period :"+ renewTimePeriodMs) @@ -115,9 +116,10 @@ object DelegationTokenCommand extends Logging { val expiryTimeStamp = renewResult.expiryTimestamp().get() val dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm") println("Completed renew operation. New expiry date : %s".format(dateFormat.format(expiryTimeStamp))) + expiryTimeStamp } - def expireToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions) = { + def expireToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions): Long = { val hmac = opts.options.valueOf(opts.hmacOpt) val expiryTimePeriodMs = opts.options.valueOf(opts.expiryTimePeriodOpt).longValue() println("Calling expire token operation with hmac :" + hmac +" , expire-time-period : "+ expiryTimePeriodMs) @@ -125,9 +127,10 @@ object DelegationTokenCommand extends Logging { val expiryTimeStamp = expireResult.expiryTimestamp().get() val dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm") println("Completed expire operation. New expiry date : %s".format(dateFormat.format(expiryTimeStamp))) + expiryTimeStamp } - def describeToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions) = { + def describeToken(adminClient: JAdminClient, opts: DelegationTokenCommandOptions): List[DelegationToken] = { val ownerPrincipals = getPrincipals(opts, opts.ownerPrincipalsOpt) if (ownerPrincipals.isEmpty) println("Calling describe token operation for current user.") @@ -137,6 +140,7 @@ object DelegationTokenCommand extends Logging { val describeResult = adminClient.describeDelegationToken(new DescribeDelegationTokenOptions().owners(ownerPrincipals.orNull)) val tokens = describeResult.delegationTokens().get().asScala.toList println("Total number of tokens : %s".format(tokens.size)); printToken(tokens) + tokens } private def createAdminClient(opts: DelegationTokenCommandOptions): JAdminClient = { diff --git a/core/src/test/scala/unit/kafka/admin/DelegationTokenCommandTest.scala b/core/src/test/scala/unit/kafka/admin/DelegationTokenCommandTest.scala new file mode 100644 index 0000000000000..0740637e4112b --- /dev/null +++ b/core/src/test/scala/unit/kafka/admin/DelegationTokenCommandTest.scala @@ -0,0 +1,147 @@ +/** + * 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.admin + +import java.util + +import kafka.admin.DelegationTokenCommand.DelegationTokenCommandOptions +import kafka.api.{KafkaSasl, SaslSetup} +import kafka.server.{BaseRequestTest, KafkaConfig} +import kafka.utils.{JaasTestUtils, TestUtils} +import org.apache.kafka.clients.admin.AdminClientConfig +import org.apache.kafka.common.security.auth.SecurityProtocol +import org.junit.Assert._ +import org.junit.{After, Before, Test} + +import scala.collection.JavaConverters._ +import scala.collection.mutable.ListBuffer +import scala.concurrent.ExecutionException + +class DelegationTokenCommandTest extends BaseRequestTest with SaslSetup { + override protected def securityProtocol = SecurityProtocol.SASL_PLAINTEXT + private val kafkaClientSaslMechanism = "PLAIN" + private val kafkaServerSaslMechanisms = List("PLAIN") + protected override val serverSaslProperties = Some(kafkaServerSaslProperties(kafkaServerSaslMechanisms, kafkaClientSaslMechanism)) + protected override val clientSaslProperties = Some(kafkaClientSaslProperties(kafkaClientSaslMechanism)) + var adminClient: org.apache.kafka.clients.admin.AdminClient = null + + override def numBrokers = 1 + + @Before + override def setUp(): Unit = { + startSasl(jaasSections(kafkaServerSaslMechanisms, Some(kafkaClientSaslMechanism), KafkaSasl, JaasTestUtils.KafkaServerContextName)) + super.setUp() + } + + override def generateConfigs = { + val props = TestUtils.createBrokerConfigs(numBrokers, zkConnect, + enableControlledShutdown = false, enableDeleteTopic = true, + interBrokerSecurityProtocol = Some(securityProtocol), + trustStoreFile = trustStoreFile, saslProperties = serverSaslProperties, enableToken = true) + props.foreach(propertyOverrides) + props.map(KafkaConfig.fromProps) + } + + def createAdminConfig():util.Map[String, Object] = { + val config = new util.HashMap[String, Object] + config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList) + val securityProps: util.Map[Object, Object] = + TestUtils.adminClientSecurityConfigs(securityProtocol, trustStoreFile, clientSaslProperties) + securityProps.asScala.foreach { case (key, value) => config.put(key.asInstanceOf[String], value) } + config + } + + @Test + def testDelegationTokenRequests(): Unit = { + adminClient = org.apache.kafka.clients.admin.AdminClient.create(createAdminConfig) + val renewer1 = "User:renewer1" + val renewer2 = "User:renewer2" + + // create token1 with renewer1 + val tokenCreated = DelegationTokenCommand.createToken(adminClient, getCreateOpts(List(renewer1))) + + var tokens = DelegationTokenCommand.describeToken(adminClient, getDescribeOpts(List())) + assertTrue(tokens.size == 1) + val token1 = tokens.head + assertEquals(token1, tokenCreated) + + // create token2 with renewer2 + val token2 = DelegationTokenCommand.createToken(adminClient, getCreateOpts(List(renewer2))) + + tokens = DelegationTokenCommand.describeToken(adminClient, getDescribeOpts(List())) + assertTrue(tokens.size == 2) + assertEquals(Set(token1, token2), tokens.toSet) + + //get tokens for renewer2 + tokens = DelegationTokenCommand.describeToken(adminClient, getDescribeOpts(List(renewer2))) + assertTrue(tokens.size == 1) + assertEquals(Set(token2), tokens.toSet) + + //test renewing tokens + val expiryTimestamp = DelegationTokenCommand.renewToken(adminClient, getRenewOpts(token1.hmacAsBase64String())) + val renewedToken = DelegationTokenCommand.describeToken(adminClient, getDescribeOpts(List(renewer1))).head + assertEquals(expiryTimestamp, renewedToken.tokenInfo().expiryTimestamp()) + + //test expire tokens + DelegationTokenCommand.expireToken(adminClient, getExpireOpts(token1.hmacAsBase64String())) + DelegationTokenCommand.expireToken(adminClient, getExpireOpts(token2.hmacAsBase64String())) + + tokens = DelegationTokenCommand.describeToken(adminClient, getDescribeOpts(List())) + assertTrue(tokens.size == 0) + + //create token with invalid renewer principal type + intercept[ExecutionException](DelegationTokenCommand.createToken(adminClient, getCreateOpts(List("Group:Renewer3")))) + + // try describing tokens for unknown owner + assertTrue(DelegationTokenCommand.describeToken(adminClient, getDescribeOpts(List("User:Unknown"))).isEmpty) + } + + def getCreateOpts(renewers: List[String]): DelegationTokenCommandOptions = { + val opts = ListBuffer("--bootstrap-server", brokerList, "--max-life-time-period", "-1", + "--command-config", "testfile", "--create") + renewers.map(renewer => opts ++= ListBuffer("--renewer-principal", renewer)) + new DelegationTokenCommandOptions(opts.toArray) + } + + def getDescribeOpts(owners: List[String]): DelegationTokenCommandOptions = { + val opts = ListBuffer("--bootstrap-server", brokerList, "--command-config", "testfile", "--describe") + owners.map(owner => opts ++= ListBuffer("--owner-principal", owner)) + new DelegationTokenCommandOptions(opts.toArray) + } + + def getRenewOpts(hmac: String): DelegationTokenCommandOptions = { + val opts = Array("--bootstrap-server", brokerList, "--command-config", "testfile", "--renew", + "--renew-time-period", "-1", + "--hmac", hmac) + new DelegationTokenCommandOptions(opts) + } + + def getExpireOpts(hmac: String): DelegationTokenCommandOptions = { + val opts = Array("--bootstrap-server", brokerList, "--command-config", "testfile", "--expire", + "--expiry-time-period", "-1", + "--hmac", hmac) + new DelegationTokenCommandOptions(opts) + } + + @After + override def tearDown(): Unit = { + if (adminClient != null) + adminClient.close() + super.tearDown() + closeSasl() + } +} From 81b40204a83d0ed4ab4f78896542fad06b6f3f72 Mon Sep 17 00:00:00 2001 From: Manikumar Reddy Date: Wed, 11 Apr 2018 10:03:34 +0530 Subject: [PATCH 4/4] Address review comments --- .../kafka/admin/DelegationTokenCommandTest.scala | 14 +++++++------- .../kafka/server/DelegationTokenRequestsTest.scala | 7 +++---- 2 files changed, 10 insertions(+), 11 deletions(-) diff --git a/core/src/test/scala/unit/kafka/admin/DelegationTokenCommandTest.scala b/core/src/test/scala/unit/kafka/admin/DelegationTokenCommandTest.scala index 0740637e4112b..6ae8f5e83ac3c 100644 --- a/core/src/test/scala/unit/kafka/admin/DelegationTokenCommandTest.scala +++ b/core/src/test/scala/unit/kafka/admin/DelegationTokenCommandTest.scala @@ -56,7 +56,7 @@ class DelegationTokenCommandTest extends BaseRequestTest with SaslSetup { props.map(KafkaConfig.fromProps) } - def createAdminConfig():util.Map[String, Object] = { + private def createAdminConfig():util.Map[String, Object] = { val config = new util.HashMap[String, Object] config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList) val securityProps: util.Map[Object, Object] = @@ -110,27 +110,27 @@ class DelegationTokenCommandTest extends BaseRequestTest with SaslSetup { assertTrue(DelegationTokenCommand.describeToken(adminClient, getDescribeOpts(List("User:Unknown"))).isEmpty) } - def getCreateOpts(renewers: List[String]): DelegationTokenCommandOptions = { + private def getCreateOpts(renewers: List[String]): DelegationTokenCommandOptions = { val opts = ListBuffer("--bootstrap-server", brokerList, "--max-life-time-period", "-1", "--command-config", "testfile", "--create") - renewers.map(renewer => opts ++= ListBuffer("--renewer-principal", renewer)) + renewers.foreach(renewer => opts ++= ListBuffer("--renewer-principal", renewer)) new DelegationTokenCommandOptions(opts.toArray) } - def getDescribeOpts(owners: List[String]): DelegationTokenCommandOptions = { + private def getDescribeOpts(owners: List[String]): DelegationTokenCommandOptions = { val opts = ListBuffer("--bootstrap-server", brokerList, "--command-config", "testfile", "--describe") - owners.map(owner => opts ++= ListBuffer("--owner-principal", owner)) + owners.foreach(owner => opts ++= ListBuffer("--owner-principal", owner)) new DelegationTokenCommandOptions(opts.toArray) } - def getRenewOpts(hmac: String): DelegationTokenCommandOptions = { + private def getRenewOpts(hmac: String): DelegationTokenCommandOptions = { val opts = Array("--bootstrap-server", brokerList, "--command-config", "testfile", "--renew", "--renew-time-period", "-1", "--hmac", hmac) new DelegationTokenCommandOptions(opts) } - def getExpireOpts(hmac: String): DelegationTokenCommandOptions = { + private def getExpireOpts(hmac: String): DelegationTokenCommandOptions = { val opts = Array("--bootstrap-server", brokerList, "--command-config", "testfile", "--expire", "--expiry-time-period", "-1", "--hmac", hmac) diff --git a/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsTest.scala b/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsTest.scala index 17975e3f119bc..a00275084fbc3 100644 --- a/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsTest.scala +++ b/core/src/test/scala/unit/kafka/server/DelegationTokenRequestsTest.scala @@ -55,7 +55,7 @@ class DelegationTokenRequestsTest extends BaseRequestTest with SaslSetup { props.map(KafkaConfig.fromProps) } - def createAdminConfig():util.Map[String, Object] = { + private def createAdminConfig():util.Map[String, Object] = { val config = new util.HashMap[String, Object] config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList) val securityProps: util.Map[Object, Object] = @@ -74,10 +74,9 @@ class DelegationTokenRequestsTest extends BaseRequestTest with SaslSetup { val tokenCreated = createResult1.delegationToken().get() //test describe token - var describeResult = adminClient.describeDelegationToken() var tokens = adminClient.describeDelegationToken().delegationTokens().get() assertTrue(tokens.size() == 1) - var token1 = describeResult.delegationTokens().get().get(0) + var token1 = tokens.get(0) assertEquals(token1, tokenCreated) // create token2 with renewer2 @@ -99,7 +98,7 @@ class DelegationTokenRequestsTest extends BaseRequestTest with SaslSetup { val renewResult = adminClient.renewDelegationToken(token1.hmac()) var expiryTimestamp = renewResult.expiryTimestamp().get() - describeResult = adminClient.describeDelegationToken() + val describeResult = adminClient.describeDelegationToken() val tokenId = token1.tokenInfo().tokenId() token1 = describeResult.delegationTokens().get().asScala.filter(dt => dt.tokenInfo().tokenId() == tokenId).head assertEquals(expiryTimestamp, token1.tokenInfo().expiryTimestamp())