This is an automated email from the ASF dual-hosted git repository.
jerrypeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 5657797 refactor AuthorizationProvider interface to allow custom
logic for determining super user (#3383)
5657797 is described below
commit 5657797e1ef75b882e85759bd3a5f52b70584d3f
Author: Boyang Jerry Peng <jerry.boyang.p...@gmail.com>
AuthorDate: Fri Jan 25 01:08:53 2019 -0600
refactor AuthorizationProvider interface to allow custom logic for
determining super user (#3383)
* refactor AuthorizationProvider interface to allow custom logic for
determining super user
* remove redundant super user checks
* addressing comments
* simplifying code
* cleaning up code
* making API backwards compatible
* fixing tests
---
.../authorization/AuthorizationProvider.java | 13 ++++-
.../broker/authorization/AuthorizationService.java | 38 +++++++++++----
.../authorization/PulsarAuthorizationProvider.java | 21 ++------
.../pulsar/broker/web/PulsarWebResource.java | 56 ++++++++++++++++------
.../pulsar/broker/service/ServerCnxTest.java | 6 +--
.../api/AuthorizationProducerConsumerTest.java | 9 ++++
6 files changed, 98 insertions(+), 45 deletions(-)
diff --git
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
index 5d8b930..b25c789 100644
---
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
+++
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java
@@ -36,9 +36,20 @@ import org.apache.pulsar.common.policies.data.AuthAction;
public interface AuthorizationProvider extends Closeable {
/**
+ * Check if specified role is a super user
+ * @param role the role to check
+ * @return a CompletableFuture containing a boolean in which true means
the role is a super user
+ * and false if it is not
+ */
+ default CompletableFuture<Boolean> isSuperUser(String role,
ServiceConfiguration serviceConfiguration) {
+ Set<String> superUserRoles = serviceConfiguration.getSuperUserRoles();
+ return CompletableFuture.completedFuture(role != null &&
superUserRoles.contains(role) ? true : false);
+ }
+
+ /**
* Perform initialization for the authorization provider
*
- * @param config
+ * @param conf
* broker config object
* @param configCache
* pulsar zk configuration cache service
diff --git
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
index 95d012f..42d935c 100644
---
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
+++
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
@@ -18,12 +18,6 @@
*/
package org.apache.pulsar.broker.authorization;
-import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.apache.pulsar.zookeeper.ZooKeeperCache.cacheTimeOutInSec;
-
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.ServiceConfiguration;
@@ -36,6 +30,14 @@ import org.apache.pulsar.common.util.FutureUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.function.Function;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.pulsar.zookeeper.ZooKeeperCache.cacheTimeOutInSec;
+
/**
* Authorization service that manages pluggable authorization provider and
authorize requests accordingly.
*
@@ -70,6 +72,13 @@ public class AuthorizationService {
}
}
+ public CompletableFuture<Boolean> isSuperUser(String user) {
+ if (provider != null) {
+ return provider.isSuperUser(user, conf);
+ }
+ return FutureUtil.failedFuture(new IllegalStateException("No
authorization provider configured"));
+ }
+
/**
*
* Grant authorization-action permission on a namespace to the given client
@@ -164,9 +173,14 @@ public class AuthorizationService {
if (!this.conf.isAuthorizationEnabled()) {
return CompletableFuture.completedFuture(true);
}
-
if (provider != null) {
- return provider.canProduceAsync(topicName, role,
authenticationData);
+ return provider.isSuperUser(role,
conf).thenComposeAsync(isSuperUser -> {
+ if (isSuperUser) {
+ return CompletableFuture.completedFuture(true);
+ } else {
+ return provider.canProduceAsync(topicName, role,
authenticationData);
+ }
+ });
}
return FutureUtil.failedFuture(new IllegalStateException("No
authorization provider configured"));
}
@@ -187,7 +201,13 @@ public class AuthorizationService {
return CompletableFuture.completedFuture(true);
}
if (provider != null) {
- return provider.canConsumeAsync(topicName, role,
authenticationData, subscription);
+ return provider.isSuperUser(role,
conf).thenComposeAsync(isSuperUser -> {
+ if (isSuperUser) {
+ return CompletableFuture.completedFuture(true);
+ } else {
+ return provider.canConsumeAsync(topicName, role,
authenticationData, subscription);
+ }
+ });
}
return FutureUtil.failedFuture(new IllegalStateException("No
authorization provider configured"));
}
diff --git
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
index 229efec..7914168 100644
---
a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
+++
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java
@@ -27,6 +27,7 @@ import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.ServiceConfiguration;
@@ -112,7 +113,7 @@ public class PulsarAuthorizationProvider implements
AuthorizationProvider {
log.debug("Policies node couldn't be found for topic :
{}", topicName);
}
} else {
- if (isNotBlank(subscription) && !isSuperUser(role)) {
+ if (isNotBlank(subscription)) {
// validate if role is authorize to access
subscription. (skip validatation if authorization
// list is empty)
Set<String> roles =
policies.get().auth_policies.subscription_auth_roles.get(subscription);
@@ -323,12 +324,8 @@ public class PulsarAuthorizationProvider implements
AuthorizationProvider {
}
private CompletableFuture<Boolean> checkAuthorization(TopicName topicName,
String role, AuthAction action) {
- if (isSuperUser(role)) {
- return CompletableFuture.completedFuture(true);
- } else {
- return checkPermission(topicName, role, action)
- .thenApply(isPermission -> isPermission &&
checkCluster(topicName));
- }
+ return checkPermission(topicName, role, action)
+ .thenApply(isPermission -> isPermission &&
checkCluster(topicName));
}
private boolean checkCluster(TopicName topicName) {
@@ -424,16 +421,6 @@ public class PulsarAuthorizationProvider implements
AuthorizationProvider {
return false;
}
- /**
- * Super user roles are allowed to do anything, used for replication
primarily
- *
- * @param role
- * the app id used to receive messages from the topic.
- */
- public boolean isSuperUser(String role) {
- Set<String> superUserRoles = conf.getSuperUserRoles();
- return role != null && superUserRoles.contains(role) ? true : false;
- }
@Override
public void close() throws IOException {
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
index 81e5a40..dc631d6 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
@@ -32,6 +32,7 @@ import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import javax.servlet.ServletContext;
import javax.servlet.http.HttpServletRequest;
@@ -176,14 +177,26 @@ public abstract class PulsarWebResource {
validateOriginalPrincipal(pulsar.getConfiguration().getProxyRoles(), appId,
originalPrincipal);
if (pulsar.getConfiguration().getProxyRoles().contains(appId)) {
- Set<String> superUserRoles =
pulsar.getConfiguration().getSuperUserRoles();
- boolean proxyAuthorized = superUserRoles.contains(appId);
- boolean originalPrincipalAuthorized =
superUserRoles.contains(originalPrincipal);
- if (!proxyAuthorized || !originalPrincipalAuthorized) {
- throw new RestException(Status.UNAUTHORIZED,
- String.format("Proxy not authorized for super-user
operation (proxy:%s,original:%s)",
- appId, originalPrincipal));
+ CompletableFuture<Boolean> proxyAuthorizedFuture;
+ CompletableFuture<Boolean> originalPrincipalAuthorizedFuture;
+
+ try {
+ proxyAuthorizedFuture = pulsar.getBrokerService()
+ .getAuthorizationService()
+ .isSuperUser(appId);
+
+ originalPrincipalAuthorizedFuture =
pulsar.getBrokerService()
+ .getAuthorizationService()
+ .isSuperUser(originalPrincipal);
+
+ if (!proxyAuthorizedFuture.get() ||
!originalPrincipalAuthorizedFuture.get()) {
+ throw new RestException(Status.UNAUTHORIZED,
+ String.format("Proxy not authorized for
super-user operation (proxy:%s,original:%s)",
+ appId, originalPrincipal));
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RestException(Status.INTERNAL_SERVER_ERROR,
e.getMessage());
}
log.debug("Successfully authorized {} (proxied by {}) as
super-user",
originalPrincipal, appId);
@@ -238,16 +251,29 @@ public abstract class PulsarWebResource {
validateOriginalPrincipal(pulsar.getConfiguration().getProxyRoles(),
clientAppId, originalPrincipal);
if
(pulsar.getConfiguration().getProxyRoles().contains(clientAppId)) {
- Set<String> superUserRoles =
pulsar.getConfiguration().getSuperUserRoles();
+
+ CompletableFuture<Boolean> isProxySuperUserFuture;
+ CompletableFuture<Boolean> isOriginalPrincipalSuperUserFuture;
+ try {
+ isProxySuperUserFuture = pulsar.getBrokerService()
+ .getAuthorizationService()
+ .isSuperUser(clientAppId);
+
+ isOriginalPrincipalSuperUserFuture =
pulsar.getBrokerService()
+ .getAuthorizationService()
+ .isSuperUser(originalPrincipal);
+
Set<String> adminRoles = tenantInfo.getAdminRoles();
- boolean proxyAuthorized = superUserRoles.contains(clientAppId)
|| adminRoles.contains(clientAppId);
+ boolean proxyAuthorized = isProxySuperUserFuture.get() ||
adminRoles.contains(clientAppId);
boolean originalPrincipalAuthorized
- = superUserRoles.contains(originalPrincipal) ||
adminRoles.contains(originalPrincipal);
-
- if (!proxyAuthorized || !originalPrincipalAuthorized) {
- throw new RestException(Status.UNAUTHORIZED,
- String.format("Proxy not authorized to access
resource (proxy:%s,original:%s)",
- clientAppId, originalPrincipal));
+ = isOriginalPrincipalSuperUserFuture.get() ||
adminRoles.contains(originalPrincipal);
+ if (!proxyAuthorized || !originalPrincipalAuthorized) {
+ throw new RestException(Status.UNAUTHORIZED,
+ String.format("Proxy not authorized to access
resource (proxy:%s,original:%s)",
+ clientAppId, originalPrincipal));
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RestException(Status.INTERNAL_SERVER_ERROR,
e.getMessage());
}
log.debug("Successfully authorized {} (proxied by {}) on
tenant {}",
originalPrincipal, clientAppId, tenant);
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
index 1145321..545817f 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
@@ -490,7 +490,7 @@ public class ServerCnxTest {
providerField.setAccessible(true);
PulsarAuthorizationProvider authorizationProvider = spy(new
PulsarAuthorizationProvider(svcConfig, configCacheService));
providerField.set(authorizationService, authorizationProvider);
-
doReturn(false).when(authorizationProvider).isSuperUser(Mockito.anyString());
+
doReturn(CompletableFuture.completedFuture(false)).when(authorizationProvider).isSuperUser(Mockito.anyString(),
Mockito.any());
// Test producer creation
resetChannel();
@@ -520,7 +520,7 @@ public class ServerCnxTest {
providerField.set(authorizationService, authorizationProvider);
doReturn(authorizationService).when(brokerService).getAuthorizationService();
doReturn(true).when(brokerService).isAuthorizationEnabled();
-
doReturn(false).when(authorizationProvider).isSuperUser(Mockito.anyString());
+
doReturn(CompletableFuture.completedFuture(false)).when(authorizationProvider).isSuperUser(Mockito.anyString(),
Mockito.any());
doReturn(CompletableFuture.completedFuture(true)).when(authorizationProvider).checkPermission(any(TopicName.class),
Mockito.anyString(),
any(AuthAction.class));
@@ -548,7 +548,7 @@ public class ServerCnxTest {
providerField.setAccessible(true);
PulsarAuthorizationProvider authorizationProvider = spy(new
PulsarAuthorizationProvider(svcConfig, configCacheService));
providerField.set(authorizationService, authorizationProvider);
-
doReturn(true).when(authorizationProvider).isSuperUser(Mockito.anyString());
+
doReturn(CompletableFuture.completedFuture(true)).when(authorizationProvider).isSuperUser(Mockito.anyString(),
Mockito.any());
// Test producer creation
resetChannel();
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java
index 033c638..72838e8 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java
@@ -402,13 +402,22 @@ public class AuthorizationProducerConsumerTest extends
ProducerConsumerBase {
public static class TestAuthorizationProvider implements
AuthorizationProvider {
+ public ServiceConfiguration conf;
+
@Override
public void close() throws IOException {
// No-op
}
@Override
+ public CompletableFuture<Boolean> isSuperUser(String role,
ServiceConfiguration serviceConfiguration) {
+ Set<String> superUserRoles =
serviceConfiguration.getSuperUserRoles();
+ return CompletableFuture.completedFuture(role != null &&
superUserRoles.contains(role) ? true : false);
+ }
+
+ @Override
public void initialize(ServiceConfiguration conf,
ConfigurationCacheService configCache) throws IOException {
+ this.conf = conf;
// No-op
}
