From 94a8f83557302ba69fbff85de9fa5b24c1ca1108 Mon Sep 17 00:00:00 2001 From: Dirk Nienhaus Date: Sat, 13 Jun 2026 17:20:00 +0200 Subject: [PATCH] Make data message decryption robust to unpadded keys and bad payloads The crypto-key and salt headers (and webpush keys generally, per RFC 8291) are transmitted without base64 '=' padding. Decoding them with a bare urlsafe_b64decode raises binascii.Error on otherwise valid input, which in downstream Home Assistant Ring use repeatedly crashed the push client: ERROR Unknown error: Incorrect padding, shutting down FcmPushClient. File "fcmpushclient.py", line 439, in _handle_data_message File "fcmpushclient.py", line 378, in _decrypt_raw_data crypto_key = urlsafe_b64decode(crypto_key_str.encode("ascii")) binascii.Error: Incorrect padding Add a small padding helper and use it for all four base64 fields in _decrypt_raw_data. Additionally, wrap the decrypt call in _handle_data_message so a single undecryptable message is logged and skipped (binascii.Error is a ValueError subclass) instead of propagating and tearing down the listen loop. Co-Authored-By: Claude Opus 4.8 --- firebase_messaging/fcmpushclient.py | 37 +++++++++++++++----- tests/test_fcmpushclient.py | 52 +++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 9 deletions(-) diff --git a/firebase_messaging/fcmpushclient.py b/firebase_messaging/fcmpushclient.py index 5afdda2..5e59001 100644 --- a/firebase_messaging/fcmpushclient.py +++ b/firebase_messaging/fcmpushclient.py @@ -45,6 +45,18 @@ OnNotificationCallable = Callable[[dict[str, Any], str, Any], None] CredentialsUpdatedCallable = Callable[[dict[str, Any]], None] + +def _urlsafe_b64decode_padded(data: str) -> bytes: + """Decode urlsafe base64 that may be missing its '=' padding. + + Webpush keys and the crypto-key/salt headers (RFC 8291) are transmitted + without padding. ``base64.urlsafe_b64decode`` raises ``binascii.Error`` + (a ``ValueError`` subclass) on unpadded input, so restore the padding + before decoding. + """ + return urlsafe_b64decode(data.encode("ascii") + b"=" * (-len(data) % 4)) + + # MCS Message Types and Tags MCS_MESSAGE_TAG = { HeartbeatPing: 0, @@ -378,12 +390,10 @@ def _decrypt_raw_data( salt_str: str, raw_data: bytes, ) -> bytes: - crypto_key = urlsafe_b64decode(crypto_key_str.encode("ascii")) - salt = urlsafe_b64decode(salt_str.encode("ascii")) - der_data_str = credentials["keys"]["private"] - der_data = urlsafe_b64decode(der_data_str.encode("ascii") + b"========") - secret_str = credentials["keys"]["secret"] - secret = urlsafe_b64decode(secret_str.encode("ascii") + b"========") + crypto_key = _urlsafe_b64decode_padded(crypto_key_str) + salt = _urlsafe_b64decode_padded(salt_str) + der_data = _urlsafe_b64decode_padded(credentials["keys"]["private"]) + secret = _urlsafe_b64decode_padded(credentials["keys"]["secret"]) privkey = load_der_private_key( der_data, password=None, backend=default_backend() ) @@ -439,9 +449,18 @@ def _handle_data_message( ) if not self.credentials: return - decrypted = self._decrypt_raw_data( - self.credentials, crypto_key, salt, msg.raw_data - ) + try: + decrypted = self._decrypt_raw_data( + self.credentials, crypto_key, salt, msg.raw_data + ) + except ValueError as ex: + # binascii.Error and http_ece decrypt failures are both ValueError. + # Skip the undecryptable message rather than tearing down the + # connection so a single bad payload can't stop the listener. + self._log_warn_with_limit( + "Failed to decrypt data for message %s: %s", msg.persistent_id, ex + ) + return decrypted_json = None with contextlib_suppress(json.JSONDecodeError, ValueError): decrypted_json = json.loads(decrypted.decode("utf-8")) diff --git a/tests/test_fcmpushclient.py b/tests/test_fcmpushclient.py index 2476b8b..bb662cc 100644 --- a/tests/test_fcmpushclient.py +++ b/tests/test_fcmpushclient.py @@ -251,3 +251,55 @@ def set_app_data_by_key(msg, key, value): ) assert raw_data_decrypted == raw_data + + +async def test_decrypt_unpadded_crypto_key_and_salt(): + """crypto-key/salt headers arrive without base64 '=' padding (RFC 8291). + + The wire format strips padding, so _decrypt_raw_data must restore it + instead of raising binascii.Error on otherwise valid input. + """ + + def get_app_data_by_key(msg, key): + for x in msg.app_data: + if x.key == key: + return x.value + + dms = load_fixture_as_msg("data_message_stanza.json", DataMessageStanza) + credentials = load_fixture_as_dict("credentials.json") + raw_data = b'{ "foo" : "bar" }' + salt_str = get_app_data_by_key(dms, "encryption")[5:] + salt = urlsafe_b64decode(salt_str.encode("ascii")) + + sender_pub = "BAGEFtID7WlmwzQ9pbjdRYAhfPe7Z8lA3ZGIPUh0SE3ikoY2PIrWUP0rmhpE4Kl8ImgMUDjKWrz0WmtLxORIHuw" + sender_pri_der = urlsafe_b64decode( + "MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgwSUpDfIqdJG3XVkn7t1GExHuW3gsqD4-J525w-rnCIihRANCAAQBhBbSA-1pZsM0PaW43UWAIXz3u2fJQN2RiD1IdEhN4pKGNjyK1lD9K5oaROCpfCJoDFA4ylq89FprS8TkSB7s".encode( + "ascii" + ) + + b"========" + ) + sender_privkey = load_der_private_key( + sender_pri_der, password=None, backend=default_backend() + ) + sender_sec = urlsafe_b64decode( + credentials["keys"]["secret"].encode("ascii") + b"========" + ) + receiver_pub_key = urlsafe_b64decode( + credentials["keys"]["public"].encode("ascii") + b"=" + ) + raw_data_encrypted = encrypt( + raw_data, + salt=salt, + private_key=sender_privkey, + dh=receiver_pub_key, + version="aesgcm", + auth_secret=sender_sec, + ) + + # Pass the crypto-key and salt WITHOUT padding, as they arrive on the wire. + assert len(sender_pub) % 4 != 0 # genuinely unpadded + raw_data_decrypted = FcmPushClient._decrypt_raw_data( + credentials, sender_pub, salt_str, raw_data_encrypted + ) + + assert raw_data_decrypted == raw_data