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

"""Git adapter: clone/fetch/push over HTTPS with askpass auth.

The token/password is supplied through a generated GIT_ASKPASS script
that cats 0600 runtime-tmpfs files -- it never appears on argv, never
inside the remote URL, and never in output (output is sanitized too).
Credentials embedded in URLs (https://user:pass@host) are rejected.
"""

from __future__ import annotations

import configparser
import os
import re
import shutil
import urllib.parse
from pathlib import Path

from ..errors import invalid_argument, origin_denied, unsafe_tool_integration
from ..policy import require_url_allowed
from ..items import item_secrets
from .common import (clean_env, run_subprocess, secret_dir, write_askpass,
                     write_secret_file)

_OPS = {"clone", "fetch", "push"}
_REFSPEC_OK = re.compile(r"^[\w\-./:+]+$")


def _validate_repo_url(item, policy, url):
    if not url:
        raise invalid_argument("missing repository url")
    if url.startswith("git@") or url.startswith("ssh://"):
        raise unsafe_tool_integration(
            "git: ssh remotes not wired to vault credentials in this phase; use https")
    p = urllib.parse.urlparse(url)
    if p.scheme not in ("https", "http"):
        raise invalid_argument(f"unsupported git remote scheme {p.scheme!r}")
    if "@" in (p.netloc or ""):
        raise invalid_argument("credentials must not be embedded in the remote URL")
    require_url_allowed(policy, url)


def _check_workdir(config_roots, workdir):
    wd = Path(workdir).resolve()
    roots = [Path(r).expanduser().resolve() for r in config_roots] or [Path.home()]
    if not any(str(wd).startswith(str(r) + os.sep) or wd == r for r in roots):
        raise origin_denied("git", f"workdir {wd} outside allowed roots")
    if not (wd / ".git").is_dir():
        raise invalid_argument(f"{wd} is not a git worktree")
    return wd


def _remote_url(workdir: Path, remote: str) -> str:
    """Read the remote's URL straight from .git/config -- the credential
    will actually go there, so THAT is the URL we validate, not whatever
    the agent claimed in `repository`."""
    cfg = configparser.RawConfigParser()
    cfg.read(workdir / ".git" / "config")
    try:
        url = cfg.get(f'remote "{remote}"', "url")
    except configparser.Error:
        raise invalid_argument(f"remote {remote!r} has no url configured")
    if "://" in url and "@" in urllib.parse.urlparse(url).netloc:
        raise invalid_argument("remote URL has embedded credentials")
    return url


def validate(item, policy, params, config):
    op = params.get("op") or params.get("operation")
    if op not in _OPS:
        raise invalid_argument(f"git op must be one of {sorted(_OPS)}")
    _validate_repo_url(item, policy, params.get("repository") or "")
    if op != "clone":
        workdir = _check_workdir(config.git_workdir_roots, params.get("workdir") or ".")
        remote = params.get("remote") or "origin"
        remote_url = _remote_url(workdir, remote)
        _validate_repo_url(item, policy, remote_url)


def run(item, policy, params, config):
    op = params.get("op") or params.get("operation")
    if op not in _OPS:
        raise invalid_argument(f"git op must be one of {sorted(_OPS)}")
    repo_url = params.get("repository") or ""
    _validate_repo_url(item, policy, repo_url)

    username = item.get("username")
    password = item.get("password")
    if not password:
        raise unsafe_tool_integration("git: item has no token/password")
    secrets = item_secrets(item)

    git = shutil.which("git") or "/usr/bin/git"
    with secret_dir() as d:
        ufile = d / "user"
        pfile = d / "pass"
        write_secret_file(ufile, username or "oauth2")
        write_secret_file(pfile, password)
        askpass = write_askpass(d, ufile, pfile)
        env = clean_env({
            "GIT_ASKPASS": str(askpass),
            "GIT_TERMINAL_PROMPT": "0",
        })

        if op == "clone":
            dest = params.get("dest")
            if not dest:
                raise invalid_argument("clone requires dest")
            argv = [git, "clone", "--", repo_url, str(dest)]
        else:
            workdir = _check_workdir(config.git_workdir_roots, params.get("workdir") or ".")
            remote = params.get("remote") or "origin"
            if not _REFSPEC_OK.match(remote):
                raise invalid_argument("bad remote name")
            remote_url = _remote_url(workdir, remote)
            if remote_url.startswith("git@") or remote_url.startswith("ssh://"):
                raise unsafe_tool_integration(
                    "git: ssh remotes not wired to vault credentials in this phase")
            _validate_repo_url(item, policy, remote_url)
            claimed = urllib.parse.urlparse(repo_url).netloc
            actual = urllib.parse.urlparse(remote_url).netloc
            if claimed.lower() != actual.lower():
                raise origin_denied(policy.ref,
                                    f"repository claim {claimed!r} != remote {actual!r}")
            argv = [git, "-C", str(workdir)]
            if op == "fetch":
                argv += ["fetch", "--", remote]
            else:
                refspec = params.get("refspec") or params.get("branch") or "HEAD"
                if not _REFSPEC_OK.match(refspec):
                    raise invalid_argument("bad refspec")
                argv += ["push", "--", remote, refspec]
        result = run_subprocess(argv, env=env, timeout=180, secrets=secrets)
        result["op"] = op
        result["repository"] = urllib.parse.urlparse(repo_url).netloc
        return result