prole/knoe/core/onepassword.py

"""1Password CLI integration for knoe secret management.

All secrets are stored in the 'knoey' vault so they are isolated from the
user's personal 1Password vaults.  The 'administrator' item holds the
database master password.
"""

from __future__ import annotations

import json
import secrets
import shutil
import string
import subprocess

_VAULT = "knoey"
_ADMIN_ITEM = "administrator"


def op_available() -> bool:
    return shutil.which("op") is not None


def _op(*args: str, check: bool = True, timeout: int = 15) -> subprocess.CompletedProcess:
    if not op_available():
        raise RuntimeError(
            "1Password CLI (op) not found. Install: brew install 1password-cli"
        )
    try:
        return subprocess.run(
            ["op", *args],
            capture_output=True,
            text=True,
            check=check,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired:
        return subprocess.CompletedProcess(
            args=["op", *args], returncode=1, stdout="", stderr="timeout"
        )


def ensure_op_signed_in() -> None:
    """Ensure the op CLI has an active session; trigger sign-in if not.

    When running non-interactively (no TTY on stdin) and no session exists,
    skip rather than hang or crash — the installer can proceed without 1Password
    for k3s/k3d modes where secrets are managed separately.
    """
    import sys

    if not op_available():
        raise RuntimeError(
            "1Password CLI (op) not found. Install: brew install 1password-cli"
        )
    result = subprocess.run(
        ["op", "whoami"],
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        if not sys.stdin.isatty():
            print(
                "[WARN] No active 1Password session and no TTY — skipping op signin.",
                flush=True,
            )
            return
        subprocess.run(["op", "signin"], check=True)


def ensure_knoey_vault() -> None:
    """Create the 'knoey' vault if it does not already exist."""
    result = _op("vault", "list", "--format", "json", check=True)
    try:
        vaults = json.loads(result.stdout or "[]")
    except json.JSONDecodeError:
        vaults = []
    names = [v.get("name", "") for v in vaults]
    if _VAULT not in names:
        _op("vault", "create", _VAULT)
        print(f"[INFO] Created 1Password vault '{_VAULT}'")
    else:
        print(f"[INFO] 1Password vault '{_VAULT}' already exists")


def get_secret(item: str, field: str = "password") -> str:
    """Retrieve a field value from an item in the knoey vault."""
    result = _op(
        "item", "get", item,
        "--vault", _VAULT,
        "--fields", field,
        "--reveal",
        check=False,
    )
    if result.returncode != 0:
        return ""
    return result.stdout.strip()


def set_secret(item: str, field: str, value: str) -> None:
    """Set a field on an existing item, or create the item if absent."""
    check_result = _op("item", "get", item, "--vault", _VAULT, check=False)
    if check_result.returncode == 0:
        _op(
            "item", "edit", item,
            "--vault", _VAULT,
            f"{field}={value}",
        )
    else:
        _op(
            "item", "create",
            "--category", "login",
            "--title", item,
            "--vault", _VAULT,
            f"{field}={value}",
        )


def get_administrator_password() -> str:
    """Return the administrator password from the knoey vault."""
    return get_secret(_ADMIN_ITEM, "password")


def ensure_administrator_secret() -> str:
    """Return the administrator password, creating the item if absent."""
    pw = get_secret(_ADMIN_ITEM, "password")
    if pw:
        return pw
    pw = _generate_password()
    _op(
        "item", "create",
        "--category", "login",
        "--title", _ADMIN_ITEM,
        "--vault", _VAULT,
        f"password={pw}",
    )
    print(f"[INFO] Created 1Password item '{_ADMIN_ITEM}' in vault '{_VAULT}'")
    return pw


def _generate_password(length: int = 32) -> str:
    alphabet = string.ascii_letters + string.digits
    return "".join(secrets.choice(alphabet) for _ in range(length))