diff --git a/src/main/java/io/cdap/plugin/batch/source/ftp/FTPBatchSource.java b/src/main/java/io/cdap/plugin/batch/source/ftp/FTPBatchSource.java
index f9d53e4..11d0ace 100644
--- a/src/main/java/io/cdap/plugin/batch/source/ftp/FTPBatchSource.java
+++ b/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);
     }
 
     /**
diff --git a/src/test/java/io/cdap/plugin/batch/source/ftp/FTPBatchSourceTest.java b/src/test/java/io/cdap/plugin/batch/source/ftp/FTPBatchSourceTest.java
index 8c89edf..6b4a588 100644
--- a/src/test/java/io/cdap/plugin/batch/source/ftp/FTPBatchSourceTest.java
+++ b/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<>();