# -*- coding: utf-8 -*- # # Copyright (C) 2021 Chris Caron <lead2gold@gmail.com> # All rights reserved. # # This code is licensed under the MIT License. # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files(the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and / or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions : # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. # # To generate a private key file for your service account: # # 1. In the Firebase console, open Settings > Service Accounts. # 2. Click Generate New Private Key, then confirm by clicking Generate Key. # 3. Securely store the JSON file containing the key. import requests import base64 import json import calendar from cryptography.hazmat import backends from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives import asymmetric from cryptography.exceptions import UnsupportedAlgorithm from datetime import datetime from datetime import timedelta from json.decoder import JSONDecodeError from urllib.parse import urlencode as _urlencode from ...logger import logger class GoogleOAuth: """ A OAuth simplified implimentation to Google's Firebase Cloud Messaging """ scopes = [ 'https://www.googleapis.com/auth/firebase.messaging', ] # 1 hour in seconds (the lifetime of our token) access_token_lifetime_sec = timedelta(seconds=3600) # The default URI to use if one is not found default_token_uri = 'https://oauth2.googleapis.com/token' # Taken right from google.auth.helpers: clock_skew = timedelta(seconds=10) def __init__(self, user_agent=None, timeout=(5, 4), verify_certificate=True): """ Initialize our OAuth object """ # Wether or not to verify ssl self.verify_certificate = verify_certificate # Our (connect, read) timeout self.request_timeout = timeout # assign our user-agent if defined self.user_agent = user_agent # initialize our other object variables self.__reset() def __reset(self): """ Reset object internal variables """ # Google Keyfile Encoding self.encoding = 'utf-8' # Our retrieved JSON content (unmangled) self.content = None # Our generated key information we cache once loaded self.private_key = None # Our keys we build using the provided content self.__refresh_token = None self.__access_token = None self.__access_token_expiry = datetime.utcnow() def load(self, path): """ Generate our SSL details """ # Reset our objects self.content = None self.private_key = None self.__access_token = None self.__access_token_expiry = datetime.utcnow() try: with open(path, mode="r", encoding=self.encoding) as fp: self.content = json.loads(fp.read()) except (OSError, IOError): logger.debug('FCM keyfile {} could not be accessed'.format(path)) return False except JSONDecodeError as e: logger.debug( 'FCM keyfile {} generated a JSONDecodeError: {}'.format( path, e)) return False if not isinstance(self.content, dict): logger.debug( 'FCM keyfile {} is incorrectly structured'.format(path)) self.__reset() return False # Verify we've got the correct tokens in our content to work with is_valid = next((False for k in ( 'client_email', 'private_key_id', 'private_key', 'type', 'project_id') if not self.content.get(k)), True) if not is_valid: logger.debug( 'FCM keyfile {} is missing required information'.format(path)) self.__reset() return False # Verify our service_account type if self.content.get('type') != 'service_account': logger.debug( 'FCM keyfile {} is not of type service_account'.format(path)) self.__reset() return False # Prepare our private key which is in PKCS8 PEM format try: self.private_key = serialization.load_pem_private_key( self.content.get('private_key').encode(self.encoding), password=None, backend=backends.default_backend()) except (TypeError, ValueError): # ValueError: If the PEM data could not be decrypted or if its # structure could not be decoded successfully. # TypeError: If a password was given and the private key was # not encrypted. Or if the key was encrypted but # no password was supplied. logger.error('FCM provided private key is invalid.') self.__reset() return False except UnsupportedAlgorithm: # If the serialized key is of a type that is not supported by # the backend. logger.error('FCM provided private key is not supported') self.__reset() return False # We've done enough validation to move on return True @property def access_token(self): """ Returns our access token (if it hasn't expired yet) - if we do not have one we'll fetch one. - if it expired, we'll renew it - if a key simply can't be acquired, then we return None """ if not self.private_key or not self.content: # invalid content (or not loaded) logger.error( 'No FCM JSON keyfile content loaded to generate a access ' 'token with.') return None if self.__access_token_expiry > datetime.utcnow(): # Return our no-expired key return self.__access_token # If we reach here we need to prepare our payload token_uri = self.content.get('token_uri', self.default_token_uri) service_email = self.content.get('client_email') key_identifier = self.content.get('private_key_id') # Generate our Assertion now = datetime.utcnow() expiry = now + self.access_token_lifetime_sec payload = { # The number of seconds since the UNIX epoch. "iat": calendar.timegm(now.utctimetuple()), "exp": calendar.timegm(expiry.utctimetuple()), # The issuer must be the service account email. "iss": service_email, # The audience must be the auth token endpoint's URI "aud": token_uri, # Our token scopes "scope": " ".join(self.scopes), } # JWT Details header = { 'typ': 'JWT', 'alg': 'RS256' if isinstance( self.private_key, asymmetric.rsa.RSAPrivateKey) else 'ES256', # Key Identifier 'kid': key_identifier, } # Encodes base64 strings removing any padding characters. segments = [ base64.urlsafe_b64encode( json.dumps(header).encode(self.encoding)).rstrip(b"="), base64.urlsafe_b64encode( json.dumps(payload).encode(self.encoding)).rstrip(b"="), ] signing_input = b".".join(segments) signature = self.private_key.sign( signing_input, asymmetric.padding.PKCS1v15(), hashes.SHA256(), ) # Finally append our segment segments.append(base64.urlsafe_b64encode(signature).rstrip(b"=")) assertion = b".".join(segments) http_payload = _urlencode({ 'assertion': assertion, 'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer', }) http_headers = { 'Content-Type': 'application/x-www-form-urlencoded', } if self.user_agent: http_headers['User-Agent'] = self.user_agent logger.info('Refreshing FCM Access Token') try: r = requests.post( token_uri, data=http_payload, headers=http_headers, verify=self.verify_certificate, timeout=self.request_timeout, ) if r.status_code != requests.codes.ok: # We had a problem logger.warning( 'Failed to update FCM Access Token error={}.' .format(r.status_code)) logger.debug( 'Response Details:\r\n%s', r.content) return None except requests.RequestException as e: logger.warning( 'A Connection error occurred refreshing FCM ' 'Access Token.' ) logger.debug('Socket Exception: %s', str(e)) return None # If we get here, we made our request successfully, now we need # to parse out the data response = json.loads(r.content) self.__access_token = response['access_token'] self.__refresh_token = response.get( 'refresh_token', self.__refresh_token) if 'expires_in' in response: delta = timedelta(seconds=int(response['expires_in'])) self.__access_token_expiry = \ delta + datetime.utcnow() - self.clock_skew else: # Allow some grace before we expire self.__access_token_expiry = expiry - self.clock_skew logger.debug( 'Access Token successfully acquired: %s', self.__access_token) # Return our token return self.__access_token @property def project_id(self): """ Returns the project id found in the file """ return None if not self.content \ else self.content.get('project_id')