From 1a6c28dd0072de05a544dbc9243bfbe6bccea5db Mon Sep 17 00:00:00 2001
From: Michael Marshall <mmarshall@apache.org>
Date: Wed, 5 Apr 2023 18:10:55 -0500
Subject: [PATCH] [fix][broker] Only validate superuser access if authz enabled
 (#19989)

### Motivation

In #19455, I added a requirement that only the proxy role could supply an original principal. That check is only supposed to apply when the broker has authorization enabled. However, in one case, that was not the case. This PR does a check and returns early when authorization is not enabled in the broker.

See https://github.com/apache/pulsar/pull/19830#issuecomment-1492262201 for additional motivation.

### Modifications

* Update the `PulsarWebResource#validateSuperUserAccessAsync` to only validate when authentication and authorization are enabled in the configuration.

### Verifying this change

This is a trivial change. It'd be good to add tests, but I didn't include them here because this is a somewhat urgent fix. There was one test that broke because of this change, so there is at least some existing coverage.

### Documentation

- [x] `doc-not-needed`

### Matching PR in forked repository

PR in forked repository: https://github.com/michaeljmarshall/pulsar/pull/39
---
 .../pulsar/broker/web/PulsarWebResource.java  | 29 +++++++------------
 .../admin/v3/AdminApiTransactionTest.java     |  1 +
 2 files changed, 12 insertions(+), 18 deletions(-)

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 c5713ebddaa2f..a182e4733fdfe 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
@@ -185,8 +185,8 @@ protected boolean hasSuperUserAccess() {
         return true;
     }
 
-    public CompletableFuture<Void> validateSuperUserAccessAsync(){
-        if (!config().isAuthenticationEnabled()) {
+    public CompletableFuture<Void> validateSuperUserAccessAsync() {
+        if (!config().isAuthenticationEnabled() || !config().isAuthorizationEnabled()) {
             return CompletableFuture.completedFuture(null);
         }
         String appId = clientAppId();
@@ -221,22 +221,15 @@ public CompletableFuture<Void> validateSuperUserAccessAsync(){
                         }
                     });
         } else {
-            if (config().isAuthorizationEnabled()) {
-                return pulsar.getBrokerService()
-                        .getAuthorizationService()
-                        .isSuperUser(appId, clientAuthData())
-                        .thenAccept(proxyAuthorizationSuccess -> {
-                            if (!proxyAuthorizationSuccess) {
-                                throw new RestException(Status.UNAUTHORIZED,
-                                        "This operation requires super-user access");
-                            }
-                        });
-            }
-            if (log.isDebugEnabled()) {
-                log.debug("Successfully authorized {} as super-user",
-                        appId);
-            }
-            return CompletableFuture.completedFuture(null);
+            return pulsar.getBrokerService()
+                    .getAuthorizationService()
+                    .isSuperUser(appId, clientAuthData())
+                    .thenAccept(proxyAuthorizationSuccess -> {
+                        if (!proxyAuthorizationSuccess) {
+                            throw new RestException(Status.UNAUTHORIZED,
+                                    "This operation requires super-user access");
+                        }
+                    });
         }
     }
 
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java
index 019b7c11fd579..0e51470da75a5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java
@@ -630,6 +630,7 @@ public void testUpdateTransactionCoordinatorNumber() throws Exception {
         Awaitility.await().until(() -> pulsar.getTransactionMetadataStoreService().getStores().size() ==
                         coordinatorSize * 2);
         pulsar.getConfiguration().setAuthenticationEnabled(true);
+        pulsar.getConfiguration().setAuthorizationEnabled(true);
         Set<String> proxyRoles = spy(Set.class);
         doReturn(true).when(proxyRoles).contains(any());
         pulsar.getConfiguration().setProxyRoles(proxyRoles);