Feature: Added support for FTP_TLS (Weak Cryptography Issue)

storages/backends/ftp.py

@@ -6,11 +6,12 @@
 # Usage:
 #
 # Add below to settings.py:
-# FTP_STORAGE_LOCATION = '[a]ftp://<user>:<pass>@<host>:<port>/[path]'
+# FTP_STORAGE_LOCATION = '[a]ftp[s]://<user>:<pass>@<host>:<port>/[path]'
 #
 # In models.py you can write:
 # from FTPStorage import FTPStorage
 # fs = FTPStorage()
+# For a TLS configuration, you must use 'ftps' protocol
 # class FTPTest(models.Model):
 #     file = models.FileField(upload_to='a/b/c/', storage=fs)
 
@@ -57,7 +58,7 @@ def _decode_location(self, location):
         splitted_url = urlparse(location)
         config = {}
 
-        if splitted_url.scheme not in ("ftp", "aftp"):
+        if splitted_url.scheme not in ("ftp", "aftp", "ftps"):
             raise ImproperlyConfigured("FTPStorage works only with FTP protocol!")
         if splitted_url.hostname == "":
             raise ImproperlyConfigured("You must at least provide hostname!")
@@ -66,6 +67,12 @@ def _decode_location(self, location):
             config["active"] = True
         else:
             config["active"] = False
+
+        if splitted_url.scheme == "ftps":
+            config["secure"] = True
+        else:
+            config["secure"] = False
+
         config["path"] = splitted_url.path
         config["host"] = splitted_url.hostname
         config["user"] = splitted_url.username
@@ -84,11 +91,13 @@ def _start_connection(self):
 
         # Real reconnect
         if self._connection is None:
-            ftp = ftplib.FTP()
+            ftp = ftplib.FTP_TLS() if self._config["secure"] else ftplib.FTP()
             ftp.encoding = self.encoding
             try:
                 ftp.connect(self._config["host"], self._config["port"])
                 ftp.login(self._config["user"], self._config["passwd"])
+                if self._config["secure"]:
+                    ftp.prot_p()
                 if self._config["active"]:
                     ftp.set_pasv(False)
                 if self._config["path"] != "":

tests/test_ftp.py

@@ -14,6 +14,9 @@
 URL = "ftp://{user}:{passwd}@{host}:{port}/".format(
     user=USER, passwd=PASSWORD, host=HOST, port=PORT
 )
+URL_TLS = "ftps://{user}:{passwd}@{host}:{port}/".format(
+    user=USER, passwd=PASSWORD, host=HOST, port=PORT
+)
 
 LIST_FIXTURE = """drwxr-xr-x   2 ftp      nogroup      4096 Jul 27 09:46 dir
 -rw-r--r--   1 ftp      nogroup      1024 Jul 27 09:45 fi
@@ -48,6 +51,7 @@ def test_decode_location(self):
             "active": False,
             "path": "/",
             "port": 2121,
+            "secure": False
         }
         self.assertEqual(config, wanted_config)
         # Test active FTP
@@ -59,6 +63,7 @@ def test_decode_location(self):
             "active": True,
             "path": "/",
             "port": 2121,
+            "secure": False
         }
         self.assertEqual(config, wanted_config)
 
@@ -239,3 +244,49 @@ def test_close(self, mock_ftp, mock_storage):
         file_.is_dirty = True
         file_.read()
         file_.close()
+
+
+class FTPTLSTest(TestCase):
+    def setUp(self):
+        self.storage = ftp.FTPStorage(location=URL_TLS)
+
+    def test_init_no_location(self):
+        with self.assertRaises(ImproperlyConfigured):
+            ftp.FTPStorage()
+
+    @patch("storages.backends.ftp.setting", return_value=URL_TLS)
+    def test_init_location_from_setting(self, mock_setting):
+        storage = ftp.FTPStorage()
+        self.assertTrue(mock_setting.called)
+        self.assertEqual(storage.location, URL_TLS)
+
+    def test_decode_location(self):
+        config = self.storage._decode_location(URL_TLS)
+        wanted_config = {
+            "passwd": "b@r",
+            "host": "localhost",
+            "user": "foo",
+            "active": False,
+            "path": "/",
+            "port": 2121,
+            "secure": True
+        }
+        self.assertEqual(config, wanted_config)
+
+    def test_decode_location_error(self):
+        with self.assertRaises(ImproperlyConfigured):
+            self.storage._decode_location("foo")
+        with self.assertRaises(ImproperlyConfigured):
+            self.storage._decode_location("http://foo.pt")
+        # TODO: Cannot not provide a port
+        # with self.assertRaises(ImproperlyConfigured):
+        #     self.storage._decode_location('ftp://')
+
+    @patch("ftplib.FTP_TLS")
+    def test_start_connection(self, mock_ftp):
+        self.storage._start_connection()
+        self.assertIsNotNone(self.storage._connection)
+        # Start active
+        storage = ftp.FTPStorage(location=URL_TLS)
+        storage._start_connection()
+