depot_tools/git_auth.py

# Copyright (c) 2024 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Defines utilities for setting up Git authentication."""

from __future__ import annotations

import enum
from collections.abc import Collection
import contextlib
import functools
import logging
import os
from typing import TYPE_CHECKING, Callable, NamedTuple, TextIO
import urllib.parse

import gerrit_util
import newauth
import scm

if TYPE_CHECKING:
    # Causes import cycle if imported normally
    import git_cl


class ConfigMode(enum.Enum):
    """Modes to pass to ConfigChanger"""
    NO_AUTH = 1
    NEW_AUTH = 2
    NEW_AUTH_SSO = 3


class ConfigChanger(object):
    """Changes Git auth config as needed for Gerrit."""

    # Can be used to determine whether this version of the config has
    # been applied to a Git repo.
    #
    # Increment this when making changes to the config, so that reliant
    # code can determine whether the config needs to be re-applied.
    VERSION: int = 6

    def __init__(
        self,
        *,
        mode: ConfigMode,
        remote_url: str,
        set_config_func: Callable[..., None] = scm.GIT.SetConfig,
    ):
        """Create a new ConfigChanger.

        Args:
            mode: How to configure auth
            remote_url: Git repository's remote URL, e.g.,
                https://chromium.googlesource.com/chromium/tools/depot_tools.git
            set_config_func: Function used to set configuration.  Used
                for testing.
        """
        self.mode: ConfigMode = mode

        self._remote_url: str = remote_url
        self._set_config_func: Callable[..., None] = set_config_func

    @functools.cached_property
    def _shortname(self) -> str:
        # Example: chromium
        parts: urllib.parse.SplitResult = urllib.parse.urlsplit(
            self._remote_url)
        return _url_shortname(parts)

    @functools.cached_property
    def _host_url(self) -> str:
        # Example: https://chromium.googlesource.com
        # Example: https://chromium-review.googlesource.com
        parts: urllib.parse.SplitResult = urllib.parse.urlsplit(
            self._remote_url)
        return _url_host_url(parts)

    @functools.cached_property
    def _root_url(self) -> str:
        # Example: https://chromium.googlesource.com/
        # Example: https://chromium-review.googlesource.com/
        parts: urllib.parse.SplitResult = urllib.parse.urlsplit(
            self._remote_url)
        return _url_root_url(parts)

    @classmethod
    def new_from_env(cls, cwd: str, cl: git_cl.Changelist) -> ConfigChanger:
        """Create a ConfigChanger by inferring from env.

        The Gerrit host is inferred from the current repo/branch.
        The user, which is used to determine the mode, is inferred using
        git-config(1) in the given `cwd`.
        """
        # This is determined either from the branch or repo config.
        #
        # Example: chromium-review.googlesource.com
        gerrit_host = cl.GetGerritHost()
        # This depends on what the user set for their remote.
        # There are a couple potential variations for the same host+repo.
        #
        # Example:
        # https://chromium.googlesource.com/chromium/tools/depot_tools.git
        remote_url = cl.GetRemoteUrl()

        if gerrit_host is None or remote_url is None:
            raise Exception(
                'Error Git auth settings inferring from environment:'
                f' {gerrit_host=} {remote_url=}')
        assert gerrit_host is not None
        assert remote_url is not None

        return cls(
            mode=cls._infer_mode(cwd, gerrit_host),
            remote_url=remote_url,
        )

    @classmethod
    def new_for_remote(cls, cwd: str, remote_url: str) -> ConfigChanger:
        """Create a ConfigChanger for the given Gerrit host.

        The user, which is used to determine the mode, is inferred using
        git-config(1) in the given `cwd`.
        """
        c = cls(
            mode=ConfigMode.NEW_AUTH,
            remote_url=remote_url,
        )
        assert c._shortname, "Short name is empty"
        c.mode = cls._infer_mode(cwd, c._shortname + '-review.googlesource.com')
        return c

    @staticmethod
    def _infer_mode(cwd: str, gerrit_host: str) -> ConfigMode:
        """Infer default mode to use."""
        if not newauth.Enabled():
            return ConfigMode.NO_AUTH
        email: str = scm.GIT.GetConfig(cwd, 'user.email') or ''
        if gerrit_util.ShouldUseSSO(gerrit_host, email):
            return ConfigMode.NEW_AUTH_SSO
        if not gerrit_util.GitCredsAuthenticator.gerrit_account_exists(
                gerrit_host):
            return ConfigMode.NO_AUTH
        return ConfigMode.NEW_AUTH

    def apply(self, cwd: str) -> None:
        """Apply config changes to the Git repo directory."""
        self._apply_cred_helper(cwd)
        self._apply_sso(cwd)
        self._apply_gitcookies(cwd)

    def apply_global(self, cwd: str) -> None:
        """Apply config changes to the global (user) Git config.

        This will make the instance's mode (e.g., SSO or not) the global
        default for the Gerrit host, if not overridden by a specific Git repo.
        """
        self._apply_global_cred_helper(cwd)
        self._apply_global_sso(cwd)

    def _apply_cred_helper(self, cwd: str) -> None:
        """Apply config changes relating to credential helper."""
        cred_key: str = f'credential.{self._host_url}.helper'
        if self.mode == ConfigMode.NEW_AUTH:
            self._set_config(cwd, cred_key, '', modify_all=True)
            self._set_config(cwd, cred_key, 'luci', append=True)
        elif self.mode == ConfigMode.NEW_AUTH_SSO:
            self._set_config(cwd, cred_key, None, modify_all=True)
        elif self.mode == ConfigMode.NO_AUTH:
            self._set_config(cwd, cred_key, None, modify_all=True)
        else:
            raise TypeError(f'Invalid mode {self.mode!r}')

        # Cleanup old from version 4
        old_key: str = f'credential.{self._root_url}.helper'
        self._set_config(cwd, old_key, None, modify_all=True)

    def _apply_sso(self, cwd: str) -> None:
        """Apply config changes relating to SSO."""
        sso_key: str = f'url.sso://{self._shortname}/.insteadOf'
        http_key: str = f'url.{self._remote_url}.insteadOf'
        if self.mode == ConfigMode.NEW_AUTH:
            self._set_config(cwd, 'protocol.sso.allow', None)
            self._set_config(cwd, sso_key, None, modify_all=True)
            # Shadow a potential global SSO rewrite rule.
            self._set_config(cwd, http_key, self._remote_url, modify_all=True)
        elif self.mode == ConfigMode.NEW_AUTH_SSO:
            self._set_config(cwd, 'protocol.sso.allow', 'always')
            self._set_config(cwd, sso_key, self._root_url, modify_all=True)
            self._set_config(cwd, http_key, None, modify_all=True)
        elif self.mode == ConfigMode.NO_AUTH:
            self._set_config(cwd, 'protocol.sso.allow', None)
            self._set_config(cwd, sso_key, None, modify_all=True)
            self._set_config(cwd, http_key, None, modify_all=True)
        else:
            raise TypeError(f'Invalid mode {self.mode!r}')

    def _apply_gitcookies(self, cwd: str) -> None:
        """Apply config changes relating to gitcookies."""
        if self.mode == ConfigMode.NEW_AUTH:
            # Override potential global setting
            self._set_config(cwd, 'http.cookieFile', '', modify_all=True)
        elif self.mode == ConfigMode.NEW_AUTH_SSO:
            # Override potential global setting
            self._set_config(cwd, 'http.cookieFile', '', modify_all=True)
        elif self.mode == ConfigMode.NO_AUTH:
            self._set_config(cwd, 'http.cookieFile', None, modify_all=True)
        else:
            raise TypeError(f'Invalid mode {self.mode!r}')

    def _apply_global_cred_helper(self, cwd: str) -> None:
        """Apply config changes relating to credential helper."""
        cred_key: str = f'credential.{self._host_url}.helper'
        if self.mode == ConfigMode.NEW_AUTH:
            self._set_config(cwd, cred_key, '', scope='global', modify_all=True)
            self._set_config(cwd, cred_key, 'luci', scope='global', append=True)
        elif self.mode == ConfigMode.NEW_AUTH_SSO:
            # Avoid editing the user's config in case they manually
            # configured something.
            pass
        elif self.mode == ConfigMode.NO_AUTH:
            # Avoid editing the user's config in case they manually
            # configured something.
            pass
        else:
            raise TypeError(f'Invalid mode {self.mode!r}')

        # Cleanup old from version 4
        old_key: str = f'credential.{self._root_url}.helper'
        self._set_config(cwd, old_key, None, modify_all=True, scope='global')

    def _apply_global_sso(self, cwd: str) -> None:
        """Apply config changes relating to SSO."""
        sso_key: str = f'url.sso://{self._shortname}/.insteadOf'
        if self.mode == ConfigMode.NEW_AUTH:
            # Do not unset protocol.sso.allow because it may be used by
            # other hosts.
            self._set_config(cwd,
                             sso_key,
                             None,
                             scope='global',
                             modify_all=True)
        elif self.mode == ConfigMode.NEW_AUTH_SSO:
            self._set_config(cwd,
                             'protocol.sso.allow',
                             'always',
                             scope='global')
            self._set_config(cwd,
                             sso_key,
                             self._root_url,
                             scope='global',
                             modify_all=True)
        elif self.mode == ConfigMode.NO_AUTH:
            # Avoid editing the user's config in case they manually
            # configured something.
            pass
        else:
            raise TypeError(f'Invalid mode {self.mode!r}')

    def _set_config(self, *args, **kwargs) -> None:
        self._set_config_func(*args, **kwargs)


