/*
    * 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.hadoop.hbase.security.access;
  
  import static org.apache.hadoop.hbase.AuthUtil.toGroupEntry;
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertTrue;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.hbase.Coprocessor;
  import org.apache.hadoop.hbase.CoprocessorEnvironment;
  import org.apache.hadoop.hbase.HBaseTestingUtility;
  import org.apache.hadoop.hbase.HColumnDescriptor;
  import org.apache.hadoop.hbase.HConstants;
  import org.apache.hadoop.hbase.HTableDescriptor;
  import org.apache.hadoop.hbase.TableName;
  import org.apache.hadoop.hbase.TableNotFoundException;
  import org.apache.hadoop.hbase.client.Connection;
  import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
  import org.apache.hadoop.hbase.coprocessor.MasterCoprocessorEnvironment;
  import org.apache.hadoop.hbase.coprocessor.ObserverContext;
  import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
  import org.apache.hadoop.hbase.coprocessor.RegionServerCoprocessorEnvironment;
  import org.apache.hadoop.hbase.master.MasterCoprocessorHost;
  import org.apache.hadoop.hbase.regionserver.HRegionServer;
  import org.apache.hadoop.hbase.regionserver.Region;
  import org.apache.hadoop.hbase.regionserver.RegionCoprocessorHost;
  import org.apache.hadoop.hbase.regionserver.RegionServerCoprocessorHost;
  import org.apache.hadoop.hbase.security.User;
  import org.apache.hadoop.hbase.testclassification.MediumTests;
  import org.apache.hadoop.hbase.util.Bytes;
  import org.apache.hadoop.hbase.util.JVMClusterUtil;
  import org.apache.log4j.Level;
  import org.apache.log4j.Logger;
  import org.junit.AfterClass;
  import org.junit.BeforeClass;
  import org.junit.Test;
  import org.junit.experimental.categories.Category;
  
  /**
   * Performs checks for reference counting w.r.t. TableAuthManager which is used by AccessController.
   */
  @Category({MediumTests.class})
  public class TestAccessController3 extends SecureTestUtil {
    private static final Log LOG = LogFactory.getLog(TestAccessController.class);
  
    static {
      Logger.getLogger(AccessController.class).setLevel(Level.TRACE);
      Logger.getLogger(AccessControlFilter.class).setLevel(Level.TRACE);
      Logger.getLogger(TableAuthManager.class).setLevel(Level.TRACE);
    }
  
    private static TableName TEST_TABLE = TableName.valueOf("testtable1");
    private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
    private static Configuration conf;
  
    /** The systemUserConnection created here is tied to the system user. In case, you are planning
     * to create AccessTestAction, DON'T use this systemUserConnection as the 'doAs' user
     * gets  eclipsed by the system user. */
    private static Connection systemUserConnection;
  
  
    // user with all permissions
    private static User SUPERUSER;
    // user granted with all global permission
    private static User USER_ADMIN;
    // user with rw permissions on column family.
    private static User USER_RW;
    // user with read-only permissions
    private static User USER_RO;
    // user is table owner. will have all permissions on table
    private static User USER_OWNER;
    // user with create table permissions alone
    private static User USER_CREATE;
    // user with no permissions
    private static User USER_NONE;
    // user with admin rights on the column family
    private static User USER_ADMIN_CF;
  
    private static final String GROUP_ADMIN = "group_admin";
    private static final String GROUP_CREATE = "group_create";
    private static final String GROUP_READ = "group_read";
   private static final String GROUP_WRITE = "group_write";
 
   private static User USER_GROUP_ADMIN;
   private static User USER_GROUP_CREATE;
   private static User USER_GROUP_READ;
   private static User USER_GROUP_WRITE;
 
   // TODO: convert this test to cover the full matrix in
   // https://hbase.apache.org/book/appendix_acl_matrix.html
   // creating all Scope x Permission combinations
 
   private static byte[] TEST_FAMILY = Bytes.toBytes("f1");
 
   private static MasterCoprocessorEnvironment CP_ENV;
   private static AccessController ACCESS_CONTROLLER;
   private static RegionServerCoprocessorEnvironment RSCP_ENV;
   private static RegionCoprocessorEnvironment RCP_ENV;
   
   private static boolean callSuperTwice = true;
 
   // class with faulty stop() method, controlled by flag
   public static class FaultyAccessController extends AccessController {
     public FaultyAccessController() {
     }
 
     @Override
     public void stop(CoprocessorEnvironment env) {
       super.stop(env);
       if (callSuperTwice) {
         super.stop(env);
       }
     }
   }
 
   @BeforeClass
   public static void setupBeforeClass() throws Exception {
     // setup configuration
     conf = TEST_UTIL.getConfiguration();
     // Up the handlers; this test needs more than usual.
     conf.setInt(HConstants.REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT, 10);
     // Enable security
     enableSecurity(conf);
     String accessControllerClassName = FaultyAccessController.class.getName();
     // In this particular test case, we can't use SecureBulkLoadEndpoint because its doAs will fail
     // to move a file for a random user
     conf.set(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY, accessControllerClassName);
     // Verify enableSecurity sets up what we require
     verifyConfiguration(conf);
 
     // Enable EXEC permission checking
     conf.setBoolean(AccessControlConstants.EXEC_PERMISSION_CHECKS_KEY, true);
 
     TEST_UTIL.startMiniCluster();
     MasterCoprocessorHost cpHost =
       TEST_UTIL.getMiniHBaseCluster().getMaster().getMasterCoprocessorHost();
     cpHost.load(FaultyAccessController.class, Coprocessor.PRIORITY_HIGHEST, conf);
     ACCESS_CONTROLLER = (AccessController) cpHost.findCoprocessor(accessControllerClassName);
     CP_ENV = cpHost.createEnvironment(AccessController.class, ACCESS_CONTROLLER,
       Coprocessor.PRIORITY_HIGHEST, 1, conf);
     RegionServerCoprocessorHost rsHost;
     do {
       rsHost = TEST_UTIL.getMiniHBaseCluster().getRegionServer(0)
           .getRegionServerCoprocessorHost();
     } while (rsHost == null);
     RSCP_ENV = rsHost.createEnvironment(AccessController.class, ACCESS_CONTROLLER,
       Coprocessor.PRIORITY_HIGHEST, 1, conf);
 
     // Wait for the ACL table to become available
     TEST_UTIL.waitUntilAllRegionsAssigned(AccessControlLists.ACL_TABLE_NAME);
 
     // create a set of test users
     SUPERUSER = User.createUserForTesting(conf, "admin", new String[] { "supergroup" });
     USER_ADMIN = User.createUserForTesting(conf, "admin2", new String[0]);
     USER_RW = User.createUserForTesting(conf, "rwuser", new String[0]);
     USER_RO = User.createUserForTesting(conf, "rouser", new String[0]);
     USER_OWNER = User.createUserForTesting(conf, "owner", new String[0]);
     USER_CREATE = User.createUserForTesting(conf, "tbl_create", new String[0]);
     USER_NONE = User.createUserForTesting(conf, "nouser", new String[0]);
     USER_ADMIN_CF = User.createUserForTesting(conf, "col_family_admin", new String[0]);
 
     USER_GROUP_ADMIN =
         User.createUserForTesting(conf, "user_group_admin", new String[] { GROUP_ADMIN });
     USER_GROUP_CREATE =
         User.createUserForTesting(conf, "user_group_create", new String[] { GROUP_CREATE });
     USER_GROUP_READ =
         User.createUserForTesting(conf, "user_group_read", new String[] { GROUP_READ });
     USER_GROUP_WRITE =
         User.createUserForTesting(conf, "user_group_write", new String[] { GROUP_WRITE });
 
     systemUserConnection = TEST_UTIL.getConnection();
     setUpTableAndUserPermissions();
   }
 
   @AfterClass
   public static void tearDownAfterClass() throws Exception {
     HRegionServer rs = null;
     for (JVMClusterUtil.RegionServerThread thread:
       TEST_UTIL.getMiniHBaseCluster().getRegionServerThreads()) {
       rs = thread.getRegionServer();
     }
     cleanUp();
     TEST_UTIL.shutdownMiniCluster();
     assertTrue("region server should have aborted due to FaultyAccessController", rs.isAborted());
   }
 
   private static void setUpTableAndUserPermissions() throws Exception {
     HTableDescriptor htd = new HTableDescriptor(TEST_TABLE);
     HColumnDescriptor hcd = new HColumnDescriptor(TEST_FAMILY);
     hcd.setMaxVersions(100);
     htd.addFamily(hcd);
     htd.setOwner(USER_OWNER);
     createTable(TEST_UTIL, htd, new byte[][] { Bytes.toBytes("s") });
 
     Region region = TEST_UTIL.getHBaseCluster().getRegions(TEST_TABLE).get(0);
     RegionCoprocessorHost rcpHost = region.getCoprocessorHost();
     RCP_ENV = rcpHost.createEnvironment(AccessController.class, ACCESS_CONTROLLER,
       Coprocessor.PRIORITY_HIGHEST, 1, conf);
 
     // Set up initial grants
 
     grantGlobal(TEST_UTIL, USER_ADMIN.getShortName(),
       Permission.Action.ADMIN,
       Permission.Action.CREATE,
       Permission.Action.READ,
       Permission.Action.WRITE);
 
     grantOnTable(TEST_UTIL, USER_RW.getShortName(),
       TEST_TABLE, TEST_FAMILY, null,
       Permission.Action.READ,
       Permission.Action.WRITE);
 
     // USER_CREATE is USER_RW plus CREATE permissions
     grantOnTable(TEST_UTIL, USER_CREATE.getShortName(),
       TEST_TABLE, null, null,
       Permission.Action.CREATE,
       Permission.Action.READ,
       Permission.Action.WRITE);
 
     grantOnTable(TEST_UTIL, USER_RO.getShortName(),
       TEST_TABLE, TEST_FAMILY, null,
       Permission.Action.READ);
 
     grantOnTable(TEST_UTIL, USER_ADMIN_CF.getShortName(),
       TEST_TABLE, TEST_FAMILY,
       null, Permission.Action.ADMIN, Permission.Action.CREATE);
 
     grantGlobal(TEST_UTIL, toGroupEntry(GROUP_ADMIN), Permission.Action.ADMIN);
     grantGlobal(TEST_UTIL, toGroupEntry(GROUP_CREATE), Permission.Action.CREATE);
     grantGlobal(TEST_UTIL, toGroupEntry(GROUP_READ), Permission.Action.READ);
     grantGlobal(TEST_UTIL, toGroupEntry(GROUP_WRITE), Permission.Action.WRITE);
 
     assertEquals(5, AccessControlLists.getTablePermissions(conf, TEST_TABLE).size());
     try {
       assertEquals(5, AccessControlClient.getUserPermissions(systemUserConnection,
           TEST_TABLE.toString()).size());
     } catch (Throwable e) {
       LOG.error("error during call of AccessControlClient.getUserPermissions. ", e);
     }
   }
 
   private static void cleanUp() throws Exception {
     // Clean the _acl_ table
     try {
       deleteTable(TEST_UTIL, TEST_TABLE);
     } catch (TableNotFoundException ex) {
       // Test deleted the table, no problem
       LOG.info("Test deleted table " + TEST_TABLE);
     }
     // Verify all table/namespace permissions are erased
     assertEquals(0, AccessControlLists.getTablePermissions(conf, TEST_TABLE).size());
     assertEquals(
       0,
       AccessControlLists.getNamespacePermissions(conf,
         TEST_TABLE.getNamespaceAsString()).size());
   }
 
   @Test
   public void testTableCreate() throws Exception {
     AccessTestAction createTable = new AccessTestAction() {
       @Override
       public Object run() throws Exception {
         HTableDescriptor htd = new HTableDescriptor(TableName.valueOf("testnewtable"));
         htd.addFamily(new HColumnDescriptor(TEST_FAMILY));
         ACCESS_CONTROLLER.preCreateTable(ObserverContext.createAndPrepare(CP_ENV, null), htd, null);
         return null;
       }
     };
 
     // verify that superuser can create tables
     verifyAllowed(createTable, SUPERUSER, USER_ADMIN, USER_GROUP_CREATE);
 
     // all others should be denied
     verifyDenied(createTable, USER_CREATE, USER_RW, USER_RO, USER_NONE, USER_GROUP_ADMIN,
       USER_GROUP_READ, USER_GROUP_WRITE);
   }
 
 }