[refactor][fn] Use AuthorizationServer more in Function Worker API

conf/functions_worker.yml

@@ -311,6 +311,8 @@ authenticationProviders:
 authorizationProvider: org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider
 # Set of role names that are treated as "super-user", meaning they will be able to access any admin-api
 superUserRoles:
+# Set of role names that are treated as "proxy" roles. These are the roles that can supply the originalPrincipal.
+proxyRoles:
 
 #### tls configuration for worker service
 # Enable TLS

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationParameters.java

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.authentication;
+
+import lombok.Builder;
+
+/**
+ * A class to collect all the common fields used for authentication. Because the authentication data source is
+ * not always consistent when using the Pulsar Protocol and the Pulsar Proxy
+ * (see <a href="https://github.com/apache/pulsar/issues/19332">19332</a>), this class is currently restricted
+ * to use only in authenticating HTTP requests.
+ */
+@Builder
+@lombok.Value
+public class AuthenticationParameters {
+
+    /**
+     * The original principal (or role) of the client.
+     * <p>
+     *     For HTTP Authentication, there are two possibilities. When the client connects directly to the broker, this
+     *     field is null (assuming the client hasn't supplied the original principal header). When the client connects
+     *     through the proxy, the proxy calculates the original principal based on the client's authentication data and
+     *     supplies it in this field.
+     * </p>
+     */
+    String originalPrincipal;
+
+    /**
+     * The client role.
+     * <p>
+     *     For HTTP Authentication, there are three possibilities. When the client connects directly to the broker, this
+     *     is the client's role. When the client connects through the proxy using mTLS, this is the role of the proxy.
+     *     In this case, the {@link #originalPrincipal} is also supplied and is the role of the original client as
+     *     determined by the proxy. When the client connects through the proxy using any other form of authentication,
+     *     this is the role of the original client. In this case, the {@link #originalPrincipal} is also the role of
+     *     the original client.
+     * </p>
+     */
+    String clientRole;
+
+    /**
+     * The authentication data source used to generate the {@link #clientRole}.
+     * <p>
+     *     For HTTP Authentication, there are three possibilities. When the client connects directly to the broker, this
+     *     is the client's {@link AuthenticationDataSource}. When the client connects through the proxy using mTLS, this
+     *     is the proxy's {@link AuthenticationDataSource}. When the client connects through the proxy using any other
+     *     form of authentication, this is the original client's {@link AuthenticationDataSource}.
+     * </p>
+     */
+    AuthenticationDataSource clientAuthenticationDataSource;
+}

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

@@ -23,10 +23,12 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import javax.ws.rs.core.Response;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.broker.PulsarServerException;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
+import org.apache.pulsar.broker.authentication.AuthenticationParameters;
 import org.apache.pulsar.broker.resources.PulsarResources;
 import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.naming.TopicName;
