/**
 * Get the file info for a specific file.
 *
 * @param srcArg The string representation of the path to the file
 * @param resolveLink whether to throw UnresolvedLinkException
 *        if src refers to a symlink
 *
 * @return object containing information regarding the file
 *         or null if file not found
 */
static HdfsFileStatus getFileInfo(
    FSDirectory fsd, String srcArg, boolean resolveLink)
    throws IOException {
  String src = srcArg;
  if (!DFSUtil.isValidName(src)) {
    throw new InvalidPathException("Invalid file name: " + src);
  }
  FSPermissionChecker pc = fsd.getPermissionChecker();
  byte[][] pathComponents = FSDirectory.getPathComponentsForReservedPath(src);
  src = fsd.resolvePath(pc, src, pathComponents);
  final INodesInPath iip = fsd.getINodesInPath(src, resolveLink);
  boolean isSuperUser = true;
  if (fsd.isPermissionEnabled()) {
    fsd.checkPermission(pc, iip, false, null, null, null, null, false);
    isSuperUser = pc.isSuperUser();
  }
  return getFileInfo(fsd, src, resolveLink,
      FSDirectory.isReservedRawName(srcArg), isSuperUser);
}
 

/**
 * Regression test for HDFS-3626. Creates a file using a non-canonical path
 * (i.e. with extra slashes between components) and makes sure that the NN
 * rejects it.
 */
@Test
public void testMkdirRpcNonCanonicalPath() throws IOException {
  MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(0).build();
  try {
    NamenodeProtocols nnrpc = cluster.getNameNodeRpc();
    
    for (String pathStr : NON_CANONICAL_PATHS) {
      try {
        nnrpc.mkdirs(pathStr, new FsPermission((short)0755), true);
        fail("Did not fail when called with a non-canonicalized path: "
           + pathStr);
      } catch (InvalidPathException ipe) {
        // expected
      }
    }
  } finally {
    cluster.shutdown();
  }
}
 

/**
 * Get the file info for a specific file.
 *
 * @param srcArg The string representation of the path to the file
 * @param resolveLink whether to throw UnresolvedLinkException
 *        if src refers to a symlink
 *
 * @return object containing information regarding the file
 *         or null if file not found
 */
static HdfsFileStatus getFileInfo(
    FSDirectory fsd, String srcArg, boolean resolveLink)
    throws IOException {
  String src = srcArg;
  if (!DFSUtil.isValidName(src)) {
    throw new InvalidPathException("Invalid file name: " + src);
  }
  FSPermissionChecker pc = fsd.getPermissionChecker();
  byte[][] pathComponents = FSDirectory.getPathComponentsForReservedPath(src);
  src = fsd.resolvePath(pc, src, pathComponents);
  final INodesInPath iip = fsd.getINodesInPath(src, resolveLink);
  boolean isSuperUser = true;
  if (fsd.isPermissionEnabled()) {
    fsd.checkPermission(pc, iip, false, null, null, null, null, false);
    isSuperUser = pc.isSuperUser();
  }
  return getFileInfo(fsd, src, resolveLink,
      FSDirectory.isReservedRawName(srcArg), isSuperUser);
}
 

/**
 * Regression test for HDFS-3626. Creates a file using a non-canonical path
 * (i.e. with extra slashes between components) and makes sure that the NN
 * rejects it.
 */
@Test
public void testMkdirRpcNonCanonicalPath() throws IOException {
  MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(0).build();
  try {
    NamenodeProtocols nnrpc = cluster.getNameNodeRpc();
    
    for (String pathStr : NON_CANONICAL_PATHS) {
      try {
        nnrpc.mkdirs(pathStr, new FsPermission((short)0755), true);
        fail("Did not fail when called with a non-canonicalized path: "
           + pathStr);
      } catch (InvalidPathException ipe) {
        // expected
      }
    }
  } finally {
    cluster.shutdown();
  }
}
 

/**
 * Same {{@link #mkdirs(String, FsPermission, boolean)} except
 * that the permissions has already been masked against umask.
 */
public boolean primitiveMkdir(String src, FsPermission absPermission, 
  boolean createParent)
  throws IOException {
  checkOpen();
  if (absPermission == null) {
    absPermission = 
      FsPermission.getDefault().applyUMask(dfsClientConf.uMask);
  } 

  if(LOG.isDebugEnabled()) {
    LOG.debug(src + ": masked=" + absPermission);
  }
  TraceScope scope = Trace.startSpan("mkdir", traceSampler);
  try {
    return namenode.mkdirs(src, absPermission, createParent);
  } catch(RemoteException re) {
    throw re.unwrapRemoteException(AccessControlException.class,
                                   InvalidPathException.class,
                                   FileAlreadyExistsException.class,
                                   FileNotFoundException.class,
                                   ParentNotDirectoryException.class,
                                   SafeModeException.class,
                                   NSQuotaExceededException.class,
                                   DSQuotaExceededException.class,
                                   UnresolvedPathException.class,
                                   SnapshotAccessControlException.class);
  } finally {
    scope.close();
  }
}
 

