mint_token.py

#!/usr/bin/env python

"""Script to test out minting a token without a scope.

Usage:

$ ./mint_token.py \
> --json-key-filename=/path/to/keyfile.json
JWT Header:
{
  "alg": "RS256",
  "kid": "9adKEYIDFOO",
  "typ": "JWT"
}
JWT Payload:
{
  "aud": "https://www.googleapis.com/oauth2/v4/token",
  "exp": 1456349646,
  "iat": 1456346046,
  "iss": "[email protected]",
  "scope": "https://www.googleapis.com/auth/userinfo.email"
}
Response Headers:
{
  "-content-encoding": "gzip",
  "alt-svc": "quic=\":443\"; ma=2592000; v=\"30,29,28,27,26,25\"",
  "alternate-protocol": "443:quic,p=1",
  "cache-control": "no-cache, no-store, max-age=0, must-revalidate",
  "content-length": "150",
  "content-type": "application/json; charset=UTF-8",
  "date": "Wed, 24 Feb 2016 20:34:13 GMT",
  "expires": "Fri, 01 Jan 1990 00:00:00 GMT",
  "pragma": "no-cache",
  "server": "GSE",
  "status": "200",
  "transfer-encoding": "chunked",
  "vary": "Origin, X-Origin",
  "x-content-type-options": "nosniff",
  "x-frame-options": "SAMEORIGIN",
  "x-xss-protection": "1; mode=block"
}
Response Body:
{
  "access_token": "ya29.THISISFORSUREATOKEN",
  "expires_in": 3600,
  "token_type": "Bearer"
}



$ ./mint_token.py \
> --json-key-filename=/path/to/keyfile.json \
> --scopeless
JWT Header:
{
  "alg": "RS256",
  "kid": "9adKEYIDFOO",
  "typ": "JWT"
}
JWT Payload:
{
  "aud": "https://bigtableclusteradmin.googleapis.com/google.bigtable.admin.cluster.v1.BigtableClusterService",
  "exp": 1456349634,
  "iat": 1456346034,
  "iss": "[email protected]",
  "sub": "[email protected]"
}
Response Headers:
{
  "-content-encoding": "gzip",
  "alt-svc": "quic=\":443\"; ma=2592000; v=\"30,29,28,27,26,25\"",
  "alternate-protocol": "443:quic,p=1",
  "cache-control": "private, max-age=0",
  "content-length": "67",
  "content-type": "application/json; charset=UTF-8",
  "date": "Wed, 24 Feb 2016 20:34:01 GMT",
  "expires": "Wed, 24 Feb 2016 20:34:01 GMT",
  "server": "GSE",
  "status": "400",
  "transfer-encoding": "chunked",
  "vary": "Origin, X-Origin",
  "x-content-type-options": "nosniff",
  "x-frame-options": "SAMEORIGIN",
  "x-xss-protection": "1; mode=block"
}
Response Body:
{
  "error": "invalid_grant",
  "error_description": "Bad Request"
}
"""

from __future__ import print_function

import argparse
import json
import time

import httplib2
from oauth2client._helpers import _urlsafe_b64encode
from oauth2client.crypt import RsaSigner
import six


SERVICE_URI = ('https://bigtableclusteradmin.googleapis.com/'
               'google.bigtable.admin.cluster.v1.BigtableClusterService')
EMAIL_SCOPE = 'https://www.googleapis.com/auth/userinfo.email'
DEFAULT_TOKEN_URI = 'https://www.googleapis.com/oauth2/v4/token'


def main(key_dict, scopeless):
    private_key_pkcs8_pem = key_dict['private_key']
    signer = RsaSigner.from_string(private_key_pkcs8_pem)

    jwt_header = {
        'alg': 'RS256',
        'kid': key_dict['private_key_id'],
        'typ': 'JWT',
    }
    now = int(time.time())
    payload = {
        'exp': now + 3600,
        'iat': now,
        'iss': key_dict['client_email'],
    }
    if scopeless:
        token_uri = DEFAULT_TOKEN_URI
        # NOTE: Potentially useful fields:
        #       jwt_header['jku'] = key_dict['client_x509_cert_url']
        #       payload['nbf'] = now + 300
        # NOTE: It seems the token URI may need to change.
        payload['aud'] = SERVICE_URI
        payload['sub'] = key_dict['client_email']
    else:
        token_uri = DEFAULT_TOKEN_URI
        payload['aud'] = DEFAULT_TOKEN_URI
        payload['scope'] = EMAIL_SCOPE

    print_jwt_info(jwt_header, payload)
    jwt_val = make_jwt(signer, jwt_header, payload)
    response, content = make_request(jwt_val, token_uri)
    print_response(response, content)


def make_jwt(signer, jwt_header, payload):
    jwt_header_seg = _urlsafe_b64encode(json.dumps(jwt_header))
    payload_seg = _urlsafe_b64encode(json.dumps(payload))
    to_sign = jwt_header_seg + b'.' + payload_seg
    signature = signer.sign(to_sign)
    signature_seg = _urlsafe_b64encode(signature)
    return to_sign + b'.' + signature_seg


def make_request(jwt_val, token_uri):
    body = six.moves.urllib.parse.urlencode({
        'assertion': jwt_val,
        'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
    })
    headers = {
        'content-type': 'application/x-www-form-urlencoded',
    }
    return httplib2.Http().request(
        token_uri, method='POST', body=body, headers=headers)


def print_jwt_info(jwt_header, payload):
    print('JWT Header:')
    print(json.dumps(jwt_header, indent=2, sort_keys=True))
    print('JWT Payload:')
    print(json.dumps(payload, indent=2, sort_keys=True))


def print_response(response, content):
    print('Response Headers:')
    print(json.dumps(dict(response), indent=2, sort_keys=True))
    print('Response Body:')
    try:
        print(json.dumps(json.loads(content), indent=2, sort_keys=True))
    except ValueError:
        print(content)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        description='Try to mint a token with scopeless credentials')
    parser.add_argument('--json-key-filename', dest='key_filename',
                        required=True)
    parser.add_argument('--scopeless', action='store_true')
    args = parser.parse_args()
    with open(args.key_filename, 'rb') as fh:
        key_dict = json.load(fh)
    main(key_dict, args.scopeless)