credentials.py

# Copyright 2014 Google Inc. All rights reserved.
#
# 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.

"""A simple wrapper around the OAuth2 credentials library."""

import base64
import datetime
import six
from six.moves.urllib.parse import urlencode  # pylint: disable=F0401

try:
    from OpenSSL import crypto
except ImportError:  # pragma: NO COVER
    # pyOpenSSL can't be installed on App Engine, but it will not
    # be needed there since app_identity is used.
    crypto = None

from oauth2client import client
from oauth2client.client import _get_application_default_credential_from_file
from oauth2client import crypt
from oauth2client import service_account
try:
    from oauth2client.appengine import AppAssertionCredentials as _GAECreds
except ImportError:
    class _GAECreds(object):
        """Dummy class if not in App Engine environment."""

try:
    from google.appengine.api import app_identity
except ImportError:
    app_identity = None

from gcloud._helpers import UTC
from gcloud._helpers import _NOW
from gcloud._helpers import _microseconds_from_datetime


def get_credentials():
    """Gets credentials implicitly from the current environment.

    .. note::

        You should not need to use this function directly. Instead, use a
        helper method which uses this method under the hood.

    Checks environment in order of precedence:

    * Google App Engine (production and testing)
    * Environment variable :envvar:`GOOGLE_APPLICATION_CREDENTIALS` pointing to
      a file with stored credentials information.
    * Stored "well known" file associated with ``gcloud`` command line tool.
    * Google Compute Engine production environment.

    The file referred to in :envvar:`GOOGLE_APPLICATION_CREDENTIALS` is
    expected to contain information about credentials that are ready to use.
    This means either service account information or user account information
    with a ready-to-use refresh token:

    .. code:: json

      {
          'type': 'authorized_user',
          'client_id': '...',
          'client_secret': '...',
          'refresh_token': '...'
      }

    or

    .. code:: json

      {
          'type': 'service_account',
          'client_id': '...',
          'client_email': '...',
          'private_key_id': '...',
          'private_key': '...'
      }

    The second of these is simply a JSON key downloaded from the Google APIs
    console. The first is a close cousin of the "client secrets" JSON file
    used by :mod:`oauth2client.clientsecrets` but differs in formatting.

    :rtype: :class:`oauth2client.client.GoogleCredentials`,
            :class:`oauth2client.contrib.appengine.AppAssertionCredentials`,
            :class:`oauth2client.contrib.gce.AppAssertionCredentials`,
            :class:`oauth2client.service_account._ServiceAccountCredentials`
    :returns: A new credentials instance corresponding to the implicit
              environment.
    """
    return client.GoogleCredentials.get_application_default()


def get_for_service_account_json(json_credentials_path, scope=None):
    """Gets the credentials for a service account with JSON key.

    :type json_credentials_path: string
    :param json_credentials_path: The path to a private key file (this file was
                                  given to you when you created the service
                                  account). This file must contain a JSON
                                  object with a private key and other
                                  credentials information (downloaded from the
                                  Google APIs console).

    :type scope: string or tuple of string
    :param scope: The scope against which to authenticate. (Different services
                  require different scopes, check the documentation for which
                  scope is required for the different levels of access to any
                  particular API.)

    :rtype: :class:`oauth2client.client.GoogleCredentials`,
            :class:`oauth2client.service_account._ServiceAccountCredentials`
    :returns: New service account or Google (for a user JSON key file)
              credentials object.
    """
    credentials = _get_application_default_credential_from_file(
        json_credentials_path)
    if scope is not None:
        credentials = credentials.create_scoped(scope)
    return credentials


def get_for_service_account_p12(client_email, private_key_path, scope=None):
    """Gets the credentials for a service account with PKCS12 / p12 key.

    .. note::
      This method is not used by default, instead :func:`get_credentials`
      is used. This method is intended to be used when the environment is
      known explicitly and detecting the environment implicitly would be
      superfluous.

    :type client_email: string
    :param client_email: The e-mail attached to the service account.

    :type private_key_path: string
    :param private_key_path: The path to a private key file (this file was
                             given to you when you created the service
                             account). This file must be in P12 format.

    :type scope: string or tuple of string
    :param scope: The scope against which to authenticate. (Different services
                  require different scopes, check the documentation for which
                  scope is required for the different levels of access to any
                  particular API.)

    :rtype: :class:`oauth2client.client.SignedJwtAssertionCredentials`
    :returns: A new ``SignedJwtAssertionCredentials`` instance with the
              needed service account settings.
    """
    return client.SignedJwtAssertionCredentials(
        service_account_name=client_email,
        private_key=open(private_key_path, 'rb').read(),
        scope=scope)