/**
 * The new rename which has the POSIX semantic.
 */
static Map.Entry<BlocksMapUpdateInfo, HdfsFileStatus> renameToInt(
    FSDirectory fsd, final String srcArg, final String dstArg,
    boolean logRetryCache, Options.Rename... options)
    throws IOException {
  String src = srcArg;
  String dst = dstArg;
  if (NameNode.stateChangeLog.isDebugEnabled()) {
    NameNode.stateChangeLog.debug("DIR* NameSystem.renameTo: with options -" +
        " " + src + " to " + dst);
  }
  if (!DFSUtil.isValidName(dst)) {
    throw new InvalidPathException("Invalid name: " + dst);
  }
  final FSPermissionChecker pc = fsd.getPermissionChecker();

  byte[][] srcComponents = FSDirectory.getPathComponentsForReservedPath(src);
  byte[][] dstComponents = FSDirectory.getPathComponentsForReservedPath(dst);
  BlocksMapUpdateInfo collectedBlocks = new BlocksMapUpdateInfo();
  src = fsd.resolvePath(pc, src, srcComponents);
  dst = fsd.resolvePath(pc, dst, dstComponents);
  renameTo(fsd, pc, src, dst, collectedBlocks, logRetryCache, options);
  INodesInPath dstIIP = fsd.getINodesInPath(dst, false);
  HdfsFileStatus resultingStat = fsd.getAuditFileInfo(dstIIP);

  return new AbstractMap.SimpleImmutableEntry<>(
      collectedBlocks, resultingStat);
}
 

/**
 * Create a snapshot
 * @param snapshotRoot The directory path where the snapshot is taken
 * @param snapshotName The name of the snapshot
 */
static String createSnapshot(
    FSDirectory fsd, SnapshotManager snapshotManager, String snapshotRoot,
    String snapshotName, boolean logRetryCache)
    throws IOException {
  final INodesInPath iip = fsd.getINodesInPath4Write(snapshotRoot);
  if (fsd.isPermissionEnabled()) {
    FSPermissionChecker pc = fsd.getPermissionChecker();
    fsd.checkOwner(pc, iip);
  }

  if (snapshotName == null || snapshotName.isEmpty()) {
    snapshotName = Snapshot.generateDefaultSnapshotName();
  } else if (!DFSUtil.isValidNameForComponent(snapshotName)) {
    throw new InvalidPathException("Invalid snapshot name: " + snapshotName);
  }

  String snapshotPath = null;
  verifySnapshotName(fsd, snapshotName, snapshotRoot);
  fsd.writeLock();
  try {
    snapshotPath = snapshotManager.createSnapshot(iip, snapshotRoot,
        snapshotName);
  } finally {
    fsd.writeUnlock();
  }
  fsd.getEditLog().logCreateSnapshot(snapshotRoot, snapshotName,
      logRetryCache);

  return snapshotPath;
}
 

private void testInvalidSymlinkTarget(NamenodeProtocols nnRpc,
    String invalidTarget, String link) throws IOException {
  try {
    FsPermission perm = FsPermission.createImmutable((short)0755);
    nnRpc.createSymlink(invalidTarget, link, perm, false);
    fail("Symbolic link creation of target " + invalidTarget + " should fail");
  } catch (InvalidPathException expected) {
    // Expected
  }
}
 

/**
 * Same {{@link #mkdirs(String, FsPermission, boolean)} except
 * that the permissions has already been masked against umask.
 */
public boolean primitiveMkdir(String src, FsPermission absPermission, 
  boolean createParent)
  throws IOException {
  checkOpen();
  if (absPermission == null) {
    absPermission = 
      FsPermission.getDefault().applyUMask(dfsClientConf.uMask);
  } 

  if(LOG.isDebugEnabled()) {
    LOG.debug(src + ": masked=" + absPermission);
  }
  TraceScope scope = Trace.startSpan("mkdir", traceSampler);
  try {
    return namenode.mkdirs(src, absPermission, createParent);
  } catch(RemoteException re) {
    throw re.unwrapRemoteException(AccessControlException.class,
                                   InvalidPathException.class,
                                   FileAlreadyExistsException.class,
                                   FileNotFoundException.class,
                                   ParentNotDirectoryException.class,
                                   SafeModeException.class,
                                   NSQuotaExceededException.class,
                                   DSQuotaExceededException.class,
                                   UnresolvedPathException.class,
                                   SnapshotAccessControlException.class);
  } finally {
    scope.close();
  }
}
 

