Add auth action for package management service

.github/workflows/ci-integration-cli.yaml

@@ -88,6 +88,11 @@ jobs:
         if: steps.docs.outputs.changed_only == 'no'
         run: mvn -B -f tests/docker-images/pom.xml install -am -Pdocker -DskipTests
 
-      - name: run integration tests
+      - name: run pulsar cli integration tests
         if: steps.docs.outputs.changed_only == 'no'
         run: mvn -B -f tests/pom.xml test -DintegrationTestSuiteFile=pulsar-cli.xml -DintegrationTests -DredirectTestOutputToFile=false
+
+      - name: run pulsar auth integration tests
+        if: steps.docs.outputs.changed_only == 'no'
+        run: mvn -B -f tests/pom.xml test -DintegrationTestSuiteFile=pulsar-auth.xml -DintegrationTests -DredirectTestOutputToFile=false
+

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

@@ -219,20 +219,20 @@ public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String rol
 
     @Override
     public CompletableFuture<Boolean> allowFunctionOpsAsync(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData) {
-        return allowFunctionSourceSinkOpsAsync(namespaceName, role, authenticationData, AuthAction.functions);
+        return allowTheSpecifiedActionOpsAsync(namespaceName, role, authenticationData, AuthAction.functions);
     }
 
     @Override
     public CompletableFuture<Boolean> allowSourceOpsAsync(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData) {
-        return allowFunctionSourceSinkOpsAsync(namespaceName, role, authenticationData, AuthAction.sources);
+        return allowTheSpecifiedActionOpsAsync(namespaceName, role, authenticationData, AuthAction.sources);
     }
 
     @Override
     public CompletableFuture<Boolean> allowSinkOpsAsync(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData) {
-        return allowFunctionSourceSinkOpsAsync(namespaceName, role, authenticationData, AuthAction.sinks);
+        return allowTheSpecifiedActionOpsAsync(namespaceName, role, authenticationData, AuthAction.sinks);
     }
 