def AutoConfigure(cwd: str, cl: git_cl.Changelist) -> None:
    """Configure Git authentication automatically.

    This tracks when the config that has already been applied and skips
    doing anything if so.

    This may modify the global Git config and the local repo config as
    needed.
    """
    latestVer: int = ConfigChanger.VERSION
    v: int = 0
    try:
        v = int(
            scm.GIT.GetConfig(cwd, 'depot-tools.gitauthautoconfigured') or '0')
    except ValueError:
        v = 0
    if v < latestVer:
        logging.debug(
            'Automatically configuring Git repo authentication'
            ' (current version: %r, latest: %r)', v, latestVer)
        Configure(cwd, cl)
        scm.GIT.SetConfig(cwd, 'depot-tools.gitAuthAutoConfigured',
                          str(latestVer))


def Configure(cwd: str, cl: git_cl.Changelist) -> None:
    """Configure Git authentication.

    This may modify the global Git config and the local repo config as
    needed.
    """
    logging.debug('Configuring Git authentication...')

    logging.debug('Configuring global Git authentication...')

    # We want the user's global config.
    # We can probably assume the root directory doesn't have any local
    # Git configuration.
    c = ConfigChanger.new_from_env('/', cl)
    c.apply_global(os.path.expanduser('~'))

    c2 = ConfigChanger.new_from_env(cwd, cl)
    if c2.mode == c.mode:
        logging.debug(
            'Local user wants same mode %s as global;'
            ' clearing local repo auth config', c2.mode)
        c2.mode = ConfigMode.NO_AUTH
        c2.apply(cwd)
        return
    logging.debug('Local user wants mode %s while global user wants mode %s',
                  c2.mode, c.mode)
    logging.debug('Configuring current Git repo authentication...')
    c2.apply(cwd)


