Allow subscribers to access subscription admin-api

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationProvider.java

@@ -101,6 +101,29 @@ CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role,
     CompletableFuture<Void> grantPermissionAsync(NamespaceName namespace, Set<AuthAction> actions, String role,
             String authDataJson);
 
+    /**
+     * Grant permission to roles that can access subscription-admin api
+     * 
+     * @param namespace
+     * @param subscriptionName
+     * @param roles
+     * @param authDataJson
+     *            additional authdata in json format
+     * @return
+     */
+    CompletableFuture<Void> grantSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName, Set<String> roles,
+            String authDataJson);
+
+    /**
+     * Revoke subscription admin-api access for a role
+     * @param namespace
+     * @param subscriptionName
+     * @param role
+     * @return
+     */
+    CompletableFuture<Void> revokeSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName,
+            String role, String authDataJson);
+
     /**
      * Grant authorization-action permission on a topic to the given client
      *

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java

@@ -94,6 +94,41 @@ public CompletableFuture<Void> grantPermissionAsync(NamespaceName namespace, Set
         return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured"));
     }
 
+    /**
+     * Grant permission to roles that can access subscription-admin api
+     * 
+     * @param namespace
+     * @param subscriptionName
+     * @param roles
+     * @param authDataJson
+     *            additional authdata in json for targeted authorization provider
+     * @return
+     */
+    public CompletableFuture<Void> grantSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName,
+            Set<String> roles, String authDataJson) {
+
+        if (provider != null) {
+            return provider.grantSubscriptionPermissionAsync(namespace, subscriptionName, roles, authDataJson);
+        }
+        return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured"));
+    }
+
+    /**
+     * Revoke subscription admin-api access for a role
+     * 
+     * @param namespace
+     * @param subscriptionName
+     * @param role
+     * @return
+     */
+    public CompletableFuture<Void> revokeSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName,
+            String role, String authDataJson) {
+        if (provider != null) {
+            return provider.revokeSubscriptionPermissionAsync(namespace, subscriptionName, role, authDataJson);
+        }
+        return FutureUtil.failedFuture(new IllegalStateException("No authorization provider configured"));
+    }
+
     /**
      * Grant authorization-action permission on a topic to the given client
      *

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java

@@ -23,6 +23,7 @@
 import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -44,6 +45,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.collect.Sets;
+
 /**
  * Default authorization provider that stores authorization policies under local-zookeeper.
  *
@@ -110,6 +113,19 @@ public CompletableFuture<Boolean> canConsumeAsync(TopicName topicName, String ro
                     }
                 } else {
                     if (isNotBlank(subscription) && !isSuperUser(role)) {
+                        // 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);
+                        if (roles != null && !roles.isEmpty() && !roles.contains(role)) {
+                            log.warn("[{}] is not authorized to subscribe on {}-{}", role, topicName, subscription);
+                            PulsarServerException ex = new PulsarServerException(
+                                    String.format("%s is not authorized to access subscription %s on topic %s", role,
+                                            subscription, topicName));
+                            permissionFuture.complete(false);
+                            return;
+                        }
+
+                        // validate if subscription-auth mode is configured
                         switch (policies.get().subscription_auth_mode) {
                         case Prefix:
                             if (!subscription.startsWith(role)) {
@@ -125,6 +141,7 @@ public CompletableFuture<Boolean> canConsumeAsync(TopicName topicName, String ro
                         }
                     }
                 }
+                // check namespace and topic level consume-permissions
                 checkAuthorization(topicName, role, AuthAction.consume).thenAccept(isAuthorized -> {
                     permissionFuture.complete(isAuthorized);
                 });
@@ -241,6 +258,70 @@ public CompletableFuture<Void> grantPermissionAsync(NamespaceName namespaceName,
         return result;
     }
 
+    @Override
+    public CompletableFuture<Void> grantSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName,
+            Set<String> roles, String authDataJson) {
+        return updateSubscriptionPermissionAsync(namespace, subscriptionName, roles, false);
+    }
+
+    @Override
+    public CompletableFuture<Void> revokeSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName,
+            String role, String authDataJson) {
+        return updateSubscriptionPermissionAsync(namespace, subscriptionName, Collections.singleton(role), true);
+    }
+
+    private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName namespace, String subscriptionName, Set<String> roles,
+            boolean remove) {
+        CompletableFuture<Void> result = new CompletableFuture<>();
+
+        try {
+            validatePoliciesReadOnlyAccess();
+        } catch (Exception e) {
+            result.completeExceptionally(e);
+        }
+
+        ZooKeeper globalZk = configCache.getZooKeeper();
+        final String policiesPath = String.format("/%s/%s/%s", "admin", POLICIES, namespace.toString());
+
+        try {
+            Stat nodeStat = new Stat();
+            byte[] content = globalZk.getData(policiesPath, null, nodeStat);
+            Policies policies = getThreadLocal().readValue(content, Policies.class);
+            if (remove) {
+                if (policies.auth_policies.subscription_auth_roles.get(subscriptionName) != null) {
+                    policies.auth_policies.subscription_auth_roles.get(subscriptionName).removeAll(roles);
+                }else {
+                    log.info("[{}] Couldn't find role {} while revoking for sub = {}", namespace, subscriptionName, roles);
+                    result.completeExceptionally(new IllegalArgumentException("couldn't find subscription"));
+                    return result;
+                }
+            } else {
+                policies.auth_policies.subscription_auth_roles.put(subscriptionName, roles);
+            }
+
+            // Write back the new policies into zookeeper
+            globalZk.setData(policiesPath, getThreadLocal().writeValueAsBytes(policies), nodeStat.getVersion());
+
+            configCache.policiesCache().invalidate(policiesPath);
+
+            log.info("[{}] Successfully granted access for role {} for sub = {}", namespace, subscriptionName, roles);
+            result.complete(null);
+        } catch (KeeperException.NoNodeException e) {
+            log.warn("[{}] Failed to set permissions for namespace {}: does not exist", subscriptionName, namespace);
+            result.completeExceptionally(new IllegalArgumentException("Namespace does not exist" + namespace));
+        } catch (KeeperException.BadVersionException e) {
+            log.warn("[{}] Failed to set permissions for {} on namespace {}: concurrent modification", subscriptionName, roles, namespace);
+            result.completeExceptionally(new IllegalStateException(
+                    "Concurrent modification on zk path: " + policiesPath + ", " + e.getMessage()));
+        } catch (Exception e) {
+            log.error("[{}] Failed to get permissions for role {} on namespace {}", subscriptionName, roles, namespace, e);
+            result.completeExceptionally(
+                    new IllegalStateException("Failed to get permissions for namespace " + namespace));
+        }
+
+        return result;
+    }
+
     private CompletableFuture<Boolean> checkAuthorization(TopicName topicName, String role, AuthAction action) {
         if (isSuperUser(role)) {
             return CompletableFuture.completedFuture(true);

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java

@@ -329,6 +329,38 @@ protected void internalGrantPermissionOnNamespace(String role, Set<AuthAction> a
         }
     }
 
+
+    protected void internalGrantPermissionOnSubscription(String subscription, Set<String> roles) {
+        /** controlled by system-admin(super-user) to prevent metadata footprint size */
+        validateSuperUserAccess();
+
+        try {
+            AuthorizationService authService = pulsar().getBrokerService().getAuthorizationService();
+            if (null != authService) {
+                authService.grantSubscriptionPermissionAsync(namespaceName, subscription, roles,
+                        null/* additional auth-data json */).get();
+            } else {
+                throw new RestException(Status.NOT_IMPLEMENTED, "Authorization is not enabled");
+            }
+        } catch (InterruptedException e) {
+            log.error("[{}] Failed to get permissions for namespace {}", clientAppId(), namespaceName, e);
+            throw new RestException(e);
+        } catch (ExecutionException e) {
+            if (e.getCause() instanceof IllegalArgumentException) {
+                log.warn("[{}] Failed to set permissions for namespace {}: does not exist", clientAppId(),
+                        namespaceName);
+                throw new RestException(Status.NOT_FOUND, "Namespace does not exist");
+            } else if (e.getCause() instanceof IllegalStateException) {
+                log.warn("[{}] Failed to set permissions for namespace {}: concurrent modification", clientAppId(),
+                        namespaceName);
+                throw new RestException(Status.CONFLICT, "Concurrent modification");
+            } else {
+                log.error("[{}] Failed to get permissions for namespace {}", clientAppId(), namespaceName, e);
+                throw new RestException(e);
+            }
+        }
+    }
+
     protected void internalRevokePermissionsOnNamespace(String role) {
         validateAdminAccessForTenant(namespaceName.getTenant());
         validatePoliciesReadOnlyAccess();
@@ -359,6 +391,19 @@ protected void internalRevokePermissionsOnNamespace(String role) {
         }
     }
 