def _get_pem_key(credentials):
    """Gets private key for a PEM payload from a credentials object.

    :type credentials: :class:`client.SignedJwtAssertionCredentials`,
                       :class:`service_account._ServiceAccountCredentials`
    :param credentials: The credentials used to create a private key
                        for signing text.

    :rtype: :class:`OpenSSL.crypto.PKey`
    :returns: A PKey object used to sign text.
    :raises: `TypeError` if `credentials` is the wrong type.
             `EnvironmentError` if `crypto` did not import successfully.
    """
    if isinstance(credentials, client.SignedJwtAssertionCredentials):
        # Take our PKCS12 (.p12) text and convert to PEM text.
        pem_text = crypt.pkcs12_key_as_pem(credentials.private_key,
                                           credentials.private_key_password)
    elif isinstance(credentials, service_account._ServiceAccountCredentials):
        pem_text = credentials._private_key_pkcs8_text
    else:
        raise TypeError((credentials,
                         'not a valid service account credentials type'))

    if crypto is None:
        raise EnvironmentError(
            'pyOpenSSL must be installed to load a private key')
    return crypto.load_privatekey(crypto.FILETYPE_PEM, pem_text)


def _get_signature_bytes(credentials, string_to_sign):
    """Uses crypto attributes of credentials to sign a string/bytes.

    :type credentials: :class:`client.SignedJwtAssertionCredentials`,
                       :class:`service_account._ServiceAccountCredentials`,
                       :class:`_GAECreds`
    :param credentials: The credentials used for signing text (typically
                        involves the creation of a PKey).

    :type string_to_sign: string
    :param string_to_sign: The string to be signed by the credentials.

    :rtype: bytes
    :returns: Signed bytes produced by the credentials.
    :raises: `EnvironmentError` if `crypto` did not import successfully.
    """
    if isinstance(credentials, _GAECreds):
        _, signed_bytes = app_identity.sign_blob(string_to_sign)
        return signed_bytes
    else:
        # Sign the string with the PKey.
        pkey = _get_pem_key(credentials)
        if not isinstance(string_to_sign, six.binary_type):
            string_to_sign = string_to_sign.encode('utf-8')
        if crypto is None:
            raise EnvironmentError(
                'pyOpenSSL must be installed to sign content using a '
                'private key')
        return crypto.sign(pkey, string_to_sign, 'SHA256')


def _get_service_account_name(credentials):
    """Determines service account name from a credentials object.

    :type credentials: :class:`client.SignedJwtAssertionCredentials`,
                       :class:`service_account._ServiceAccountCredentials`,
                       :class:`_GAECreds`
    :param credentials: The credentials used to determine the service
                        account name.

    :rtype: string
    :returns: Service account name associated with the credentials.
    :raises: :class:`ValueError` if the credentials are not a valid service
             account type.
    """
    service_account_name = None
    if isinstance(credentials, client.SignedJwtAssertionCredentials):
        service_account_name = credentials.service_account_name
    elif isinstance(credentials, service_account._ServiceAccountCredentials):
        service_account_name = credentials._service_account_email
    elif isinstance(credentials, _GAECreds):
        service_account_name = app_identity.get_service_account_name()

    if service_account_name is None:
        raise ValueError('Service account name could not be determined '
                         'from credentials')
    return service_account_name


