Method deserialize_json

authlib/jose/rfc7516/jwe.py:513–660

Extract JWE JSON Serialization. :param obj: JWE JSON Serialization as dict or str :param key: Private key used to decrypt payload (optionally can be a tuple of kid and essentially key) :param decode: Function to decode payload data :param sender_key: Send

(self, obj, key, decode=None, sender_key=None)

        return {"header": protected, "payload": payload}

    def deserialize_json(self, obj, key, decode=None, sender_key=None):  # noqa: C901
        """Extract JWE JSON Serialization.

        :param obj: JWE JSON Serialization as dict or str
        :param key: Private key used to decrypt payload
            (optionally can be a tuple of kid and essentially key)
        :param decode: Function to decode payload data
        :param sender_key: Sender's public key in case
            JWEAlgorithmWithTagAwareKeyAgreement is used
        :return: dict with `header` and `payload` keys where `header` value is
            a dict containing `protected`, `unprotected`, `recipients` and/or
            `aad` keys
        """
        obj = ensure_dict(obj, "JWE")
        obj = deepcopy(obj)

        if "protected" in obj:
            protected = extract_header(to_bytes(obj["protected"]), DecodeError)
        else:
            protected = None

        unprotected = obj.get("unprotected")

        recipients = obj["recipients"]
        for recipient in recipients:
            if "header" not in recipient:
                recipient["header"] = {}
            recipient["encrypted_key"] = extract_segment(
                to_bytes(recipient["encrypted_key"]), DecodeError, "encrypted key"
            )

        if "aad" in obj:
            jwe_aad = extract_segment(to_bytes(obj["aad"]), DecodeError, "JWE AAD")
        else:
            jwe_aad = None

        iv = extract_segment(to_bytes(obj["iv"]), DecodeError, "initialization vector")

        ciphertext = extract_segment(
            to_bytes(obj["ciphertext"]), DecodeError, "ciphertext"
        )

        tag = extract_segment(to_bytes(obj["tag"]), DecodeError, "authentication tag")

        shared_header = JWESharedHeader(protected, unprotected)

        alg = self.get_header_alg(shared_header)
        enc = self.get_header_enc(shared_header)
        zip_alg = self.get_header_zip(shared_header)

        self._validate_sender_key(sender_key, alg)
        self._validate_private_headers(shared_header, alg)
        for recipient in recipients:
            self._validate_private_headers(recipient["header"], alg)

        kid = None
        if isinstance(key, tuple) and len(key) == 2:
            # Extract separately provided kid and essentially key