Class swarmauri_crypto_jwe.JweCrypto.JweCrypto

swarmauri_crypto_jwe.JweCrypto.JweCrypto

Utility class for working with JSON Web Encryption.

The helpers create and parse compact JWE strings. All operations are asynchronous to integrate with async workflows.

encrypt_compact async

encrypt_compact(
    *,
    payload,
    alg,
    enc,
    key,
    kid=None,
    header_extra=None,
    aad=None,
)

Encrypt a payload into a compact JWE string.

payload (Union[bytes, str, Mapping[str, Any]]): Data to encrypt. Strings and mappings are encoded as UTF-8 JSON. alg (JWAAlg): Key management algorithm. enc (JWAAlg): Content encryption algorithm. key (Mapping[str, Any]): Key material used for encryption. kid (str): Optional key identifier placed in the protected header. header_extra (Mapping[str, Any]): Additional protected header fields. aad (Union[bytes, str]): Additional authenticated data. RETURNS (JWECompact): Compact JWE representation.

Source code in swarmauri_crypto_jwe/JweCrypto.py

async def encrypt_compact(
    self,
    *,
    payload: Union[bytes, str, Mapping[str, Any]],
    alg: JWAAlg,
    enc: JWAAlg,
    key: Mapping[str, Any],
    kid: Optional[str] = None,
    header_extra: Optional[Mapping[str, Any]] = None,
    aad: Optional[Union[bytes, str]] = None,
) -> JWECompact:
    """Encrypt a payload into a compact JWE string.

    payload (Union[bytes, str, Mapping[str, Any]]): Data to encrypt. Strings and
        mappings are encoded as UTF-8 JSON.
    alg (JWAAlg): Key management algorithm.
    enc (JWAAlg): Content encryption algorithm.
    key (Mapping[str, Any]): Key material used for encryption.
    kid (str): Optional key identifier placed in the protected header.
    header_extra (Mapping[str, Any]): Additional protected header fields.
    aad (Union[bytes, str]): Additional authenticated data.
    RETURNS (JWECompact): Compact JWE representation.
    """

    if isinstance(payload, str):
        pt = payload.encode("utf-8")
    elif isinstance(payload, (bytes, bytearray)):
        pt = bytes(payload)
    else:
        pt = _json_dumps(payload)

    protected: Dict[str, Any] = {"alg": alg.value, "enc": enc.value}
    if kid:
        protected["kid"] = kid
    if header_extra:
        protected.update(dict(header_extra))

    if protected.get("zip") == "DEF":
        pt = zlib.compress(pt)

    cek: bytes
    encrypted_key_b64 = ""
    epk_header: Optional[Dict[str, Any]] = None

    if alg == JWAAlg.DIR:
        secret = key.get("k")
        secret_b = (
            secret.encode("utf-8") if isinstance(secret, str) else bytes(secret)
        )
        if len(secret_b) != _cek_len_for_enc(enc):
            raise ValueError(
                f"'dir' key size must equal enc key size ({_cek_len_for_enc(enc)} bytes)"
            )
        cek = secret_b
    elif alg in (JWAAlg.RSA_OAEP, JWAAlg.RSA_OAEP_256):
        cek = _rand(_cek_len_for_enc(enc))
        pk = _load_rsa_public(key.get("pub") or key)
        hash_alg = _hash_for_oaep(alg)
        ekey = pk.encrypt(
            cek,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hash_alg),
                algorithm=hash_alg,
                label=None,
            ),
        )
        encrypted_key_b64 = _b64u(ekey)
    elif alg == JWAAlg.ECDH_ES:
        crv, rpk = _load_ecdh_recipient_public(key.get("pub") or key)
        if crv == "X25519":
            esk = x25519.X25519PrivateKey.generate()
            epk = esk.public_key()
            z = esk.exchange(rpk)  # type: ignore[arg-type]
            epk_header = _x25519_jwk_from_public_key(epk)
        else:
            curve = {
                "P-256": ec.SECP256R1(),
                "P-384": ec.SECP384R1(),
                "P-521": ec.SECP521R1(),
            }[crv]
            esk = ec.generate_private_key(curve)
            epk = esk.public_key()
            z = esk.exchange(ec.ECDH(), rpk)  # type: ignore[arg-type]
            epk_header = _ec_jwk_from_public_key(epk)
        apu_b = None
        apv_b = None
        if "apu" in (header_extra or {}):
            apu_b = (
                _b64u_dec(header_extra["apu"])
                if isinstance(header_extra["apu"], str)
                else header_extra["apu"]
            )
        if "apv" in (header_extra or {}):
            apv_b = (
                _b64u_dec(header_extra["apv"])
                if isinstance(header_extra["apv"], str)
                else header_extra["apv"]
            )
        cek = _concat_kdf(z, enc, hashes.SHA256(), apu_b, apv_b)
        protected["epk"] = epk_header
    else:
        raise ValueError(f"Unsupported alg '{alg.value}'")

    iv = _rand(12)
    aadd = _ensure_bytes(aad) if aad is not None else None

    protected_b64 = _b64u(_json_dumps(protected))
    aead_aad = _compute_aad(protected_b64, aadd)

    aesgcm = AESGCM(cek)
    ct = aesgcm.encrypt(iv, pt, aead_aad)
    ciphertext, tag = ct[:-16], ct[-16:]

    return ".".join(
        [
            protected_b64,
            encrypted_key_b64,
            _b64u(iv),
            _b64u(ciphertext),
            _b64u(tag),
        ]
    )

