pprovenzano commented on code in PR #14083:
URL: https://github.com/apache/kafka/pull/14083#discussion_r1293786529
##########
core/src/main/scala/kafka/server/ControllerApis.scala:
##########
@@ -842,6 +847,75 @@ class ControllerApis(val requestChannel: RequestChannel,
}
}
+ def handleCreateDelegationTokenRequest(request: RequestChannel.Request):
CompletableFuture[Unit] = {
+ val alterRequest = request.body[CreateDelegationTokenRequest]
+
+ val requester = request.context.principal
+ val ownerPrincipalName = alterRequest.data.ownerPrincipalName
+ val ownerPrincipalType = alterRequest.data.ownerPrincipalType
+ val owner = if (ownerPrincipalName == null || ownerPrincipalName.isEmpty) {
+ request.context.principal
+ } else {
+ new KafkaPrincipal(ownerPrincipalType, ownerPrincipalName)
+ }
+
+ // Requester is always allowed to create token for self
+ if (!owner.equals(requester) &&
+ !authHelper.authorize(request.context, CREATE_TOKENS, USER,
owner.toString)) {
+ requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
+
CreateDelegationTokenResponse.prepareResponse(request.context.requestVersion,
requestThrottleMs,
+ Errors.DELEGATION_TOKEN_AUTHORIZATION_FAILED, owner, requester))
+ }
+
+ val context = new ControllerRequestContext(request.context.header.data,
request.context.principal,
+ OptionalLong.empty())
+
+ // Copy the response data to a new response so we can apply the request
version
+ controller.createDelegationToken(context, alterRequest.data)
+ .thenApply[Unit] { response =>
+ requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
+ CreateDelegationTokenResponse.prepareResponse(
+ request.context.requestVersion,
+ requestThrottleMs,
+ Errors.forCode(response.errorCode()),
+ new KafkaPrincipal(response.principalType(),
response.principalName()),
+ new KafkaPrincipal(response.tokenRequesterPrincipalType(),
response.tokenRequesterPrincipalName()),
+ response.issueTimestampMs(),
+ response.expiryTimestampMs(),
+ response.maxTimestampMs(),
+ response.tokenId(),
+ ByteBuffer.wrap(response.hmac())))
+ }
+ }
+
+ def handleRenewDelegationTokenRequest(request: RequestChannel.Request):
CompletableFuture[Unit] = {
+ val alterRequest = request.body[RenewDelegationTokenRequest]
Review Comment:
Fixed
##########
core/src/main/scala/kafka/server/ControllerApis.scala:
##########
@@ -842,6 +847,75 @@ class ControllerApis(val requestChannel: RequestChannel,
}
}
+ def handleCreateDelegationTokenRequest(request: RequestChannel.Request):
CompletableFuture[Unit] = {
+ val alterRequest = request.body[CreateDelegationTokenRequest]
+
+ val requester = request.context.principal
+ val ownerPrincipalName = alterRequest.data.ownerPrincipalName
+ val ownerPrincipalType = alterRequest.data.ownerPrincipalType
+ val owner = if (ownerPrincipalName == null || ownerPrincipalName.isEmpty) {
+ request.context.principal
+ } else {
+ new KafkaPrincipal(ownerPrincipalType, ownerPrincipalName)
+ }
+
+ // Requester is always allowed to create token for self
+ if (!owner.equals(requester) &&
+ !authHelper.authorize(request.context, CREATE_TOKENS, USER,
owner.toString)) {
+ requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
+
CreateDelegationTokenResponse.prepareResponse(request.context.requestVersion,
requestThrottleMs,
+ Errors.DELEGATION_TOKEN_AUTHORIZATION_FAILED, owner, requester))
+ }
+
+ val context = new ControllerRequestContext(request.context.header.data,
request.context.principal,
+ OptionalLong.empty())
+
+ // Copy the response data to a new response so we can apply the request
version
+ controller.createDelegationToken(context, alterRequest.data)
+ .thenApply[Unit] { response =>
+ requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
+ CreateDelegationTokenResponse.prepareResponse(
+ request.context.requestVersion,
+ requestThrottleMs,
+ Errors.forCode(response.errorCode()),
+ new KafkaPrincipal(response.principalType(),
response.principalName()),
+ new KafkaPrincipal(response.tokenRequesterPrincipalType(),
response.tokenRequesterPrincipalName()),
+ response.issueTimestampMs(),
+ response.expiryTimestampMs(),
+ response.maxTimestampMs(),
+ response.tokenId(),
+ ByteBuffer.wrap(response.hmac())))
+ }
+ }
+
+ def handleRenewDelegationTokenRequest(request: RequestChannel.Request):
CompletableFuture[Unit] = {
+ val alterRequest = request.body[RenewDelegationTokenRequest]
+
+ val context = new ControllerRequestContext(
+ request.context.header.data,
+ request.context.principal,
+ OptionalLong.empty())
+ controller.renewDelegationToken(context, alterRequest.data)
+ .thenApply[Unit] { response =>
+ requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
+ new
RenewDelegationTokenResponse(response.setThrottleTimeMs(requestThrottleMs)))
+ }
+ }
+
+ def handleExpireDelegationTokenRequest(request: RequestChannel.Request):
CompletableFuture[Unit] = {
+ val alterRequest = request.body[ExpireDelegationTokenRequest]
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
