From e6f48401001609a8f99e71fcaf355fb895d508a8 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jos=C3=A9=20Padilla?= <jpadilla@users.noreply.github.com>
Date: Wed, 9 Sep 2026 09:46:01 -0400
Subject: [PATCH] Reject non-canonical compact JWS segments

---
 jwt/api_jws.py               | 35 +++++++++++++++++++++++------------
 tests/test_api_jws.py        | 11 +++++++++++
 tests/test_compressed_jwt.py |  8 ++++----
 3 files changed, 38 insertions(+), 16 deletions(-)

Index: pyjwt-2.12.1/jwt/api_jws.py
===================================================================
--- pyjwt-2.12.1.orig/jwt/api_jws.py
+++ pyjwt-2.12.1/jwt/api_jws.py
@@ -28,6 +28,9 @@ if TYPE_CHECKING:
     from .types import SigOptions
 
 _ALGORITHM_UNSET = object()
+_BASE64URL_ALPHABET = (
+    b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"
+)
 
 
 class PyJWS:
@@ -318,6 +321,23 @@ class PyJWS:
 
         return headers
 
+    @staticmethod
+    def _decode_base64url_segment(segment: bytes, name: str) -> bytes:
+        if len(segment) % 4 == 1 or any(
+            character not in _BASE64URL_ALPHABET for character in segment
+        ):
+            raise DecodeError(f"Invalid {name} padding")
+
+        try:
+            decoded = base64url_decode(segment)
+        except (TypeError, binascii.Error) as err:
+            raise DecodeError(f"Invalid {name} padding") from err
+
+        if base64url_encode(decoded) != segment:
+            raise DecodeError(f"Invalid {name} padding")
+
+        return decoded
+
     def _load(self, jwt: str | bytes) -> tuple[bytes, bytes, dict[str, Any], bytes]:
         if isinstance(jwt, str):
             jwt = jwt.encode("utf-8")
@@ -331,10 +351,7 @@ class PyJWS:
         except ValueError as err:
             raise DecodeError("Not enough segments") from err
 
-        try:
-            header_data = base64url_decode(header_segment)
-        except (TypeError, binascii.Error) as err:
-            raise DecodeError("Invalid header padding") from err
+        header_data = self._decode_base64url_segment(header_segment, "header")
 
         try:
             header: dict[str, Any] = json.loads(header_data)
@@ -358,15 +375,9 @@ class PyJWS:
                 )
             payload = b""
         else:
-            try:
-                payload = base64url_decode(payload_segment)
-            except (TypeError, binascii.Error) as err:
-                raise DecodeError("Invalid payload padding") from err
+            payload = self._decode_base64url_segment(payload_segment, "payload")
 
-        try:
-            signature = base64url_decode(crypto_segment)
-        except (TypeError, binascii.Error) as err:
-            raise DecodeError("Invalid crypto padding") from err
+        signature = self._decode_base64url_segment(crypto_segment, "crypto")
 
         return (payload, signing_input, header, signature)
 
Index: pyjwt-2.12.1/tests/test_api_jws.py
===================================================================
--- pyjwt-2.12.1.orig/tests/test_api_jws.py
+++ pyjwt-2.12.1/tests/test_api_jws.py
@@ -657,6 +657,17 @@ class TestJWS:
 
         assert "Invalid crypto padding" in str(exc.value)
 
+    def test_decode_rejects_non_canonical_crypto_segment(
+        self, jws: PyJWS, payload: bytes
+    ) -> None:
+        secret = "a" * 32
+        token = jws.encode(payload, secret, algorithm="HS256")
+        header, encoded_payload, signature = token.split(".")
+        mutated_token = ".".join((header, encoded_payload, f"{signature}!!!!"))
+
+        with pytest.raises(DecodeError, match="Invalid crypto padding"):
+            jws.decode(mutated_token, secret, algorithms=["HS256"])
+
     def test_decode_with_algo_none_should_fail(
         self, jws: PyJWS, payload: bytes
     ) -> None:
Index: pyjwt-2.12.1/tests/test_compressed_jwt.py
===================================================================
--- pyjwt-2.12.1.orig/tests/test_compressed_jwt.py
+++ pyjwt-2.12.1/tests/test_compressed_jwt.py
@@ -22,8 +22,8 @@ def test_decodes_complete_valid_jwt_with
     # Buffer.from(pako.deflateRaw('{"hello": "world"}')).toString('base64')
     example_jwt = (
         b"eyJhbGciOiAiSFMyNTYiLCAidHlwIjogIkpXVCJ9"
-        b".q1bKSM3JyVeyUlAqzy/KSVGqBQA="
-        b".08wHYeuh1rJXmcBcMrz6NxmbxAnCQp2rGTKfRNIkxiw="
+        b".q1bKSM3JyVeyUlAqzy_KSVGqBQA"
+        b".AAn1elCJC5MQCYFwTwa2tjjtyLgqLUVU-Y1vFBsU8jo"
     )
     decoded = CompressedPyJWT().decode_complete(
         example_jwt, example_secret, algorithms=["HS256"]
@@ -33,7 +33,7 @@ def test_decodes_complete_valid_jwt_with
         "header": {"alg": "HS256", "typ": "JWT"},
         "payload": example_payload,
         "signature": (
-            b"\xd3\xcc\x07a\xeb\xa1\xd6\xb2W\x99\xc0\\2\xbc\xfa7"
-            b"\x19\x9b\xc4\t\xc2B\x9d\xab\x192\x9fD\xd2$\xc6,"
+            b"\x00\t\xf5zP\x89\x0b\x93\x10\t\x81pO\x06\xb6\xb68"
+            b"\xed\xc8\xb8*-ET\xf9\x8do\x14\x1b\x14\xf2:"
         ),
     }
