legacy client: work with a borg 1.x "borg serve"

docs/changes.rst

@@ -201,6 +201,9 @@ Fixes:
 - crypto: the AEAD ciphers (AES-OCB, ChaCha20-Poly1305) feed their input to OpenSSL in
   <= 1GB chunks, to overcome the 32bit size limit of the OpenSSL API.
 - repository: raise DoesNotExist for a missing ssh:// repo, #10365
+- --from-borg1 via ssh:// failed with a borg 1.x "borg serve" (e.g. borg
+  transfer): the legacy client used RPC methods borg 1.x does not have and did
+  not convert the bytes a borg 1.x server sends back.
 - compact:
 
   - build the chunk index once, not three times

src/borg/legacy/remote.py

@@ -25,6 +25,8 @@
 from ..fslocking import LockTimeout, NotLocked, NotMyLock, LockFailed
 from ..logger import create_logger, borg_serve_log_queue
 from ..helpers import msgpack
+from ..helpers import safe_encode
+from ..manifest import Manifest, NoManifestError
 from .repository import LegacyRepository
 from ..repository import Repository, StoreObjectNotFound
 from ..version import parse_version, format_version
@@ -38,6 +40,21 @@
 
 MAX_INFLIGHT = 100
 
+# the RPC methods whose results are bytes (or tuples of bytes)
+BYTES_RESULT_METHODS = ("open", "get", "list", "scan", "load_key")
+
+
+def bytes_from_borg1(value):
+    """Convert a result of a borg 1.x server back to bytes (or a tuple of bytes).
+
+    borg 1.x packs bytes into the msgpack "raw" type, which we unpack to str (see helpers.msgpack).
+    """
+    if isinstance(value, str):
+        return safe_encode(value)
+    if isinstance(value, (list, tuple)):
+        return tuple(bytes_from_borg1(v) for v in value)
+    return value
+
 
 class ConnectionClosed(Error):
     """Connection closed by remote host."""
