Add test for WebSocketProxy

pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthorizationTest.java

@@ -106,8 +106,18 @@ public void test() throws Exception {
         assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds2"), "other-role", null));
         assertTrue(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
         assertFalse(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds2"), "other-role", null));
-        assertTrue(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds2"), "other-role", null, null));
-        assertFalse(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds2"), "no-access-role", null,null));
+        // Include a subscription name. The subscription doesn't need to exist for the purpose of this test, but this
+        // tests the case when service.getConfig().isGrantImplicitPermissionOnSubscription() is true because we
+        // have not granted permission for this role on the subscription named "sub".
+        assertTrue(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds2"), "other-role", null, "sub"));
+        assertFalse(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds2"), "no-access-role", null,"sub"));
+
+        // Grant permission to a different role for sub and expect failure
+        admin.namespaces().grantPermissionOnSubscription("p1/c1/ns1", "sub", Set.of("no-ones-role"));
+        // Even though other-role has permission to consume from the topic, the "sub" subscription is locked down and
+        // only roles with permission granted via grantPermissionOnSubscription have permission to consume from that
+        // subscription.
+        assertFalse(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds2"), "other-role", null, "sub"));
 
         assertFalse(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "no-access-role", null));
 

pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyAuthorizationWithoutImplicitPermissionOnSubscriptionTest.java

@@ -0,0 +1,142 @@
+package org.apache.pulsar.websocket.proxy;
+
+import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+import com.google.common.collect.Sets;
+import java.net.URI;
+import java.util.EnumSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.Future;
+import javax.servlet.http.HttpServletResponse;
+import org.apache.pulsar.broker.authorization.AuthorizationService;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.AuthAction;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.metadata.impl.ZKMetadataStore;
+import org.apache.pulsar.websocket.WebSocketService;
+import org.apache.pulsar.websocket.service.ProxyServer;
+import org.apache.pulsar.websocket.service.WebSocketProxyConfiguration;
+import org.apache.pulsar.websocket.service.WebSocketServiceStarter;
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.api.UpgradeException;
+import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
+import org.eclipse.jetty.websocket.client.WebSocketClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Class that initializes the WebSocketService disabling {@link WebSocketProxyConfiguration#setGrantImplicitPermissionOnSubscription(boolean)}.
+ * We must have this class on its own because the WebSocketProxyConfiguration is converted to the ServiceConfiguration
+ * on start up, so it is not a dynamic property that we can change after the service has started.
+ */
+
+@Test(groups = "websocket")
+public class ProxyAuthorizationWithoutImplicitPermissionOnSubscriptionTest extends ProducerConsumerBase {
+    private static final Logger log = LoggerFactory.getLogger(ProxyPublishConsumeTest.class);
+    private WebSocketService service;
+    private final String configClusterName = "c1";
+
+    @BeforeClass
+    @Override
+    protected void setup() throws Exception {
+        conf.setClusterName(configClusterName);
+        internalSetup();
+
+        WebSocketProxyConfiguration config = new WebSocketProxyConfiguration();
+        Set<String> superUser = Sets.newHashSet("");
+        config.setAuthorizationEnabled(true);
+        config.setSuperUserRoles(superUser);
+        config.setClusterName("c1");
+        config.setWebServicePort(Optional.of(0));
+        config.setConfigurationMetadataStoreUrl(GLOBAL_DUMMY_VALUE);
+        config.setGrantImplicitPermissionOnSubscription(false);
+        service = spyWithClassAndConstructorArgs(WebSocketService.class, config);
+        doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service)
+                .createConfigMetadataStore(anyString(), anyInt(), anyBoolean());
+        service.start();
+    }
+
+    @AfterMethod(alwaysRun = true)
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+        if (service != null) {
+            service.close();
+        }
+        log.info("Finished Cleaning Up Test setup");
+    }
+
+
+    @Test
+    public void testAuthorizationServiceDirectly() throws Exception {
+        AuthorizationService auth = service.getAuthorizationService();
+
+        assertFalse(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
+
+        admin.clusters().createCluster(configClusterName, ClusterData.builder().build());
+        admin.tenants().createTenant("p1", new TenantInfoImpl(Sets.newHashSet("role1"), Sets.newHashSet("c1")));
+        waitForChange();
+        admin.namespaces().createNamespace("p1/c1/ns1");
+        waitForChange();
+
+        assertFalse(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
+
+        admin.namespaces().grantPermissionOnNamespace("p1/c1/ns1", "my-role", EnumSet.of(AuthAction.produce));
+        waitForChange();
+
+        assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
+        assertTrue(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
+
+        admin.topics().grantPermission("persistent://p1/c1/ns1/ds2", "other-role",
+                EnumSet.of(AuthAction.consume));
+        waitForChange();
+
+        assertTrue(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds2"), "other-role", null));
+        assertTrue(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
+        assertFalse(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds2"), "other-role", null));
+
+        // Expect false because we disabled the implicit permission on subscription
+        assertFalse(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds2"), "other-role", null, "sub"));
+        assertFalse(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds2"), "no-access-role", null,"sub"));
+
+        // Grant permission
+        admin.namespaces().grantPermissionOnSubscription("p1/c1/ns1", "sub", Set.of("other-role"));
+
+        // Expect only true for "other-role" because we granted permission for only that one
+        assertTrue(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds2"), "other-role", null, "sub"));
+        assertFalse(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds2"), "no-access-role", null,"sub"));
+
+
+        assertFalse(auth.canLookup(TopicName.get("persistent://p1/c1/ns1/ds1"), "no-access-role", null));
+
+        admin.namespaces().grantPermissionOnNamespace("p1/c1/ns1", "my-role", EnumSet.allOf(AuthAction.class));
+        waitForChange();
+
+        assertTrue(auth.canProduce(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null));
+        assertTrue(auth.canConsume(TopicName.get("persistent://p1/c1/ns1/ds1"), "my-role", null, null));
+
+        admin.namespaces().deleteNamespace("p1/c1/ns1");
+        admin.tenants().deleteTenant("p1");
+        admin.clusters().deleteCluster("c1");
+    }
+
+    private static void waitForChange() {
+        try {
+            Thread.sleep(100);
+        } catch (InterruptedException e) {
+        }
+    }
+}