def ConfigureGlobal(cwd: str, remote_url: str) -> None:
    """Configure global/user Git authentication."""
    logging.debug('Configuring global Git authentication for %s', remote_url)
    # Checks to ensure this doesn't error when called with "bad" URLs.
    #
    # Don't try to configure auth for local files.
    if remote_url.startswith('file://'):
        return
    # Skip for local files that aren't even URIs.
    if '://' not in remote_url:
        return
    ConfigChanger.new_for_remote(cwd, remote_url).apply_global(cwd)


def ClearRepoConfig(cwd: str, cl: git_cl.Changelist) -> None:
    """Clear the current Git repo authentication."""
    logging.debug('Clearing current Git repo authentication...')
    c = ConfigChanger.new_from_env(cwd, cl)
    c.mode = ConfigMode.NO_AUTH
    c.apply(cwd)


class _ConfigError(Exception):
    """Subclass for errors raised by ConfigWizard.

    This may be unused, but keep this around so that anyone who needs it
    when tweaking ConfigWizard can use it.
    """


class _ConfigMethod(enum.Enum):
    """Enum used in _ConfigInfo."""
    OAUTH = 1
    SSO = 2


class _ConfigInfo(NamedTuple):
    """Result for ConfigWizard._configure."""
    method: _ConfigMethod


class _GitcookiesSituation(NamedTuple):
    """Result for ConfigWizard._check_gitcookies."""
    gitcookies_exists: bool
    cookiefile: str
    cookiefile_exists: bool
    divergent_cookiefiles: bool


_InputChecker = Callable[['UserInterface', str], bool]


def _check_any(ui: UserInterface, line: str) -> bool:
    """Allow any input."""
    return True


def _check_nonempty(ui: UserInterface, line: str) -> bool:
    """Reject nonempty input."""
    if line:
        return True
    ui.write('Input cannot be empty.\n')
    return False


