We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Allow for specification of cipher in SftpHook
<!--

Welcome to Apache Airflow!  For a smooth issue process, try to answer the following questions.
Don't worry if they're not all applicable; just try to include what you can :-)

If you need to include code snippets or logs, please put them in fenced code
blocks.  If they're super-long, please use the details tag like
<details><summary>super-long log</summary> lots of stuff </details>

Please delete these comment blocks before submitting the issue.

-->

**Description**

I want to specify the cipher to use when creating a SftpHook. Currently this is not possible because there is not a way to propagate the value to CnOpts. 

https://github.com/apache/airflow/blob/master/airflow/providers/sftp/hooks/sftp.py
https://pysftp.readthedocs.io/en/release_0.2.9/pysftp.html?#pysftp.CnOpts
https://pysftp.readthedocs.io/en/release_0.2.9/cookbook.html

**Use case / motivation**

<!-- What do you want to happen?

Rather than telling us how you might implement this solution, try to take a
step back and describe what you are trying to achieve.

-->

I want to be able to access SFTP servers that have disabled access from connection with weak ciphers.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/providers/sftp/hooks/sftp.py`
Content:
```
1 #
2 # Licensed to the Apache Software Foundation (ASF) under one
3 # or more contributor license agreements.  See the NOTICE file
4 # distributed with this work for additional information
5 # regarding copyright ownership.  The ASF licenses this file
6 # to you under the Apache License, Version 2.0 (the
7 # "License"); you may not use this file except in compliance
8 # with the License.  You may obtain a copy of the License at
9 #
10 #   http://www.apache.org/licenses/LICENSE-2.0
11 #
12 # Unless required by applicable law or agreed to in writing,
13 # software distributed under the License is distributed on an
14 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 # KIND, either express or implied.  See the License for the
16 # specific language governing permissions and limitations
17 # under the License.
18 """This module contains SFTP hook."""
19 import datetime
20 import stat
21 from typing import Dict, List, Optional, Tuple
22 
23 import pysftp
24 
25 from airflow.providers.ssh.hooks.ssh import SSHHook
26 
27 
28 class SFTPHook(SSHHook):
29     """
30     This hook is inherited from SSH hook. Please refer to SSH hook for the input
31     arguments.
32 
33     Interact with SFTP. Aims to be interchangeable with FTPHook.
34 
35     :Pitfalls::
36 
37         - In contrast with FTPHook describe_directory only returns size, type and
38           modify. It doesn't return unix.owner, unix.mode, perm, unix.group and
39           unique.
40         - retrieve_file and store_file only take a local full path and not a
41            buffer.
42         - If no mode is passed to create_directory it will be created with 777
43           permissions.
44 
45     Errors that may occur throughout but should be handled downstream.
46     """
47 
48     def __init__(self, ftp_conn_id: str = 'sftp_default', *args, **kwargs) -> None:
49         kwargs['ssh_conn_id'] = ftp_conn_id
50         super().__init__(*args, **kwargs)
51 
52         self.conn = None
53         self.private_key_pass = None
54 
55         # Fail for unverified hosts, unless this is explicitly allowed
56         self.no_host_key_check = False
57 
58         if self.ssh_conn_id is not None:
59             conn = self.get_connection(self.ssh_conn_id)
60             if conn.extra is not None:
61                 extra_options = conn.extra_dejson
62                 if 'private_key_pass' in extra_options:
63                     self.private_key_pass = extra_options.get('private_key_pass')
64 
65                 # For backward compatibility
66                 # TODO: remove in Airflow 2.1
67                 import warnings
68 
69                 if 'ignore_hostkey_verification' in extra_options:
70                     warnings.warn(
71                         'Extra option `ignore_hostkey_verification` is deprecated.'
72                         'Please use `no_host_key_check` instead.'
73                         'This option will be removed in Airflow 2.1',
74                         DeprecationWarning,
75                         stacklevel=2,
76                     )
77                     self.no_host_key_check = (
78                         str(extra_options['ignore_hostkey_verification']).lower() == 'true'
79                     )
80 
81                 if 'no_host_key_check' in extra_options:
82                     self.no_host_key_check = str(extra_options['no_host_key_check']).lower() == 'true'
83 
84                 if 'private_key' in extra_options:
85                     warnings.warn(
86                         'Extra option `private_key` is deprecated.'
87                         'Please use `key_file` instead.'
88                         'This option will be removed in Airflow 2.1',
89                         DeprecationWarning,
90                         stacklevel=2,
91                     )
92                     self.key_file = extra_options.get('private_key')
93 
94     def get_conn(self) -> pysftp.Connection:
95         """Returns an SFTP connection object"""
96         if self.conn is None:
97             cnopts = pysftp.CnOpts()
98             if self.no_host_key_check:
99                 cnopts.hostkeys = None
100             cnopts.compression = self.compress
101             conn_params = {
102                 'host': self.remote_host,
103                 'port': self.port,
104                 'username': self.username,
105                 'cnopts': cnopts,
106             }
107             if self.password and self.password.strip():
108                 conn_params['password'] = self.password
109             if self.key_file:
110                 conn_params['private_key'] = self.key_file
111             if self.private_key_pass:
112                 conn_params['private_key_pass'] = self.private_key_pass
113 
114             self.conn = pysftp.Connection(**conn_params)
115         return self.conn
116 
117     def close_conn(self) -> None:
118         """Closes the connection"""
119         if self.conn is not None:
120             self.conn.close()
121             self.conn = None
122 
123     def describe_directory(self, path: str) -> Dict[str, Dict[str, str]]:
124         """
125         Returns a dictionary of {filename: {attributes}} for all files
126         on the remote system (where the MLSD command is supported).
127 
128         :param path: full path to the remote directory
129         :type path: str
130         """
131         conn = self.get_conn()
132         flist = conn.listdir_attr(path)
133         files = {}
134         for f in flist:
135             modify = datetime.datetime.fromtimestamp(f.st_mtime).strftime('%Y%m%d%H%M%S')
136             files[f.filename] = {
137                 'size': f.st_size,
138                 'type': 'dir' if stat.S_ISDIR(f.st_mode) else 'file',
139                 'modify': modify,
140             }
141         return files
142 
143     def list_directory(self, path: str) -> List[str]:
144         """
145         Returns a list of files on the remote system.
146 
147         :param path: full path to the remote directory to list
148         :type path: str
149         """
150         conn = self.get_conn()
151         files = conn.listdir(path)
152         return files
153 
154     def create_directory(self, path: str, mode: int = 777) -> None:
155         """
156         Creates a directory on the remote system.
157 
158         :param path: full path to the remote directory to create
159         :type path: str
160         :param mode: int representation of octal mode for directory
161         """
162         conn = self.get_conn()
163         conn.makedirs(path, mode)
164 
165     def delete_directory(self, path: str) -> None:
166         """
167         Deletes a directory on the remote system.
168 
169         :param path: full path to the remote directory to delete
170         :type path: str
171         """
172         conn = self.get_conn()
173         conn.rmdir(path)
174 
175     def retrieve_file(self, remote_full_path: str, local_full_path: str) -> None:
176         """
177         Transfers the remote file to a local location.
178         If local_full_path is a string path, the file will be put
179         at that location
180 
181         :param remote_full_path: full path to the remote file
182         :type remote_full_path: str
183         :param local_full_path: full path to the local file
184         :type local_full_path: str
185         """
186         conn = self.get_conn()
187         self.log.info('Retrieving file from FTP: %s', remote_full_path)
188         conn.get(remote_full_path, local_full_path)
189         self.log.info('Finished retrieving file from FTP: %s', remote_full_path)
190 
191     def store_file(self, remote_full_path: str, local_full_path: str) -> None:
192         """
193         Transfers a local file to the remote location.
194         If local_full_path_or_buffer is a string path, the file will be read
195         from that location
196 
197         :param remote_full_path: full path to the remote file
198         :type remote_full_path: str
199         :param local_full_path: full path to the local file
200         :type local_full_path: str
201         """
202         conn = self.get_conn()
203         conn.put(local_full_path, remote_full_path)
204 
205     def delete_file(self, path: str) -> None:
206         """
207         Removes a file on the FTP Server
208 
209         :param path: full path to the remote file
210         :type path: str
211         """
212         conn = self.get_conn()
213         conn.remove(path)
214 
215     def get_mod_time(self, path: str) -> str:
216         """
217         Returns modification time.
218 
219         :param path: full path to the remote file
220         :type path: str
221         """
222         conn = self.get_conn()
223         ftp_mdtm = conn.stat(path).st_mtime
224         return datetime.datetime.fromtimestamp(ftp_mdtm).strftime('%Y%m%d%H%M%S')
225 
226     def path_exists(self, path: str) -> bool:
227         """
228         Returns True if a remote entity exists
229 
230         :param path: full path to the remote file or directory
231         :type path: str
232         """
233         conn = self.get_conn()
234         return conn.exists(path)
235 
236     @staticmethod
237     def _is_path_match(path: str, prefix: Optional[str] = None, delimiter: Optional[str] = None) -> bool:
238         """
239         Return True if given path starts with prefix (if set) and ends with delimiter (if set).
240 
241         :param path: path to be checked
242         :type path: str
243         :param prefix: if set path will be checked is starting with prefix
244         :type prefix: str
245         :param delimiter: if set path will be checked is ending with suffix
246         :type delimiter: str
247         :return: bool
248         """
249         if prefix is not None and not path.startswith(prefix):
250             return False
251         if delimiter is not None and not path.endswith(delimiter):
252             return False
253         return True
254 
255     def get_tree_map(
256         self, path: str, prefix: Optional[str] = None, delimiter: Optional[str] = None
257     ) -> Tuple[List[str], List[str], List[str]]:
258         """
259         Return tuple with recursive lists of files, directories and unknown paths from given path.
260         It is possible to filter results by giving prefix and/or delimiter parameters.
261 
262         :param path: path from which tree will be built
263         :type path: str
264         :param prefix: if set paths will be added if start with prefix
265         :type prefix: str
266         :param delimiter: if set paths will be added if end with delimiter
267         :type delimiter: str
268         :return: tuple with list of files, dirs and unknown items
269         :rtype: Tuple[List[str], List[str], List[str]]
270         """
271         conn = self.get_conn()
272         files, dirs, unknowns = [], [], []  # type: List[str], List[str], List[str]
273 
274         def append_matching_path_callback(list_):
275             return lambda item: list_.append(item) if self._is_path_match(item, prefix, delimiter) else None
276 
277         conn.walktree(
278             remotepath=path,
279             fcallback=append_matching_path_callback(files),
280             dcallback=append_matching_path_callback(dirs),
281             ucallback=append_matching_path_callback(unknowns),
282             recurse=True,
283         )
284 
285         return files, dirs, unknowns
286
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/airflow/providers/sftp/hooks/sftp.py b/airflow/providers/sftp/hooks/sftp.py
--- a/airflow/providers/sftp/hooks/sftp.py
+++ b/airflow/providers/sftp/hooks/sftp.py
@@ -51,6 +51,7 @@
 
         self.conn = None
         self.private_key_pass = None
