We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Secsh channel 10 open FAILED: open failed: Connect failed
**Please provide information about your setup**
DVC version(i.e. `dvc --version`), Platform and method of installation (pip, homebrew, pkg Mac, exe (Windows), DEB(Linux), RPM(Linux))

version 0.88.0
installed via pip on Linux and Mac.
Syncing to a Ubuntu host via ssh.

Everything seems to be working but I get a cryptic warning message every time I do anything. For example:

dvc push
  0% Querying cache in ssh://bror@zotac.bror.us/media/sda2/dvc/first_day| |0/42 Secsh channel 10 open FAILED: open failed: Connect failed
Secsh channel 10 open FAILED: open failed: Connect failed
Secsh channel 10 open FAILED: open failed: Connect failed
  2% /media/sda2/dvc/first_day/4e/4b31f0c5784a2e185d88a3120cac19| |1/42 [00:02<0Secsh channel 10 open FAILED: open failed: Connect failed
Everything is up to date.                          

This is probably an edge case due to my setup but I'm not sure how to quiet the message or resolve the issue.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `dvc/remote/ssh/connection.py`
Content:
```
1 import errno
2 import logging
3 import os
4 import posixpath
5 import stat
6 from contextlib import suppress
7 
8 from funcy import cached_property
9 
10 try:
11     import paramiko
12 except ImportError:
13     paramiko = None
14 
15 from dvc.utils import tmp_fname
16 from dvc.progress import Tqdm
17 from dvc.exceptions import DvcException
18 from dvc.remote.base import RemoteCmdError
19 
20 
21 logger = logging.getLogger(__name__)
22 
23 
24 def sizeof_fmt(num, suffix="B"):
25     """ Convert number of bytes to human-readable string """
26     for unit in ["", "K", "M", "G", "T", "P", "E", "Z"]:
27         if abs(num) < 1024.0:
28             return "%3.1f%s%s" % (num, unit, suffix)
29         num /= 1024.0
30     return "%.1f%s%s" % (num, "Y", suffix)
31 
32 
33 class SSHConnection:
34     def __init__(self, host, *args, **kwargs):
35         logger.debug(
36             "Establishing ssh connection with '{host}' "
37             "through port '{port}' as user '{username}'".format(
38                 host=host, **kwargs
39             )
40         )
41         self.timeout = kwargs.get("timeout", 1800)
42 
43         self._ssh = paramiko.SSHClient()
44         self._ssh.load_system_host_keys()
45         self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
46 
47         self._ssh.connect(host, *args, **kwargs)
48         self._ssh.get_transport().set_keepalive(10)
49         self._sftp_channels = []
50 
51     @property
52     def sftp(self):
53         if not self._sftp_channels:
54             self._sftp_channels = [self._ssh.open_sftp()]
55         return self._sftp_channels[0]
56 
57     def close(self):
58         for sftp in self._sftp_channels:
59             sftp.close()
60         self._ssh.close()
61 
62     def st_mode(self, path):
63         with suppress(FileNotFoundError):
64             return self.sftp.lstat(path).st_mode
65 
66         return 0
67 
68     def getsize(self, path):
69         with suppress(FileNotFoundError):
70             return self.sftp.lstat(path).st_size
71 
72         return 0
73 
74     def exists(self, path, sftp=None):
75         return bool(self.st_mode(path))
76 
77     def isdir(self, path):
78         return stat.S_ISDIR(self.st_mode(path))
79 
80     def isfile(self, path):
81         return stat.S_ISREG(self.st_mode(path))
82 
83     def islink(self, path):
84         return stat.S_ISLNK(self.st_mode(path))
85 
86     def makedirs(self, path):
87         # Single stat call will say whether this is a dir, a file or a link
88         st_mode = self.st_mode(path)
89 
90         if stat.S_ISDIR(st_mode):
91             return
92 
93         if stat.S_ISREG(st_mode) or stat.S_ISLNK(st_mode):
94             raise DvcException(
95                 "a file with the same name '{}' already exists".format(path)
96             )
97 
98         head, tail = posixpath.split(path)
99 
100         if head:
101             self.makedirs(head)
102 
103         if tail:
104             try:
105                 self.sftp.mkdir(path)
106             except IOError as exc:
107                 # Since paramiko errors are very vague we need to recheck
108                 # whether it's because path already exists or something else
109                 if exc.errno == errno.EACCES or not self.exists(path):
110                     raise DvcException(
111                         "unable to create remote directory '{}'".format(path)
112                     ) from exc
113 
114     def walk(self, directory, topdown=True):
115         # NOTE: original os.walk() implementation [1] with default options was
116         # used as a template.
117         #
118         # [1] https://github.com/python/cpython/blob/master/Lib/os.py
119         try:
120             dir_entries = self.sftp.listdir_attr(directory)
121         except IOError as exc:
122             raise DvcException(
123                 "couldn't get the '{}' remote directory files list".format(
124                     directory
125                 )
126             ) from exc
127 
128         dirs = []
129         nondirs = []
130         for entry in dir_entries:
131             name = entry.filename
132             if stat.S_ISDIR(entry.st_mode):
133                 dirs.append(name)
134             else:
135                 nondirs.append(name)
136 
137         if topdown:
138             yield directory, dirs, nondirs
139 
140         for dname in dirs:
141             newpath = posixpath.join(directory, dname)
142             yield from self.walk(newpath, topdown=topdown)
143 
144         if not topdown:
145             yield directory, dirs, nondirs
146 
147     def walk_files(self, directory):
148         for root, dirs, files in self.walk(directory):
149             for fname in files:
150                 yield posixpath.join(root, fname)
151 
152     def _remove_file(self, path):
153         with suppress(FileNotFoundError):
154             self.sftp.remove(path)
155 
156     def _remove_dir(self, path):
157         for root, dirs, files in self.walk(path, topdown=False):
158             for fname in files:
159                 with suppress(FileNotFoundError):
160                     self._remove_file(posixpath.join(root, fname))
161 
162             for dname in dirs:
163                 with suppress(FileNotFoundError):
164                     self.sftp.rmdir(posixpath.join(root, dname))
165 
166         with suppress(FileNotFoundError):
167             self.sftp.rmdir(path)
168 
169     def remove(self, path):
170         if self.isdir(path):
171             self._remove_dir(path)
172         else:
173             self._remove_file(path)
174 
175     def download(self, src, dest, no_progress_bar=False, progress_title=None):
176         with Tqdm(
177             desc=progress_title or os.path.basename(src),
178             disable=no_progress_bar,
179             bytes=True,
180         ) as pbar:
181             self.sftp.get(src, dest, callback=pbar.update_to)
182 
183     def move(self, src, dst):
184         """Rename src to dst, if it is not possible (in case src and dst are
185         on different filesystems) and actual physical copying of data is
186         happening.
187         """
188         self.makedirs(posixpath.dirname(dst))
189 
190         try:
191             self.sftp.rename(src, dst)
192         except OSError:
193             self.atomic_copy(src, dst)
194             self.remove(src)
195 
196     def atomic_copy(self, src, dst):
197         tmp = tmp_fname(dst)
198 
199         try:
200             self.copy(src, tmp)
201             self.sftp.rename(tmp, dst)
202         finally:
203             self.remove(tmp)
204 
205     def upload(self, src, dest, no_progress_bar=False, progress_title=None):
206         self.makedirs(posixpath.dirname(dest))
207         tmp_file = tmp_fname(dest)
208         if not progress_title:
209             progress_title = posixpath.basename(dest)
210 
211         with Tqdm(
212             desc=progress_title, disable=no_progress_bar, bytes=True
213         ) as pbar:
214             self.sftp.put(src, tmp_file, callback=pbar.update_to)
215 
216         self.sftp.rename(tmp_file, dest)
217 
218     def execute(self, cmd):
219         stdin, stdout, stderr = self._ssh.exec_command(cmd)
220         channel = stdout.channel
221 
222         stdin.close()
223         channel.shutdown_write()
224 
225         stdout_chunks = []
226         stderr_chunks = []
227         while (
228             not channel.closed
229             or channel.recv_ready()
230             or channel.recv_stderr_ready()
231         ):
232             import select
233 
234             got_chunk = False
235             readq, _, _ = select.select([stdout.channel], [], [], self.timeout)
236             for c in readq:
237                 if c.recv_ready():
238                     stdout_chunks.append(stdout.channel.recv(len(c.in_buffer)))
239                     got_chunk = True
240 
241                 if c.recv_stderr_ready():
242                     stderr_len = len(c.in_stderr_buffer)
243                     s = stderr.channel.recv_stderr(stderr_len)
244                     stderr_chunks.append(s)
245                     got_chunk = True
246 
247             if (
248                 not got_chunk
249                 and stdout.channel.exit_status_ready()
250                 and not stderr.channel.recv_stderr_ready()
251                 and not stdout.channel.recv_ready()
252             ):
253                 stdout.channel.shutdown_read()
254                 stdout.channel.close()
255                 break
256 
257         stdout.close()
258         stderr.close()
259 
260         ret = stdout.channel.recv_exit_status()
261         if ret != 0:
262             err = b"".join(stderr_chunks).decode("utf-8")
263             raise RemoteCmdError("ssh", cmd, ret, err)
264 
265         return b"".join(stdout_chunks).decode("utf-8")
266 
267     @cached_property
268     def uname(self):
269         return self.execute("uname").strip()
270 
271     def md5(self, path):
272         """
273         Use different md5 commands depending on the OS:
274 
275          - Darwin's `md5` returns BSD-style checksums by default
276          - Linux's `md5sum` needs the `--tag` flag for a similar output
277 
278          Example:
279               MD5 (foo.txt) = f3d220a856b52aabbf294351e8a24300
280         """
281         if self.uname == "Linux":
282             md5 = self.execute("md5sum " + path).split()[0]
283         elif self.uname == "Darwin":
284             md5 = self.execute("md5 " + path).split()[-1]
285         else:
286             raise DvcException(
287                 "'{}' is not supported as a SSH remote".format(self.uname)
288             )
289 
290         assert len(md5) == 32
291         return md5
292 
293     def copy(self, src, dest):
294         self.execute("cp {} {}".format(src, dest))
295 
296     def open_max_sftp_channels(self):
297         # If there are more than 1 it means we've already opened max amount
298         if len(self._sftp_channels) <= 1:
299             while True:
300                 try:
301                     self._sftp_channels.append(self._ssh.open_sftp())
302                 except paramiko.ssh_exception.ChannelException:
303                     if not self._sftp_channels:
304                         raise
305                     break
306         return self._sftp_channels
307 
308     def open(self, *args, **kwargs):
309         return self.sftp.open(*args, **kwargs)
310 
311     def symlink(self, src, dest):
312         self.sftp.symlink(src, dest)
313 
314     def reflink(self, src, dest):
315         if self.uname == "Linux":
316             return self.execute("cp --reflink {} {}".format(src, dest))
317 
318         if self.uname == "Darwin":
319             return self.execute("cp -c {} {}".format(src, dest))
320 
321         raise DvcException(
322             "'{}' is not supported as a SSH remote".format(self.uname)
323         )
324 
325     def hardlink(self, src, dest):
326         self.execute("ln {} {}".format(src, dest))
327
```
--- 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/dvc/remote/ssh/connection.py b/dvc/remote/ssh/connection.py
--- a/dvc/remote/ssh/connection.py
+++ b/dvc/remote/ssh/connection.py
@@ -41,6 +41,14 @@
         self.timeout = kwargs.get("timeout", 1800)
 
         self._ssh = paramiko.SSHClient()
+
+        # Explicitly disable paramiko logger. Due to how paramiko dynamically
+        # loads loggers, it is not disabled by DVC disable_other_loggers().
+        # See https://github.com/iterative/dvc/issues/3482
+        self._ssh.set_log_channel("dvc.paramiko")
+        logging.getLogger("dvc.paramiko").disabled = True
+        logging.getLogger("dvc.paramiko.sftp").disabled = True
+
         self._ssh.load_system_host_keys()
         self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

{"golden_diff": "diff --git a/dvc/remote/ssh/connection.py b/dvc/remote/ssh/connection.py\n--- a/dvc/remote/ssh/connection.py\n+++ b/dvc/remote/ssh/connection.py\n@@ -41,6 +41,14 @@\n         self.timeout = kwargs.get(\"timeout\", 1800)\n \n         self._ssh = paramiko.SSHClient()\n+\n+        # Explicitly disable paramiko logger. Due to how paramiko dynamically\n+        # loads loggers, it is not disabled by DVC disable_other_loggers().\n+        # See https://github.com/iterative/dvc/issues/3482\n+        self._ssh.set_log_channel(\"dvc.paramiko\")\n+        logging.getLogger(\"dvc.paramiko\").disabled = True\n+        logging.getLogger(\"dvc.paramiko.sftp\").disabled = True\n+\n         self._ssh.load_system_host_keys()\n         self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())\n", "issue": "Secsh channel 10 open FAILED: open failed: Connect failed\n**Please provide information about your setup**\r\nDVC version(i.e. `dvc --version`), Platform and method of installation (pip, homebrew, pkg Mac, exe (Windows), DEB(Linux), RPM(Linux))\r\n\r\nversion 0.88.0\r\ninstalled via pip on Linux and Mac.\r\nSyncing to a Ubuntu host via ssh.\r\n\r\nEverything seems to be working but I get a cryptic warning message every time I do anything. For example:\r\n\r\ndvc push\r\n  0% Querying cache in ssh://bror@zotac.bror.us/media/sda2/dvc/first_day| |0/42 Secsh channel 10 open FAILED: open failed: Connect failed\r\nSecsh channel 10 open FAILED: open failed: Connect failed\r\nSecsh channel 10 open FAILED: open failed: Connect failed\r\n  2% /media/sda2/dvc/first_day/4e/4b31f0c5784a2e185d88a3120cac19| |1/42 [00:02<0Secsh channel 10 open FAILED: open failed: Connect failed\r\nEverything is up to date.                          \r\n\r\nThis is probably an edge case due to my setup but I'm not sure how to quiet the message or resolve the issue.\n", "before_files": [{"content": "import errno\nimport logging\nimport os\nimport posixpath\nimport stat\nfrom contextlib import suppress\n\nfrom funcy import cached_property\n\ntry:\n    import paramiko\nexcept ImportError:\n    paramiko = None\n\nfrom dvc.utils import tmp_fname\nfrom dvc.progress import Tqdm\nfrom dvc.exceptions import DvcException\nfrom dvc.remote.base import RemoteCmdError\n\n\nlogger = logging.getLogger(__name__)\n\n\ndef sizeof_fmt(num, suffix=\"B\"):\n    \"\"\" Convert number of bytes to human-readable string \"\"\"\n    for unit in [\"\", \"K\", \"M\", \"G\", \"T\", \"P\", \"E\", \"Z\"]:\n        if abs(num) < 1024.0:\n            return \"%3.1f%s%s\" % (num, unit, suffix)\n        num /= 1024.0\n    return \"%.1f%s%s\" % (num, \"Y\", suffix)\n\n\nclass SSHConnection:\n    def __init__(self, host, *args, **kwargs):\n        logger.debug(\n            \"Establishing ssh connection with '{host}' \"\n            \"through port '{port}' as user '{username}'\".format(\n                host=host, **kwargs\n            )\n        )\n        self.timeout = kwargs.get(\"timeout\", 1800)\n\n        self._ssh = paramiko.SSHClient()\n        self._ssh.load_system_host_keys()\n        self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())\n\n        self._ssh.connect(host, *args, **kwargs)\n        self._ssh.get_transport().set_keepalive(10)\n        self._sftp_channels = []\n\n    @property\n    def sftp(self):\n        if not self._sftp_channels:\n            self._sftp_channels = [self._ssh.open_sftp()]\n        return self._sftp_channels[0]\n\n    def close(self):\n        for sftp in self._sftp_channels:\n            sftp.close()\n        self._ssh.close()\n\n    def st_mode(self, path):\n        with suppress(FileNotFoundError):\n            return self.sftp.lstat(path).st_mode\n\n        return 0\n\n    def getsize(self, path):\n        with suppress(FileNotFoundError):\n            return self.sftp.lstat(path).st_size\n\n        return 0\n\n    def exists(self, path, sftp=None):\n        return bool(self.st_mode(path))\n\n    def isdir(self, path):\n        return stat.S_ISDIR(self.st_mode(path))\n\n    def isfile(self, path):\n        return stat.S_ISREG(self.st_mode(path))\n\n    def islink(self, path):\n        return stat.S_ISLNK(self.st_mode(path))\n\n    def makedirs(self, path):\n        # Single stat call will say whether this is a dir, a file or a link\n        st_mode = self.st_mode(path)\n\n        if stat.S_ISDIR(st_mode):\n            return\n\n        if stat.S_ISREG(st_mode) or stat.S_ISLNK(st_mode):\n            raise DvcException(\n                \"a file with the same name '{}' already exists\".format(path)\n            )\n\n        head, tail = posixpath.split(path)\n\n        if head:\n            self.makedirs(head)\n\n        if tail:\n            try:\n                self.sftp.mkdir(path)\n            except IOError as exc:\n                # Since paramiko errors are very vague we need to recheck\n                # whether it's because path already exists or something else\n                if exc.errno == errno.EACCES or not self.exists(path):\n                    raise DvcException(\n                        \"unable to create remote directory '{}'\".format(path)\n                    ) from exc\n\n    def walk(self, directory, topdown=True):\n        # NOTE: original os.walk() implementation [1] with default options was\n        # used as a template.\n        #\n        # [1] https://github.com/python/cpython/blob/master/Lib/os.py\n        try:\n            dir_entries = self.sftp.listdir_attr(directory)\n        except IOError as exc:\n            raise DvcException(\n                \"couldn't get the '{}' remote directory files list\".format(\n                    directory\n                )\n            ) from exc\n\n        dirs = []\n        nondirs = []\n        for entry in dir_entries:\n            name = entry.filename\n            if stat.S_ISDIR(entry.st_mode):\n                dirs.append(name)\n            else:\n                nondirs.append(name)\n\n        if topdown:\n            yield directory, dirs, nondirs\n\n        for dname in dirs:\n            newpath = posixpath.join(directory, dname)\n            yield from self.walk(newpath, topdown=topdown)\n\n        if not topdown:\n            yield directory, dirs, nondirs\n\n    def walk_files(self, directory):\n        for root, dirs, files in self.walk(directory):\n            for fname in files:\n                yield posixpath.join(root, fname)\n\n    def _remove_file(self, path):\n        with suppress(FileNotFoundError):\n            self.sftp.remove(path)\n\n    def _remove_dir(self, path):\n        for root, dirs, files in self.walk(path, topdown=False):\n            for fname in files:\n                with suppress(FileNotFoundError):\n                    self._remove_file(posixpath.join(root, fname))\n\n            for dname in dirs:\n                with suppress(FileNotFoundError):\n                    self.sftp.rmdir(posixpath.join(root, dname))\n\n        with suppress(FileNotFoundError):\n            self.sftp.rmdir(path)\n\n    def remove(self, path):\n        if self.isdir(path):\n            self._remove_dir(path)\n        else:\n            self._remove_file(path)\n\n    def download(self, src, dest, no_progress_bar=False, progress_title=None):\n        with Tqdm(\n            desc=progress_title or os.path.basename(src),\n            disable=no_progress_bar,\n            bytes=True,\n        ) as pbar:\n            self.sftp.get(src, dest, callback=pbar.update_to)\n\n    def move(self, src, dst):\n        \"\"\"Rename src to dst, if it is not possible (in case src and dst are\n        on different filesystems) and actual physical copying of data is\n        happening.\n        \"\"\"\n        self.makedirs(posixpath.dirname(dst))\n\n        try:\n            self.sftp.rename(src, dst)\n        except OSError:\n            self.atomic_copy(src, dst)\n            self.remove(src)\n\n    def atomic_copy(self, src, dst):\n        tmp = tmp_fname(dst)\n\n        try:\n            self.copy(src, tmp)\n            self.sftp.rename(tmp, dst)\n        finally:\n            self.remove(tmp)\n\n    def upload(self, src, dest, no_progress_bar=False, progress_title=None):\n        self.makedirs(posixpath.dirname(dest))\n        tmp_file = tmp_fname(dest)\n        if not progress_title:\n            progress_title = posixpath.basename(dest)\n\n        with Tqdm(\n            desc=progress_title, disable=no_progress_bar, bytes=True\n        ) as pbar:\n            self.sftp.put(src, tmp_file, callback=pbar.update_to)\n\n        self.sftp.rename(tmp_file, dest)\n\n    def execute(self, cmd):\n        stdin, stdout, stderr = self._ssh.exec_command(cmd)\n        channel = stdout.channel\n\n        stdin.close()\n        channel.shutdown_write()\n\n        stdout_chunks = []\n        stderr_chunks = []\n        while (\n            not channel.closed\n            or channel.recv_ready()\n            or channel.recv_stderr_ready()\n        ):\n            import select\n\n            got_chunk = False\n            readq, _, _ = select.select([stdout.channel], [], [], self.timeout)\n            for c in readq:\n                if c.recv_ready():\n                    stdout_chunks.append(stdout.channel.recv(len(c.in_buffer)))\n                    got_chunk = True\n\n                if c.recv_stderr_ready():\n                    stderr_len = len(c.in_stderr_buffer)\n                    s = stderr.channel.recv_stderr(stderr_len)\n                    stderr_chunks.append(s)\n                    got_chunk = True\n\n            if (\n                not got_chunk\n                and stdout.channel.exit_status_ready()\n                and not stderr.channel.recv_stderr_ready()\n                and not stdout.channel.recv_ready()\n            ):\n                stdout.channel.shutdown_read()\n                stdout.channel.close()\n                break\n\n        stdout.close()\n        stderr.close()\n\n        ret = stdout.channel.recv_exit_status()\n        if ret != 0:\n            err = b\"\".join(stderr_chunks).decode(\"utf-8\")\n            raise RemoteCmdError(\"ssh\", cmd, ret, err)\n\n        return b\"\".join(stdout_chunks).decode(\"utf-8\")\n\n    @cached_property\n    def uname(self):\n        return self.execute(\"uname\").strip()\n\n    def md5(self, path):\n        \"\"\"\n        Use different md5 commands depending on the OS:\n\n         - Darwin's `md5` returns BSD-style checksums by default\n         - Linux's `md5sum` needs the `--tag` flag for a similar output\n\n         Example:\n              MD5 (foo.txt) = f3d220a856b52aabbf294351e8a24300\n        \"\"\"\n        if self.uname == \"Linux\":\n            md5 = self.execute(\"md5sum \" + path).split()[0]\n        elif self.uname == \"Darwin\":\n            md5 = self.execute(\"md5 \" + path).split()[-1]\n        else:\n            raise DvcException(\n                \"'{}' is not supported as a SSH remote\".format(self.uname)\n            )\n\n        assert len(md5) == 32\n        return md5\n\n    def copy(self, src, dest):\n        self.execute(\"cp {} {}\".format(src, dest))\n\n    def open_max_sftp_channels(self):\n        # If there are more than 1 it means we've already opened max amount\n        if len(self._sftp_channels) <= 1:\n            while True:\n                try:\n                    self._sftp_channels.append(self._ssh.open_sftp())\n                except paramiko.ssh_exception.ChannelException:\n                    if not self._sftp_channels:\n                        raise\n                    break\n        return self._sftp_channels\n\n    def open(self, *args, **kwargs):\n        return self.sftp.open(*args, **kwargs)\n\n    def symlink(self, src, dest):\n        self.sftp.symlink(src, dest)\n\n    def reflink(self, src, dest):\n        if self.uname == \"Linux\":\n            return self.execute(\"cp --reflink {} {}\".format(src, dest))\n\n        if self.uname == \"Darwin\":\n            return self.execute(\"cp -c {} {}\".format(src, dest))\n\n        raise DvcException(\n            \"'{}' is not supported as a SSH remote\".format(self.uname)\n        )\n\n    def hardlink(self, src, dest):\n        self.execute(\"ln {} {}\".format(src, dest))\n", "path": "dvc/remote/ssh/connection.py"}], "after_files": [{"content": "import errno\nimport logging\nimport os\nimport posixpath\nimport stat\nfrom contextlib import suppress\n\nfrom funcy import cached_property\n\ntry:\n    import paramiko\nexcept ImportError:\n    paramiko = None\n\nfrom dvc.utils import tmp_fname\nfrom dvc.progress import Tqdm\nfrom dvc.exceptions import DvcException\nfrom dvc.remote.base import RemoteCmdError\n\n\nlogger = logging.getLogger(__name__)\n\n\ndef sizeof_fmt(num, suffix=\"B\"):\n    \"\"\" Convert number of bytes to human-readable string \"\"\"\n    for unit in [\"\", \"K\", \"M\", \"G\", \"T\", \"P\", \"E\", \"Z\"]:\n        if abs(num) < 1024.0:\n            return \"%3.1f%s%s\" % (num, unit, suffix)\n        num /= 1024.0\n    return \"%.1f%s%s\" % (num, \"Y\", suffix)\n\n\nclass SSHConnection:\n    def __init__(self, host, *args, **kwargs):\n        logger.debug(\n            \"Establishing ssh connection with '{host}' \"\n            \"through port '{port}' as user '{username}'\".format(\n                host=host, **kwargs\n            )\n        )\n        self.timeout = kwargs.get(\"timeout\", 1800)\n\n        self._ssh = paramiko.SSHClient()\n\n        # Explicitly disable paramiko logger. Due to how paramiko dynamically\n        # loads loggers, it is not disabled by DVC disable_other_loggers().\n        # See https://github.com/iterative/dvc/issues/3482\n        self._ssh.set_log_channel(\"dvc.paramiko\")\n        logging.getLogger(\"dvc.paramiko\").disabled = True\n        logging.getLogger(\"dvc.paramiko.sftp\").disabled = True\n\n        self._ssh.load_system_host_keys()\n        self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())\n\n        self._ssh.connect(host, *args, **kwargs)\n        self._ssh.get_transport().set_keepalive(10)\n        self._sftp_channels = []\n\n    @property\n    def sftp(self):\n        if not self._sftp_channels:\n            self._sftp_channels = [self._ssh.open_sftp()]\n        return self._sftp_channels[0]\n\n    def close(self):\n        for sftp in self._sftp_channels:\n            sftp.close()\n        self._ssh.close()\n\n    def st_mode(self, path):\n        with suppress(FileNotFoundError):\n            return self.sftp.lstat(path).st_mode\n\n        return 0\n\n    def getsize(self, path):\n        with suppress(FileNotFoundError):\n            return self.sftp.lstat(path).st_size\n\n        return 0\n\n    def exists(self, path, sftp=None):\n        return bool(self.st_mode(path))\n\n    def isdir(self, path):\n        return stat.S_ISDIR(self.st_mode(path))\n\n    def isfile(self, path):\n        return stat.S_ISREG(self.st_mode(path))\n\n    def islink(self, path):\n        return stat.S_ISLNK(self.st_mode(path))\n\n    def makedirs(self, path):\n        # Single stat call will say whether this is a dir, a file or a link\n        st_mode = self.st_mode(path)\n\n        if stat.S_ISDIR(st_mode):\n            return\n\n        if stat.S_ISREG(st_mode) or stat.S_ISLNK(st_mode):\n            raise DvcException(\n                \"a file with the same name '{}' already exists\".format(path)\n            )\n\n        head, tail = posixpath.split(path)\n\n        if head:\n            self.makedirs(head)\n\n        if tail:\n            try:\n                self.sftp.mkdir(path)\n            except IOError as exc:\n                # Since paramiko errors are very vague we need to recheck\n                # whether it's because path already exists or something else\n                if exc.errno == errno.EACCES or not self.exists(path):\n                    raise DvcException(\n                        \"unable to create remote directory '{}'\".format(path)\n                    ) from exc\n\n    def walk(self, directory, topdown=True):\n        # NOTE: original os.walk() implementation [1] with default options was\n        # used as a template.\n        #\n        # [1] https://github.com/python/cpython/blob/master/Lib/os.py\n        try:\n            dir_entries = self.sftp.listdir_attr(directory)\n        except IOError as exc:\n            raise DvcException(\n                \"couldn't get the '{}' remote directory files list\".format(\n                    directory\n                )\n            ) from exc\n\n        dirs = []\n        nondirs = []\n        for entry in dir_entries:\n            name = entry.filename\n            if stat.S_ISDIR(entry.st_mode):\n                dirs.append(name)\n            else:\n                nondirs.append(name)\n\n        if topdown:\n            yield directory, dirs, nondirs\n\n        for dname in dirs:\n            newpath = posixpath.join(directory, dname)\n            yield from self.walk(newpath, topdown=topdown)\n\n        if not topdown:\n            yield directory, dirs, nondirs\n\n    def walk_files(self, directory):\n        for root, dirs, files in self.walk(directory):\n            for fname in files:\n                yield posixpath.join(root, fname)\n\n    def _remove_file(self, path):\n        with suppress(FileNotFoundError):\n            self.sftp.remove(path)\n\n    def _remove_dir(self, path):\n        for root, dirs, files in self.walk(path, topdown=False):\n            for fname in files:\n                with suppress(FileNotFoundError):\n                    self._remove_file(posixpath.join(root, fname))\n\n            for dname in dirs:\n                with suppress(FileNotFoundError):\n                    self.sftp.rmdir(posixpath.join(root, dname))\n\n        with suppress(FileNotFoundError):\n            self.sftp.rmdir(path)\n\n    def remove(self, path):\n        if self.isdir(path):\n            self._remove_dir(path)\n        else:\n            self._remove_file(path)\n\n    def download(self, src, dest, no_progress_bar=False, progress_title=None):\n        with Tqdm(\n            desc=progress_title or os.path.basename(src),\n            disable=no_progress_bar,\n            bytes=True,\n        ) as pbar:\n            self.sftp.get(src, dest, callback=pbar.update_to)\n\n    def move(self, src, dst):\n        \"\"\"Rename src to dst, if it is not possible (in case src and dst are\n        on different filesystems) and actual physical copying of data is\n        happening.\n        \"\"\"\n        self.makedirs(posixpath.dirname(dst))\n\n        try:\n            self.sftp.rename(src, dst)\n        except OSError:\n            self.atomic_copy(src, dst)\n            self.remove(src)\n\n    def atomic_copy(self, src, dst):\n        tmp = tmp_fname(dst)\n\n        try:\n            self.copy(src, tmp)\n            self.sftp.rename(tmp, dst)\n        finally:\n            self.remove(tmp)\n\n    def upload(self, src, dest, no_progress_bar=False, progress_title=None):\n        self.makedirs(posixpath.dirname(dest))\n        tmp_file = tmp_fname(dest)\n        if not progress_title:\n            progress_title = posixpath.basename(dest)\n\n        with Tqdm(\n            desc=progress_title, disable=no_progress_bar, bytes=True\n        ) as pbar:\n            self.sftp.put(src, tmp_file, callback=pbar.update_to)\n\n        self.sftp.rename(tmp_file, dest)\n\n    def execute(self, cmd):\n        stdin, stdout, stderr = self._ssh.exec_command(cmd)\n        channel = stdout.channel\n\n        stdin.close()\n        channel.shutdown_write()\n\n        stdout_chunks = []\n        stderr_chunks = []\n        while (\n            not channel.closed\n            or channel.recv_ready()\n            or channel.recv_stderr_ready()\n        ):\n            import select\n\n            got_chunk = False\n            readq, _, _ = select.select([stdout.channel], [], [], self.timeout)\n            for c in readq:\n                if c.recv_ready():\n                    stdout_chunks.append(stdout.channel.recv(len(c.in_buffer)))\n                    got_chunk = True\n\n                if c.recv_stderr_ready():\n                    stderr_len = len(c.in_stderr_buffer)\n                    s = stderr.channel.recv_stderr(stderr_len)\n                    stderr_chunks.append(s)\n                    got_chunk = True\n\n            if (\n                not got_chunk\n                and stdout.channel.exit_status_ready()\n                and not stderr.channel.recv_stderr_ready()\n                and not stdout.channel.recv_ready()\n            ):\n                stdout.channel.shutdown_read()\n                stdout.channel.close()\n                break\n\n        stdout.close()\n        stderr.close()\n\n        ret = stdout.channel.recv_exit_status()\n        if ret != 0:\n            err = b\"\".join(stderr_chunks).decode(\"utf-8\")\n            raise RemoteCmdError(\"ssh\", cmd, ret, err)\n\n        return b\"\".join(stdout_chunks).decode(\"utf-8\")\n\n    @cached_property\n    def uname(self):\n        return self.execute(\"uname\").strip()\n\n    def md5(self, path):\n        \"\"\"\n        Use different md5 commands depending on the OS:\n\n         - Darwin's `md5` returns BSD-style checksums by default\n         - Linux's `md5sum` needs the `--tag` flag for a similar output\n\n         Example:\n              MD5 (foo.txt) = f3d220a856b52aabbf294351e8a24300\n        \"\"\"\n        if self.uname == \"Linux\":\n            md5 = self.execute(\"md5sum \" + path).split()[0]\n        elif self.uname == \"Darwin\":\n            md5 = self.execute(\"md5 \" + path).split()[-1]\n        else:\n            raise DvcException(\n                \"'{}' is not supported as a SSH remote\".format(self.uname)\n            )\n\n        assert len(md5) == 32\n        return md5\n\n    def copy(self, src, dest):\n        self.execute(\"cp {} {}\".format(src, dest))\n\n    def open_max_sftp_channels(self):\n        # If there are more than 1 it means we've already opened max amount\n        if len(self._sftp_channels) <= 1:\n            while True:\n                try:\n                    self._sftp_channels.append(self._ssh.open_sftp())\n                except paramiko.ssh_exception.ChannelException:\n                    if not self._sftp_channels:\n                        raise\n                    break\n        return self._sftp_channels\n\n    def open(self, *args, **kwargs):\n        return self.sftp.open(*args, **kwargs)\n\n    def symlink(self, src, dest):\n        self.sftp.symlink(src, dest)\n\n    def reflink(self, src, dest):\n        if self.uname == \"Linux\":\n            return self.execute(\"cp --reflink {} {}\".format(src, dest))\n\n        if self.uname == \"Darwin\":\n            return self.execute(\"cp -c {} {}\".format(src, dest))\n\n        raise DvcException(\n            \"'{}' is not supported as a SSH remote\".format(self.uname)\n        )\n\n    def hardlink(self, src, dest):\n        self.execute(\"ln {} {}\".format(src, dest))\n", "path": "dvc/remote/ssh/connection.py"}]}