@@ -224,6 +241,8 @@ def __init__(self, location, create=False, exclusive=False, lock_wait=None, lock
         self.shutdown_time = None
         self.unpacker = get_limited_unpacker("client")
         self.server_version = None  # we update this after server sends its version
+        # borg 1.x servers do not have the info, close and get_manifest methods and they send bytes as str.
+        self.borg1_server = False
         self.p = None
         self._args = args
         if self.location.proto == "ssh":
@@ -273,6 +292,7 @@ def __init__(self, location, create=False, exclusive=False, lock_wait=None, lock
                 self.server_version = version["server_version"]
             else:
                 raise Exception("Server insisted on using unsupported protocol version %s" % version)
+            self.borg1_server = self.server_version < parse_version("2.0.0a1")
 
             self.id = self.open(
                 path=self.location.path,
@@ -282,8 +302,11 @@ def __init__(self, location, create=False, exclusive=False, lock_wait=None, lock
                 exclusive=exclusive,
                 v1_legacy=True,  # make remote use LegacyRepository
             )
-            info = self.info()
-            self.version = info["version"]
+            if self.borg1_server:
+                self.version = 1  # borg 1.x servers only serve v1 repositories.
+            else:
+                info = self.info()
+                self.version = info["version"]
 
         except Exception:
             self.close()
@@ -458,7 +481,10 @@ def handle_error(unpacked):
                         unpacked = self.responses.pop(waiting_for[0])
                         waiting_for.pop(0)
                         handle_error(unpacked)
-                        yield unpacked[RESULT]
+                        result = unpacked[RESULT]
+                        if self.borg1_server and cmd in BYTES_RESULT_METHODS:
+                            result = bytes_from_borg1(result)
+                        yield result
                         if not waiting_for and not calls:
                             return
                     except KeyError:
@@ -614,7 +640,9 @@ def break_lock(self):
 
     def close(self):
         if self.p:
-            self.call("close", {}, wait=True)
+            if not self.borg1_server:
+                # borg 1.x servers close the repository when stdin gets closed, they have no close method.
+                self.call("close", {}, wait=True)
             self.p.stdin.close()
             self.p.stdout.close()
             self.p.wait()
@@ -624,9 +652,14 @@ def async_response(self, wait=True):
         for resp in self.call_many("async_responses", calls=[], wait=True, async_wait=wait):
             return resp
 
-    @api(since=parse_version("2.0.0b8"))
     def get_manifest(self):
-        """actual remoting is done via self.call in the @api decorator"""
+        if not self.borg1_server:
+            return self.call("get_manifest", {})
+        # borg 1.x servers have no get_manifest method, so get the manifest object like LegacyRepository does.
+        try:
+            return self.get(Manifest.MANIFEST_ID)
+        except LegacyRepository.ObjectNotFound:
+            raise NoManifestError
 
     @api(since=parse_version("2.0.0b8"))
     def put_manifest(self, data):

src/borg/testsuite/legacyrepository_test.py

@@ -12,7 +12,8 @@
 from ..helpers import msgpack
 from ..fslocking import Lock, LockFailed
 from ..platformflags import is_win32
-from ..legacy.remote import LegacyRemoteRepository, InvalidRPCMethod
+from ..legacy.remote import LegacyRemoteRepository, InvalidRPCMethod, bytes_from_borg1
+from ..manifest import Manifest
 from ..legacy.repository import LegacyRepository, LoggedIO
 from ..legacy.repository import MAGIC, MAX_DATA_SIZE, TAG_DELETE, TAG_PUT, TAG_COMMIT
 from ..compress import CNONE
@@ -987,3 +988,66 @@ def test_remote_borg_cmd(remote_repository, monkeypatch):
         assert remote_repository.borg_cmd(args, testing=False) == ["borg", "serve", "--info"]
         monkeypatch.setenv("BORG_RSH", "ssh -i foo")
         assert remote_repository.ssh_cmd(Location("ssh://example.com/foo")) == ["ssh", "-i", "foo", "example.com"]
+
+
+# A borg 1.x "borg serve", emulated with the legacy server: it has no info, close and get_manifest methods,
+# it packs bytes into the msgpack "raw" type and it closes the repository when the client closes stdin.
+BORG1_SERVE = """
+import msgpack
+from borg.logger import setup_logging
+from borg.legacy import remote
+
+
+class Borg1Msgpack:
+    @staticmethod
+    def packb(o):
+        return msgpack.packb(o, use_bin_type=False)
+
+
+remote.msgpack = Borg1Msgpack
+remote.BORG_VERSION = (1, 4, 5)
+remote.RepositoryServer._legacy_rpc_methods = tuple(
+    m for m in remote.RepositoryServer._legacy_rpc_methods if m not in ("info", "close", "get_manifest")
+)
+setup_logging(is_serve=True)
+server = remote.RepositoryServer(restrict_to_paths=None, restrict_to_repositories=None)
+try:
+    server.serve()
+except AssertionError:
+    # the legacy server expects a close call before stdin gets closed, borg 1.x did not.
+    if server.repository is None:
+        raise
+    server.repository.close()
+"""
+
+
+@pytest.mark.skipif(is_win32, reason="Remote repository does not yet work on Windows.")
+def test_remote_borg1_server(tmp_path, monkeypatch):
+    path = os.fspath(tmp_path / "repository")
+    data = fchunk(b"\xff\xfe not utf-8")
+    with LegacyRepository(path, exclusive=True, create=True) as repository:
+        repository.put(Manifest.MANIFEST_ID, fchunk(b"manifest"))
+        repository.put(H(1), data)
+        repository.commit(compact=False)
+        repository_id = repository.id
+    server = tmp_path / "borg1_serve.py"
+    server.write_text(BORG1_SERVE)
+    monkeypatch.setattr(LegacyRemoteRepository, "borg_cmd", lambda self, args, testing: [sys.executable, str(server)])
+    with LegacyRemoteRepository(Location("ssh://__testsuite__/" + path), exclusive=True) as remote_repository:
+        assert remote_repository.borg1_server
+        assert remote_repository.version == 1
+        assert remote_repository.id == repository_id
+        assert remote_repository.get_manifest() == fchunk(b"manifest")
+        assert remote_repository.get(H(1)) == data
+        assert set(remote_repository.list()) == {Manifest.MANIFEST_ID, H(1)}
+    # the server closed the repository (and released its lock) when the client closed the connection.
+    with LegacyRepository(path, exclusive=True) as repository:
+        assert repository.get(H(1)) == data
+
+
+def test_bytes_from_borg1():
+    # borg 1.x packed bytes into the msgpack "raw" type, which the client unpacks to str.
+    value = (b"\x00\xff\xfe not utf-8", H(1))
+    unpacked = msgpack.unpackb(msgpack.packb(value, use_bin_type=False))
+    assert all(isinstance(v, str) for v in unpacked)
+    assert bytes_from_borg1(unpacked) == value