vault2vault.py

"""CLI tool for recursively rekeying ansible-vault encrypted secrets"""
import argparse
import getpass
import logging
import re
import shutil
import sys
from pathlib import Path
from typing import Any
from typing import Iterable
from typing import List
from typing import Optional

import ruamel.yaml

try:
    import ansible.constants
    from ansible.parsing.vault import VaultSecret
    from ansible.parsing.vault import VaultLib
    from ansible.parsing.vault import AnsibleVaultError
except ImportError:
    print(
        "FATAL: No supported version of Ansible could be imported under the current python interpreter",
        file=sys.stderr,
    )
    sys.exit(1)


__title__ = "vault2vault"
__summary__ = "Recursively rekey ansible-vault encrypted files and in-line variables"
__version__ = "0.1.3"
__url__ = "https://github.com/enpaul/vault2vault/"
__license__ = "MIT"
__authors__ = ["Ethan Paul <24588726+enpaul@users.noreply.github.com>"]


YAML_FILE_EXTENSIONS = (".yml", ".yaml")

yaml = ruamel.yaml.YAML(typ="rt")

ruamel.yaml.add_constructor(
    "!vault",
    lambda loader, node: node.value,
    constructor=ruamel.yaml.SafeConstructor,
)


def rekey(
    old: VaultLib,
    new: VaultLib,
    content: bytes,
) -> bytes:
    """Rekey vaulted content to use a new vault password

    :param old: ``VaultLib`` object populated with the vault password the content is
                currently encrypted with
    :param new: ``VaultLib`` object populated with the vault password the content will
                be re-encrypted with
    :param content: Content to decrypt using ``old`` and re-encrypt using ``new``
    :returns: The value of ``content`` decrypted using the existing vault password and
              re-encrypted using the new vault password
    """
    return new.encrypt(old.decrypt(content))


