codekingpro/portable-devtools

1# Copyright 2020 Google LLC2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#      http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14 15"""Google ID Token helpers.16 17Provides support for verifying `OpenID Connect ID Tokens`_, especially ones18generated by Google infrastructure.19 20To parse and verify an ID Token issued by Google's OAuth 2.0 authorization21server use :func:`verify_oauth2_token`. To verify an ID Token issued by22Firebase, use :func:`verify_firebase_token`.23 24A general purpose ID Token verifier is available as :func:`verify_token`.25 26Example::27 28    from google.oauth2 import _id_token_async29    from google.auth.transport import aiohttp_requests30 31    request = aiohttp_requests.Request()32 33    id_info = await _id_token_async.verify_oauth2_token(34        token, request, 'my-client-id.example.com')35 36    if id_info['iss'] != 'https://accounts.google.com':37        raise ValueError('Wrong issuer.')38 39    userid = id_info['sub']40 41By default, this will re-fetch certificates for each verification. Because42Google's public keys are only changed infrequently (on the order of once per43day), you may wish to take advantage of caching to reduce latency and the44potential for network errors. This can be accomplished using an external45library like `CacheControl`_ to create a cache-aware46:class:`google.auth.transport.Request`::47 48    import cachecontrol49    import google.auth.transport.requests50    import requests51 52    session = requests.session()53    cached_session = cachecontrol.CacheControl(session)54    request = google.auth.transport.requests.Request(session=cached_session)55 56.. _OpenID Connect ID Token:57    http://openid.net/specs/openid-connect-core-1_0.html#IDToken58.. _CacheControl: https://cachecontrol.readthedocs.io59"""60 61import http.client as http_client62import json63import os64 65from google.auth import environment_vars66from google.auth import exceptions67from google.auth import jwt68from google.auth.transport import requests69from google.oauth2 import id_token as sync_id_token70 71 72async def _fetch_certs(request, certs_url):73    """Fetches certificates.74 75    Google-style cerificate endpoints return JSON in the format of76    ``{'key id': 'x509 certificate'}``.77 78    Args:79        request (google.auth.transport.Request): The object used to make80            HTTP requests. This must be an aiohttp request.81        certs_url (str): The certificate endpoint URL.82 83    Returns:84        Mapping[str, str]: A mapping of public key ID to x.509 certificate85            data.86    """87    response = await request(certs_url, method="GET")88 89    if response.status != http_client.OK:90        raise exceptions.TransportError(91            "Could not fetch certificates at {}".format(certs_url)92        )93 94    data = await response.content()95 96    return json.loads(data)97 98 99async def verify_token(100    id_token,101    request,102    audience=None,103    certs_url=sync_id_token._GOOGLE_OAUTH2_CERTS_URL,104    clock_skew_in_seconds=0,105):106    """Verifies an ID token and returns the decoded token.107 108    Args:109        id_token (Union[str, bytes]): The encoded token.110        request (google.auth.transport.Request): The object used to make111            HTTP requests. This must be an aiohttp request.112        audience (str): The audience that this token is intended for. If None113            then the audience is not verified.114        certs_url (str): The URL that specifies the certificates to use to115            verify the token. This URL should return JSON in the format of116            ``{'key id': 'x509 certificate'}``.117        clock_skew_in_seconds (int): The clock skew used for `iat` and `exp`118            validation.119 120    Returns:121        Mapping[str, Any]: The decoded token.122    """123    certs = await _fetch_certs(request, certs_url)124 125    return jwt.decode(126        id_token,127        certs=certs,128        audience=audience,129        clock_skew_in_seconds=clock_skew_in_seconds,130    )131 132 133async def verify_oauth2_token(134    id_token, request, audience=None, clock_skew_in_seconds=0135):136    """Verifies an ID Token issued by Google's OAuth 2.0 authorization server.137 138    Args:139        id_token (Union[str, bytes]): The encoded token.140        request (google.auth.transport.Request): The object used to make141            HTTP requests. This must be an aiohttp request.142        audience (str): The audience that this token is intended for. This is143            typically your application's OAuth 2.0 client ID. If None then the144            audience is not verified.145        clock_skew_in_seconds (int): The clock skew used for `iat` and `exp`146            validation.147 148    Returns:149        Mapping[str, Any]: The decoded token.150 151    Raises:152        exceptions.GoogleAuthError: If the issuer is invalid.153    """154    idinfo = await verify_token(155        id_token,156        request,157        audience=audience,158        certs_url=sync_id_token._GOOGLE_OAUTH2_CERTS_URL,159        clock_skew_in_seconds=clock_skew_in_seconds,160    )161 162    if idinfo["iss"] not in sync_id_token._GOOGLE_ISSUERS:163        raise exceptions.GoogleAuthError(164            "Wrong issuer. 'iss' should be one of the following: {}".format(165                sync_id_token._GOOGLE_ISSUERS166            )167        )168 169    return idinfo170 171 172async def verify_firebase_token(173    id_token, request, audience=None, clock_skew_in_seconds=0174):175    """Verifies an ID Token issued by Firebase Authentication.176 177    Args:178        id_token (Union[str, bytes]): The encoded token.179        request (google.auth.transport.Request): The object used to make180            HTTP requests. This must be an aiohttp request.181        audience (str): The audience that this token is intended for. This is182            typically your Firebase application ID. If None then the audience183            is not verified.184        clock_skew_in_seconds (int): The clock skew used for `iat` and `exp`185            validation.186 187    Returns:188        Mapping[str, Any]: The decoded token.189    """190    return await verify_token(191        id_token,192        request,193        audience=audience,194        certs_url=sync_id_token._GOOGLE_APIS_CERTS_URL,195        clock_skew_in_seconds=clock_skew_in_seconds,196    )197 198 199async def fetch_id_token(request, audience):200    """Fetch the ID Token from the current environment.201 202    This function acquires ID token from the environment in the following order.203    See https://google.aip.dev/auth/4110.204 205    1. If the environment variable ``GOOGLE_APPLICATION_CREDENTIALS`` is set206       to the path of a valid service account JSON file, then ID token is207       acquired using this service account credentials.208    2. If the application is running in Compute Engine, App Engine or Cloud Run,209       then the ID token are obtained from the metadata server.210    3. If metadata server doesn't exist and no valid service account credentials211       are found, :class:`~google.auth.exceptions.DefaultCredentialsError` will212       be raised.213 214    Example::215 216        import google.oauth2._id_token_async217        import google.auth.transport.aiohttp_requests218 219        request = google.auth.transport.aiohttp_requests.Request()220        target_audience = "https://pubsub.googleapis.com"221 222        id_token = await google.oauth2._id_token_async.fetch_id_token(request, target_audience)223 224    Args:225        request (google.auth.transport.aiohttp_requests.Request): A callable used to make226            HTTP requests.227        audience (str): The audience that this ID token is intended for.228 229    Returns:230        str: The ID token.231 232    Raises:233        ~google.auth.exceptions.DefaultCredentialsError:234            If metadata server doesn't exist and no valid service account235            credentials are found.236    """237    # 1. Try to get credentials from the GOOGLE_APPLICATION_CREDENTIALS environment238    # variable.239    credentials_filename = os.environ.get(environment_vars.CREDENTIALS)240    if credentials_filename:241        if not (242            os.path.exists(credentials_filename)243            and os.path.isfile(credentials_filename)244        ):245            raise exceptions.DefaultCredentialsError(246                "GOOGLE_APPLICATION_CREDENTIALS path is either not found or invalid."247            )248 249        try:250            with open(credentials_filename, "r") as f:251                from google.oauth2 import _service_account_async as service_account252 253                info = json.load(f)254                if info.get("type") == "service_account":255                    credentials = service_account.IDTokenCredentials.from_service_account_info(256                        info, target_audience=audience257                    )258                    await credentials.refresh(request)259                    return credentials.token260        except ValueError as caught_exc:261            new_exc = exceptions.DefaultCredentialsError(262                "GOOGLE_APPLICATION_CREDENTIALS is not valid service account credentials.",263                caught_exc,264            )265            raise new_exc from caught_exc266 267    # 2. Try to fetch ID token from metada server if it exists. The code works268    # for GAE and Cloud Run metadata server as well.269    try:270        from google.auth import compute_engine271        from google.auth.compute_engine import _metadata272 273        request_new = requests.Request()274        if _metadata.ping(request_new):275            credentials = compute_engine.IDTokenCredentials(276                request_new, audience, use_metadata_identity_endpoint=True277            )278            credentials.refresh(request_new)279            return credentials.token280    except (ImportError, exceptions.TransportError):281        pass282 283    raise exceptions.DefaultCredentialsError(284        "Neither metadata server or valid service account credentials are found."285    )286