Class swarmauri_signing_jws.JwsSignerVerifier.JwsSignerVerifier

swarmauri_signing_jws.JwsSignerVerifier.JwsSignerVerifier

JwsSignerVerifier(
    *,
    hmac_signer=None,
    rsa_signer=None,
    ecdsa_signer=None,
    eddsa_signer=None,
    k1_signer=None,
)

Source code in swarmauri_signing_jws/JwsSignerVerifier.py

def __init__(
    self,
    *,
    hmac_signer: Optional[HmacEnvelopeSigner] = None,
    rsa_signer: Optional[RSAEnvelopeSigner] = None,
    ecdsa_signer: Optional[EcdsaEnvelopeSigner] = None,
    eddsa_signer: Optional[Ed25519EnvelopeSigner] = None,
    k1_signer: Optional[Any] = None,
) -> None:
    self.hmac = hmac_signer or HmacEnvelopeSigner()
    self.rsa = rsa_signer or RSAEnvelopeSigner()
    self.ecdsa = ecdsa_signer or EcdsaEnvelopeSigner()
    self.eddsa = eddsa_signer or Ed25519EnvelopeSigner()
    self.k1 = k1_signer or (Secp256k1EnvelopeSigner() if _HAS_K1 else None)

hmac instance-attribute

hmac = hmac_signer or HmacEnvelopeSigner()

rsa instance-attribute

rsa = rsa_signer or RSAEnvelopeSigner()

ecdsa instance-attribute

ecdsa = ecdsa_signer or EcdsaEnvelopeSigner()

eddsa instance-attribute

eddsa = eddsa_signer or Ed25519EnvelopeSigner()

k1 instance-attribute

k1 = k1_signer or (
    Secp256k1EnvelopeSigner() if _HAS_K1 else None
)

sign_compact async

sign_compact(
    *,
    payload,
    alg,
    key,
    kid=None,
    header_extra=None,
    typ=None,
)

Source code in swarmauri_signing_jws/JwsSignerVerifier.py

async def sign_compact(
    self,
    *,
    payload: Union[bytes, str, Mapping[str, Any]],
    alg: JWAAlg | str,
    key: Mapping[str, Any],
    kid: Optional[str] = None,
    header_extra: Optional[Mapping[str, Any]] = None,
    typ: Optional[str] = None,
) -> JWSCompact:
    if isinstance(payload, str):
        payload_b = payload.encode("utf-8")
    elif isinstance(payload, (bytes, bytearray)):
        payload_b = bytes(payload)
    else:
        payload_b = json.dumps(
            payload, separators=(",", ":"), ensure_ascii=False
        ).encode("utf-8")

    alg_token = alg if isinstance(alg, JWAAlg) else JWAAlg(str(alg))
    header: Dict[str, Any] = {"alg": alg_token.value}
    if kid:
        header["kid"] = kid
    if typ:
        header["typ"] = typ
    if header_extra:
        header.update(dict(header_extra))

    b64_header = _b64u(
        json.dumps(header, separators=(",", ":"), ensure_ascii=False).encode(
            "utf-8"
        )
    )
    b64_payload = _b64u(payload_b)
    signing_input = f"{b64_header}.{b64_payload}".encode("ascii")

    sig_bytes = await self._sign_for_alg(signing_input, alg_token, key)
    b64sig = _b64u(sig_bytes)
    return f"{b64_header}.{b64_payload}.{b64sig}"

verify_compact async

verify_compact(
    jws,
    *,
    hmac_keys=None,
    rsa_pubkeys=None,
    ec_pubkeys=None,
    ed_pubkeys=None,
    k1_pubkeys=None,
    jwks_resolver=None,
    alg_allowlist=None,
)

Source code in swarmauri_signing_jws/JwsSignerVerifier.py

