DIDs & keys

Signing keys

openvc.keys

openvc.keys — key backends implementing the SigningKey protocol.

Two software backends are provided:

• Ed25519SigningKey -> alg "EdDSA" (default for general OB 3.0 issuance; opt into the RFC 9864 fully-specified name "Ed25519" with alg="Ed25519")
• P256SigningKey -> alg "ES256" (P-256, required for EBSI/EUDI)

Both produce JWS-compatible signatures, which is the subtle part:

• Ed25519 : cryptography already returns the raw 64-byte signature — no change.
• ES256 : cryptography returns a DER-encoded ECDSA signature. JOSE requires the fixed-length raw form R||S (32 + 32 = 64 bytes). We convert on sign and back to DER on verify. Getting this wrong is the classic reason a locally-produced ES256 token fails to verify in another stack.

HSM / Vault backends: anything implementing SigningKey (alg, kid, sign returning R||S for ES256) is a drop-in replacement — e.g. a PKCS#11 or Vault Transit backend whose sign performs the operation without the private key ever entering the process. These software classes are for dev, tests, and low-assurance issuance.

Ed25519SigningKey

An in-process Ed25519 (EdDSA) signing key — the general OB 3.0 default.

The JOSE algorithm name emitted in the header defaults to the (still-supported) polymorphic "EdDSA"; pass alg="Ed25519" to emit the RFC 9864 fully-specified name instead. Both name the same Ed25519 signature.

Source code in src/openvc/keys.py

class Ed25519SigningKey:
    """An in-process Ed25519 (``EdDSA``) signing key — the general OB 3.0 default.

    The JOSE algorithm name emitted in the header defaults to the (still-supported)
    polymorphic ``"EdDSA"``; pass ``alg="Ed25519"`` to emit the RFC 9864
    fully-specified name instead. Both name the same Ed25519 signature.
    """

    def __init__(
        self, private_key: ed25519.Ed25519PrivateKey, kid: str, *, alg: str = "EdDSA"
    ) -> None:
        if alg not in ED25519_ALG_NAMES:
            raise InvalidKey(f"Ed25519 alg must be one of {sorted(ED25519_ALG_NAMES)}, "
                             f"got {alg!r}")
        self._sk = private_key
        self._kid = kid
        self._alg = alg

    @property
    def alg(self) -> str:
        """The JOSE algorithm name in the header — ``"EdDSA"`` (default) or the RFC
        9864 fully-specified ``"Ed25519"``."""
        return self._alg

    @property
    def kid(self) -> str:
        """The verification-method id this key signs as."""
        return self._kid

    def sign(self, signing_input: bytes) -> bytes:
        # cryptography returns the raw 64-byte signature; JOSE-ready as-is.
        """Sign *signing_input*; returns the raw 64-byte Ed25519 signature."""
        return self._sk.sign(signing_input)

    def public_jwk(self) -> dict[str, Any]:
        """The public key as an OKP JWK."""
        raw = self._sk.public_key().public_bytes(
            serialization.Encoding.Raw, serialization.PublicFormat.Raw
        )
        return {"kty": "OKP", "crv": "Ed25519", "x": _b64url_encode(raw)}

    def public_key_raw(self) -> bytes:
        """Raw 32-byte public key (used by the did:key encoder, multicodec 0xed01)."""
        return self._sk.public_key().public_bytes(
            serialization.Encoding.Raw, serialization.PublicFormat.Raw
        )

    # -- constructors ------------------------------------------------------ #

    @classmethod
    def generate(cls, kid: str, *, alg: str = "EdDSA") -> "Ed25519SigningKey":
        return cls(ed25519.Ed25519PrivateKey.generate(), kid, alg=alg)

    @classmethod
    def from_jwk(cls, jwk: dict[str, Any], kid: str, *, alg: str = "EdDSA") -> "Ed25519SigningKey":
        if jwk.get("kty") != "OKP" or jwk.get("crv") != "Ed25519" or "d" not in jwk:
            raise InvalidKey("not an Ed25519 private JWK")
        sk = ed25519.Ed25519PrivateKey.from_private_bytes(_b64url_decode(jwk["d"]))
        return cls(sk, kid, alg=alg)

    @classmethod
    def from_pem(
        cls, pem: bytes, kid: str, password: bytes | None = None, *, alg: str = "EdDSA"
    ) -> "Ed25519SigningKey":
        sk = serialization.load_pem_private_key(pem, password=password)
        if not isinstance(sk, ed25519.Ed25519PrivateKey):
            raise InvalidKey("PEM is not an Ed25519 private key")
        return cls(sk, kid, alg=alg)

alg property

The JOSE algorithm name in the header — "EdDSA" (default) or the RFC 9864 fully-specified "Ed25519".

kid property

The verification-method id this key signs as.

sign(signing_input)

Sign signing_input; returns the raw 64-byte Ed25519 signature.

Source code in src/openvc/keys.py

def sign(self, signing_input: bytes) -> bytes:
    # cryptography returns the raw 64-byte signature; JOSE-ready as-is.
    """Sign *signing_input*; returns the raw 64-byte Ed25519 signature."""
    return self._sk.sign(signing_input)

public_jwk()

The public key as an OKP JWK.

Source code in src/openvc/keys.py

def public_jwk(self) -> dict[str, Any]:
    """The public key as an OKP JWK."""
    raw = self._sk.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    )
    return {"kty": "OKP", "crv": "Ed25519", "x": _b64url_encode(raw)}

public_key_raw()

Raw 32-byte public key (used by the did:key encoder, multicodec 0xed01).

Source code in src/openvc/keys.py

def public_key_raw(self) -> bytes:
    """Raw 32-byte public key (used by the did:key encoder, multicodec 0xed01)."""
    return self._sk.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    )

P256SigningKey

An in-process P-256 (ES256) signing key — required for EBSI/EUDI.

Source code in src/openvc/keys.py

class P256SigningKey:
    """An in-process P-256 (``ES256``) signing key — required for EBSI/EUDI."""
    alg = "ES256"

    def __init__(self, private_key: ec.EllipticCurvePrivateKey, kid: str) -> None:
        if not isinstance(private_key.curve, ec.SECP256R1):
            raise InvalidKey("ES256 requires a P-256 (secp256r1) key")
        self._sk = private_key
        self._kid = kid

    @property
    def kid(self) -> str:
        """The verification-method id this key signs as."""
        return self._kid

    def sign(self, signing_input: bytes) -> bytes:
        """Sign *signing_input*; returns the raw R‖S signature (JOSE ES256, not DER)."""
        der = self._sk.sign(signing_input, ec.ECDSA(hashes.SHA256()))
        r, s = decode_dss_signature(der)                       # DER -> (int, int)
        return _int_to_fixed(r, P256_COORD_BYTES) + _int_to_fixed(s, P256_COORD_BYTES)

    def public_jwk(self) -> dict[str, Any]:
        """The public key as an EC P-256 JWK."""
        nums = self._sk.public_key().public_numbers()
        return {
            "kty": "EC",
            "crv": "P-256",
            "x": _b64url_encode(_int_to_fixed(nums.x, P256_COORD_BYTES)),
            "y": _b64url_encode(_int_to_fixed(nums.y, P256_COORD_BYTES)),
        }

    def public_key_raw(self, *, compressed: bool = True) -> bytes:
        """SEC1 point (compressed by default) — used by the did:key encoder
        (multicodec 0x1200 for P-256)."""
        fmt = (serialization.PublicFormat.CompressedPoint if compressed
               else serialization.PublicFormat.UncompressedPoint)
        return self._sk.public_key().public_bytes(serialization.Encoding.X962, fmt)

    # -- constructors ------------------------------------------------------ #

    @classmethod
    def generate(cls, kid: str) -> "P256SigningKey":
        return cls(ec.generate_private_key(ec.SECP256R1()), kid)

    @classmethod
    def from_jwk(cls, jwk: dict[str, Any], kid: str) -> "P256SigningKey":
        if jwk.get("kty") != "EC" or jwk.get("crv") != "P-256" or "d" not in jwk:
            raise InvalidKey("not a P-256 private JWK")
        d = int.from_bytes(_b64url_decode(jwk["d"]), "big")
        return cls(ec.derive_private_key(d, ec.SECP256R1()), kid)

    @classmethod
    def from_pem(cls, pem: bytes, kid: str, password: bytes | None = None) -> "P256SigningKey":
        sk = serialization.load_pem_private_key(pem, password=password)
        if not isinstance(sk, ec.EllipticCurvePrivateKey):
            raise InvalidKey("PEM is not an EC private key")
        return cls(sk, kid)

