org.apache.hadoop.hbase.security.User#runAs ( )源码实例Demo

/**
 * Create a new {@link AsyncClusterConnection} instance.
 * <p/>
 * Unlike what we have done in {@link ConnectionFactory}, here we just return an
 * {@link AsyncClusterConnection} instead of a {@link java.util.concurrent.CompletableFuture},
 * which means this method could block on fetching the cluster id. This is just used to simplify
 * the implementation, as when starting new region servers, we do not need to be event-driven. Can
 * change later if we want a {@link java.util.concurrent.CompletableFuture} here.
 */
public static AsyncClusterConnection createAsyncClusterConnection(Configuration conf,
    SocketAddress localAddress, User user) throws IOException {
  ConnectionRegistry registry = ConnectionRegistryFactory.getRegistry(conf);
  String clusterId = FutureUtils.get(registry.getClusterId());
  Class<? extends AsyncClusterConnection> clazz =
    conf.getClass(HBASE_SERVER_CLUSTER_CONNECTION_IMPL, AsyncClusterConnectionImpl.class,
      AsyncClusterConnection.class);
  try {
    return user
      .runAs((PrivilegedExceptionAction<? extends AsyncClusterConnection>) () -> ReflectionUtils
        .newInstance(clazz, conf, registry, clusterId, localAddress, user));
  } catch (Exception e) {
    throw new IOException(e);
  }
}
 

/** This fails only in case of ADE or empty list for any of the actions. */
public static void verifyAllowed(User user, AccessTestAction... actions) throws Exception {
  for (AccessTestAction action : actions) {
    try {
      Object obj = user.runAs(action);
      if (obj != null && obj instanceof List<?>) {
        List<?> results = (List<?>) obj;
        if (results != null && results.isEmpty()) {
          fail("Empty non null results from action for user '" + user.getShortName() + "'");
        }
      }
    } catch (AccessDeniedException ade) {
      fail("Expected action to pass for user '" + user.getShortName() + "' but was denied");
    }
  }
}
 

private void verifiedDeniedServiceException(User user, Action action) throws Exception {
  user.runAs((PrivilegedExceptionAction<?>) () -> {
    boolean accessDenied = false;
    try (Connection conn = ConnectionFactory.createConnection(conf);
        Admin admin = conn.getAdmin()) {
      action.run(admin);
    } catch (ServiceException e) {
      // For MasterRpcServices.execService.
      if (e.getCause() instanceof AccessDeniedException) {
        accessDenied = true;
      }
    }
    assertTrue("Expected access to be denied", accessDenied);
    return null;
  });

}
 

private void verifyUserDeniedForDeleteMultipleVersions(final User user, final byte[] row,
    final byte[] q1, final byte[] q2) throws IOException, InterruptedException {
  user.runAs(new PrivilegedExceptionAction<Void>() {
    @Override
    public Void run() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(conf)) {
        try (Table t = connection.getTable(testTable.getTableName())) {
          Delete d = new Delete(row);
          d.addColumns(TEST_FAMILY1, q1);
          d.addColumns(TEST_FAMILY1, q2);
          t.delete(d);
          fail(user.getShortName() + " should not be allowed to delete the row");
        } catch (Exception e) {

        }
      }
      return null;
    }
  });
}
 

private void verifyUserDeniedForDeleteExactVersion(final User user, final byte[] row,
    final byte[] q1, final byte[] q2) throws IOException, InterruptedException {
  user.runAs(new PrivilegedExceptionAction<Void>() {
    @Override
    public Void run() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(conf)) {
        try (Table t = connection.getTable(testTable.getTableName())) {
          Delete d = new Delete(row, 127);
          d.addColumns(TEST_FAMILY1, q1);
          d.addColumns(TEST_FAMILY1, q2);
          d.addFamily(TEST_FAMILY2, 129);
          t.delete(d);
          fail(user.getShortName() + " can not do the delete");
        } catch (Exception e) {

        }
      }
      return null;
    }
  });
}
 

