diff --git a/Doc/whatsnew/2.7.rst b/Doc/whatsnew/2.7.rst
index bc297ef4ee89fb..71d410bcd1fbf0 100644
--- a/Doc/whatsnew/2.7.rst
+++ b/Doc/whatsnew/2.7.rst
@@ -2777,6 +2777,15 @@ It has been replaced by the new ``make regen-all`` target.
 
 .. _acks27:
 
+Security fix for FTP
+================================
+
+A security fix alters the :class:`ftplib.FTP` behavior to not trust the
+IPv4 address sent from the remote server when setting up a passive data
+channel.  We reuse the ftp server IP address instead.  For unusual code
+requiring the old behavior, set a ``trust_server_pasv_ipv4_address``
+attribute on your FTP instance to ``True``.  (See :issue:`43285`)
+
 Acknowledgements
 ================
 
diff --git a/Lib/ftplib.py b/Lib/ftplib.py
index 6644554792791b..7c772e6ee5ee16 100644
--- a/Lib/ftplib.py
+++ b/Lib/ftplib.py
@@ -108,23 +108,46 @@ class FTP:
     file = None
     welcome = None
     passiveserver = 1
+    encoding = "latin-1"
+    # # Disables https://bugs.python.org/issue43285 security if set to True.
+    # trust_server_pasv_ipv4_address = False
 
     # Initialization method (called by class instantiation).
     # Initialize host to localhost, port to standard ftp port
     # Optional arguments are host (for connect()),
     # and user, passwd, acct (for login())
     def __init__(self, host='', user='', passwd='', acct='',
-                 timeout=_GLOBAL_DEFAULT_TIMEOUT):
+                 timeout=_GLOBAL_DEFAULT_TIMEOUT, source_address=None):
+        self.source_address = source_address
+        # Disables https://bugs.python.org/issue43285 security if set to True.
+        self.trust_server_pasv_ipv4_address = False
         self.timeout = timeout
         if host:
             self.connect(host)
             if user:
                 self.login(user, passwd, acct)
 
-    def connect(self, host='', port=0, timeout=-999):
+    def __enter__(self):
+        return self
+
+    # Context management protocol: try to quit() if active
+    def __exit__(self, *args):
+        if self.sock is not None:
+            try:
+                self.quit()
+            except (OSError, EOFError):
+                pass
+            finally:
+                if self.sock is not None:
+                    self.close()
+
+    def connect(self, host='', port=0, timeout=-999, source_address=None):
         '''Connect to host.  Arguments are:
          - host: hostname to connect to (string, default previous host)
          - port: port to connect to (integer, default previous port)
+         - timeout: the timeout to set against the ftp socket(s)
+         - source_address: a 2-tuple (host, port) for the socket to bind
+           to as its source address before connecting.
         '''
         if host != '':
             self.host = host
@@ -132,6 +155,8 @@ def connect(self, host='', port=0, timeout=-999):
             self.port = port
         if timeout != -999:
             self.timeout = timeout
+        if source_address is not None:
+            self.source_address = source_address
         self.sock = socket.create_connection((self.host, self.port), self.timeout)
         self.af = self.sock.family
         self.file = self.sock.makefile('rb')
@@ -310,8 +335,13 @@ def makeport(self):
         return sock
 
     def makepasv(self):
+        """Internal: Does the PASV or EPSV handshake -> (address, port)"""
         if self.af == socket.AF_INET:
-            host, port = parse227(self.sendcmd('PASV'))
+            untrusted_host, port = parse227(self.sendcmd('PASV'))
+            if self.trust_server_pasv_ipv4_address:
+                host = untrusted_host
+            else:
+                host = self.sock.getpeername()[0]
         else:
             host, port = parse229(self.sendcmd('EPSV'), self.sock.getpeername())
         return host, port
diff --git a/Lib/test/test_ftplib.py b/Lib/test/test_ftplib.py
index 8a3eb067a4462b..a1d725206de7c3 100644
--- a/Lib/test/test_ftplib.py
+++ b/Lib/test/test_ftplib.py
@@ -15,6 +15,7 @@
 except ImportError:
     ssl = None
 
+from contextlib import closing
 from unittest import TestCase, SkipTest, skipUnless
 from test import test_support
 from test.test_support import HOST, HOSTv6
@@ -67,6 +68,10 @@ def __init__(self, conn):
         self.rest = None
         self.next_retr_data = RETR_DATA
         self.push('220 welcome')
+        # We use this as the string IPv4 address to direct the client
+        # to in response to a PASV command.  To test security behavior.
+        # https://bugs.python.org/issue43285/.
+        self.fake_pasv_server_ip = '252.253.254.255'
 
     def collect_incoming_data(self, data):
         self.in_buffer.append(data)
@@ -109,13 +114,13 @@ def cmd_pasv(self, arg):
         sock.bind((self.socket.getsockname()[0], 0))
         sock.listen(5)
         sock.settimeout(10)
-        ip, port = sock.getsockname()[:2]
-        ip = ip.replace('.', ',')
-        p1, p2 = divmod(port, 256)
+        port = sock.getsockname()[1]
+        ip = self.fake_pasv_server_ip
+        ip = ip.replace('.', ','); p1 = port / 256; p2 = port % 256
         self.push('227 entering passive mode (%s,%d,%d)' %(ip, p1, p2))
         conn, addr = sock.accept()
         self.dtp = self.dtp_handler(conn, baseclass=self)
-
+        
     def cmd_eprt(self, arg):
         af, ip, port = arg.split(arg[0])[1:-1]
         port = int(port)
@@ -577,6 +582,107 @@ def test_makepasv(self):
         # IPv4 is in use, just make sure send_epsv has not been used
         self.assertEqual(self.server.handler_instance.last_received_cmd, 'pasv')
 
