Merge branch 'main' into 446-better-error-if-batou-already-running

CHANGES.md

@@ -5,6 +5,8 @@
 ---------------------
 
 - fix Component.require_one raising a configuration error with strict=False
+- Adds command `./batou secrets decrypttostdout` to decrypt a secrets file to stdout.
+  - Useful for integration with git diff using textconv, see documentation for installation instructions.
 - The error message shown when a batou deployment fails to start due to another
   deployment running at the same time, now correctly informs the user that
   another deployment is running instead of showing a traceback. (#446)

doc/source/cli/index.txt

@@ -151,3 +151,29 @@ you want to update the set of public keys fetched from a key server.
       --environments ENVIRONMENTS
                             The environments to update. Update all if not
                             specified.
+
+batou secrets decrypttostdout
+-----------------------------
+
+Decrypts a secret file to stdout.
+It's intended usage is for allowing git to decrypt secrets before displaying a diff to the user.
+You can set up the command withing your project like this:
+
+.. code-block:: console
+
+    git config diff.batou.textconv "./batou secrets decrypttostdout"
+    echo "*.gpg diff=batou" >> .gitattributes
+    echo "*.age diff=batou" >> .gitattributes
+    echo "*.age-diffable diff=batou" >> .gitattributes
+
+This only works if your batou binary is in the root of your git project.
+.. code-block:: console
+
+    usage: batou secrets decrypttostdout [-h] file
+
+    positional arguments:
+      file        The secret file to decrypt, should be contained in an
+                  environment.
+
+    optional arguments:
+      -h, --help  show this help message and exit

src/batou/main.py

@@ -182,6 +182,17 @@ def main(args: Optional[list] = None) -> None:
     )
     p.set_defaults(func=batou.secrets.manage.reencrypt)
 
+    p = sp.add_parser(
+        "decrypttostdout",
+        help="Decrypt a secret file to stdout, useful for git diff.",
+    )
+    p.set_defaults(func=p.print_usage)
+    p.add_argument(
+        "file",
+        help="The secret file to decrypt, should be contained in an environment.",
+    )
+    p.set_defaults(func=batou.secrets.manage.decrypt_to_stdout)
+
     # migrate
     migrate = subparsers.add_parser(
         "migrate",

src/batou/secrets/encryption.py

@@ -18,6 +18,8 @@
 
 
 class EncryptedFile:
+    file_ending = None
+
     def __init__(self, path: "pathlib.Path", writeable: bool = False):
         self.path = path
         self.writeable = writeable
@@ -131,6 +133,8 @@ def _unlock(self):
 
 
 class GPGEncryptedFile(EncryptedFile):
+    file_ending = ".gpg"
+
     def decrypt(self):
         if not self.locked:
             raise RuntimeError("File not locked")
@@ -300,6 +304,8 @@ def get_passphrase(identity: str) -> str:
 
 
 class AGEEncryptedFile(EncryptedFile):
+    file_ending = ".age"
+
     def decrypt(self):
         if not self.locked:
             raise ValueError("File is not locked")
@@ -452,6 +458,8 @@ def age(cls):
 
 
 class DiffableAGEEncryptedFile(EncryptedFile):
+    file_ending = ".age-diffable"
+
     def __init__(self, path: "pathlib.Path", writeable: bool = False):
         super().__init__(path, writeable)
         self._decrypted_content = None
@@ -549,3 +557,21 @@ def _write(
             f.write(str(config))
 
         self.is_new = False
+
+
+all_encrypted_file_types = [
+    NoBackingEncryptedFile,
+    GPGEncryptedFile,
+    AGEEncryptedFile,
+    DiffableAGEEncryptedFile,
+]
+
+
+def get_encrypted_file(
+    path: "pathlib.Path", writeable: bool = False
+) -> EncryptedFile:
+    """Return the appropriate EncryptedFile object for the given path."""
+    for ef in all_encrypted_file_types:
+        if ef.file_ending and path.name.endswith(ef.file_ending):
+            return ef(path, writeable)
+    raise ValueError(f"Unknown encrypted file type for {path}")

src/batou/secrets/manage.py

@@ -1,9 +1,11 @@
+import pathlib
 import sys
 
 from configupdater import ConfigUpdater
 
 from batou import AgeCallError, GPGCallError
 from batou.environment import Environment, UnknownEnvironmentError
+from batou.secrets.encryption import get_encrypted_file
 
 
 def summary():
@@ -82,3 +84,13 @@ def reencrypt(environments, **kw):
             environment.secret_provider.write_config(
                 str(config).encode("utf-8"), force_reencrypt=True
             )
+
+
+def decrypt_to_stdout(file: str):
+    """Decrypt a file and write the content to stdout."""
+    file_path = pathlib.Path(file).absolute()
+    with get_encrypted_file(file_path) as encrypted_file:
+        decrypted = encrypted_file.decrypt()
+    sys.stdout.buffer.write(decrypted)
+    sys.stdout.flush()
+    return 0

src/batou/secrets/tests/test_manage.py

@@ -8,7 +8,15 @@
 
 from batou.environment import UnknownEnvironmentError
 
-from ..manage import add_user, reencrypt, remove_user, summary
+from ..encryption import GPGEncryptedFile
+from ..manage import (
+    add_user,
+    decrypt_to_stdout,
+    reencrypt,
+    remove_user,
+    summary,
+)
+from .test_secrets import cleartext_file, encrypted_file
 
 
 @pytest.mark.parametrize("func", (add_user, remove_user))
@@ -136,3 +144,14 @@ def test_manage__reencrypt__1(tmp_path, monkeypatch, capsys):
         assert old[path] != new[path]
 
     assert set(old) == set(new)
+
+
+def test_manage__decrypt_to_stdout__1(encrypted_file, capsys):
+    """It decrypts a file and writes the content to stdout."""
+    with GPGEncryptedFile(encrypted_file) as secret:
+        with open(cleartext_file) as cleartext:
+            # assert cleartext.read().strip() == secret.cleartext.strip()
+            assert decrypt_to_stdout(str(encrypted_file)) == 0
+            out, err = capsys.readouterr()
+            assert out == cleartext.read()
+            assert err == ""

src/batou/secrets/tests/test_secrets.py

@@ -19,7 +19,7 @@
 
 FIXTURE = pathlib.Path(__file__).parent / "fixture"
 cleartext_file = FIXTURE / "cleartext.cfg"
-FIXTURE_ENCRYPTED_CONFIG = FIXTURE / "encrypted.cfg"
+FIXTURE_ENCRYPTED_CONFIG = FIXTURE / "encrypted.cfg.gpg"
 
 
 @pytest.fixture(scope="function")