secrets-engine/src/secrets_engine/exec_delivery.py

"""Exec-time delivery: make a secret available only to a child process.

The default and preferred delivery mode. The value is fetched from OpenBao,
injected into the child's environment / a temp config, the child runs, and the
injection is destroyed afterward — on success, failure, or interruption.

Supported here:
  - npm-config: write a temporary .npmrc with the auth token and point the child
    at it via NPM_CONFIG_USERCONFIG. Preferred for `npm publish`.
  - exec-env:   inject the value as an environment variable for the child only.

The parent shell never sees the value; the value is never logged. Child stdout/
stderr is streamed through a redactor as a backstop.
"""
from __future__ import annotations

import json
import os
import signal
import subprocess
import sys
import tempfile
from contextlib import contextmanager
from pathlib import Path
from typing import Iterator

from secrets_engine.catalog import CatalogEntry
from secrets_engine.errors import DeliveryError
from secrets_engine.openbao import OpenBaoClient
from secrets_engine.publication_policy import PublicationPolicy, resolve
from secrets_engine.redact import redact_text


def resolve_npm_token_env(entry: CatalogEntry, *, policy_dir=None) -> str:
    """Resolve the env-var name to inject for a lane via the publication policy."""
    if policy_dir is None:
        from secrets_engine.config import Config

        policy_dir = Config.load().policy_dir
    npm = entry.npm
    policy = PublicationPolicy.load(policy_dir)
    res = resolve(
        policy,
        org=entry.org,
        repo=entry.repo,
        npm_scope=npm.get("scope", ""),
        package_maturity=npm.get("maturity", "maturity-build"),
        token_env_override=npm.get("token_env", ""),
    )
    return res.token_env


def _fetch_value(client: OpenBaoClient, entry: CatalogEntry, field: str) -> str:
    """Read the field value via an approle-scoped token. Held in memory only."""
    try:
        token = client.approle_login_token(entry.role_name)
    except Exception as e:
        raise DeliveryError(f"could not obtain scoped token for delivery: {e}") from e
    scoped = OpenBaoClient(addr=client.addr, token=token, bao_bin=client.bao_bin)
    proc = scoped._run(["kv", "get", "-format=json", f"{entry.mount}/{entry.path}"])
    if proc.returncode != 0:
        raise DeliveryError(f"scoped read failed for lane '{entry.id}' (denied or absent)")
    try:
        data = json.loads(proc.stdout)["data"]["data"]
    except (json.JSONDecodeError, KeyError) as e:
        raise DeliveryError(f"malformed KV response for lane '{entry.id}'") from e
    if field not in data:
        raise DeliveryError(f"field '{field}' absent in lane '{entry.id}'")
    return data[field]


def _registry_authkey(registry: str) -> str:
    """Turn a registry URL into the npm `//host/path/:_authToken` config key."""
    no_scheme = registry.split("://", 1)[-1]
    if not no_scheme.endswith("/"):
        no_scheme += "/"
    return "//" + no_scheme


@contextmanager
def _npm_userconfig(registry: str, scope: str, token_env: str) -> Iterator[Path]:
    """Write a mode-0600 temp .npmrc for the configured registry/scope.

    The token itself is NOT written to the file — npm expands ${<token_env>}
    from the child environment, so the value never touches disk. `token_env` is
    resolved from the netkingdom publication-scope policy, so its name reflects
    the lane's effective publication scope.
    """
    fd, name = tempfile.mkstemp(prefix="se-npmrc-", suffix=".ini")
    path = Path(name)
    try:
        os.fchmod(fd, 0o600)
        authkey = _registry_authkey(registry)
        with os.fdopen(fd, "w") as fh:
            # e.g. @whynot:registry=https://gitea.coulomb.social/api/packages/coulomb/npm/
            fh.write(f"{scope}:registry={registry}\n")
            fh.write(f"{authkey}:_authToken=${{{token_env}}}\n")
        yield path
    finally:
        try:
            path.unlink()
        except FileNotFoundError:
            pass


def _stream_redacted(proc: subprocess.Popen, secret: str) -> None:
    """Stream child output through the redactor (backstop)."""
    assert proc.stdout is not None
    for line in proc.stdout:
        sys.stdout.write(redact_text(line, extra=[secret]))
        sys.stdout.flush()


def exec_with_secret(
    client: OpenBaoClient,
    entry: CatalogEntry,
    field: str,
    command: list[str],
    *,
    mode: str = "auto",
    policy_dir=None,
) -> int:
    """Run `command` with the lane's secret injected for the child only.

    Returns the child's exit code. Raises DeliveryError if setup is unsafe.
    """
    if not command:
        raise DeliveryError("no command given to exec")

    declared = set(entry.delivery_modes)
    if mode == "auto":
        mode = "npm-config" if "npm-config" in declared else (
            "exec-env" if "exec-env" in declared else ""
        )
    if not mode:
        raise DeliveryError(
            f"lane '{entry.id}' declares no exec-capable delivery mode "
            f"({sorted(declared)})"
        )
    if mode not in declared:
        raise DeliveryError(
            f"delivery mode '{mode}' not permitted for lane '{entry.id}' "
            f"(allowed {sorted(declared)})"
        )

    value = _fetch_value(client, entry, field)
    child_env = dict(os.environ)

    if mode == "npm-config":
        npm = entry.npm
        registry = npm.get("registry", "")
        scope = npm.get("scope", "")
        if not registry or not scope:
            raise DeliveryError(
                f"lane '{entry.id}' npm-config delivery needs "
                "delivery_config.npm.registry and .scope"
            )
        token_env = resolve_npm_token_env(entry, policy_dir=policy_dir)
        with _npm_userconfig(registry, scope, token_env) as npmrc:
            child_env["NPM_CONFIG_USERCONFIG"] = str(npmrc)
            child_env[token_env] = value
            rc = _spawn(command, child_env, value)
        return rc

    if mode == "exec-env":
        # Inject under a conventional name derived from the field.
        env_name = field.upper()
        child_env[env_name] = value
        return _spawn(command, child_env, value)

    raise DeliveryError(f"unsupported delivery mode '{mode}'")


def _spawn(command: list[str], env: dict[str, str], secret: str) -> int:
    """Spawn the child, stream redacted output, propagate signals, ensure cleanup."""
    try:
        proc = subprocess.Popen(
            command,
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
        )
    except FileNotFoundError as e:
        raise DeliveryError(f"command not found: {command[0]}") from e

    def _forward(signum, _frame):
        proc.send_signal(signum)

    old_int = signal.signal(signal.SIGINT, _forward)
    old_term = signal.signal(signal.SIGTERM, _forward)
    try:
        _stream_redacted(proc, secret)
        return proc.wait()
    finally:
        signal.signal(signal.SIGINT, old_int)
        signal.signal(signal.SIGTERM, old_term)
        # env dict goes out of scope; the temp npmrc is removed by its context mgr.