feat(secrets): add --hide-secrets to keep values out of the output

alakazam.py

@@ -57,6 +57,7 @@ ALL_CONFIGS = {}
 SETTINGS = {}
 SETTINGS_PATH = ""  # path to the alakazam settings file
 GROUP_PATH = None  # path this run was invoked with, either an instance file or a group directory
+HIDE_SECRETS = False  # keep secret values out of the output, for runs whose logs are archived
 ABRA_DIR = None  # path to the abra data directory
 FETCH_MAX_AGE = 600 # seconds; recipe repos are refetched at most once per hour
 ABRA_RETRIES = 6 # number of attempts per abra command that failed to reach the server
@@ -75,6 +76,8 @@ CONNECTION_ERRORS = (
 )
 # marker in the abra output that indicates a secret which is not stored on the server
 MISSING_SECRET_ERROR = "doesn't exist on server"
+# abra subcommands that take a secret as an argument or return one
+SECRET_COMMANDS = ("insert", "generate")
 # markers in the abra output for a secret run that had nothing left to do, which abra exits 1 on
 NO_SECRETS_GENERATED = "no secrets generated"
 NO_SECRETS_TO_REMOVE = "no secrets to remove"
@@ -584,6 +587,19 @@ def extend_shared_secrets(connection_config: Dict[str, Any]) -> None:
                 target_conf['shared_secrets'] = {source_app: shared_secrets}
 
 
+def is_secret_command(args: Tuple[str, ...]) -> bool:
+    """
+    Checks whether an abra command carries a secret value in its arguments or in its output.
+
+    Args:
+        args (tuple): The arguments passed to abra()
+
+    Returns:
+        bool: True for the commands that insert or generate a secret
+    """
+    return args[:2] == ("app", "secret") and len(args) > 2 and args[2] in SECRET_COMMANDS
+
+
 def is_connection_error(output: str) -> bool:
     """
     Checks whether the output of an abra command indicates that the server could not be reached.
@@ -643,7 +659,11 @@ def abra(*args: str, machine_output: bool = False, ignore_error: bool = False, s
     command = [arg for arg in command if arg]
     if machine_output:
         command.append("-m")
-    logging.debug(f"run command: {' '.join(command)}")
+    # the command line of 'app secret insert' ends in the secret itself, and the output of
+    # 'app secret generate' contains the generated values
+    quiet = HIDE_SECRETS and is_secret_command(args)
+    if not quiet:
+        logging.debug(f"run command: {' '.join(command)}")
     for attempt in range(1, ABRA_RETRIES + 1):
         process = run_streamed(command) if stream else subprocess.run(command, capture_output=True)
         if not process.returncode or attempt == ABRA_RETRIES:
@@ -657,7 +677,7 @@ def abra(*args: str, machine_output: bool = False, ignore_error: bool = False, s
         sleep(delay)
     if process.stderr and ignore_error:
         logging.warning(process.stderr.decode())
-    if process.stdout and not stream:
+    if process.stdout and not stream and not quiet:
         logging.debug(process.stdout.decode())
     if process.returncode and not ignore_error:
         #breakpoint()
@@ -858,7 +878,8 @@ def generate_all_secrets(domain: str) -> None:
             return
         print(f"secrets for {domain} generated")
         for gen_sec in generated_secrets:
-            print(f"\t {gen_sec['name']}: {gen_sec['value']}")
+            value = "[hidden]" if HIDE_SECRETS else gen_sec['value']
+            print(f"\t {gen_sec['name']}: {value}")
 
 
 def resolve_path(path_str: str, base: Optional[Path] = None) -> Path:
@@ -1267,7 +1288,10 @@ def insert_secret(domain: str, secret_name: str, secret: str) -> None:
     """
     # Fix extra quotes around secrets
     secret = unquote_strings(secret)
-    logging.debug(f"Insert secret {secret_name}: {secret} into {domain}")
+    if HIDE_SECRETS:
+        logging.debug(f"Insert secret {secret_name} into {domain}")
+    else:
+        logging.debug(f"Insert secret {secret_name}: {secret} into {domain}")
     abra("app", "secret", "insert", domain, secret_name, "v1", secret)
 
 