kid property

The verification-method id this key signs as.

sign(signing_input)

Sign signing_input; returns the raw R‖S signature (JOSE ES256, not DER).

Source code in src/openvc/keys.py

def sign(self, signing_input: bytes) -> bytes:
    """Sign *signing_input*; returns the raw R‖S signature (JOSE ES256, not DER)."""
    der = self._sk.sign(signing_input, ec.ECDSA(hashes.SHA256()))
    r, s = decode_dss_signature(der)                       # DER -> (int, int)
    return _int_to_fixed(r, P256_COORD_BYTES) + _int_to_fixed(s, P256_COORD_BYTES)

public_jwk()

The public key as an EC P-256 JWK.

Source code in src/openvc/keys.py

def public_jwk(self) -> dict[str, Any]:
    """The public key as an EC P-256 JWK."""
    nums = self._sk.public_key().public_numbers()
    return {
        "kty": "EC",
        "crv": "P-256",
        "x": _b64url_encode(_int_to_fixed(nums.x, P256_COORD_BYTES)),
        "y": _b64url_encode(_int_to_fixed(nums.y, P256_COORD_BYTES)),
    }

public_key_raw(*, compressed=True)

SEC1 point (compressed by default) — used by the did:key encoder (multicodec 0x1200 for P-256).

Source code in src/openvc/keys.py

def public_key_raw(self, *, compressed: bool = True) -> bytes:
    """SEC1 point (compressed by default) — used by the did:key encoder
    (multicodec 0x1200 for P-256)."""
    fmt = (serialization.PublicFormat.CompressedPoint if compressed
           else serialization.PublicFormat.UncompressedPoint)
    return self._sk.public_key().public_bytes(serialization.Encoding.X962, fmt)

P384SigningKey

An in-process P-384 (ES384) signing key — the P-384 leg of the Data Integrity ECDSA cryptosuites (ecdsa-jcs-2019 / ecdsa-rdfc-2019).

Source code in src/openvc/keys.py

class P384SigningKey:
    """An in-process P-384 (``ES384``) signing key — the P-384 leg of the
    Data Integrity ECDSA cryptosuites (``ecdsa-jcs-2019`` / ``ecdsa-rdfc-2019``)."""
    alg = "ES384"

    def __init__(self, private_key: ec.EllipticCurvePrivateKey, kid: str) -> None:
        if not isinstance(private_key.curve, ec.SECP384R1):
            raise InvalidKey("ES384 requires a P-384 (secp384r1) key")
        self._sk = private_key
        self._kid = kid

    @property
    def kid(self) -> str:
        """The verification-method id this key signs as."""
        return self._kid

    def sign(self, signing_input: bytes) -> bytes:
        """Sign *signing_input* over SHA-384; returns the raw R‖S signature (JOSE
        ES384, 96 bytes — not DER)."""
        der = self._sk.sign(signing_input, ec.ECDSA(hashes.SHA384()))
        r, s = decode_dss_signature(der)
        return _int_to_fixed(r, P384_COORD_BYTES) + _int_to_fixed(s, P384_COORD_BYTES)

    def public_jwk(self) -> dict[str, Any]:
        """The public key as an EC P-384 JWK."""
        nums = self._sk.public_key().public_numbers()
        return {
            "kty": "EC",
            "crv": "P-384",
            "x": _b64url_encode(_int_to_fixed(nums.x, P384_COORD_BYTES)),
            "y": _b64url_encode(_int_to_fixed(nums.y, P384_COORD_BYTES)),
        }

    def public_key_raw(self, *, compressed: bool = True) -> bytes:
        """SEC1 point (compressed by default) — used by the did:key encoder
        (multicodec 0x1201 for P-384)."""
        fmt = (serialization.PublicFormat.CompressedPoint if compressed
               else serialization.PublicFormat.UncompressedPoint)
        return self._sk.public_key().public_bytes(serialization.Encoding.X962, fmt)

    # -- constructors ------------------------------------------------------ #

    @classmethod
    def generate(cls, kid: str) -> "P384SigningKey":
        return cls(ec.generate_private_key(ec.SECP384R1()), kid)

    @classmethod
    def from_jwk(cls, jwk: dict[str, Any], kid: str) -> "P384SigningKey":
        if jwk.get("kty") != "EC" or jwk.get("crv") != "P-384" or "d" not in jwk:
            raise InvalidKey("not a P-384 private JWK")
        d = int.from_bytes(_b64url_decode(jwk["d"]), "big")
        try:                                              # d=0 / d>=n -> not a valid scalar
            return cls(ec.derive_private_key(d, ec.SECP384R1()), kid)
        except ValueError as exc:
            raise InvalidKey(f"invalid P-384 private scalar: {exc}") from exc

    @classmethod
    def from_pem(cls, pem: bytes, kid: str, password: bytes | None = None) -> "P384SigningKey":
        sk = serialization.load_pem_private_key(pem, password=password)
        if not isinstance(sk, ec.EllipticCurvePrivateKey):
            raise InvalidKey("PEM is not an EC private key")
        return cls(sk, kid)

kid property

The verification-method id this key signs as.

sign(signing_input)

Sign signing_input over SHA-384; returns the raw R‖S signature (JOSE ES384, 96 bytes — not DER).

Source code in src/openvc/keys.py

def sign(self, signing_input: bytes) -> bytes:
    """Sign *signing_input* over SHA-384; returns the raw R‖S signature (JOSE
    ES384, 96 bytes — not DER)."""
    der = self._sk.sign(signing_input, ec.ECDSA(hashes.SHA384()))
    r, s = decode_dss_signature(der)
    return _int_to_fixed(r, P384_COORD_BYTES) + _int_to_fixed(s, P384_COORD_BYTES)

public_jwk()

The public key as an EC P-384 JWK.

Source code in src/openvc/keys.py

def public_jwk(self) -> dict[str, Any]:
    """The public key as an EC P-384 JWK."""
    nums = self._sk.public_key().public_numbers()
    return {
        "kty": "EC",
        "crv": "P-384",
        "x": _b64url_encode(_int_to_fixed(nums.x, P384_COORD_BYTES)),
        "y": _b64url_encode(_int_to_fixed(nums.y, P384_COORD_BYTES)),
    }

