Skip to content

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
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
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
111
112
113
114
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
116
117
118
119
120
121
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
123
124
125
126
127
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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
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
171
172
173
174
175
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
177
178
179
180
181
182
183
184
185
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
187
188
189
190
191
192
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
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
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
235
236
237
238
239
240
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
242
243
244
245
246
247
248
249
250
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
252
253
254
255
256
257
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
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
@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
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
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
321
322
323
324
325
326
327
328
329
330
331
332
333
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
335
336
337
338
339
340
341
342
343
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
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
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
436
437
438
439
440
441
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
443
444
445
446
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
448
449
450
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
390
391
392
393
394
395
396
397
398
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

jwk_thumbprint_bytes(jwk, *, hash_name=_DEFAULT_THUMBPRINT_HASH)

The raw RFC 7638 JWK Thumbprint digest of jwk.

The digest covers only the members RFC 7638 declares required for the key type, serialized with no whitespace and ordered lexicographically by code point. A private key and its public half therefore produce the same thumbprint.

Fails closed: an unknown kty, a missing required member, a non-string member value, or a member holding a lone UTF-16 surrogate (JSON "\ud800", which has no UTF-8 form to digest) raises :class:InvalidKey rather than hashing a partial key.

Source code in src/openvc/keys.py
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
def jwk_thumbprint_bytes(
    jwk: Mapping[str, Any], *, hash_name: str = _DEFAULT_THUMBPRINT_HASH
) -> bytes:
    """The raw RFC 7638 JWK Thumbprint digest of *jwk*.

    The digest covers **only** the members RFC 7638 declares required for the key
    type, serialized with no whitespace and ordered lexicographically by code point.
    A private key and its public half therefore produce the *same* thumbprint.

    Fails closed: an unknown ``kty``, a missing required member, a non-string
    member value, or a member holding a lone UTF-16 surrogate (JSON ``"\\ud800"``, which
    has no UTF-8 form to digest) raises :class:`InvalidKey` rather than hashing a partial
    key.
    """
    try:
        hasher = _THUMBPRINT_HASHES[hash_name]
    except KeyError as exc:
        raise InvalidKey(f"unsupported thumbprint hash {hash_name!r}") from exc

    kty = jwk.get("kty")
    if not isinstance(kty, str):
        raise InvalidKey("JWK has no string 'kty'; cannot compute a thumbprint")
    try:
        members = _THUMBPRINT_MEMBERS[kty]
    except KeyError as exc:
        raise InvalidKey(f"unsupported JWK kty {kty!r} for a thumbprint") from exc

    canonical: dict[str, str] = {}
    for name in members:
        value = jwk.get(name)
        if not isinstance(value, str):
            # The canonical form is defined over string members. A missing or
            # non-string one means we would be digesting something other than the
            # key the caller believes they hold — reject instead of guessing.
            raise InvalidKey(
                f"JWK of kty {kty!r} needs a string {name!r} member for a thumbprint")
        # ensure_ascii=False below passes a lone UTF-16 surrogate (JSON "\ud800") through,
        # and it has no UTF-8 form to digest. Refuse it here, naming the member, not its text.
        _json.utf8(value, error=InvalidKey, what=f"JWK member {name!r}")
        canonical[name] = value

    # sort_keys gives RFC 7638 §3's lexicographic-by-code-point order; the separators
    # drop the whitespace it forbids; ensure_ascii=False keeps the UTF-8 serialization
    # the RFC specifies rather than \u-escaping it into a different digest.
    serialized = json.dumps(
        canonical, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    return hasher(serialized.encode("utf-8")).digest()

jwk_thumbprint(jwk, *, hash_name=_DEFAULT_THUMBPRINT_HASH)

The RFC 7638 JWK Thumbprint of jwk, base64url-encoded without padding.

This is the form used as a kid, in a cnf claim, and as the DPoP jkt. For the raw digest — e.g. the mdoc_jwk_thumbprint an ISO 18013-7 SessionTranscript needs — use :func:jwk_thumbprint_bytes.

Fails closed exactly as :func:jwk_thumbprint_bytes does: :class:InvalidKey for an unknown kty, a missing or non-string required member, or a member holding a lone UTF-16 surrogate.

Source code in src/openvc/keys.py
551
552
553
554
555
556
557
558
559
560
561
562
563
564
def jwk_thumbprint(
    jwk: Mapping[str, Any], *, hash_name: str = _DEFAULT_THUMBPRINT_HASH
) -> str:
    """The RFC 7638 JWK Thumbprint of *jwk*, base64url-encoded without padding.

    This is the form used as a `kid`, in a `cnf` claim, and as the DPoP `jkt`. For
    the raw digest — e.g. the ``mdoc_jwk_thumbprint`` an ISO 18013-7 SessionTranscript
    needs — use :func:`jwk_thumbprint_bytes`.

    Fails closed exactly as :func:`jwk_thumbprint_bytes` does: :class:`InvalidKey` for an
    unknown ``kty``, a missing or non-string required member, or a member holding a lone
    UTF-16 surrogate.
    """
    return _b64url_encode(jwk_thumbprint_bytes(jwk, hash_name=hash_name))

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
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
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
629
630
631
632
633
634
635
636
637
638
639
640
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
49
50
51
52
53
54
55
56
57
58
59
@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
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@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.
        This is the JOSE lanes' lookup; a Data Integrity proof's key is selected by
        the stricter, exact-id :meth:`key_for_purpose`."""
        if kid is None:
            return self.verification_methods[0] if len(self.verification_methods) == 1 else None
        if not isinstance(kid, str):
            return 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 _base(self) -> str | None:
        """The document ``id`` when it is a DID (DID Core 1.0 §3.1), the only base a
        ``#fragment`` resolves against (§3.2.2); else ``None``.

        A lookup calls this a fixed number of times, never once per method or
        reference: the ``id`` (as long as the DID the proof names) and the number of
        references are both the document host's to choose, so re-checking the ``id``
        per reference would cost their product."""
        doc_id = self.id
        return doc_id if isinstance(doc_id, str) and _DID_SYNTAX.fullmatch(doc_id) else None

    @staticmethod
    def _absolute(url: Any, base: str | None) -> str | None:
        """*url* — a method ``id`` or a relationship reference — as an absolute DID URL,
        or ``None`` when it does not resolve to one (DID Core 1.0 §3.2.2).

        A ``did:`` URL is returned unchanged. A ``#fragment`` is resolved against the
        document ``id`` (RFC 3986 §5), and only when that ``id`` is a DID (§3.1), which
        is what *base* is (:meth:`_base`): ``#key-1`` in ``did:example:123`` is
        ``did:example:123#key-1``. Anything else is ``None``: a bare ``key-1`` is a
        relative *path* reference, and a ``?query``, ``/path`` or ``https:`` URL can
        never resolve to ``<did>#fragment`` either, so treating them as unresolvable
        fails closed instead of resolving them in full."""
        if not isinstance(url, str):
            return None
        if url.startswith("did:"):
            return url
        if url.startswith("#") and base is not None:
            return base + url
        return None

    @staticmethod
    def _spellings(absolute: Any, base: str | None) -> tuple[str, ...]:
        """Every string that :meth:`_absolute` resolves to *absolute* against *base*:
        *absolute* itself, when it is a ``did:`` URL, and, when it is *base* followed by
        a ``#fragment``, that fragment. Nothing resolves to a value that is not a
        ``did:`` URL, so that is ``()``.

        A method id or reference is then matched by comparing it with these, not by
        resolving it: each comparison costs at most the candidate's own length, so a
        lookup stays linear in the document size. Resolving every ``#fragment`` would
        copy the document ``id`` once per reference."""
        if not isinstance(absolute, str) or not absolute.startswith("did:"):
            return ()
        if (base is not None and absolute.startswith(base)
                and absolute.startswith("#", len(base))):
            return (absolute, absolute[len(base):])
        return (absolute,)

    def _methods_with_id(self, vm_id: Any, base: str | None) -> list[VerificationMethod]:
        """The methods whose absolute ``id`` (against *base*, :meth:`_base`) is exactly
        *vm_id* (CID 1.0 §3.3 step 9): a plain, case-sensitive string comparison, with
        no fallback to the fragment."""
        spellings = self._spellings(vm_id, base)
        return [vm for vm in self.verification_methods
                if isinstance(vm.id, str) and vm.id in spellings]

    def key_for_purpose(
        self, kid: str | None, proof_purpose: str
    ) -> VerificationMethod | None:
        """The verification method *kid* names, if it is authorized for *proof_purpose*.

        This is the Data Integrity lookup, and it is stricter than :meth:`key_by_kid`
        (which the JOSE lanes keep, since a JWS ``kid`` is only a hint). Method ids and
        relationship references are compared as absolute DID URLs, resolved as DID
        Core 1.0 §3.2.2 says (see :meth:`_absolute`): ``#key-1`` in the document for
        ``did:example:123`` is ``did:example:123#key-1``, while a bare ``key-1``, a
        ``?query``, a ``/path`` or a non-DID URL resolves to nothing. The comparison is
        a plain, case-sensitive string equality, as in CID 1.0 §3.3 (Retrieve
        Verification Method):

        1. **select** (step 9): *kid* must equal the absolute ``id`` of exactly one
           method. Nothing falls back to the fragment, and a second method with the
           same absolute ``id`` is not resolved by document order: returns None.
        2. **grant** (step 11): when the document declares the relationship named by
           *proof_purpose*, one of its references (a string, or the ``id`` of an
           embedded method) must resolve to that same absolute ``id``. A reference to
           another DID, another version, a path or a non-DID URL grants nothing, even
           when its fragment matches; returns None (the key exists but is not usable
           for this purpose). A declared but empty relationship grants nothing.

        If the document does not declare the relationship at all, the binding cannot
        be enforced and the selected method is returned as-is. A *kid* of None falls
        back to the document's sole method, if it has exactly one, and that method
        must then pass step 2. A *proof_purpose* that is not a string (including
        None) names no relationship and returns None.

        The document ``id`` is checked a fixed number of times, and only the selected
        method's own relative ``id`` is resolved; no other method id or reference is:
        each is compared with the spellings of the one absolute ``id`` it would have to
        resolve to (:meth:`_spellings`), computed once per lookup. The cost is linear in
        the document size, however long its ``id`` and however many references it
        lists."""
        if not isinstance(proof_purpose, str):                 # unhashable -> no key
            return None
        base = self._base()                    # once per lookup, never per reference
        if kid is None:
            vm = self.key_by_kid(None)                         # the sole-key fallback
        else:
            matches = self._methods_with_id(kid, base)
            vm = matches[0] if len(matches) == 1 else None
        if vm is None:
            return None
        refs = self.relationships.get(proof_purpose)
        if refs is None:                       # relationship not declared -> lenient
            return vm
        # an unresolvable method id has no spellings, so nothing grants it
        spellings = self._spellings(self._absolute(vm.id, base), base)
        if any(isinstance(ref, str) and ref in spellings 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. This is the JOSE lanes' lookup; a Data Integrity proof's key is selected by the stricter, exact-id :meth:key_for_purpose.

Source code in src/openvc/did/base.py
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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.
    This is the JOSE lanes' lookup; a Data Integrity proof's key is selected by
    the stricter, exact-id :meth:`key_for_purpose`."""
    if kid is None:
        return self.verification_methods[0] if len(self.verification_methods) == 1 else None
    if not isinstance(kid, str):
        return 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)

