PLUGIN-1525 fix colon in password for sftp

fix the logic to extract the password to be a little better and
work properly if the password contains a colon. This does not
fix issues if the username contains a colon, and only fixes
issues for SFTP and not FTP. This is because the underlying
Hadoop FTPFileSystem also does not handle colons in username or
password properly, while the SFTPFileSystem does not have that
same bug.

src/main/java/io/cdap/plugin/batch/source/ftp/FTPBatchSource.java

@@ -330,12 +330,19 @@ String getPathFromConfig() {
     }
 
     /**
-     * This method extracts the password from url
+     * This method extracts the password from url.
+     * This does some manual URI parsing and will not work properly if the username contains a colon,
+     * and will not work properly in all sorts of edge cases with malformed URIs.
+     * However, the underlying Hadoop FTP library also does not work properly on usernames with colon (see PLUGIN-1525).
      */
     public String extractPasswordFromUrl() {
+      if (!path.startsWith("ftp://") && !path.startsWith("sftp://")) {
+        throw new IllegalArgumentException("Invalid path. Please ensure that it starts with ftp:// or sftp://");
+      }
       int getLastIndexOfAtSign = path.lastIndexOf("@");
       String authentication = path.substring(0, getLastIndexOfAtSign);
-      return authentication.substring(authentication.lastIndexOf(":") + 1);
+      int usernameEndIndex = authentication.indexOf(':', authentication.indexOf(':') + 1);
+      return authentication.substring(usernameEndIndex + 1);
     }
 
     /**

src/test/java/io/cdap/plugin/batch/source/ftp/FTPBatchSourceTest.java

@@ -44,7 +44,7 @@
 public class FTPBatchSourceTest {
 
   private static final String USER = "user";
-  private static final String PASSWORD_WITH_SPECIAL_CHARACTERS = "wex^Yz@123#456!";
+  private static final String PASSWORD_WITH_SPECIAL_CHARACTERS = "we:/_x^Yz @123#456!";
   private static final String PASSWORD_WITHOUT_SPECIAL_CHARACTERS = "wexYz123456";
   private static final String HOST = "localhost";
   private static final int FTP_DEFAULT_PORT = 21;
@@ -82,8 +82,7 @@ public static void ftpTeardown() {
 
   @Test
   public void testSchemaDetection() {
-    String path = String.format("ftp://user2:%s@%s:%d/tmp/file1.txt",
-                                PASSWORD_WITH_SPECIAL_CHARACTERS, HOST, ftpServerPort);
+    String path = String.format("ftp://user:password@%s:%d/tmp/file1.txt", HOST, ftpServerPort);
     FTPBatchSource.FTPBatchSourceConfig config = new FTPBatchSource.FTPBatchSourceConfig(path, "csv", false);
     FTPBatchSource ftpBatchSource = new FTPBatchSource(config);
     Map<String, Object> plugins = new HashMap<>();