@@ -49,6 +51,7 @@
 public class AuthorizationService {
     private static final Logger log = LoggerFactory.getLogger(AuthorizationService.class);
 
+    private final PulsarResources resources;
     private final AuthorizationProvider provider;
     private final ServiceConfiguration conf;
 
@@ -61,6 +64,7 @@ public AuthorizationService(ServiceConfiguration conf, PulsarResources pulsarRes
                 provider = (AuthorizationProvider) Class.forName(providerClassname)
                         .getDeclaredConstructor().newInstance();
                 provider.initialize(conf, pulsarResources);
+                this.resources = pulsarResources;
                 log.info("{} has been loaded.", providerClassname);
             } else {
                 throw new PulsarServerException("No authorization providers are present.");
@@ -72,6 +76,23 @@ public AuthorizationService(ServiceConfiguration conf, PulsarResources pulsarRes
         }
     }
 
+    public CompletableFuture<Boolean> isSuperUser(AuthenticationParameters authParams) {
+        if (!isValidOriginalPrincipal(authParams)) {
+            return CompletableFuture.completedFuture(false);
+        }
+        if (isProxyRole(authParams.getClientRole())) {
+            CompletableFuture<Boolean> isRoleAuthorizedFuture = isSuperUser(authParams.getClientRole(),
+                    authParams.getClientAuthenticationDataSource());
+            // The current paradigm is to pass the client auth data when we don't have access to the original auth data.
+            CompletableFuture<Boolean> isOriginalAuthorizedFuture = isSuperUser(authParams.getOriginalPrincipal(),
+                    authParams.getClientAuthenticationDataSource());
+            return isRoleAuthorizedFuture.thenCombine(isOriginalAuthorizedFuture,
+                    (isRoleAuthorized, isOriginalAuthorized) -> isRoleAuthorized && isOriginalAuthorized);
+        } else {
+            return isSuperUser(authParams.getClientRole(), authParams.getClientAuthenticationDataSource());
+        }
+    }
+
     public CompletableFuture<Boolean> isSuperUser(String user, AuthenticationDataSource authenticationData) {
         return provider.isSuperUser(user, authenticationData, conf);
     }
@@ -279,17 +300,118 @@ public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String rol
 
     public CompletableFuture<Boolean> allowFunctionOpsAsync(NamespaceName namespaceName, String role,
                                                             AuthenticationDataSource authenticationData) {
-        return provider.allowFunctionOpsAsync(namespaceName, role, authenticationData);
+        return isSuperUserOrAdmin(namespaceName, role, authenticationData)
+                .thenCompose(isSuperUserOrAdmin -> isSuperUserOrAdmin
+                        ? CompletableFuture.completedFuture(true)
+                        : provider.allowFunctionOpsAsync(namespaceName, role, authenticationData)
+                );
+    }
+
+    public CompletableFuture<Boolean> allowFunctionOpsAsync(NamespaceName namespaceName,
+                                                            AuthenticationParameters authParams) {
+        if (!isValidOriginalPrincipal(authParams)) {
+            return CompletableFuture.completedFuture(false);
+        }
+        if (isProxyRole(authParams.getClientRole())) {
+            CompletableFuture<Boolean> isRoleAuthorizedFuture = allowFunctionOpsAsync(namespaceName,
+                    authParams.getClientRole(), authParams.getClientAuthenticationDataSource());
+            // The current paradigm is to pass the client auth data when we don't have access to the original auth data.
+            CompletableFuture<Boolean> isOriginalAuthorizedFuture = allowFunctionOpsAsync(
+                    namespaceName, authParams.getOriginalPrincipal(), authParams.getClientAuthenticationDataSource());
+            return isRoleAuthorizedFuture.thenCombine(isOriginalAuthorizedFuture,
+                    (isRoleAuthorized, isOriginalAuthorized) -> isRoleAuthorized && isOriginalAuthorized);
+        } else {
+            return allowFunctionOpsAsync(namespaceName, authParams.getClientRole(),
+                    authParams.getClientAuthenticationDataSource());
+        }
     }
 
     public CompletableFuture<Boolean> allowSourceOpsAsync(NamespaceName namespaceName, String role,
                                                           AuthenticationDataSource authenticationData) {
-        return provider.allowSourceOpsAsync(namespaceName, role, authenticationData);
+        return isSuperUserOrAdmin(namespaceName, role, authenticationData)
+                .thenCompose(isSuperUserOrAdmin -> isSuperUserOrAdmin
+                        ? CompletableFuture.completedFuture(true)
+                        : provider.allowSourceOpsAsync(namespaceName, role, authenticationData)
+                );
+    }
+
+    public CompletableFuture<Boolean> allowSourceOpsAsync(NamespaceName namespaceName,
+                                                          AuthenticationParameters authParams) {
+        if (!isValidOriginalPrincipal(authParams)) {
+            return CompletableFuture.completedFuture(false);
+        }
+        if (isProxyRole(authParams.getClientRole())) {
+            CompletableFuture<Boolean> isRoleAuthorizedFuture = allowSourceOpsAsync(namespaceName,
+                    authParams.getClientRole(), authParams.getClientAuthenticationDataSource());
+            // The current paradigm is to pass the client auth data when we don't have access to the original auth data.
+            CompletableFuture<Boolean> isOriginalAuthorizedFuture = allowSourceOpsAsync(
+                    namespaceName, authParams.getOriginalPrincipal(), authParams.getClientAuthenticationDataSource());
+            return isRoleAuthorizedFuture.thenCombine(isOriginalAuthorizedFuture,
+                    (isRoleAuthorized, isOriginalAuthorized) -> isRoleAuthorized && isOriginalAuthorized);
+        } else {
+            return allowSourceOpsAsync(namespaceName, authParams.getClientRole(),
+                    authParams.getClientAuthenticationDataSource());
+        }
     }
 
     public CompletableFuture<Boolean> allowSinkOpsAsync(NamespaceName namespaceName, String role,
                                                         AuthenticationDataSource authenticationData) {
-        return provider.allowSinkOpsAsync(namespaceName, role, authenticationData);
+        return isSuperUserOrAdmin(namespaceName, role, authenticationData)
+                .thenCompose(isSuperUserOrAdmin -> isSuperUserOrAdmin
+                        ? CompletableFuture.completedFuture(true)
+                        : provider.allowSinkOpsAsync(namespaceName, role, authenticationData)
+                );
+    }
+
+    public CompletableFuture<Boolean> allowSinkOpsAsync(NamespaceName namespaceName,
+                                                        AuthenticationParameters authParams) {
+        if (!isValidOriginalPrincipal(authParams)) {
+            return CompletableFuture.completedFuture(false);
+        }
+        if (isProxyRole(authParams.getClientRole())) {
+            CompletableFuture<Boolean> isRoleAuthorizedFuture = allowSinkOpsAsync(namespaceName,
+                    authParams.getClientRole(), authParams.getClientAuthenticationDataSource());
+            // The current paradigm is to pass the client auth data when we don't have access to the original auth data.
+            CompletableFuture<Boolean> isOriginalAuthorizedFuture = allowSinkOpsAsync(
+                    namespaceName, authParams.getOriginalPrincipal(), authParams.getClientAuthenticationDataSource());
+            return isRoleAuthorizedFuture.thenCombine(isOriginalAuthorizedFuture,
+                    (isRoleAuthorized, isOriginalAuthorized) -> isRoleAuthorized && isOriginalAuthorized);
+        } else {
+            return allowSinkOpsAsync(namespaceName, authParams.getClientRole(),
+                    authParams.getClientAuthenticationDataSource());
+        }
+    }
+
+    /**
+     * Functions, sources, and sinks each have their own method in this class. This method first checks for
+     * tenant admin access, then for namespace level permission.
+     */
+    private CompletableFuture<Boolean> isSuperUserOrAdmin(NamespaceName namespaceName,
+                                                          String role,
+                                                          AuthenticationDataSource authenticationData) {
+        return isSuperUser(role, authenticationData)
+                .thenCompose(isSuperUserOrAdmin -> isSuperUserOrAdmin
+                        ? CompletableFuture.completedFuture(true)
+                        : isTenantAdmin(namespaceName.getTenant(), role, authenticationData));
+    }
+
+    private CompletableFuture<Boolean> isTenantAdmin(String tenant, String role,
+                                                    AuthenticationDataSource authData) {
+        return resources.getTenantResources()
+                .getTenantAsync(tenant)
+                .thenCompose(op -> {
+                    if (op.isPresent()) {
+                        return isTenantAdmin(tenant, role, op.get(), authData);
+                    } else {
+                        return CompletableFuture.failedFuture(new RestException(Response.Status.NOT_FOUND,
+                                "Tenant does not exist"));
+                    }
+                });
+    }
+
+    private boolean isValidOriginalPrincipal(AuthenticationParameters authParams) {
+        return isValidOriginalPrincipal(authParams.getClientRole(),
+                authParams.getOriginalPrincipal(), authParams.getClientAuthenticationDataSource());
     }
 
     /**

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -1860,6 +1860,7 @@ public static WorkerConfig initializeWorkerConfigFromBrokerConfig(ServiceConfigu
 
         // inherit super users
         workerConfig.setSuperUserRoles(brokerConfig.getSuperUserRoles());
+        workerConfig.setProxyRoles(brokerConfig.getProxyRoles());
         workerConfig.setFunctionsWorkerEnablePackageManagement(brokerConfig.isFunctionsWorkerEnablePackageManagement());
 
         // inherit the nar package locations

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

@@ -159,6 +159,7 @@ public LoadManagerReport getLoadReport() throws Exception {
 
     protected Map<Long, Collection<ResourceUnit>> internalBrokerResourceAvailability(NamespaceName namespace) {
         try {
+            validateSuperUserAccess();
             LoadManager lm = pulsar().getLoadManager().get();
             if (lm instanceof SimpleLoadManagerImpl) {
                 return ((SimpleLoadManagerImpl) lm).getResourceAvailabilityFor(namespace).asMap();