Ring doorbel Ding Fix

"""Fixes for firebase_messaging bugs that silently break Ring's real-time push."""

import json
import logging
from base64 import urlsafe_b64decode
from contextlib import suppress as contextlib_suppress

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.serialization import load_der_private_key
from http_ece import decrypt as http_decrypt
from firebase_messaging import fcmpushclient
from firebase_messaging.fcmpushclient import ErrorType

_LOGGER = logging.getLogger(__name__)
DOMAIN = "fcm_padding_fix"


def _b64pad(s: str) -> bytes:
    return (s + "=" * (-len(s) % 4)).encode("ascii")


def _patched_decrypt_raw_data(credentials, crypto_key_str, salt_str, raw_data):
    crypto_key = urlsafe_b64decode(_b64pad(crypto_key_str))
    salt = urlsafe_b64decode(_b64pad(salt_str))
    der_data = urlsafe_b64decode(_b64pad(credentials["keys"]["private"]))
    secret = urlsafe_b64decode(_b64pad(credentials["keys"]["secret"]))
    privkey = load_der_private_key(der_data, password=None, backend=default_backend())
    return http_decrypt(
        raw_data,
        salt=salt,
        private_key=privkey,
        dh=crypto_key,
        version="aesgcm",
        auth_secret=secret,
    )


def _patched_handle_data_message(self, msg):
    _LOGGER.debug(
        "Received data message Stream ID: %s, Last: %s, Status: %s",
        msg.stream_id,
        msg.last_stream_id_received,
        msg.status,
    )

    if self._app_data_by_key(msg, "message_type", do_not_raise=True) == "deleted_messages":
        return

    crypto_key_raw = salt_raw = None
    try:
        crypto_key_raw = self._app_data_by_key(msg, "crypto-key")
        salt_raw = self._app_data_by_key(msg, "encryption")
        crypto_key = crypto_key_raw.removeprefix("dh=").split(";", 1)[0]
        salt = salt_raw.removeprefix("salt=").split(";", 1)[0]
        subtype = self._app_data_by_key(msg, "subtype")
        if subtype != self.credentials["gcm"]["app_id"]:
            self._log_warn_with_limit(
                "Subtype %s in data message does not match app id client was registered with %s",
                subtype,
                self.credentials["gcm"]["app_id"],
            )
        if not self.credentials:
            return
        decrypted = self._decrypt_raw_data(self.credentials, crypto_key, salt, msg.raw_data)
        decrypted_json = None
        with contextlib_suppress(json.JSONDecodeError, ValueError):
            decrypted_json = json.loads(decrypted.decode("utf-8"))
        if not decrypted_json:
            self._log_warn_with_limit("Failed to decrypt data for message %s", msg.persistent_id)
        ret_val = decrypted_json if decrypted_json else decrypted
        self._log_verbose("Data for message %s is: %s", msg.persistent_id, ret_val)
        if not isinstance(ret_val, dict):
            ret_val = {"message": ret_val}
        self.callback(ret_val, msg.persistent_id, self.callback_context)
        self._reset_error_count(ErrorType.NOTIFY)
    except Exception:
        _LOGGER.exception(
            "fcm_padding_fix: skipping one malformed/undecodable push message "
            "instead of tearing down the listener. raw crypto-key=%r encryption=%r "
            "persistent_id=%s",
            crypto_key_raw,
            salt_raw,
            getattr(msg, "persistent_id", None),
        )
        self._try_increment_error_count(ErrorType.NOTIFY)


async def async_setup(hass, config):
    fcmpushclient.FcmPushClient._decrypt_raw_data = staticmethod(_patched_decrypt_raw_data)
    fcmpushclient.FcmPushClient._handle_data_message = _patched_handle_data_message
    _LOGGER.warning(
        "fcm_padding_fix: patched FcmPushClient._decrypt_raw_data and _handle_data_message"
    )
    return True

{
  "domain": "fcm_padding_fix",
  "name": "FCM padding fix",
  "version": "1.1.0",
  "requirements": [
    "firebase-messaging @ git+https://github.com/sdb9696/firebase-messaging.git@dc5acc31a8bafa68f530f58ce964641ad8322394"
  ],
  "codeowners": [],
  "iot_class": "calculated"
}