Async verification

openvc.aio is the async counterpart of the sync verification pipeline, for asyncio servers (FastAPI / Starlette): a handler awaits verify_credential_async instead of offloading the whole call to a thread pool, and verify_many_async verifies a presentation cascade concurrently instead of serialising N blocking fetches.

It reuses every proof suite, status/schema codec and binding check of the sync path unchanged — only the I/O sequencing is re-expressed with await, so there is no second implementation of any signature check to drift. The batteries-included async fetch (openvc.fetch.https_json_fetch_async) runs the exact same SSRF / DNS-rebind guard as the sync fetch under asyncio.to_thread; a caller may inject an httpx.AsyncClient-backed fetch instead. See ADR-0002 for the design and its trade-offs.

import asyncio
from openvc import verify_credential_async
from openvc.aio import default_async_resolver

async def main():
    result = await verify_credential_async(token, resolver=default_async_resolver())
    print(result.issuer)

asyncio.run(main())

openvc.aio

openvc.aio — an additive async verification surface over the sync pipeline.

:func:verify_credential_async mirrors :func:openvc.verify.verify_credential exactly — same formats, same policy, same fail-closed guarantees — but await\s its I/O (DID resolution, jwt-vc-issuer discovery, status-list and credentialSchema fetches) instead of blocking. It reuses every pure/CPU helper of the sync path unchanged (the proof suites, the status/schema codecs, the issuer binding and type checks); only the sequencing around the I/O boundaries is re-expressed with await. There is no second implementation of any signature check to drift — see docs/adr/ADR-0002-async-verification.md.

Injection mirrors the sync path with async counterparts: resolver is an :class:~openvc.did.base.AsyncDidResolver; resolve_status_list / resolve_status_list_token / resolve_credential_schema / jwt_vc_issuer_fetch each return an awaitable. The caller supplies the transport — an httpx.AsyncClient-backed fetch, or the batteries-included :func:openvc.fetch.https_json_fetch_async (the exact same SSRF/DNS-rebind guard as the sync fetch, run off the event loop). :func:default_async_resolver wires the offline did:key / did:jwk resolvers (via :func:~openvc.did.base.as_async_resolver) and the SSRF-guarded async did:web resolver.

:func:verify_many_async verifies a batch concurrently (asyncio.gather), each item independently fail-closed — the fix for a presentation cascade that would otherwise serialise N blocking fetches.

Data Integrity note: the embedded-proof suites resolve the proof's verificationMethod synchronously; the async path resolves that DID first (awaiting the injected resolver) and hands the suite a one-shot in-memory resolver, so the suite's own resolution does no I/O. The canonicalisation + signature check then run inline on the event loop (bounded CPU).

default_async_resolver()

The async counterpart of :func:openvc.verify.default_resolver: the offline did:key / did:jwk resolvers (adapted via :func:as_async_resolver) plus the SSRF-guarded async did:web and did:webvh resolvers.

Source code in src/openvc/aio.py

def default_async_resolver() -> AsyncDidResolverRegistry:
    """The async counterpart of :func:`openvc.verify.default_resolver`: the offline
    ``did:key`` / ``did:jwk`` resolvers (adapted via :func:`as_async_resolver`) plus
    the SSRF-guarded async ``did:web`` and ``did:webvh`` resolvers."""
    from .did.did_jwk import DidJwkResolver
    from .did.did_key import DidKeyResolver
    from .fetch import default_async_did_web_resolver, default_async_did_webvh_resolver
    return AsyncDidResolverRegistry([
        as_async_resolver(DidKeyResolver()),
        as_async_resolver(DidJwkResolver()),
        default_async_did_web_resolver(),
        default_async_did_webvh_resolver(),
    ])

verify_credential_async(credential, *, policy=None, resolver=None, resolve_status_list=None, resolve_status_list_token=None, resolve_credential_schema=None, jwt_vc_issuer_fetch=None, x5c_trust_anchors=None, extra_contexts=None, _depth=0) async

Async :func:openvc.verify.verify_credential. Every keyword means what it does there, with async counterparts for the injected I/O: resolver is an :class:~openvc.did.base.AsyncDidResolver (default :func:default_async_resolver), and the status/schema/issuer fetches return awaitables. Same formats, same policy, same fail-closed behaviour and same :class:VerificationResult.

Source code in src/openvc/aio.py

