feat: support self-signed JWT flow for service accounts

gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/base.py.j2

@@ -2,10 +2,12 @@
 
 {% block content %}
 import abc
-import typing
+from typing import Awaitable, Callable, Dict, Optional, Sequence, Union
+import packaging.version
 import pkg_resources
 
 from google import auth  # type: ignore
+import google.api_core  # type: ignore
 from google.api_core import exceptions  # type: ignore
 from google.api_core import gapic_v1    # type: ignore
 from google.api_core import retry as retries  # type: ignore
@@ -34,6 +36,18 @@ try:
 except pkg_resources.DistributionNotFound:
     DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo()
 
+try:
+    # google.auth.__version__ was added in 1.26.0
+    _GOOGLE_AUTH_VERSION = auth.__version__
+except AttributeError:
+    try:  # try pkg_resources if it is available
+        _GOOGLE_AUTH_VERSION = pkg_resources.get_distribution("google-auth").version
+    except pkg_resources.DistributionNotFound:  # pragma: NO COVER
+        _GOOGLE_AUTH_VERSION = None
+
+_API_CORE_VERSION = google.api_core.__version__
+
+
 class {{ service.name }}Transport(abc.ABC):
     """Abstract transport class for {{ service.name }}."""
 
@@ -43,13 +57,15 @@ class {{ service.name }}Transport(abc.ABC):
         {%- endfor %}
     )
 
+    DEFAULT_HOST: str = {% if service.host %}'{{ service.host }}'{% else %}{{ '' }}{% endif %}
+
     def __init__(
             self, *,
-            host: str{% if service.host %} = '{{ service.host }}'{% endif %},
+            host: str = DEFAULT_HOST,
             credentials: credentials.Credentials = None,
-            credentials_file: typing.Optional[str] = None,
-            scopes: typing.Optional[typing.Sequence[str]] = AUTH_SCOPES,
-            quota_project_id: typing.Optional[str] = None,
+            credentials_file: Optional[str] = None,
+            scopes: Optional[Sequence[str]] = None,
+            quota_project_id: Optional[str] = None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
             **kwargs,       
             ) -> None:
@@ -66,7 +82,7 @@ class {{ service.name }}Transport(abc.ABC):
             credentials_file (Optional[str]): A file with credentials that can
                 be loaded with :func:`google.auth.load_credentials_from_file`.
                 This argument is mutually exclusive with credentials.
-            scope (Optional[Sequence[str]]): A list of scopes.
+            scopes (Optional[Sequence[str]]): A list of scopes.
             quota_project_id (Optional[str]): An optional project to use for billing
                 and quota.
             client_info (google.api_core.gapic_v1.client_info.ClientInfo):	
@@ -80,8 +96,10 @@ class {{ service.name }}Transport(abc.ABC):
             host += ':443'
         self._host = host
 
+        scopes_kwargs = self._get_scopes_kwargs(self._host, scopes)
+
         # Save the scopes.
-        self._scopes = scopes or self.AUTH_SCOPES
+        self._scopes = scopes_kwargs["scopes"]
 
         # If no credentials are provided, then determine the appropriate
         # defaults.
@@ -91,17 +109,73 @@ class {{ service.name }}Transport(abc.ABC):
         if credentials_file is not None:
             credentials, _ = auth.load_credentials_from_file(
                                 credentials_file,
-                                scopes=self._scopes,
+                                **scopes_kwargs,
                                 quota_project_id=quota_project_id
                             )
 
         elif credentials is None:
-            credentials, _ = auth.default(scopes=self._scopes, quota_project_id=quota_project_id)
+            credentials, _ = auth.default(**scopes_kwargs, quota_project_id=quota_project_id)
 
         # Save the credentials.
         self._credentials = credentials
 
 
