Added support for legacy SCP protocol.

tbot/machine/linux/copy.py

@@ -19,6 +19,7 @@ def _scp_copy(
     ssh_config: typing.List[str],
     authenticator: auth.Authenticator,
     use_multiplexing: bool,
+    use_legacy_protocol: bool
 ) -> None:
     local_host = local_path.host
 
@@ -39,6 +40,9 @@ def _scp_copy(
             "-o",
             f"ControlPath={multiplexing_dir.at_host(local_host)}/%C",
         ]
+
+    if use_legacy_protocol:
+        scp_command += ["-O"]
 
     if isinstance(authenticator, auth.NoneAuthenticator):
         scp_command += ["-o", "BatchMode=yes"]
@@ -70,7 +74,7 @@ def _scp_copy(
         )
 
 
-def copy(p1: linux.Path[H1], p2: linux.Path[H2]) -> None:
+def copy(p1: linux.Path[H1], p2: linux.Path[H2], use_legacy_protocol: bool=False) -> None:
     """
     Copy a file, possibly from one host to another.
 
@@ -93,6 +97,8 @@ def copy(p1: linux.Path[H1], p2: linux.Path[H2]) -> None:
 
     :param linux.Path p1: Exisiting path to be copied
     :param linux.Path p2: Target where ``p1`` should be copied
+    :param bool use_legacy_protocol: Use the legacy SCP protocol for file transfers 
+        instead of the  SFTP  protocol (use -O option)
 
     .. note::
 
@@ -129,6 +135,7 @@ def copy(p1: linux.Path[H1], p2: linux.Path[H2]) -> None:
             ssh_config=p1.host.ssh_config,
             authenticator=p1.host.authenticator,
             use_multiplexing=p1.host.use_multiplexing,
+            use_legacy_protocol=use_legacy_protocol,
         )
     elif isinstance(p2.host, connector.SSHConnector) and p2.host.host is p1.host:
         # Copy to an SSH machine
@@ -143,6 +150,7 @@ def copy(p1: linux.Path[H1], p2: linux.Path[H2]) -> None:
             ssh_config=p2.host.ssh_config,
             authenticator=p2.host.authenticator,
             use_multiplexing=p2.host.use_multiplexing,
+            use_legacy_protocol=use_legacy_protocol,
         )
     elif isinstance(p1.host, connector.SubprocessConnector) and (
         isinstance(p2.host, connector.ParamikoConnector)
@@ -160,6 +168,7 @@ def copy(p1: linux.Path[H1], p2: linux.Path[H2]) -> None:
             ssh_config=getattr(p2.host, "ssh_config", []),
             authenticator=p2.host.authenticator,
             use_multiplexing=p2.host.use_multiplexing,
+            use_legacy_protocol=use_legacy_protocol,
         )
     elif isinstance(p2.host, connector.SubprocessConnector) and (
         isinstance(p1.host, connector.ParamikoConnector)
@@ -177,6 +186,7 @@ def copy(p1: linux.Path[H1], p2: linux.Path[H2]) -> None:
             ssh_config=getattr(p2.host, "ssh_config", []),
             authenticator=p1.host.authenticator,
             use_multiplexing=p1.host.use_multiplexing,
+            use_legacy_protocol=use_legacy_protocol,
         )
     else:
         raise NotImplementedError(f"Can't copy from {p1.host} to {p2.host}!")

tbot_contrib/utils.py

@@ -189,6 +189,7 @@ def copy_to_dir(
     dest_dir: linux.Path[H2],
     *,
     hashcmp: bool = False,
+    use_legacy_protocol: bool=False,
 ) -> linux.Path[H2]:
     pass
 
@@ -199,6 +200,7 @@ def copy_to_dir(
     dest_dir: linux.Path[H2],
     *,
     hashcmp: bool = False,
+    use_legacy_protocol: bool=False,
 ) -> typing.List[linux.Path[H2]]:
     pass
 
@@ -208,6 +210,7 @@ def copy_to_dir(
     dest_dir: linux.Path[H2],
     *,
     hashcmp: bool = False,
+    use_legacy_protocol: bool=False,
 ) -> typing.Union[linux.Path[H2], typing.List[linux.Path[H2]]]:
     """
     Copy one or more files to a directory
@@ -230,6 +233,9 @@ def copy_to_dir(
         make the function verify checksums of each file before performing the
         copy. This is very useful to skip superfluous copying operations.
 
+    :param bool use_legacy_protocol: Use the legacy SCP protocol for file transfers 
+        instead of the  SFTP  protocol (use -O option)
+
     :returns: If a single ``sources`` path was passed, a single path is
         returned which points to the newly created copy.  If multiple
         ``sources`` were passed (as an iterable), a list of paths for each
@@ -306,9 +312,9 @@ def copy_to_dir(
 
         if hashcmp:
             if not _hashcmp(source, dest):
-                linux.copy(source, dest)
+                linux.copy(source, dest, use_legacy_protocol)
         else:
-            linux.copy(source, dest)
+            linux.copy(source, dest, use_legacy_protocol)
 
     if isinstance(sources, linux.Path):
         return dest_list[0]