async def verify_compact(
    self,
    jws: JWSCompact,
    *,
    hmac_keys: Optional[Sequence[Mapping[str, Any]]] = None,
    rsa_pubkeys: Optional[Sequence[Any]] = None,
    ec_pubkeys: Optional[Sequence[Any]] = None,
    ed_pubkeys: Optional[Sequence[Any]] = None,
    k1_pubkeys: Optional[Sequence[Any]] = None,
    jwks_resolver: Optional[callable] = None,
    alg_allowlist: Optional[Iterable[JWAAlg | str]] = None,
) -> JwsResult:
    parts = jws.split(".")
    if len(parts) != 3:
        raise ValueError("Invalid JWS compact serialization.")
    b64_header, b64_payload, b64_sig = parts
    try:
        header = json.loads(_b64u_dec(b64_header))
    except Exception as e:  # pragma: no cover - invalid header
        raise ValueError(f"Invalid JWS header: {e}")
    alg_raw = header.get("alg")
    kid = header.get("kid")
    if not isinstance(alg_raw, str):
        raise ValueError("Missing/invalid 'alg' in JWS header.")
    alg = JWAAlg(alg_raw)
    if alg_allowlist:
        allowed = {
            a if isinstance(a, JWAAlg) else JWAAlg(str(a)) for a in alg_allowlist
        }
        if alg not in allowed:
            raise ValueError(f"Rejected alg '{alg.value}' (not in allowlist).")

    payload_b = _b64u_dec(b64_payload)
    sig = _b64u_dec(b64_sig)
    signing_input = f"{b64_header}.{b64_payload}".encode("ascii")

    if jwks_resolver:
        jwk = jwks_resolver(kid, alg.value)
        ok = await self._verify_one(signing_input, alg, sig, jwk=jwk)
        if not ok:
            raise ValueError("Invalid JWS signature.")
        return JwsResult(header=header, payload=payload_b)

    if _is_hmac(alg):
        keys = hmac_keys or []
        if not keys:
            raise ValueError("No HMAC keys provided.")
        ok = await self.hmac.verify_bytes(
            signing_input,
            [{"alg": alg.value, "sig": sig}],
            require={"min_signers": 1, "algs": [alg]},
            opts={"keys": list(keys)},
        )
        if not ok:
            raise ValueError("Invalid JWS (HMAC) signature.")
        return JwsResult(header=header, payload=payload_b)

    if _is_rsa(alg):
        if not rsa_pubkeys:
            raise ValueError("No RSA public keys provided.")
        ok = await self.rsa.verify_bytes(
            signing_input,
            [{"alg": _map_rsa_alg(alg), "sig": sig}],
            require={"min_signers": 1, "algs": [_map_rsa_alg(alg)]},
            opts={"pubkeys": list(rsa_pubkeys)},
        )
        if not ok:
            raise ValueError("Invalid JWS (RSA) signature.")
        return JwsResult(header=header, payload=payload_b)

    if _is_ec(alg):
        if not ec_pubkeys:
            raise ValueError("No EC public keys provided.")
        ok = await self.ecdsa.verify_bytes(
            signing_input,
            [{"alg": alg.value, "sig": sig, "sigfmt": "raw"}],
            require={"min_signers": 1, "algs": [alg.value]},
            opts={"pubkeys": list(ec_pubkeys)},
        )
        if not ok:
            raise ValueError("Invalid JWS (ECDSA) signature.")
        return JwsResult(header=header, payload=payload_b)

    if _is_k1(alg):
        if not self.k1:
            raise ValueError(
                "ES256K requested but secp256k1 signer is not available"
            )
        if not k1_pubkeys:
            raise ValueError("No secp256k1 public keys provided.")
        ok = await self.k1.verify_bytes(
            signing_input,
            [{"alg": "ES256K", "sig": sig, "sigfmt": "raw"}],
            require={"min_signers": 1, "algs": ["ES256K"]},
            opts={"pubkeys": list(k1_pubkeys)},
        )
        if not ok:
            raise ValueError("Invalid JWS (ES256K) signature.")
        return JwsResult(header=header, payload=payload_b)

    if _is_eddsa(alg):
        pubs = []
        for x in ed_pubkeys or []:
            if isinstance(x, dict) and x.get("kty") == "OKP":
                pubs.append(_jwk_to_pub_for_signer(x))
            else:
                pubs.append(x)
        if not pubs:
            raise ValueError("No Ed25519 public keys provided.")
        ok = await self.eddsa.verify_bytes(
            signing_input,
            [{"alg": "Ed25519", "sig": sig}],
            require={"min_signers": 1, "algs": ["Ed25519"]},
            opts={"pubkeys": pubs},
        )
        if not ok:
            raise ValueError("Invalid JWS (EdDSA) signature.")
        return JwsResult(header=header, payload=payload_b)

    raise ValueError(f"Unsupported alg: {alg}")

sign_general_json async

sign_general_json(*, payload, signatures)

Source code in swarmauri_signing_jws/JwsSignerVerifier.py

