[git_cl] Make EnsureCanUploadPatchset work for all auth methods.

Previously EnsureCanUploadPatchset only had a working implementation
for the CookiesAuthenticator, relying on being able to parse the
user name out of the .gitcookies file.

Additionally, the previous implementation assumed that you would
always authenticate as your primary Gerrit account OR you had
a matching `user.email` gitconfig entry, even though neither of these
is a strict requirement for the upload to work.

The new implementation still short-circuits if issue_owner matches
the configured user.email, but other than this it just asks Gerrit
what the full list of linked emails is for the currently authenticated
account.

The new approach is not only correct, but will now work for all auth
schemes in exactly the same way.

When the accounts do mismatch, you will now see output like:

```
WARNING: Change 5590262 is owned by iannucci@chromium.org, but Gerrit knows you as:
  * user@example.org
  * other.user@example.com
  * primary@real.example.com (preferred)
Uploading may fail due to lack of permissions.
```

R=ayatane@chromium.org, yiwzhang@google.com

Bug: 336351842
Change-Id: I89c1b121c9110e00d1348884aaf025fc783542d0
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/tools/depot_tools/+/5590262
Commit-Queue: Robbie Iannucci <iannucci@chromium.org>
Reviewed-by: Yiwei Zhang <yiwzhang@google.com>
Auto-Submit: Robbie Iannucci <iannucci@chromium.org>

gerrit_util.py

@@ -30,7 +30,7 @@ from dataclasses import dataclass
 from io import StringIO
 from multiprocessing.pool import ThreadPool
 from typing import Any, Container, Dict, List, Optional
-from typing import Tuple, Type, TypedDict, cast
+from typing import Tuple, TypedDict, cast
 
 import httplib2
 import httplib2.socks
@@ -1640,6 +1640,32 @@ def GetAccountDetails(host, account_id='self'):
     return ReadHttpJsonResponse(conn, accept_statuses=[200, 404])
 
 
+class EmailRecord(TypedDict):
+    email: str
+    preferred: bool  # This should be NotRequired[bool] in 3.11+
+
+
+def GetAccountEmails(host, account_id='self') -> Optional[List[EmailRecord]]:
+    """Returns all emails for this account, and an indication of which of these
+    is preferred.
+
+    If account_id is not given, uses magic value 'self' which corresponds to
+    whichever account user is authenticating as.
+
+    Requires Modify Account permission to view emails other than 'self'.
+
+    Documentation:
+    https://gerrit-review.googlesource.com/Documentation/rest-api-accounts.html#list-account-emails
+
+    Returns None if account is not found (i.e. Gerrit returned 404).
+    """
+    conn = CreateHttpConn(host, '/accounts/%s/emails' % account_id)
+    resp = ReadHttpJsonResponse(conn, accept_statuses=[200, 404])
+    if resp is None:
+        return None
+    return cast(List[EmailRecord], resp)
+
+
 def ValidAccounts(host, accounts, max_threads=10):
     """Returns a mapping from valid account to its details.
 

git_cl.py

@@ -2320,7 +2320,8 @@ class Changelist(object):
             DieWithError(msg)
 
     def EnsureCanUploadPatchset(self, force):
-        if not self.GetIssue():
+        issue = self.GetIssue()
+        if not issue:
             return
 
         status = self._GetChangeDetail()['status']
@@ -2338,30 +2339,41 @@ class Changelist(object):
             self.SetIssue()
             return
 
-        # TODO(vadimsh): For some reason the chunk of code below was skipped if
+        # Check to see if the currently authenticated account is the issue
-        # 'is_gce' is True. I'm just refactoring it to be 'skip if not cookies'.
+        # owner.
-        # Apparently this check is not very important? Otherwise get_auth_email
+
-        # could have been added to other implementations of Authenticator.
+        # first, grab the issue owner email
-        cookies_auth = gerrit_util.Authenticator.get()
+        owner = self.GetIssueOwner()
-        if not isinstance(cookies_auth, gerrit_util.CookiesAuthenticator):
+
+        # do a quick check to see if this matches the local git config's
+        # configured email.
+        git_config_email = scm.GIT.GetConfig(settings.GetRoot(), 'user.email')
+        if git_config_email == owner:
+            # Good enough - Gerrit will reject this if the user is doing funny things
+            # with user.email.
             return
 
-        cookies_user = cookies_auth.get_auth_email(self.GetGerritHost())
+        # However, the user may have linked accounts in Gerrit, so pull up the
-        if self.GetIssueOwner() == cookies_user:
+        # list of all known emails for the currently authenticated account.
+        emails = gerrit_util.GetAccountEmails(self.GetGerritHost(), 'self')
+        if not emails:
+            print('WARNING: Gerrit does not have a record for your account.')
+            print('Please browse to https://{self.GetGerritHost()} and log in.')
             return
-        logging.debug('change %s owner is %s, cookies user is %s',
+
-                      self.GetIssue(), self.GetIssueOwner(), cookies_user)
+        # If the issue owner is one of the emails for the currently
-        # Maybe user has linked accounts or something like that,
+        # authenticated account, Gerrit will accept the upload.
-        # so ask what Gerrit thinks of this user.
+        if any(owner == e['email'] for e in emails):
-        details = gerrit_util.GetAccountDetails(self.GetGerritHost(), 'self')
-        if details['email'] == self.GetIssueOwner():
             return
+
         if not force:
             print(
-                'WARNING: Change %s is owned by %s, but you authenticate to Gerrit '
+                f'WARNING: Change {issue} is owned by {owner}, but Gerrit knows you as:'
-                'as %s.\n'
+            )
-                'Uploading may fail due to lack of permissions.' %
+            for email in emails:
-                (self.GetIssue(), self.GetIssueOwner(), details['email']))
+                tag = ' (preferred)' if email.get('preferred') else ''
+                print(f'  * {email["email"]}{tag}')
+            print('Uploading may fail due to lack of permissions.')
             confirm_or_exit(action='upload')
 
     def GetStatus(self):