better support other repo by misc. passphrase env vars, fixes #8457

- borg repo-create and borg transfer not only support --repo / --other-repo options,
  but also already supported related BORG_REPO and BORG_OTHER_REPO env vars.
- similar to that, the passphrases now come from BORG_[OTHER_]PASSPHRASE, BORG_[OTHER_]PASSCOMMAND or BORG_[OTHER_]PASSPHRASE_FD.
- borg repo-create --repo B --other-repo A does not silently copy the passphrase of key A
  to key B anymore, but either asks for the passphrase or reads it from env vars.

docs/usage/general/environment.rst.inc

@@ -9,12 +9,12 @@ General:
         Use this so you do not need to type ``--repo /path/to/my/repo`` all the time.
     BORG_OTHER_REPO
         Similar to BORG_REPO, but gives the default for ``--other-repo``.
-    BORG_PASSPHRASE
+    BORG_PASSPHRASE (and BORG_OTHER_PASSPHRASE)
         When set, use the value to answer the passphrase question for encrypted repositories.
         It is used when a passphrase is needed to access an encrypted repo as well as when a new
         passphrase should be initially set when initializing an encrypted repo.
         See also BORG_NEW_PASSPHRASE.
-    BORG_PASSCOMMAND
+    BORG_PASSCOMMAND (and BORG_OTHER_PASSCOMMAND)
         When set, use the standard output of the command (trailing newlines are stripped) to answer the
         passphrase question for encrypted repositories.
         It is used when a passphrase is needed to access an encrypted repo as well as when a new
@@ -22,7 +22,7 @@ General:
         is executed without a shell. So variables, like ``$HOME`` will work, but ``~`` won't.
         If BORG_PASSPHRASE is also set, it takes precedence.
         See also BORG_NEW_PASSPHRASE.
-    BORG_PASSPHRASE_FD
+    BORG_PASSPHRASE_FD (and BORG_OTHER_PASSPHRASE_FD)
         When set, specifies a file descriptor to read a passphrase
         from. Programs starting borg may choose to open an anonymous pipe
         and use it to pass a passphrase. This is safer than passing via

src/borg/archiver/_common.py

@@ -123,7 +123,7 @@ def with_repository(
                         f"You can use 'borg transfer' to copy archives from old to new repos."
                     )
                 if manifest or cache:
-                    manifest_ = Manifest.load(repository, compatibility)
+                    manifest_ = Manifest.load(repository, compatibility, other=False)
                     kwargs["manifest"] = manifest_
                     if "compression" in args:
                         manifest_.repo_objs.compressor = args.compression.compressor
@@ -192,7 +192,7 @@ def with_other_repository(manifest=False, cache=False, compatibility=None):
                 kwargs["other_repository"] = repository
                 if manifest or cache:
                     manifest_ = Manifest.load(
-                        repository, compatibility, ro_cls=RepoObj if repository.version > 1 else RepoObj1
+                        repository, compatibility, other=True, ro_cls=RepoObj if repository.version > 1 else RepoObj1
                     )
                     assert_secure(repository, manifest_)
                     if manifest:

src/borg/crypto/key.py

@@ -101,10 +101,10 @@ def identify_key(manifest_data):
         raise UnsupportedPayloadError(key_type)
 
 
-def key_factory(repository, manifest_chunk, *, ro_cls=RepoObj):
+def key_factory(repository, manifest_chunk, *, other=False, ro_cls=RepoObj):
     manifest_data = ro_cls.extract_crypted_data(manifest_chunk)
     assert manifest_data, "manifest data must not be zero bytes long"
-    return identify_key(manifest_data).detect(repository, manifest_data)
+    return identify_key(manifest_data).detect(repository, manifest_data, other=other)
 
 
 def uses_same_chunker_secret(other_key, key):
@@ -236,7 +236,7 @@ class PlaintextKey(KeyBase):
         return cls(repository)
 
     @classmethod
-    def detect(cls, repository, manifest_data):
+    def detect(cls, repository, manifest_data, *, other=False):
         return cls(repository)
 
     def id_hash(self, data):
@@ -359,11 +359,11 @@ class FlexiKey:
     STORAGE: ClassVar[str] = KeyBlobStorage.NO_STORAGE  # override in subclass
 
     @classmethod
-    def detect(cls, repository, manifest_data):
+    def detect(cls, repository, manifest_data, *, other=False):
         key = cls(repository)
         target = key.find_key()
         prompt = "Enter passphrase for key %s: " % target
-        passphrase = Passphrase.env_passphrase()
+        passphrase = Passphrase.env_passphrase(other=other)
         if passphrase is None:
             passphrase = Passphrase()
             if not key.load(target, passphrase):
@@ -541,10 +541,9 @@ class FlexiKey:
                 # borg transfer re-encrypts all data anyway, thus we can default to a new, random AE key
                 crypt_key = os.urandom(64)
             key.init_from_given_data(crypt_key=crypt_key, id_key=other_key.id_key, chunk_seed=other_key.chunk_seed)
-            passphrase = other_key._passphrase
         else:
             key.init_from_random_data()
-            passphrase = Passphrase.new(allow_empty=True)
+        passphrase = Passphrase.new(allow_empty=True)
         key.init_ciphers()
         target = key.get_new_target(args)
         key.save(target, passphrase, create=True, algorithm=KEY_ALGORITHMS["argon2"])

src/borg/helpers/passphrase.py