-    private CompletableFuture<Boolean> allowFunctionSourceSinkOpsAsync(NamespaceName namespaceName, String role,
+    private CompletableFuture<Boolean> allowTheSpecifiedActionOpsAsync(NamespaceName namespaceName, String role,
                                                                        AuthenticationDataSource authenticationData,
                                                                        AuthAction authAction) {
         CompletableFuture<Boolean> permissionFuture = new CompletableFuture<>();
@@ -538,7 +538,21 @@ public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName nam
                                                                    String role,
                                                                    NamespaceOperation operation,
                                                                    AuthenticationDataSource authData) {
-        return validateTenantAdminAccess(namespaceName.getTenant(), role, authData);
+        CompletableFuture<Boolean> isAuthorizedFuture;
+        if (operation == NamespaceOperation.PACKAGES) {
+            isAuthorizedFuture = allowTheSpecifiedActionOpsAsync(namespaceName, role, authData, AuthAction.packages);
+        } else {
+            isAuthorizedFuture = CompletableFuture.completedFuture(false);
+        }
+        CompletableFuture<Boolean> isTenantAdminFuture = validateTenantAdminAccess(namespaceName.getTenant(), role, authData);
+        return isTenantAdminFuture.thenCombine(isAuthorizedFuture, (isTenantAdmin, isAuthorized) -> {
+            if (log.isDebugEnabled()) {
+                log.debug("Verify if role {} is allowed to {} to topic {}:"
+                        + " isTenantAdmin={}, isAuthorized={}",
+                    role, operation, namespaceName, isTenantAdmin, isAuthorized);
+            }
+            return isTenantAdmin || isAuthorized;
+        });
     }
 
     @Override

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

@@ -21,21 +21,27 @@
 import java.io.InputStream;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import javax.ws.rs.WebApplicationException;
 import javax.ws.rs.container.AsyncResponse;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.StreamingOutput;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.admin.AdminResource;
+import org.apache.pulsar.broker.authorization.AuthorizationService;
 import org.apache.pulsar.broker.web.RestException;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.apache.pulsar.common.policies.data.NamespaceOperation;
 import org.apache.pulsar.packages.management.core.PackagesManagement;
 import org.apache.pulsar.packages.management.core.common.PackageMetadata;
 import org.apache.pulsar.packages.management.core.common.PackageName;
 import org.apache.pulsar.packages.management.core.common.PackageType;
 import org.apache.pulsar.packages.management.core.exceptions.PackagesManagementException;
 
-
 @Slf4j
 public class PackagesBase extends AdminResource {
+
+    private AuthorizationService authorizationService;
+
     private PackagesManagement getPackagesManagement() {
         return pulsar().getPackagesManagement();
     }
@@ -57,30 +63,46 @@ private Void handleError(Throwable throwable, AsyncResponse asyncResponse) {
             asyncResponse.resume(new RestException(Response.Status.PRECONDITION_FAILED, throwable.getMessage()));
         } else if (throwable instanceof PackagesManagementException.NotFoundException) {
             asyncResponse.resume(new RestException(Response.Status.NOT_FOUND, throwable.getMessage()));
+        } else if (throwable instanceof WebApplicationException) {
+            asyncResponse.resume(throwable);
         } else {
+            log.error("Encountered unexpected error", throwable);
             asyncResponse.resume(new RestException(Response.Status.INTERNAL_SERVER_ERROR, throwable.getMessage()));
         }
         return null;
     }
 
     protected void internalGetMetadata(String type, String tenant, String namespace, String packageName,
                                                   String version, AsyncResponse asyncResponse) {
-        getPackageNameAsync(type, tenant, namespace, packageName, version)
+        checkPermissions(tenant, namespace)
+            .thenCompose(ignore -> getPackageNameAsync(type, tenant, namespace, packageName, version))
             .thenCompose(name -> getPackagesManagement().getMeta(name))
             .thenAccept(asyncResponse::resume)
             .exceptionally(e -> handleError(e.getCause(), asyncResponse));
     }
 
     protected void internalUpdateMetadata(String type, String tenant, String namespace, String packageName,
                                           String version, PackageMetadata metadata, AsyncResponse asyncResponse) {
-        getPackageNameAsync(type, tenant, namespace, packageName, version)
+        checkPermissions(tenant, namespace)
+            .thenCompose(ignore -> getPackageNameAsync(type, tenant, namespace, packageName, version))
             .thenCompose(name -> getPackagesManagement().updateMeta(name, metadata))
             .thenAccept(ignore -> asyncResponse.resume(Response.noContent().build()))
             .exceptionally(e -> handleError(e.getCause(), asyncResponse));
     }
 
     protected StreamingOutput internalDownload(String type, String tenant, String namespace,
                                                String packageName, String version) {
+        try {
+            checkPermissions(tenant, namespace).get();
+        } catch (InterruptedException e) {
+            throw new RestException(Response.Status.INTERNAL_SERVER_ERROR, e.getMessage());
+        } catch (ExecutionException e) {
+            if (e.getCause() instanceof WebApplicationException) {
+                throw (WebApplicationException) e.getCause();
+            } else {
+                throw new RestException(Response.Status.INTERNAL_SERVER_ERROR, e.getCause().getMessage());
+            }
+        }
         try {
             PackageName name = PackageName.get(type, tenant, namespace, packageName, version);
             return output -> {
@@ -104,23 +126,26 @@ protected StreamingOutput internalDownload(String type, String tenant, String na
     protected void internalUpload(String type, String tenant, String namespace, String packageName, String version,
                                   PackageMetadata metadata, InputStream uploadedInputStream,
                                   AsyncResponse asyncResponse) {
-        getPackageNameAsync(type, tenant, namespace, packageName, version)
+        checkPermissions(tenant, namespace)
+            .thenCompose(ignore -> getPackageNameAsync(type, tenant, namespace, packageName, version))
             .thenCompose(name -> getPackagesManagement().upload(name, metadata, uploadedInputStream))
             .thenAccept(ignore -> asyncResponse.resume(Response.noContent().build()))
             .exceptionally(e -> handleError(e.getCause(), asyncResponse));
     }
 
     protected void internalDelete(String type, String tenant, String namespace, String packageName, String version,
                                   AsyncResponse asyncResponse) {
-        getPackageNameAsync(type, tenant, namespace, packageName, version)
+        checkPermissions(tenant, namespace)
+            .thenCompose(ignore -> getPackageNameAsync(type, tenant, namespace, packageName, version))
             .thenCompose(name -> getPackagesManagement().delete(name))
             .thenAccept(ignore -> asyncResponse.resume(Response.noContent().build()))
             .exceptionally(e -> handleError(e.getCause(), asyncResponse));
     }
 
     protected void internalListVersions(String type, String tenant, String namespace, String packageName,
                                                      AsyncResponse asyncResponse) {
-        getPackageNameAsync(type, tenant, namespace, packageName, "")
+        checkPermissions(tenant, namespace)
+            .thenCompose(ignore -> getPackageNameAsync(type, tenant, namespace, packageName, ""))
             .thenCompose(name -> getPackagesManagement().list(name))
             .thenAccept(asyncResponse::resume)
             .exceptionally(e -> handleError(e.getCause(), asyncResponse));
@@ -129,11 +154,51 @@ protected void internalListVersions(String type, String tenant, String namespace
     protected void internalListPackages(String type, String tenant, String namespace, AsyncResponse asyncResponse) {
         try {
             PackageType packageType = PackageType.getEnum(type);
-            getPackagesManagement().list(packageType, tenant, namespace)
+            checkPermissions(tenant, namespace)
+                .thenCompose(ignore -> getPackagesManagement().list(packageType, tenant, namespace))
                 .thenAccept(asyncResponse::resume)
                 .exceptionally(e -> handleError(e.getCause(), asyncResponse));
         } catch (IllegalArgumentException iae) {
             asyncResponse.resume(new RestException(Response.Status.PRECONDITION_FAILED, iae.getMessage()));
         }
     }
+
+    private CompletableFuture<Void> checkPermissions(String tenant, String namespace) {
+        CompletableFuture<Void> future = new CompletableFuture<>();
+        if (config().isAuthenticationEnabled()) {
+            NamespaceName namespaceName;
+            try {
+                namespaceName = NamespaceName.get(tenant, namespace);
+            } catch (Exception e) {
+                future.completeExceptionally(e);
+                return future;
+            }
+            getAuthorizationService()
+                .allowNamespaceOperationAsync(namespaceName, NamespaceOperation.PACKAGES, originalPrincipal(),
+                    clientAppId(), clientAuthData())
+                .whenComplete((hasPermission, throwable) -> {
+                    if (throwable != null) {
+                        future.completeExceptionally(throwable);
+                        return;
+                    }
+                    if (hasPermission) {
+                        future.complete(null);
+                    } else {
+                        future.completeExceptionally(new RestException(Response.Status.UNAUTHORIZED, String.format(
+                            "Role %s has not the 'package' permission to do the packages operations.", clientAppId())));
+                    }
+                });
+        } else {
+            future.complete(null);
+        }
+        return future;
+    }
+
+    private AuthorizationService getAuthorizationService() {
+        if (authorizationService == null) {
+            authorizationService = pulsar().getBrokerService().getAuthorizationService();
+            return authorizationService;
+        }
+        return authorizationService;
+    }
 }

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/AuthAction.java

@@ -36,4 +36,7 @@ public enum AuthAction {
 
     /** Permissions for sinks ops. **/
     sinks,
+
+    /** Permissions for packages ops. **/
+    packages
 }

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/NamespaceOperation.java

@@ -38,4 +38,6 @@ public enum NamespaceOperation {
 
     CLEAR_BACKLOG,
     UNSUBSCRIBE,
+
+    PACKAGES,
 }