SickGear/lib/apprise/plugins/NotifyFCM/oauth.py

# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2023, Chris Caron <lead2gold@gmail.com>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
#    this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
#    this list of conditions and the following disclaimer in the documentation
#    and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
# 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 datetime import timezone
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.now(timezone.utc)

    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.now(timezone.utc)

        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.now(timezone.utc):
            # 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.now(timezone.utc)
        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.now(timezone.utc) - 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')