+        self.ciphers = None
 
         # Fail for unverified hosts, unless this is explicitly allowed
         self.no_host_key_check = False
@@ -80,6 +81,9 @@
 
                 if 'no_host_key_check' in extra_options:
                     self.no_host_key_check = str(extra_options['no_host_key_check']).lower() == 'true'
+                    
+                if 'ciphers' in extra_options:
+                    self.ciphers = extra_options['ciphers']
 
                 if 'private_key' in extra_options:
                     warnings.warn(
@@ -98,6 +102,7 @@
             if self.no_host_key_check:
                 cnopts.hostkeys = None
             cnopts.compression = self.compress
+            cnopts.ciphers = self.ciphers
             conn_params = {
                 'host': self.remote_host,
                 'port': self.port,

{"golden_diff": "diff --git a/airflow/providers/sftp/hooks/sftp.py b/airflow/providers/sftp/hooks/sftp.py\n--- a/airflow/providers/sftp/hooks/sftp.py\n+++ b/airflow/providers/sftp/hooks/sftp.py\n@@ -51,6 +51,7 @@\n \n         self.conn = None\n         self.private_key_pass = None\n+        self.ciphers = None\n \n         # Fail for unverified hosts, unless this is explicitly allowed\n         self.no_host_key_check = False\n@@ -80,6 +81,9 @@\n \n                 if 'no_host_key_check' in extra_options:\n                     self.no_host_key_check = str(extra_options['no_host_key_check']).lower() == 'true'\n+                    \n+                if 'ciphers' in extra_options:\n+                    self.ciphers = extra_options['ciphers']\n \n                 if 'private_key' in extra_options:\n                     warnings.warn(\n@@ -98,6 +102,7 @@\n             if self.no_host_key_check:\n                 cnopts.hostkeys = None\n             cnopts.compression = self.compress\n+            cnopts.ciphers = self.ciphers\n             conn_params = {\n                 'host': self.remote_host,\n                 'port': self.port,\n", "issue": "Allow for specification of cipher in SftpHook\n<!--\r\n\r\nWelcome to Apache Airflow!  For a smooth issue process, try to answer the following questions.\r\nDon't worry if they're not all applicable; just try to include what you can :-)\r\n\r\nIf you need to include code snippets or logs, please put them in fenced code\r\nblocks.  If they're super-long, please use the details tag like\r\n<details><summary>super-long log</summary> lots of stuff </details>\r\n\r\nPlease delete these comment blocks before submitting the issue.\r\n\r\n-->\r\n\r\n**Description**\r\n\r\nI want to specify the cipher to use when creating a SftpHook. Currently this is not possible because there is not a way to propagate the value to CnOpts. \r\n\r\nhttps://github.com/apache/airflow/blob/master/airflow/providers/sftp/hooks/sftp.py\r\nhttps://pysftp.readthedocs.io/en/release_0.2.9/pysftp.html?#pysftp.CnOpts\r\nhttps://pysftp.readthedocs.io/en/release_0.2.9/cookbook.html\r\n\r\n**Use case / motivation**\r\n\r\n<!-- What do you want to happen?\r\n\r\nRather than telling us how you might implement this solution, try to take a\r\nstep back and describe what you are trying to achieve.\r\n\r\n-->\r\n\r\nI want to be able to access SFTP servers that have disabled access from connection with weak ciphers.\r\n\n", "before_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"This module contains SFTP hook.\"\"\"\nimport datetime\nimport stat\nfrom typing import Dict, List, Optional, Tuple\n\nimport pysftp\n\nfrom airflow.providers.ssh.hooks.ssh import SSHHook\n\n\nclass SFTPHook(SSHHook):\n    \"\"\"\n    This hook is inherited from SSH hook. Please refer to SSH hook for the input\n    arguments.\n\n    Interact with SFTP. Aims to be interchangeable with FTPHook.\n\n    :Pitfalls::\n\n        - In contrast with FTPHook describe_directory only returns size, type and\n          modify. It doesn't return unix.owner, unix.mode, perm, unix.group and\n          unique.\n        - retrieve_file and store_file only take a local full path and not a\n           buffer.\n        - If no mode is passed to create_directory it will be created with 777\n          permissions.\n\n    Errors that may occur throughout but should be handled downstream.\n    \"\"\"\n\n    def __init__(self, ftp_conn_id: str = 'sftp_default', *args, **kwargs) -> None:\n        kwargs['ssh_conn_id'] = ftp_conn_id\n        super().__init__(*args, **kwargs)\n\n        self.conn = None\n        self.private_key_pass = None\n\n        # Fail for unverified hosts, unless this is explicitly allowed\n        self.no_host_key_check = False\n\n        if self.ssh_conn_id is not None:\n            conn = self.get_connection(self.ssh_conn_id)\n            if conn.extra is not None:\n                extra_options = conn.extra_dejson\n                if 'private_key_pass' in extra_options:\n                    self.private_key_pass = extra_options.get('private_key_pass')\n\n                # For backward compatibility\n                # TODO: remove in Airflow 2.1\n                import warnings\n\n                if 'ignore_hostkey_verification' in extra_options:\n                    warnings.warn(\n                        'Extra option `ignore_hostkey_verification` is deprecated.'\n                        'Please use `no_host_key_check` instead.'\n                        'This option will be removed in Airflow 2.1',\n                        DeprecationWarning,\n                        stacklevel=2,\n                    )\n                    self.no_host_key_check = (\n                        str(extra_options['ignore_hostkey_verification']).lower() == 'true'\n                    )\n\n                if 'no_host_key_check' in extra_options:\n                    self.no_host_key_check = str(extra_options['no_host_key_check']).lower() == 'true'\n\n                if 'private_key' in extra_options:\n                    warnings.warn(\n                        'Extra option `private_key` is deprecated.'\n                        'Please use `key_file` instead.'\n                        'This option will be removed in Airflow 2.1',\n                        DeprecationWarning,\n                        stacklevel=2,\n                    )\n                    self.key_file = extra_options.get('private_key')\n\n    def get_conn(self) -> pysftp.Connection:\n        \"\"\"Returns an SFTP connection object\"\"\"\n        if self.conn is None:\n            cnopts = pysftp.CnOpts()\n            if self.no_host_key_check:\n                cnopts.hostkeys = None\n            cnopts.compression = self.compress\n            conn_params = {\n                'host': self.remote_host,\n                'port': self.port,\n                'username': self.username,\n                'cnopts': cnopts,\n            }\n            if self.password and self.password.strip():\n                conn_params['password'] = self.password\n            if self.key_file:\n                conn_params['private_key'] = self.key_file\n            if self.private_key_pass:\n                conn_params['private_key_pass'] = self.private_key_pass\n\n            self.conn = pysftp.Connection(**conn_params)\n        return self.conn\n\n    def close_conn(self) -> None:\n        \"\"\"Closes the connection\"\"\"\n        if self.conn is not None:\n            self.conn.close()\n            self.conn = None\n\n    def describe_directory(self, path: str) -> Dict[str, Dict[str, str]]:\n        \"\"\"\n        Returns a dictionary of {filename: {attributes}} for all files\n        on the remote system (where the MLSD command is supported).\n\n        :param path: full path to the remote directory\n        :type path: str\n        \"\"\"\n        conn = self.get_conn()\n        flist = conn.listdir_attr(path)\n        files = {}\n        for f in flist:\n            modify = datetime.datetime.fromtimestamp(f.st_mtime).strftime('%Y%m%d%H%M%S')\n            files[f.filename] = {\n                'size': f.st_size,\n                'type': 'dir' if stat.S_ISDIR(f.st_mode) else 'file',\n                'modify': modify,\n            }\n        return files\n\n    def list_directory(self, path: str) -> List[str]:\n        \"\"\"\n        Returns a list of files on the remote system.\n\n        :param path: full path to the remote directory to list\n        :type path: str\n        \"\"\"\n        conn = self.get_conn()\n        files = conn.listdir(path)\n        return files\n\n    def create_directory(self, path: str, mode: int = 777) -> None:\n        \"\"\"\n        Creates a directory on the remote system.\n\n        :param path: full path to the remote directory to create\n        :type path: str\n        :param mode: int representation of octal mode for directory\n        \"\"\"\n        conn = self.get_conn()\n        conn.makedirs(path, mode)\n\n    def delete_directory(self, path: str) -> None:\n        \"\"\"\n        Deletes a directory on the remote system.\n\n        :param path: full path to the remote directory to delete\n        :type path: str\n        \"\"\"\n        conn = self.get_conn()\n        conn.rmdir(path)\n\n    def retrieve_file(self, remote_full_path: str, local_full_path: str) -> None:\n        \"\"\"\n        Transfers the remote file to a local location.\n        If local_full_path is a string path, the file will be put\n        at that location\n\n        :param remote_full_path: full path to the remote file\n        :type remote_full_path: str\n        :param local_full_path: full path to the local file\n        :type local_full_path: str\n        \"\"\"\n        conn = self.get_conn()\n        self.log.info('Retrieving file from FTP: %s', remote_full_path)\n        conn.get(remote_full_path, local_full_path)\n        self.log.info('Finished retrieving file from FTP: %s', remote_full_path)\n\n    def store_file(self, remote_full_path: str, local_full_path: str) -> None:\n        \"\"\"\n        Transfers a local file to the remote location.\n        If local_full_path_or_buffer is a string path, the file will be read\n        from that location\n\n        :param remote_full_path: full path to the remote file\n        :type remote_full_path: str\n        :param local_full_path: full path to the local file\n        :type local_full_path: str\n        \"\"\"\n        conn = self.get_conn()\n        conn.put(local_full_path, remote_full_path)\n\n    def delete_file(self, path: str) -> None:\n        \"\"\"\n        Removes a file on the FTP Server\n\n        :param path: full path to the remote file\n        :type path: str\n        \"\"\"\n        conn = self.get_conn()\n        conn.remove(path)\n\n    def get_mod_time(self, path: str) -> str:\n        \"\"\"\n        Returns modification time.\n\n        :param path: full path to the remote file\n        :type path: str\n        \"\"\"\n        conn = self.get_conn()\n        ftp_mdtm = conn.stat(path).st_mtime\n        return datetime.datetime.fromtimestamp(ftp_mdtm).strftime('%Y%m%d%H%M%S')\n\n    def path_exists(self, path: str) -> bool:\n        \"\"\"\n        Returns True if a remote entity exists\n\n        :param path: full path to the remote file or directory\n        :type path: str\n        \"\"\"\n        conn = self.get_conn()\n        return conn.exists(path)\n\n    @staticmethod\n    def _is_path_match(path: str, prefix: Optional[str] = None, delimiter: Optional[str] = None) -> bool:\n        \"\"\"\n        Return True if given path starts with prefix (if set) and ends with delimiter (if set).\n\n        :param path: path to be checked\n        :type path: str\n        :param prefix: if set path will be checked is starting with prefix\n        :type prefix: str\n        :param delimiter: if set path will be checked is ending with suffix\n        :type delimiter: str\n        :return: bool\n        \"\"\"\n        if prefix is not None and not path.startswith(prefix):\n            return False\n        if delimiter is not None and not path.endswith(delimiter):\n            return False\n        return True\n\n    def get_tree_map(\n        self, path: str, prefix: Optional[str] = None, delimiter: Optional[str] = None\n    ) -> Tuple[List[str], List[str], List[str]]:\n        \"\"\"\n        Return tuple with recursive lists of files, directories and unknown paths from given path.\n        It is possible to filter results by giving prefix and/or delimiter parameters.\n\n        :param path: path from which tree will be built\n        :type path: str\n        :param prefix: if set paths will be added if start with prefix\n        :type prefix: str\n        :param delimiter: if set paths will be added if end with delimiter\n        :type delimiter: str\n        :return: tuple with list of files, dirs and unknown items\n        :rtype: Tuple[List[str], List[str], List[str]]\n        \"\"\"\n        conn = self.get_conn()\n        files, dirs, unknowns = [], [], []  # type: List[str], List[str], List[str]\n\n        def append_matching_path_callback(list_):\n            return lambda item: list_.append(item) if self._is_path_match(item, prefix, delimiter) else None\n\n        conn.walktree(\n            remotepath=path,\n            fcallback=append_matching_path_callback(files),\n            dcallback=append_matching_path_callback(dirs),\n            ucallback=append_matching_path_callback(unknowns),\n            recurse=True,\n        )\n\n        return files, dirs, unknowns\n", "path": "airflow/providers/sftp/hooks/sftp.py"}], "after_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"This module contains SFTP hook.\"\"\"\nimport datetime\nimport stat\nfrom typing import Dict, List, Optional, Tuple\n\nimport pysftp\n\nfrom airflow.providers.ssh.hooks.ssh import SSHHook\n\n\nclass SFTPHook(SSHHook):\n    \"\"\"\n    This hook is inherited from SSH hook. Please refer to SSH hook for the input\n    arguments.\n\n    Interact with SFTP. Aims to be interchangeable with FTPHook.\n\n    :Pitfalls::\n\n        - In contrast with FTPHook describe_directory only returns size, type and\n          modify. It doesn't return unix.owner, unix.mode, perm, unix.group and\n          unique.\n        - retrieve_file and store_file only take a local full path and not a\n           buffer.\n        - If no mode is passed to create_directory it will be created with 777\n          permissions.\n\n    Errors that may occur throughout but should be handled downstream.\n    \"\"\"\n\n    def __init__(self, ftp_conn_id: str = 'sftp_default', *args, **kwargs) -> None:\n        kwargs['ssh_conn_id'] = ftp_conn_id\n        super().__init__(*args, **kwargs)\n\n        self.conn = None\n        self.private_key_pass = None\n        self.ciphers = None\n\n        # Fail for unverified hosts, unless this is explicitly allowed\n        self.no_host_key_check = False\n\n        if self.ssh_conn_id is not None:\n            conn = self.get_connection(self.ssh_conn_id)\n            if conn.extra is not None:\n                extra_options = conn.extra_dejson\n                if 'private_key_pass' in extra_options:\n                    self.private_key_pass = extra_options.get('private_key_pass')\n\n                # For backward compatibility\n                # TODO: remove in Airflow 2.1\n                import warnings\n\n                if 'ignore_hostkey_verification' in extra_options:\n                    warnings.warn(\n                        'Extra option `ignore_hostkey_verification` is deprecated.'\n                        'Please use `no_host_key_check` instead.'\n                        'This option will be removed in Airflow 2.1',\n                        DeprecationWarning,\n                        stacklevel=2,\n                    )\n                    self.no_host_key_check = (\n                        str(extra_options['ignore_hostkey_verification']).lower() == 'true'\n                    )\n\n                if 'no_host_key_check' in extra_options:\n                    self.no_host_key_check = str(extra_options['no_host_key_check']).lower() == 'true'\n                    \n                if 'ciphers' in extra_options:\n                    self.ciphers = extra_options['ciphers']\n\n                if 'private_key' in extra_options:\n                    warnings.warn(\n                        'Extra option `private_key` is deprecated.'\n                        'Please use `key_file` instead.'\n                        'This option will be removed in Airflow 2.1',\n                        DeprecationWarning,\n                        stacklevel=2,\n                    )\n                    self.key_file = extra_options.get('private_key')\n\n    def get_conn(self) -> pysftp.Connection:\n        \"\"\"Returns an SFTP connection object\"\"\"\n        if self.conn is None:\n            cnopts = pysftp.CnOpts()\n            if self.no_host_key_check:\n                cnopts.hostkeys = None\n            cnopts.compression = self.compress\n            cnopts.ciphers = self.ciphers\n            conn_params = {\n                'host': self.remote_host,\n                'port': self.port,\n                'username': self.username,\n                'cnopts': cnopts,\n            }\n            if self.password and self.password.strip():\n                conn_params['password'] = self.password\n            if self.key_file:\n                conn_params['private_key'] = self.key_file\n            if self.private_key_pass:\n                conn_params['private_key_pass'] = self.private_key_pass\n\n            self.conn = pysftp.Connection(**conn_params)\n        return self.conn\n\n    def close_conn(self) -> None:\n        \"\"\"Closes the connection\"\"\"\n        if self.conn is not None:\n            self.conn.close()\n            self.conn = None\n\n    def describe_directory(self, path: str) -> Dict[str, Dict[str, str]]:\n        \"\"\"\n        Returns a dictionary of {filename: {attributes}} for all files\n        on the remote system (where the MLSD command is supported).\n\n        :param path: full path to the remote directory\n        :type path: str\n        \"\"\"\n        conn = self.get_conn()\n        flist = conn.listdir_attr(path)\n        files = {}\n        for f in flist:\n            modify = datetime.datetime.fromtimestamp(f.st_mtime).strftime('%Y%m%d%H%M%S')\n            files[f.filename] = {\n                'size': f.st_size,\n                'type': 'dir' if stat.S_ISDIR(f.st_mode) else 'file',\n                'modify': modify,\n            }\n        return files\n\n    def list_directory(self, path: str) -> List[str]:\n        \"\"\"\n        Returns a list of files on the remote system.\n\n        :param path: full path to the remote directory to list\n        :type path: str\n        \"\"\"\n        conn = self.get_conn()\n        files = conn.listdir(path)\n        return files\n\n    def create_directory(self, path: str, mode: int = 777) -> None:\n        \"\"\"\n        Creates a directory on the remote system.\n\n        :param path: full path to the remote directory to create\n        :type path: str\n        :param mode: int representation of octal mode for directory\n        \"\"\"\n        conn = self.get_conn()\n        conn.makedirs(path, mode)\n\n    def delete_directory(self, path: str) -> None:\n        \"\"\"\n        Deletes a directory on the remote system.\n\n        :param path: full path to the remote directory to delete\n        :type path: str\n        \"\"\"\n        conn = self.get_conn()\n        conn.rmdir(path)\n\n    def retrieve_file(self, remote_full_path: str, local_full_path: str) -> None:\n        \"\"\"\n        Transfers the remote file to a local location.\n        If local_full_path is a string path, the file will be put\n        at that location\n\n        :param remote_full_path: full path to the remote file\n        :type remote_full_path: str\n        :param local_full_path: full path to the local file\n        :type local_full_path: str\n        \"\"\"\n        conn = self.get_conn()\n        self.log.info('Retrieving file from FTP: %s', remote_full_path)\n        conn.get(remote_full_path, local_full_path)\n        self.log.info('Finished retrieving file from FTP: %s', remote_full_path)\n\n    def store_file(self, remote_full_path: str, local_full_path: str) -> None:\n        \"\"\"\n        Transfers a local file to the remote location.\n        If local_full_path_or_buffer is a string path, the file will be read\n        from that location\n\n        :param remote_full_path: full path to the remote file\n        :type remote_full_path: str\n        :param local_full_path: full path to the local file\n        :type local_full_path: str\n        \"\"\"\n        conn = self.get_conn()\n        conn.put(local_full_path, remote_full_path)\n\n    def delete_file(self, path: str) -> None:\n        \"\"\"\n        Removes a file on the FTP Server\n\n        :param path: full path to the remote file\n        :type path: str\n        \"\"\"\n        conn = self.get_conn()\n        conn.remove(path)\n\n    def get_mod_time(self, path: str) -> str:\n        \"\"\"\n        Returns modification time.\n\n        :param path: full path to the remote file\n        :type path: str\n        \"\"\"\n        conn = self.get_conn()\n        ftp_mdtm = conn.stat(path).st_mtime\n        return datetime.datetime.fromtimestamp(ftp_mdtm).strftime('%Y%m%d%H%M%S')\n\n    def path_exists(self, path: str) -> bool:\n        \"\"\"\n        Returns True if a remote entity exists\n\n        :param path: full path to the remote file or directory\n        :type path: str\n        \"\"\"\n        conn = self.get_conn()\n        return conn.exists(path)\n\n    @staticmethod\n    def _is_path_match(path: str, prefix: Optional[str] = None, delimiter: Optional[str] = None) -> bool:\n        \"\"\"\n        Return True if given path starts with prefix (if set) and ends with delimiter (if set).\n\n        :param path: path to be checked\n        :type path: str\n        :param prefix: if set path will be checked is starting with prefix\n        :type prefix: str\n        :param delimiter: if set path will be checked is ending with suffix\n        :type delimiter: str\n        :return: bool\n        \"\"\"\n        if prefix is not None and not path.startswith(prefix):\n            return False\n        if delimiter is not None and not path.endswith(delimiter):\n            return False\n        return True\n\n    def get_tree_map(\n        self, path: str, prefix: Optional[str] = None, delimiter: Optional[str] = None\n    ) -> Tuple[List[str], List[str], List[str]]:\n        \"\"\"\n        Return tuple with recursive lists of files, directories and unknown paths from given path.\n        It is possible to filter results by giving prefix and/or delimiter parameters.\n\n        :param path: path from which tree will be built\n        :type path: str\n        :param prefix: if set paths will be added if start with prefix\n        :type prefix: str\n        :param delimiter: if set paths will be added if end with delimiter\n        :type delimiter: str\n        :return: tuple with list of files, dirs and unknown items\n        :rtype: Tuple[List[str], List[str], List[str]]\n        \"\"\"\n        conn = self.get_conn()\n        files, dirs, unknowns = [], [], []  # type: List[str], List[str], List[str]\n\n        def append_matching_path_callback(list_):\n            return lambda item: list_.append(item) if self._is_path_match(item, prefix, delimiter) else None\n\n        conn.walktree(\n            remotepath=path,\n            fcallback=append_matching_path_callback(files),\n            dcallback=append_matching_path_callback(dirs),\n            ucallback=append_matching_path_callback(unknowns),\n            recurse=True,\n        )\n\n        return files, dirs, unknowns\n", "path": "airflow/providers/sftp/hooks/sftp.py"}]}