def _check_choice(choices: Collection[str]) -> _InputChecker:
    """Allow specified choices."""

    def func(ui: UserInterface, line: str) -> bool:
        if line in choices:
            return True
        ui.write('Invalid choice.\n')
        return False

    return func


class UserInterface(object):
    """Abstracts user interaction for ConfigWizard.

    This implementation supports regular terminals.
    """

    _prompts = {
        None: 'y/n',
        True: 'Y/n',
        False: 'y/N',
    }

    def __init__(self, stdin: TextIO, stdout: TextIO):
        self._stdin = stdin
        self._stdout = stdout

    def read_yn(self, prompt: str, *, default: bool | None = None) -> bool:
        """Reads a yes/no response.

        The prompt should end in '?'.
        """
        prompt = f'{prompt} [{self._prompts[default]}]: '
        while True:
            self._stdout.write(prompt)
            self._stdout.flush()
            response = self._stdin.readline().strip().lower()
            if response in ('y', 'yes'):
                return True
            if response in ('n', 'no'):
                return False
            if not response and default is not None:
                return default
            self._stdout.write('Type y or n.\n')

    def read_line(self,
                  prompt: str,
                  *,
                  check: _InputChecker = _check_any) -> str:
        """Reads a line of input.

        Trailing whitespace is stripped from the read string.
        The prompt should not end in any special indicator like a colon.

        Optionally, an input check function may be provided.  This
        method will continue to prompt for input until it passes the
        check.  The check should print some explanation for rejected
        inputs.
        """
        while True:
            self._stdout.write(f'{prompt}: ')
            self._stdout.flush()
            s = self._stdin.readline().rstrip()
            if check(self, s):
                return s

    def read_enter(self, text: str = 'Press Enter to proceed.') -> None:
        """Reads an Enter.

        Used to interactively proceed.
        """
        self._stdout.write(text)
        self._stdout.flush()
        self._stdin.readline()

    def write(self, s: str) -> None:
        """Write string as-is.

        The string should usually end in a newline.
        """
        self._stdout.write(s)


RemoteURLFunc = Callable[[], str]