/**
 * The new rename which has the POSIX semantic.
 */
static Map.Entry<BlocksMapUpdateInfo, HdfsFileStatus> renameToInt(
    FSDirectory fsd, final String srcArg, final String dstArg,
    boolean logRetryCache, Options.Rename... options)
    throws IOException {
  String src = srcArg;
  String dst = dstArg;
  if (NameNode.stateChangeLog.isDebugEnabled()) {
    NameNode.stateChangeLog.debug("DIR* NameSystem.renameTo: with options -" +
        " " + src + " to " + dst);
  }
  if (!DFSUtil.isValidName(dst)) {
    throw new InvalidPathException("Invalid name: " + dst);
  }
  final FSPermissionChecker pc = fsd.getPermissionChecker();

  byte[][] srcComponents = FSDirectory.getPathComponentsForReservedPath(src);
  byte[][] dstComponents = FSDirectory.getPathComponentsForReservedPath(dst);
  BlocksMapUpdateInfo collectedBlocks = new BlocksMapUpdateInfo();
  src = fsd.resolvePath(pc, src, srcComponents);
  dst = fsd.resolvePath(pc, dst, dstComponents);
  renameTo(fsd, pc, src, dst, collectedBlocks, logRetryCache, options);
  INodesInPath dstIIP = fsd.getINodesInPath(dst, false);
  HdfsFileStatus resultingStat = fsd.getAuditFileInfo(dstIIP);

  return new AbstractMap.SimpleImmutableEntry<>(
      collectedBlocks, resultingStat);
}
 

/**
 * Create a snapshot
 * @param snapshotRoot The directory path where the snapshot is taken
 * @param snapshotName The name of the snapshot
 */
static String createSnapshot(
    FSDirectory fsd, SnapshotManager snapshotManager, String snapshotRoot,
    String snapshotName, boolean logRetryCache)
    throws IOException {
  final INodesInPath iip = fsd.getINodesInPath4Write(snapshotRoot);
  if (fsd.isPermissionEnabled()) {
    FSPermissionChecker pc = fsd.getPermissionChecker();
    fsd.checkOwner(pc, iip);
  }

  if (snapshotName == null || snapshotName.isEmpty()) {
    snapshotName = Snapshot.generateDefaultSnapshotName();
  } else if (!DFSUtil.isValidNameForComponent(snapshotName)) {
    throw new InvalidPathException("Invalid snapshot name: " + snapshotName);
  }

  String snapshotPath = null;
  verifySnapshotName(fsd, snapshotName, snapshotRoot);
  fsd.writeLock();
  try {
    snapshotPath = snapshotManager.createSnapshot(iip, snapshotRoot,
        snapshotName);
  } finally {
    fsd.writeUnlock();
  }
  fsd.getEditLog().logCreateSnapshot(snapshotRoot, snapshotName,
      logRetryCache);

  return snapshotPath;
}
 

private void testInvalidSymlinkTarget(NamenodeProtocols nnRpc,
    String invalidTarget, String link) throws IOException {
  try {
    FsPermission perm = FsPermission.createImmutable((short)0755);
    nnRpc.createSymlink(invalidTarget, link, perm, false);
    fail("Symbolic link creation of target " + invalidTarget + " should fail");
  } catch (InvalidPathException expected) {
    // Expected
  }
}
 

/** {@inheritDoc} */
@Override protected void checkPath(Path path) {
    URI uri = path.toUri();

    if (uri.isAbsolute()) {
        if (!F.eq(uri.getScheme(), IGFS_SCHEME))
            throw new InvalidPathException("Wrong path scheme [expected=" + IGFS_SCHEME + ", actual=" +
                uri.getAuthority() + ']');

        if (!F.eq(uri.getAuthority(), uriAuthority))
            throw new InvalidPathException("Wrong path authority [expected=" + uriAuthority + ", actual=" +
                uri.getAuthority() + ']');
    }
}
 

/** {@inheritDoc} */
@Override public void checkPath(Path path) {
    URI uri = path.toUri();

    if (uri.isAbsolute()) {
        if (!F.eq(uri.getScheme(), IGFS_SCHEME))
            throw new InvalidPathException("Wrong path scheme [expected=" + IGFS_SCHEME + ", actual=" +
                uri.getAuthority() + ']');

        if (!F.eq(uri.getAuthority(), uriAuthority))
            throw new InvalidPathException("Wrong path authority [expected=" + uriAuthority + ", actual=" +
                uri.getAuthority() + ']');
    }
}
 