@@ -1502,8 +1526,9 @@ def execute_cmds(app_config: Dict[str, Any], commands: Tuple[str] = tuple(), ini
 @click.group(context_settings={"help_option_names": ['-h', '--help']})
 @click.option('-l', '--log', 'loglevel', help='Desired logging level ("debug", "info", "warning", "error", "critical")')
 @click.option('-e', '--exclude', help='Path to a directory that contains a group of instance configurations to be excluded.', multiple=True, type=click.Path(exists=True))
+@click.option('hide_secrets', '--hide-secrets', is_flag=True, help='Keep secret values out of the output, for runs whose logs are kept.')
 @click.argument('group_path', type=click.Path(exists=True))
-def cli(loglevel: str, group_path: str, exclude:Tuple[str]) -> None:
+def cli(loglevel: str, group_path: str, exclude: Tuple[str], hide_secrets: bool) -> None:
     """
     Alakazam is a meta-configuration app-connector and an abra wrapper, designed as a proof-of-concept to simplify the management of environment configuration files across multiple instances.
 
@@ -1514,8 +1539,10 @@ def cli(loglevel: str, group_path: str, exclude:Tuple[str]) -> None:
     global SETTINGS
     global SETTINGS_PATH
     global GROUP_PATH
+    global HIDE_SECRETS
     global ABRA_DIR
     global ROOT_PATH
+    HIDE_SECRETS = hide_secrets
     if loglevel:
         numeric_level = getattr(logging, loglevel.upper(), None)
         if not isinstance(numeric_level, int):

tests/test_hide_secrets.py

@@ -0,0 +1,100 @@
+"""Tests for keeping secret values out of the output of a run whose logs are kept."""
+
+import logging
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+import alakazam
+from alakazam import generate_all_secrets, insert_secret, is_secret_command
+
+VALUE = "hunter2-do-not-log-me"
+
+
+@pytest.fixture(params=[False, True], ids=["visible", "hidden"])
+def hide(request, monkeypatch):
+    monkeypatch.setattr(alakazam, "HIDE_SECRETS", request.param)
+    return request.param
+
+
+class TestIsSecretCommand:
+    @pytest.mark.parametrize("args", [
+        ("app", "secret", "insert", "login.a.org", "db_password", "v1", VALUE),
+        ("app", "secret", "generate", "-a", "login.a.org"),
+    ])
+    def test_commands_carrying_a_value(self, args):
+        assert is_secret_command(args)
+
+    @pytest.mark.parametrize("args", [
+        ("app", "secret", "ls", "login.a.org"),
+        ("app", "secret", "rm", "login.a.org", "db_password"),
+        ("app", "ls"),
+        ("app", "secret"),
+    ])
+    def test_commands_without_a_value(self, args):
+        assert not is_secret_command(args)
+
+
+class TestGeneratedValues:
+    def install(self, monkeypatch, created=False):
+        def abra(*args, **kwargs):
+            if args[:3] == ("app", "secret", "ls"):
+                return [{"name": "db_password", "created on server": str(created).lower()}]
+            if args[:3] == ("app", "secret", "generate"):
+                return [{"name": "db_password", "value": VALUE}]
+            raise AssertionError(f"unexpected: {args}")
+        monkeypatch.setattr(alakazam, "abra", abra)
+
+    def test_the_name_is_always_reported(self, hide, monkeypatch, capsys):
+        self.install(monkeypatch)
+        generate_all_secrets("login.a.org")
+        assert "db_password" in capsys.readouterr().out
+
+    def test_the_value_follows_the_switch(self, hide, monkeypatch, capsys):
+        self.install(monkeypatch)
+        generate_all_secrets("login.a.org")
+        out = capsys.readouterr().out
+        assert (VALUE in out) is not hide
+        assert ("[hidden]" in out) is hide
+
+
+class TestDebugLog:
+    """-l debug is the level a pipeline reaches for when something breaks."""
+
+    def run_abra(self, monkeypatch, args):
+        class Process:
+            returncode = 0
+            stdout = f'[{{"name":"db_password","value":"{VALUE}"}}]'.encode()
+            stderr = b""
+        monkeypatch.setattr(alakazam.subprocess, "run", lambda cmd, capture_output: Process())
+        return alakazam.abra(*args)
+
+    def test_the_command_line_is_not_logged_when_hidden(self, monkeypatch, caplog):
+        monkeypatch.setattr(alakazam, "HIDE_SECRETS", True)
+        with caplog.at_level(logging.DEBUG):
+            self.run_abra(monkeypatch, ("app", "secret", "insert", "login.a.org", "db_password", "v1", VALUE))
+        assert VALUE not in caplog.text
+
+    def test_the_generated_output_is_not_logged_when_hidden(self, monkeypatch, caplog):
+        monkeypatch.setattr(alakazam, "HIDE_SECRETS", True)
+        with caplog.at_level(logging.DEBUG):
+            self.run_abra(monkeypatch, ("app", "secret", "generate", "-a", "login.a.org"))
+        assert VALUE not in caplog.text
+
+    def test_other_commands_are_still_logged(self, monkeypatch, caplog):
+        """Hiding secrets must not turn the debug log off altogether."""
+        monkeypatch.setattr(alakazam, "HIDE_SECRETS", True)
+        with caplog.at_level(logging.DEBUG):
+            self.run_abra(monkeypatch, ("app", "ls"))
+        assert "run command" in caplog.text
+
+    def test_insert_logs_the_name_but_not_the_value(self, monkeypatch, caplog):
+        monkeypatch.setattr(alakazam, "HIDE_SECRETS", True)
+        monkeypatch.setattr(alakazam, "abra", lambda *a, **k: "")
+        with caplog.at_level(logging.DEBUG):
+            insert_secret("login.a.org", "db_password", VALUE)
+        assert "db_password" in caplog.text
+        assert VALUE not in caplog.text