Add JWT credentials (#21)
diff --git a/google/auth/jwt.py b/google/auth/jwt.py
index f68fc4b..394a7d5 100644
--- a/google/auth/jwt.py
+++ b/google/auth/jwt.py
@@ -42,8 +42,13 @@
import base64
import collections
+import datetime
+import io
import json
+from six.moves import urllib
+
+from google.auth import credentials
from google.auth import crypt
from google.auth import _helpers
@@ -234,3 +239,240 @@
claim_audience, audience))
return payload
+
+
+class Credentials(credentials.Signing,
+ credentials.Credentials):
+ """Credentials that use a JWT as the bearer token.
+
+ These credentials require an "audience" claim. This claim identifies the
+ intended recipient of the bearer token. You can set the audience when
+ you construct these credentials, however, these credentials can also set
+ the audience claim automatically if not specified. In this case, whenever
+ a request is made the credentials will automatically generate a one-time
+ JWT with the request URI as the audience.
+
+ The constructor arguments determine the claims for the JWT that is
+ sent with requests. Usually, you'll construct these credentials with
+ one of the helper constructors as shown in the next section.
+
+ To create JWT credentials using a Google service account private key
+ JSON file::
+
+ credentials = jwt.Credentials.from_service_account_file(
+ 'service-account.json')
+
+ If you already have the service account file loaded and parsed::
+
+ service_account_info = json.load(open('service_account.json'))
+ credentials = jwt.Credentials.from_service_account_info(
+ service_account_info)
+
+ Both helper methods pass on arguments to the constructor, so you can
+ specify the JWT claims::
+
+ credentials = jwt.Credentials.from_service_account_file(
+ 'service-account.json',
+ audience='https://speech.googleapis.com',
+ additional_claims={'meta': 'data'})
+
+ You can also construct the credentials directly if you have a
+ :class:`~google.auth.crypt.Signer` instance::
+
+ credentials = jwt.Credentials(
+ signer, issuer='your-issuer', subject='your-subject')
+
+ The claims are considered immutable. If you want to modify the claims,
+ you can easily create another instance using :meth:`with_claims`::
+
+ new_credentials = credentials.with_claims(
+ audience='https://vision.googleapis.com')
+ """
+
+ def __init__(self, signer, issuer=None, subject=None, audience=None,
+ additional_claims=None,
+ token_lifetime=_DEFAULT_TOKEN_LIFETIME_SECS):
+ """
+ Args:
+ signer (google.auth.crypt.Signer): The signer used to sign JWTs.
+ issuer (str): The `iss` claim.
+ subject (str): The `sub` claim.
+ audience (str): the `aud` claim. The intended audience for the
+ credentials. If not specified, a new JWT will be generated for
+ every request and will use the request URI as the audience.
+ additional_claims (Mapping[str, str]): Any additional claims for
+ the JWT payload.
+ token_lifetime (int): The amount of time in seconds for
+ which the token is valid. Defaults to 1 hour.
+ """
+ super(Credentials, self).__init__()
+ self._signer = signer
+ self._issuer = issuer
+ self._subject = subject
+ self._audience = audience
+ self._additional_claims = additional_claims or {}
+ self._token_lifetime = token_lifetime
+
+ @classmethod
+ def from_service_account_info(cls, info, **kwargs):
+ """Creates a Credentials instance from parsed service account info.
+
+ Args:
+ info (Mapping[str, str]): The service account info in Google
+ format.
+ kwargs: Additional arguments to pass to the constructor.
+
+ Returns:
+ google.auth.jwt.Credentials: The constructed credentials.
+
+ Raises:
+ ValueError: If the info is not in the expected format.
+ """
+
+ try:
+ email = info['client_email']
+ key_id = info['private_key_id']
+ private_key = info['private_key']
+ except KeyError:
+ raise ValueError(
+ 'Service account info was not in the expected format.')
+
+ signer = crypt.Signer.from_string(private_key, key_id)
+
+ kwargs.setdefault('subject', email)
+ return cls(signer, issuer=email, **kwargs)
+
+ @classmethod
+ def from_service_account_file(cls, filename, **kwargs):
+ """Creates a Credentials instance from a service account json file.
+
+ Args:
+ filename (str): The path to the service account json file.
+ kwargs: Additional arguments to pass to the constructor.
+
+ Returns:
+ google.auth.jwt.Credentials: The constructed credentials.
+ """
+ with io.open(filename, 'r', encoding='utf-8') as json_file:
+ info = json.load(json_file)
+ return cls.from_service_account_info(info, **kwargs)
+
+ def with_claims(self, issuer=None, subject=None, audience=None,
+ additional_claims=None):
+ """Returns a copy of these credentials with modified claims.
+
+ Args:
+ issuer (str): The `iss` claim. If unspecified the current issuer
+ claim will be used.
+ subject (str): The `sub` claim. If unspecified the current subject
+ claim will be used.
+ audience (str): the `aud` claim. If not specified, a new
+ JWT will be generated for every request and will use
+ the request URI as the audience.
+ additional_claims (Mapping[str, str]): Any additional claims for
+ the JWT payload. This will be merged with the current
+ additional claims.
+
+ Returns:
+ google.auth.jwt.Credentials: A new credentials instance.
+ """
+ return Credentials(
+ self._signer,
+ issuer=issuer if issuer is not None else self._issuer,
+ subject=subject if subject is not None else self._subject,
+ audience=audience if audience is not None else self._audience,
+ additional_claims=self._additional_claims.copy().update(
+ additional_claims or {}))
+
+ def _make_jwt(self, audience=None):
+ """Make a signed JWT.
+
+ Args:
+ audience (str): Overrides the instance's current audience claim.
+
+ Returns:
+ Tuple(bytes, datetime): The encoded JWT and the expiration.
+ """
+ now = _helpers.utcnow()
+ lifetime = datetime.timedelta(seconds=self._token_lifetime)
+ expiry = now + lifetime
+
+ payload = {
+ 'iss': self._issuer,
+ 'sub': self._subject or self._issuer,
+ 'iat': _helpers.datetime_to_secs(now),
+ 'exp': _helpers.datetime_to_secs(expiry),
+ 'aud': audience or self._audience,
+ }
+
+ payload.update(self._additional_claims)
+
+ jwt = encode(self._signer, payload)
+
+ return jwt, expiry
+
+ def _make_one_time_jwt(self, uri):
+ """Makes a one-off JWT with the URI as the audience.
+
+ Args:
+ uri (str): The request URI.
+
+ Returns:
+ bytes: The encoded JWT.
+ """
+ parts = urllib.parse.urlsplit(uri)
+ # Strip query string and fragment
+ audience = urllib.parse.urlunsplit(
+ (parts.scheme, parts.netloc, parts.path, None, None))
+ token, _ = self._make_jwt(audience=audience)
+ return token
+
+ def refresh(self, request):
+ """Refreshes the access token.
+
+ Args:
+ request (Any): Unused.
+ """
+ # pylint: disable=unused-argument
+ # (pylint doesn't correctly recognize overridden methods.)
+ self.token, self.expiry = self._make_jwt()
+
+ def sign_bytes(self, message):
+ """Signs the given message.
+
+ Args:
+ message (bytes): The message to sign.
+
+ Returns:
+ bytes: The message signature.
+ """
+ return self._signer.sign(message)
+
+ def before_request(self, request, method, url, headers):
+ """Performs credential-specific before request logic.
+
+ If an audience is specified it will refresh the credentials if
+ necessary. If no audience is specified it will generate a one-time
+ token for the request URI. In either case, it will set the
+ authorization header in headers to the token.
+
+ Args:
+ request (Any): Unused.
+ method (str): The request's HTTP method.
+ url (str): The request's URI.
+ headers (Mapping): The request's headers.
+ """
+ # pylint: disable=unused-argument
+ # (pylint doesn't correctly recognize overridden methods.)
+
+ # If this set of credentials has a pre-set audience, just ensure that
+ # there is a valid token and apply the auth headers.
+ if self._audience:
+ if not self.valid:
+ self.refresh(request)
+ self.apply(headers)
+ # Otherwise, generate a one-time token using the URL
+ # (without the query string and fragment) as the audience.
+ else:
+ token = self._make_one_time_jwt(url)
+ self.apply(headers, token=token)
diff --git a/tests/test_jwt.py b/tests/test_jwt.py
index 69628e5..b6c07df 100644
--- a/tests/test_jwt.py
+++ b/tests/test_jwt.py
@@ -14,8 +14,10 @@
import base64
import datetime
+import json
import os
+import mock
import pytest
from google.auth import _helpers
@@ -34,6 +36,11 @@
with open(os.path.join(DATA_DIR, 'other_cert.pem'), 'rb') as fh:
OTHER_CERT_BYTES = fh.read()
+SERVICE_ACCOUNT_JSON_FILE = os.path.join(DATA_DIR, 'service_account.json')
+
+with open(SERVICE_ACCOUNT_JSON_FILE, 'r') as fh:
+ SERVICE_ACCOUNT_INFO = json.load(fh)
+
@pytest.fixture
def signer():
@@ -187,3 +194,147 @@
certs = {'2': OTHER_CERT_BYTES, '3': PUBLIC_CERT_BYTES}
payload = jwt.decode(token, certs)
assert payload['user'] == 'billy bob'
+
+
+class TestCredentials:
+ SERVICE_ACCOUNT_EMAIL = 'service-account@example.com'
+ SUBJECT = 'subject'
+ AUDIENCE = 'audience'
+ ADDITIONAL_CLAIMS = {'meta': 'data'}
+ credentials = None
+
+ @pytest.fixture(autouse=True)
+ def credentials_fixture(self, signer):
+ self.credentials = jwt.Credentials(
+ signer, self.SERVICE_ACCOUNT_EMAIL)
+
+ def test_from_service_account_info(self):
+ with open(SERVICE_ACCOUNT_JSON_FILE, 'r') as fh:
+ info = json.load(fh)
+
+ credentials = jwt.Credentials.from_service_account_info(info)
+
+ assert credentials._signer.key_id == info['private_key_id']
+ assert credentials._issuer == info['client_email']
+ assert credentials._subject == info['client_email']
+
+ def test_from_service_account_info_args(self):
+ info = SERVICE_ACCOUNT_INFO.copy()
+
+ credentials = jwt.Credentials.from_service_account_info(
+ info, subject=self.SUBJECT, audience=self.AUDIENCE,
+ additional_claims=self.ADDITIONAL_CLAIMS)
+
+ assert credentials._signer.key_id == info['private_key_id']
+ assert credentials._issuer == info['client_email']
+ assert credentials._subject == self.SUBJECT
+ assert credentials._audience == self.AUDIENCE
+ assert credentials._additional_claims == self.ADDITIONAL_CLAIMS
+
+ def test_from_service_account_bad_private_key(self):
+ info = SERVICE_ACCOUNT_INFO.copy()
+ info['private_key'] = 'garbage'
+
+ with pytest.raises(ValueError) as excinfo:
+ jwt.Credentials.from_service_account_info(info)
+
+ assert excinfo.match(r'No key could be detected')
+
+ def test_from_service_account_bad_format(self):
+ with pytest.raises(ValueError):
+ jwt.Credentials.from_service_account_info({})
+
+ def test_from_service_account_file(self):
+ info = SERVICE_ACCOUNT_INFO.copy()
+
+ credentials = jwt.Credentials.from_service_account_file(
+ SERVICE_ACCOUNT_JSON_FILE)
+
+ assert credentials._signer.key_id == info['private_key_id']
+ assert credentials._issuer == info['client_email']
+ assert credentials._subject == info['client_email']
+
+ def test_from_service_account_file_args(self):
+ info = SERVICE_ACCOUNT_INFO.copy()
+
+ credentials = jwt.Credentials.from_service_account_file(
+ SERVICE_ACCOUNT_JSON_FILE, subject=self.SUBJECT,
+ audience=self.AUDIENCE, additional_claims=self.ADDITIONAL_CLAIMS)
+
+ assert credentials._signer.key_id == info['private_key_id']
+ assert credentials._issuer == info['client_email']
+ assert credentials._subject == self.SUBJECT
+ assert credentials._audience == self.AUDIENCE
+ assert credentials._additional_claims == self.ADDITIONAL_CLAIMS
+
+ def test_default_state(self):
+ assert not self.credentials.valid
+ # Expiration hasn't been set yet
+ assert not self.credentials.expired
+
+ def test_sign_bytes(self):
+ to_sign = b'123'
+ signature = self.credentials.sign_bytes(to_sign)
+ assert crypt.verify_signature(to_sign, signature, PUBLIC_CERT_BYTES)
+
+ def _verify_token(self, token):
+ payload = jwt.decode(token, PUBLIC_CERT_BYTES)
+ assert payload['iss'] == self.SERVICE_ACCOUNT_EMAIL
+ return payload
+
+ def test_refresh(self):
+ self.credentials.refresh(None)
+ assert self.credentials.valid
+ assert not self.credentials.expired
+
+ def test_expired(self):
+ assert not self.credentials.expired
+
+ self.credentials.refresh(None)
+ assert not self.credentials.expired
+
+ with mock.patch('google.auth._helpers.utcnow') as now:
+ one_day = datetime.timedelta(days=1)
+ now.return_value = self.credentials.expiry + one_day
+ assert self.credentials.expired
+
+ def test_before_request_one_time_token(self):
+ headers = {}
+
+ self.credentials.refresh(None)
+ self.credentials.before_request(
+ mock.Mock(), 'GET', 'http://example.com?a=1#3', headers)
+
+ header_value = headers['authorization']
+ _, token = header_value.split(' ')
+
+ # This should be a one-off token, so it shouldn't be the same as the
+ # credentials' stored token.
+ assert token != self.credentials.token
+
+ payload = self._verify_token(token)
+ assert payload['aud'] == 'http://example.com'
+
+ def test_before_request_with_preset_audience(self):
+ headers = {}
+
+ credentials = self.credentials.with_claims(audience=self.AUDIENCE)
+ credentials.refresh(None)
+ credentials.before_request(
+ None, 'GET', 'http://example.com?a=1#3', headers)
+
+ header_value = headers['authorization']
+ _, token = header_value.split(' ')
+
+ # Since the audience is set, it should use the existing token.
+ assert token.encode('utf-8') == credentials.token
+
+ payload = self._verify_token(token)
+ assert payload['aud'] == self.AUDIENCE
+
+ def test_before_request_refreshes(self):
+ credentials = self.credentials.with_claims(audience=self.AUDIENCE)
+ assert not credentials.valid
+ credentials.before_request(
+ None, 'GET', 'http://example.com?a=1#3', {})
+ assert credentials.valid