+    @classmethod
+    def _get_user_scopes(cls, host: str, scopes: Optional[Sequence[str]]) -> Optional[Sequence[str]]:
+        if scopes is None:
+            # If a custom API endpoint is set, set user scopes to ensure the auth
+            # library does not used the self-signed JWT flow for service
+            # accounts.
+            if host.split(":")[0] != cls.DEFAULT_HOST.split(":")[0]:
+                scopes = cls.AUTH_SCOPES
+
+        return scopes
+
+
+    # TODO(busunkim): These two class methods are in the base transport 
+    # to avoid duplicating code across the transport classes. These functions
+    # should be deleted once the minimum required versions of google-api-core
+    # and google-auth are increased.
+
+    # TODO: Remove this function once google-auth >= 1.25.0 is required
+    @classmethod
+    def _get_scopes_kwargs(cls, host: str, scopes: Optional[Sequence[str]]) -> Dict[str, Optional[Sequence[str]]]:
+        """Returns scopes kwargs to pass to google-auth methods depending on the google-auth version"""
+
+        scopes = cls._get_user_scopes(host, scopes)
+        scopes_kwargs = {}
+
+        if _GOOGLE_AUTH_VERSION and (
+            packaging.version.parse(_GOOGLE_AUTH_VERSION)
+            >= packaging.version.parse("1.25.0")
+        ):
+            scopes_kwargs = {"scopes": scopes, "default_scopes": cls.AUTH_SCOPES}
+        else:
+            scopes_kwargs = {"scopes": scopes or cls.AUTH_SCOPES}
+
+        return scopes_kwargs
+
+    # TODO: Remove this function once google-api-core >= 1.26.0 is required
+    @classmethod
+    def _get_self_signed_jwt_kwargs(cls, host: str, scopes: Optional[Sequence[str]]) -> Dict[str, Union[Optional[Sequence[str]], str]]:
+        """Returns kwargs to pass to grpc_helpers.create_channel depending on the google-api-core version"""
+        scopes = cls._get_user_scopes(host, scopes)
+
+        self_signed_jwt_kwargs: Dict[str, Union[Optional[Sequence[str]], str]] = {}
+
+        if _API_CORE_VERSION and (
+            packaging.version.parse(_API_CORE_VERSION)
+            >= packaging.version.parse("1.26.0")
+        ):
+            self_signed_jwt_kwargs["default_scopes"] = cls.AUTH_SCOPES
+            self_signed_jwt_kwargs["scopes"] = scopes
+            self_signed_jwt_kwargs["default_host"] = cls.DEFAULT_HOST
+        else:
+            self_signed_jwt_kwargs["scopes"] = scopes or cls.AUTH_SCOPES
+
+        return self_signed_jwt_kwargs
+
+
     def _prep_wrapped_messages(self, client_info):
         # Precompute the wrapped methods.
         self._wrapped_methods = {
@@ -137,11 +211,11 @@ class {{ service.name }}Transport(abc.ABC):
     {%- for method in service.methods.values() %}
 
     @property
-    def {{ method.name|snake_case }}(self) -> typing.Callable[
+    def {{ method.name|snake_case }}(self) -> Callable[
             [{{ method.input.ident }}],
-            typing.Union[
+            Union[
                 {{ method.output.ident }},
-                typing.Awaitable[{{ method.output.ident }}]
+                Awaitable[{{ method.output.ident }}]
             ]]:
         raise NotImplementedError()
     {%- endfor %}
@@ -151,29 +225,29 @@ class {{ service.name }}Transport(abc.ABC):
     @property
     def set_iam_policy(
         self,
-    ) -> typing.Callable[
+    ) -> Callable[
         [iam_policy.SetIamPolicyRequest],
-        typing.Union[policy.Policy, typing.Awaitable[policy.Policy]],
+        Union[policy.Policy, Awaitable[policy.Policy]],
     ]:
         raise NotImplementedError()
 
     @property
     def get_iam_policy(
         self,
-    ) -> typing.Callable[
+    ) -> Callable[
         [iam_policy.GetIamPolicyRequest],
-        typing.Union[policy.Policy, typing.Awaitable[policy.Policy]],
+        Union[policy.Policy, Awaitable[policy.Policy]],
     ]:
         raise NotImplementedError()
 
     @property
     def test_iam_permissions(
         self,
-    ) -> typing.Callable[
+    ) -> Callable[
         [iam_policy.TestIamPermissionsRequest],
-        typing.Union[
+        Union[
             iam_policy.TestIamPermissionsResponse,
-            typing.Awaitable[iam_policy.TestIamPermissionsResponse],
+            Awaitable[iam_policy.TestIamPermissionsResponse],
         ],
     ]:
         raise NotImplementedError()

gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc.py.j2

@@ -2,7 +2,7 @@
 
 {% block content %}
 import warnings
-from typing import Callable, Dict, Optional, Sequence, Tuple
+from typing import Callable, Dict, Optional, Sequence, Tuple, Union
 
 from google.api_core import grpc_helpers   # type: ignore
 {%- if service.has_lro %}
@@ -202,13 +202,15 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
             google.api_core.exceptions.DuplicateCredentialArgs: If both ``credentials``
               and ``credentials_file`` are passed.
         """
-        scopes = scopes or cls.AUTH_SCOPES
+
+        self_signed_jwt_kwargs = cls._get_self_signed_jwt_kwargs(host, scopes)
+
         return grpc_helpers.create_channel(
             host,
             credentials=credentials,
             credentials_file=credentials_file,
-            scopes=scopes,
             quota_project_id=quota_project_id,
+            **self_signed_jwt_kwargs,
             **kwargs
         )
 

gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2

@@ -2,7 +2,7 @@
 
 {% block content %}
 import warnings
-from typing import Awaitable, Callable, Dict, Optional, Sequence, Tuple
+from typing import Awaitable, Callable, Dict, Optional, Sequence, Tuple, Union
 
 from google.api_core import gapic_v1                   # type: ignore
 from google.api_core import grpc_helpers_async         # type: ignore
@@ -12,6 +12,7 @@ from google.api_core import operations_v1              # type: ignore
 from google import auth                                # type: ignore
 from google.auth import credentials                    # type: ignore
 from google.auth.transport.grpc import SslCredentials  # type: ignore
+import packaging.version
 
 import grpc                        # type: ignore
 from grpc.experimental import aio  # type: ignore
@@ -75,13 +76,15 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
         Returns:
             aio.Channel: A gRPC AsyncIO channel object.
         """
-        scopes = scopes or cls.AUTH_SCOPES
+
+        self_signed_jwt_kwargs = cls._get_self_signed_jwt_kwargs(host, scopes)
+
         return grpc_helpers_async.create_channel(
             host,
             credentials=credentials,
             credentials_file=credentials_file,
-            scopes=scopes,
             quota_project_id=quota_project_id,
+            **self_signed_jwt_kwargs,
             **kwargs
         )
 
@@ -163,7 +166,6 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
             # If a channel was explicitly provided, set it.
             self._grpc_channel = channel
             self._ssl_channel_credentials = None
-
         else:
             if api_mtls_endpoint:
                 host = api_mtls_endpoint

gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2

@@ -81,12 +81,14 @@ class {{ service.name }}RestTransport({{ service.name }}Transport):
         """
         # Run the base constructor
         # TODO(yon-mg): resolve other ctor params i.e. scopes, quota, etc.
+        # TODO: When custom host (api_endpoint) is set, `scopes` must *also* be set on the
+        # credentials object 
         super().__init__(
             host=host,
             credentials=credentials,
             client_info=client_info,
         )
-        self._session = AuthorizedSession(self._credentials)
+        self._session = AuthorizedSession(self._credentials, default_host=self.DEFAULT_HOST)
         {%- if service.has_lro %}
         self._operations_client = None
         {%- endif %}
@@ -106,11 +108,14 @@ class {{ service.name }}RestTransport({{ service.name }}Transport):
         # Sanity check: Only create a new client if we do not already have one.
         if self._operations_client is None:
             from google.api_core import grpc_helpers
+
+            self_signed_jwt_kwargs = cls._get_self_signed_jwt_kwargs(self._host, self._scopes)
+
             self._operations_client = operations_v1.OperationsClient(
                 grpc_helpers.create_channel(
                     self._host,
                     credentials=self._credentials,
-                    scopes=self.AUTH_SCOPES,
+                    **self_signed_jwt_kwargs,
                     options=[
                         ("grpc.max_send_message_length", -1),
                         ("grpc.max_receive_message_length", -1),

gapic/templates/.coveragerc.j2

@@ -2,7 +2,6 @@
 branch = True
 
 [report]
-fail_under = 100
 show_missing = True
 omit =
     {{ api.naming.module_namespace|join("/") }}/{{ api.naming.module_name }}/__init__.py

gapic/templates/noxfile.py.j2

@@ -2,10 +2,28 @@
 
 {% block content %}
 import os
+import pathlib
 import shutil
+import subprocess
+import sys
+
 
 import nox  # type: ignore
 
+CURRENT_DIRECTORY = pathlib.Path(__file__).parent.absolute()
+
+LOWER_BOUND_CONSTRAINTS_FILE = CURRENT_DIRECTORY / "constraints.txt"
+PACKAGE_NAME = subprocess.check_output([sys.executable, "setup.py", "--name"], encoding="utf-8")
+
+
+nox.sessions = [
+    "unit",
+    "cover",
+    "mypy",
+    "check_lower_bounds"
+    # exclude update_lower_bounds from default
+    "docs",
+]
 
 @nox.session(python=['3.6', '3.7', '3.8', '3.9'])
 def unit(session):
@@ -25,6 +43,18 @@ def unit(session):
     )
 
 
+@nox.session(python='3.7')
+def cover(session):
+    """Run the final coverage report.
+    This outputs the coverage report aggregating coverage from the unit
+    test runs (not system test runs), and then erases coverage data.
+    """
+    session.install("coverage", "pytest-cov")
+    session.run("coverage", "report", "--show-missing", "--fail-under=100")
+
+    session.run("coverage", "erase")
+
+
 @nox.session(python=['3.6', '3.7'])
 def mypy(session):
     """Run the type checker."""
@@ -40,6 +70,38 @@ def mypy(session):
         {%- endif %}
     )
 
+
+@nox.session
+def update_lower_bounds(session):
+    """Update lower bounds in constraints.txt to match setup.py"""
+    session.install('google-cloud-testutils')
+    session.install('.')
+
+    session.run(
+        'lower-bound-checker',
+        'update',
+        '--package-name',
+        PACKAGE_NAME,
+        '--constraints-file',
+        str(LOWER_BOUND_CONSTRAINTS_FILE),
+    )
+
+
+@nox.session
+def check_lower_bounds(session):
+    """Check lower bounds in setup.py are reflected in constraints file"""
+    session.install('google-cloud-testutils')
+    session.install('.')
+
+    session.run(
+        'lower-bound-checker',
+        'check',
+        '--package-name',
+        PACKAGE_NAME,
+        '--constraints-file',
+        str(LOWER_BOUND_CONSTRAINTS_FILE),
+    )
+
 @nox.session(python='3.6')
 def docs(session):
     """Build the docs for this library."""