Merge pull request #181 from soualid/feature/cp

implemented a cp function

Author: Joachim Jablon

docs/howto/organize.rst

@@ -7,9 +7,15 @@ content of the vault.
 Copy secrets and folders
 ------------------------
 
-This is planned but not implemented yet. Please refer to `#119`__
+.. code:: console
+
+    $ vault-cli set a b=c
 
-.. __: https://github.com/peopledoc/vault-cli/issues/119
+    $ vault-cli cp a d/e
+    Copy 'a' to 'd/e'
+
+``vault-cli cp`` follows the ``safe-write`` parameter (see :ref:`safe-write`) and
+has a ``--force`` flag, like ``vault-cli set``.
 
 Move secrets and folders
 ------------------------

tests/unit/test_cli.py

@@ -588,6 +588,64 @@ def test_mv_mix_secrets_folders(cli_runner, vault_with_token):
     assert result.exit_code != 0
 
 
+def test_cp(cli_runner, vault_with_token):
+    vault_with_token.db = {
+        "a/b": {"value": "c"},
+        "d/e": {"value": "f"},
+        "d/g": {"value": "h"},
+    }
+
+    result = cli_runner.invoke(cli.cli, ["cp", "d", "a"])
+
+    assert result.output.splitlines() == ["Copy 'd/e' to 'a/e'", "Copy 'd/g' to 'a/g'"]
+    assert vault_with_token.db == {
+        "a/b": {"value": "c"},
+        "a/e": {"value": "f"},
+        "a/g": {"value": "h"},
+        "d/e": {"value": "f"},
+        "d/g": {"value": "h"},
+    }
+    assert result.exit_code == 0
+
+
+def test_cp_overwrite_safe(cli_runner, vault_with_token):
+    vault_with_token.db = {"a/b": {"value": "c"}, "d/b": {"value": "f"}}
+
+    vault_with_token.safe_write = True
+
+    result = cli_runner.invoke(cli.cli, ["cp", "d", "a"])
+
+    assert vault_with_token.db == {"a/b": {"value": "c"}, "d/b": {"value": "f"}}
+    assert result.exit_code != 0
+
+
+def test_cp_overwrite_force(cli_runner, vault_with_token):
+    vault_with_token.db = {"a/b": {"value": "c"}, "d/b": {"value": "f"}}
+
+    result = cli_runner.invoke(cli.cli, ["cp", "d", "a", "--force"])
+
+    assert vault_with_token.db == {"a/b": {"value": "f"}, "d/b": {"value": "f"}}
+    assert result.exit_code == 0
+
+
+def test_cp_mix_folders_secrets(cli_runner, vault_with_token):
+    vault_with_token.db = {"a/b": {"value": "c"}, "d": {"value": "e"}}
+
+    result = cli_runner.invoke(cli.cli, ["cp", "d", "a"])
+
+    assert vault_with_token.db == {"a/b": {"value": "c"}, "d": {"value": "e"}}
+    assert result.exit_code != 0
+
+
+def test_cp_mix_secrets_folders(cli_runner, vault_with_token):
+    vault_with_token.db = {"a/b": {"value": "c"}, "d": {"value": "e"}}
+
+    result = cli_runner.invoke(cli.cli, ["cp", "a", "d"])
+
+    assert vault_with_token.db == {"a/b": {"value": "c"}, "d": {"value": "e"}}
+    assert result.exit_code != 0
+
+
 def test_template_from_stdin(cli_runner, vault_with_token):
     vault_with_token.db = {"a/b": {"value": "c"}}
 

tests/unit/test_client_base.py

@@ -392,6 +392,70 @@ def test_vault_client_move_secrets_overwrite_force(vault):
     assert vault.db == {"b": {"value": "c"}}
 
 
+def test_vault_client_copy_secrets(vault):
+
+    vault.db = {"a/b": {"value": "c"}, "a/d": {"value": "e"}}
+
+    vault.copy_secrets("a", "d")
+
+    assert vault.db == {
+        "a/b": {"value": "c"},
+        "a/d": {"value": "e"},
+        "d/b": {"value": "c"},
+        "d/d": {"value": "e"},
+    }
+
+
+def test_vault_client_copy_secrets_generator(vault):
+
+    vault.db = {"a/b": {"value": "c"}, "a/d": {"value": "e"}}
+
+    result = vault.copy_secrets("a", "f", generator=True)
+
+    assert next(result) == ("a/b", "f/b")
+
+    assert vault.db == {"a/b": {"value": "c"}, "a/d": {"value": "e"}}
+
+    assert next(result) == ("a/d", "f/d")
+
+    assert vault.db == {
+        "f/b": {"value": "c"},
+        "a/b": {"value": "c"},
+        "a/d": {"value": "e"},
+    }
+
+    with pytest.raises(StopIteration):
+        next(result)
+
+    assert vault.db == {
+        "a/b": {"value": "c"},
+        "a/d": {"value": "e"},
+        "f/b": {"value": "c"},
+        "f/d": {"value": "e"},
+    }
+
+
+def test_vault_client_copy_secrets_overwrite_safe(vault):
+
+    vault.db = {"a": {"value": "c"}, "b": {"value": "d"}}
+
+    vault.safe_write = True
+
+    with pytest.raises(exceptions.VaultOverwriteSecretError):
+        vault.copy_secrets("a", "b")
+
+    assert vault.db == {"a": {"value": "c"}, "b": {"value": "d"}}
+
+
+def test_vault_client_copy_secrets_overwrite_force(vault):
+
+    vault.db = {"a": {"value": "c"}, "b": {"value": "d"}}
+
+    vault.copy_secrets("a", "b", force=True)
+
+    assert vault.db == {"a": {"value": "c"}, "b": {"value": "c"}}
+
+
 def test_vault_client_base_render_template(vault):
 
     vault.db = {"a/b": {"value": "c"}}