+    protected void internalRevokePermissionsOnSubscription(String subscriptionName, String role) {
+        validateAdminAccessForTenant(namespaceName.getTenant());
+        validatePoliciesReadOnlyAccess();
+
+        AuthorizationService authService = pulsar().getBrokerService().getAuthorizationService();
+        if (null != authService) {
+            authService.revokeSubscriptionPermissionAsync(namespaceName, subscriptionName, role,
+                    null/* additional auth-data json */);
+        } else {
+            throw new RestException(Status.NOT_IMPLEMENTED, "Authorization is not enabled");
+        }
+    }
+
     protected Set<String> internalGetNamespaceReplicationClusters() {
         if (!namespaceName.isGlobal()) {
             throw new RestException(Status.PRECONDITION_FAILED,

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -235,6 +235,36 @@ public void validateAdminOperationOnTopic(boolean authoritative) {
         validateTopicOwnership(topicName, authoritative);
     }
 
+    protected void validateAdminAccessForSubscriber(String subscriptionName, boolean authoritative) {
+        validateTopicOwnership(topicName, authoritative);
+        try {
+            validateAdminAccessForTenant(topicName.getTenant());
+        } catch (Exception e) {
+            if (log.isDebugEnabled()) {
+                log.debug("[{}] failed to validate admin access for {}", topicName, clientAppId());
+            }
+            validateAdminAccessForSubscriber(subscriptionName);
+        }
+    }
+
+    private void validateAdminAccessForSubscriber(String subscriptionName) {
+        try {
+            if (!pulsar().getBrokerService().getAuthorizationService().canConsume(topicName, clientAppId(),
+                    clientAuthData(), subscriptionName)) {
+                log.warn("[{}} Subscriber {} is not authorized to access api", topicName, clientAppId());
+                throw new RestException(Status.UNAUTHORIZED,
+                        String.format("Subscriber %s is not authorized to access this operation", clientAppId()));
+            }
+        } catch (RestException re) {
+            throw re;
+        } catch (Exception e) {
+            // unknown error marked as internal server error
+            log.warn("Unexpected error while authorizing request. topic={}, role={}. Error: {}", topicName,
+                    clientAppId(), e.getMessage(), e);
+            throw new RestException(e);
+        }
+    }
+
     protected void internalGrantPermissionsOnTopic(String role, Set<AuthAction> actions) {
         // This operation should be reading from zookeeper and it should be allowed without having admin privileges
         validateAdminAccessForTenant(namespaceName.getTenant());
@@ -654,7 +684,7 @@ protected void internalDeleteSubscription(String subName, boolean authoritative)
                 }
             }
         } else {
-            validateAdminOperationOnTopic(authoritative);
+            validateAdminAccessForSubscriber(subName, authoritative);
             Topic topic = getTopicReference(topicName);
             try {
                 Subscription sub = topic.getSubscription(subName);
@@ -690,7 +720,7 @@ protected void internalSkipAllMessages(String subName, boolean authoritative) {
                 throw new RestException(e);
             }
         } else {
-            validateAdminOperationOnTopic(authoritative);
+            validateAdminAccessForSubscriber(subName, authoritative);
             PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
             try {
                 if (subName.startsWith(topic.replicatorPrefix)) {
@@ -721,7 +751,7 @@ protected void internalSkipMessages(String subName, int numMessages, boolean aut
         if (partitionMetadata.partitions > 0) {
             throw new RestException(Status.METHOD_NOT_ALLOWED, "Skip messages on a partitioned topic is not allowed");
         }
-        validateAdminOperationOnTopic(authoritative);
+        validateAdminAccessForSubscriber(subName, authoritative);
         PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
         try {
             if (subName.startsWith(topic.replicatorPrefix)) {
@@ -810,7 +840,7 @@ protected void internalResetCursor(String subName, long timestamp, boolean autho
             }
 
         } else {
-            validateAdminOperationOnTopic(authoritative);
+            validateAdminAccessForSubscriber(subName, authoritative);
             log.info("[{}][{}] received reset cursor on subscription {} to time {}", clientAppId(), topicName,
                     subName, timestamp);
             PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
@@ -883,7 +913,7 @@ protected void internalCreateSubscription(String subscriptionName, MessageIdImpl
                     throw exception.get();
                 }
             } else {
-                validateAdminOperationOnTopic(authoritative);
+                validateAdminAccessForSubscriber(subscriptionName, authoritative);
 
                 PersistentTopic topic = (PersistentTopic) getOrCreateTopic(topicName);
 
@@ -925,7 +955,7 @@ protected void internalResetCursorOnPosition(String subName, boolean authoritati
             throw new RestException(Status.METHOD_NOT_ALLOWED,
                     "Reset-cursor at position is not allowed for partitioned-topic");
         } else {
-            validateAdminOperationOnTopic(authoritative);
+            validateAdminAccessForSubscriber(subName, authoritative);
             PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
             if (topic == null) {
                 throw new RestException(Status.NOT_FOUND, "Topic not found");
@@ -960,7 +990,7 @@ protected Response internalPeekNthMessage(String subName, int messagePosition, b
         if (partitionMetadata.partitions > 0) {
             throw new RestException(Status.METHOD_NOT_ALLOWED, "Peek messages on a partitioned topic is not allowed");
         }
-        validateAdminOperationOnTopic(authoritative);
+        validateAdminAccessForSubscriber(subName, authoritative);
         if (!(getTopicReference(topicName) instanceof PersistentTopic)) {
             log.error("[{}] Not supported operation of non-persistent topic {} {}", clientAppId(), topicName,
                     subName);
@@ -1112,7 +1142,7 @@ protected void internalExpireMessages(String subName, int expireTimeInSeconds, b
             }
         } else {
             // validate ownership and redirect if current broker is not owner
-            validateAdminOperationOnTopic(authoritative);
+            validateAdminAccessForSubscriber(subName, authoritative);
             if (!(getTopicReference(topicName) instanceof PersistentTopic)) {
                 log.error("[{}] Not supported operation of non-persistent topic {} {}", clientAppId(), topicName,
                         subName);

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/Namespaces.java

@@ -226,6 +226,19 @@ public void grantPermissionOnNamespace(@PathParam("property") String property, @
         internalGrantPermissionOnNamespace(role, actions);
     }
 
+    @POST
+    @Path("/{property}/{cluster}/{namespace}/permissions/subscription/{subscription}")
+    @ApiOperation(hidden = true, value = "Grant a new permission to roles for a subscription.")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Property or cluster or namespace doesn't exist"),
+            @ApiResponse(code = 409, message = "Concurrent modification"),
+            @ApiResponse(code = 501, message = "Authorization is not enabled")})
+    public void grantPermissionOnSubscription(@PathParam("property") String property, @PathParam("cluster") String cluster,
+            @PathParam("namespace") String namespace, @PathParam("subscription") String subscription, Set<String> roles) {
+        validateNamespaceName(property, cluster, namespace);
+        internalGrantPermissionOnSubscription(subscription, roles);
+    }
+
     @DELETE
     @Path("/{property}/{cluster}/{namespace}/permissions/{role}")
     @ApiOperation(hidden = true, value = "Revoke all permissions to a role on a namespace.")
@@ -238,6 +251,18 @@ public void revokePermissionsOnNamespace(@PathParam("property") String property,
         internalRevokePermissionsOnNamespace(role);
     }
 
+    @DELETE
+    @Path("/{property}/{cluster}/{namespace}/permissions/{subscription}/{role}")
+    @ApiOperation(hidden = true, value = "Revoke subscription admin-api access permission for a role.")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Property or cluster or namespace doesn't exist") })
+    public void revokePermissionOnSubscription(@PathParam("property") String property,
+            @PathParam("cluster") String cluster, @PathParam("namespace") String namespace,
+            @PathParam("subscription") String subscription, @PathParam("role") String role) {
+        validateNamespaceName(property, cluster, namespace);
+        internalRevokePermissionsOnSubscription(subscription, role);
+    }
+
     @GET
     @Path("/{property}/{cluster}/{namespace}/replication")
     @ApiOperation(hidden = true, value = "Get the replication clusters for a namespace.", response = String.class, responseContainer = "List")