# This whole function needs to be rebuilt from the ground up so I don't
# feel bad about disabling this warning
def _process_file(  # pylint: disable=too-many-statements
    path: Path,
    old: VaultLib,
    new: VaultLib,
    interactive: bool,
    backup: bool,
    ignore: bool,
) -> None:
    """Determine whether a filepath includes vaulted data and if so, rekey it

    :param path: Path to the file to check
    :param old: VaultLib object with the current (old) vault password encoded in it
    :param new: VaultLib object with the target (new) vault password encoded in it
    :param interactive: Whether to prompt interactively for confirmation before each
                        rekey operation
    :param backup: Whether to copy the original file to a backup before making any
                   in-place changes
    :param ignore: Whether to ignore any errors that come from failing to decrypt
                   any vaulted data
    """

    logger = logging.getLogger(__name__)

    logger.debug(f"Processing file {path}")

    def _process_yaml_data(  # pylint: disable=too-many-locals
        content: bytes, data: Any, ignore: bool, name: str = ""
    ):
        if isinstance(data, dict):
            for key, value in data.items():
                content = _process_yaml_data(
                    content, value, ignore, name=f"{name}.{key}"
                )
        elif isinstance(data, list):
            for index, item in enumerate(data):
                content = _process_yaml_data(
                    content, item, ignore, name=f"{name}.{index}"
                )
        elif isinstance(data, ruamel.yaml.comments.TaggedScalar) and old.is_encrypted(
            data.value
        ):
            logger.info(f"Identified vaulted content in {path} at {name}")
            confirm = (
                _confirm(f"Rekey vault encrypted variable {name} in file {path}?")
                if interactive
                else True
            )

            if not confirm:
                logger.debug(
                    f"User skipped vault encrypted content in {path} at {name} via interactive mode"
                )
                return content

            try:
                new_data = rekey(old, new, data.value.encode())
            except AnsibleVaultError as err:
                msg = f"Failed to decrypt vault encrypted data in {path} at {name} with provided vault secret"
                if ignore:
                    logger.warning(msg)
                    return content
                raise RuntimeError(msg) from err
            content_decoded = content.decode("utf-8")

            # Ok so this next section is probably the worst possible way to do this, but I did
            # it this way to solve a very specific problem that would absolutely prevent people
            # from using this tool: round trip YAML format preservation. Namely, that it's impossible.
            # Ruamel gets the closest to achieving this: it can do round trip format preservation
            # when the starting state is in _some_ known state (this is better than competitors which
            # require the starting state to be in a _specific_ known state). But given how many
            # ways there are to write YAML- and by extension, how many opinions there are on the
            # "correct" way to write YAML- it is not possible to configure ruamel to account for all of
            # them, even if everyones YAML style was compatible with ruamel's roundtrip formatting (note:
            # they aren't). So there's the problem: to be useful, this tool would need to reformat every
            # YAML file it touched, which means nobody would use it.
            #
            # To avoid the YAML formatting problem, we need a way to replace the target content
            # in the raw text of the file without dumping the parsed YAML. We want to preserve
            # indendation, remove any extra newlines that would be left over, add any necessary
            # newlines without clobbering the following lines, and ideally avoid reimplementing
            # a YAML formatter. The answer to this problem- as the answer to so many stupid problems
            # seems to be- is a regex. If this is too janky for you (I know it is for me) go support
            # the estraven project I'm trying to get off the ground: https://github.com/enpaul/estraven
            #
            # Ok, thanks for sticking with me as I was poetic about this. The solution below...
            # is awful, I can admit that. But it does work, so I'll leave it up to
            # your judgement as to whether it's worthwhile or not. Here's how it works:
            #
            # 1. First we take the first line of the original (unmodified) vaulted content. This line
            #    of text has several important qualities: 1) it exists in the raw text of the file, 2)
            #    it is pseudo-guaranteed to be unique, and 3) it is guaranteed to exist (vaulted content
            #    will be at least one line long, but possibly no more)
            search_data = data.value.split("\n")[1]
            try:
                # 2. Next we use a regex to grab the full line of text from the file that includes the above
                #    string. This is important because the full line of text will include the leading
                #    whitespace, which ruamel helpfully strips out from the parsed data.
                # 3. Next we grab the number of leading spaces on the line using the capture group from the
                #    regex
                padding = len(
                    re.search(rf"\n(\s*){search_data}\n", content_decoded).groups()[0]
                )
            except (TypeError, AttributeError):
                # This is to handle an edgecase where the vaulted content is actually a yaml anchor. For
                # example, if a single vaulted secret needs to be stored under multiple variable names.
                # In that case, the vaulted content iself will only appear once in the file, but the data
                # parsed by ruamel will include it twice. If we fail to get a match on the first line, then
                # we check whether the data is a yaml anchor and, if it is, we skip it.
                if data.anchor.value:
                    logger.debug(
                        f"Content replacement for encrypted content in {path} at {name} was not found, so replacement will be skipped because target is a YAML anchor"
                    )
                    return content
                raise

            # 4. Now with the leading whitespace padding, we add this same number of spaces to each line
            #    of *both* the old vaulted data and the new vaulted data. It's important to do both because
            #    we'll need to do a replacement in a moment so we need to know both what we're replacing
            #    and what we're replacing it with.
            padded_old_data = "\n".join(
                [f"{' ' * padding}{item}" for item in data.value.split("\n") if item]
            )
            padded_new_data = "\n".join(
                [
                    f"{' ' * padding}{item}"
                    for item in new_data.decode("utf-8").split("\n")
                    if item
                ]
            )

            # 5. Finally, we actually replace the content. This needs to have a count=1 so that if the same
            #    encrypted block appears twice in the same file we only replace the first occurance of it,
            #    otherwise the later replacement attempts will fail. We also need to re-encode it back to
            #    bytes because all file operations with vault are done in bytes mode
            content = content_decoded.replace(
                padded_old_data, padded_new_data, 1
            ).encode()
        return content

    with path.open("rb") as infile:
        raw = infile.read()

    # The 'is_encrypted' check doesn't rely on the vault secret in the VaultLib matching the
    # secret the data was encrypted with, it just checks that the data is encrypted with some
    # vault secret. We could use either `old` or `new` for this check, it doesn't actually matter.
    if old.is_encrypted(raw):
        logger.info(f"Identified vault encrypted file: {path}")

        confirm = (
            _confirm(f"Rekey vault encrypted file {path}?") if interactive else True
        )

        if not confirm:
            logger.debug(
                f"User skipped vault encrypted file {path} via interactive mode"
            )
            return

        if backup:
            path.rename(f"{path}.bak")

        try:
            updated = rekey(old, new, raw)
        except AnsibleVaultError:
            msg = f"Failed to decrypt vault encrypted file {path} with provided vault secret"
            if ignore:
                logger.warning(msg)
                return
            raise RuntimeError(msg) from None
    elif path.suffix.lower() in YAML_FILE_EXTENSIONS:
        logger.debug(f"Identified YAML file: {path}")

        confirm = (
            _confirm(f"Search YAML file {path} for vault encrypted variables?")
            if interactive
            else True
        )

        data = yaml.load(raw)

        if not confirm:
            logger.debug(
                f"User skipped processing YAML file {path} via interactive mode"
            )
            return

        if backup:
            shutil.copy(path, f"{path}.bak")

        updated = _process_yaml_data(raw, data, ignore=ignore)
    else:
        logger.debug(f"Skipping non-vault file {path}")
        return

    logger.debug(f"Writing updated file contents to {path}")

    with path.open("wb") as outfile:
        outfile.write(updated)