@Test
public void testCheckPath_shouldThrowExceptionForMismatchingBucket()
    throws IOException, URISyntaxException {
  Configuration config = GoogleHadoopFileSystemIntegrationHelper.getTestConfig();
  GoogleHadoopFS ghfs = new GoogleHadoopFS(initUri, config);

  Path testPath = new Path("gs://fake/file");

  InvalidPathException e =
      assertThrows(InvalidPathException.class, () -> ghfs.checkPath(testPath));

  assertThat(e).hasMessageThat().startsWith("Invalid path");
}
 

static HdfsFileStatus createSymlinkInt(
    FSNamesystem fsn, String target, final String linkArg,
    PermissionStatus dirPerms, boolean createParent, boolean logRetryCache)
    throws IOException {
  FSDirectory fsd = fsn.getFSDirectory();
  String link = linkArg;
  if (!DFSUtil.isValidName(link)) {
    throw new InvalidPathException("Invalid link name: " + link);
  }
  if (FSDirectory.isReservedName(target) || target.isEmpty()) {
    throw new InvalidPathException("Invalid target name: " + target);
  }

  if (NameNode.stateChangeLog.isDebugEnabled()) {
    NameNode.stateChangeLog.debug("DIR* NameSystem.createSymlink: target="
        + target + " link=" + link);
  }

  FSPermissionChecker pc = fsn.getPermissionChecker();
  byte[][] pathComponents = FSDirectory.getPathComponentsForReservedPath(link);
  INodesInPath iip;
  fsd.writeLock();
  try {
    link = fsd.resolvePath(pc, link, pathComponents);
    iip = fsd.getINodesInPath4Write(link, false);
    if (!createParent) {
      fsd.verifyParentDir(iip, link);
    }
    if (!fsd.isValidToCreate(link, iip)) {
      throw new IOException(
          "failed to create link " + link +
              " either because the filename is invalid or the file exists");
    }
    if (fsd.isPermissionEnabled()) {
      fsd.checkAncestorAccess(pc, iip, FsAction.WRITE);
    }
    // validate that we have enough inodes.
    fsn.checkFsObjectLimit();

    // add symbolic link to namespace
    addSymlink(fsd, link, iip, target, dirPerms, createParent, logRetryCache);
  } finally {
    fsd.writeUnlock();
  }
  NameNode.getNameNodeMetrics().incrCreateSymlinkOps();
  return fsd.getAuditFileInfo(iip);
}
 

static HdfsFileStatus createSymlinkInt(
    FSNamesystem fsn, String target, final String linkArg,
    PermissionStatus dirPerms, boolean createParent, boolean logRetryCache)
    throws IOException {
  FSDirectory fsd = fsn.getFSDirectory();
  String link = linkArg;
  if (!DFSUtil.isValidName(link)) {
    throw new InvalidPathException("Invalid link name: " + link);
  }
  if (FSDirectory.isReservedName(target) || target.isEmpty()) {
    throw new InvalidPathException("Invalid target name: " + target);
  }

  if (NameNode.stateChangeLog.isDebugEnabled()) {
    NameNode.stateChangeLog.debug("DIR* NameSystem.createSymlink: target="
        + target + " link=" + link);
  }

  FSPermissionChecker pc = fsn.getPermissionChecker();
  byte[][] pathComponents = FSDirectory.getPathComponentsForReservedPath(link);
  INodesInPath iip;
  fsd.writeLock();
  try {
    link = fsd.resolvePath(pc, link, pathComponents);
    iip = fsd.getINodesInPath4Write(link, false);
    if (!createParent) {
      fsd.verifyParentDir(iip, link);
    }
    if (!fsd.isValidToCreate(link, iip)) {
      throw new IOException(
          "failed to create link " + link +
              " either because the filename is invalid or the file exists");
    }
    if (fsd.isPermissionEnabled()) {
      fsd.checkAncestorAccess(pc, iip, FsAction.WRITE);
    }
    // validate that we have enough inodes.
    fsn.checkFsObjectLimit();

    // add symbolic link to namespace
    addSymlink(fsd, link, iip, target, dirPerms, createParent, logRetryCache);
  } finally {
    fsd.writeUnlock();
  }
  NameNode.getNameNodeMetrics().incrCreateSymlinkOps();
  return fsd.getAuditFileInfo(iip);
}