private void verifyUserDeniedForIncrementMultipleVersions(final User user, final byte[] row,
    final byte[] q1) throws IOException, InterruptedException {
  user.runAs(new PrivilegedExceptionAction<Void>() {
    @Override
    public Void run() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(conf)) {
        try (Table t = connection.getTable(testTable.getTableName())) {
          Increment inc = new Increment(row);
          inc.setTimeRange(0, 127);
          inc.addColumn(TEST_FAMILY1, q1, 2L);
          t.increment(inc);
          fail(user.getShortName() + " cannot do the increment.");
        } catch (Exception e) {

        }
      }
      return null;
    }
  });
}
 

private void verifyUserDeniedForPutMultipleVersions(final User user, final byte[] row,
    final byte[] q1, final byte[] q2, final byte[] value) throws IOException,
    InterruptedException {
  user.runAs(new PrivilegedExceptionAction<Void>() {
    @Override
    public Void run() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(conf)) {
        try (Table t = connection.getTable(testTable.getTableName())) {
          Put p = new Put(row);
          // column Q1 covers version at 123 fr which user2 do not have permission
          p.addColumn(TEST_FAMILY1, q1, 124, value);
          p.addColumn(TEST_FAMILY1, q2, value);
          t.put(p);
          fail(user.getShortName() + " cannot do the put.");
        } catch (Exception e) {

        }
      }
      return null;
    }
  });
}
 

private void verifyUserAllowedforCheckAndDelete(final User user, final byte[] row,
    final byte[] q1, final byte[] value) throws IOException, InterruptedException {
  user.runAs(new PrivilegedExceptionAction<Void>() {
    @Override
    public Void run() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(conf)) {
        try (Table t = connection.getTable(testTable.getTableName())) {
          Delete d = new Delete(row);
          d.addColumn(TEST_FAMILY1, q1, 120);
          t.checkAndMutate(row, TEST_FAMILY1).qualifier(q1).ifEquals(value).thenDelete(d);
        }
      }
      return null;
    }
  });
}
 

private void verifyUserDeniedForCheckAndDelete(final User user, final byte[] row,
    final byte[] value) throws IOException, InterruptedException {
  user.runAs(new PrivilegedExceptionAction<Void>() {
    @Override
    public Void run() throws Exception {
      try (Connection connection = ConnectionFactory.createConnection(conf)) {
        try (Table t = connection.getTable(testTable.getTableName())) {
          Delete d = new Delete(row);
          d.addColumns(TEST_FAMILY1, TEST_Q1);
          t.checkAndMutate(row, TEST_FAMILY1).qualifier(TEST_Q1).ifEquals(value).thenDelete(d);
          fail(user.getShortName() + " should not be allowed to do checkAndDelete");
        } catch (Exception e) {
        }
      }
      return null;
    }
  });
}
 

@Test
public void testAccessControlClientUserPerms() throws Exception {
  final TableName tableName = TableName.valueOf(name.getMethodName());
  createTestTable(tableName);
  try {
    final String regex = tableName.getNameWithNamespaceInclAsString();
    User testUserPerms = User.createUserForTesting(conf, "testUserPerms", new String[0]);
    assertEquals(0, testUserPerms.runAs(getPrivilegedAction(regex)).size());
    // Grant TABLE ADMIN privs to testUserPerms
    grantOnTable(TEST_UTIL, testUserPerms.getShortName(), tableName, null, null, Action.ADMIN);
    List<UserPermission> perms = testUserPerms.runAs(getPrivilegedAction(regex));
    assertNotNull(perms);
    // Superuser, testUserPerms
    assertEquals(2, perms.size());
  } finally {
    deleteTable(TEST_UTIL, tableName);
  }
}
 

public static <T, E extends Exception> T execute(SupplierException<T, E> action, SecurityContext securityContext, User user)
        throws E, PrivilegedActionException, IOException, InterruptedException {
    if (user != null && SecurityUtil.isKerberosAuthenticated(securityContext)) {
        LOG.debug("Executing action [{}] for user [{}] with security context [{}] using Kerberos authentication",
                action, securityContext, user);
        return user.runAs((PrivilegedExceptionAction<T>) action::get);
    } else {
        LOG.debug("Executing action [{}] for user [{}] with security context [{}] without Kerberos authentication",
                action, securityContext, user);
        return action.get();
    }
}
 

