From 675fa10db578886ee6cfd1df688236f69560ced4 Mon Sep 17 00:00:00 2001
From: Daniel Miles <daniel.t.miles@gmail.com>
Date: Thu, 19 May 2022 07:34:19 -0700
Subject: adding support for compressed payloads (#753)

* adding support for compressed payloads

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* adding test to cover all lines in patch

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* addressing flake8 unused variable and cyclomatic complexity complaints

* expanding test for better coverage

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
---
 jwt/api_jwt.py        | 23 ++++++++++++++++++++++-
 tests/test_api_jwt.py | 33 +++++++++++++++++++++++++++++++++
 2 files changed, 55 insertions(+), 1 deletion(-)

diff --git a/jwt/api_jwt.py b/jwt/api_jwt.py
index 7d2177b..f04a88d 100644
--- a/jwt/api_jwt.py
+++ b/jwt/api_jwt.py
@@ -1,5 +1,6 @@
 import json
 import warnings
+import zlib
 from calendar import timegm
 from collections.abc import Iterable, Mapping
 from datetime import datetime, timedelta, timezone
@@ -108,7 +109,7 @@ class PyJWT:
         try:
             payload = json.loads(decoded["payload"])
         except ValueError as e:
-            raise DecodeError(f"Invalid payload string: {e}")
+            payload = self._decompress_payload(decoded["payload"], e)
         if not isinstance(payload, dict):
             raise DecodeError("Invalid payload string: must be a json object")
 
@@ -118,6 +119,26 @@ class PyJWT:
         decoded["payload"] = payload
         return decoded
 
+    @staticmethod
+    def _decompress_payload(payload, e):
+        """
+        Smart Health cards use a raw-compressed (no header or crc) payload,
+        so before surfacing a UnicodeDecodeError, find out if it can be
+        uncompressed successfully
+        noinspection PyBroadException
+        """
+        if isinstance(e, UnicodeDecodeError):
+            try:
+                payload = json.loads(
+                    # wbits=-15 has zlib not worry about headers of crc's
+                    zlib.decompress(payload, wbits=-15).decode("utf-8")
+                )
+            except Exception:
+                payload = None
+        if payload is not None:
+            return payload
+        raise DecodeError(f"Invalid payload string: {e}")
+
     def decode(
         self,
         jwt: str,
diff --git a/tests/test_api_jwt.py b/tests/test_api_jwt.py
index 84e41e0..491ef9a 100644
--- a/tests/test_api_jwt.py
+++ b/tests/test_api_jwt.py
@@ -64,6 +64,39 @@ class TestJWT:
             ),
         }
 
+    def test_decodes_complete_valid_jwt_with_compressed_payload(self, jwt):
+        example_payload = {"hello": "world"}
+        example_secret = "secret"
+        # payload made with the pako (https://nodeca.github.io/pako/) library in Javascript:
+        # Buffer.from(pako.deflateRaw('{"hello": "world"}')).toString('base64')
+        example_jwt = (
+            b"eyJhbGciOiAiSFMyNTYiLCAidHlwIjogIkpXVCJ9"
+            b".q1bKSM3JyVeyUlAqzy/KSVGqBQA="
+            b".08wHYeuh1rJXmcBcMrz6NxmbxAnCQp2rGTKfRNIkxiw="
+        )
+        decoded = jwt.decode_complete(example_jwt, example_secret, algorithms=["HS256"])
+
+        assert decoded == {
+            "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,"
+            ),
+        }
+
+    def test_decodes_complete_valid_jwt_with_invalid_compressed_payload(self, jwt):
+        # payload made with the pako (https://nodeca.github.io/pako/) library in Javascript:
+        # Buffer.from(pako.deflateRaw('{"this is": "not valid" json')).toString('base64')
+        example_secret = "secret"
+        example_jwt = (
+            b"eyJhbGciOiAiSFMyNTYiLCAidHlwIjogIkpXVCJ9"
+            b".q1YqycgsVsgsVrJSUMrLL1EoS8zJTFFSyCrOzwMA"
+            b".SIAxephyoWBJrw+KdBfksJlhdg+mQtm+vjaRXV1qGJ4="
+        )
+        with pytest.raises(DecodeError):
+            jwt.decode_complete(example_jwt, example_secret, algorithms=["HS256"])
+
     def test_load_verify_valid_jwt(self, jwt):
         example_payload = {"hello": "world"}
         example_secret = "secret"
-- 
cgit v1.2.1