def _get_signed_query_params(credentials, expiration, string_to_sign):
    """Gets query parameters for creating a signed URL.

    :type credentials: :class:`client.SignedJwtAssertionCredentials`,
                       :class:`service_account._ServiceAccountCredentials`
    :param credentials: The credentials used to create a private key
                        for signing text.

    :type expiration: int or long
    :param expiration: When the signed URL should expire.

    :type string_to_sign: string
    :param string_to_sign: The string to be signed by the credentials.

    :rtype: dict
    :returns: Query parameters matching the signing credentials with a
              signed payload.
    """
    signature_bytes = _get_signature_bytes(credentials, string_to_sign)
    signature = base64.b64encode(signature_bytes)
    service_account_name = _get_service_account_name(credentials)
    return {
        'GoogleAccessId': service_account_name,
        'Expires': str(expiration),
        'Signature': signature,
    }


def _get_expiration_seconds(expiration):
    """Convert 'expiration' to a number of seconds in the future.

    :type expiration: int, long, datetime.datetime, datetime.timedelta
    :param expiration: When the signed URL should expire.

    :rtype: int
    :returns: a timestamp as an absolute number of seconds.
    """
    # If it's a timedelta, add it to `now` in UTC.
    if isinstance(expiration, datetime.timedelta):
        now = _NOW().replace(tzinfo=UTC)
        expiration = now + expiration

    # If it's a datetime, convert to a timestamp.
    if isinstance(expiration, datetime.datetime):
        micros = _microseconds_from_datetime(expiration)
        expiration = micros // 10**6

    if not isinstance(expiration, six.integer_types):
        raise TypeError('Expected an integer timestamp, datetime, or '
                        'timedelta. Got %s' % type(expiration))
    return expiration


def generate_signed_url(credentials, resource, expiration,
                        api_access_endpoint='',
                        method='GET', content_md5=None,
                        content_type=None, response_type=None,
                        response_disposition=None, generation=None):
    """Generate signed URL to provide query-string auth'n to a resource.

    .. note::

        If you are on Google Compute Engine, you can't generate a signed URL.
        Follow `Issue 922`_ for updates on this. If you'd like to be able to
        generate a signed URL from GCE, you can use a standard service account
        from a JSON file rather than a GCE service account.

    See headers `reference`_ for more details on optional arguments.

    .. _Issue 922: https://github.com/GoogleCloudPlatform/\
                   gcloud-python/issues/922
    .. _reference: https://cloud.google.com/storage/docs/reference-headers

    :type credentials: :class:`oauth2client.appengine.AppAssertionCredentials`
    :param credentials: Credentials object with an associated private key to
                        sign text.

    :type resource: string
    :param resource: A pointer to a specific resource
                     (typically, ``/bucket-name/path/to/blob.txt``).

    :type expiration: :class:`int`, :class:`long`, :class:`datetime.datetime`,
                      :class:`datetime.timedelta`
    :param expiration: When the signed URL should expire.

    :type api_access_endpoint: str
    :param api_access_endpoint: Optional URI base. Defaults to empty string.

    :type method: str
    :param method: The HTTP verb that will be used when requesting the URL.
                   Defaults to ``'GET'``.

    :type content_md5: str
    :param content_md5: (Optional) The MD5 hash of the object referenced by
                        ``resource``.

    :type content_type: str
    :param content_type: (Optional) The content type of the object referenced
                         by ``resource``.

    :type response_type: str
    :param response_type: (Optional) Content type of responses to requests for
                          the signed URL. Used to over-ride the content type of
                          the underlying resource.

    :type response_disposition: str
    :param response_disposition: (Optional) Content disposition of responses to
                                 requests for the signed URL.

    :type generation: str
    :param generation: (Optional) A value that indicates which generation of
                       the resource to fetch.

    :rtype: string
    :returns: A signed URL you can use to access the resource
              until expiration.
    """
    expiration = _get_expiration_seconds(expiration)

    # Generate the string to sign.
    string_to_sign = '\n'.join([
        method,
        content_md5 or '',
        content_type or '',
        str(expiration),
        resource])

    # Set the right query parameters.
    query_params = _get_signed_query_params(credentials,
                                            expiration,
                                            string_to_sign)
    if response_type is not None:
        query_params['response-content-type'] = response_type
    if response_disposition is not None:
        query_params['response-content-disposition'] = response_disposition
    if generation is not None:
        query_params['generation'] = generation

    # Return the built URL.
    return '{endpoint}{resource}?{querystring}'.format(
        endpoint=api_access_endpoint, resource=resource,
        querystring=urlencode(query_params))