content: /*
   * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
   * (the "License"). You may not use this work except in compliance with the License, which is
   * available at www.apache.org/licenses/LICENSE-2.0
   *
   * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
   * either express or implied, as more fully set forth in the License.
   *
   * See the NOTICE file distributed with this work for information regarding copyright ownership.
   */
  
  package alluxio.underfs.hdfs.acl;
  
  import alluxio.collections.Pair;
  import alluxio.security.authorization.AccessControlList;
  import alluxio.security.authorization.AclAction;
  import alluxio.security.authorization.DefaultAccessControlList;
  import alluxio.underfs.hdfs.HdfsAclProvider;
  
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  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.AclStatus;
  import org.apache.hadoop.fs.permission.FsAction;
  import org.apache.hadoop.hdfs.protocol.AclException;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.List;
  import javax.annotation.concurrent.ThreadSafe;
  
  /**
   * Provides the HDFS ACLs. This is only supported in Hadoop versions 2.4 and greater.
   */
  @ThreadSafe
  public class SupportedHdfsAclProvider implements HdfsAclProvider {
    private static final Logger LOG = LoggerFactory.getLogger(SupportedHdfsAclProvider.class);
  
    @Override
    public Pair<AccessControlList, DefaultAccessControlList> getAcl(FileSystem hdfs, String path)
        throws IOException {
      AclStatus hdfsAcl;
      Path filePath = new Path(path);
      boolean isDir = hdfs.isDirectory(filePath);
      try {
        hdfsAcl = hdfs.getAclStatus(filePath);
      } catch (AclException e) {
        // When dfs.namenode.acls.enabled is false, getAclStatus throws AclException.
        return new Pair<>(null, null);
      }
      AccessControlList acl = new AccessControlList();
      DefaultAccessControlList defaultAcl = new DefaultAccessControlList();
  
      acl.setOwningUser(hdfsAcl.getOwner().intern());
      acl.setOwningGroup(hdfsAcl.getGroup().intern());
      defaultAcl.setOwningUser(hdfsAcl.getOwner());
      defaultAcl.setOwningGroup(hdfsAcl.getGroup());
      for (AclEntry entry : hdfsAcl.getEntries()) {
        alluxio.security.authorization.AclEntry.Builder builder =
            new alluxio.security.authorization.AclEntry.Builder();
        builder.setType(getAclEntryType(entry));
        builder.setSubject(entry.getName() == null ? "" : entry.getName());
        FsAction permission = entry.getPermission();
        if (permission.implies(FsAction.READ)) {
          builder.addAction(AclAction.READ);
        } else if (permission.implies(FsAction.WRITE)) {
          builder.addAction(AclAction.WRITE);
        } else if (permission.implies(FsAction.EXECUTE)) {
          builder.addAction(AclAction.EXECUTE);
        }
        if (entry.getScope().equals(AclEntryScope.ACCESS)) {
          acl.setEntry(builder.build());
        } else {
          // default ACL, must be a directory
          defaultAcl.setEntry(builder.build());
        }
      }
      if (isDir) {
        return new Pair<>(acl, defaultAcl);
      } else {
        // a null defaultACL indicates this is a file
        return new Pair<>(acl, null);
      }
    }
  
    @Override
    public void setAclEntries(FileSystem hdfs, String path,
        List<alluxio.security.authorization.AclEntry> aclEntries) throws IOException {
      // convert AccessControlList into hdfsAcl
      List<AclEntry> aclSpecs = new ArrayList<>();
  
      for (alluxio.security.authorization.AclEntry entry : aclEntries) {
        AclEntry hdfsAclEntry = getHdfsAclEntry(entry);
        aclSpecs.add(hdfsAclEntry);
      }
      // set hdfsAcl;
      try {
        hdfs.setAcl(new Path(path), aclSpecs);
      } catch (UnsupportedOperationException e) {
        // noop if hdfs does not support acl
      }
    }
  
    private AclEntry getHdfsAclEntry(alluxio.security.authorization.AclEntry entry)
        throws IOException {
      AclEntry.Builder builder = new AclEntry.Builder();
      // Do not set name for unnamed entries
      if (entry.getType() != alluxio.security.authorization.AclEntryType.OWNING_USER
          && entry.getType() != alluxio.security.authorization.AclEntryType.OWNING_GROUP) {
        builder.setName(entry.getSubject());
      }
  
      builder.setScope(entry.isDefault() ? AclEntryScope.DEFAULT : AclEntryScope.ACCESS);
      builder.setType(getHdfsAclEntryType(entry));
      FsAction permission = FsAction.getFsAction(entry.getActions().toCliString());
      builder.setPermission(permission);
      return builder.build();
    }
  
    /**
     * @param aclEntry an alluxio acl entry
     * @return hdfs acl entry type
     */
    private AclEntryType getHdfsAclEntryType(alluxio.security.authorization.AclEntry aclEntry)
        throws IOException {
      switch (aclEntry.getType()) {
        case OWNING_USER:
        case NAMED_USER:
          return AclEntryType.USER;
        case OWNING_GROUP:
        case NAMED_GROUP:
          return AclEntryType.GROUP;
        case MASK:
          return AclEntryType.MASK;
        case OTHER:
          return AclEntryType.OTHER;
        default:
          throw new IOException("Unknown Alluxio ACL entry type: " + aclEntry.getType());
      }
    }
  
    /**
     * @param entry an hdfs acl entry
     * @return alluxio acl entry type
     */
    private alluxio.security.authorization.AclEntryType getAclEntryType(AclEntry entry)
        throws IOException {
      switch (entry.getType()) {
        case USER:
          return entry.getName() == null || entry.getName().isEmpty()
              ? alluxio.security.authorization.AclEntryType.OWNING_USER
              : alluxio.security.authorization.AclEntryType.NAMED_USER;
        case GROUP:
          return entry.getName() == null || entry.getName().isEmpty()
              ? alluxio.security.authorization.AclEntryType.OWNING_GROUP
              : alluxio.security.authorization.AclEntryType.NAMED_GROUP;
        case MASK:
          return alluxio.security.authorization.AclEntryType.MASK;
        case OTHER:
          return alluxio.security.authorization.AclEntryType.OTHER;
        default:
          throw new IOException("Unknown HDFS ACL entry type: " + entry.getType());
      }
    }
  }
  