The verification method kid names, if it is authorized for proof_purpose.

This is the Data Integrity lookup, and it is stricter than :meth:key_by_kid (which the JOSE lanes keep, since a JWS kid is only a hint). Method ids and relationship references are compared as absolute DID URLs, resolved as DID Core 1.0 §3.2.2 says (see :meth:_absolute): #key-1 in the document for did:example:123 is did:example:123#key-1, while a bare key-1, a ?query, a /path or a non-DID URL resolves to nothing. The comparison is a plain, case-sensitive string equality, as in CID 1.0 §3.3 (Retrieve Verification Method):

  1. select (step 9): kid must equal the absolute id of exactly one method. Nothing falls back to the fragment, and a second method with the same absolute id is not resolved by document order: returns None.
  2. grant (step 11): when the document declares the relationship named by proof_purpose, one of its references (a string, or the id of an embedded method) must resolve to that same absolute id. A reference to another DID, another version, a path or a non-DID URL grants nothing, even when its fragment matches; returns None (the key exists but is not usable for this purpose). A declared but empty relationship grants nothing.

If the document does not declare the relationship at all, the binding cannot be enforced and the selected method is returned as-is. A kid of None falls back to the document's sole method, if it has exactly one, and that method must then pass step 2. A proof_purpose that is not a string (including None) names no relationship and returns None.

