agent-credential-broker/agent_credential_broker/adapters/cli_profile.py

"""CLI-profile adapter: bounded subprocess execution of pre-declared
tools only. There is intentionally NO generic "run with secret" -- every
profile is a declarative allowlist:

    {
      "id": "example-cli",
      "executable": "example-cli",
      "injection": {"method": "env", "name": "EXAMPLE_TOKEN"},
      "allowed_subcommands": ["list", "status"],
      "arg_regex": "^[\\w .\\-/:]+$",
      "forbidden_regex": "[;&|`$<>(){}\\[\\]]",
      "target_hosts": ["api.example.com"],
      "timeout": 30,
      "max_output_bytes": 65536
    }

Injection methods (in order of preference):
  stdin -- secret written to child stdin (no argv, no env, no file)
  file  -- 0600 tmpfs file rendered from file_template ({SECRET}
           placeholder); the child gets its PATH via env_name and/or
           profile args_template entries containing {SECRET_FILE}
  env   -- secret exported ONLY into the child environment
  none  -- no secret reaches the child (metadata-only calls)

Profile args_template entries are broker-supplied args (they may embed
{SECRET_FILE} but NEVER the secret itself), so tools like curl can pick
credentials up from config files without argv exposure. Agent-supplied
args are validated separately.
argv is never an injection channel.
"""

from __future__ import annotations

import json
import re
import shutil
import urllib.parse
from pathlib import Path

from ..config import profiles_dir
from ..errors import denied, invalid_argument
from ..items import item_secrets
from .common import clean_env, run_subprocess, secret_dir, write_secret_file

_DEFAULT_ARG_RE = r"^[\w ./:@\-+=%]+$"
_DEFAULT_FORBIDDEN_RE = r"[;&|`$<>(){}\[\]\n!#]|(^|/)\.\.(/|$)"
_MAX_ARGS = 64


def load_profile(profile_id: str) -> dict:
    if not profile_id or not re.match(r"^[\w.\-]+$", profile_id):
        raise invalid_argument("bad profile id")
    path = profiles_dir() / f"{profile_id}.json"
    if not path.is_file():
        raise denied(f"unknown cli profile {profile_id!r}")
    try:
        profile = json.loads(path.read_text())
    except Exception:
        raise denied(f"invalid cli profile {profile_id!r}")
    if profile.get("id") != profile_id:
        raise denied(f"profile id mismatch in {profile_id!r}")
    if profile.get("disabled"):
        raise denied(f"cli profile {profile_id!r} is disabled")
    return profile


def _validate_args(profile, args):
    if not isinstance(args, list) or len(args) > _MAX_ARGS:
        raise invalid_argument("bad args")
    arg_re = re.compile(profile.get("arg_regex") or _DEFAULT_ARG_RE)
    forbidden = profile.get("forbidden_regex")
    for a in args:
        if not isinstance(a, str) or not a or len(a) > 500:
            raise invalid_argument("bad argument")
        if not arg_re.match(a):
            raise invalid_argument(f"argument {a!r} rejected by profile arg_regex")
        if forbidden and re.search(forbidden, a):
            raise invalid_argument(f"argument {a!r} rejected by profile forbidden_regex")
    subs = profile.get("allowed_subcommands")
    if subs:
        positional = [a for a in args if not a.startswith("-")]
        if not positional or positional[0] not in subs:
            raise invalid_argument(
                f"first positional must be one of {subs}")
    targets = {h.lower() for h in (profile.get("target_hosts") or [])}
    if targets:
        for a in args:
            if "://" in a:
                host = (urllib.parse.urlparse(a).hostname or "").lower()
                if host not in targets:
                    raise invalid_argument(
                        f"url target {host!r} not in profile target_hosts")


def validate(item, policy, params, config):
    profile_id = params.get("profile")
    profile = load_profile(profile_id)
    if policy.cli_profile and policy.cli_profile != profile_id:
        raise denied(f"credential {policy.ref!r} is bound to profile {policy.cli_profile!r}")
    _validate_args(profile, params.get("args") or [])


def run(item, policy, params, config):
    profile_id = params.get("profile")
    profile = load_profile(profile_id)
    if policy.cli_profile and policy.cli_profile != profile_id:
        raise denied(f"credential {policy.ref!r} is bound to profile {policy.cli_profile!r}")

    executable = profile.get("executable") or ""
    if "/" in executable or ".." in executable:
        raise denied("profile executable must be a bare name")
    exe = shutil.which(executable)
    if not exe:
        raise denied(f"executable {executable!r} not found")

    args = params.get("args") or []
    _validate_args(profile, args)

    injection = profile.get("injection") or {}
    method = injection.get("method")
    secret = item.get("password") or ""
    secrets = item_secrets(item)
    timeout = int(profile.get("timeout") or 30)
    argv = [exe] + [str(a) for a in args]
    input_text = None

    if method == "stdin":
        input_text = secret + "\n"
        env = clean_env(profile.get("env") or {})
    elif method == "file":
        with secret_dir() as d:
            sfile = d / "secret"
            template = injection.get("file_template") or "{SECRET}"
            if "{SECRET}" not in template:
                raise denied("file_template must contain {SECRET}")
            write_secret_file(sfile, template.replace("{SECRET}", secret))
            env = clean_env(profile.get("env") or {})
            env_name = injection.get("env_name")
            if env_name:
                env[env_name] = str(sfile)
            extra = [a.replace("{SECRET_FILE}", str(sfile))
                     for a in (profile.get("args_template") or [])]
            argv = [exe] + extra + argv[1:]
            res = run_subprocess(argv, env=env, input_text=None,
                                 timeout=timeout, secrets=secrets)
            res["profile"] = profile_id
            res["injection"] = "file"
            return _cap(profile, res)
    elif method == "env":
        env = clean_env(profile.get("env") or {})
        env[injection.get("name") or "CREDENTIAL"] = secret
    else:
        env = clean_env(profile.get("env") or {})
        if method not in (None, "none"):
            raise denied(f"unknown injection method {method!r}")

    res = run_subprocess(argv, env=env, input_text=input_text,
                         timeout=timeout, secrets=secrets)
    res["profile"] = profile_id
    res["injection"] = method or "none"
    return _cap(profile, res)


def _cap(profile, res):
    cap = int(profile.get("max_output_bytes") or 65536)
    res["stdout"] = res["stdout"][:cap]
    res["stderr"] = res["stderr"][:cap]
    return res