public_key_raw(*, compressed=True)

SEC1 point (compressed by default) — used by the did:key encoder (multicodec 0x1201 for P-384).

Source code in src/openvc/keys.py

def public_key_raw(self, *, compressed: bool = True) -> bytes:
    """SEC1 point (compressed by default) — used by the did:key encoder
    (multicodec 0x1201 for P-384)."""
    fmt = (serialization.PublicFormat.CompressedPoint if compressed
           else serialization.PublicFormat.UncompressedPoint)
    return self._sk.public_key().public_bytes(serialization.Encoding.X962, fmt)

KeyAgreementKey

Bases: Protocol

The recipient's static private half for JWE ECDH-ES key agreement.

The encryption counterpart of :class:~openvc.proof.vc_jwt.SigningKey: an HSM/Vault backend that runs the raw ECDH on-device (never exporting the private scalar) drops in by implementing crv, kid and agree — which returns the raw ECDH shared secret Z; the public JWE Concat KDF is applied by the caller (:mod:openvc.jwe), so no secret-derived material beyond Z crosses the boundary.

Source code in src/openvc/keys.py

@runtime_checkable
class KeyAgreementKey(Protocol):
    """The recipient's static private half for JWE ECDH-ES key agreement.

    The encryption counterpart of :class:`~openvc.proof.vc_jwt.SigningKey`: an
    HSM/Vault backend that runs the raw ECDH on-device (never exporting the private
    scalar) drops in by implementing ``crv``, ``kid`` and ``agree`` — which returns
    the raw ECDH shared secret ``Z``; the public JWE Concat KDF is applied by the
    caller (:mod:`openvc.jwe`), so no secret-derived material beyond ``Z`` crosses the
    boundary.
    """
    crv: str

    @property
    def kid(self) -> str: ...

    def agree(self, peer_public_jwk: dict[str, Any]) -> bytes: ...

P256KeyAgreementKey

An in-process P-256 ECDH key-agreement key (JWE ECDH-ES, HAIP responses).

Source code in src/openvc/keys.py

class P256KeyAgreementKey:
    """An in-process P-256 ECDH key-agreement key (JWE ``ECDH-ES``, HAIP responses)."""
    crv = "P-256"

    def __init__(self, private_key: ec.EllipticCurvePrivateKey, kid: str) -> None:
        if not isinstance(private_key.curve, ec.SECP256R1):
            raise InvalidKey("ECDH-ES over P-256 requires a P-256 (secp256r1) key")
        self._sk = private_key
        self._kid = kid

    @property
    def kid(self) -> str:
        """The id the recipient key is published under (JWE ``kid``)."""
        return self._kid

    def agree(self, peer_public_jwk: dict[str, Any]) -> bytes:
        """Return the raw ECDH shared secret ``Z`` (the 32-byte big-endian X
        coordinate of the shared point) with the peer's ephemeral public key (the JWE
        ``epk``). A non-P-256 peer key is rejected before any curve operation."""
        if peer_public_jwk.get("kty") != "EC" or peer_public_jwk.get("crv") != "P-256":
            raise InvalidKey("ECDH-ES peer (epk) must be an EC P-256 public JWK")
        try:
            x = int.from_bytes(_b64url_decode(peer_public_jwk["x"]), "big")
            y = int.from_bytes(_b64url_decode(peer_public_jwk["y"]), "big")
            peer = ec.EllipticCurvePublicNumbers(x, y, ec.SECP256R1()).public_key()
        except (KeyError, ValueError, TypeError) as exc:               # malformed epk
            raise InvalidKey(f"invalid ECDH-ES peer public key: {exc}") from exc
        return self._sk.exchange(ec.ECDH(), peer)

    def public_jwk(self) -> dict[str, Any]:
        """The public key as an EC P-256 JWK marked ``use:"enc"`` — the verifier
        publishes this (in ``client_metadata.jwks``) for the wallet to encrypt to."""
        nums = self._sk.public_key().public_numbers()
        return {
            "kty": "EC", "crv": "P-256", "use": "enc",
            "x": _b64url_encode(_int_to_fixed(nums.x, P256_COORD_BYTES)),
            "y": _b64url_encode(_int_to_fixed(nums.y, P256_COORD_BYTES)),
        }

    # -- constructors ------------------------------------------------------ #

    @classmethod
    def generate(cls, kid: str) -> "P256KeyAgreementKey":
        return cls(ec.generate_private_key(ec.SECP256R1()), kid)

    @classmethod
    def from_jwk(cls, jwk: dict[str, Any], kid: str) -> "P256KeyAgreementKey":
        if jwk.get("kty") != "EC" or jwk.get("crv") != "P-256" or "d" not in jwk:
            raise InvalidKey("not a P-256 private JWK")
        d = int.from_bytes(_b64url_decode(jwk["d"]), "big")
        return cls(ec.derive_private_key(d, ec.SECP256R1()), kid)

    @classmethod
    def from_pem(
        cls, pem: bytes, kid: str, password: bytes | None = None
    ) -> "P256KeyAgreementKey":
        sk = serialization.load_pem_private_key(pem, password=password)
        if not isinstance(sk, ec.EllipticCurvePrivateKey):
            raise InvalidKey("PEM is not an EC private key")
        return cls(sk, kid)

kid property

The id the recipient key is published under (JWE kid).

agree(peer_public_jwk)

Return the raw ECDH shared secret Z (the 32-byte big-endian X coordinate of the shared point) with the peer's ephemeral public key (the JWE epk). A non-P-256 peer key is rejected before any curve operation.

Source code in src/openvc/keys.py

def agree(self, peer_public_jwk: dict[str, Any]) -> bytes:
    """Return the raw ECDH shared secret ``Z`` (the 32-byte big-endian X
    coordinate of the shared point) with the peer's ephemeral public key (the JWE
    ``epk``). A non-P-256 peer key is rejected before any curve operation."""
    if peer_public_jwk.get("kty") != "EC" or peer_public_jwk.get("crv") != "P-256":
        raise InvalidKey("ECDH-ES peer (epk) must be an EC P-256 public JWK")
    try:
        x = int.from_bytes(_b64url_decode(peer_public_jwk["x"]), "big")
        y = int.from_bytes(_b64url_decode(peer_public_jwk["y"]), "big")
        peer = ec.EllipticCurvePublicNumbers(x, y, ec.SECP256R1()).public_key()
    except (KeyError, ValueError, TypeError) as exc:               # malformed epk
        raise InvalidKey(f"invalid ECDH-ES peer public key: {exc}") from exc
    return self._sk.exchange(ec.ECDH(), peer)

public_jwk()

The public key as an EC P-256 JWK marked use:"enc" — the verifier publishes this (in client_metadata.jwks) for the wallet to encrypt to.

Source code in src/openvc/keys.py

def public_jwk(self) -> dict[str, Any]:
    """The public key as an EC P-256 JWK marked ``use:"enc"`` — the verifier
    publishes this (in ``client_metadata.jwks``) for the wallet to encrypt to."""
    nums = self._sk.public_key().public_numbers()
    return {
        "kty": "EC", "crv": "P-256", "use": "enc",
        "x": _b64url_encode(_int_to_fixed(nums.x, P256_COORD_BYTES)),
        "y": _b64url_encode(_int_to_fixed(nums.y, P256_COORD_BYTES)),
    }