@@ -47,20 +47,22 @@ class Passphrase(str):
             return cls(passphrase)
 
     @classmethod
-    def env_passphrase(cls, default=None):
-        passphrase = cls._env_passphrase("BORG_PASSPHRASE", default)
+    def env_passphrase(cls, default=None, other=False):
+        env_var = "BORG_OTHER_PASSPHRASE" if other else "BORG_PASSPHRASE"
+        passphrase = cls._env_passphrase(env_var, default)
         if passphrase is not None:
             return passphrase
-        passphrase = cls.env_passcommand()
+        passphrase = cls.env_passcommand(other=other)
         if passphrase is not None:
             return passphrase
-        passphrase = cls.fd_passphrase()
+        passphrase = cls.fd_passphrase(other=other)
         if passphrase is not None:
             return passphrase
 
     @classmethod
-    def env_passcommand(cls, default=None):
-        passcommand = os.environ.get("BORG_PASSCOMMAND", None)
+    def env_passcommand(cls, default=None, other=False):
+        env_var = "BORG_OTHER_PASSCOMMAND" if other else "BORG_PASSCOMMAND"
+        passcommand = os.environ.get(env_var, None)
         if passcommand is not None:
             # passcommand is a system command (not inside pyinstaller env)
             env = prepare_subprocess_env(system=True)
@@ -71,9 +73,10 @@ class Passphrase(str):
             return cls(passphrase.rstrip("\n"))
 
     @classmethod
-    def fd_passphrase(cls):
+    def fd_passphrase(cls, other=False):
+        env_var = "BORG_OTHER_PASSPHRASE_FD" if other else "BORG_PASSPHRASE_FD"
         try:
-            fd = int(os.environ.get("BORG_PASSPHRASE_FD"))
+            fd = int(os.environ.get(env_var))
         except (ValueError, TypeError):
             return None
         with os.fdopen(fd, mode="r") as f:

src/borg/manifest.py

@@ -491,13 +491,13 @@ class Manifest:
         return parse_timestamp(self.timestamp)
 
     @classmethod
-    def load(cls, repository, operations, key=None, *, ro_cls=RepoObj):
+    def load(cls, repository, operations, key=None, *, other=False, ro_cls=RepoObj):
         from .item import ManifestItem
         from .crypto.key import key_factory
 
         cdata = repository.get_manifest()
         if not key:
-            key = key_factory(repository, cdata, ro_cls=ro_cls)
+            key = key_factory(repository, cdata, other=other, ro_cls=ro_cls)
         manifest = cls(key, repository, ro_cls=ro_cls)
         _, data = manifest.repo_objs.parse(cls.MANIFEST_ID, cdata, ro_type=ROBJ_MANIFEST)
         manifest_dict = key.unpack_manifest(data)

src/borg/testsuite/archiver/transfer_cmd_test.py

@@ -13,7 +13,7 @@ from . import cmd, create_test_files, RK_ENCRYPTION, open_archive, generate_arch
 pytest_generate_tests = lambda metafunc: generate_archiver_tests(metafunc, kinds="local,remote,binary")  # NOQA
 
 
-def test_transfer(archivers, request):
+def test_transfer(archivers, request, monkeypatch):
     archiver = request.getfixturevalue(archivers)
     original_location, input_path = archiver.repository_location, archiver.input_path
 
@@ -29,6 +29,7 @@ def test_transfer(archivers, request):
     create_test_files(input_path)
     archiver.repository_location = original_location + "1"
 
+    monkeypatch.setenv("BORG_PASSPHRASE", "pw1")
     cmd(archiver, "repo-create", RK_ENCRYPTION)
     cmd(archiver, "create", "arch1", "input")
     cmd(archiver, "create", "arch2", "input")
@@ -36,6 +37,8 @@ def test_transfer(archivers, request):
 
     archiver.repository_location = original_location + "2"
     other_repo1 = f"--other-repo={original_location}1"
+    monkeypatch.setenv("BORG_PASSPHRASE", "pw2")
+    monkeypatch.setenv("BORG_OTHER_PASSPHRASE", "pw1")
     cmd(archiver, "repo-create", RK_ENCRYPTION, other_repo1)
     cmd(archiver, "transfer", other_repo1, "--dry-run")
     cmd(archiver, "transfer", other_repo1)
@@ -43,7 +46,7 @@ def test_transfer(archivers, request):
     check_repo()
 
 
-def test_transfer_upgrade(archivers, request):
+def test_transfer_upgrade(archivers, request, monkeypatch):
     archiver = request.getfixturevalue(archivers)
     if archiver.get_kind() in ["remote", "binary"]:
         pytest.skip("only works locally")
@@ -72,7 +75,8 @@ def test_transfer_upgrade(archivers, request):
     other_repo1 = f"--other-repo={original_location}1"
     archiver.repository_location = original_location + "2"
 
-    assert os.environ.get("BORG_PASSPHRASE") == "waytooeasyonlyfortests"
+    monkeypatch.setenv("BORG_PASSPHRASE", "pw2")
+    monkeypatch.setenv("BORG_OTHER_PASSPHRASE", "waytooeasyonlyfortests")
     os.environ["BORG_TESTONLY_WEAKEN_KDF"] = "0"  # must use the strong kdf here or it can't decrypt the key
 
     cmd(archiver, "repo-create", RK_ENCRYPTION, other_repo1, "--from-borg1")