SecureWriter(final Configuration conf, final UserProvider userProvider,
    final Token userToken, final List<SequenceFile.Writer.Option> opts)
    throws IOException {
  User user = getActiveUser(userProvider, userToken);
  try {
    SequenceFile.Writer sequenceFileWriter =
        user.runAs((PrivilegedExceptionAction<SequenceFile.Writer>) () ->
            SequenceFile.createWriter(conf,
                opts.toArray(new SequenceFile.Writer.Option[opts.size()])));
    privilegedWriter = new PrivilegedWriter(user, sequenceFileWriter);
  } catch (InterruptedException e) {
    throw new IOException(e);
  }
}
 

public JVMClusterUtil.RegionServerThread addRegionServer(
    final Configuration config, final int index, User user)
throws IOException, InterruptedException {
  return user.runAs(
      new PrivilegedExceptionAction<JVMClusterUtil.RegionServerThread>() {
        @Override
        public JVMClusterUtil.RegionServerThread run() throws Exception {
          return addRegionServer(config, index);
        }
      });
}
 

public JVMClusterUtil.MasterThread addMaster(
    final Configuration c, final int index, User user)
throws IOException, InterruptedException {
  return user.runAs(
      new PrivilegedExceptionAction<JVMClusterUtil.MasterThread>() {
        @Override
        public JVMClusterUtil.MasterThread run() throws Exception {
          return addMaster(c, index);
        }
      });
}
 

private void verifyAllowed(User user, AccessTestAction... actions) throws Exception {
    for (AccessTestAction action : actions) {
        try {
            Object obj = user.runAs(action);
            if (obj != null && obj instanceof List<?>) {
                List<?> results = (List<?>) obj;
                if (results.isEmpty()) {
                    fail("Empty non null results from action for user '" + user.getShortName() + "'");
                }
            }
        } catch (AccessDeniedException ade) {
            fail("Expected action to pass for user '" + user.getShortName() + "' but was denied");
        }
    }
}
 

/** This passes only in case of null for all users. */
public static void verifyIfNull(AccessTestAction  action, User... users) throws Exception {
  for (User user : users) {
    try {
      Object obj = user.runAs(action);
      if (obj != null) {
        fail("Non null results from action for user '" + user.getShortName() + "' : " + obj);
      }
    } catch (AccessDeniedException ade) {
      fail("Expected action to pass for user '" + user.getShortName() + "' but was denied");
    }
  }
}
 

private void verifyAllowed(User user, Action action) throws Exception {
  user.runAs((PrivilegedExceptionAction<?>) () -> {
    try (Connection conn = ConnectionFactory.createConnection(conf);
        Admin admin = conn.getAdmin()) {
      action.run(admin);
    } catch (IOException e) {
      fail(e.toString());
    }
    return null;
  });
}
 

private void verifyDenied(User user, Action action) throws Exception {
  user.runAs((PrivilegedExceptionAction<?>) () -> {
    boolean accessDenied = false;
    try (Connection conn = ConnectionFactory.createConnection(conf);
        Admin admin = conn.getAdmin()) {
      action.run(admin);
    } catch (AccessDeniedException e) {
      accessDenied = true;
    }
    assertTrue("Expected access to be denied", accessDenied);
    return null;
  });
}
 

