HBASE-11077 [AccessController] Restore compatible early-out access denial

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1591524 13f79535-47bb-0310-9956-ffa450edef68

hbase-client/src/main/java/org/apache/hadoop/hbase/security/access/AccessControlConstants.java

@@ -25,6 +25,22 @@ import org.apache.hadoop.classification.InterfaceStability;
 @InterfaceStability.Evolving
 public interface AccessControlConstants {
 
+  /**
+   * Configuration option that toggles whether EXEC permission checking is
+   * performed during coprocessor endpoint invocations.
+   */
+  public static final String EXEC_PERMISSION_CHECKS_KEY = "hbase.security.exec.permission.checks";
+  /** Default setting for hbase.security.exec.permission.checks; false */
+  public static final boolean DEFAULT_EXEC_PERMISSION_CHECKS = false;
+
+  /**
+   * Configuration or CF schema option for early termination of access checks
+   * if table or CF permissions grant access. Pre-0.98 compatible behavior
+   */
+  public static final String CF_ATTRIBUTE_EARLY_OUT = "hbase.security.access.early_out";
+  /** Default setting for hbase.security.access.early_out */
+  public static final boolean DEFAULT_ATTRIBUTE_EARLY_OUT = true;
+
   // Operation attributes for cell level security
 
   /** Cell level ACL */

hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessControlFilter.java

@@ -22,10 +22,10 @@ import java.io.IOException;
 import java.util.Map;
 
 import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.exceptions.DeserializationException;
 import org.apache.hadoop.hbase.filter.FilterBase;
-import org.apache.hadoop.hbase.filter.Filter.ReturnCode;
 import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.util.ByteRange;
 import org.apache.hadoop.hbase.util.Bytes;
@@ -48,11 +48,20 @@ import org.apache.hadoop.hbase.util.SimpleByteRange;
  */
 class AccessControlFilter extends FilterBase {
 
+  public static enum Strategy {
+    /** Filter only by checking the table or CF permissions */
+    CHECK_TABLE_AND_CF_ONLY,
+    /** Cell permissions can override table or CF permissions */
+    CHECK_CELL_DEFAULT,
+    /** Cell permissions must authorize */
+    CHECK_CELL_FIRST,
+  };
+
   private TableAuthManager authManager;
   private TableName table;
   private User user;
   private boolean isSystemTable;
-  private boolean cellFirstStrategy;
+  private Strategy strategy;
   private Map<ByteRange, Integer> cfVsMaxVersions;
   private int familyMaxVersions;
   private int currentVersions;
@@ -66,12 +75,12 @@ class AccessControlFilter extends FilterBase {
   }
 
   AccessControlFilter(TableAuthManager mgr, User ugi, TableName tableName,
-      boolean cellFirstStrategy, Map<ByteRange, Integer> cfVsMaxVersions) {
+      Strategy strategy, Map<ByteRange, Integer> cfVsMaxVersions) {
     authManager = mgr;
     table = tableName;
     user = ugi;
     isSystemTable = tableName.isSystemTable();
-    this.cellFirstStrategy = cellFirstStrategy;
+    this.strategy = strategy;
     this.cfVsMaxVersions = cfVsMaxVersions;
     this.prevFam = new SimpleByteRange();
     this.prevQual = new SimpleByteRange();
@@ -103,12 +112,37 @@ class AccessControlFilter extends FilterBase {
     if (currentVersions > familyMaxVersions) {
       return ReturnCode.SKIP;
     }
-    if (authManager.authorize(user, table, cell, cellFirstStrategy, Permission.Action.READ)) {
+    // XXX: Compare in place, don't clone
-      return ReturnCode.INCLUDE;
+    byte[] family = CellUtil.cloneFamily(cell);
+    byte[] qualifier = CellUtil.cloneQualifier(cell);
+    switch (strategy) {
+      // Filter only by checking the table or CF permissions
+      case CHECK_TABLE_AND_CF_ONLY: {
+        if (authManager.authorize(user, table, family, qualifier, Permission.Action.READ)) {
+          return ReturnCode.INCLUDE;
+        }
+      }
+      break;
+      // Cell permissions can override table or CF permissions
+      case CHECK_CELL_DEFAULT: {
+        if (authManager.authorize(user, table, family, qualifier, Permission.Action.READ) ||
+            authManager.authorize(user, table, cell, Permission.Action.READ)) {
+          return ReturnCode.INCLUDE;
+        }
+      }
+      break;
+      // Cell permissions must authorize
+      case CHECK_CELL_FIRST: {
+        if (authManager.authorize(user, table, cell, Permission.Action.READ) &&
+            authManager.authorize(user, table, family, qualifier, Permission.Action.READ)) {
+          return ReturnCode.INCLUDE;
+        }
+      }
+      break;
+      default:
+        throw new RuntimeException("Unhandled strategy " + strategy);
     }
-    // Before per cell ACLs we used to return the NEXT_COL hint, but we can
+
-    // no longer do that since, given the possibility of per cell ACLs
-    // anywhere, we now need to examine all KVs with this filter.
     return ReturnCode.SKIP;
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AuthResult.java

@@ -33,12 +33,12 @@ import org.apache.hadoop.hbase.util.Bytes;
  */
 @InterfaceAudience.Private
 public class AuthResult {
-  private final boolean allowed;
+  private boolean allowed;
   private final String namespace;
   private final TableName table;
   private final Permission.Action action;
   private final String request;
-  private final String reason;
+  private String reason;
   private final User user;
 
   // "family" and "qualifier" should only be used if "families" is null.
@@ -121,6 +121,14 @@ public class AuthResult {
     return request;
   }
 
+  public void setAllowed(boolean allowed) {
+    this.allowed = allowed;
+  }
+
+  public void setReason(String reason) {
+    this.reason = reason;
+  }
+
   String toFamilyString() {
     StringBuilder sb = new StringBuilder();
     if (families != null) {

hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/TableAuthManager.java

@@ -28,7 +28,6 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.exceptions.DeserializationException;
 import org.apache.hadoop.hbase.security.User;
@@ -348,12 +347,14 @@ public class TableAuthManager {
     return false;
   }
 
-  private boolean checkCellPermissions(User user, Cell cell, Permission.Action action) {
+  /**
+   * Authorize a user for a given KV. This is called from AccessControlFilter.
+   */
+  public boolean authorize(User user, TableName table, Cell cell, Permission.Action action) {
     try {
       List<Permission> perms = AccessControlLists.getCellPermissionsForUser(user, cell);
       if (LOG.isTraceEnabled()) {
-        LOG.trace("Perms for user " + user.getShortName() + " in cell " +
+        LOG.trace("Perms for user " + user.getShortName() + " in cell " + cell + ": " + perms);
-          cell + ": " + perms);
       }
       for (Permission p: perms) {
         if (p.implies(action)) {
@@ -369,46 +370,6 @@ public class TableAuthManager {
     return false;
   }
 
-  private boolean checkTableColumnPermissions(User user, TableName table, Cell cell,
-      Permission.Action action) {
-    // TODO: Do not clone here
-    byte[] family = CellUtil.cloneFamily(cell);
-    byte[] qualifier = CellUtil.cloneQualifier(cell);
-    // User is authorized at table or CF level
-    if (authorizeUser(user, table, family, qualifier, action)) {
-      return true;
-    }
-    String groupNames[] = user.getGroupNames();
-    if (groupNames != null) {
-      for (String group: groupNames) {
-        // TODO: authorizeGroup should check qualifier too?
-        // Group is authorized at table or CF level
-        if (authorizeGroup(group, table, family, action)) {
-          return true;
-        }
-      }
-    }
-    return false;
-  }
-
-  /**
-   * Authorize a user for a given KV. This is called from AccessControlFilter.
-   */
-  public boolean authorize(User user, TableName table, Cell cell, boolean cellFirstStrategy,
-      Permission.Action action) {
-    if (cellFirstStrategy) {
-      if (checkCellPermissions(user, cell, action)) {
-        return true;
-      }
-      return checkTableColumnPermissions(user, table, cell, action);
-    } else {
-      if (checkTableColumnPermissions(user, table, cell, action)) {
-        return true;
-      }
-      return checkCellPermissions(user, cell, action);
-    }
-  }
-
   public boolean authorize(User user, String namespace, Permission.Action action) {
     // Global authorizations supercede namespace level
     if (authorizeUser(user, action)) {

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/SecureTestUtil.java

@@ -41,6 +41,7 @@ import org.apache.hadoop.hbase.Waiter.Predicate;
 import org.apache.hadoop.hbase.client.HTable;
 import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
 import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
+import org.apache.hadoop.hbase.io.hfile.HFile;
 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
 import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.AccessControlService;
 import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.CheckPermissionsRequest;
@@ -84,7 +85,7 @@ public class SecureTestUtil {
     }
     conf.set("hbase.superuser", sb.toString());
     // Need HFile V3 for tags for security features
-    conf.setInt("hfile.format.version", 3);
+    conf.setInt(HFile.FORMAT_VERSION_KEY, 3);
   }
 
   public static void verifyConfiguration(Configuration conf) {
@@ -96,9 +97,12 @@ public class SecureTestUtil {
         AccessController.class.getName()))) {
       throw new RuntimeException("AccessController is missing from a system coprocessor list");
     }
+    if (conf.getInt(HFile.FORMAT_VERSION_KEY, 2) < HFile.MIN_FORMAT_VERSION_WITH_TAGS) {
+      throw new RuntimeException("Post 0.96 security features require HFile version >= 3");
+    }
   }
 
-  public void checkTablePerms(Configuration conf, byte[] table, byte[] family, byte[] column,
+  public static void checkTablePerms(Configuration conf, byte[] table, byte[] family, byte[] column,
       Permission.Action... actions) throws IOException {
     Permission[] perms = new Permission[actions.length];
     for (int i = 0; i < actions.length; i++) {
@@ -108,7 +112,7 @@ public class SecureTestUtil {
     checkTablePerms(conf, table, perms);
   }
 
-  public void checkTablePerms(Configuration conf, byte[] table, Permission... perms) throws IOException {
+  public static void checkTablePerms(Configuration conf, byte[] table, Permission... perms) throws IOException {
     CheckPermissionsRequest.Builder request = CheckPermissionsRequest.newBuilder();
     for (Permission p : perms) {
       request.addPermission(ProtobufUtil.toPermission(p));
@@ -139,7 +143,7 @@ public class SecureTestUtil {
    */
   static interface AccessTestAction extends PrivilegedExceptionAction<Object> { }
 
-  public void verifyAllowed(User user, AccessTestAction... actions) throws Exception {
+  public static void verifyAllowed(User user, AccessTestAction... actions) throws Exception {
     for (AccessTestAction action : actions) {
       try {
         Object obj = user.runAs(action);
@@ -155,13 +159,13 @@ public class SecureTestUtil {
     }
   }
 
-  public void verifyAllowed(AccessTestAction action, User... users) throws Exception {
+  public static void verifyAllowed(AccessTestAction action, User... users) throws Exception {
     for (User user : users) {
       verifyAllowed(user, action);
     }
   }
 
-  public void verifyAllowed(User user, AccessTestAction action, int count) throws Exception {
+  public static void verifyAllowed(User user, AccessTestAction action, int count) throws Exception {
     try {
       Object obj = user.runAs(action);
       if (obj != null && obj instanceof List<?>) {
@@ -176,14 +180,34 @@ public class SecureTestUtil {
     }
   }
 
-  public void verifyDenied(User user, AccessTestAction... actions) throws Exception {
+  public static void verifyDeniedWithException(User user, AccessTestAction... actions)
+      throws Exception {
+    verifyDenied(user, true, actions);
+  }
+
+  public static void verifyDeniedWithException(AccessTestAction action, User... users)
+      throws Exception {
+    for (User user : users) {
+      verifyDenied(user, true, action);
+    }
+  }
+
+  public static void verifyDenied(User user, AccessTestAction... actions) throws Exception {
+    verifyDenied(user, false, actions);
+  }
+
+  public static void verifyDenied(User user, boolean requireException,
+      AccessTestAction... actions) throws Exception {
     for (AccessTestAction action : actions) {
       try {
         Object obj = user.runAs(action);
+        if (requireException) {
+          fail("Expected exception was not thrown for user '" + user.getShortName() + "'");
+        }
         if (obj != null && obj instanceof List<?>) {
           List<?> results = (List<?>) obj;
           if (results != null && !results.isEmpty()) {
-            fail("Expected no results for user '" + user.getShortName() + "'");
+            fail("Unexpected results for user '" + user.getShortName() + "'");
           }
         }
       } catch (IOException e) {
@@ -211,7 +235,7 @@ public class SecureTestUtil {
           } while((ex = ex.getCause()) != null);
         }
         if (!isAccessDeniedException) {
-          fail("Not receiving AccessDeniedException for user '" + user.getShortName() + "'");
+          fail("Expected exception was not thrown for user '" + user.getShortName() + "'");
         }
       } catch (UndeclaredThrowableException ute) {
         // TODO why we get a PrivilegedActionException, which is unexpected?
@@ -226,12 +250,12 @@ public class SecureTestUtil {
             return;
           }
         }
-        fail("Not receiving AccessDeniedException for user '" + user.getShortName() + "'");
+        fail("Expected exception was not thrown for user '" + user.getShortName() + "'");
       }
     }
   }
 
-  public void verifyDenied(AccessTestAction action, User... users) throws Exception {
+  public static void verifyDenied(AccessTestAction action, User... users) throws Exception {
     for (User user : users) {
       verifyDenied(user, action);
     }

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestAccessControlFilter.java

@@ -47,7 +47,7 @@ import org.junit.experimental.categories.Category;
 import org.junit.rules.TestName;
 
 @Category(LargeTests.class)
-public class TestAccessControlFilter {
+public class TestAccessControlFilter extends SecureTestUtil {
   @Rule public TestName name = new TestName();
   private static HBaseTestingUtility TEST_UTIL;
 
@@ -69,12 +69,14 @@ public class TestAccessControlFilter {
   public static void setupBeforeClass() throws Exception {
     TEST_UTIL = new HBaseTestingUtility();
     Configuration conf = TEST_UTIL.getConfiguration();
-    SecureTestUtil.enableSecurity(conf);
+    enableSecurity(conf);
-    String baseuser = User.getCurrent().getShortName();
+    verifyConfiguration(conf);
-    conf.set("hbase.superuser", conf.get("hbase.superuser", "") +
+
-        String.format(",%s.hfs.0,%s.hfs.1,%s.hfs.2", baseuser, baseuser, baseuser));
+    // We expect 0.98 scanning semantics
+    conf.setBoolean(AccessControlConstants.CF_ATTRIBUTE_EARLY_OUT, false);
+
     TEST_UTIL.startMiniCluster();
-    TEST_UTIL.waitTableEnabled(AccessControlLists.ACL_TABLE_NAME.getName());
+    TEST_UTIL.waitTableEnabled(AccessControlLists.ACL_TABLE_NAME.getName(), 50000);
 
     READER = User.createUserForTesting(conf, "reader", new String[0]);
     LIMITED = User.createUserForTesting(conf, "limited", new String[0]);

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestAccessController.java

@@ -170,7 +170,7 @@ public class TestAccessController extends SecureTestUtil {
     verifyConfiguration(conf);
 
     // Enable EXEC permission checking
-    conf.setBoolean(AccessController.EXEC_PERMISSION_CHECKS_KEY, true);
+    conf.setBoolean(AccessControlConstants.EXEC_PERMISSION_CHECKS_KEY, true);
 
     TEST_UTIL.startMiniCluster();
     MasterCoprocessorHost cpHost =

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestCellACLWithMultipleVersions.java

@@ -45,10 +45,8 @@ import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.TestTableName;
-
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
-
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Before;
@@ -89,18 +87,13 @@ public class TestCellACLWithMultipleVersions extends SecureTestUtil {
   public static void setupBeforeClass() throws Exception {
     // setup configuration
     conf = TEST_UTIL.getConfiguration();
-    conf.set("hbase.master.hfilecleaner.plugins",
-        "org.apache.hadoop.hbase.master.cleaner.HFileLinkCleaner,"
-            + "org.apache.hadoop.hbase.master.snapshot.SnapshotHFileCleaner");
-    conf.set("hbase.master.logcleaner.plugins",
-        "org.apache.hadoop.hbase.master.snapshot.SnapshotLogCleaner");
     // Enable security
     enableSecurity(conf);
     // Verify enableSecurity sets up what we require
     verifyConfiguration(conf);
 
-    // Enable EXEC permission checking
+    // We expect 0.98 cell ACL semantics
-    conf.setBoolean(AccessController.EXEC_PERMISSION_CHECKS_KEY, true);
+    conf.setBoolean(AccessControlConstants.CF_ATTRIBUTE_EARLY_OUT, false);
 
     TEST_UTIL.startMiniCluster();
     MasterCoprocessorHost cpHost = TEST_UTIL.getMiniHBaseCluster().getMaster()

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestCellACLs.java

@@ -43,12 +43,11 @@ import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.master.MasterCoprocessorHost;
 import org.apache.hadoop.hbase.regionserver.RegionServerCoprocessorHost;
 import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.security.access.Permission.Action;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.TestTableName;
-
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
-
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Before;
@@ -79,6 +78,7 @@ public class TestCellACLs extends SecureTestUtil {
   private static final byte[] TEST_Q3 = Bytes.toBytes("q3");
   private static final byte[] TEST_Q4 = Bytes.toBytes("q4");
   private static final byte[] ZERO = Bytes.toBytes(0L);
+  private static final byte[] ONE = Bytes.toBytes(1L);
 
   private static Configuration conf;
 
@@ -89,18 +89,13 @@ public class TestCellACLs extends SecureTestUtil {
   public static void setupBeforeClass() throws Exception {
     // setup configuration
     conf = TEST_UTIL.getConfiguration();
-    conf.set("hbase.master.hfilecleaner.plugins",
-        "org.apache.hadoop.hbase.master.cleaner.HFileLinkCleaner,"
-            + "org.apache.hadoop.hbase.master.snapshot.SnapshotHFileCleaner");
-    conf.set("hbase.master.logcleaner.plugins",
-        "org.apache.hadoop.hbase.master.snapshot.SnapshotLogCleaner");
     // Enable security
     enableSecurity(conf);
     // Verify enableSecurity sets up what we require
     verifyConfiguration(conf);
 
-    // Enable EXEC permission checking
+    // We expect 0.98 cell ACL semantics
-    conf.setBoolean(AccessController.EXEC_PERMISSION_CHECKS_KEY, true);
+    conf.setBoolean(AccessControlConstants.CF_ATTRIBUTE_EARLY_OUT, false);
 
     TEST_UTIL.startMiniCluster();
     MasterCoprocessorHost cpHost = TEST_UTIL.getMiniHBaseCluster().getMaster()
@@ -150,12 +145,11 @@ public class TestCellACLs extends SecureTestUtil {
           Put p;
           // with ro ACL
           p = new Put(TEST_ROW).add(TEST_FAMILY, TEST_Q1, ZERO);
-          p.setACL(USER_OTHER.getShortName(), new Permission(Permission.Action.READ));
+          p.setACL(USER_OTHER.getShortName(), new Permission(Action.READ));
           t.put(p);
           // with rw ACL
           p = new Put(TEST_ROW).add(TEST_FAMILY, TEST_Q2, ZERO);
-          p.setACL(USER_OTHER.getShortName(), new Permission(Permission.Action.READ,
+          p.setACL(USER_OTHER.getShortName(), new Permission(Action.READ, Action.WRITE));
-            Permission.Action.WRITE));
           t.put(p);
           // no ACL
           p = new Put(TEST_ROW)
@@ -308,7 +302,7 @@ public class TestCellACLs extends SecureTestUtil {
       public Object run() throws Exception {
         Increment i = new Increment(TEST_ROW).addColumn(TEST_FAMILY, TEST_Q2, 1L);
         // Tag this increment with an ACL that denies write permissions to USER_OTHER
-        i.setACL(USER_OTHER.getShortName(), new Permission(Permission.Action.READ));
+        i.setACL(USER_OTHER.getShortName(), new Permission(Action.READ));
         HTable t = new HTable(conf, TEST_TABLE.getTableName());
         try {
           t.increment(i);
@@ -379,6 +373,164 @@ public class TestCellACLs extends SecureTestUtil {
     verifyAllowed(deleteQ1, USER_OWNER);
   }
 
+  /**
+   * Insure we are not granting access in the absence of any cells found
+   * when scanning for covered cells.
+   */
+  @Test
+  public void testCoveringCheck() throws Exception {
+    // Grant read access to USER_OTHER
+    grantOnTable(TEST_UTIL, USER_OTHER.getShortName(), TEST_TABLE.getTableName(),
+      TEST_FAMILY, null, Action.READ);
+
+    // A write by USER_OTHER should be denied.
+    // This is where we could have a big problem if there is an error in the
+    // covering check logic.
+    verifyDenied(new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          Put p;
+          p = new Put(TEST_ROW).add(TEST_FAMILY, TEST_Q1, ZERO);
+          t.put(p);
+        } finally {
+          t.close();
+        }
+        return null;
+      }
+    }, USER_OTHER);
+
+    // Add the cell
+    verifyAllowed(new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          Put p;
+          p = new Put(TEST_ROW).add(TEST_FAMILY, TEST_Q1, ZERO);
+          t.put(p);
+        } finally {
+          t.close();
+        }
+        return null;
+      }
+    }, USER_OWNER);
+
+    // A write by USER_OTHER should still be denied, just to make sure
+    verifyDenied(new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          Put p;
+          p = new Put(TEST_ROW).add(TEST_FAMILY, TEST_Q1, ONE);
+          t.put(p);
+        } finally {
+          t.close();
+        }
+        return null;
+      }
+    }, USER_OTHER);
+
+    // A read by USER_OTHER should be allowed, just to make sure
+    verifyAllowed(new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          return t.get(new Get(TEST_ROW).addColumn(TEST_FAMILY, TEST_Q1));
+        } finally {
+          t.close();
+        }
+      }
+    }, USER_OTHER);
+  }
+
+  @Test
+  public void testCellStrategy() throws Exception {
+    // Set up our test actions
+    AccessTestAction readQ1Default = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          return t.get(new Get(TEST_ROW).addColumn(TEST_FAMILY, TEST_Q1));
+        } finally {
+          t.close();
+        }
+      }
+    };
+    AccessTestAction readQ2Default = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          return t.get(new Get(TEST_ROW).addColumn(TEST_FAMILY, TEST_Q2));
+        } finally {
+          t.close();
+        }
+      }
+    };
+    AccessTestAction readQ1CellFirst = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          Get get = new Get(TEST_ROW).addColumn(TEST_FAMILY, TEST_Q1);
+          get.setACLStrategy(true);
+          return t.get(get);
+        } finally {
+          t.close();
+        }
+      }
+    };
+
+    // Add test data
+    verifyAllowed(new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          Put p;
+          // The empty permission set on Q1
+          p = new Put(TEST_ROW).add(TEST_FAMILY, TEST_Q1, ZERO);
+          p.setACL(USER_OTHER.getShortName(), new Permission());
+          t.put(p);
+          // Read permissions on Q2
+          p = new Put(TEST_ROW).add(TEST_FAMILY, TEST_Q2, ZERO);
+          p.setACL(USER_OTHER.getShortName(), new Permission(Action.READ));
+          t.put(p);
+        } finally {
+          t.close();
+        }
+        return null;
+      }
+    }, USER_OWNER);
+
+    // A read by USER_OTHER will be denied with the default cell strategy as
+    // there is no visibility without a grant and a cell ACL giving
+    // explicit permission
+    verifyDenied(readQ1Default, USER_OTHER);
+
+    // A read will be allowed by the default cell strategy if there is a cell
+    // ACL giving explicit permission.
+    verifyAllowed(readQ2Default, USER_OTHER);
+
+    // Grant read access to USER_OTHER
+    grantOnTable(TEST_UTIL, USER_OTHER.getShortName(), TEST_TABLE.getTableName(),
+      TEST_FAMILY, null, Action.READ);
+
+    // A read by USER_OTHER will now be allowed with the default cell strategy
+    // because we have a CF level grant and we take the union of permissions.
+    verifyAllowed(readQ1Default, USER_OTHER);
+
+    // A read by USER_OTHER will be denied with the cell first strategy
+    // because the empty perm set for USER_OTHER in the cell ACL there
+    // revokes access.
+    verifyDenied(readQ1CellFirst, USER_OTHER);
+  }
+
   @After
   public void tearDown() throws Exception {
     // Clean the _acl_ table

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestScanEarlyTermination.java

@@ -0,0 +1,299 @@
+/*
+ * 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.junit.Assert.*;
+
+import java.util.UUID;
+
+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.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.MediumTests;
+import org.apache.hadoop.hbase.TableNotFoundException;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.master.MasterCoprocessorHost;
+import org.apache.hadoop.hbase.regionserver.RegionServerCoprocessorHost;
+import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.security.access.Permission.Action;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.TestTableName;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category(MediumTests.class)
+public class TestScanEarlyTermination extends SecureTestUtil {
+  private static final Log LOG = LogFactory.getLog(TestScanEarlyTermination.class);
+
+  static {
+    Logger.getLogger(AccessController.class).setLevel(Level.TRACE);
+    Logger.getLogger(AccessControlFilter.class).setLevel(Level.TRACE);
+    Logger.getLogger(TableAuthManager.class).setLevel(Level.TRACE);
+  }
+
+  @Rule
+  public TestTableName TEST_TABLE = new TestTableName();
+  private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
+  private static final byte[] TEST_FAMILY1 = Bytes.toBytes("f1");
+  private static final byte[] TEST_FAMILY2 = Bytes.toBytes("f2");
+  private static final byte[] TEST_ROW = Bytes.toBytes("testrow");
+  private static final byte[] TEST_Q1 = Bytes.toBytes("q1");
+  private static final byte[] TEST_Q2 = Bytes.toBytes("q2");
+  private static final byte[] ZERO = Bytes.toBytes(0L);
+
+  private static Configuration conf;
+
+  private static User USER_OWNER;
+  private static User USER_OTHER;
+
+  @BeforeClass
+  public static void setupBeforeClass() throws Exception {
+    // setup configuration
+    conf = TEST_UTIL.getConfiguration();
+    // Enable security
+    enableSecurity(conf);
+    // Verify enableSecurity sets up what we require
+    verifyConfiguration(conf);
+
+    TEST_UTIL.startMiniCluster();
+    MasterCoprocessorHost cpHost = TEST_UTIL.getMiniHBaseCluster().getMaster()
+        .getMasterCoprocessorHost();
+    cpHost.load(AccessController.class, Coprocessor.PRIORITY_HIGHEST, conf);
+    AccessController ac = (AccessController)
+      cpHost.findCoprocessor(AccessController.class.getName());
+    cpHost.createEnvironment(AccessController.class, ac, Coprocessor.PRIORITY_HIGHEST, 1, conf);
+    RegionServerCoprocessorHost rsHost = TEST_UTIL.getMiniHBaseCluster().getRegionServer(0)
+        .getRegionServerCoprocessorHost();
+    rsHost.createEnvironment(AccessController.class, ac, Coprocessor.PRIORITY_HIGHEST, 1, conf);
+
+    // Wait for the ACL table to become available
+    TEST_UTIL.waitTableEnabled(AccessControlLists.ACL_TABLE_NAME.getName());
+
+    // create a set of test users
+    USER_OWNER = User.createUserForTesting(conf, "owner", new String[0]);
+    USER_OTHER = User.createUserForTesting(conf, "other", new String[0]);
+  }
+
+  @AfterClass
+  public static void tearDownAfterClass() throws Exception {
+    TEST_UTIL.shutdownMiniCluster();
+  }
+
+  @Before
+  public void setUp() throws Exception {
+    HBaseAdmin admin = TEST_UTIL.getHBaseAdmin();
+    HTableDescriptor htd = new HTableDescriptor(TEST_TABLE.getTableName());
+    htd.setOwner(USER_OWNER);
+    HColumnDescriptor hcd = new HColumnDescriptor(TEST_FAMILY1);
+    hcd.setMaxVersions(10);
+    htd.addFamily(hcd);
+    hcd = new HColumnDescriptor(TEST_FAMILY2);
+    hcd.setMaxVersions(10);
+    htd.addFamily(hcd);
+
+    // Enable backwards compatible early termination behavior in the HTD. We
+    // want to confirm that the per-table configuration is properly picked up.
+    htd.setConfiguration(AccessControlConstants.CF_ATTRIBUTE_EARLY_OUT, "true");
+
+    admin.createTable(htd);
+
+    TEST_UTIL.waitTableEnabled(TEST_TABLE.getTableName().getName());
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    // Clean the _acl_ table
+    try {
+      TEST_UTIL.deleteTable(TEST_TABLE.getTableName());
+    } catch (TableNotFoundException ex) {
+      // Test deleted the table, no problem
+      LOG.info("Test deleted table " + TEST_TABLE.getTableName());
+    }
+    assertEquals(0, AccessControlLists.getTablePermissions(conf, TEST_TABLE.getTableName()).size());
+  }
+
+  @Test
+  public void testEarlyScanTermination() throws Exception {
+    // Grant USER_OTHER access to TEST_FAMILY1 only
+    grantOnTable(TEST_UTIL, USER_OTHER.getShortName(), TEST_TABLE.getTableName(), TEST_FAMILY1,
+      null, Action.READ);
+
+    // Set up test data
+    verifyAllowed(new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        // force a new RS connection
+        conf.set("testkey", UUID.randomUUID().toString());
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          Put put = new Put(TEST_ROW).add(TEST_FAMILY1, TEST_Q1, ZERO);
+          t.put(put);
+          // Set a READ cell ACL for USER_OTHER on this value in FAMILY2
+          put = new Put(TEST_ROW).add(TEST_FAMILY2, TEST_Q1, ZERO);
+          put.setACL(USER_OTHER.getShortName(), new Permission(Action.READ));
+          t.put(put);
+          // Set an empty cell ACL for USER_OTHER on this other value in FAMILY2
+          put = new Put(TEST_ROW).add(TEST_FAMILY2, TEST_Q2, ZERO);
+          put.setACL(USER_OTHER.getShortName(), new Permission());
+          t.put(put);
+        } finally {
+          t.close();
+        }
+        return null;
+      }
+    }, USER_OWNER);
+
+    // A scan of FAMILY1 will be allowed
+    verifyAllowed(new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        // force a new RS connection
+        conf.set("testkey", UUID.randomUUID().toString());
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          Scan scan = new Scan().addFamily(TEST_FAMILY1);
+          Result result = t.getScanner(scan).next();
+          if (result != null) {
+            assertTrue("Improper exclusion", result.containsColumn(TEST_FAMILY1, TEST_Q1));
+            assertFalse("Improper inclusion", result.containsColumn(TEST_FAMILY2, TEST_Q1));
+            return result.listCells();
+          }
+          return null;
+        } finally {
+          t.close();
+        }
+      }
+    }, USER_OTHER);
+
+    // A scan of FAMILY1 and FAMILY2 will produce results for FAMILY1 without
+    // throwing an exception, however no cells from FAMILY2 will be returned
+    // because we early out checks at the CF level.
+    verifyAllowed(new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        // force a new RS connection
+        conf.set("testkey", UUID.randomUUID().toString());
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          Scan scan = new Scan();
+          Result result = t.getScanner(scan).next();
+          if (result != null) {
+            assertTrue("Improper exclusion", result.containsColumn(TEST_FAMILY1, TEST_Q1));
+            assertFalse("Improper inclusion", result.containsColumn(TEST_FAMILY2, TEST_Q1));
+            return result.listCells();
+          }
+          return null;
+        } finally {
+          t.close();
+        }
+      }
+    }, USER_OTHER);
+
+    // A scan of FAMILY2 will throw an AccessDeniedException
+    verifyDeniedWithException(new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        // force a new RS connection
+        conf.set("testkey", UUID.randomUUID().toString());
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          Scan scan = new Scan().addFamily(TEST_FAMILY2);
+          Result result = t.getScanner(scan).next();
+          if (result != null) {
+            return result.listCells();
+          }
+          return null;
+        } finally {
+          t.close();
+        }
+      }
+    }, USER_OTHER);
+
+    // Now grant USER_OTHER access to TEST_FAMILY2:TEST_Q2
+    grantOnTable(TEST_UTIL, USER_OTHER.getShortName(), TEST_TABLE.getTableName(), TEST_FAMILY2,
+      TEST_Q2, Action.READ);
+
+    // A scan of FAMILY1 and FAMILY2 will produce combined results. In FAMILY2
+    // we have access granted to Q2 at the CF level. Because we early out
+    // checks at the CF level the cell ACL on Q1 also granting access is ignored.
+    verifyAllowed(new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        // force a new RS connection
+        conf.set("testkey", UUID.randomUUID().toString());
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          Scan scan = new Scan();
+          Result result = t.getScanner(scan).next();
+          if (result != null) {
+            assertTrue("Improper exclusion", result.containsColumn(TEST_FAMILY1, TEST_Q1));
+            assertFalse("Improper inclusion", result.containsColumn(TEST_FAMILY2, TEST_Q1));
+            assertTrue("Improper exclusion", result.containsColumn(TEST_FAMILY2, TEST_Q2));
+            return result.listCells();
+          }
+          return null;
+        } finally {
+          t.close();
+        }
+      }
+    }, USER_OTHER);
+
+    // A scan of FAMILY1 and FAMILY2 will produce combined results. If we use
+    // a cell first strategy then cell ACLs come into effect. In FAMILY2, that
+    // cell ACL on Q1 now grants access and the empty permission set on Q2 now
+    // denies access.
+    verifyAllowed(new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        // force a new RS connection
+        conf.set("testkey", UUID.randomUUID().toString());
+        HTable t = new HTable(conf, TEST_TABLE.getTableName());
+        try {
+          Scan scan = new Scan();
+          scan.setACLStrategy(true);
+          Result result = t.getScanner(scan).next();
+          if (result != null) {
+            assertTrue("Improper exclusion", result.containsColumn(TEST_FAMILY1, TEST_Q1));
+            assertTrue("Improper exclusion", result.containsColumn(TEST_FAMILY2, TEST_Q1));
+            assertFalse("Improper inclusion", result.containsColumn(TEST_FAMILY2, TEST_Q2));
+            return result.listCells();
+          }
+          return null;
+        } finally {
+          t.close();
+        }
+      }
+    }, USER_OTHER);
+
+  }
+}