The document id is checked a fixed number of times, and only the selected method's own relative id is resolved; no other method id or reference is: each is compared with the spellings of the one absolute id it would have to resolve to (:meth:_spellings), computed once per lookup. The cost is linear in the document size, however long its id and however many references it lists.

Source code in src/openvc/did/base.py
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def key_for_purpose(
    self, kid: str | None, proof_purpose: str
) -> VerificationMethod | None:
    """The verification method *kid* names, if it is authorized for *proof_purpose*.

    This is the Data Integrity lookup, and it is stricter than :meth:`key_by_kid`
    (which the JOSE lanes keep, since a JWS ``kid`` is only a hint). Method ids and
    relationship references are compared as absolute DID URLs, resolved as DID
    Core 1.0 §3.2.2 says (see :meth:`_absolute`): ``#key-1`` in the document for
    ``did:example:123`` is ``did:example:123#key-1``, while a bare ``key-1``, a
    ``?query``, a ``/path`` or a non-DID URL resolves to nothing. The comparison is
    a plain, case-sensitive string equality, as in CID 1.0 §3.3 (Retrieve
    Verification Method):

    1. **select** (step 9): *kid* must equal the absolute ``id`` of exactly one
       method. Nothing falls back to the fragment, and a second method with the
       same absolute ``id`` is not resolved by document order: returns None.
    2. **grant** (step 11): when the document declares the relationship named by
       *proof_purpose*, one of its references (a string, or the ``id`` of an
       embedded method) must resolve to that same absolute ``id``. A reference to
       another DID, another version, a path or a non-DID URL grants nothing, even
       when its fragment matches; returns None (the key exists but is not usable
       for this purpose). A declared but empty relationship grants nothing.

    If the document does not declare the relationship at all, the binding cannot
    be enforced and the selected method is returned as-is. A *kid* of None falls
    back to the document's sole method, if it has exactly one, and that method
    must then pass step 2. A *proof_purpose* that is not a string (including
    None) names no relationship and returns None.

    The document ``id`` is checked a fixed number of times, and only the selected
    method's own relative ``id`` is resolved; no other method id or reference is:
    each is compared with the spellings of the one absolute ``id`` it would have to
    resolve to (:meth:`_spellings`), computed once per lookup. The cost is linear in
    the document size, however long its ``id`` and however many references it
    lists."""
    if not isinstance(proof_purpose, str):                 # unhashable -> no key
        return None
    base = self._base()                    # once per lookup, never per reference
    if kid is None:
        vm = self.key_by_kid(None)                         # the sole-key fallback
    else:
        matches = self._methods_with_id(kid, base)
        vm = matches[0] if len(matches) == 1 else None
    if vm is None:
        return None
    refs = self.relationships.get(proof_purpose)
    if refs is None:                       # relationship not declared -> lenient
        return vm
    # an unresolvable method id has no spellings, so nothing grants it
    spellings = self._spellings(self._absolute(vm.id, base), base)
    if any(isinstance(ref, str) and ref in spellings for ref in refs):
        return vm
    return None

