fix: reject crit header in unprotected headers

Author: lepture

src/joserfc/_rfc7515/json.py

@@ -12,6 +12,7 @@
     FlattenedJSONSerialization,
 )
 from .registry import JWSRegistry
+from ..registry import reject_unprotected_crit_header
 from ..util import (
     json_b64encode,
     json_b64decode,
@@ -65,6 +66,7 @@ def sign_flattened_json(
 def sign_json_member(
     payload_segment: bytes, member: HeaderMember, registry: JWSRegistry, find_key: FindKey
 ) -> JSONSignatureDict:
+    reject_unprotected_crit_header(member.header)
     headers = member.headers()
     registry.check_header(headers)
     alg = registry.get_alg(headers["alg"])
@@ -131,6 +133,7 @@ def verify_signature(
     registry: JWSRegistry,
     find_key: FindKey,
 ) -> bool:
+    reject_unprotected_crit_header(member.header)
     headers = member.headers()
     registry.check_header(headers)
     alg = registry.get_alg(headers["alg"])

src/joserfc/jwe.py

@@ -25,7 +25,7 @@
 from .jwa import setup_jwe_algorithms
 from .jwk import Key, KeySet, ECKey, OKPKey, KeyFlexible, guess_key
 from .util import to_bytes
-from .registry import Header
+from .registry import Header, reject_unprotected_crit_header
 
 __all__ = [
     # types
@@ -195,6 +195,7 @@ def encrypt_json(
     elif registry is None:
         registry = default_registry
 
+    reject_unprotected_crit_header(obj.unprotected)
     for recipient in obj.recipients:
         if sender_key and not recipient.sender_key:
             recipient.sender_key = _guess_sender_key(recipient, sender_key, True)
@@ -232,6 +233,7 @@ def decrypt_json(
     elif registry is None:
         registry = default_registry
 
+    reject_unprotected_crit_header(data.get("unprotected"))
     if "recipients" in data:
         general_obj = extract_general_json(data)  # type: ignore[arg-type]
         _attach_recipient_keys(general_obj.recipients, private_key, sender_key)

src/joserfc/registry.py

@@ -211,3 +211,8 @@ def check_crit_header(registry: HeaderRegistryDict, header: Header) -> None:
         raise MissingCritHeaderError(",".join(missing_crit_headers))
     elif unsupported_crit_headers:
         raise UnsupportedHeaderError(f"Unsupported {unsupported_crit_headers} in header")
+
+
+def reject_unprotected_crit_header(unprotected: Header) -> None:
+    if unprotected and "crit" in unprotected:
+        raise UnsupportedHeaderError("'crit' header MUST be protected header")

tests/jws/test_rfc7797.py

@@ -5,6 +5,7 @@
     BadSignatureError,
     InvalidHeaderValueError,
     MissingCritHeaderError,
+    UnsupportedHeaderError,
 )
 from joserfc.util import to_bytes
 from joserfc.jws import HeaderDict
@@ -70,10 +71,7 @@ def test_compact_use_registry(self):
     def test_json_without_protected_header(self):
         header = {"alg": "HS256", "b64": False, "crit": ["b64"]}
         member: HeaderDict = {"header": header}
-        value = jws.serialize_json(member, "hello", default_key)
-        obj = jws.deserialize_json(value, default_key)
-        self.assertTrue(obj.flattened)
-        self.assertEqual(obj.headers(), header)
+        self.assertRaises(UnsupportedHeaderError, jws.serialize_json, member, "hello", default_key)
 
     def test_general_json(self):
         member: HeaderDict = {"protected": {"alg": "HS256"}}