vault_cli/cli.py

@@ -132,7 +132,7 @@ def repr_octal(value: Optional[int]) -> Optional[str]:
     "--safe-write/--unsafe-write",
     default=settings.DEFAULTS.safe_write,
     help="When activated, you can't overwrite a secret without "
-    'passing "--force" (in commands "set", "mv", etc)',
+    'passing "--force" (in commands "set", "mv", "cp", etc)',
 )
 @click.option(
     "--render/--no-render",
@@ -551,6 +551,38 @@ def mv(
         raise click.ClickException(str(exc))
 
 
+@cli.command()
+@click.argument("source", required=True)
+@click.argument("dest", required=True)
+@click.option(
+    "--force/--no-force",
+    "-f",
+    is_flag=True,
+    default=None,
+    help="In case the path already holds a secret, allow overwriting it "
+    "(this is necessary only if --safe-write is set).",
+)
+@click.pass_obj
+@handle_errors()
+def cp(
+    client_obj: client.VaultClientBase, source: str, dest: str, force: Optional[bool]
+) -> None:
+    """
+    Recursively copy secrets from source to destination path.
+    """
+    try:
+        for old_path, new_path in client_obj.copy_secrets(
+            source=source, dest=dest, force=force, generator=True
+        ):
+            click.echo(f"Copy '{old_path}' to '{new_path}'")
+    except exceptions.VaultOverwriteSecretError as exc:
+        raise click.ClickException(
+            f"Secret already exists at {exc.path}. Use -f to force overwriting."
+        )
+    except exceptions.VaultMixSecretAndFolder as exc:
+        raise click.ClickException(str(exc))
+
+
 @cli.command()
 @click.argument(
     "template",

vault_cli/client.py

@@ -427,8 +427,12 @@ def delete_all_secrets(self, *paths: str, generator: bool = False) -> Iterable[s
             return iterator
         return list(iterator)
 
-    def move_secrets_iter(
-        self, source: str, dest: str, force: Optional[bool] = None
+    def copy_secrets_iter(
+        self,
+        source: str,
+        dest: str,
+        force: Optional[bool] = None,
+        delete_source: Optional[bool] = False,
     ) -> Iterable[Tuple[str, str]]:
 
         source_secrets = self.get_secrets(path=source, render=False)
@@ -441,7 +445,13 @@ def move_secrets_iter(
 
             secret_ = cast(types.JSONDict, secret)
             self.set_secret(new_path, secret_, force=force)
-            self.delete_secret(old_path)
+            if delete_source:
+                self.delete_secret(old_path)
+
+    def move_secrets_iter(
+        self, source: str, dest: str, force: Optional[bool] = None
+    ) -> Iterable[Tuple[str, str]]:
+        return self.copy_secrets_iter(source, dest, force, delete_source=True)
 
     def move_secrets(
         self,
@@ -475,6 +485,38 @@ def move_secrets(
             return iterator
         return list(iterator)
 
+    def copy_secrets(
+        self,
+        source: str,
+        dest: str,
+        force: Optional[bool] = None,
+        generator: bool = False,
+    ) -> Iterable[Tuple[str, str]]:
+        """
+        Yield current and new paths, then copy a secret or a folder
+        to a new path
+
+        Parameters
+        ----------
+        source : str
+            Path of the secret to move
+        dest : str
+            New path for the secret
+        force : Optional[bool], optional
+            Allow overwriting exiting secret, if safe_mode is True
+        generator : bool, optional
+            Whether of not to yield before move, by default False
+
+        Returns
+        -------
+        Iterable[Tuple[str, str]]
+            [(Current path, new path)]
+        """
+        iterator = self.copy_secrets_iter(source=source, dest=dest, force=force)
+        if generator:
+            return iterator
+        return list(iterator)
+
     template_prefix = "!template!"
 
     def _render_template_value(self, secret: types.JSONValue) -> types.JSONValue: