contrib/git-credential-google-gerrit

#!/usr/bin/env python3
# Copyright (C) 2026 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Reference git credential helper: Google OAuth tokens for Gerrit Git-over-HTTP.

This is an UNSUPPORTED reference/test utility. For most deployments prefer a
maintained helper (Git Credential Manager, git-credential-oauth) pointed at a
Desktop client listed in the plugin's `trusted-audience` (see config-google.md).

It runs the browser OAuth handshake once, stores the refresh token in a 0600 file
under ~/.config (like git's own credential-store), and mints fresh access tokens
on demand -- so `git clone/fetch/push` work without pasting tokens.

Use a dedicated Google **Desktop** client whose secret you treat as
non-confidential. Do NOT feed this helper Gerrit's confidential Web SSO secret;
Web-client mode is only for a throwaway/test or operator-controlled client.

Setup (once, per Gerrit host) -- --client-id is a Google Desktop client:
    git-credential-google-gerrit login \\
        --host https://gerrit.example.com \\
        --client-id <CLI_CLIENT_ID> \\
        --gerrit-username <USERNAME>
    # login prompts for the Google client secret without echo. Google's Desktop
    # flow requires the secret in the token exchange (it is not confidential
    # there); never pass it on argv or via the environment.

Secrets (refresh token, client secret) are written to a 0600 file, never passed
on the command line, so they do not appear in `ps`.

Enable for git (the empty first value resets the helper chain so a platform
keychain -- osxkeychain, libsecret -- does not cache the short-lived token ahead
of this helper):
    git config --global credential.https://gerrit.example.com.helper ""
    git config --global --add credential.https://gerrit.example.com.helper google-gerrit
    git config --global credential.https://gerrit.example.com.username <USERNAME>

Other commands:
    git-credential-google-gerrit token  --host <URL>   # print a fresh access token
    git-credential-google-gerrit logout --host <URL>   # forget the host
"""

import argparse
import base64
import getpass
import hashlib
import http.server
import json
import os
import secrets
import sys
import threading
import time
import urllib.parse
import urllib.request
import webbrowser

AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"
TOKEN_URL = "https://oauth2.googleapis.com/token"
DEFAULT_SCOPES = "openid email profile"
DEFAULT_PORT = 8090
STORE_NAME = "git-credential-google-gerrit"
# Refresh a little before the token actually expires to avoid edge-of-expiry 401s.
EXPIRY_SKEW_SECONDS = 120


def log(msg):
    print(msg, file=sys.stderr)


def die(msg, code=1):
    log("error: " + msg)
    sys.exit(code)


# --- Storage: a 0600 file per host, like git's own credential-store -----------
#
# Secrets go through the filesystem with 0600 perms, never through argv (which is
# world-visible via `ps`). We deliberately avoid the macOS `security` CLI: it
# takes the secret on argv (leaks) or, via stdin, truncates at ~128 chars, which
# is shorter than a Google refresh token. A production helper should bind the
# Keychain/Secret-Service API directly; that is out of scope for this stdlib
# reference.


def _file_store_path():
    base = os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config"))
    d = os.path.join(base, STORE_NAME)
    os.makedirs(d, mode=0o700, exist_ok=True)
    os.chmod(d, 0o700)  # repair perms if the dir pre-existed with a looser mode
    return d


def _record_path(host):
    return os.path.join(_file_store_path(), _host_key(host))


def store_load(host):
    path = _record_path(host)
    if os.path.exists(path):
        with open(path) as f:
            return json.load(f)
    return None


def store_save(host, data):
    path = _record_path(host)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        f.write(json.dumps(data))
    os.chmod(path, 0o600)  # repair perms if the file pre-existed 0644


def store_delete(host):
    path = _record_path(host)
    if os.path.exists(path):
        os.remove(path)


def _host_key(host):
    return hashlib.sha256(host.encode()).hexdigest()


# --- OAuth --------------------------------------------------------------------


def _post_token(params):
    body = urllib.parse.urlencode(params).encode()
    req = urllib.request.Request(TOKEN_URL, data=body)
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        detail = e.read().decode(errors="replace")
        die("token endpoint HTTP %d: %s" % (e.code, detail))


def _pkce_pair():
    verifier = base64.urlsafe_b64encode(secrets.token_bytes(40)).rstrip(b"=").decode()
    challenge = (
        base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest())
        .rstrip(b"=")
        .decode()
    )
    return verifier, challenge


class _CodeCatcher(http.server.BaseHTTPRequestHandler):
    code = None
    state = None

    def do_GET(self):  # noqa: N802 (http.server API)
        qs = urllib.parse.urlparse(self.path).query
        params = urllib.parse.parse_qs(qs)
        _CodeCatcher.code = params.get("code", [None])[0]
        _CodeCatcher.state = params.get("state", [None])[0]
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.end_headers()
        self.wfile.write(
            b"<html><body><h2>Authenticated.</h2>"
            b"You can close this tab and return to the terminal.</body></html>"
        )

    def log_message(self, *args):  # silence the default stderr logging
        pass


def interactive_login(client_id, client_secret, scopes, port):
    verifier, challenge = _pkce_pair()
    state = secrets.token_urlsafe(16)
    redirect_uri = "http://localhost:%d/" % port

    server = http.server.HTTPServer(("127.0.0.1", port), _CodeCatcher)
    threading.Thread(target=server.handle_request, daemon=True).start()

    auth_params = {
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "response_type": "code",
        "scope": scopes,
        "state": state,
        "code_challenge": challenge,
        "code_challenge_method": "S256",
        "access_type": "offline",
        "prompt": "consent",
    }
    url = AUTH_URL + "?" + urllib.parse.urlencode(auth_params)
    log("Opening browser for Google sign-in ...")
    log("If it does not open, visit:\n  " + url)
    webbrowser.open(url)

    deadline = time.time() + 300
    while _CodeCatcher.code is None and time.time() < deadline:
        time.sleep(0.2)
    server.server_close()
    if _CodeCatcher.code is None:
        die("timed out waiting for the OAuth redirect")
    if _CodeCatcher.state != state:
        die("OAuth state mismatch (possible CSRF); aborting")

    params = {
        "client_id": client_id,
        "code": _CodeCatcher.code,
        "code_verifier": verifier,
        "grant_type": "authorization_code",
        "redirect_uri": redirect_uri,
    }
    if client_secret:
        params["client_secret"] = client_secret
    tok = _post_token(params)
    if "refresh_token" not in tok:
        die("Google did not return a refresh_token (need access_type=offline + "
            "prompt=consent, and a first-time consent)")
    return tok


def refresh_access_token(rec):
    params = {
        "client_id": rec["client_id"],
        "grant_type": "refresh_token",
        "refresh_token": rec["refresh_token"],
    }
    if rec.get("client_secret"):
        params["client_secret"] = rec["client_secret"]
    tok = _post_token(params)
    return tok


def fresh_access_token(host):
    rec = store_load(host)
    if not rec:
        die("no stored credentials for %s; run: %s login --host %s ..."
            % (host, os.path.basename(sys.argv[0]), host))
    now = int(time.time())
    if rec.get("access_token") and rec.get("access_expiry", 0) - EXPIRY_SKEW_SECONDS > now:
        return rec["access_token"], rec.get("gerrit_username", "")
    tok = refresh_access_token(rec)
    rec["access_token"] = tok["access_token"]
    rec["access_expiry"] = now + int(tok.get("expires_in", 3600))
    # Google may rotate the refresh token; persist the new one if present.
    if tok.get("refresh_token"):
        rec["refresh_token"] = tok["refresh_token"]
    store_save(host, rec)
    return rec["access_token"], rec.get("gerrit_username", "")


# --- git credential protocol --------------------------------------------------


def read_git_attrs():
    attrs = {}
    for line in sys.stdin:
        line = line.rstrip("\n")
        if not line:
            break
        k, _, v = line.partition("=")
        attrs[k] = v
    return attrs


def host_from_attrs(attrs):
    proto = attrs.get("protocol", "https")
    host = attrs.get("host", "")
    return "%s://%s" % (proto, host)


def cmd_get():
    attrs = read_git_attrs()
    host = host_from_attrs(attrs)
    token, stored_username = fresh_access_token(host)
    # Gerrit Basic auth needs a real username; there is no meaningful default.
    username = attrs.get("username") or stored_username
    if not username:
        die("no Gerrit username for %s: set credential.<url>.username or re-run "
            "`login --gerrit-username <name>`" % host)
    sys.stdout.write("username=%s\n" % username)
    sys.stdout.write("password=%s\n" % token)


def cmd_store_or_erase():
    # We manage our own store via `login`/`logout`; ignore git's store/erase input.
    read_git_attrs()


def cmd_login(args):
    if not args.host or not args.client_id:
        die("login requires --host and --client-id")
    # Google's Desktop-client flow requires client_secret in the token exchange.
    # Prompt without echo so the secret never lands in argv, shell history, or
    # the environment. For a Desktop client this secret is not confidential.
    client_secret = getpass.getpass("Google client secret (input hidden): ")
    if not client_secret:
        die("a client secret is required (Google needs it even for Desktop clients)")
    tok = interactive_login(
        args.client_id, client_secret, args.scopes, args.port
    )
    rec = {
        "client_id": args.client_id,
        "client_secret": client_secret,
        "gerrit_username": args.gerrit_username or "",
        "refresh_token": tok["refresh_token"],
        "access_token": tok.get("access_token", ""),
        "access_expiry": int(time.time()) + int(tok.get("expires_in", 3600)),
    }
    store_save(args.host, rec)
    log("Stored credentials for %s." % args.host)
    log("Enable git (the empty value resets the helper chain so a platform")
    log("keychain does not cache the short-lived token ahead of this helper):")
    log("  git config --global credential.%s.helper ''" % args.host)
    log("  git config --global --add credential.%s.helper google-gerrit" % args.host)
    log("  git config --global credential.%s.username %s"
        % (args.host, args.gerrit_username))


def cmd_token(args):
    if not args.host:
        die("token requires --host")
    token, _ = fresh_access_token(args.host)
    print(token)


def cmd_logout(args):
    if not args.host:
        die("logout requires --host")
    store_delete(args.host)
    log("Forgot credentials for %s." % args.host)


def main():
    p = argparse.ArgumentParser(add_help=True, description=__doc__)
    sub = p.add_subparsers(dest="command")

    # git invokes: <helper> get|store|erase  (operation as the sole positional)
    for op in ("get", "store", "erase"):
        sub.add_parser(op)

    pl = sub.add_parser("login")
    pl.add_argument("--host", required=True)
    pl.add_argument("--client-id", required=True)
    # Required: Gerrit's Basic-auth path resolves the account by username.
    pl.add_argument("--gerrit-username", required=True)
    pl.add_argument("--scopes", default=DEFAULT_SCOPES)
    pl.add_argument("--port", type=int, default=DEFAULT_PORT)

    pt = sub.add_parser("token")
    pt.add_argument("--host", required=True)

    po = sub.add_parser("logout")
    po.add_argument("--host", required=True)

    args = p.parse_args()
    if args.command == "get":
        cmd_get()
    elif args.command in ("store", "erase"):
        cmd_store_or_erase()
    elif args.command == "login":
        cmd_login(args)
    elif args.command == "token":
        cmd_token(args)
    elif args.command == "logout":
        cmd_logout(args)
    else:
        p.print_help()
        sys.exit(2)


if __name__ == "__main__":
    main()