MLDSASigningKey

Experimental post-quantum ML-DSA (RFC 9964) signing key — opt-in, not a production trust path (ADR-0004). Implements the :class:SigningKey protocol: alg is the parameter set (ML-DSA-44 / ML-DSA-65 / ML-DSA-87), the private key is a 32-byte seed, and sign returns the raw ML-DSA signature — no R‖S transform, it is already the JOSE-wire form.

An HSM/Vault backend uses external-mu (sign_mu) so the message never reaches the device in the clear; this in-process backend is for dev, tests and low-assurance issuance, exactly like the EC / Ed backends.

Source code in src/openvc/keys.py

class MLDSASigningKey:
    """**Experimental** post-quantum ML-DSA (RFC 9964) signing key — opt-in, **not** a
    production trust path (ADR-0004). Implements the :class:`SigningKey` protocol: ``alg``
    is the parameter set (``ML-DSA-44`` / ``ML-DSA-65`` / ``ML-DSA-87``), the private key
    is a 32-byte seed, and ``sign`` returns the raw ML-DSA signature — no R‖S transform,
    it is already the JOSE-wire form.

    An HSM/Vault backend uses external-mu (``sign_mu``) so the message never reaches the
    device in the clear; this in-process backend is for dev, tests and low-assurance
    issuance, exactly like the EC / Ed backends."""

    def __init__(self, private_key: Any, kid: str, alg: str) -> None:
        if alg not in MLDSA_ALGS:
            raise InvalidKey(f"ML-DSA alg must be one of {sorted(MLDSA_ALGS)}, got {alg!r}")
        self._sk = private_key
        self._kid = kid
        self._alg = alg

    @property
    def alg(self) -> str:
        return self._alg

    @property
    def kid(self) -> str:
        return self._kid

    def sign(self, signing_input: bytes) -> bytes:
        """Sign *signing_input*; returns the raw ML-DSA signature (JOSE-wire form)."""
        try:
            return self._sk.sign(signing_input)
        except Exception as exc:                       # OpenSSL < 3.5 -> typed, fail closed
            raise InvalidKey(f"ML-DSA signing failed ({self._alg}): {exc}") from exc

    def public_jwk(self) -> dict[str, Any]:
        """The public key as an RFC 9964 ``AKP`` JWK (``kty`` / ``alg`` / ``pub``)."""
        return {"kty": "AKP", "alg": self._alg,
                "pub": _b64url_encode(self._sk.public_key().public_bytes_raw())}

    def private_jwk(self) -> dict[str, Any]:
        """The private ``AKP`` JWK — adds the 32-byte seed ``priv``. Handle with care."""
        return dict(self.public_jwk(), priv=_b64url_encode(self._sk.private_bytes_raw()))

    @classmethod
    def generate(cls, kid: str, *, alg: str = "ML-DSA-65") -> "MLDSASigningKey":
        priv_cls, _ = _mldsa_classes(alg)
        try:
            return cls(priv_cls.generate(), kid, alg)
        except InvalidKey:
            raise
        except Exception as exc:                       # OpenSSL < 3.5 -> typed
            raise InvalidKey(f"ML-DSA unavailable ({alg}): {exc}") from exc

    @classmethod
    def from_seed(cls, seed: bytes, kid: str, *, alg: str = "ML-DSA-65") -> "MLDSASigningKey":
        priv_cls, _ = _mldsa_classes(alg)
        if len(seed) != _MLDSA_SEED_BYTES:
            raise InvalidKey(f"ML-DSA seed must be {_MLDSA_SEED_BYTES} bytes, got {len(seed)}")
        return cls(priv_cls.from_seed_bytes(seed), kid, alg)

    @classmethod
    def from_jwk(cls, jwk: dict[str, Any], kid: str) -> "MLDSASigningKey":
        if jwk.get("kty") != "AKP" or "priv" not in jwk:
            raise InvalidKey("not an AKP (ML-DSA) private JWK")
        alg = jwk.get("alg")
        if not isinstance(alg, str) or alg not in MLDSA_ALGS:
            raise InvalidKey(f"unsupported AKP alg {alg!r}")
        return cls.from_seed(_b64url_decode(jwk["priv"]), kid, alg=alg)

sign(signing_input)

Sign signing_input; returns the raw ML-DSA signature (JOSE-wire form).

Source code in src/openvc/keys.py

def sign(self, signing_input: bytes) -> bytes:
    """Sign *signing_input*; returns the raw ML-DSA signature (JOSE-wire form)."""
    try:
        return self._sk.sign(signing_input)
    except Exception as exc:                       # OpenSSL < 3.5 -> typed, fail closed
        raise InvalidKey(f"ML-DSA signing failed ({self._alg}): {exc}") from exc

public_jwk()

The public key as an RFC 9964 AKP JWK (kty / alg / pub).

Source code in src/openvc/keys.py

def public_jwk(self) -> dict[str, Any]:
    """The public key as an RFC 9964 ``AKP`` JWK (``kty`` / ``alg`` / ``pub``)."""
    return {"kty": "AKP", "alg": self._alg,
            "pub": _b64url_encode(self._sk.public_key().public_bytes_raw())}

private_jwk()

The private AKP JWK — adds the 32-byte seed priv. Handle with care.

Source code in src/openvc/keys.py

def private_jwk(self) -> dict[str, Any]:
    """The private ``AKP`` JWK — adds the 32-byte seed ``priv``. Handle with care."""
    return dict(self.public_jwk(), priv=_b64url_encode(self._sk.private_bytes_raw()))

mldsa_available()

True if ML-DSA is usable here — the module imports and the OpenSSL build supports it (a wheel built against OpenSSL < 3.5 raises at first use even on cryptography 49). Gate experimental ML-DSA paths and skip tests where unavailable.

Source code in src/openvc/keys.py

def mldsa_available() -> bool:
    """``True`` if ML-DSA is usable here — the module imports **and** the OpenSSL build
    supports it (a wheel built against OpenSSL < 3.5 raises at first use even on
    cryptography 49). Gate experimental ML-DSA paths and skip tests where unavailable."""
    try:
        _mldsa_module().MLDSA44PrivateKey.generate()
        return True
    except Exception:
        return False

verify_signature(*, alg, public_jwk, signing_input, signature)

Verify a JOSE signature directly from a public JWK, without PyJWT.

Handy for did:key (where the key is inside the DID) and for round-trip tests. The VC verification path uses the proof suite instead; this is complementary.

Source code in src/openvc/keys.py

