vaultexec.py

#!/usr/bin/env python3
"""vaultexec: resolve secret:// references at spawn time, then exec the real command.

Env var values and command arguments may contain secret://SERVICE/ACCOUNT
(keyring:// is an alias) references. Each resolves against the first backend
that knows it; the real command is then exec'd so plaintext exists only in the
child process environment — never in config files, never in a wrapper process.
"""
import getpass
import os
import re
import shutil
import subprocess
import sys

__version__ = "0.2.1"

REF = re.compile(r"(?:secret|keyring)://([A-Za-z0-9._-]+)/([A-Za-z0-9._@-]+)")

USAGE = """\
usage: vaultexec [--] CMD [ARGS...]     resolve refs in env + args, then exec CMD
       vaultexec get REF                resolve one reference and print it
       vaultexec set SERVICE ACCOUNT    store a secret (stdin or hidden prompt)
       vaultexec --version

References look like secret://SERVICE/ACCOUNT; see README for backends."""


def _run(cmd, input=None):
    return subprocess.run(cmd, capture_output=True, text=True, input=input)


def keychain_get(service, account):
    p = _run(["security", "find-generic-password", "-s", service, "-a", account, "-w"])
    return p.stdout.rstrip("\n") if p.returncode == 0 else None


def secretservice_get(service, account):
    if not shutil.which("secret-tool"):
        return None
    p = _run(["secret-tool", "lookup", "service", service, "account", account])
    return p.stdout.rstrip("\n") if p.returncode == 0 and p.stdout else None


def systemdcreds_get(service, account):
    path = "/etc/credstore.encrypted/%s.%s.cred" % (service, account)
    if not shutil.which("systemd-creds") or not os.path.exists(path):
        return None
    p = _run(["systemd-creds", "decrypt", path, "-"])
    return p.stdout.rstrip("\n") if p.returncode == 0 else None


def age_get(service, account):
    store = os.environ.get("VAULTEXEC_AGE_FILE",
                           os.path.expanduser("~/.config/vaultexec/secrets.age"))
    ident = os.environ.get("VAULTEXEC_AGE_IDENTITY",
                           os.path.expanduser("~/.config/vaultexec/identity.txt"))
    if not shutil.which("age") or not os.path.exists(store):
        return None
    p = _run(["age", "-d", "-i", ident, store])
    if p.returncode != 0:
        return None
    for line in p.stdout.splitlines():
        key, sep, value = line.partition("=")
        if sep and key.strip() == "%s/%s" % (service, account):
            return value
    return None


def file_get(service, account):
    name = "%s.%s" % (service, account)
    for d in (os.environ.get("CREDENTIALS_DIRECTORY"), "/run/secrets"):
        if not d:
            continue
        path = os.path.join(d, name)
        if os.path.isfile(path):
            with open(path) as f:
                return f.read().rstrip("\n")
    return None


BACKENDS = {
    "keychain": keychain_get,
    "file": file_get,
    "secret-service": secretservice_get,
    "systemd-creds": systemdcreds_get,
    "age": age_get,
}


def backend_chain():
    forced = os.environ.get("VAULTEXEC_BACKEND")
    if forced:
        names = [n.strip() for n in forced.split(",") if n.strip()]
        unknown = [n for n in names if n not in BACKENDS]
        if unknown:
            raise LookupError("unknown backend in VAULTEXEC_BACKEND: %s (known: %s)"
                              % (", ".join(unknown), ", ".join(BACKENDS)))
        return [(n, BACKENDS[n]) for n in names]
    if sys.platform == "darwin":
        return [("keychain", keychain_get)]
    return [(n, BACKENDS[n]) for n in ("file", "secret-service", "systemd-creds", "age")]


def resolve(text, chain, cache):
    def lookup(match):
        key = (match.group(1), match.group(2))
        if key not in cache:
            for _, backend in chain:
                value = backend(*key)
                if value is not None:
                    cache[key] = value
                    break
            else:
                raise LookupError("could not resolve %s (tried: %s)"
                                  % (match.group(0), ", ".join(n for n, _ in chain)))
        return cache[key]

    return REF.sub(lookup, text)


def cmd_set(service, account):
    if not (re.fullmatch(r"[A-Za-z0-9._-]+", service)
            and re.fullmatch(r"[A-Za-z0-9._@-]+", account)):
        raise LookupError("service/account may only contain letters, digits, "
                          ". _ - (account may also contain @)")
    if sys.stdin.isatty():
        value = getpass.getpass("Secret value for %s/%s: " % (service, account))
    else:
        value = sys.stdin.read().rstrip("\n")
    if not value:
        raise LookupError("empty secret value")
    if sys.platform == "darwin":
        # secret goes to `security -i` over stdin so it never appears in any argv
        if "\n" in value:
            raise LookupError("multi-line values not supported on the keychain backend")
        quoted = '"%s"' % value.replace("\\", "\\\\").replace('"', '\\"')
        p = _run(["security", "-i"],
                 input="add-generic-password -U -s %s -a %s -w %s\n"
                       % (service, account, quoted))
    elif shutil.which("secret-tool"):
        p = _run(["secret-tool", "store",
                  "--label", "vaultexec: %s/%s" % (service, account),
                  "service", service, "account", account], input=value)
    else:
        raise LookupError("no writable backend here; "
                          "see README for systemd-creds and age setup")
    if p.returncode != 0:
        raise LookupError("store failed: %s" % (p.stderr or "").strip())
    print("stored secret://%s/%s" % (service, account), file=sys.stderr)


def main(argv=None):
    argv = sys.argv[1:] if argv is None else list(argv)
    if not argv or argv[0] in ("-h", "--help"):
        print(USAGE, file=sys.stderr)
        sys.exit(2 if not argv else 0)
    if argv[0] == "--version":
        print("vaultexec %s" % __version__)
        sys.exit(0)
    try:
        if argv[0] == "get":
            if len(argv) != 2 or not REF.search(argv[1]):
                print(USAGE, file=sys.stderr)
                sys.exit(2)
            print(resolve(argv[1], backend_chain(), {}))
            sys.exit(0)
        if argv[0] == "set":
            if len(argv) != 3:
                print(USAGE, file=sys.stderr)
                sys.exit(2)
            cmd_set(argv[1], argv[2])
            sys.exit(0)
        if argv[0] == "--":
            argv = argv[1:]
        if not argv:
            print(USAGE, file=sys.stderr)
            sys.exit(2)
        chain, cache = backend_chain(), {}
        env = {k: resolve(v, chain, cache) for k, v in os.environ.items()}
        args = [resolve(a, chain, cache) for a in argv]
    except LookupError as e:
        print("vaultexec: %s" % e, file=sys.stderr)
        sys.exit(2)
    try:
        os.execvpe(args[0], args, env)
    except FileNotFoundError:
        print("vaultexec: command not found: %s" % args[0], file=sys.stderr)
        sys.exit(127)


if __name__ == "__main__":
    main()