decrypt_compact async

decrypt_compact(
    jwe,
    *,
    dir_key=None,
    rsa_private_pem=None,
    rsa_private_password=None,
    ecdh_private_key=None,
    expected_algs=None,
    expected_encs=None,
    aad=None,
)

Decrypt a compact JWE string.

jwe (JWECompact): Serialized JWE to decode. dir_key (Union[bytes, str]): Symmetric key when alg='dir' is used. rsa_private_pem (Union[str, bytes]): RSA private key in PEM encoding for RSA-OAEP algorithms. rsa_private_password (Union[str, bytes]): Password for the RSA key. ecdh_private_key (Any): Private key for ECDH-ES. expected_algs (Iterable[JWAAlg]): Allowed algorithm values. expected_encs (Iterable[JWAAlg]): Allowed encryption values. aad (Union[bytes, str]): Additional authenticated data. RETURNS (JweDecryptResult): Header and plaintext of the decrypted token.

Source code in swarmauri_crypto_jwe/JweCrypto.py

async def decrypt_compact(
    self,
    jwe: JWECompact,
    *,
    dir_key: Optional[Union[bytes, str]] = None,
    rsa_private_pem: Optional[Union[str, bytes]] = None,
    rsa_private_password: Optional[Union[str, bytes]] = None,
    ecdh_private_key: Optional[Any] = None,
    expected_algs: Optional[Iterable[JWAAlg]] = None,
    expected_encs: Optional[Iterable[JWAAlg]] = None,
    aad: Optional[Union[bytes, str]] = None,
) -> JweDecryptResult:
    """Decrypt a compact JWE string.

    jwe (JWECompact): Serialized JWE to decode.
    dir_key (Union[bytes, str]): Symmetric key when ``alg='dir'`` is used.
    rsa_private_pem (Union[str, bytes]): RSA private key in PEM encoding for
        RSA-OAEP algorithms.
    rsa_private_password (Union[str, bytes]): Password for the RSA key.
    ecdh_private_key (Any): Private key for ECDH-ES.
    expected_algs (Iterable[JWAAlg]): Allowed algorithm values.
    expected_encs (Iterable[JWAAlg]): Allowed encryption values.
    aad (Union[bytes, str]): Additional authenticated data.
    RETURNS (JweDecryptResult): Header and plaintext of the decrypted token.
    """

    parts = jwe.split(".")
    if len(parts) != 5:
        raise ValueError("Invalid JWE compact: expected 5 dot-separated parts.")
    b64_prot, b64_ekey, b64_iv, b64_ct, b64_tag = parts

    header = json.loads(_b64u_dec(b64_prot))
    try:
        alg = JWAAlg(header.get("alg"))
        enc = JWAAlg(header.get("enc"))
    except Exception as exc:
        raise ValueError("JWE header missing 'alg' or 'enc'.") from exc
    if expected_algs and alg not in set(expected_algs):
        raise ValueError(f"Unexpected alg '{alg.value}'.")
    if expected_encs and enc not in set(expected_encs):
        raise ValueError(f"Unexpected enc '{enc.value}'.")

    iv = _b64u_dec(b64_iv)
    if len(iv) != 12:
        raise ValueError("Invalid IV length for AES-GCM (must be 96-bit).")
    ciphertext = _b64u_dec(b64_ct)
    tag = _b64u_dec(b64_tag)
    aadd = _ensure_bytes(aad) if aad is not None else None
    aead_aad = _compute_aad(b64_prot, aadd)

    if alg == JWAAlg.DIR:
        if dir_key is None:
            raise ValueError("dir_key is required for alg='dir'.")
        cek = (
            dir_key.encode("utf-8") if isinstance(dir_key, str) else bytes(dir_key)
        )
        if len(cek) != _cek_len_for_enc(enc):
            raise ValueError("dir_key length mismatch for enc.")
    elif alg in (JWAAlg.RSA_OAEP, JWAAlg.RSA_OAEP_256):
        if rsa_private_pem is None:
            raise ValueError("rsa_private_pem is required for RSA-OAEP decryption.")
        sk = _load_rsa_private(rsa_private_pem, password=rsa_private_password)
        ekey = _b64u_dec(b64_ekey)
        hash_alg = _hash_for_oaep(alg)
        cek = sk.decrypt(
            ekey,
            padding.OAEP(
                mgf=padding.MGF1(algorithm=hash_alg), algorithm=hash_alg, label=None
            ),
        )
    elif alg == JWAAlg.ECDH_ES:
        if ecdh_private_key is None:
            raise ValueError("ecdh_private_key is required for ECDH-ES decryption.")
        epk = header.get("epk")
        if not isinstance(epk, Mapping):
            raise ValueError("Missing/invalid 'epk' in JWE header for ECDH-ES.")
        if epk.get("kty") == "OKP" and epk.get("crv") == "X25519":
            if not isinstance(ecdh_private_key, x25519.X25519PrivateKey):
                raise TypeError("ECDH-ES with X25519 requires an X25519PrivateKey.")
            z = ecdh_private_key.exchange(
                x25519.X25519PublicKey.from_public_bytes(_b64u_dec(epk["x"]))
            )
        elif epk.get("kty") == "EC":
            crv = epk.get("crv")
            curve = {
                "P-256": ec.SECP256R1(),
                "P-384": ec.SECP384R1(),
                "P-521": ec.SECP521R1(),
            }.get(crv)
            if curve is None:
                raise ValueError(f"Unsupported EC curve in epk: {crv}")
            if not isinstance(ecdh_private_key, ec.EllipticCurvePrivateKey):
                raise TypeError(
                    "ECDH-ES with EC requires an EllipticCurvePrivateKey."
                )
            x = int.from_bytes(_b64u_dec(epk["x"]), "big")
            y = int.from_bytes(_b64u_dec(epk["y"]), "big")
            rpk = ec.EllipticCurvePublicNumbers(x, y, curve).public_key()
            z = ecdh_private_key.exchange(ec.ECDH(), rpk)
        else:
            raise ValueError("Unsupported 'epk' kty/crv.")
        apu_b = _b64u_dec(header["apu"]) if "apu" in header else None
        apv_b = _b64u_dec(header["apv"]) if "apv" in header else None
        cek = _concat_kdf(z, enc, hashes.SHA256(), apu_b, apv_b)
    else:
        raise ValueError(f"Unsupported alg '{alg.value}'")

    aesgcm = AESGCM(cek)
    try:
        pt = aesgcm.decrypt(iv, ciphertext + tag, aead_aad)
    except Exception as exc:  # noqa: BLE001
        raise ValueError(
            "JWE decryption failed (invalid tag or wrong key)."
        ) from exc

    if header.get("zip") == "DEF":
        try:
            pt = zlib.decompress(pt)
        except Exception as exc:  # noqa: BLE001
            raise ValueError(f"Failed to decompress (zip=DEF): {exc}") from exc

    return JweDecryptResult(header=header, plaintext=pt)