class ConfigWizard(object):
    """Wizard for setting up user's Git config Gerrit authentication.

    Instances carry internal state, so cannot be reused.
    """

    def __init__(self, *, ui: UserInterface, remote_url_func: RemoteURLFunc):
        self._ui = ui
        self._remote_url_func = remote_url_func

        # Internal state
        self._user_actions = []

    def run(self, *, force_global: bool) -> None:
        with self._handle_config_errors():
            self._run(force_global=force_global)

    def _run(self, *, force_global: bool) -> None:
        self._println('This tool will help check your Gerrit authentication.')
        self._println(
            '(Report any issues to https://issues.chromium.org/issues/new?component=1456702&template=2076315)'
        )
        self._println()
        self._fix_gitcookies()
        self._println()
        self._println('Checking for SSO helper...')
        if self._check_sso_helper():
            self._println('SSO helper is available.')
            self._set_config('protocol.sso.allow', 'always', scope='global')
        self._println()
        self._run_gerrit_host_configuration(force_global=force_global)
        self._println()
        self._println('Successfully finished auth configuration check.')
        self._print_actions_for_user()

    def _run_gerrit_host_configuration(self, *, force_global: bool) -> None:
        remote_url = self._remote_url_func()
        if _is_gerrit_url(remote_url):
            if force_global:
                self._println(
                    'We will pretend to be running outside of a Gerrit repository'
                )
                self._println(
                    'and check your global Git configuration since you passed --global.'
                )
                self._run_outside_repo()
            else:
                self._println(
                    'Looks like we are running inside a Gerrit repository,')
                self._println('so we will check your Git configuration for it.')
                self._run_inside_repo()
        else:
            self._println(
                'Looks like we are running outside of a Gerrit repository,')
            self._println('so we will check your global Git configuration.')
            self._run_outside_repo()

    def _run_outside_repo(self) -> None:
        global_email = self._check_global_email()

        self._println()
        self._println('Since we are not running in a Gerrit repository,')
        self._println('we do not know which Gerrit host to check.')
        self._println(
            'You can re-run this command inside a Gerrit repository to check a specific host,'
        )
        self._println('or we can set up some commonly used Gerrit hosts.')

        self._println()
        self._println(
            "(If you haven't already set up auth for these Gerrit hosts,")
        self._println(
            "and you skip this, then you won't be able to auth to those hosts.")
        self._println(
            "This means lots of things will fail, like gclient sync.)")

        self._println()
        if not self._read_yn('Set up commonly used Gerrit hosts?',
                             default=True):
            self._println('Okay, skipping Gerrit host setup.')
            self._println(
                'You can re-run this command later or follow the instructions for manual configuration.'
            )
            self._print_manual_instructions()
            return

        hosts = [
            'android.googlesource.com',
            'aomedia.googlesource.com',
            'beto-core.googlesource.com',
            'boringssl.googlesource.com',
            'chromium.googlesource.com',
            'chrome-internal.googlesource.com',
            'dawn.googlesource.com',
            'pdfium.googlesource.com',
            'quiche.googlesource.com',
            'skia.googlesource.com',
            'swiftshader.googlesource.com',
            'webrtc.googlesource.com',
        ]

        self._println()
        self._println('We will set up auth for the following hosts:')
        for host in hosts:
            self._println(f'- {host}')
        self._println()
        self._read_enter()

        used_oauth = False
        for host in hosts:
            self._println()
            self._println(f'Checking authentication config for {host}')
            parts = urllib.parse.urlsplit(f'https://{host}/')
            info = self._configure_host(parts, global_email, scope='global')
            if info.method == _ConfigMethod.OAUTH:
                used_oauth = True
        if used_oauth:
            self._print_oauth_instructions()

        self._println()
        self._println(
            "If you need to set up any uncommonly used hosts that we didn't set up above,"
        )
        self._println('you can set them up manually.')
        self._print_manual_instructions()

    def _run_inside_repo(self) -> None:
        global_email = self._check_global_email()
        info = self._configure_repo(global_email=global_email)
        # This repo should be confirmed to be Gerrit by this point.
        assert info is not None
        if info.method == _ConfigMethod.OAUTH:
            self._print_oauth_instructions()

        dirs = list(scm.GIT.ListSubmodules(os.getcwd()))
        if dirs:
            self._println()
            self._println('This repository appears to have submodules.')
            self._println(
                'These may use different Gerrit hosts and need to be configured separately.'
            )
            self._println_action(
                "If you haven't yet, run `git cl creds-check --global` to configure common Gerrit hosts."
            )

    # Configuring Git for Gerrit auth

    def _configure_repo(self, *, global_email: str) -> _ConfigInfo | None:
        """Configure current Git repo for Gerrit auth.

        Returns None if current Git repo doesn't have Gerrit remote.
        """
        self._println()
        self._println(f'Configuring Gerrit auth for {os.getcwd()}')

        remote_url = self._remote_url_func()
        if not _is_gerrit_url(remote_url):
            self._println(
                f"Current repo remote {remote_url} doesn't look like Gerrit, so skipping"
            )
            return None
        self._println(f"Repo remote is {remote_url}")

        local_email = self._check_local_email()
        email = global_email
        scope = 'global'
        if local_email and local_email != global_email:
            self._println()
            self._println(
                'You have different emails configured locally vs globally.')
            self._println(
                'We will configure Gerrit authentication for your local repo only.'
            )
            email = local_email
            scope = 'local'
        self._println()
        parts = urllib.parse.urlsplit(remote_url)
        return self._configure_host(parts, email, scope=scope)

    def _configure_host(self, parts: urllib.parse.SplitResult, email: str, *,
                        scope: scm.GitConfigScope) -> _ConfigInfo:
        """Configure auth for one Gerrit host."""
        use_sso = self._check_use_sso(parts, email)
        if use_sso:
            self._configure_sso(parts, scope=scope)
            return _ConfigInfo(method=_ConfigMethod.SSO)
        self._configure_oauth(parts, scope=scope)
        return _ConfigInfo(method=_ConfigMethod.OAUTH)

    def _configure_sso(self, parts: urllib.parse.SplitResult, *,
                       scope: scm.GitConfigScope) -> None:
        if parts.scheme == 'sso':
            self._println(f'Your remote URL {parts.geturl()} already uses SSO')
        else:
            self._set_sso_rewrite(parts, scope=scope)
        self._clear_url_rewrite_override(parts, scope=scope)

    def _configure_oauth(self, parts: urllib.parse.SplitResult, *,
                         scope: scm.GitConfigScope) -> None:
        self._set_oauth_helper(parts, scope=scope)
        if scope == 'local':
            # Override a potential SSO rewrite set in the global config
            self._set_url_rewrite_override(parts, scope=scope)
        self._clear_sso_rewrite(parts, scope=scope)

    # Fixing gitcookies

    def _fix_gitcookies(self):
        sit = self._check_gitcookies()
        if not sit.cookiefile:
            self._println(
                "You don't have a cookie file configured in Git (good).")
            if sit.gitcookies_exists:
                self._println(
                    'However, you have a .gitcookies file (which is not configured for Git).'
                )
                self._println(
                    'This won"t affect Git authentication, but may cause issues for'
                )
                self._println('other Gerrit operations in depot_tools.')
                if self._read_yn(
                        'Shall we move your .gitcookies file (to a backup location)?',
                        default=True):
                    self._move_file(self._gitcookies())
                    self._println(
                        'Note that some tools may still use the (legacy) .gitcookies file.'
                    )
                    self._println(
                        'If you encounter an issue, please report it.')
            return

        self._println('You appear to have a cookie file configured for Git.')
        self._println(f'http.cookiefile={sit.cookiefile!r}')

        if not sit.cookiefile_exists:
            self._println('However, this file does not exist.')
            self._println(
                'This will not affect anything, but we suggest removing the http.cookiefile from your Git config.'
            )
            if self._read_yn('Shall we remove it for you?', default=True):
                self._set_config('http.cookiefile', None, scope='global')
            return

        if sit.divergent_cookiefiles:
            self._println()
            self._println(
                'You also have a .gitcookies file, which is different from the cookefile in your Git config.'
            )
            self._println('We cannot handle this unusual case right now.')
            raise _ConfigError('unusual gitcookie setup')

        with open(sit.cookiefile, 'r') as f:
            info = _parse_cookiefile(f)

        if not info.contains_gerrit:
            self._println(
                "The cookie file doesn't appear to contain any Gerrit cookies,")
            self._println('so we will ignore it.')
            return

        if info.contains_nongerrit:
            self._println(
                'The cookie file contains Gerrit cookies and non-Gerrit cookies.'
            )
            self._println(
                'Cookie auth is deprecated, and these cookies may interfere with Gerrit authentication.'
            )
            self._println(
                "Since you have non-Gerrit cookies too, we won't try to fix it for you."
            )
            self._println_action(
                f'Please remove the Gerrit cookies (lines containing .googlesource.com) from {sit.cookiefile}'
            )
            return

        self._println('The cookie file contains Gerrit cookies.')
        self._println(
            'Cookie auth is deprecated, and these cookies may interfere with Gerrit authentication.'
        )
        if not self._read_yn(
                'Shall we move your cookie file (to a backup location)?',
                default=True):
            self._println(
                'Okay, we recommend that you move (or remove) it later to avoid issues.'
            )
            return

        self._move_file(sit.cookiefile)
        self._set_config('http.cookiefile', None, scope='global')

    # Self-contained checks for specific things

    def _check_gitcookies(self) -> _GitcookiesSituation:
        """Checks various things about the user's gitcookies situation."""
        gitcookies = self._gitcookies()
        gitcookies_exists = os.path.exists(gitcookies)
        cookiefile = scm.GIT.GetConfig(
            os.getcwd(), 'http.cookiefile', scope='global') or ''
        cookiefile_exists = os.path.exists(cookiefile)
        divergent_cookiefiles = gitcookies_exists and cookiefile_exists and not os.path.samefile(
            gitcookies, cookiefile)
        return _GitcookiesSituation(
            gitcookies_exists=gitcookies_exists,
            cookiefile=cookiefile,
            cookiefile_exists=cookiefile_exists,
            divergent_cookiefiles=divergent_cookiefiles,
        )

    def _check_global_email(self) -> str:
        """Checks and returns user's global Git email.

        Prompts the user to set it if it isn't set.
        """
        email = scm.GIT.GetConfig(os.getcwd(), 'user.email',
                                  scope='global') or ''
        if email:
            self._println(f'Your global Git email is: {email}')
            return email
        self._println()
        self._println(
            'You do not have an email configured in your global Git config.')
        if not self._read_yn('Do you want to set one now?', default=True):
            self._println('Will attempt to continue without a global email.')
            return ''
        name = scm.GIT.GetConfig(os.getcwd(), 'user.name', scope='global') or ''
        if not name:
            name = self._read_line('Enter your name (e.g., John Doe)',
                                   check=_check_nonempty)
            self._set_config('user.name', name, scope='global')
        email = self._read_line('Enter your email', check=_check_nonempty)
        self._set_config('user.email', email, scope='global')
        return email

    def _check_local_email(self) -> str:
        """Checks and returns the user's local Git email."""
        email = scm.GIT.GetConfig(os.getcwd(), 'user.email',
                                  scope='local') or ''
        if email:
            self._println(
                f'You have an email configured in your local repo: {email}')
        return email

    def _check_use_sso(self, parts: urllib.parse.SplitResult,
                       email: str) -> bool:
        """Checks whether SSO is needed for the given user and host."""
        if not self._check_sso_helper():
            return False
        host = _url_review_host(parts)
        self._println(f'Checking SSO requirement for {email!r} on {host}')
        self._println(
            '(Note that this check may require SSO; if you get an error,')
        self._println('you will need to login to SSO and re-run this command.)')
        result = gerrit_util.CheckShouldUseSSO(host, email)
        text = 'use' if result.status else 'not use'
        self._println(f'Decided we should {text} SSO for {email!r} on {host}')
        self._println(f'Reason: {result.reason}')
        self._println()
        return result.status

    def _check_sso_helper(self) -> bool:
        """Checks and returns whether SSO helper is available."""
        return bool(gerrit_util.ssoHelper.find_cmd())

    # Reused instruction printing

    def _print_manual_instructions(self) -> None:
        """Prints manual instructions for setting up auth."""
        self._println()
        self._println(
            'Instructions for manually configuring Gerrit authentication:')
        self._println(
            'https://commondatastorage.googleapis.com/chrome-infra-docs/flat/depot_tools/docs/html/depot_tools_gerrit_auth.html'
        )

    def _print_oauth_instructions(self) -> None:
        """Prints instructions for setting up OAuth helper."""
        self._println()
        self._println('We have configured Git to use an OAuth helper.')
        self._println('The OAuth helper requires its own login.')
        self._println_action(
            "If you haven't yet, run `git credential-luci login` using the same email as Git."
        )
        self._println(
            "(If you have already done this, you don't need to do it again.)")
        self._println(
            '(However, if you changed your email, you should do this again')
        self._println("to ensure you're using the right account.)")

    # Low level Git config manipulation

    def _set_oauth_helper(self, parts: urllib.parse.SplitResult, *,
                          scope: scm.GitConfigScope) -> None:
        cred_key = _creds_helper_key(parts)
        self._set_config(cred_key, '', modify_all=True, scope=scope)
        self._set_config(cred_key, 'luci', append=True, scope=scope)

    def _set_sso_rewrite(self, parts: urllib.parse.SplitResult, *,
                         scope: scm.GitConfigScope) -> None:
        sso_key = _sso_rewrite_key(parts)
        self._set_config(sso_key,
                         _url_root_url(parts),
                         modify_all=True,
                         scope=scope)

    def _clear_sso_rewrite(self, parts: urllib.parse.SplitResult, *,
                           scope: scm.GitConfigScope) -> None:
        sso_key = _sso_rewrite_key(parts)
        self._set_config(sso_key, None, modify_all=True, scope=scope)

    def _set_url_rewrite_override(self, parts: urllib.parse.SplitResult, *,
                                  scope: scm.GitConfigScope) -> None:
        url_key = _url_rewrite_key(parts)
        self._set_config(url_key, parts.geturl(), modify_all=True, scope=scope)

    def _clear_url_rewrite_override(self, parts: urllib.parse.SplitResult, *,
                                    scope: scm.GitConfigScope) -> None:
        url_key = _url_rewrite_key(parts)
        self._set_config(url_key, None, scope=scope, modify_all=True)

    def _set_config(self,
                    key: str,
                    value: str | None,
                    *,
                    scope: scm.GitConfigScope,
                    modify_all: bool = False,
                    append: bool = False) -> None:
        scope_msg = f'In your {scope} Git config,'
        if append:
            assert value is not None
            self._println_notify(
                f'{scope_msg} we appended {key}={value!r} to existing values')
        else:
            if value is None:
                action = f"we cleared {'all values' if modify_all else 'the value'} for {key}"
            else:
                action = f'we set {key}={value!r}'
                if modify_all:
                    action += ', replacing any existing values'
            self._println_notify(f'{scope_msg} {action}')

        scm.GIT.SetConfig(os.getcwd(),
                          key,
                          value,
                          scope=scope,
                          modify_all=modify_all,
                          append=append)

    # Low level misc helpers

    def _move_file(self, path: str) -> None:
        """Move file to a backup path."""
        backup = f'{path}.bak'
        n = 1
        while os.path.exists(backup):
            n += 1
            backup = f'{path}.bak{n}'
        os.rename(path, backup)
        self._println_notify(f'Moved {path!r} to {backup!r}')

    @contextlib.contextmanager
    def _handle_config_errors(self):
        try:
            yield None
        except _ConfigError as e:
            self._println(f'ConfigError: {e!s}')

    def _print_actions_for_user(self) -> None:
        """Print manual actions requested from user.

        Aggregates any actions printed throughout the wizard run so it's
        easier for the user.
        """
        if not self._user_actions:
            return
        self._println()
        self._println(
            "However, there are some manual actions that are suggested")
        self._println("(you don't have to re-run this command afterward):")
        for s in self._user_actions:
            self._println(f'- {s}')

    def _println_action(self, s: str) -> None:
        """Print a notification about a manual action request from user.

        Also queues up the action for _print_actions_for_user.
        """
        self._println(f'!!! {s}')
        self._user_actions.append(s)

    def _println_notify(self, s: str) -> None:
        """Print a notification about a change we made."""
        self._println(f'>>> {s}')

    def _println(self, s: str = '') -> None:
        self._ui.write(s)
        self._ui.write('\n')

    def _read_yn(self, prompt: str, *, default: bool | None = None) -> bool:
        ret = self._ui.read_yn(prompt, default=default)
        self._ui.write('\n')
        return ret

    def _read_line(self,
                   prompt: str,
                   *,
                   check: _InputChecker = _check_any) -> str:
        ret = self._ui.read_line(prompt, check=check)
        self._ui.write('\n')
        return ret

    def _read_enter(self) -> None:
        self._ui.read_enter()
        self._ui.write('\n')

    @staticmethod
    def _gitcookies() -> str:
        """Path to user's gitcookies.

        Can be mocked for testing.
        """
        return os.path.expanduser('~/.gitcookies')