def verify_signature(
    *, alg: str, public_jwk: dict[str, Any], signing_input: bytes, signature: bytes
) -> bool:
    """Verify a JOSE signature directly from a public JWK, without PyJWT.

    Handy for did:key (where the key is inside the DID) and for round-trip tests.
    The VC verification path uses the proof suite instead; this is complementary.
    """
    try:
        if alg in ED25519_ALG_NAMES:                       # "EdDSA" or RFC 9864 "Ed25519"
            raw = _b64url_decode(public_jwk["x"])
            ed25519.Ed25519PublicKey.from_public_bytes(raw).verify(signature, signing_input)
            return True
        if alg == "ES256":
            if len(signature) != 2 * P256_COORD_BYTES:
                raise InvalidKey("ES256 signature must be 64-byte R||S")
            r = int.from_bytes(signature[:P256_COORD_BYTES], "big")
            s = int.from_bytes(signature[P256_COORD_BYTES:], "big")
            der = encode_dss_signature(r, s)                   # R||S -> DER for verify
            pub = ec.EllipticCurvePublicNumbers(
                int.from_bytes(_b64url_decode(public_jwk["x"]), "big"),
                int.from_bytes(_b64url_decode(public_jwk["y"]), "big"),
                ec.SECP256R1(),
            ).public_key()
            pub.verify(der, signing_input, ec.ECDSA(hashes.SHA256()))
            return True
        if alg == "ES384":
            if len(signature) != 2 * P384_COORD_BYTES:
                raise InvalidKey("ES384 signature must be 96-byte R||S")
            r = int.from_bytes(signature[:P384_COORD_BYTES], "big")
            s = int.from_bytes(signature[P384_COORD_BYTES:], "big")
            der = encode_dss_signature(r, s)                   # R||S -> DER for verify
            pub = ec.EllipticCurvePublicNumbers(
                int.from_bytes(_b64url_decode(public_jwk["x"]), "big"),
                int.from_bytes(_b64url_decode(public_jwk["y"]), "big"),
                ec.SECP384R1(),
            ).public_key()
            pub.verify(der, signing_input, ec.ECDSA(hashes.SHA384()))
            return True
        if alg in MLDSA_ALGS:                              # experimental PQ (RFC 9964)
            _, pub_cls = _mldsa_classes(alg)               # typed InvalidKey if cryptography<48
            pub_cls.from_public_bytes(_b64url_decode(public_jwk["pub"])).verify(
                signature, signing_input)                  # raises InvalidSignature on failure
            return True
        raise InvalidKey(f"unsupported alg {alg!r}")
    except InvalidSignature:
        return False
    except (ValueError, KeyError, TypeError) as exc:
        # A malformed / mismatched public JWK (wrong-curve coords, an OKP key with no
        # "y", a bad-length Ed25519 x) must fail closed as a typed InvalidKey, not leak
        # a bare ValueError/KeyError to callers that catch only KeyBackendError.
        raise InvalidKey(f"malformed public key for {alg}: {exc}") from exc

signing_key_from_jwk(jwk, kid)

Dispatch to the right backend from a private JWK.

Source code in src/openvc/keys.py

def signing_key_from_jwk(jwk: dict[str, Any], kid: str) -> "SigningKey":
    """Dispatch to the right backend from a private JWK."""
    kty, crv = jwk.get("kty"), jwk.get("crv")
    if kty == "OKP" and crv == "Ed25519":
        return Ed25519SigningKey.from_jwk(jwk, kid)
    if kty == "EC" and crv == "P-256":
        return P256SigningKey.from_jwk(jwk, kid)
    if kty == "EC" and crv == "P-384":
        return P384SigningKey.from_jwk(jwk, kid)
    if kty == "AKP":                                   # experimental ML-DSA (RFC 9964)
        return MLDSASigningKey.from_jwk(jwk, kid)
    raise InvalidKey(f"unsupported key type kty={kty!r} crv={crv!r}")

DID resolution

openvc.did.base

openvc.did.base — core DID resolution primitives.

Home of the generic types every DID method shares: VerificationMethod, DidDocument, the DidResolver protocol, a shared W3C DID-document parser, and a registry that dispatches a DID to the backend that supports it. No network, no method specifics.

(These types used to live in the EBSI plugin; they are generic to did:key, did:web and did:ebsi alike, so they belong in the core.)

VerificationMethod dataclass

Source code in src/openvc/did/base.py

@dataclass(frozen=True)
class VerificationMethod:
    id: str                       # e.g. did:...#key-1
    type: str
    controller: str
    public_key_jwk: dict[str, Any]

    @property
    def kid(self) -> str:
        """The verification-method id — the ``#fragment`` key identifier."""
        return self.id.split("#", 1)[-1]

kid property

The verification-method id — the #fragment key identifier.

DidDocument dataclass

Source code in src/openvc/did/base.py

@dataclass(frozen=True)
class DidDocument:
    id: str
    verification_methods: list[VerificationMethod]
    raw: dict[str, Any]
    # {relationship -> [verificationMethod id, ...]} for the relationships the
    # document actually declares. A relationship absent from this mapping was not
    # declared by the document (distinct from declared-but-empty).
    relationships: dict[str, list[str]] = field(default_factory=dict)

    def key_by_kid(self, kid: str | None) -> VerificationMethod | None:
        """Match on the full verificationMethod id or its fragment (a JWS `kid`
        may carry either). If kid is None, fall back to the sole key if unique."""
        if kid is None:
            return self.verification_methods[0] if len(self.verification_methods) == 1 else None
        fragment = kid.split("#", 1)[-1]
        return next(
            (vm for vm in self.verification_methods if vm.id == kid or vm.kid == fragment),
            None,
        )

    def key_for_purpose(
        self, kid: str | None, proof_purpose: str
    ) -> VerificationMethod | None:
        """Like :meth:`key_by_kid`, but authorized for *proof_purpose*.

        If the document declares that relationship, the method must be referenced
        by it (returns None otherwise — the key exists but is not usable for this
        purpose). If the document does not declare the relationship at all, the
        binding cannot be enforced and the matched key is returned as-is."""
        vm = self.key_by_kid(kid)
        if vm is None:
            return None
        refs = self.relationships.get(proof_purpose)
        if refs is None:                       # relationship not declared -> lenient
            return vm
        if any(ref == vm.id or ref.split("#", 1)[-1] == vm.kid for ref in refs):
            return vm
        return None

key_by_kid(kid)

Match on the full verificationMethod id or its fragment (a JWS kid may carry either). If kid is None, fall back to the sole key if unique.

Source code in src/openvc/did/base.py

def key_by_kid(self, kid: str | None) -> VerificationMethod | None:
    """Match on the full verificationMethod id or its fragment (a JWS `kid`
    may carry either). If kid is None, fall back to the sole key if unique."""
    if kid is None:
        return self.verification_methods[0] if len(self.verification_methods) == 1 else None
    fragment = kid.split("#", 1)[-1]
    return next(
        (vm for vm in self.verification_methods if vm.id == kid or vm.kid == fragment),
        None,
    )

key_for_purpose(kid, proof_purpose)

Like :meth:key_by_kid, but authorized for proof_purpose.

If the document declares that relationship, the method must be referenced by it (returns None otherwise — the key exists but is not usable for this purpose). If the document does not declare the relationship at all, the binding cannot be enforced and the matched key is returned as-is.

Source code in src/openvc/did/base.py

def key_for_purpose(
    self, kid: str | None, proof_purpose: str
) -> VerificationMethod | None:
    """Like :meth:`key_by_kid`, but authorized for *proof_purpose*.

    If the document declares that relationship, the method must be referenced
    by it (returns None otherwise — the key exists but is not usable for this
    purpose). If the document does not declare the relationship at all, the
    binding cannot be enforced and the matched key is returned as-is."""
    vm = self.key_by_kid(kid)
    if vm is None:
        return None
    refs = self.relationships.get(proof_purpose)
    if refs is None:                       # relationship not declared -> lenient
        return vm
    if any(ref == vm.id or ref.split("#", 1)[-1] == vm.kid for ref in refs):
        return vm
    return None

