Harden encrypted push payload decoding

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"))

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