Authlib: JWS/JWT accepts unknown crit headers (RFC violation → possible authz bypass)

@@ -33,6 +33,14 @@ def __init__(self, name):
         super().__init__(description=description)
 
 
+class InvalidCritHeaderParameterNameError(JoseError):
+    error = "invalid_crit_header_parameter_name"
+
+    def __init__(self, name):
+        description = f"Invalid Header Parameter Name: {name}"
+        super().__init__(description=description)
+
+
 class InvalidEncryptionAlgorithmForECDH1PUWithKeyWrappingError(JoseError):
     error = "invalid_encryption_algorithm_for_ECDH_1PU_with_key_wrapping"
 

@@ -4,6 +4,7 @@
 from authlib.common.encoding import urlsafe_b64encode
 from authlib.jose.errors import BadSignatureError
 from authlib.jose.errors import DecodeError
+from authlib.jose.errors import InvalidCritHeaderParameterNameError
 from authlib.jose.errors import InvalidHeaderParameterNameError
 from authlib.jose.errors import MissingAlgorithmError
 from authlib.jose.errors import UnsupportedAlgorithmError
@@ -64,6 +65,7 @@ def serialize_compact(self, protected, payload, key):
         """
         jws_header = JWSHeader(protected, None)
         self._validate_private_headers(protected)
+        self._validate_crit_headers(protected)
         algorithm, key = self._prepare_algorithm_key(protected, payload, key)
 
         protected_segment = json_b64encode(jws_header.protected)
@@ -95,6 +97,7 @@ def deserialize_compact(self, s, key, decode=None):
             raise DecodeError("Not enough segments") from exc
 
         protected = _extract_header(protected_segment)
+        self._validate_crit_headers(protected)
         jws_header = JWSHeader(protected, None)
 
         payload = _extract_payload(payload_segment)
@@ -132,6 +135,11 @@ def serialize_json(self, header_obj, payload, key):
 
         def _sign(jws_header):
             self._validate_private_headers(jws_header)
+            # RFC 7515 §4.1.11: 'crit' MUST be integrity-protected.
+            # Reject if present in unprotected header, and validate only
+            # against the protected header parameters.
+            self._reject_unprotected_crit(jws_header.header)
+            self._validate_crit_headers(jws_header.protected)
             _alg, _key = self._prepare_algorithm_key(jws_header, payload, key)
 
             protected_segment = json_b64encode(jws_header.protected)
@@ -272,6 +280,28 @@ def _validate_private_headers(self, header):
                 if k not in names:
                     raise InvalidHeaderParameterNameError(k)
 
+    def _reject_unprotected_crit(self, unprotected_header):
+        """Reject 'crit' when found in the unprotected header (RFC 7515 §4.1.11)."""
+        if unprotected_header and "crit" in unprotected_header:
+            raise InvalidHeaderParameterNameError("crit")
+
+    def _validate_crit_headers(self, header):
+        if "crit" in header:
+            crit_headers = header["crit"]
+            # Type enforcement for robustness and predictable errors
+            if not isinstance(crit_headers, list) or not all(
+                isinstance(x, str) for x in crit_headers
+            ):
+                raise InvalidHeaderParameterNameError("crit")
+            names = self.REGISTERED_HEADER_PARAMETER_NAMES.copy()
+            if self._private_headers:
+                names = names.union(self._private_headers)
+            for k in crit_headers:
+                if k not in names:
+                    raise InvalidCritHeaderParameterNameError(k)
+                elif k not in header:
+                    raise InvalidCritHeaderParameterNameError(k)
+
     def _validate_json_jws(self, payload_segment, payload, header_obj, key):
         protected_segment = header_obj.get("protected")
         if not protected_segment:
@@ -286,7 +316,14 @@ def _validate_json_jws(self, payload_segment, payload, header_obj, key):
         header = header_obj.get("header")
         if header and not isinstance(header, dict):
             raise DecodeError('Invalid "header" value')
-
+        # RFC 7515 §4.1.11: 'crit' MUST be integrity-protected. If present in
+        # the unprotected header object, reject the JWS.
+        self._reject_unprotected_crit(header)
+
+        # Enforce must-understand semantics for names listed in protected
+        # 'crit'. This will also ensure each listed name is present in the
+        # protected header.
+        self._validate_crit_headers(protected)
         jws_header = JWSHeader(protected, header)
         algorithm, key = self._prepare_algorithm_key(jws_header, payload, key)
         signing_input = b".".join([protected_segment, payload_segment])

@@ -226,6 +226,55 @@ def test_validate_header():
     assert isinstance(s, dict)
 
 
+def test_validate_crit_header_with_serialize():
+    jws = JsonWebSignature()
+    protected = {"alg": "HS256", "kid": "1", "crit": ["kid"]}
+    jws.serialize(protected, b"hello", "secret")
+
+    protected = {"alg": "HS256", "crit": ["kid"]}
+    with pytest.raises(errors.InvalidCritHeaderParameterNameError):
+        jws.serialize(protected, b"hello", "secret")
+
+    protected = {"alg": "HS256", "invalid": "1", "crit": ["invalid"]}
+    with pytest.raises(errors.InvalidCritHeaderParameterNameError):
+        jws.serialize(protected, b"hello", "secret")
+
+
+def test_validate_crit_header_with_deserialize():
+    jws = JsonWebSignature()
+    case1 = "eyJhbGciOiJIUzI1NiIsImNyaXQiOlsia2lkIl19.aGVsbG8.RVimhJH2LRGAeHy0ZcbR9xsgKhzhxIBkHs7S_TDgWvc"
+    with pytest.raises(errors.InvalidCritHeaderParameterNameError):
+        jws.deserialize(case1, "secret")
+
+    case2 = (
+        "eyJhbGciOiJIUzI1NiIsImludmFsaWQiOiIxIiwiY3JpdCI6WyJpbnZhbGlkIl19."
+        "aGVsbG8.ifW_D1AQWzggrpd8npcnmpiwMD9dp5FTX66lCkYFENM"
+    )
+    with pytest.raises(errors.InvalidCritHeaderParameterNameError):
+        jws.deserialize(case2, "secret")
+
+
+def test_unprotected_crit_rejected_in_json_serialize():
+    jws = JsonWebSignature()
+    protected = {"alg": "HS256", "kid": "a"}
+    # Place 'crit' in unprotected header; must be rejected
+    header = {"protected": protected, "header": {"kid": "a", "crit": ["kid"]}}
+    with pytest.raises(errors.InvalidHeaderParameterNameError):
+        jws.serialize_json(header, b"hello", "secret")
+
+
+def test_unprotected_crit_rejected_in_json_deserialize():
+    jws = JsonWebSignature()
+    protected = {"alg": "HS256", "kid": "a"}
+    header = {"protected": protected, "header": {"kid": "a"}}
+    data = jws.serialize_json(header, b"hello", "secret")
+    # Tamper by adding 'crit' into the unprotected header; must be rejected
+    data_tampered = dict(data)
+    data_tampered["header"] = {"kid": "a", "crit": ["kid"]}
+    with pytest.raises(errors.InvalidHeaderParameterNameError):
+        jws.deserialize_json(data_tampered, "secret")
+
+
 def test_ES512_alg():
     jws = JsonWebSignature()
     private_key = read_file_path("secp521r1-private.json")