async def sign_general_json(
    self,
    *,
    payload: Union[bytes, str, Mapping[str, Any]],
    signatures: Sequence[
        Tuple[
            JWAAlg | str,
            Mapping[str, Any],
            Optional[str],
            Optional[Mapping[str, Any]],
        ]
    ],
) -> Dict[str, Any]:
    if isinstance(payload, str):
        payload_b = payload.encode("utf-8")
    elif isinstance(payload, (bytes, bytearray)):
        payload_b = bytes(payload)
    else:
        payload_b = json.dumps(
            payload, separators=(",", ":"), ensure_ascii=False
        ).encode("utf-8")

    b64_payload = _b64u(payload_b)
    out_sigs = []
    for alg, key, kid, header_extra in signatures:
        alg_token = alg if isinstance(alg, JWAAlg) else JWAAlg(str(alg))
        header = {"alg": alg_token.value}
        if kid:
            header["kid"] = kid
        if header_extra:
            header.update(dict(header_extra))
        b64_hdr = _b64u(
            json.dumps(header, separators=(",", ":"), ensure_ascii=False).encode(
                "utf-8"
            )
        )
        signing_input = f"{b64_hdr}.{b64_payload}".encode("ascii")
        sig_bytes = await self._sign_for_alg(signing_input, alg_token, key)
        out_sigs.append({"protected": b64_hdr, "signature": _b64u(sig_bytes)})
    return {"payload": b64_payload, "signatures": out_sigs}

verify_general_json async

verify_general_json(
    jws_obj,
    *,
    hmac_keys=None,
    rsa_pubkeys=None,
    ec_pubkeys=None,
    ed_pubkeys=None,
    k1_pubkeys=None,
    jwks_resolver=None,
    require_any=None,
    min_signers=1,
)

Source code in swarmauri_signing_jws/JwsSignerVerifier.py

async def verify_general_json(
    self,
    jws_obj: Mapping[str, Any],
    *,
    hmac_keys: Optional[Sequence[Mapping[str, Any]]] = None,
    rsa_pubkeys: Optional[Sequence[Any]] = None,
    ec_pubkeys: Optional[Sequence[Any]] = None,
    ed_pubkeys: Optional[Sequence[Any]] = None,
    k1_pubkeys: Optional[Sequence[Any]] = None,
    jwks_resolver: Optional[callable] = None,
    require_any: Optional[Iterable[JWAAlg | str]] = None,
    min_signers: int = 1,
) -> Tuple[int, bytes]:
    b64_payload = jws_obj.get("payload")
    if not isinstance(b64_payload, str):
        raise ValueError("Invalid JWS JSON: missing payload.")
    sigs = jws_obj.get("signatures")
    if not isinstance(sigs, list) or not sigs:
        raise ValueError("Invalid JWS JSON: missing signatures.")
    payload_b = _b64u_dec(b64_payload)

    accepted = 0
    allowed = (
        {a if isinstance(a, JWAAlg) else JWAAlg(str(a)) for a in require_any}
        if require_any
        else None
    )

    for s in sigs:
        b64_hdr = s.get("protected")
        b64sig = s.get("signature")
        if not (isinstance(b64_hdr, str) and isinstance(b64sig, str)):
            continue
        try:
            header = json.loads(_b64u_dec(b64_hdr))
        except Exception:  # pragma: no cover - invalid header
            continue
        alg_raw = header.get("alg")
        kid = header.get("kid")
        if not isinstance(alg_raw, str):
            continue
        try:
            alg = JWAAlg(alg_raw)
        except ValueError:
            continue
        if allowed and alg not in allowed:
            continue
        signing_input = f"{b64_hdr}.{b64_payload}".encode("ascii")
        sig = _b64u_dec(b64sig)

        if jwks_resolver:
            jwk = jwks_resolver(kid, alg.value)
            ok = await self._verify_one(signing_input, alg, sig, jwk=jwk)
        else:
            ok = await self._verify_one(
                signing_input,
                alg,
                sig,
                hmac_keys=hmac_keys,
                rsa_pubkeys=rsa_pubkeys,
                ec_pubkeys=ec_pubkeys,
                ed_pubkeys=ed_pubkeys,
                k1_pubkeys=k1_pubkeys,
            )
        if ok:
            accepted += 1
            if accepted >= min_signers:
                return accepted, payload_b

    return accepted, payload_b