[fix][broker] Allow proxy to pass same role for authRole and original…

…Role (apache#19557)

### Motivation

I broke the Pulsar Proxy with apache#19455 because that PR requires that when `X-Original-Principal` is supplied, the auth role must be a proxy role. This is not always the case for proxied admin requests. This PR seeks to fix that incorrect assumption by changing the way verification is done for the roles. Specifically, when the two roles are the same and they are not a proxy role, we will consider it a valid combination.

Note that there is no inefficiency in this solution because When the `authenticatedPrincipal` is not a proxy role, that is the only role that is authenticated. Note also that we do not let the binary protocol authenticate this way, and that is consistent with the way the pulsar proxy forwards authentication data.

Currently, we do the following when authentication is enabled in the proxy:

1. Authenticate the client's http request and put the resulting role in the `X-Original-Principal` header for the call to the broker.
https://github.com/apache/pulsar/blob/38555851359f9cfc172650c387a58c5a03809e97/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java#L370-L373

2. Copy the `Authorization` header into the broker's http request:
https://github.com/apache/pulsar/blob/38555851359f9cfc172650c387a58c5a03809e97/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java#L232-L236

3. Configure the proxy's http client to use client TLS authentication (when configured):
https://github.com/apache/pulsar/blob/38555851359f9cfc172650c387a58c5a03809e97/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java#L269-L277

The problem with apache#19455 is that it assumes the proxy supplies its own authentication data. However, that only happens when using TLS authentication. Otherwise, the proxy forwards the client's authentication data in the `Authorization` header. As such, calls will fail because the `X-Original-Principal` header supplied without using a proxy role.

### Modifications

* Consider the `authenticatedPrincipal` and the `originalPrincipal` a valid pair when they are equal and are not a `proxyRole` for http requests.

### Alternative Solutions

I initially proposed that we only add the `X-Original-Principal` when we are using the proxy's authentication (see the first commit). I decided this solution is not ideal because it doesn't solve the problem, it doesn't make the brokers backwards compatible, and there isn't actually any inefficiency in passing the role as a header.

### Verifying this change

When cherry-picking apache#19455 to branch-2.9, I discovered that `PackagesOpsWithAuthTest#testPackagesOps` was consistently failing because of the way the proxy supplies authentication data when proxying http requests. That test was removed by apache#12771, which explains why I didn't catch the error sooner. This PR includes a test that fails  without this change.

Note that the primary issue must be that we didn't have any tests doing authentication forwarding through the proxy. Now we will have both relevant tests where the proxy is and is not authenticating.

### Does this pull request potentially affect one of the following parts:

This is not a breaking change.

### Documentation

- [x] `doc-required`

### Matching PR in forked repository

PR in forked repository: michaeljmarshall#31

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

@@ -292,31 +292,45 @@ public CompletableFuture<Boolean> allowSinkOpsAsync(NamespaceName namespaceName,
         return provider.allowSinkOpsAsync(namespaceName, role, authenticationData);
     }
 
+    /**
+     * Whether the authenticatedPrincipal and the originalPrincipal form a valid pair. This method assumes that
+     * authenticatedPrincipal and originalPrincipal can be equal, as long as they are not a proxy role. This use
+     * case is relvant for the admin server because of the way the proxy handles authentication. The binary protocol
+     * should not use this method.
+     * @return true when roles are a valid combination and false when roles are an invalid combination
+     */
     public boolean isValidOriginalPrincipal(String authenticatedPrincipal,
                                             String originalPrincipal,
                                             AuthenticationDataSource authDataSource) {
         SocketAddress remoteAddress = authDataSource != null ? authDataSource.getPeerAddress() : null;
-        return isValidOriginalPrincipal(authenticatedPrincipal, originalPrincipal, remoteAddress);
+        return isValidOriginalPrincipal(authenticatedPrincipal, originalPrincipal, remoteAddress, true);
     }
 
     /**
      * Validates that the authenticatedPrincipal and the originalPrincipal are a valid combination.
-     * Valid combinations fulfill the following rule: the authenticatedPrincipal is in
-     * {@link ServiceConfiguration#getProxyRoles()}, if, and only if, the originalPrincipal is set to a role
-     * that is not also in {@link ServiceConfiguration#getProxyRoles()}.
+     * Valid combinations fulfill one of the following two rules:
+     * <p>
+     * 1. The authenticatedPrincipal is in {@link ServiceConfiguration#getProxyRoles()}, if, and only if,
+     * the originalPrincipal is set to a role that is not also in {@link ServiceConfiguration#getProxyRoles()}.
+     * <p>
+     * 2. The authenticatedPrincipal and the originalPrincipal are the same, but are not a proxyRole, when
+     * allowNonProxyPrincipalsToBeEqual is true.
+     *
      * @return true when roles are a valid combination and false when roles are an invalid combination
      */
     public boolean isValidOriginalPrincipal(String authenticatedPrincipal,
                                             String originalPrincipal,
-                                            SocketAddress remoteAddress) {
+                                            SocketAddress remoteAddress,
+                                            boolean allowNonProxyPrincipalsToBeEqual) {
         String errorMsg = null;
         if (conf.getProxyRoles().contains(authenticatedPrincipal)) {
             if (StringUtils.isBlank(originalPrincipal)) {
                 errorMsg = "originalPrincipal must be provided when connecting with a proxy role.";
             } else if (conf.getProxyRoles().contains(originalPrincipal)) {
                 errorMsg = "originalPrincipal cannot be a proxy role.";
             }
-        } else if (StringUtils.isNotBlank(originalPrincipal)) {
+        } else if (StringUtils.isNotBlank(originalPrincipal)
+                && !(allowNonProxyPrincipalsToBeEqual && originalPrincipal.equals(authenticatedPrincipal))) {
             errorMsg = "cannot specify originalPrincipal when connecting without valid proxy role.";
         }
         if (errorMsg != null) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -673,7 +673,7 @@ private void completeConnect(int clientProtoVersion, String clientVersion) {
         if (service.isAuthenticationEnabled()) {
             if (service.isAuthorizationEnabled()) {
                 if (!service.getAuthorizationService()
-                    .isValidOriginalPrincipal(authRole, originalPrincipal, remoteAddress)) {
+                    .isValidOriginalPrincipal(authRole, originalPrincipal, remoteAddress, false)) {
                     state = State.Failed;
                     service.getPulsarStats().recordConnectionCreateFail();
                     final ByteBuf msg = Commands.newError(-1, ServerError.AuthorizationError, "Invalid roles.");

pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationTest.java

@@ -242,27 +242,31 @@ public void testOriginalRoleValidation() throws Exception {
         AuthorizationService auth = new AuthorizationService(conf, Mockito.mock(PulsarResources.class));
 
         // Original principal should be supplied when authenticatedPrincipal is proxy role
-        assertTrue(auth.isValidOriginalPrincipal("proxy", "client", (SocketAddress) null));
+        assertTrue(auth.isValidOriginalPrincipal("proxy", "client", (SocketAddress) null, false));
 
         // Non proxy role should not supply originalPrincipal
-        assertTrue(auth.isValidOriginalPrincipal("client", "", (SocketAddress) null));
-        assertTrue(auth.isValidOriginalPrincipal("client", null, (SocketAddress) null));
+        assertTrue(auth.isValidOriginalPrincipal("client", "", (SocketAddress) null, false));
+        assertTrue(auth.isValidOriginalPrincipal("client", null, (SocketAddress) null, false));
+
+        // Edge cases that differ because binary protocol and http protocol have different expectations
+        assertTrue(auth.isValidOriginalPrincipal("client", "client", (SocketAddress) null, true));
+        assertFalse(auth.isValidOriginalPrincipal("client", "client", (SocketAddress) null, false));
 
         // Only likely in cases when authentication is disabled, but we still define these to be valid.
-        assertTrue(auth.isValidOriginalPrincipal(null, null, (SocketAddress) null));
-        assertTrue(auth.isValidOriginalPrincipal(null, "", (SocketAddress) null));
-        assertTrue(auth.isValidOriginalPrincipal("", null, (SocketAddress) null));
-        assertTrue(auth.isValidOriginalPrincipal("", "", (SocketAddress) null));
+        assertTrue(auth.isValidOriginalPrincipal(null, null, (SocketAddress) null, false));
+        assertTrue(auth.isValidOriginalPrincipal(null, "", (SocketAddress) null, false));
+        assertTrue(auth.isValidOriginalPrincipal("", null, (SocketAddress) null, false));
+        assertTrue(auth.isValidOriginalPrincipal("", "", (SocketAddress) null, false));
 
         // Proxy role must supply an original principal
-        assertFalse(auth.isValidOriginalPrincipal("proxy", "", (SocketAddress) null));
-        assertFalse(auth.isValidOriginalPrincipal("proxy", null, (SocketAddress) null));
+        assertFalse(auth.isValidOriginalPrincipal("proxy", "", (SocketAddress) null, false));
+        assertFalse(auth.isValidOriginalPrincipal("proxy", null, (SocketAddress) null, false));
 
         // OriginalPrincipal cannot be proxy role
-        assertFalse(auth.isValidOriginalPrincipal("proxy", "proxy", (SocketAddress) null));
-        assertFalse(auth.isValidOriginalPrincipal("client", "proxy", (SocketAddress) null));
-        assertFalse(auth.isValidOriginalPrincipal("", "proxy", (SocketAddress) null));
-        assertFalse(auth.isValidOriginalPrincipal(null, "proxy", (SocketAddress) null));
+        assertFalse(auth.isValidOriginalPrincipal("proxy", "proxy", (SocketAddress) null, false));
+        assertFalse(auth.isValidOriginalPrincipal("client", "proxy", (SocketAddress) null, false));
+        assertFalse(auth.isValidOriginalPrincipal("", "proxy", (SocketAddress) null, false));
+        assertFalse(auth.isValidOriginalPrincipal(null, "proxy", (SocketAddress) null, false));
 
         // Must gracefully handle a missing AuthenticationDataSource
         assertTrue(auth.isValidOriginalPrincipal("proxy", "client", (AuthenticationDataSource) null));

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java

@@ -613,6 +613,8 @@ public void testConnectCommandWithInvalidRoleCombinations() throws Exception {
         verifyAuthRoleAndOriginalPrincipalBehavior(authMethodName, "pass.client", "pass.proxy");
         // Invalid combinations where the original principal is set to a non-proxy role
         verifyAuthRoleAndOriginalPrincipalBehavior(authMethodName, "pass.client1", "pass.client");
+        verifyAuthRoleAndOriginalPrincipalBehavior(authMethodName, "pass.client", "pass.client");
+        verifyAuthRoleAndOriginalPrincipalBehavior(authMethodName, "pass.client", "pass.client1");
     }
 
     private void verifyAuthRoleAndOriginalPrincipalBehavior(String authMethodName, String authData,