DidResolver

Bases: Protocol

Source code in src/openvc/did/base.py

@runtime_checkable
class DidResolver(Protocol):
    def supports(self, did: str) -> bool:
        """Whether this resolver handles *did* (its DID method)."""
    def resolve(self, did: str) -> DidDocument:
        """Resolve *did* to a :class:`DidDocument`, or raise :class:`DidResolutionError`."""

supports(did)

Whether this resolver handles did (its DID method).

Source code in src/openvc/did/base.py

def supports(self, did: str) -> bool:
    """Whether this resolver handles *did* (its DID method)."""

resolve(did)

Resolve did to a :class:DidDocument, or raise :class:DidResolutionError.

Source code in src/openvc/did/base.py

def resolve(self, did: str) -> DidDocument:
    """Resolve *did* to a :class:`DidDocument`, or raise :class:`DidResolutionError`."""

DidResolverRegistry

A registry that dispatches resolve to the first resolver that supports the DID.

Source code in src/openvc/did/base.py

class DidResolverRegistry:
    """A registry that dispatches ``resolve`` to the first resolver that supports the DID."""

    def __init__(self, resolvers: list[DidResolver] | None = None) -> None:
        self._resolvers: list[DidResolver] = list(resolvers or [])

    def register(self, resolver: DidResolver) -> None:
        """Add a resolver (consulted in registration order)."""
        self._resolvers.append(resolver)

    def resolve(self, did: str) -> DidDocument:
        """Resolve *did* via the first matching resolver, else raise ``UnsupportedDidMethod``."""
        for r in self._resolvers:
            if r.supports(did):
                return r.resolve(did)
        raise UnsupportedDidMethod(f"no resolver for {did!r}")

register(resolver)

Add a resolver (consulted in registration order).

Source code in src/openvc/did/base.py

def register(self, resolver: DidResolver) -> None:
    """Add a resolver (consulted in registration order)."""
    self._resolvers.append(resolver)

resolve(did)

Resolve did via the first matching resolver, else raise UnsupportedDidMethod.

Source code in src/openvc/did/base.py

def resolve(self, did: str) -> DidDocument:
    """Resolve *did* via the first matching resolver, else raise ``UnsupportedDidMethod``."""
    for r in self._resolvers:
        if r.supports(did):
            return r.resolve(did)
    raise UnsupportedDidMethod(f"no resolver for {did!r}")

AsyncDidResolver

Bases: Protocol

The async counterpart of :class:DidResolver: supports stays a plain predicate (no I/O), resolve is awaitable so a backend can await a non-blocking fetch. A sync resolver is adapted to this shape by :func:as_async_resolver.

Source code in src/openvc/did/base.py

@runtime_checkable
class AsyncDidResolver(Protocol):
    """The async counterpart of :class:`DidResolver`: ``supports`` stays a plain
    predicate (no I/O), ``resolve`` is awaitable so a backend can await a non-blocking
    fetch. A sync resolver is adapted to this shape by :func:`as_async_resolver`."""
    def supports(self, did: str) -> bool:
        """Whether this resolver handles *did* (its DID method)."""
    def resolve(self, did: str) -> Awaitable[DidDocument]:
        """Resolve *did* to a :class:`DidDocument`, or raise :class:`DidResolutionError`."""

supports(did)

Whether this resolver handles did (its DID method).

Source code in src/openvc/did/base.py

def supports(self, did: str) -> bool:
    """Whether this resolver handles *did* (its DID method)."""

resolve(did)

Resolve did to a :class:DidDocument, or raise :class:DidResolutionError.

Source code in src/openvc/did/base.py

def resolve(self, did: str) -> Awaitable[DidDocument]:
    """Resolve *did* to a :class:`DidDocument`, or raise :class:`DidResolutionError`."""

AsyncDidResolverRegistry

The async counterpart of :class:DidResolverRegistry: dispatches an awaitable resolve to the first :class:AsyncDidResolver that supports the DID.

Source code in src/openvc/did/base.py

class AsyncDidResolverRegistry:
    """The async counterpart of :class:`DidResolverRegistry`: dispatches an awaitable
    ``resolve`` to the first :class:`AsyncDidResolver` that supports the DID."""

    def __init__(self, resolvers: list[AsyncDidResolver] | None = None) -> None:
        self._resolvers: list[AsyncDidResolver] = list(resolvers or [])

    def register(self, resolver: AsyncDidResolver) -> None:
        """Add a resolver (consulted in registration order)."""
        self._resolvers.append(resolver)

    async def resolve(self, did: str) -> DidDocument:
        """Resolve *did* via the first matching resolver, else raise ``UnsupportedDidMethod``."""
        for r in self._resolvers:
            if r.supports(did):
                return await r.resolve(did)
        raise UnsupportedDidMethod(f"no resolver for {did!r}")

register(resolver)

Add a resolver (consulted in registration order).

Source code in src/openvc/did/base.py

def register(self, resolver: AsyncDidResolver) -> None:
    """Add a resolver (consulted in registration order)."""
    self._resolvers.append(resolver)

resolve(did) async

Resolve did via the first matching resolver, else raise UnsupportedDidMethod.

Source code in src/openvc/did/base.py

async def resolve(self, did: str) -> DidDocument:
    """Resolve *did* via the first matching resolver, else raise ``UnsupportedDidMethod``."""
    for r in self._resolvers:
        if r.supports(did):
            return await r.resolve(did)
    raise UnsupportedDidMethod(f"no resolver for {did!r}")

multikey_to_jwk(multibase_key)

Decode a publicKeyMultibase Multikey to a public JWK.

Handles the key types openvc verifies with — Ed25519 (multicodec 0xed) and the NIST curves P-256 (0x1200) / P-384 (0x1201, SEC1 compressed points) — the same set as :mod:openvc.did.did_key. Raises :class:~openvc.multibase.MultibaseError (or ValueError for an off-curve point / unknown codec) so the caller can skip an undecodable method rather than crash.

Source code in src/openvc/did/base.py

def multikey_to_jwk(multibase_key: str) -> dict[str, Any]:
    """Decode a ``publicKeyMultibase`` Multikey to a public JWK.

    Handles the key types openvc verifies with — Ed25519 (multicodec ``0xed``) and the
    NIST curves P-256 (``0x1200``) / P-384 (``0x1201``, SEC1 compressed points) — the same
    set as :mod:`openvc.did.did_key`. Raises :class:`~openvc.multibase.MultibaseError`
    (or ``ValueError`` for an off-curve point / unknown codec) so the caller can skip an
    undecodable method rather than crash."""
    raw = decode_multibase(multibase_key)
    code, off = read_varint(raw)
    key = raw[off:]
    if code == _MC_ED25519:
        if len(key) != 32:
            raise ValueError(f"Ed25519 Multikey must be 32 bytes, got {len(key)}")
        return {"kty": "OKP", "crv": "Ed25519", "x": _b64url(key)}
    if code in (_MC_P256, _MC_P384):
        from cryptography.hazmat.primitives.asymmetric import ec

        curve, crv, size = (
            (ec.SECP256R1(), "P-256", 32) if code == _MC_P256 else (ec.SECP384R1(), "P-384", 48))
        pub = ec.EllipticCurvePublicKey.from_encoded_point(curve, key)  # validates on-curve
        nums = pub.public_numbers()
        return {"kty": "EC", "crv": crv,
                "x": _b64url(nums.x.to_bytes(size, "big")),
                "y": _b64url(nums.y.to_bytes(size, "big"))}
    raise ValueError(f"unsupported Multikey multicodec 0x{code:x}")