+    def test_makepasv_issue43285_security_disabled(self):
+        """Test the opt-in to the old vulnerable behavior."""
+        self.client.trust_server_pasv_ipv4_address = True
+        bad_host, port = self.client.makepasv()
+        self.assertEqual(
+                bad_host, self.server.handler_instance.fake_pasv_server_ip)
+        # Opening and closing a connection keeps the dummy server happy
+        # instead of timing out on accept.
+        socket.create_connection((self.client.sock.getpeername()[0], port),
+                                 timeout=TIMEOUT).close()
+
+    def test_makepasv_issue43285_security_enabled_default(self):
+        self.assertFalse(self.client.trust_server_pasv_ipv4_address)
+        trusted_host, port = self.client.makepasv()
+        self.assertNotEqual(
+                trusted_host, self.server.handler_instance.fake_pasv_server_ip)
+        # Opening and closing a connection keeps the dummy server happy
+        # instead of timing out on accept.
+        socket.create_connection((trusted_host, port), timeout=TIMEOUT).close()
+
+    def test_with_statement(self):
+        self.client.quit()
+
+        def is_client_connected():
+            if self.client.sock is None:
+                return False
+            try:
+                self.client.sendcmd('noop')
+            except (OSError, EOFError):
+                return False
+            return True
+
+        # base test
+        with ftplib.FTP(timeout=TIMEOUT) as self.client:
+            self.client.connect(self.server.host, self.server.port)
+            self.client.sendcmd('noop')
+            self.assertTrue(is_client_connected())
+        self.assertEqual(self.server.handler_instance.last_received_cmd, 'quit')
+        self.assertFalse(is_client_connected())
+
+        # QUIT sent inside the with block
+        with ftplib.FTP(timeout=TIMEOUT) as self.client:
+            self.client.connect(self.server.host, self.server.port)
+            self.client.sendcmd('noop')
+            self.client.quit()
+        self.assertEqual(self.server.handler_instance.last_received_cmd, 'quit')
+        self.assertFalse(is_client_connected())
+
+        # force a wrong response code to be sent on QUIT: error_perm
+        # is expected and the connection is supposed to be closed
+        try:
+            with ftplib.FTP(timeout=TIMEOUT) as self.client:
+                self.client.connect(self.server.host, self.server.port)
+                self.client.sendcmd('noop')
+                self.server.handler_instance.next_response = '550 error on quit'
+        except ftplib.error_perm as err:
+            self.assertEqual(str(err), '550 error on quit')
+        else:
+            self.fail('Exception not raised')
+        # needed to give the threaded server some time to set the attribute
+        # which otherwise would still be == 'noop'
+        time.sleep(0.1)
+        self.assertEqual(self.server.handler_instance.last_received_cmd, 'quit')
+        self.assertFalse(is_client_connected())
+
+    def test_source_address(self):
+        self.client.quit()
+        port = test_support.find_unused_port()
+        try:
+            self.client.connect(self.server.host, self.server.port,
+                                source_address=(HOST, port))
+            self.assertEqual(self.client.sock.getsockname()[1], port)
+            self.client.quit()
+        except OSError as e:
+            if e.errno == errno.EADDRINUSE:
+                self.skipTest("couldn't bind to port %d" % port)
+            raise
+
+    def test_source_address_passive_connection(self):
+        port = test_support.find_unused_port()
+        self.client.source_address = (HOST, port)
+        try:
+            with closing(self.client.transfercmd('list')) as sock:
+                self.assertEqual(sock.getsockname()[1], port)
+        except OSError as e:
+            if e.errno == errno.EADDRINUSE:
+                self.skipTest("couldn't bind to port %d" % port)
+            raise
+
+    def test_parse257(self):
+        self.assertEqual(ftplib.parse257('257 "/foo/bar"'), '/foo/bar')
+        self.assertEqual(ftplib.parse257('257 "/foo/bar" created'), '/foo/bar')
+        self.assertEqual(ftplib.parse257('257 ""'), '')
+        self.assertEqual(ftplib.parse257('257 "" created'), '')
+        self.assertRaises(ftplib.error_reply, ftplib.parse257, '250 "/foo/bar"')
+        # The 257 response is supposed to include the directory
+        # name and in case it contains embedded double-quotes
+        # they must be doubled (see RFC-959, chapter 7, appendix 2).
+        self.assertEqual(ftplib.parse257('257 "/foo/b""ar"'), '/foo/b"ar')
+        self.assertEqual(ftplib.parse257('257 "/foo/b""ar" created'), '/foo/b"ar')
+
     def test_line_too_long(self):
         self.assertRaises(ftplib.Error, self.client.sendcmd,
                           'x' * self.client.maxline * 2)
diff --git a/Misc/NEWS.d/next/Security/2021-03-13-03-48-14.bpo-43285.g-Hah3.rst b/Misc/NEWS.d/next/Security/2021-03-13-03-48-14.bpo-43285.g-Hah3.rst
new file mode 100644
index 00000000000000..8312b7e885441d
--- /dev/null
+++ b/Misc/NEWS.d/next/Security/2021-03-13-03-48-14.bpo-43285.g-Hah3.rst
@@ -0,0 +1,8 @@
+:mod:`ftplib` no longer trusts the IP address value returned from the server
+in response to the PASV command by default.  This prevents a malicious FTP
+server from using the response to probe IPv4 address and port combinations
+on the client network.
+
+Code that requires the former vulnerable behavior may set a
+``trust_server_pasv_ipv4_address`` attribute on their
+:class:`ftplib.FTP` instances to ``True`` to re-enable it.