sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryAuthorizationProvider.java - incubator-sentry - Git at Google

 /**
  * 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 permission and
  * limitations under the License.
  */
 package org.apache.sentry.hdfs;

 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;

 import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.UnresolvedLinkException;
 import org.apache.hadoop.fs.permission.AclEntry;
 import org.apache.hadoop.fs.permission.AclEntryScope;
 import org.apache.hadoop.fs.permission.AclEntryType;
 import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.server.namenode.AclFeature;
 import org.apache.hadoop.hdfs.server.namenode.AuthorizationProvider;
 import org.apache.hadoop.security.AccessControlException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;

 public class SentryAuthorizationProvider
     extends AuthorizationProvider implements Configurable {

   static class SentryAclFeature extends AclFeature {
     public SentryAclFeature(ImmutableList<AclEntry> entries) {
       super(entries);
     }
   }

   private static Logger LOG =
       LoggerFactory.getLogger(SentryAuthorizationProvider.class);

   private boolean started;
   private Configuration conf;
   private AuthorizationProvider defaultAuthzProvider;
   private String user;
   private String group;
   private FsPermission permission;
   private boolean originalAuthzAsAcl;
   private SentryAuthorizationInfo authzInfo;

   public SentryAuthorizationProvider() {
     this(null);
   }

   @VisibleForTesting
   SentryAuthorizationProvider(SentryAuthorizationInfo authzInfo) {
     this.authzInfo = authzInfo;
   }

   @Override
   public void setConf(Configuration conf) {
     this.conf = conf;
   }

   @Override
   public Configuration getConf() {
     return conf;
   }

   @Override
   public synchronized void start() {
     if (started) {
       throw new IllegalStateException("Provider already started");
     }
     started = true;
     try {
       if (!conf.getBoolean(DFSConfigKeys.DFS_NAMENODE_ACLS_ENABLED_KEY, false)) {
         throw new RuntimeException("HDFS ACLs must be enabled");
       }

       defaultAuthzProvider = AuthorizationProvider.get();
       defaultAuthzProvider.start();
       // Configuration is read from hdfs-sentry.xml and NN configuration, in
       // that order of precedence.
       Configuration conf = new Configuration(this.conf);
       conf.addResource(SentryAuthorizationConstants.CONFIG_FILE);
       user = conf.get(SentryAuthorizationConstants.HDFS_USER_KEY,
           SentryAuthorizationConstants.HDFS_USER_DEFAULT);
       group = conf.get(SentryAuthorizationConstants.HDFS_GROUP_KEY,
           SentryAuthorizationConstants.HDFS_GROUP_DEFAULT);
       permission = FsPermission.createImmutable(
           (short) conf.getLong(SentryAuthorizationConstants.HDFS_PERMISSION_KEY,
               SentryAuthorizationConstants.HDFS_PERMISSION_DEFAULT)
       );
       originalAuthzAsAcl = conf.getBoolean(
           SentryAuthorizationConstants.INCLUDE_HDFS_AUTHZ_AS_ACL_KEY,
           SentryAuthorizationConstants.INCLUDE_HDFS_AUTHZ_AS_ACL_DEFAULT);

       LOG.info("Starting");
       LOG.info("Config: hdfs-user[{}] hdfs-group[{}] hdfs-permission[{}] " +
           "include-hdfs-authz-as-acl[{}]", new Object[]
           {user, group, permission, originalAuthzAsAcl});

       if (authzInfo == null) {
         authzInfo = new SentryAuthorizationInfo(conf);
       }
       authzInfo.start();
     } catch (Exception ex) {
       throw new RuntimeException(ex);
     }
   }

   @Override
   public synchronized void stop() {
     LOG.debug("Stopping");
     authzInfo.stop();
     defaultAuthzProvider.stop();
     defaultAuthzProvider = null;
   }

   @Override
   public void setSnaphottableDirs(Map<INodeAuthorizationInfo, Integer>
       snapshotableDirs) {
     defaultAuthzProvider.setSnaphottableDirs(snapshotableDirs);
   }

   @Override
   public void addSnapshottable(INodeAuthorizationInfo dir) {
     defaultAuthzProvider.addSnapshottable(dir);
   }

   @Override
   public void removeSnapshottable(INodeAuthorizationInfo dir) {
     defaultAuthzProvider.removeSnapshottable(dir);
   }

   @Override
   public void createSnapshot(INodeAuthorizationInfo dir, int snapshotId)
       throws IOException{
     defaultAuthzProvider.createSnapshot(dir, snapshotId);
   }

   @Override
   public void removeSnapshot(INodeAuthorizationInfo dir, int snapshotId)
       throws IOException {
     defaultAuthzProvider.removeSnapshot(dir, snapshotId);
   }

   @Override
   public void checkPermission(String user, Set<String> groups,
       INodeAuthorizationInfo[] inodes, int snapshotId,
       boolean doCheckOwner, FsAction ancestorAccess, FsAction parentAccess,
       FsAction access, FsAction subAccess, boolean ignoreEmptyDir)
       throws AccessControlException, UnresolvedLinkException {
     defaultAuthzProvider.checkPermission(user, groups, inodes, snapshotId,
         doCheckOwner, ancestorAccess, parentAccess, access, subAccess,
         ignoreEmptyDir);
   }

   private static final String[] EMPTY_STRING_ARRAY = new String[0];

   private String[] getPathElements(INodeAuthorizationInfo node) {
     return getPathElements(node, 0);
   }

   private String[] getPathElements(INodeAuthorizationInfo node, int idx) {
     String[] paths;
     INodeAuthorizationInfo parent = node.getParent();
     if (parent == null) {
       paths = (idx > 0) ? new String[idx] : EMPTY_STRING_ARRAY;
     } else {
       paths = getPathElements(parent, idx + 1);
       paths[paths.length - 1 - idx] = node.getLocalName();
     }
     return paths;
   }

   @Override
   public void setUser(INodeAuthorizationInfo node, String user) {
     defaultAuthzProvider.setUser(node, user);
   }

   @Override
   public String getUser(INodeAuthorizationInfo node, int snapshotId) {
     String user;
     String[] pathElements = getPathElements(node);
     if (!authzInfo.isManaged(pathElements)) {
       user = defaultAuthzProvider.getUser(node, snapshotId);
     } else if (!authzInfo.doesBelongToAuthzObject(pathElements)) {
       user = defaultAuthzProvider.getUser(node, snapshotId);
     } else {
         user = this.user;
     }
     return user;
   }

   @Override
   public void setGroup(INodeAuthorizationInfo node, String group) {
     defaultAuthzProvider.setGroup(node, group);
   }

   @Override
   public String getGroup(INodeAuthorizationInfo node, int snapshotId) {
     String group;
     String[] pathElements = getPathElements(node);
     if (!authzInfo.isManaged(pathElements)) {
       group = getDefaultProviderGroup(node, snapshotId);
     } else if (!authzInfo.doesBelongToAuthzObject(pathElements)) {
       group = getDefaultProviderGroup(node, snapshotId);
     } else {
       group = this.group;
     }
     return group;
   }

   @Override
   public void setPermission(INodeAuthorizationInfo node,
       FsPermission permission) {
     defaultAuthzProvider.setPermission(node, permission);
   }

   @Override
   public FsPermission getFsPermission(
       INodeAuthorizationInfo node, int snapshotId) {
     FsPermission permission;
     String[] pathElements = getPathElements(node);
     if (!authzInfo.isManaged(pathElements)) {
       permission = defaultAuthzProvider.getFsPermission(node, snapshotId);
     } else if (!authzInfo.doesBelongToAuthzObject(pathElements)) {
       permission = defaultAuthzProvider.getFsPermission(node, snapshotId);
     }
      else {
       FsPermission returnPerm = this.permission;
       // Handle case when prefix directory is itself associated with an
       // authorizable object (default db directory in hive)
       // An executable permission needs to be set on the the prefix directory
       // in this case.. else, subdirectories (which map to other dbs) will
       // not be travesible.
       for (String [] prefixPath : authzInfo.getPathPrefixes()) {
         if (Arrays.equals(prefixPath, pathElements)) {
           returnPerm = FsPermission.createImmutable((short)(returnPerm.toShort() | 0x01));
           break;
         }
       }
       permission = returnPerm;
     }
     return permission;
   }

   private List<AclEntry> createAclEntries(String user, String group,
       FsPermission permission) {
     List<AclEntry> list = new ArrayList<AclEntry>();
     AclEntry.Builder builder = new AclEntry.Builder();
     FsPermission fsPerm = new FsPermission(permission);
     builder.setName(user);
     builder.setType(AclEntryType.USER);
     builder.setScope(AclEntryScope.ACCESS);
     builder.setPermission(fsPerm.getUserAction());
     list.add(builder.build());
     builder.setName(group);
     builder.setType(AclEntryType.GROUP);
     builder.setScope(AclEntryScope.ACCESS);
     builder.setPermission(fsPerm.getGroupAction());
     list.add(builder.build());
     builder.setName(null);
     return list;
   }
   /*
   Returns hadoop acls if
   - Not managed
   - Not stale and not an auth obj
   Returns hive:hive
   - If stale
   Returns sentry acls
   - Otherwise, if not stale and auth obj
    */
   @Override
   public AclFeature getAclFeature(INodeAuthorizationInfo node, int snapshotId) {
     AclFeature f = null;
     String[] pathElements = getPathElements(node);
     String p = Arrays.toString(pathElements);
     boolean isManaged = false;
     boolean isStale = false;
     boolean hasAuthzObj = false;
     Map<String, AclEntry> aclMap = null;
     if (!authzInfo.isManaged(pathElements)) {
       isManaged = false;
       f = defaultAuthzProvider.getAclFeature(node, snapshotId);
     } else if (!authzInfo.doesBelongToAuthzObject(pathElements)) {
       isManaged = true;
       f = defaultAuthzProvider.getAclFeature(node, snapshotId);
     } else {
       isManaged = true;
       hasAuthzObj = true;
       aclMap = new HashMap<String, AclEntry>();
       if (originalAuthzAsAcl) {
         String user = defaultAuthzProvider.getUser(node, snapshotId);
         String group = getDefaultProviderGroup(node, snapshotId);
         FsPermission perm = defaultAuthzProvider.getFsPermission(node, snapshotId);
         addToACLMap(aclMap, createAclEntries(user, group, perm));
       } else {
         addToACLMap(aclMap,
             createAclEntries(this.user, this.group, this.permission));
       }
       if (!authzInfo.isStale()) {
         isStale = false;
         addToACLMap(aclMap, authzInfo.getAclEntries(pathElements));
         f = new SentryAclFeature(ImmutableList.copyOf(aclMap.values()));
       } else {
         isStale = true;
         f = new SentryAclFeature(ImmutableList.copyOf(aclMap.values()));
       }
     }
     if (LOG.isDebugEnabled()) {
       LOG.debug("### getAclEntry \n[" + (p == null ? "null" : p) + "] : ["
           + "isManaged=" + isManaged
           + ", isStale=" + isStale
           + ", hasAuthzObj=" + hasAuthzObj
           + ", origAuthzAsAcl=" + originalAuthzAsAcl + "]\n"
           + "[" + (aclMap == null ? "null" : aclMap) + "]\n"
           + "[" + (f == null ? "null" : f.getEntries()) + "]\n");
     }
     return f;
   }

   private void addToACLMap(Map<String, AclEntry> map,
       Collection<AclEntry> entries) {
     for (AclEntry ent : entries) {
       String key = (ent.getName() == null ? "" : ent.getName())
           + ent.getScope() + ent.getType();
       AclEntry aclEntry = map.get(key);
       if (aclEntry == null) {
         map.put(key, ent);
       } else {
         map.put(key,
             new AclEntry.Builder().
             setName(ent.getName()).
             setScope(ent.getScope()).
             setType(ent.getType()).
             setPermission(ent.getPermission().or(aclEntry.getPermission())).
             build());
       }
     }
   }

   private String getDefaultProviderGroup(INodeAuthorizationInfo node,
       int snapshotId) {
     String group = defaultAuthzProvider.getGroup(node, snapshotId);
     INodeAuthorizationInfo pNode = node.getParent();
     while  (group == null && pNode != null) {
       group = defaultAuthzProvider.getGroup(pNode, snapshotId);
       pNode = pNode.getParent();
     }
     return group;
   }

   @Override
   public void removeAclFeature(INodeAuthorizationInfo node) {
     AclFeature aclFeature = node.getAclFeature(CURRENT_STATE_ID);
     if (aclFeature.getClass() != SentryAclFeature.class) {
       defaultAuthzProvider.removeAclFeature(node);
     }
   }

   @Override
   public void addAclFeature(INodeAuthorizationInfo node, AclFeature f) {
     String[] pathElements = getPathElements(node);
     if (!authzInfo.isManaged(pathElements)) {
       defaultAuthzProvider.addAclFeature(node, f);
     }
   }

 }
	/**
	* 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 permission and
	* limitations under the License.
	*/
	package org.apache.sentry.hdfs;

	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.Collection;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;

	import org.apache.hadoop.conf.Configurable;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fs.UnresolvedLinkException;
	import org.apache.hadoop.fs.permission.AclEntry;
	import org.apache.hadoop.fs.permission.AclEntryScope;
	import org.apache.hadoop.fs.permission.AclEntryType;
	import org.apache.hadoop.fs.permission.FsAction;
	import org.apache.hadoop.fs.permission.FsPermission;
	import org.apache.hadoop.hdfs.DFSConfigKeys;
	import org.apache.hadoop.hdfs.server.namenode.AclFeature;
	import org.apache.hadoop.hdfs.server.namenode.AuthorizationProvider;
	import org.apache.hadoop.security.AccessControlException;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.google.common.annotations.VisibleForTesting;
	import com.google.common.collect.ImmutableList;

	public class SentryAuthorizationProvider
	extends AuthorizationProvider implements Configurable {

	static class SentryAclFeature extends AclFeature {
	public SentryAclFeature(ImmutableList<AclEntry> entries) {
	super(entries);
	}
	}

	private static Logger LOG =
	LoggerFactory.getLogger(SentryAuthorizationProvider.class);

	private boolean started;
	private Configuration conf;
	private AuthorizationProvider defaultAuthzProvider;
	private String user;
	private String group;
	private FsPermission permission;
	private boolean originalAuthzAsAcl;
	private SentryAuthorizationInfo authzInfo;

	public SentryAuthorizationProvider() {
	this(null);
	}

	@VisibleForTesting
	SentryAuthorizationProvider(SentryAuthorizationInfo authzInfo) {
	this.authzInfo = authzInfo;
	}

	@Override
	public void setConf(Configuration conf) {
	this.conf = conf;
	}

	@Override
	public Configuration getConf() {
	return conf;
	}

	@Override
	public synchronized void start() {
	if (started) {
	throw new IllegalStateException("Provider already started");
	}
	started = true;
	try {
	if (!conf.getBoolean(DFSConfigKeys.DFS_NAMENODE_ACLS_ENABLED_KEY, false)) {
	throw new RuntimeException("HDFS ACLs must be enabled");
	}

	defaultAuthzProvider = AuthorizationProvider.get();
	defaultAuthzProvider.start();
	// Configuration is read from hdfs-sentry.xml and NN configuration, in
	// that order of precedence.
	Configuration conf = new Configuration(this.conf);
	conf.addResource(SentryAuthorizationConstants.CONFIG_FILE);
	user = conf.get(SentryAuthorizationConstants.HDFS_USER_KEY,
	SentryAuthorizationConstants.HDFS_USER_DEFAULT);
	group = conf.get(SentryAuthorizationConstants.HDFS_GROUP_KEY,
	SentryAuthorizationConstants.HDFS_GROUP_DEFAULT);
	permission = FsPermission.createImmutable(
	(short) conf.getLong(SentryAuthorizationConstants.HDFS_PERMISSION_KEY,
	SentryAuthorizationConstants.HDFS_PERMISSION_DEFAULT)
	);
	originalAuthzAsAcl = conf.getBoolean(
	SentryAuthorizationConstants.INCLUDE_HDFS_AUTHZ_AS_ACL_KEY,
	SentryAuthorizationConstants.INCLUDE_HDFS_AUTHZ_AS_ACL_DEFAULT);

	LOG.info("Starting");
	LOG.info("Config: hdfs-user[{}] hdfs-group[{}] hdfs-permission[{}] " +
	"include-hdfs-authz-as-acl[{}]", new Object[]
	{user, group, permission, originalAuthzAsAcl});

	if (authzInfo == null) {
	authzInfo = new SentryAuthorizationInfo(conf);
	}
	authzInfo.start();
	} catch (Exception ex) {
	throw new RuntimeException(ex);
	}
	}

	@Override
	public synchronized void stop() {
	LOG.debug("Stopping");
	authzInfo.stop();
	defaultAuthzProvider.stop();
	defaultAuthzProvider = null;
	}

	@Override
	public void setSnaphottableDirs(Map<INodeAuthorizationInfo, Integer>
	snapshotableDirs) {
	defaultAuthzProvider.setSnaphottableDirs(snapshotableDirs);
	}

	@Override
	public void addSnapshottable(INodeAuthorizationInfo dir) {
	defaultAuthzProvider.addSnapshottable(dir);
	}

	@Override
	public void removeSnapshottable(INodeAuthorizationInfo dir) {
	defaultAuthzProvider.removeSnapshottable(dir);
	}

	@Override
	public void createSnapshot(INodeAuthorizationInfo dir, int snapshotId)
	throws IOException{
	defaultAuthzProvider.createSnapshot(dir, snapshotId);
	}

	@Override
	public void removeSnapshot(INodeAuthorizationInfo dir, int snapshotId)
	throws IOException {
	defaultAuthzProvider.removeSnapshot(dir, snapshotId);
	}

	@Override
	public void checkPermission(String user, Set<String> groups,
	INodeAuthorizationInfo[] inodes, int snapshotId,
	boolean doCheckOwner, FsAction ancestorAccess, FsAction parentAccess,
	FsAction access, FsAction subAccess, boolean ignoreEmptyDir)
	throws AccessControlException, UnresolvedLinkException {
	defaultAuthzProvider.checkPermission(user, groups, inodes, snapshotId,
	doCheckOwner, ancestorAccess, parentAccess, access, subAccess,
	ignoreEmptyDir);
	}

	private static final String[] EMPTY_STRING_ARRAY = new String[0];

	private String[] getPathElements(INodeAuthorizationInfo node) {
	return getPathElements(node, 0);
	}

	private String[] getPathElements(INodeAuthorizationInfo node, int idx) {
	String[] paths;
	INodeAuthorizationInfo parent = node.getParent();
	if (parent == null) {
	paths = (idx > 0) ? new String[idx] : EMPTY_STRING_ARRAY;
	} else {
	paths = getPathElements(parent, idx + 1);
	paths[paths.length - 1 - idx] = node.getLocalName();
	}
	return paths;
	}

	@Override
	public void setUser(INodeAuthorizationInfo node, String user) {
	defaultAuthzProvider.setUser(node, user);
	}

	@Override
	public String getUser(INodeAuthorizationInfo node, int snapshotId) {
	String user;
	String[] pathElements = getPathElements(node);
	if (!authzInfo.isManaged(pathElements)) {
	user = defaultAuthzProvider.getUser(node, snapshotId);
	} else if (!authzInfo.doesBelongToAuthzObject(pathElements)) {
	user = defaultAuthzProvider.getUser(node, snapshotId);
	} else {
	user = this.user;
	}
	return user;
	}

	@Override
	public void setGroup(INodeAuthorizationInfo node, String group) {
	defaultAuthzProvider.setGroup(node, group);
	}

	@Override
	public String getGroup(INodeAuthorizationInfo node, int snapshotId) {
	String group;
	String[] pathElements = getPathElements(node);
	if (!authzInfo.isManaged(pathElements)) {
	group = getDefaultProviderGroup(node, snapshotId);
	} else if (!authzInfo.doesBelongToAuthzObject(pathElements)) {
	group = getDefaultProviderGroup(node, snapshotId);
	} else {
	group = this.group;
	}
	return group;
	}

	@Override
	public void setPermission(INodeAuthorizationInfo node,
	FsPermission permission) {
	defaultAuthzProvider.setPermission(node, permission);
	}

	@Override
	public FsPermission getFsPermission(
	INodeAuthorizationInfo node, int snapshotId) {
	FsPermission permission;
	String[] pathElements = getPathElements(node);
	if (!authzInfo.isManaged(pathElements)) {
	permission = defaultAuthzProvider.getFsPermission(node, snapshotId);
	} else if (!authzInfo.doesBelongToAuthzObject(pathElements)) {
	permission = defaultAuthzProvider.getFsPermission(node, snapshotId);
	}
	else {
	FsPermission returnPerm = this.permission;
	// Handle case when prefix directory is itself associated with an
	// authorizable object (default db directory in hive)
	// An executable permission needs to be set on the the prefix directory
	// in this case.. else, subdirectories (which map to other dbs) will
	// not be travesible.
	for (String [] prefixPath : authzInfo.getPathPrefixes()) {
	if (Arrays.equals(prefixPath, pathElements)) {
	returnPerm = FsPermission.createImmutable((short)(returnPerm.toShort() \| 0x01));
	break;
	}
	}
	permission = returnPerm;
	}
	return permission;
	}

	private List<AclEntry> createAclEntries(String user, String group,
	FsPermission permission) {
	List<AclEntry> list = new ArrayList<AclEntry>();
	AclEntry.Builder builder = new AclEntry.Builder();
	FsPermission fsPerm = new FsPermission(permission);
	builder.setName(user);
	builder.setType(AclEntryType.USER);
	builder.setScope(AclEntryScope.ACCESS);
	builder.setPermission(fsPerm.getUserAction());
	list.add(builder.build());
	builder.setName(group);
	builder.setType(AclEntryType.GROUP);
	builder.setScope(AclEntryScope.ACCESS);
	builder.setPermission(fsPerm.getGroupAction());
	list.add(builder.build());
	builder.setName(null);
	return list;
	}
	/*
	Returns hadoop acls if
	- Not managed
	- Not stale and not an auth obj
	Returns hive:hive
	- If stale
	Returns sentry acls
	- Otherwise, if not stale and auth obj
	*/
	@Override
	public AclFeature getAclFeature(INodeAuthorizationInfo node, int snapshotId) {
	AclFeature f = null;
	String[] pathElements = getPathElements(node);
	String p = Arrays.toString(pathElements);
	boolean isManaged = false;
	boolean isStale = false;
	boolean hasAuthzObj = false;
	Map<String, AclEntry> aclMap = null;
	if (!authzInfo.isManaged(pathElements)) {
	isManaged = false;
	f = defaultAuthzProvider.getAclFeature(node, snapshotId);
	} else if (!authzInfo.doesBelongToAuthzObject(pathElements)) {
	isManaged = true;
	f = defaultAuthzProvider.getAclFeature(node, snapshotId);
	} else {
	isManaged = true;
	hasAuthzObj = true;
	aclMap = new HashMap<String, AclEntry>();
	if (originalAuthzAsAcl) {
	String user = defaultAuthzProvider.getUser(node, snapshotId);
	String group = getDefaultProviderGroup(node, snapshotId);
	FsPermission perm = defaultAuthzProvider.getFsPermission(node, snapshotId);
	addToACLMap(aclMap, createAclEntries(user, group, perm));
	} else {
	addToACLMap(aclMap,
	createAclEntries(this.user, this.group, this.permission));
	}
	if (!authzInfo.isStale()) {
	isStale = false;
	addToACLMap(aclMap, authzInfo.getAclEntries(pathElements));
	f = new SentryAclFeature(ImmutableList.copyOf(aclMap.values()));
	} else {
	isStale = true;
	f = new SentryAclFeature(ImmutableList.copyOf(aclMap.values()));
	}
	}
	if (LOG.isDebugEnabled()) {
	LOG.debug("### getAclEntry \n[" + (p == null ? "null" : p) + "] : ["
	+ "isManaged=" + isManaged
	+ ", isStale=" + isStale
	+ ", hasAuthzObj=" + hasAuthzObj
	+ ", origAuthzAsAcl=" + originalAuthzAsAcl + "]\n"
	+ "[" + (aclMap == null ? "null" : aclMap) + "]\n"
	+ "[" + (f == null ? "null" : f.getEntries()) + "]\n");
	}
	return f;
	}

	private void addToACLMap(Map<String, AclEntry> map,
	Collection<AclEntry> entries) {
	for (AclEntry ent : entries) {
	String key = (ent.getName() == null ? "" : ent.getName())
	+ ent.getScope() + ent.getType();
	AclEntry aclEntry = map.get(key);
	if (aclEntry == null) {
	map.put(key, ent);
	} else {
	map.put(key,
	new AclEntry.Builder().
	setName(ent.getName()).
	setScope(ent.getScope()).
	setType(ent.getType()).
	setPermission(ent.getPermission().or(aclEntry.getPermission())).
	build());
	}
	}
	}

	private String getDefaultProviderGroup(INodeAuthorizationInfo node,
	int snapshotId) {
	String group = defaultAuthzProvider.getGroup(node, snapshotId);
	INodeAuthorizationInfo pNode = node.getParent();
	while (group == null && pNode != null) {
	group = defaultAuthzProvider.getGroup(pNode, snapshotId);
	pNode = pNode.getParent();
	}
	return group;
	}

	@Override
	public void removeAclFeature(INodeAuthorizationInfo node) {
	AclFeature aclFeature = node.getAclFeature(CURRENT_STATE_ID);
	if (aclFeature.getClass() != SentryAclFeature.class) {
	defaultAuthzProvider.removeAclFeature(node);
	}
	}

	@Override
	public void addAclFeature(INodeAuthorizationInfo node, AclFeature f) {
	String[] pathElements = getPathElements(node);
	if (!authzInfo.isManaged(pathElements)) {
	defaultAuthzProvider.addAclFeature(node, f);
	}
	}

	}