operation-bin/gopass.py

#!/usr/bin/env python3
"""Minimal Python interface to the Bernina gopass secrets store.

Shells out to the `gopass` CLI rather than reimplementing GPG decryption
or git-based sync - the CLI already handles the crypto, the git remote,
and gpg-agent passphrase caching correctly, and shelling out is a much
smaller surface to get wrong. See gopass.md (same directory) for how the
underlying store is set up, and its "Limitations & workarounds" section
(store location, multi-host access) - this module doesn't change any of
that, it just gives scripts a way to call into it.

Usage as a library:

    import sys
    sys.path.insert(0, "/sf/bernina/bin")
    from gopass import GopassStore

    store = GopassStore("bernina")
    password = store.get_password("instrument/eiger-web-ui")
    names = store.list()

Usage as a CLI:

    /sf/bernina/bin/gopass.py get instrument/eiger-web-ui
    /sf/bernina/bin/gopass.py get instrument/eiger-web-ui --full
    /sf/bernina/bin/gopass.py list
    /sf/bernina/bin/gopass.py list instrument

Every call that actually decrypts a secret (get/get_password/exists) needs
a usable GPG setup on the machine it runs on: your key imported, and
either a cached gpg-agent passphrase or a terminal gpg-agent can prompt
on (pinentry). It will hang or fail in a context with neither (e.g. a
plain cron job with no agent primed) - see gopass.md's section on
service-account access for how to set that up unattended.
"""
from __future__ import annotations

import argparse
import subprocess
import sys


class GopassError(RuntimeError):
    """Raised when the gopass CLI itself reports failure."""


def _run(args: list[str], input_text: str | None = None, timeout: int = 30) -> subprocess.CompletedProcess:
    try:
        result = subprocess.run(
            ["gopass", *args],
            input=input_text,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except FileNotFoundError as exc:
        raise GopassError("'gopass' not found on PATH") from exc
    except subprocess.TimeoutExpired as exc:
        raise GopassError(
            f"gopass {' '.join(args)} timed out after {timeout}s - likely "
            "waiting on a GPG passphrase prompt with no terminal/agent "
            "available (see this module's docstring)"
        ) from exc
    if result.returncode != 0:
        msg = result.stderr.strip() or f"gopass {' '.join(args)} exited {result.returncode}"
        raise GopassError(msg)
    return result


class GopassStore:
    """A thin wrapper bound to one gopass mount/prefix, e.g. "bernina"."""

    def __init__(self, prefix: str = "bernina", sync: bool = True, timeout: int = 30):
        self.prefix = prefix.strip("/")
        self.sync = sync
        self.timeout = timeout

    def _path(self, name: str) -> str:
        name = name.strip("/")
        if not self.prefix:
            return name
        return f"{self.prefix}/{name}" if name else self.prefix

    def _sync_flags(self) -> list[str]:
        return [] if self.sync else ["--nosync"]

    def get_password(self, name: str) -> str:
        """Return just the password (first line) of a secret, decrypted."""
        result = _run(
            ["show", *self._sync_flags(), "--password", self._path(name)],
            timeout=self.timeout,
        )
        return result.stdout.rstrip("\n")

    def get(self, name: str) -> str:
        """Return the full decrypted secret body (password + any extra lines)."""
        result = _run(
            ["show", *self._sync_flags(), "--noparsing", self._path(name)],
            timeout=self.timeout,
        )
        return result.stdout

    def exists(self, name: str) -> bool:
        """True if the secret exists AND is decryptable with the current key."""
        try:
            _run(
                ["show", *self._sync_flags(), "--password", self._path(name)],
                timeout=self.timeout,
            )
            return True
        except GopassError:
            return False

    def list(self, subpath: str = "") -> list[str]:
        """List secret names under this store/subpath. No decryption needed.

        Returns [] for an empty (but existing) store/subpath rather than
        raising - gopass itself errors with "Entry not found" for that
        case, which isn't really an error from a listing function's point
        of view.
        """
        target = self._path(subpath)
        try:
            result = _run(["list", "--flat", target], timeout=self.timeout)
        except GopassError as exc:
            if "not found" in str(exc).lower():
                return []
            raise
        prefix = f"{target}/" if target else ""
        names = []
        for line in result.stdout.splitlines():
            line = line.strip()
            if not line:
                continue
            names.append(line[len(prefix):] if prefix and line.startswith(prefix) else line)
        return names

    def insert(self, name: str, password: str, extra: str | None = None, force: bool = True) -> None:
        """Insert or overwrite a secret. Needs local write + push access."""
        body = password if extra is None else f"{password}\n{extra}"
        if not body.endswith("\n"):
            body += "\n"
        args = ["insert", "-m"]
        if force:
            args.append("-f")
        args.append(self._path(name))
        _run(args, input_text=body, timeout=self.timeout)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Minimal CLI around the Bernina gopass store (see gopass.md)."
    )
    parser.add_argument("--store", default="bernina", help="gopass mount/prefix (default: bernina)")
    parser.add_argument("--nosync", action="store_true", help="skip the git pull/push gopass normally does")
    sub = parser.add_subparsers(dest="command", required=True)

    p_get = sub.add_parser("get", help="print a secret's password (or full body with --full)")
    p_get.add_argument("name")
    p_get.add_argument("--full", action="store_true", help="print the full decrypted body, not just the password")

    p_list = sub.add_parser("list", help="list secret names (no decryption needed)")
    p_list.add_argument("subpath", nargs="?", default="")

    p_exists = sub.add_parser("exists", help="exit 0 if the secret exists and is decryptable, 1 otherwise")
    p_exists.add_argument("name")

    args = parser.parse_args(argv)
    store = GopassStore(args.store, sync=not args.nosync)

    try:
        if args.command == "get":
            print(store.get(args.name) if args.full else store.get_password(args.name))
        elif args.command == "list":
            for name in store.list(args.subpath):
                print(name)
        elif args.command == "exists":
            return 0 if store.exists(args.name) else 1
    except GopassError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())