Initial commit.

Author: githubChenchi

ftplibtls.py

@@ -0,0 +1,143 @@
+# Copyright (c) Quectel Wireless Solution, Co., Ltd.All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# -*- coding: utf-8 -*-
+"""An FTP subclass which adds FTP-over-TLS support as described in RFC-4217.
+
+Example with FTP-over-TLS. Note that you must call the ``prot_b`` method after
+connecting and authentication to actually establish secure communication for
+data transfers::
+
+    >>> from ftplibtls import FTP_TLS
+    >>> ftp = FTP_TLS('servername')  # default port 21
+    >>> ftp.login('username', 'password')
+    >>> ftp.prot_p()
+    >>> ftp.retrlines('LIST')
+
+"""
+
+# Based on CPython standard library implementation adapted for MicroPython
+
+try:
+    import _ssl
+except ImportError:
+    import ussl as _ssl
+
+try:
+    import socket as _socket
+except ImportError:
+    import usocket as _socket
+
+import ftplib
+
+
+class FTP_TLS(ftplib.FTP):
+    """An FTP subclass which adds FTP-over-TLS support as described in RFC-4217.
+
+    Connect as usual to port 21, implicitly securing the FTP control connection
+    before authenticating.
+
+    Securing the data connection requires the user to explicitly ask for it by
+    calling the ``prot_p()`` method.
+
+    The ``ssl`` module of the ESP8266 port of MicroPython does not support
+    certficate validation, so the following instantiation argument is
+    ignored:
+
+    * ``cert_reqs``
+
+    See the module docstring for a usage example.
+
+    """
+
+    def __init__(self, host=None, port=None, user=None, passwd=None, acct=None,
+                 keyfile=None, certfile=None, cert_reqs=None,
+                 timeout=ftplib._GLOBAL_DEFAULT_TIMEOUT, source_address=None, ipvtype = 0):
+        self._prot_p = False
+        self.keyfile = keyfile
+        self.certfile = certfile
+        self._keydata = None
+        self._certdata = None
+        super().__init__(host, port, user, passwd, acct, timeout, source_address, ipvtype = ipvtype)
+
+    def login(self, user=None, passwd=None, acct=None, secure=True):
+        if secure and isinstance(self.sock._sock, _socket.socket):
+            self.auth()
+
+        return super().login(user, passwd, acct)
+
+    def auth(self):
+        """Set up secure control connection by using TLS/SSL."""
+        if not isinstance(self.sock._sock, _socket.socket):
+            raise ValueError("Already using TLS")
+
+        resp = self.voidcmd('AUTH TLS')
+        self.sock._sock = self._wrap_socket(self.sock._sock)
+        self.file = self.sock._sock
+        return resp
+
+    def ccc(self):
+        """Switch back to a clear-text control connection."""
+        if isinstance(self.sock._sock, _socket.socket):
+            raise ValueError("Not using TLS")
+
+        resp = self.voidcmd('CCC')
+        self.sock._sock = self.sock.unwrap()
+        self.file = self.sock._sock
+        return resp
+
+    def prot_p(self):
+        """Set up secure data connection."""
+        # PROT defines whether or not the data channel is to be protected.
+        # Though RFC-2228 defines four possible protection levels,
+        # RFC-4217 only recommends two, Clear and Private.
+        # Clear (PROT C) means that no security is to be used on the
+        # data-channel, Private (PROT P) means that the data-channel
+        # should be protected by TLS.
+        # PBSZ command MUST still be issued, but must have a parameter of
+        # '0' to indicate that no buffering is taking place and the data
+        # connection should not be encapsulated.
+        self.voidcmd('PBSZ 0')
+        resp = self.voidcmd('PROT P')
+        self._prot_p = True
+        return resp
+
+    def prot_c(self):
+        """Set up clear text data connection."""
+        resp = self.voidcmd('PROT C')
+        self._prot_p = False
+        return resp
+
+    # Overridden FTP methods
+
+    def ntransfercmd(self, cmd, rest=None):
+        conn, size = super().ntransfercmd(cmd, rest)
+
+        if self._prot_p:
+            conn._sock = self._wrap_socket(conn._sock)
+
+        return conn, size
+
+    # Internal helper methods
+
+    def _wrap_socket(self, socket):
+        if self.keyfile and not self._keydata:
+                with open(self.keyfile, 'rb') as f:
+                    self._keydata = f.read()
+
+        if self.certfile and not self._certdata:
+            with open(self.certfile, 'rb') as f:
+                self._certdata = f.read()
+
+        return _ssl.wrap_socket(socket, key=self._keydata, cert=self._certdata)