/**
 * 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.client.api;

import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.mockito.Mockito.spy;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import javax.naming.AuthenticationException;
import lombok.Cleanup;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.authentication.AuthenticationDataCommand;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
import org.apache.pulsar.broker.authentication.AuthenticationProvider;
import org.apache.pulsar.broker.authorization.AuthorizationProvider;
import org.apache.pulsar.broker.authorization.AuthorizationService;
import org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider;
import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.common.naming.NamespaceName;
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.DelayedDeliveryPolicies;
import org.apache.pulsar.common.policies.data.DispatchRate;
import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode;
import org.apache.pulsar.common.policies.data.InactiveTopicPolicies;
import org.apache.pulsar.common.policies.data.NamespaceOperation;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
import org.apache.pulsar.common.policies.data.PublishRate;
import org.apache.pulsar.common.policies.data.SubscribeRate;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.policies.data.TenantOperation;
import org.apache.pulsar.common.policies.data.TopicOperation;
import org.apache.pulsar.common.util.RestException;
import org.apache.pulsar.packages.management.core.MockedPackagesStorageProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.Test;

@Test(groups = "broker-api")
public class AuthorizationProducerConsumerTest extends ProducerConsumerBase {
    private static final Logger log = LoggerFactory.getLogger(AuthorizationProducerConsumerTest.class);

    private static final String clientRole = "plugbleRole";
    private static final Set<String> clientAuthProviderSupportedRoles = Sets.newHashSet(clientRole);

    protected void setup() throws Exception {

        conf.setAuthenticationEnabled(true);
        conf.setAuthorizationEnabled(true);

        Set<String> superUserRoles = new HashSet<>();
        superUserRoles.add("superUser");
        conf.setSuperUserRoles(superUserRoles);

        Set<String> providers = new HashSet<>();
        providers.add(TestAuthenticationProvider.class.getName());
        conf.setAuthenticationProviders(providers);

        conf.setClusterName("test");

        super.init();
    }

    @AfterMethod(alwaysRun = true)
    @Override
    protected void cleanup() throws Exception {
        super.internalCleanup();
    }

    /**
     * It verifies plugable authorization service
     *
     * <pre>
     * 1. Client passes correct authorization plugin-name + correct auth role: SUCCESS
     * 2. Client passes correct authorization plugin-name + incorrect auth-role: FAIL
     * 3. Client passes incorrect authorization plugin-name + correct auth-role: FAIL
     * </pre>
     *
     * @throws Exception
     */
    @Test
    public void testProducerAndConsumerAuthorization() throws Exception {
        log.info("-- Starting {} test --", methodName);

        conf.setAuthorizationProvider(TestAuthorizationProvider.class.getName());
        setup();

        Authentication adminAuthentication = new ClientAuthentication("superUser");

        @Cleanup
        PulsarAdmin admin = spy(
                PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString()).authentication(adminAuthentication).build());

        String lookupUrl = pulsar.getBrokerServiceUrl();

        Authentication authentication = new ClientAuthentication(clientRole);
        Authentication authenticationInvalidRole = new ClientAuthentication("test-role");

        @Cleanup
        PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(lookupUrl).authentication(authentication)
                .operationTimeout(1000, TimeUnit.MILLISECONDS).build();

        @Cleanup
        PulsarClient pulsarClientInvalidRole = PulsarClient.builder().serviceUrl(lookupUrl)
                .operationTimeout(1000, TimeUnit.MILLISECONDS)
                .authentication(authenticationInvalidRole).build();

        admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());

        admin.tenants().createTenant("my-property",
                new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
        admin.namespaces().createNamespace("my-property/my-ns", Sets.newHashSet("test"));

        // (1) Valid Producer and consumer creation
        Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("persistent://my-property/my-ns/my-topic")
                .subscriptionName("my-subscriber-name").subscribe();
        Producer<byte[]> producer = pulsarClient.newProducer().topic("persistent://my-property/my-ns/my-topic")
                .create();
        consumer.close();
        producer.close();

        // (2) InValid user auth-role will be rejected by authorization service
        try {
            consumer = pulsarClientInvalidRole.newConsumer().topic("persistent://my-property/my-ns/my-topic")
                    .subscriptionName("my-subscriber-name").subscribe();
            Assert.fail("should have failed with authorization error");
        } catch (PulsarClientException.AuthorizationException pa) {
            // Ok
        }
        try {
            producer = pulsarClientInvalidRole.newProducer().topic("persistent://my-property/my-ns/my-topic")
                    .create();
            Assert.fail("should have failed with authorization error");
        } catch (PulsarClientException.AuthorizationException pa) {
            // Ok
        }

        log.info("-- Exiting {} test --", methodName);
    }

    @Test
    public void testNamespacePoliciesPermission() throws Exception {
        log.info("-- Starting {} test --", methodName);

        conf.setAuthorizationProvider(PulsarAuthorizationProvider.class.getName());
        setup();

        final String tenantRole = "tenant-role";
        final String namespace = "my-property/my-ns-sub-auth";
        Authentication adminAuthentication = new ClientAuthentication("superUser");

        @Cleanup
        PulsarAdmin superAdmin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
                .authentication(adminAuthentication).build());

        Authentication tenantAdminAuthentication = new ClientAuthentication(tenantRole);
        @Cleanup
        PulsarAdmin tenantAdmin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
                .authentication(tenantAdminAuthentication).build());

        superAdmin.clusters().createCluster("test",
                ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
        superAdmin.tenants().createTenant("my-property",
                new TenantInfoImpl(Sets.newHashSet(tenantRole), Sets.newHashSet("test")));
        superAdmin.namespaces().createNamespace(namespace, Sets.newHashSet("test"));

        // verify super-user and tenant-administrator are able to perform all namespace-policies admin api
        // test for `publish-rate`
        PublishRate publishRate = new PublishRate(10, 100);
        superAdmin.namespaces().setPublishRate(namespace, publishRate);
        assertEquals(superAdmin.namespaces().getPublishRate(namespace), publishRate);
        superAdmin.namespaces().removePublishRate(namespace);
        assertNull(superAdmin.namespaces().getPublishRate(namespace));
        tenantAdmin.namespaces().setPublishRate(namespace, publishRate);
        assertEquals(tenantAdmin.namespaces().getPublishRate(namespace), publishRate);
        tenantAdmin.namespaces().removePublishRate(namespace);
        assertNull(tenantAdmin.namespaces().getPublishRate(namespace));

        // test for `topic-dispatch-rate`
        DispatchRate dispatchRate = DispatchRate.builder()
                .dispatchThrottlingRateInByte(4096)
                .dispatchThrottlingRateInMsg(1000)
                .build();
        superAdmin.namespaces().setDispatchRate(namespace, dispatchRate);
        assertEquals(superAdmin.namespaces().getDispatchRate(namespace), dispatchRate);
        superAdmin.namespaces().removeDispatchRate(namespace);
        assertNull(superAdmin.namespaces().getDispatchRate(namespace));
        tenantAdmin.namespaces().setDispatchRate(namespace, dispatchRate);
        assertEquals(tenantAdmin.namespaces().getDispatchRate(namespace), dispatchRate);
        tenantAdmin.namespaces().removeDispatchRate(namespace);
        assertNull(tenantAdmin.namespaces().getDispatchRate(namespace));

        // test for `subscription-dispatch-rate`
        superAdmin.namespaces().setSubscriptionDispatchRate(namespace, dispatchRate);
        assertEquals(superAdmin.namespaces().getSubscriptionDispatchRate(namespace), dispatchRate);
        superAdmin.namespaces().removeSubscriptionDispatchRate(namespace);
        assertNull(superAdmin.namespaces().getSubscriptionDispatchRate(namespace));
        tenantAdmin.namespaces().setSubscriptionDispatchRate(namespace, dispatchRate);
        assertEquals(tenantAdmin.namespaces().getSubscriptionDispatchRate(namespace), dispatchRate);
        tenantAdmin.namespaces().removeSubscriptionDispatchRate(namespace);
        assertNull(tenantAdmin.namespaces().getSubscriptionDispatchRate(namespace));

        // test for `subscribe-rate`
        SubscribeRate subscribeRate = new SubscribeRate(10, 60);
        superAdmin.namespaces().setSubscribeRate(namespace, subscribeRate);
        assertEquals(superAdmin.namespaces().getSubscribeRate(namespace), subscribeRate);
        superAdmin.namespaces().removeSubscribeRate(namespace);
        assertNull(superAdmin.namespaces().getSubscribeRate(namespace));
        tenantAdmin.namespaces().setSubscribeRate(namespace, subscribeRate);
        assertEquals(tenantAdmin.namespaces().getSubscribeRate(namespace), subscribeRate);
        tenantAdmin.namespaces().removeSubscribeRate(namespace);
        assertNull(tenantAdmin.namespaces().getSubscribeRate(namespace));

        // test for `replicator-dispatch-rate`
        superAdmin.namespaces().setReplicatorDispatchRate(namespace, dispatchRate);
        assertEquals(superAdmin.namespaces().getReplicatorDispatchRate(namespace), dispatchRate);
        superAdmin.namespaces().removeReplicatorDispatchRate(namespace);
        assertNull(superAdmin.namespaces().getReplicatorDispatchRate(namespace));
        tenantAdmin.namespaces().setReplicatorDispatchRate(namespace, dispatchRate);
        assertEquals(tenantAdmin.namespaces().getReplicatorDispatchRate(namespace), dispatchRate);
        tenantAdmin.namespaces().removeReplicatorDispatchRate(namespace);
        assertNull(tenantAdmin.namespaces().getReplicatorDispatchRate(namespace));

        // test for `inactive-topic`
        InactiveTopicPolicies inactiveTopicPolicies = new InactiveTopicPolicies(
                InactiveTopicDeleteMode.delete_when_no_subscriptions, 30, true);
        superAdmin.namespaces().setInactiveTopicPolicies(namespace, inactiveTopicPolicies);
        assertEquals(superAdmin.namespaces().getInactiveTopicPolicies(namespace), inactiveTopicPolicies);
        superAdmin.namespaces().removeInactiveTopicPolicies(namespace);
        assertNull(superAdmin.namespaces().getInactiveTopicPolicies(namespace));
        tenantAdmin.namespaces().setInactiveTopicPolicies(namespace, inactiveTopicPolicies);
        assertEquals(tenantAdmin.namespaces().getInactiveTopicPolicies(namespace), inactiveTopicPolicies);
        tenantAdmin.namespaces().removeInactiveTopicPolicies(namespace);
        assertNull(tenantAdmin.namespaces().getInactiveTopicPolicies(namespace));

        // test for `delayed-delivery`
        DelayedDeliveryPolicies delayedDeliveryPolicies = DelayedDeliveryPolicies.builder()
                .tickTime(60)
                .active(true)
                .build();
        superAdmin.namespaces().setDelayedDeliveryMessages(namespace, delayedDeliveryPolicies);
        assertEquals(superAdmin.namespaces().getDelayedDelivery(namespace), delayedDeliveryPolicies);
        superAdmin.namespaces().removeDelayedDeliveryMessages(namespace);
        assertNull(superAdmin.namespaces().getDelayedDelivery(namespace));
        tenantAdmin.namespaces().setDelayedDeliveryMessages(namespace, delayedDeliveryPolicies);
        assertEquals(tenantAdmin.namespaces().getDelayedDelivery(namespace), delayedDeliveryPolicies);
        tenantAdmin.namespaces().removeDelayedDeliveryMessages(namespace);
        assertNull(tenantAdmin.namespaces().getDelayedDelivery(namespace));

        // test for `max-subscriptions-per-topic`
        superAdmin.namespaces().setMaxSubscriptionsPerTopic(namespace, 10);
        assertEquals(superAdmin.namespaces().getMaxSubscriptionsPerTopic(namespace).intValue(), 10);
        superAdmin.namespaces().removeMaxSubscriptionsPerTopic(namespace);
        assertNull(superAdmin.namespaces().getMaxSubscriptionsPerTopic(namespace));
        tenantAdmin.namespaces().setMaxSubscriptionsPerTopic(namespace, 20);
        assertEquals(tenantAdmin.namespaces().getMaxSubscriptionsPerTopic(namespace).intValue(), 20);
        tenantAdmin.namespaces().removeMaxSubscriptionsPerTopic(namespace);
        assertNull(tenantAdmin.namespaces().getMaxSubscriptionsPerTopic(namespace));

        log.info("-- Exiting {} test --", methodName);
    }

    @Test
    public void testSubscriberPermission() throws Exception {
        log.info("-- Starting {} test --", methodName);

        conf.setEnablePackagesManagement(true);
        conf.setPackagesManagementStorageProvider(MockedPackagesStorageProvider.class.getName());
        conf.setAuthorizationProvider(PulsarAuthorizationProvider.class.getName());
        setup();

        final String tenantRole = "tenant-role";
        final String subscriptionRole = "sub1-role";
        final String subscriptionName = "sub1";
        final String subscriptionName2 = "sub2";
        final String namespace = "my-property/my-ns-sub-auth";
        final String topicName = "persistent://" + namespace + "/my-topic";
        Authentication adminAuthentication = new ClientAuthentication("superUser");

        clientAuthProviderSupportedRoles.add(subscriptionRole);

        @Cleanup
        PulsarAdmin superAdmin = spy(
                PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString()).authentication(adminAuthentication).build());

        Authentication tenantAdminAuthentication = new ClientAuthentication(tenantRole);
        @Cleanup
        PulsarAdmin tenantAdmin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
                .authentication(tenantAdminAuthentication).build());

        Authentication subAdminAuthentication = new ClientAuthentication(subscriptionRole);
        @Cleanup
        PulsarAdmin sub1Admin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
                .authentication(subAdminAuthentication).build());

        Authentication authentication = new ClientAuthentication(subscriptionRole);

        superAdmin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());

        superAdmin.tenants().createTenant("my-property",
                new TenantInfoImpl(Sets.newHashSet(tenantRole), Sets.newHashSet("test")));
        superAdmin.namespaces().createNamespace(namespace, Sets.newHashSet("test"));
        assertNull(superAdmin.namespaces().getPublishRate(namespace));

        // subscriptionRole doesn't have topic-level authorization, so it will fail to get topic stats-internal info
        try {
            sub1Admin.topics().getInternalStats(topicName, true);
            fail("should have failed with authorization exception");
        } catch (Exception e) {
            assertTrue(e.getMessage().startsWith(
                    "Unauthorized to validateTopicOperation for operation [GET_STATS]"));
        }
        try {
            sub1Admin.topics().getBacklogSizeByMessageId(topicName, MessageId.earliest);
            fail("should have failed with authorization exception");
        } catch (Exception e) {
            assertTrue(e.getMessage().startsWith(
                    "Unauthorized to validateTopicOperation for operation"));
        }

        // grant topic consume authorization to the subscriptionRole
        tenantAdmin.topics().grantPermission(topicName, subscriptionRole,
                Collections.singleton(AuthAction.consume));

        replacePulsarClient(PulsarClient.builder()
                .serviceUrl(pulsar.getBrokerServiceUrl())
                .authentication(authentication));

        // (1) Create subscription name
        Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName).subscriptionName(subscriptionName)
                .subscribe();
        Consumer<byte[]> consumer2 = pulsarClient.newConsumer().topic(topicName).subscriptionName(subscriptionName2)
                .subscribe();
        consumer.close();
        consumer2.close();

        List<String> subscriptions = sub1Admin.topics().getSubscriptions(topicName);
        assertEquals(subscriptions.size(), 2);

        // now, subscriptionRole have consume authorization on topic, so it will successfully get topic internal stats
        PersistentTopicInternalStats internalStats = sub1Admin.topics().getInternalStats(topicName, true);
        assertNotNull(internalStats);
        Long backlogSize = sub1Admin.topics().getBacklogSizeByMessageId(topicName, MessageId.earliest);
        assertEquals(backlogSize.longValue(), 0);

        // verify tenant is able to perform all subscription-admin api
        tenantAdmin.topics().skipAllMessages(topicName, subscriptionName);
        tenantAdmin.topics().skipMessages(topicName, subscriptionName, 1);
        try {
            tenantAdmin.topics().expireMessages(topicName, subscriptionName, 10);
        } catch (Exception e) {
            // my-sub1 has no msg backlog, so expire message won't be issued on that subscription
            assertTrue(e.getMessage().startsWith("Expire message by timestamp not issued on topic"));
        }
        tenantAdmin.topics().expireMessages(topicName, subscriptionName, new MessageIdImpl(-1, -1, -1), true);
        tenantAdmin.topics().peekMessages(topicName, subscriptionName, 1);
        tenantAdmin.topics().resetCursor(topicName, subscriptionName, 10);
        tenantAdmin.topics().resetCursor(topicName, subscriptionName, MessageId.earliest);

        // subscriptionRole doesn't have namespace-level authorization, so it will fail to unsubscribe namespace
        try {
            sub1Admin.namespaces().unsubscribeNamespace(namespace, subscriptionName2);
            fail("should have failed with authorization exception");
        } catch (Exception e) {
            assertTrue(e.getMessage().startsWith(
                    "Unauthorized to validateNamespaceOperation for operation [UNSUBSCRIBE]"));
        }
        try {
            sub1Admin.namespaces().getTopics(namespace);
            fail("should have failed with authorization exception");
        } catch (Exception e) {
            assertTrue(e.getMessage().startsWith(
                    "Unauthorized to validateNamespaceOperation for operation [GET_TOPICS]"));
        }
        try {
            sub1Admin.packages().listPackages("function", namespace);
            fail("should have failed with authorization exception");
        } catch (Exception e) {
            assertTrue(e.getMessage().startsWith(
                    "Role sub1-role has not the 'package' permission to do the packages operations"));
        }

        // grant namespace-level authorization to the subscriptionRole
        tenantAdmin.namespaces().grantPermissionOnNamespace(namespace, subscriptionRole,
                Sets.newHashSet(AuthAction.consume, AuthAction.packages));

        // now, subscriptionRole have consume authorization on namespace, so it will successfully unsubscribe namespace
        sub1Admin.namespaces().unsubscribeNamespaceBundle(namespace, "0x00000000_0xffffffff", subscriptionName2);
        subscriptions = sub1Admin.topics().getSubscriptions(topicName);
        assertEquals(subscriptions.size(), 1);
        List<String> topics = sub1Admin.namespaces().getTopics(namespace);
        assertEquals(topics.size(), 1);
        List<String> packages = sub1Admin.packages().listPackages("function", namespace);
        assertEquals(packages.size(), 0);

        // subscriptionRole has namespace-level authorization
        sub1Admin.topics().resetCursor(topicName, subscriptionName, 10);

        // grant subscription access to specific different role and only that role can access the subscription
        String otherPrincipal = "Principal-1-to-access-sub";
        tenantAdmin.namespaces().grantPermissionOnSubscription(namespace, subscriptionName,
                Collections.singleton(otherPrincipal));
        TreeMap<String, Set<String>> permissionOnSubscription = new TreeMap<>();
        permissionOnSubscription.put(subscriptionName, Collections.singleton(otherPrincipal));
        Assert.assertEquals(tenantAdmin.namespaces().getPermissionOnSubscription(namespace), permissionOnSubscription);

        // now, subscriptionRole doesn't have subscription level access so, it will fail to access subscription
        try {
            sub1Admin.topics().resetCursor(topicName, subscriptionName, 10);
            fail("should have fail with authorization exception");
        } catch (org.apache.pulsar.client.admin.PulsarAdminException.NotAuthorizedException e) {
            // Ok
        }

        // reset on position
        try {
            sub1Admin.topics().resetCursor(topicName, subscriptionName, MessageId.earliest);
            fail("should have fail with authorization exception");
        } catch (org.apache.pulsar.client.admin.PulsarAdminException.NotAuthorizedException e) {
            // Ok
        }

        // now, grant subscription-access to subscriptionRole as well
        superAdmin.namespaces().grantPermissionOnSubscription(namespace, subscriptionName,
                Sets.newHashSet(otherPrincipal, subscriptionRole));
        TreeMap<String, Set<String>> permissionOnSubscription1 = new TreeMap<>();
        permissionOnSubscription1.put(subscriptionName, Sets.newHashSet(otherPrincipal, subscriptionRole));
        Assert.assertEquals(tenantAdmin.namespaces().getPermissionOnSubscription(namespace), permissionOnSubscription1);

        sub1Admin.topics().skipAllMessages(topicName, subscriptionName);
        sub1Admin.topics().skipMessages(topicName, subscriptionName, 1);
        try {
            tenantAdmin.topics().expireMessages(topicName, subscriptionName, 10);
        } catch (Exception e) {
            // my-sub1 has no msg backlog, so expire message won't be issued on that subscription
            assertTrue(e.getMessage().startsWith("Expire message by timestamp not issued on topic"));
        }        sub1Admin.topics().peekMessages(topicName, subscriptionName, 1);
        sub1Admin.topics().resetCursor(topicName, subscriptionName, 10);
        sub1Admin.topics().resetCursor(topicName, subscriptionName, MessageId.earliest);

        superAdmin.namespaces().revokePermissionOnSubscription(namespace, subscriptionName, subscriptionRole);

        try {
            sub1Admin.topics().resetCursor(topicName, subscriptionName, 10);
            fail("should have fail with authorization exception");
        } catch (org.apache.pulsar.client.admin.PulsarAdminException.NotAuthorizedException e) {
            // Ok
        }

        log.info("-- Exiting {} test --", methodName);
    }

    @Test
    public void testClearBacklogPermission() throws Exception {
        log.info("-- Starting {} test --", methodName);

        conf.setAuthorizationProvider(PulsarAuthorizationProvider.class.getName());
        setup();

        final String tenantRole = "tenant-role";
        final String subscriptionRole = "sub-role";
        final String subscriptionName = "sub1";
        final String namespace = "my-property/my-ns-sub-auth";
        final String topicName = "persistent://" + namespace + "/my-topic";
        Authentication adminAuthentication = new ClientAuthentication("superUser");

        clientAuthProviderSupportedRoles.add(subscriptionRole);

        @Cleanup
        PulsarAdmin superAdmin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
                .authentication(adminAuthentication).build());

        Authentication tenantAdminAuthentication = new ClientAuthentication(tenantRole);
        @Cleanup
        PulsarAdmin tenantAdmin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
                .authentication(tenantAdminAuthentication).build());

        Authentication subAdminAuthentication = new ClientAuthentication(subscriptionRole);
        @Cleanup
        PulsarAdmin sub1Admin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
                .authentication(subAdminAuthentication).build());

        superAdmin.clusters().createCluster("test",
                ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
        superAdmin.tenants().createTenant("my-property",
                new TenantInfoImpl(Sets.newHashSet(tenantRole), Sets.newHashSet("test")));
        superAdmin.namespaces().createNamespace(namespace, Sets.newHashSet("test"));
        superAdmin.topics().createPartitionedTopic(topicName, 1);
        assertEquals(tenantAdmin.topics().getPartitionedTopicList(namespace),
                Lists.newArrayList(topicName));

        // grant topic consume&produce authorization to the subscriptionRole
        superAdmin.topics().grantPermission(topicName, subscriptionRole,
                Sets.newHashSet(AuthAction.produce, AuthAction.consume));
        replacePulsarClient(PulsarClient.builder()
                .serviceUrl(pulsar.getBrokerServiceUrl())
                .authentication(subAdminAuthentication));

        @Cleanup
        Producer<byte[]> batchProducer = pulsarClient.newProducer().topic(topicName)
                .enableBatching(false)
                .create();

        @Cleanup
        Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topicName)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .subscriptionName(subscriptionName)
                .subscribe();

        CompletableFuture<MessageId> completableFuture = new CompletableFuture<>();
        for (int i = 0; i < 10; i++) {
            completableFuture = batchProducer.sendAsync("a".getBytes());
        }
        completableFuture.get();
        assertEquals(sub1Admin.topics().getStats(topicName + "-partition-0").getSubscriptions()
                .get(subscriptionName).getMsgBacklog(), 10);

        // subscriptionRole doesn't have namespace-level authorization, so it will fail to clear backlog
        try {
            sub1Admin.topics().getPartitionedTopicList(namespace);
            fail("should have failed with authorization exception");
        } catch (Exception e) {
            assertTrue(e.getMessage().startsWith(
                    "Unauthorized to validateNamespaceOperation for operation [GET_TOPICS]"));
        }
        try {
            sub1Admin.namespaces().clearNamespaceBundleBacklog(namespace, "0x00000000_0xffffffff");
            fail("should have failed with authorization exception");
        } catch (Exception e) {
            assertTrue(e.getMessage().startsWith(
                    "Unauthorized to validateNamespaceOperation for operation [CLEAR_BACKLOG]"));
        }

        superAdmin.namespaces().grantPermissionOnNamespace(namespace, subscriptionRole,
                Sets.newHashSet(AuthAction.consume));
        // now, subscriptionRole have consume authorization on namespace, so it will successfully clear backlog
        assertEquals(sub1Admin.topics().getPartitionedTopicList(namespace),
                Lists.newArrayList(topicName));
        sub1Admin.namespaces().clearNamespaceBundleBacklog(namespace, "0x00000000_0xffffffff");
        assertEquals(sub1Admin.topics().getStats(topicName + "-partition-0").getSubscriptions()
                .get(subscriptionName).getMsgBacklog(), 0);

        log.info("-- Exiting {} test --", methodName);
    }

    @Test
    public void testSubscriptionPrefixAuthorization() throws Exception {
        log.info("-- Starting {} test --", methodName);

        conf.setAuthorizationProvider(TestAuthorizationProviderWithSubscriptionPrefix.class.getName());
        setup();

        Authentication adminAuthentication = new ClientAuthentication("superUser");
        @Cleanup
        PulsarAdmin admin = spy(
                PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString()).authentication(adminAuthentication).build());

        Authentication authentication = new ClientAuthentication(clientRole);

        replacePulsarClient(PulsarClient.builder()
                .serviceUrl(pulsar.getBrokerServiceUrl())
                .authentication(authentication));


        admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());

        admin.tenants().createTenant("prop-prefix",
                new TenantInfoImpl(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
        admin.namespaces().createNamespace("prop-prefix/ns", Sets.newHashSet("test"));

        // (1) Valid subscription name will be approved by authorization service
        Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("persistent://prop-prefix/ns/t1")
                .subscriptionName(clientRole + "-sub1").subscribe();
        consumer.close();

        // (2) InValid subscription name will be rejected by authorization service
        try {
            consumer = pulsarClient.newConsumer().topic("persistent://prop-prefix/ns/t1").subscriptionName("sub1")
                    .subscribe();
            Assert.fail("should have failed with authorization error");
        } catch (PulsarClientException.AuthorizationException pa) {
            // Ok
        }

        log.info("-- Exiting {} test --", methodName);
    }

    @Test
    public void testGrantPermission() throws Exception {
        log.info("-- Starting {} test --", methodName);

        conf.setAuthorizationProvider(TestAuthorizationProviderWithGrantPermission.class.getName());
        setup();

        AuthorizationService authorizationService = new AuthorizationService(conf, null);
        TopicName topicName = TopicName.get("persistent://prop/cluster/ns/t1");
        String role = "test-role";
        Assert.assertFalse(authorizationService.canProduce(topicName, role, null));
        Assert.assertFalse(authorizationService.canConsume(topicName, role, null, "sub1"));
        authorizationService.grantPermissionAsync(topicName, null, role, "auth-json").get();
        Assert.assertTrue(authorizationService.canProduce(topicName, role, null));
        Assert.assertTrue(authorizationService.canConsume(topicName, role, null, "sub1"));

        log.info("-- Exiting {} test --", methodName);
    }

    @Test
    public void testAuthData() throws Exception {
        log.info("-- Starting {} test --", methodName);

        conf.setAuthorizationProvider(TestAuthorizationProviderWithGrantPermission.class.getName());
        setup();

        AuthorizationService authorizationService = new AuthorizationService(conf, null);
        TopicName topicName = TopicName.get("persistent://prop/cluster/ns/t1");
        String role = "test-role";
        authorizationService.grantPermissionAsync(topicName, null, role, "auth-json").get();
        Assert.assertEquals(TestAuthorizationProviderWithGrantPermission.authDataJson, "auth-json");
        Assert.assertTrue(authorizationService.canProduce(topicName, role, new AuthenticationDataCommand("prod-auth")));
        Assert.assertEquals(TestAuthorizationProviderWithGrantPermission.authenticationData.getCommandData(),
                "prod-auth");
        Assert.assertTrue(
                authorizationService.canConsume(topicName, role, new AuthenticationDataCommand("cons-auth"), "sub1"));
        Assert.assertEquals(TestAuthorizationProviderWithGrantPermission.authenticationData.getCommandData(),
                "cons-auth");

        log.info("-- Exiting {} test --", methodName);
    }

    public static class ClientAuthentication implements Authentication {
        String user;

        public ClientAuthentication(String user) {
            this.user = user;
        }

        @Override
        public void close() throws IOException {
            // No-op
        }

        @Override
        public String getAuthMethodName() {
            return "test";
        }

        @Override
        public AuthenticationDataProvider getAuthData() throws PulsarClientException {
            AuthenticationDataProvider provider = new AuthenticationDataProvider() {
                public boolean hasDataForHttp() {
                    return true;
                }

                @SuppressWarnings("unchecked")
                public Set<Map.Entry<String, String>> getHttpHeaders() {
                    return Sets.newHashSet(Maps.immutableEntry("user", user));
                }

                public boolean hasDataFromCommand() {
                    return true;
                }

                public String getCommandData() {
                    return user;
                }
            };
            return provider;
        }

        @Override
        public void configure(Map<String, String> authParams) {
            // No-op
        }

        @Override
        public void start() throws PulsarClientException {
            // No-op
        }

    }

    public static class TestAuthenticationProvider implements AuthenticationProvider {

        @Override
        public void close() throws IOException {
            // no-op
        }

        @Override
        public void initialize(ServiceConfiguration config) throws IOException {
            // No-op
        }

        @Override
        public String getAuthMethodName() {
            return "test";
        }

        @Override
        public String authenticate(AuthenticationDataSource authData) throws AuthenticationException {
            return authData.getCommandData() != null ? authData.getCommandData() : authData.getHttpHeader("user");
        }

    }

    public static class TestAuthorizationProvider implements AuthorizationProvider {

        public ServiceConfiguration conf;

        @Override
        public void close() throws IOException {
            // No-op
        }

        @Override
        public CompletableFuture<Boolean> isSuperUser(String role,
                                                      ServiceConfiguration serviceConfiguration) {
            Set<String> superUserRoles = serviceConfiguration.getSuperUserRoles();
            return CompletableFuture.completedFuture(role != null && superUserRoles.contains(role) ? true : false);
        }

        @Override
        public void initialize(ServiceConfiguration conf, PulsarResources pulsarResources) throws IOException {
            this.conf = conf;
            // No-op
        }

        @Override
        public CompletableFuture<Boolean> canProduceAsync(TopicName topicName, String role,
                AuthenticationDataSource authenticationData) {
            return CompletableFuture.completedFuture(clientAuthProviderSupportedRoles.contains(role));
        }

        @Override
        public CompletableFuture<Boolean> canConsumeAsync(TopicName topicName, String role,
                AuthenticationDataSource authenticationData, String subscription) {
            return CompletableFuture.completedFuture(clientAuthProviderSupportedRoles.contains(role));
        }

        @Override
        public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role,
                AuthenticationDataSource authenticationData) {
            return CompletableFuture.completedFuture(clientAuthProviderSupportedRoles.contains(role));
        }

        @Override
        public CompletableFuture<Boolean> allowFunctionOpsAsync(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData) {
            return null;
        }

        @Override
        public CompletableFuture<Boolean> allowSourceOpsAsync(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData) {
            return null;
        }

        @Override
        public CompletableFuture<Boolean> allowSinkOpsAsync(NamespaceName namespaceName, String role, AuthenticationDataSource authenticationData) {
            return null;
        }

        @Override
        public CompletableFuture<Void> grantPermissionAsync(NamespaceName namespace, Set<AuthAction> actions,
                String role, String authenticationData) {
            return CompletableFuture.completedFuture(null);
        }

        @Override
        public CompletableFuture<Void> grantPermissionAsync(TopicName topicname, Set<AuthAction> actions, String role,
                String authenticationData) {
            return CompletableFuture.completedFuture(null);
        }

        @Override
        public CompletableFuture<Void> grantSubscriptionPermissionAsync(NamespaceName namespace,
                String subscriptionName, Set<String> roles, String authDataJson) {
            return CompletableFuture.completedFuture(null);
        }

        @Override
        public CompletableFuture<Void> revokeSubscriptionPermissionAsync(NamespaceName namespace,
                String subscriptionName, String role, String authDataJson) {
            return CompletableFuture.completedFuture(null);
        }

        @Override
        public CompletableFuture<Boolean> isTenantAdmin(String tenant, String role, TenantInfo tenantInfo, AuthenticationDataSource authenticationData) {
            return CompletableFuture.completedFuture(true);
        }

        @Override
        public CompletableFuture<Boolean> allowTenantOperationAsync(
            String tenantName, String role, TenantOperation operation, AuthenticationDataSource authData) {
            return CompletableFuture.completedFuture(true);
        }

        @Override
        public Boolean allowTenantOperation(
            String tenantName, String role, TenantOperation operation, AuthenticationDataSource authData) {
            return true;
        }

        @Override
        public CompletableFuture<Boolean> allowNamespaceOperationAsync(
            NamespaceName namespaceName, String role, NamespaceOperation operation, AuthenticationDataSource authData) {
            return CompletableFuture.completedFuture(true);
        }

        @Override
        public Boolean allowNamespaceOperation(
            NamespaceName namespaceName, String role, NamespaceOperation operation, AuthenticationDataSource authData) {
            return null;
        }

        @Override
        public CompletableFuture<Boolean> allowTopicOperationAsync(
            TopicName topic, String role, TopicOperation operation, AuthenticationDataSource authData) {
            CompletableFuture<Boolean> isAuthorizedFuture;

            if (role.equals("plugbleRole")) {
                isAuthorizedFuture = CompletableFuture.completedFuture(true);
            } else {
                isAuthorizedFuture = CompletableFuture.completedFuture(false);
            }

            return isAuthorizedFuture;
        }

        @Override
        public Boolean allowTopicOperation(
            TopicName topicName, String role, TopicOperation operation, AuthenticationDataSource authData) {
            try {
                return allowTopicOperationAsync(topicName, role, operation, authData).get();
            } catch (InterruptedException e) {
                throw new RestException(e);
            } catch (ExecutionException e) {
                throw new RestException(e);
            }
        }
    }

    /**
     * This provider always fails authorization on consumer and passes on producer
     *
     */
    public static class TestAuthorizationProvider2 extends TestAuthorizationProvider {

        @Override
        public CompletableFuture<Boolean> canProduceAsync(TopicName topicName, String role,
                AuthenticationDataSource authenticationData) {
            return CompletableFuture.completedFuture(true);
        }

        @Override
        public CompletableFuture<Boolean> canConsumeAsync(TopicName topicName, String role,
                AuthenticationDataSource authenticationData, String subscription) {
            return CompletableFuture.completedFuture(false);
        }

        @Override
        public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role,
                AuthenticationDataSource authenticationData) {
            return CompletableFuture.completedFuture(true);
        }
    }

    public static class TestAuthorizationProviderWithSubscriptionPrefix extends TestAuthorizationProvider {
        @Override
        public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topic,
                                                                   String role,
                                                                   TopicOperation operation,
                                                                   AuthenticationDataSource authData) {
            CompletableFuture<Boolean> future = new CompletableFuture<>();
            if (authData.hasSubscription()) {
                String subscription = authData.getSubscription();
                if (isNotBlank(subscription)) {
                    if (!subscription.startsWith(role)) {
                        future.completeExceptionally(new PulsarServerException(
                                "The subscription name needs to be prefixed by the authentication role"));
                    }
                }
            }
            future.complete(clientRole.equals(role));
            return future;
        }
    }

    public static class TestAuthorizationProviderWithGrantPermission extends TestAuthorizationProvider {

        private Set<String> grantRoles = Sets.newHashSet();
        static AuthenticationDataSource authenticationData;
        static String authDataJson;

        @Override
        public CompletableFuture<Boolean> canProduceAsync(TopicName topicName, String role,
                AuthenticationDataSource authenticationData) {
            this.authenticationData = authenticationData;
            return CompletableFuture.completedFuture(grantRoles.contains(role));
        }

        @Override
        public CompletableFuture<Boolean> canConsumeAsync(TopicName topicName, String role,
                AuthenticationDataSource authenticationData, String subscription) {
            this.authenticationData = authenticationData;
            return CompletableFuture.completedFuture(grantRoles.contains(role));
        }

        @Override
        public CompletableFuture<Boolean> canLookupAsync(TopicName topicName, String role,
                AuthenticationDataSource authenticationData) {
            this.authenticationData = authenticationData;
            return CompletableFuture.completedFuture(grantRoles.contains(role));
        }

        @Override
        public CompletableFuture<Void> grantPermissionAsync(NamespaceName namespace, Set<AuthAction> actions,
                String role, String authData) {
            this.authDataJson = authData;
            grantRoles.add(role);
            return CompletableFuture.completedFuture(null);
        }

        @Override
        public CompletableFuture<Void> grantPermissionAsync(TopicName topicname, Set<AuthAction> actions, String role,
                String authData) {
            this.authDataJson = authData;
            grantRoles.add(role);
            return CompletableFuture.completedFuture(null);
        }
    }

}