You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/05/21 01:00:08 UTC
[GitHub] [pulsar] codelipenghui opened a new pull request, #15694: [fix][security] Add timeout of sync methods for AuthoriationService
codelipenghui opened a new pull request, #15694:
URL: https://github.com/apache/pulsar/pull/15694
### Documentation
Check the box below or label this PR directly.
Need to update docs?
- [x] `no-need-doc`
(Please explain why)
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Demogorgon314 commented on a diff in pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
Demogorgon314 commented on code in PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#discussion_r878699358
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -3880,46 +3879,53 @@ public static CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMet
PulsarService pulsar, String clientAppId, String originalPrincipal,
AuthenticationDataSource authenticationData, TopicName topicName) {
CompletableFuture<PartitionedTopicMetadata> metadataFuture = new CompletableFuture<>();
- try {
- // (1) authorize client
- try {
- checkAuthorization(pulsar, topicName, clientAppId, authenticationData);
- } catch (RestException e) {
- try {
- validateAdminAccessForTenant(pulsar,
- clientAppId, originalPrincipal, topicName.getTenant(), authenticationData,
- pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds(), SECONDS);
- } catch (RestException authException) {
- log.warn("Failed to authorize {} on topic {}", clientAppId, topicName);
- throw new PulsarClientException(String.format("Authorization failed %s on topic %s with error %s",
- clientAppId, topicName, authException.getMessage()));
- }
- } catch (Exception ex) {
- // throw without wrapping to PulsarClientException that considers: unknown error marked as internal
- // server error
- log.warn("Failed to authorize {} on topic {}", clientAppId, topicName, ex);
- throw ex;
- }
+ CompletableFuture<Void> authorizationFuture = new CompletableFuture<>();
+ checkAuthorizationAsync(pulsar, topicName, clientAppId, authenticationData)
+ .thenRun(() -> authorizationFuture.complete(null))
+ .exceptionally(e -> {
+ if (FutureUtil.unwrapCompletionException(e) instanceof RestException) {
+ validateAdminAccessForTenantAsync(pulsar,
+ clientAppId, originalPrincipal, topicName.getTenant(), authenticationData)
+ .thenRun(() -> {
+ authorizationFuture.complete(null);
+ }).exceptionally(ex -> {
+ if (FutureUtil.unwrapCompletionException(ex) instanceof RestException) {
+ log.warn("Failed to authorize {} on topic {}", clientAppId, topicName);
+ authorizationFuture.completeExceptionally(new PulsarClientException(
+ String.format("Authorization failed %s on topic %s with error %s",
+ clientAppId, topicName, ex.getCause().getMessage())));
Review Comment:
Same as the previous comment.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#issuecomment-1133875106
/pulsarbot run-failure-checks
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nodece commented on a diff in pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#discussion_r878653408
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java:
##########
@@ -220,49 +220,52 @@ public static CompletableFuture<ByteBuf> lookupTopicAsync(PulsarService pulsarSe
cluster);
}
validationFuture.complete(newLookupResponse(differentClusterData.getBrokerServiceUrl(),
- differentClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect, requestId, false));
+ differentClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect,
+ requestId, false));
} else {
// (2) authorize client
- try {
- checkAuthorization(pulsarService, topicName, clientAppId, authenticationData);
- } catch (RestException authException) {
- log.warn("Failed to authorized {} on cluster {}", clientAppId, topicName.toString());
- validationFuture.complete(newLookupErrorResponse(ServerError.AuthorizationError,
- authException.getMessage(), requestId));
- return;
- } catch (Exception e) {
- log.warn("Unknown error while authorizing {} on cluster {}", clientAppId, topicName.toString());
- validationFuture.completeExceptionally(e);
- return;
- }
- // (3) validate global namespace
- checkLocalOrGetPeerReplicationCluster(pulsarService, topicName.getNamespaceObject())
- .thenAccept(peerClusterData -> {
- if (peerClusterData == null) {
- // (4) all validation passed: initiate lookup
- validationFuture.complete(null);
- return;
- }
- // if peer-cluster-data is present it means namespace is owned by that peer-cluster and
- // request should be redirect to the peer-cluster
- if (StringUtils.isBlank(peerClusterData.getBrokerServiceUrl())
- && StringUtils.isBlank(peerClusterData.getBrokerServiceUrlTls())) {
- validationFuture.complete(newLookupErrorResponse(ServerError.MetadataError,
- "Redirected cluster's brokerService url is not configured", requestId));
- return;
+ checkAuthorizationAsync(pulsarService, topicName, clientAppId, authenticationData).thenRun(() ->
+ validationFuture.complete(null))
+ .exceptionally(e -> {
+ if (e.getCause() instanceof RestException) {
+ log.warn("Failed to authorized {} on cluster {}", clientAppId, topicName);
+ validationFuture.complete(newLookupErrorResponse(ServerError.AuthorizationError,
+ e.getCause().getMessage(), requestId));
+ } else {
+ log.warn("Unknown error while authorizing {} on cluster {}", clientAppId, topicName);
+ validationFuture.completeExceptionally(e);
}
- validationFuture.complete(newLookupResponse(peerClusterData.getBrokerServiceUrl(),
- peerClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect, requestId,
- false));
+ return null;
+ });
- }).exceptionally(ex -> {
+ // (3) validate global namespace
+ validationFuture.thenCompose(__ ->
+ checkLocalOrGetPeerReplicationCluster(pulsarService,
+ topicName.getNamespaceObject())).thenAccept(peerClusterData -> {
+ if (peerClusterData == null) {
+ // (4) all validation passed: initiate lookup
+ validationFuture.complete(null);
+ return;
+ }
+ // if peer-cluster-data is present it means namespace is owned by that peer-cluster and
+ // request should be redirect to the peer-cluster
+ if (StringUtils.isBlank(peerClusterData.getBrokerServiceUrl())
+ && StringUtils.isBlank(peerClusterData.getBrokerServiceUrlTls())) {
+ validationFuture.complete(newLookupErrorResponse(ServerError.MetadataError,
+ "Redirected cluster's brokerService url is not configured", requestId));
+ return;
+ }
+ validationFuture.complete(newLookupResponse(peerClusterData.getBrokerServiceUrl(),
+ peerClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect, requestId,
+ false));
+ }).exceptionally(ex -> {
validationFuture.complete(
- newLookupErrorResponse(ServerError.MetadataError, ex.getMessage(), requestId));
+ newLookupErrorResponse(ServerError.MetadataError, ex.getCause().getMessage(), requestId));
return null;
});
}
}).exceptionally(ex -> {
- validationFuture.completeExceptionally(ex);
+ validationFuture.completeExceptionally(ex.getCause());
Review Comment:
```suggestion
validationFuture.completeExceptionally(FutureUtil.unwrapCompletionException(ex));
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Demogorgon314 commented on a diff in pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
Demogorgon314 commented on code in PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#discussion_r878698477
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java:
##########
@@ -220,49 +220,52 @@ public static CompletableFuture<ByteBuf> lookupTopicAsync(PulsarService pulsarSe
cluster);
}
validationFuture.complete(newLookupResponse(differentClusterData.getBrokerServiceUrl(),
- differentClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect, requestId, false));
+ differentClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect,
+ requestId, false));
} else {
// (2) authorize client
- try {
- checkAuthorization(pulsarService, topicName, clientAppId, authenticationData);
- } catch (RestException authException) {
- log.warn("Failed to authorized {} on cluster {}", clientAppId, topicName.toString());
- validationFuture.complete(newLookupErrorResponse(ServerError.AuthorizationError,
- authException.getMessage(), requestId));
- return;
- } catch (Exception e) {
- log.warn("Unknown error while authorizing {} on cluster {}", clientAppId, topicName.toString());
- validationFuture.completeExceptionally(e);
- return;
- }
- // (3) validate global namespace
- checkLocalOrGetPeerReplicationCluster(pulsarService, topicName.getNamespaceObject())
- .thenAccept(peerClusterData -> {
- if (peerClusterData == null) {
- // (4) all validation passed: initiate lookup
- validationFuture.complete(null);
- return;
- }
- // if peer-cluster-data is present it means namespace is owned by that peer-cluster and
- // request should be redirect to the peer-cluster
- if (StringUtils.isBlank(peerClusterData.getBrokerServiceUrl())
- && StringUtils.isBlank(peerClusterData.getBrokerServiceUrlTls())) {
- validationFuture.complete(newLookupErrorResponse(ServerError.MetadataError,
- "Redirected cluster's brokerService url is not configured", requestId));
- return;
+ checkAuthorizationAsync(pulsarService, topicName, clientAppId, authenticationData).thenRun(() ->
+ validationFuture.complete(null))
+ .exceptionally(e -> {
+ if (FutureUtil.unwrapCompletionException(e) instanceof RestException) {
+ log.warn("Failed to authorized {} on cluster {}", clientAppId, topicName);
+ validationFuture.complete(newLookupErrorResponse(ServerError.AuthorizationError,
+ FutureUtil.unwrapCompletionException(e).getMessage(), requestId));
+ } else {
+ log.warn("Unknown error while authorizing {} on cluster {}", clientAppId, topicName);
+ validationFuture.completeExceptionally(e);
Review Comment:
```suggestion
Throwable throwable = FutureUtil.unwrapCompletionException(e);
if (throwable instanceof RestException) {
log.warn("Failed to authorized {} on cluster {}", clientAppId, topicName);
validationFuture.complete(newLookupErrorResponse(ServerError.AuthorizationError,
throwable.getMessage(), requestId));
} else {
log.warn("Unknown error while authorizing {} on cluster {}", clientAppId, topicName);
validationFuture.completeExceptionally(throwable);
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nodece commented on a diff in pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#discussion_r878649918
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -3880,46 +3879,53 @@ public static CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMet
PulsarService pulsar, String clientAppId, String originalPrincipal,
AuthenticationDataSource authenticationData, TopicName topicName) {
CompletableFuture<PartitionedTopicMetadata> metadataFuture = new CompletableFuture<>();
- try {
- // (1) authorize client
- try {
- checkAuthorization(pulsar, topicName, clientAppId, authenticationData);
- } catch (RestException e) {
- try {
- validateAdminAccessForTenant(pulsar,
- clientAppId, originalPrincipal, topicName.getTenant(), authenticationData,
- pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds(), SECONDS);
- } catch (RestException authException) {
- log.warn("Failed to authorize {} on topic {}", clientAppId, topicName);
- throw new PulsarClientException(String.format("Authorization failed %s on topic %s with error %s",
- clientAppId, topicName, authException.getMessage()));
- }
- } catch (Exception ex) {
- // throw without wrapping to PulsarClientException that considers: unknown error marked as internal
- // server error
- log.warn("Failed to authorize {} on topic {}", clientAppId, topicName, ex);
- throw ex;
- }
+ CompletableFuture<Void> authorizationFuture = new CompletableFuture<>();
+ checkAuthorizationAsync(pulsar, topicName, clientAppId, authenticationData)
+ .thenRun(() -> authorizationFuture.complete(null))
+ .exceptionally(e -> {
+ if (e.getCause() instanceof RestException) {
Review Comment:
```suggestion
if (FutureUtil.unwrapCompletionException(e) instanceof RestException) {
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nodece commented on a diff in pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#discussion_r878653299
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -3880,46 +3879,53 @@ public static CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMet
PulsarService pulsar, String clientAppId, String originalPrincipal,
AuthenticationDataSource authenticationData, TopicName topicName) {
CompletableFuture<PartitionedTopicMetadata> metadataFuture = new CompletableFuture<>();
- try {
- // (1) authorize client
- try {
- checkAuthorization(pulsar, topicName, clientAppId, authenticationData);
- } catch (RestException e) {
- try {
- validateAdminAccessForTenant(pulsar,
- clientAppId, originalPrincipal, topicName.getTenant(), authenticationData,
- pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds(), SECONDS);
- } catch (RestException authException) {
- log.warn("Failed to authorize {} on topic {}", clientAppId, topicName);
- throw new PulsarClientException(String.format("Authorization failed %s on topic %s with error %s",
- clientAppId, topicName, authException.getMessage()));
- }
- } catch (Exception ex) {
- // throw without wrapping to PulsarClientException that considers: unknown error marked as internal
- // server error
- log.warn("Failed to authorize {} on topic {}", clientAppId, topicName, ex);
- throw ex;
- }
+ CompletableFuture<Void> authorizationFuture = new CompletableFuture<>();
+ checkAuthorizationAsync(pulsar, topicName, clientAppId, authenticationData)
+ .thenRun(() -> authorizationFuture.complete(null))
+ .exceptionally(e -> {
+ if (e.getCause() instanceof RestException) {
+ validateAdminAccessForTenantAsync(pulsar,
+ clientAppId, originalPrincipal, topicName.getTenant(), authenticationData)
+ .thenRun(() -> {
+ authorizationFuture.complete(null);
+ }).exceptionally(ex -> {
+ if (ex.getCause() instanceof RestException) {
+ log.warn("Failed to authorize {} on topic {}", clientAppId, topicName);
+ authorizationFuture.completeExceptionally(new PulsarClientException(
+ String.format("Authorization failed %s on topic %s with error %s",
+ clientAppId, topicName, ex.getCause().getMessage())));
+ } else {
+ authorizationFuture.completeExceptionally(e);
+ }
+ return null;
+ });
+ } else {
+ // throw without wrapping to PulsarClientException that considers: unknown error marked as
+ // internal server error
+ log.warn("Failed to authorize {} on topic {}", clientAppId, topicName, e);
+ authorizationFuture.completeExceptionally(e);
+ }
+ return null;
+ });
- // validates global-namespace contains local/peer cluster: if peer/local cluster present then lookup can
- // serve/redirect request else fail partitioned-metadata-request so, client fails while creating
- // producer/consumer
- checkLocalOrGetPeerReplicationCluster(pulsar, topicName.getNamespaceObject())
- .thenCompose(res -> pulsar.getBrokerService()
- .fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName))
- .thenAccept(metadata -> {
- if (log.isDebugEnabled()) {
- log.debug("[{}] Total number of partitions for topic {} is {}", clientAppId, topicName,
- metadata.partitions);
- }
- metadataFuture.complete(metadata);
- }).exceptionally(ex -> {
- metadataFuture.completeExceptionally(ex.getCause());
- return null;
- });
- } catch (Exception ex) {
- metadataFuture.completeExceptionally(ex);
- }
+ // validates global-namespace contains local/peer cluster: if peer/local cluster present then lookup can
+ // serve/redirect request else fail partitioned-metadata-request so, client fails while creating
+ // producer/consumer
+ authorizationFuture.thenCompose(__ ->
+ checkLocalOrGetPeerReplicationCluster(pulsar, topicName.getNamespaceObject()))
+ .thenCompose(res ->
+ pulsar.getBrokerService().fetchPartitionedTopicMetadataCheckAllowAutoCreationAsync(topicName))
+ .thenAccept(metadata -> {
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Total number of partitions for topic {} is {}", clientAppId, topicName,
+ metadata.partitions);
+ }
+ metadataFuture.complete(metadata);
+ })
+ .exceptionally(e -> {
+ metadataFuture.completeExceptionally(e.getCause());
Review Comment:
```suggestion
metadataFuture.completeExceptionally(FutureUtil.unwrapCompletionException(e));
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nodece commented on a diff in pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#discussion_r878653346
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java:
##########
@@ -220,49 +220,52 @@ public static CompletableFuture<ByteBuf> lookupTopicAsync(PulsarService pulsarSe
cluster);
}
validationFuture.complete(newLookupResponse(differentClusterData.getBrokerServiceUrl(),
- differentClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect, requestId, false));
+ differentClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect,
+ requestId, false));
} else {
// (2) authorize client
- try {
- checkAuthorization(pulsarService, topicName, clientAppId, authenticationData);
- } catch (RestException authException) {
- log.warn("Failed to authorized {} on cluster {}", clientAppId, topicName.toString());
- validationFuture.complete(newLookupErrorResponse(ServerError.AuthorizationError,
- authException.getMessage(), requestId));
- return;
- } catch (Exception e) {
- log.warn("Unknown error while authorizing {} on cluster {}", clientAppId, topicName.toString());
- validationFuture.completeExceptionally(e);
- return;
- }
- // (3) validate global namespace
- checkLocalOrGetPeerReplicationCluster(pulsarService, topicName.getNamespaceObject())
- .thenAccept(peerClusterData -> {
- if (peerClusterData == null) {
- // (4) all validation passed: initiate lookup
- validationFuture.complete(null);
- return;
- }
- // if peer-cluster-data is present it means namespace is owned by that peer-cluster and
- // request should be redirect to the peer-cluster
- if (StringUtils.isBlank(peerClusterData.getBrokerServiceUrl())
- && StringUtils.isBlank(peerClusterData.getBrokerServiceUrlTls())) {
- validationFuture.complete(newLookupErrorResponse(ServerError.MetadataError,
- "Redirected cluster's brokerService url is not configured", requestId));
- return;
+ checkAuthorizationAsync(pulsarService, topicName, clientAppId, authenticationData).thenRun(() ->
+ validationFuture.complete(null))
+ .exceptionally(e -> {
+ if (e.getCause() instanceof RestException) {
+ log.warn("Failed to authorized {} on cluster {}", clientAppId, topicName);
+ validationFuture.complete(newLookupErrorResponse(ServerError.AuthorizationError,
+ e.getCause().getMessage(), requestId));
+ } else {
+ log.warn("Unknown error while authorizing {} on cluster {}", clientAppId, topicName);
+ validationFuture.completeExceptionally(e);
}
- validationFuture.complete(newLookupResponse(peerClusterData.getBrokerServiceUrl(),
- peerClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect, requestId,
- false));
+ return null;
+ });
- }).exceptionally(ex -> {
+ // (3) validate global namespace
+ validationFuture.thenCompose(__ ->
+ checkLocalOrGetPeerReplicationCluster(pulsarService,
+ topicName.getNamespaceObject())).thenAccept(peerClusterData -> {
+ if (peerClusterData == null) {
+ // (4) all validation passed: initiate lookup
+ validationFuture.complete(null);
+ return;
+ }
+ // if peer-cluster-data is present it means namespace is owned by that peer-cluster and
+ // request should be redirect to the peer-cluster
+ if (StringUtils.isBlank(peerClusterData.getBrokerServiceUrl())
+ && StringUtils.isBlank(peerClusterData.getBrokerServiceUrlTls())) {
+ validationFuture.complete(newLookupErrorResponse(ServerError.MetadataError,
+ "Redirected cluster's brokerService url is not configured", requestId));
+ return;
+ }
+ validationFuture.complete(newLookupResponse(peerClusterData.getBrokerServiceUrl(),
+ peerClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect, requestId,
+ false));
+ }).exceptionally(ex -> {
validationFuture.complete(
- newLookupErrorResponse(ServerError.MetadataError, ex.getMessage(), requestId));
+ newLookupErrorResponse(ServerError.MetadataError, ex.getCause().getMessage(), requestId));
Review Comment:
```suggestion
newLookupErrorResponse(ServerError.MetadataError, FutureUtil.unwrapCompletionException(ex).getMessage(), requestId));
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nodece commented on a diff in pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#discussion_r878653229
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java:
##########
@@ -220,49 +220,52 @@ public static CompletableFuture<ByteBuf> lookupTopicAsync(PulsarService pulsarSe
cluster);
}
validationFuture.complete(newLookupResponse(differentClusterData.getBrokerServiceUrl(),
- differentClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect, requestId, false));
+ differentClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect,
+ requestId, false));
} else {
// (2) authorize client
- try {
- checkAuthorization(pulsarService, topicName, clientAppId, authenticationData);
- } catch (RestException authException) {
- log.warn("Failed to authorized {} on cluster {}", clientAppId, topicName.toString());
- validationFuture.complete(newLookupErrorResponse(ServerError.AuthorizationError,
- authException.getMessage(), requestId));
- return;
- } catch (Exception e) {
- log.warn("Unknown error while authorizing {} on cluster {}", clientAppId, topicName.toString());
- validationFuture.completeExceptionally(e);
- return;
- }
- // (3) validate global namespace
- checkLocalOrGetPeerReplicationCluster(pulsarService, topicName.getNamespaceObject())
- .thenAccept(peerClusterData -> {
- if (peerClusterData == null) {
- // (4) all validation passed: initiate lookup
- validationFuture.complete(null);
- return;
- }
- // if peer-cluster-data is present it means namespace is owned by that peer-cluster and
- // request should be redirect to the peer-cluster
- if (StringUtils.isBlank(peerClusterData.getBrokerServiceUrl())
- && StringUtils.isBlank(peerClusterData.getBrokerServiceUrlTls())) {
- validationFuture.complete(newLookupErrorResponse(ServerError.MetadataError,
- "Redirected cluster's brokerService url is not configured", requestId));
- return;
+ checkAuthorizationAsync(pulsarService, topicName, clientAppId, authenticationData).thenRun(() ->
+ validationFuture.complete(null))
+ .exceptionally(e -> {
+ if (e.getCause() instanceof RestException) {
Review Comment:
```suggestion
if (FutureUtil.unwrapCompletionException(e) instanceof RestException) {
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #15694: [fix][security] Add timeout of sync methods for AuthoriationService
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#issuecomment-1133520742
@nodece For the pulsar codebase, we don't have a place to call `allowTenantOperation`, but since they are public methods, which might be called in the plugins? This is my understanding of why the method have have any reference in Pulsar but did not removed yet.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nodece commented on a diff in pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#discussion_r878653154
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -3880,46 +3879,53 @@ public static CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMet
PulsarService pulsar, String clientAppId, String originalPrincipal,
AuthenticationDataSource authenticationData, TopicName topicName) {
CompletableFuture<PartitionedTopicMetadata> metadataFuture = new CompletableFuture<>();
- try {
- // (1) authorize client
- try {
- checkAuthorization(pulsar, topicName, clientAppId, authenticationData);
- } catch (RestException e) {
- try {
- validateAdminAccessForTenant(pulsar,
- clientAppId, originalPrincipal, topicName.getTenant(), authenticationData,
- pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds(), SECONDS);
- } catch (RestException authException) {
- log.warn("Failed to authorize {} on topic {}", clientAppId, topicName);
- throw new PulsarClientException(String.format("Authorization failed %s on topic %s with error %s",
- clientAppId, topicName, authException.getMessage()));
- }
- } catch (Exception ex) {
- // throw without wrapping to PulsarClientException that considers: unknown error marked as internal
- // server error
- log.warn("Failed to authorize {} on topic {}", clientAppId, topicName, ex);
- throw ex;
- }
+ CompletableFuture<Void> authorizationFuture = new CompletableFuture<>();
+ checkAuthorizationAsync(pulsar, topicName, clientAppId, authenticationData)
+ .thenRun(() -> authorizationFuture.complete(null))
+ .exceptionally(e -> {
+ if (e.getCause() instanceof RestException) {
Review Comment:
```suggestion
if (FutureUtil.unwrapCompletionException(e) instanceof RestException) {
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#discussion_r878783911
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java:
##########
@@ -220,49 +220,52 @@ public static CompletableFuture<ByteBuf> lookupTopicAsync(PulsarService pulsarSe
cluster);
}
validationFuture.complete(newLookupResponse(differentClusterData.getBrokerServiceUrl(),
- differentClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect, requestId, false));
+ differentClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect,
+ requestId, false));
} else {
// (2) authorize client
- try {
- checkAuthorization(pulsarService, topicName, clientAppId, authenticationData);
- } catch (RestException authException) {
- log.warn("Failed to authorized {} on cluster {}", clientAppId, topicName.toString());
- validationFuture.complete(newLookupErrorResponse(ServerError.AuthorizationError,
- authException.getMessage(), requestId));
- return;
- } catch (Exception e) {
- log.warn("Unknown error while authorizing {} on cluster {}", clientAppId, topicName.toString());
- validationFuture.completeExceptionally(e);
- return;
- }
- // (3) validate global namespace
- checkLocalOrGetPeerReplicationCluster(pulsarService, topicName.getNamespaceObject())
- .thenAccept(peerClusterData -> {
- if (peerClusterData == null) {
- // (4) all validation passed: initiate lookup
- validationFuture.complete(null);
- return;
- }
- // if peer-cluster-data is present it means namespace is owned by that peer-cluster and
- // request should be redirect to the peer-cluster
- if (StringUtils.isBlank(peerClusterData.getBrokerServiceUrl())
- && StringUtils.isBlank(peerClusterData.getBrokerServiceUrlTls())) {
- validationFuture.complete(newLookupErrorResponse(ServerError.MetadataError,
- "Redirected cluster's brokerService url is not configured", requestId));
- return;
+ checkAuthorizationAsync(pulsarService, topicName, clientAppId, authenticationData).thenRun(() ->
+ validationFuture.complete(null))
+ .exceptionally(e -> {
+ if (FutureUtil.unwrapCompletionException(e) instanceof RestException) {
+ log.warn("Failed to authorized {} on cluster {}", clientAppId, topicName);
+ validationFuture.complete(newLookupErrorResponse(ServerError.AuthorizationError,
+ FutureUtil.unwrapCompletionException(e).getMessage(), requestId));
+ } else {
+ log.warn("Unknown error while authorizing {} on cluster {}", clientAppId, topicName);
+ validationFuture.completeExceptionally(e);
Review Comment:
Fixed
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -3880,46 +3879,53 @@ public static CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMet
PulsarService pulsar, String clientAppId, String originalPrincipal,
AuthenticationDataSource authenticationData, TopicName topicName) {
CompletableFuture<PartitionedTopicMetadata> metadataFuture = new CompletableFuture<>();
- try {
- // (1) authorize client
- try {
- checkAuthorization(pulsar, topicName, clientAppId, authenticationData);
- } catch (RestException e) {
- try {
- validateAdminAccessForTenant(pulsar,
- clientAppId, originalPrincipal, topicName.getTenant(), authenticationData,
- pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds(), SECONDS);
- } catch (RestException authException) {
- log.warn("Failed to authorize {} on topic {}", clientAppId, topicName);
- throw new PulsarClientException(String.format("Authorization failed %s on topic %s with error %s",
- clientAppId, topicName, authException.getMessage()));
- }
- } catch (Exception ex) {
- // throw without wrapping to PulsarClientException that considers: unknown error marked as internal
- // server error
- log.warn("Failed to authorize {} on topic {}", clientAppId, topicName, ex);
- throw ex;
- }
+ CompletableFuture<Void> authorizationFuture = new CompletableFuture<>();
+ checkAuthorizationAsync(pulsar, topicName, clientAppId, authenticationData)
+ .thenRun(() -> authorizationFuture.complete(null))
+ .exceptionally(e -> {
+ if (FutureUtil.unwrapCompletionException(e) instanceof RestException) {
+ validateAdminAccessForTenantAsync(pulsar,
+ clientAppId, originalPrincipal, topicName.getTenant(), authenticationData)
+ .thenRun(() -> {
+ authorizationFuture.complete(null);
+ }).exceptionally(ex -> {
+ if (FutureUtil.unwrapCompletionException(ex) instanceof RestException) {
+ log.warn("Failed to authorize {} on topic {}", clientAppId, topicName);
+ authorizationFuture.completeExceptionally(new PulsarClientException(
+ String.format("Authorization failed %s on topic %s with error %s",
+ clientAppId, topicName, ex.getCause().getMessage())));
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nodece commented on a diff in pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#discussion_r878653162
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java:
##########
@@ -3880,46 +3879,53 @@ public static CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMet
PulsarService pulsar, String clientAppId, String originalPrincipal,
AuthenticationDataSource authenticationData, TopicName topicName) {
CompletableFuture<PartitionedTopicMetadata> metadataFuture = new CompletableFuture<>();
- try {
- // (1) authorize client
- try {
- checkAuthorization(pulsar, topicName, clientAppId, authenticationData);
- } catch (RestException e) {
- try {
- validateAdminAccessForTenant(pulsar,
- clientAppId, originalPrincipal, topicName.getTenant(), authenticationData,
- pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds(), SECONDS);
- } catch (RestException authException) {
- log.warn("Failed to authorize {} on topic {}", clientAppId, topicName);
- throw new PulsarClientException(String.format("Authorization failed %s on topic %s with error %s",
- clientAppId, topicName, authException.getMessage()));
- }
- } catch (Exception ex) {
- // throw without wrapping to PulsarClientException that considers: unknown error marked as internal
- // server error
- log.warn("Failed to authorize {} on topic {}", clientAppId, topicName, ex);
- throw ex;
- }
+ CompletableFuture<Void> authorizationFuture = new CompletableFuture<>();
+ checkAuthorizationAsync(pulsar, topicName, clientAppId, authenticationData)
+ .thenRun(() -> authorizationFuture.complete(null))
+ .exceptionally(e -> {
+ if (e.getCause() instanceof RestException) {
+ validateAdminAccessForTenantAsync(pulsar,
+ clientAppId, originalPrincipal, topicName.getTenant(), authenticationData)
+ .thenRun(() -> {
+ authorizationFuture.complete(null);
+ }).exceptionally(ex -> {
+ if (ex.getCause() instanceof RestException) {
Review Comment:
```suggestion
if (FutureUtil.unwrapCompletionException(ex) instanceof RestException) {
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#issuecomment-1199074831
Move the `release/2.8.4` label to https://github.com/apache/pulsar/pull/16831
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui merged pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
codelipenghui merged PR #15694:
URL: https://github.com/apache/pulsar/pull/15694
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #15694: [fix][security] Add timeout of sync methods for AuthoriationService
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#issuecomment-1133529311
@RobertIndie @mattisonchao Please help review again, I have changed the `allowTopicOperation` callers to use the async method `allowTopicOperationAsync` so that we don't have async calls for `AuthoriationService` anymore in the master branch.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] nodece commented on a diff in pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
nodece commented on code in PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#discussion_r878653270
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/lookup/TopicLookupBase.java:
##########
@@ -220,49 +220,52 @@ public static CompletableFuture<ByteBuf> lookupTopicAsync(PulsarService pulsarSe
cluster);
}
validationFuture.complete(newLookupResponse(differentClusterData.getBrokerServiceUrl(),
- differentClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect, requestId, false));
+ differentClusterData.getBrokerServiceUrlTls(), true, LookupType.Redirect,
+ requestId, false));
} else {
// (2) authorize client
- try {
- checkAuthorization(pulsarService, topicName, clientAppId, authenticationData);
- } catch (RestException authException) {
- log.warn("Failed to authorized {} on cluster {}", clientAppId, topicName.toString());
- validationFuture.complete(newLookupErrorResponse(ServerError.AuthorizationError,
- authException.getMessage(), requestId));
- return;
- } catch (Exception e) {
- log.warn("Unknown error while authorizing {} on cluster {}", clientAppId, topicName.toString());
- validationFuture.completeExceptionally(e);
- return;
- }
- // (3) validate global namespace
- checkLocalOrGetPeerReplicationCluster(pulsarService, topicName.getNamespaceObject())
- .thenAccept(peerClusterData -> {
- if (peerClusterData == null) {
- // (4) all validation passed: initiate lookup
- validationFuture.complete(null);
- return;
- }
- // if peer-cluster-data is present it means namespace is owned by that peer-cluster and
- // request should be redirect to the peer-cluster
- if (StringUtils.isBlank(peerClusterData.getBrokerServiceUrl())
- && StringUtils.isBlank(peerClusterData.getBrokerServiceUrlTls())) {
- validationFuture.complete(newLookupErrorResponse(ServerError.MetadataError,
- "Redirected cluster's brokerService url is not configured", requestId));
- return;
+ checkAuthorizationAsync(pulsarService, topicName, clientAppId, authenticationData).thenRun(() ->
+ validationFuture.complete(null))
+ .exceptionally(e -> {
+ if (e.getCause() instanceof RestException) {
+ log.warn("Failed to authorized {} on cluster {}", clientAppId, topicName);
+ validationFuture.complete(newLookupErrorResponse(ServerError.AuthorizationError,
+ e.getCause().getMessage(), requestId));
Review Comment:
```suggestion
FutureUtil.unwrapCompletionException(e).getMessage(), requestId));
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Technoboy- commented on pull request #15694: [fix][security] Add timeout of sync methods and avoid call sync method for AuthoriationService
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on PR #15694:
URL: https://github.com/apache/pulsar/pull/15694#issuecomment-1133586667
Better to rebase
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org