async def verify_credential_async(
    credential: Any,
    *,
    policy: VerificationPolicy | None = None,
    resolver: Any = None,
    resolve_status_list: Any = None,
    resolve_status_list_token: Any = None,
    resolve_credential_schema: Any = None,
    jwt_vc_issuer_fetch: Any = None,
    x5c_trust_anchors: Any = None,
    extra_contexts: Any = None,
    _depth: int = 0,
) -> VerificationResult:
    """Async :func:`openvc.verify.verify_credential`. Every keyword means what it does
    there, with async counterparts for the injected I/O: *resolver* is an
    :class:`~openvc.did.base.AsyncDidResolver` (default :func:`default_async_resolver`),
    and the status/schema/issuer fetches return awaitables. Same formats, same policy,
    same fail-closed behaviour and same :class:`VerificationResult`."""
    policy = policy or VerificationPolicy()
    resolver = resolver if resolver is not None else default_async_resolver()

    fmt = detect_format(credential)
    logger.debug("verify_async: format=%s", fmt)
    with span("openvc.verify_credential_async", format=fmt):
        try:
            result = await _verify_by_format_async(
                credential, fmt, policy, resolver, resolve_status_list,
                resolve_status_list_token, resolve_credential_schema,
                jwt_vc_issuer_fetch, x5c_trust_anchors, extra_contexts, _depth)
        except OpenvcError as exc:
            logger.info("verify_async failed: format=%s error=%s", fmt, type(exc).__name__)
            raise
        logger.debug("verify_async ok: format=%s issuer=%s", result.format, result.issuer)
    return result

verify_many_async(credentials, *, policy=None, resolver=None, resolve_status_list=None, resolve_status_list_token=None, resolve_credential_schema=None, jwt_vc_issuer_fetch=None, x5c_trust_anchors=None, extra_contexts=None) async

Verify many credentials concurrently (asyncio.gather), each independently fail-closed: one failure becomes that item's error and never aborts the others. Returns one :class:BatchResult per input, in input order. Unlike the sync :func:openvc.verify.verify_many, this does not de-duplicate shared resolutions across credentials (that cache is not concurrency-safe — ADR-0002 D4); overlapping the I/O is the win here.

Source code in src/openvc/aio.py

async def verify_many_async(
    credentials: Sequence[Any],
    *,
    policy: VerificationPolicy | None = None,
    resolver: Any = None,
    resolve_status_list: Any = None,
    resolve_status_list_token: Any = None,
    resolve_credential_schema: Any = None,
    jwt_vc_issuer_fetch: Any = None,
    x5c_trust_anchors: Any = None,
    extra_contexts: Any = None,
) -> list[BatchResult]:
    """Verify many credentials **concurrently** (``asyncio.gather``), each independently
    fail-closed: one failure becomes that item's ``error`` and never aborts the others.
    Returns one :class:`BatchResult` per input, in input order. Unlike the sync
    :func:`openvc.verify.verify_many`, this does not de-duplicate shared resolutions
    across credentials (that cache is not concurrency-safe — ADR-0002 D4); overlapping
    the I/O is the win here."""
    resolver = resolver if resolver is not None else default_async_resolver()

    async def _one(index: int, credential: Any) -> BatchResult:
        try:
            result = await verify_credential_async(
                credential, policy=policy, resolver=resolver,
                resolve_status_list=resolve_status_list,
                resolve_status_list_token=resolve_status_list_token,
                resolve_credential_schema=resolve_credential_schema,
                jwt_vc_issuer_fetch=jwt_vc_issuer_fetch,
                x5c_trust_anchors=x5c_trust_anchors, extra_contexts=extra_contexts)
            return BatchResult(index=index, result=result)
        except OpenvcError as exc:          # per-credential fail-closed
            return BatchResult(index=index, error=exc)

    with span("openvc.verify_many_async", count=len(credentials)):
        results = await asyncio.gather(
            *(_one(i, c) for i, c in enumerate(credentials)))
    ok = sum(1 for r in results if r.ok)
    logger.debug("verify_many_async: %d credentials, %d ok", len(results), ok)
    return list(results)

Async resolvers & fetch

The async DID-resolution and fetch primitives the pipeline awaits.

openvc.did.base.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`."""

openvc.did.base.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}")

openvc.did.base.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_web.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)

openvc.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)

openvc.fetch.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)

openvc.fetch.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)

openvc.fetch.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)