parse_did_document(raw)

Parse a W3C DID document. Tolerates a didDocument wrapper or a bare doc (the EBSI DID Registry returns it bare, as application/did+ld+json). Verification methods may carry publicKeyJwk or a publicKeyMultibase Multikey (did:webvh and modern did:web documents use the latter).

The parser is context-agnostic — it reads the document shape (verificationMethod / relationships), never @context — so DID 1.1 / CID 1.0 documents (the https://www.w3.org/ns/did/v1.1 context) resolve unchanged. The DID 1.1 relationship-semantics diff is revisited when it reaches Proposed Recommendation.

Source code in src/openvc/did/base.py

def parse_did_document(raw: dict[str, Any]) -> DidDocument:
    """Parse a W3C DID document. Tolerates a `didDocument` wrapper or a bare doc
    (the EBSI DID Registry returns it bare, as application/did+ld+json). Verification
    methods may carry ``publicKeyJwk`` or a ``publicKeyMultibase`` Multikey (did:webvh and
    modern did:web documents use the latter).

    The parser is **context-agnostic** — it reads the document *shape*
    (``verificationMethod`` / relationships), never ``@context`` — so **DID 1.1 / CID 1.0**
    documents (the ``https://www.w3.org/ns/did/v1.1`` context) resolve unchanged. The DID 1.1
    relationship-semantics diff is revisited when it reaches Proposed Recommendation."""
    doc = raw.get("didDocument", raw)
    vms = [
        VerificationMethod(
            id=vm["id"],
            type=vm.get("type", ""),
            controller=vm.get("controller", doc.get("id", "")),
            public_key_jwk=jwk,
        )
        for vm in doc.get("verificationMethod", [])
        if isinstance(vm, dict) and "id" in vm and (jwk := _vm_public_jwk(vm)) is not None
    ]
    relationships = {
        key: _relationship_refs(doc, key) for key in RELATIONSHIP_KEYS if key in doc
    }
    return DidDocument(
        id=doc.get("id", ""), verification_methods=vms, raw=doc,
        relationships=relationships,
    )

as_async_resolver(resolver)

Wrap a sync :class:DidResolver as an :class:AsyncDidResolver (its resolve awaits nothing). Lets an offline did:key / did:jwk resolver — or any sync resolver whose blocking is acceptable — drop into an async registry.

Source code in src/openvc/did/base.py

def as_async_resolver(resolver: DidResolver) -> AsyncDidResolver:
    """Wrap a sync :class:`DidResolver` as an :class:`AsyncDidResolver` (its
    ``resolve`` awaits nothing). Lets an offline did:key / did:jwk resolver — or any
    sync resolver whose blocking is acceptable — drop into an async registry."""
    return _SyncResolverAsAsync(resolver)

openvc.did.did_key

openvc.did.did_key — offline resolver for the did:key method (Ed25519, P-256).

did:key is self-contained: the public key is encoded in the identifier itself, so resolution is pure decoding — no network. Format:

did:key:z<base58btc( <multicodec-varint> || <raw-public-key> )>

Supported multicodecs

0xed Ed25519 public key -> OKP / Ed25519 JWK 0x1200 P-256 public key -> EC / P-256 JWK (from the compressed point)

openvc.did.did_jwk

openvc.did.did_jwk — offline resolver for the did:jwk method.

did:jwk is self-contained like did:key: the public key is the identifier, a base64url-encoded JSON JWK, so resolution is pure decoding — no network. Format:

did:jwk:<base64url-nopad( utf8( JSON public JWK ) )>

The verification method fragment is always #0, and it is referenced by every signing verification relationship. Common in EUDI / OID4VC test stacks.

openvc.did.did_web

openvc.did.did_web — resolver for the did:web method.

did:web maps a DID to an https URL and fetches the DID document from the issuer's own domain:

did:web:example.edu                 -> https://example.edu/.well-known/did.json
did:web:example.edu:issuers:physics -> https://example.edu/issuers/physics/did.json
did:web:example.edu%3A3000          -> https://example.edu:3000/.well-known/did.json

SSRF note

did:web is intentionally cross-host — the whole point is to resolve a controller's own domain — so a fixed host allow-list does NOT apply here (unlike the EBSI client). Pass a general-purpose fetch for this resolver, ideally one that still enforces https and blocks private/link-local address ranges. Do NOT reuse the EBSI client, whose allow-list would reject every legitimate did:web host.

AsyncDidWebResolver

The async counterpart of :class:DidWebResolver: same URL mapping and id integrity check, but awaits an injected async fetch (pass :func:openvc.fetch.https_json_fetch_async for the SSRF-guarded one).

Source code in src/openvc/did/did_web.py

class AsyncDidWebResolver:
    """The async counterpart of :class:`DidWebResolver`: same URL mapping and id
    integrity check, but awaits an injected async fetch (pass
    :func:`openvc.fetch.https_json_fetch_async` for the SSRF-guarded one)."""
    def __init__(self, fetch: AsyncFetch) -> None:
        self._fetch = fetch

    def supports(self, did: str) -> bool:
        return did.startswith("did:web:")

    async def resolve(self, did: str) -> DidDocument:
        return _validated_document(await self._fetch(_did_web_url(did)), did)

SSRF-guarded fetch

openvc.fetch

openvc.fetch — an SSRF-guarded, stdlib-only https JSON fetch for did:web.

did:web is intentionally cross-host (it resolves a controller's own domain), so it cannot use a host allow-list like the EBSI client. This fetch instead blocks the dangerous targets: it requires https, resolves the host and refuses any private / loopback / link-local / reserved / multicast address (the cloud metadata endpoint 169.254.169.254 is link-local, so it is covered), and refuses HTTP redirects — a common SSRF pivot.

DNS-rebinding is closed, not just documented: the connection is pinned to the validated IP (we resolve, validate every resolved address, then open the TCP socket to that exact IP) while TLS SNI, certificate validation, and the Host header still use the hostname. So an attacker who flips DNS between the check and the connect cannot redirect us to an internal address — we never re-resolve.

Dependency-light on purpose: pure stdlib (http.client + ssl + socket + ipaddress). Pass https_json_fetch wherever a Fetch is expected, or use default_did_web_resolver().

UnsafeUrlError

Bases: DidResolutionError

The URL or a resolved address is not allowed (scheme, host, or IP range).

Source code in src/openvc/fetch.py

class UnsafeUrlError(DidResolutionError):
    """The URL or a resolved address is not allowed (scheme, host, or IP range)."""

https_json_fetch(url, *, timeout_s=DEFAULT_TIMEOUT_S, max_bytes=MAX_RESPONSE_BYTES)

Fetch url as a JSON object with the SSRF guards above. Returns the parsed object.

Raises :class:UnsafeUrlError for a disallowed scheme/host/address or a redirect (a subclass of :class:~openvc.did.base.DidResolutionError), and DidResolutionError for transport / oversize / non-JSON failures.

Source code in src/openvc/fetch.py

def https_json_fetch(
    url: str,
    *,
    timeout_s: float = DEFAULT_TIMEOUT_S,
    max_bytes: int = MAX_RESPONSE_BYTES,
) -> dict[str, Any]:
    """Fetch *url* as a JSON object with the SSRF guards above. Returns the parsed
    object.

    Raises :class:`UnsafeUrlError` for a disallowed scheme/host/address or a
    redirect (a subclass of :class:`~openvc.did.base.DidResolutionError`), and
    ``DidResolutionError`` for transport / oversize / non-JSON failures.
    """
    raw = _https_fetch_guarded(url, timeout_s=timeout_s, max_bytes=max_bytes)
    try:
        data = json.loads(raw)
    except (ValueError, json.JSONDecodeError, RecursionError) as exc:
        raise DidResolutionError(f"response is not JSON: {exc}") from exc
    if not isinstance(data, dict):
        raise DidResolutionError("response must be a JSON object")
    return data

https_text_fetch(url, *, timeout_s=DEFAULT_TIMEOUT_S, max_bytes=MAX_RESPONSE_BYTES)

Fetch url as UTF-8 text with the same SSRF guards as :func:https_json_fetch.

For resources that are not JSON objects — a compact-JWS status-list credential (VC-JWT) or an IETF statuslist+jwt token. Raises the same error family.

Source code in src/openvc/fetch.py

def https_text_fetch(
    url: str,
    *,
    timeout_s: float = DEFAULT_TIMEOUT_S,
    max_bytes: int = MAX_RESPONSE_BYTES,
) -> str:
    """Fetch *url* as UTF-8 text with the same SSRF guards as :func:`https_json_fetch`.

    For resources that are not JSON objects — a compact-JWS status-list credential
    (VC-JWT) or an IETF ``statuslist+jwt`` token. Raises the same error family."""
    raw = _https_fetch_guarded(url, timeout_s=timeout_s, max_bytes=max_bytes)
    try:
        return raw.decode("utf-8")
    except UnicodeDecodeError as exc:
        raise DidResolutionError(f"response is not UTF-8 text: {exc}") from exc

https_bytes_fetch(url, *, timeout_s=DEFAULT_TIMEOUT_S, max_bytes=MAX_RESPONSE_BYTES)

Fetch url as raw bytes with the same SSRF guards as :func:https_json_fetch.

For resources whose exact bytes matter — a credentialSchema whose digestSRI is verified over the response before parsing. Raises the same error family.

Source code in src/openvc/fetch.py

def https_bytes_fetch(
    url: str,
    *,
    timeout_s: float = DEFAULT_TIMEOUT_S,
    max_bytes: int = MAX_RESPONSE_BYTES,
) -> bytes:
    """Fetch *url* as raw bytes with the same SSRF guards as :func:`https_json_fetch`.

    For resources whose *exact bytes* matter — a ``credentialSchema`` whose
    ``digestSRI`` is verified over the response before parsing. Raises the same
    error family."""
    return _https_fetch_guarded(url, timeout_s=timeout_s, max_bytes=max_bytes)

default_did_web_resolver()

A :class:~openvc.did.did_web.DidWebResolver wired to the SSRF-guarded fetch — the batteries-included way to resolve did:web offline of any HTTP client dependency.

Source code in src/openvc/fetch.py

def default_did_web_resolver() -> DidWebResolver:
    """A :class:`~openvc.did.did_web.DidWebResolver` wired to the SSRF-guarded
    fetch — the batteries-included way to resolve did:web offline of any HTTP
    client dependency."""
    return DidWebResolver(https_json_fetch)

default_did_webvh_resolver()

A :class:~openvc.did.did_webvh.DidWebvhResolver wired to the SSRF-guarded text fetch (did.jsonl is JSON Lines, not one JSON object) — the batteries-included did:webvh resolver.

Source code in src/openvc/fetch.py

def default_did_webvh_resolver() -> "DidWebvhResolver":
    """A :class:`~openvc.did.did_webvh.DidWebvhResolver` wired to the SSRF-guarded
    **text** fetch (``did.jsonl`` is JSON Lines, not one JSON object) — the
    batteries-included did:webvh resolver."""
    from .did.did_webvh import DidWebvhResolver
    return DidWebvhResolver(https_text_fetch)

https_json_fetch_async(url, *, timeout_s=DEFAULT_TIMEOUT_S, max_bytes=MAX_RESPONSE_BYTES) async

Async :func:https_json_fetch — same SSRF guards, run off the event loop.

Source code in src/openvc/fetch.py

async def https_json_fetch_async(
    url: str, *, timeout_s: float = DEFAULT_TIMEOUT_S, max_bytes: int = MAX_RESPONSE_BYTES,
) -> dict[str, Any]:
    """Async :func:`https_json_fetch` — same SSRF guards, run off the event loop."""
    return await asyncio.to_thread(https_json_fetch, url, timeout_s=timeout_s, max_bytes=max_bytes)

https_text_fetch_async(url, *, timeout_s=DEFAULT_TIMEOUT_S, max_bytes=MAX_RESPONSE_BYTES) async

Async :func:https_text_fetch — same SSRF guards, run off the event loop.

Source code in src/openvc/fetch.py

async def https_text_fetch_async(
    url: str, *, timeout_s: float = DEFAULT_TIMEOUT_S, max_bytes: int = MAX_RESPONSE_BYTES,
) -> str:
    """Async :func:`https_text_fetch` — same SSRF guards, run off the event loop."""
    return await asyncio.to_thread(https_text_fetch, url, timeout_s=timeout_s, max_bytes=max_bytes)

https_bytes_fetch_async(url, *, timeout_s=DEFAULT_TIMEOUT_S, max_bytes=MAX_RESPONSE_BYTES) async

Async :func:https_bytes_fetch — same SSRF guards, run off the event loop.

Source code in src/openvc/fetch.py

async def https_bytes_fetch_async(
    url: str, *, timeout_s: float = DEFAULT_TIMEOUT_S, max_bytes: int = MAX_RESPONSE_BYTES,
) -> bytes:
    """Async :func:`https_bytes_fetch` — same SSRF guards, run off the event loop."""
    return await asyncio.to_thread(https_bytes_fetch, url, timeout_s=timeout_s, max_bytes=max_bytes)

default_async_did_web_resolver()

An :class:~openvc.did.did_web.AsyncDidWebResolver wired to the SSRF-guarded async fetch — the async counterpart of :func:default_did_web_resolver.

Source code in src/openvc/fetch.py

def default_async_did_web_resolver() -> AsyncDidWebResolver:
    """An :class:`~openvc.did.did_web.AsyncDidWebResolver` wired to the SSRF-guarded
    async fetch — the async counterpart of :func:`default_did_web_resolver`."""
    return AsyncDidWebResolver(https_json_fetch_async)

default_async_did_webvh_resolver()

The async counterpart of :func:default_did_webvh_resolver — an :class:~openvc.did.did_webvh.AsyncDidWebvhResolver on the SSRF-guarded async text fetch.

Source code in src/openvc/fetch.py

def default_async_did_webvh_resolver() -> "AsyncDidWebvhResolver":
    """The async counterpart of :func:`default_did_webvh_resolver` — an
    :class:`~openvc.did.did_webvh.AsyncDidWebvhResolver` on the SSRF-guarded async
    text fetch."""
    from .did.did_webvh import AsyncDidWebvhResolver
    return AsyncDidWebvhResolver(https_text_fetch_async)