DidResolver

Bases: Protocol

Source code in src/openvc/did/base.py
209
210
211
212
213
214
@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
211
212
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
213
214
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
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
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
369
370
371
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
373
374
375
376
377
378
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
385
386
387
388
389
390
391
392
393
@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
390
391
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
392
393
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
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
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
424
425
426
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
428
429
430
431
432
433
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
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
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.

The document is attacker-hosted JSON, so its shape is checked, not assumed. A document or didDocument wrapper that is not an object, and a verificationMethod or relationship (assertionMethod …) that is present but not an array, raise :class:DidResolutionError — never read as an absent member, since an undeclared relationship leaves the key purpose unenforced. The check is per document, not per lane: a relationship that is not an array rejects the whole document, even for a lane that never reads relationships. A method whose id is not a string is dropped and a relationship entry that is neither an id string nor an object with a string id is dropped. A dropped method is never selected, by kid or by purpose, and a dropped entry grants nothing. The kid-less fallback to the sole key counts the methods that remain, so a dropped method that carried a key no longer makes it ambiguous (as a method with an undecodable Multikey never did). An id / type / controller that is not a string reads as "".

Source code in src/openvc/did/base.py
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
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.

    The document is attacker-hosted JSON, so its *shape* is checked, not assumed. A
    document or ``didDocument`` wrapper that is not an object, and a ``verificationMethod``
    or relationship (``assertionMethod`` …) that is present but not an array, raise
    :class:`DidResolutionError` — never read as an absent member, since an undeclared
    relationship leaves the key purpose unenforced. The check is per document, not per
    lane: a relationship that is not an array rejects the whole document, even for a
    lane that never reads relationships. A method whose ``id`` is not a string is dropped
    and a relationship entry that is neither an id string nor an object with a string
    ``id`` is dropped. A dropped method is never selected, by ``kid`` or by purpose, and
    a dropped entry grants nothing. The ``kid``-less fallback to the sole key counts the
    methods that remain, so a dropped method that carried a key no longer makes it
    ambiguous (as a method with an undecodable Multikey never did). An ``id`` / ``type`` /
    ``controller`` that is not a string reads as ``""``."""
    if not isinstance(raw, dict):            # an injected fetch can hand over any JSON value
        raise DidResolutionError(
            f"DID document must be a JSON object, not {type(raw).__name__}")
    if _json.nesting_exceeds(raw):           # an injected fetch hands it over already parsed
        raise DidResolutionError(
            f"DID document is nested deeper than {_json.MAX_DEPTH} levels")
    doc = raw["didDocument"] if "didDocument" in raw else raw    # present-but-null is not absent
    if not isinstance(doc, dict):
        raise DidResolutionError(
            f"didDocument must be an object, not {type(doc).__name__}")
    doc_id = _text(doc.get("id"))
    vms = [
        VerificationMethod(
            id=vm["id"],
            type=_text(vm.get("type")),
            controller=_text(vm["controller"]) if "controller" in vm else doc_id,
            public_key_jwk=jwk,
        )
        for vm in _array(doc.get("verificationMethod", []), "verificationMethod")
        # an entry without a string id names no key (VerificationMethod.kid splits it)
        if isinstance(vm, dict) and isinstance(vm.get("id"), str)
        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_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
410
411
412
413
414
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
111
112
113
114
115
116
117
118
119
120
121
122
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 address that is not globally routable (private / loopback / link-local / reserved / multicast / unspecified, and RFC 6598 CGNAT 100.64.0.0/10 — 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
47
48
class UnsafeUrlError(DidResolutionError):
    """The URL or a resolved address is not allowed (scheme, host, or IP range)."""

ip_is_forbidden(ip_str)

True for any non-globally-routable / SSRF-sensitive address.

The flag union covers private / loopback / link-local / reserved / multicast / unspecified. not is_global additionally closes RFC 6598 shared space (100.64.0.0/10, CGNAT), which is_private does not set — multicast stays in the union because some multicast addresses report is_global=True, and so does the deprecated IPv6 site-local prefix fec0::/10 (RFC 3879), which is_site_local catches. IPv4-mapped forms inherit the same classification (::ffff:100.64.0.1).

Public so a caller that brings its own HTTP client applies the same rule instead of a copy of it (the _ip_is_forbidden name is kept as the same function). Raises ValueError for a string that is not an IP address; the caller decides what that means.

Source code in src/openvc/fetch.py
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def ip_is_forbidden(ip_str: str) -> bool:
    """True for any non-globally-routable / SSRF-sensitive address.

    The flag union covers private / loopback / link-local / reserved /
    multicast / unspecified. ``not is_global`` additionally closes RFC 6598
    shared space (``100.64.0.0/10``, CGNAT), which ``is_private`` does not
    set — multicast stays in the union because some multicast addresses
    report ``is_global=True``, and so does the deprecated IPv6 site-local
    prefix ``fec0::/10`` (RFC 3879), which ``is_site_local`` catches.
    IPv4-mapped forms inherit the same classification (``::ffff:100.64.0.1``).

    Public so a caller that brings its own HTTP client applies the same rule instead of
    a copy of it (the ``_ip_is_forbidden`` name is kept as the same function). Raises
    ``ValueError`` for a string that is not an IP address; the caller decides what that
    means.
    """
    ip = ipaddress.ip_address(ip_str)
    return (
        ip.is_private or ip.is_loopback or ip.is_link_local
        or ip.is_reserved or ip.is_multicast or ip.is_unspecified
        or not ip.is_global
        # fec0::/10 reports is_global=True and no other flag; only IPv6 has the property.
        or (isinstance(ip, ipaddress.IPv6Address) and ip.is_site_local)
    )

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
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
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
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
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
241
242
243
244
245
246
247
248
249
250
251
252
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
255
256
257
258
259
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
262
263
264
265
266
267
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
282
283
284
285
286
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
289
290
291
292
293
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
296
297
298
299
300
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
303
304
305
306
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
309
310
311
312
313
314
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)