@Test
public void test() throws Exception {
  TableName tableName = TableName.valueOf("test-table");
  String userName1 = "user1";
  String userName2 = "user2";
  User user2 = User.createUserForTesting(TEST_UTIL.getConfiguration(), userName2, new String[0]);
  Permission permission =
      Permission.newBuilder(tableName).withActions(Permission.Action.READ).build();
  UserPermission userPermission = new UserPermission(userName1, permission);

  // grant user1 table permission
  admin.grant(userPermission, false).get();

  // get table permissions
  List<UserPermission> userPermissions =
      admin.getUserPermissions(GetUserPermissionsRequest.newBuilder(tableName).build()).get();
  assertEquals(1, userPermissions.size());
  assertEquals(userPermission, userPermissions.get(0));

  // get table permissions
  userPermissions =
      admin
          .getUserPermissions(
            GetUserPermissionsRequest.newBuilder(tableName).withUserName(userName1).build())
          .get();
  assertEquals(1, userPermissions.size());
  assertEquals(userPermission, userPermissions.get(0));

  userPermissions =
      admin
          .getUserPermissions(
            GetUserPermissionsRequest.newBuilder(tableName).withUserName(userName2).build())
          .get();
  assertEquals(0, userPermissions.size());

  // has user permission
  List<Permission> permissions = Lists.newArrayList(permission);
  boolean hasPermission =
      admin.hasUserPermissions(userName1, permissions).get().get(0).booleanValue();
  assertTrue(hasPermission);
  hasPermission = admin.hasUserPermissions(userName2, permissions).get().get(0).booleanValue();
  assertFalse(hasPermission);

  AccessTestAction hasPermissionAction = new AccessTestAction() {
    @Override
    public Object run() throws Exception {
      try (AsyncConnection conn =
          ConnectionFactory.createAsyncConnection(TEST_UTIL.getConfiguration()).get()) {
        return conn.getAdmin().hasUserPermissions(userName1, permissions).get().get(0);
      }
    }
  };
  try {
    user2.runAs(hasPermissionAction);
    fail("Should not come here");
  } catch (Exception e) {
    LOG.error("Call has permission error", e);
  }

  // check permission
  admin.hasUserPermissions(permissions);
  AccessTestAction checkPermissionsAction = new AccessTestAction() {
    @Override
    public Object run() throws Exception {
      try (AsyncConnection conn =
          ConnectionFactory.createAsyncConnection(TEST_UTIL.getConfiguration()).get()) {
        return conn.getAdmin().hasUserPermissions(permissions).get().get(0);
      }
    }
  };
  assertFalse((Boolean) user2.runAs(checkPermissionsAction));
}
 

@Test
public void testHandleErrorsInFlush() throws Exception {
  LOG.info("Setting up a faulty file system that cannot write");

  final Configuration conf = HBaseConfiguration.create(TEST_UTIL.getConfiguration());
  User user = User.createUserForTesting(conf,
      "testhandleerrorsinflush", new String[]{"foo"});
  // Inject our faulty LocalFileSystem
  conf.setClass("fs.file.impl", FaultyFileSystem.class,
      FileSystem.class);
  user.runAs(new PrivilegedExceptionAction<Object>() {
    @Override
    public Object run() throws Exception {
      // Make sure it worked (above is sensitive to caching details in hadoop core)
      FileSystem fs = FileSystem.get(conf);
      assertEquals(FaultyFileSystem.class, fs.getClass());

      // Initialize region
      init(name.getMethodName(), conf);

      LOG.info("Adding some data");
      store.add(new KeyValue(row, family, qf1, 1, (byte[])null), null);
      store.add(new KeyValue(row, family, qf2, 1, (byte[])null), null);
      store.add(new KeyValue(row, family, qf3, 1, (byte[])null), null);

      LOG.info("Before flush, we should have no files");

      Collection<StoreFileInfo> files =
        store.getRegionFileSystem().getStoreFiles(store.getColumnFamilyName());
      assertEquals(0, files != null ? files.size() : 0);

      //flush
      try {
        LOG.info("Flushing");
        flush(1);
        fail("Didn't bubble up IOE!");
      } catch (IOException ioe) {
        assertTrue(ioe.getMessage().contains("Fault injected"));
      }

      LOG.info("After failed flush, we should still have no files!");
      files = store.getRegionFileSystem().getStoreFiles(store.getColumnFamilyName());
      assertEquals(0, files != null ? files.size() : 0);
      store.getHRegion().getWAL().close();
      return null;
    }
  });
  FileSystem.closeAllForUGI(user.getUGI());
}