class _CookiefileInfo(NamedTuple):
    """Result for _parse_cookiefile."""
    contains_gerrit: bool
    contains_nongerrit: bool


def _parse_cookiefile(f: TextIO) -> _CookiefileInfo:
    """Checks cookie file contents.

    Used to guide auth configuration.
    """
    contains_gerrit = False
    contains_nongerrit = False
    for line in f:
        if line.lstrip().startswith('#'):
            continue
        if not line.strip():
            continue
        if '.googlesource.com' in line:
            contains_gerrit = True
        else:
            contains_nongerrit = True
    return _CookiefileInfo(
        contains_gerrit=contains_gerrit,
        contains_nongerrit=contains_nongerrit,
    )


def _is_gerrit_url(url: str) -> bool:
    """Checks if URL is for a Gerrit host."""
    if not url:
        return False
    parts = urllib.parse.urlsplit(url)
    if parts.netloc.endswith('.googlesource.com') or parts.netloc.endswith(
            '.git.corp.google.com'):
        return True
    return False


def _creds_helper_key(parts: urllib.parse.SplitResult) -> str:
    """Return Git config key for credential helpers."""
    return f'credential.{_url_host_url(parts)}.helper'


def _sso_rewrite_key(parts: urllib.parse.SplitResult) -> str:
    """Return Git config key for SSO URL rewrites."""
    return f'url.sso://{_url_shortname(parts)}/.insteadOf'


def _url_rewrite_key(parts: urllib.parse.SplitResult) -> str:
    """Return Git config key for rewriting the full URL."""
    return f'url.{parts.geturl()}.insteadOf'


def _url_review_host(parts: urllib.parse.SplitResult) -> str:
    """Format URL as Gerrit review host.

    Example: chromium-review.googlesource.com
    """
    return f'{_url_shortname(parts)}-review.googlesource.com'


def _url_shortname(parts: urllib.parse.SplitResult) -> str:
    """Format URL as Gerrit host shortname.

    Example: chromium
    """
    name: str = parts.netloc.split('.')[0]
    if name.endswith('-review'):
        name = name[:-len('-review')]
    return name


def _url_host_url(parts: urllib.parse.SplitResult) -> str:
    """Format URL with host only (no path).

    Example: https://chromium.googlesource.com
    Example: https://chromium-review.googlesource.com
    """
    return parts._replace(path='', query='', fragment='').geturl()


def _url_root_url(parts: urllib.parse.SplitResult) -> str:
    """Format URL with root path.

    Example: https://chromium.googlesource.com/
    Example: https://chromium-review.googlesource.com/
    """
    return parts._replace(path='/', query='', fragment='').geturl()