def _get_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        prog=__title__,
        description=__summary__,
    )

    parser.add_argument(
        "--version", help="Show program version and exit", action="store_true"
    )
    parser.add_argument(
        "--interactive",
        help="Step through files and variables interactively, prompting for confirmation before making each change",
        action="store_true",
    )
    parser.add_argument(
        "-v",
        "--verbose",
        help="Increase verbosity; can be repeated",
        action="count",
        default=0,
    )
    parser.add_argument(
        "-b",
        "--backup",
        help="Write a backup of every file to be modified, suffixed with '.bak'",
        action="store_true",
    )
    parser.add_argument(
        "-i",
        "--vault-id",
        help="Limit rekeying to encrypted secrets with the specified Vault ID",
        type=str,
        default=ansible.constants.DEFAULT_VAULT_IDENTITY,
    )
    parser.add_argument(
        "--ignore-undecryptable",
        help="Ignore any file or variable that is not decryptable with the provided vault secret instead of raising an error",
        action="store_true",
    )
    parser.add_argument(
        "--old-pass-file",
        help="Path to a file with the old vault password to decrypt secrets with",
        type=str,
        dest="old_pass_file",
    )
    parser.add_argument(
        "--new-pass-file",
        help="Path to a file with the new vault password to rekey secrets with",
        type=str,
        dest="new_pass_file",
    )
    parser.add_argument(
        "paths", help="Paths to search for Ansible Vault encrypted content", nargs="*"
    )

    return parser.parse_args()


def _confirm(prompt: str, default: bool = True) -> bool:
    while True:
        confirm = input(f"{prompt} [{'YES/no' if default else 'yes/NO'}]: ")
        if not confirm:
            return default
        if confirm.lower() in ["yes", "y"]:
            return True
        if confirm.lower() in ["no", "n"]:
            return False
        print("Please input one of the specified options", file=sys.stderr)


def _expand_paths(paths: Iterable[Path]) -> List[Path]:
    logger = logging.getLogger(__name__)

    results = []
    for path in paths:
        path = Path(path).resolve()
        if path.is_file():
            logger.debug(f"Including file {path}")
            results.append(path)
        elif path.is_dir():
            logger.debug(f"Identifying files under {path}")
            results += _expand_paths(path.iterdir())
        else:
            logger.debug(f"Discarding path {path}")
    return results


def _load_password(
    fpath: Optional[str], desc: str = "", confirm: bool = True
) -> VaultSecret:
    """Load a password from a file or interactively

    :param fpath: Optional path to the file containing the vault password. If not provided then
                  the password will be prompted for interactively.
    :param desc: Description text to inject into the interactive password prompt. Useful when using
                 this function multiple times to identify different passwords to the user.
    :param confirm: Whether to prompt twice for the input and check that the two inputs match
    :returns: Populated vault secret object with the loaded password
    """

    logger = logging.getLogger(__name__)

    if fpath:
        try:
            with Path(fpath).resolve().open("rb") as infile:
                return VaultSecret(infile.read())
        except (FileNotFoundError, PermissionError) as err:
            raise RuntimeError(
                f"Specified vault password file '{fpath}' does not exist or is unreadable"
            ) from err

    logger.debug("No vault password file provided, prompting for interactive input")

    password_1 = getpass.getpass(
        prompt=f"Enter {desc} Ansible Vault password: ", stream=sys.stderr
    )

    if confirm:
        password_2 = getpass.getpass(
            prompt=f"Confirm (re-enter) {desc} Ansible Vault password: ",
            stream=sys.stderr,
        )

        if password_1 != password_2:
            raise RuntimeError(f"Provided {desc} passwords do not match")

    return VaultSecret(password_1.encode("utf-8"))


def main():
    """Main program entrypoint and CLI interface"""
    args = _get_args()

    logger = logging.getLogger(__name__)

    logging.basicConfig(
        stream=sys.stderr,
        format="%(levelname)s: %(message)s",
        level=max(logging.WARNING - (args.verbose * 10), 0),
    )

    if args.version:
        print(f"{__title__} {__version__}")
        sys.exit(0)

    if not args.paths:
        logger.warning("No paths provided, nothing to do!")
        sys.exit(0)

    try:
        old_pass = _load_password(args.old_pass_file, desc="existing", confirm=False)
        new_pass = _load_password(args.new_pass_file, desc="new", confirm=True)

        in_vault = VaultLib([(args.vault_id, old_pass)])
        out_vault = VaultLib([(args.vault_id, new_pass)])
    except RuntimeError as err:
        logger.error(str(err))
        sys.exit(1)
    except KeyboardInterrupt:
        sys.exit(130)

    logger.info(
        f"Identifying all files under {len(args.paths)} input paths: {', '.join(args.paths)}"
    )
    files = _expand_paths(args.paths)
    logger.info(f"Identified {len(files)} files for processing")

    for filepath in files:
        _process_file(
            filepath,
            in_vault,
            out_vault,
            args.interactive,
            args.backup,
            args.ignore_undecryptable,
        )


if __name__ == "__main__":
    main()