diff --git a/CHANGELOG.md b/CHANGELOG.md index 48597b8..7eb2663 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,55 @@ All notable changes to **openvc** are documented here. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and the project aims for [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.22.0] — unreleased + +### Added + +- **ETSI TS 119 602 Lists of Trusted Entities (LoTE) — the JSON trusted-list + lane** ([#135](https://github.com/luisgf/openvc/issues/135)). TS 119 602 is + the successor data model to the TS 119 612 XML Trusted Lists, and its EU + profiles are the EUDI wallet anchor lists: **Annex F** (WRPAC providers — + who may issue relying-party *access* certificates) and **Annex G** (WRPRC + providers — the registrar anchors `verify_rp_registration_certificate` + consumes). One interface, two encodings: `openvc.trustlist` grows + `parse_lote` / `consume_lote` / `walk_lote` plus the + `EU_WRPAC_PROVIDERS_PROFILE` / `EU_WRPRC_PROVIDERS_PROFILE` conformance + gates (`LoteProfile`, `LoteType`, `LoteServiceType`, + `TrustListProfileError`), all distilling into the same `TrustAnchorSet` + as `walk_lotl` — `.certificates` feeds the existing X.509 path unchanged. + + A JSON LoTE travels as a **compact JAdES baseline-B** JWS (clause 6.8 / + Annex G.4), so verification runs on the library's own JOSE primitives — the + `{ES256, ES384, EdDSA, Ed25519}` allow-list before any crypto, the WRPRC + lane's allow-listed `crit`, the signer from `x5c` authenticated against + **caller-pinned** certificates (byte-for-byte or by path validation), plus + clause 6.8's DN binding (signing-certificate `organizationName` ↔ + `SchemeOperatorName`, `countryName` ↔ `SchemeTerritory`). Parsing is strict + and fail-closed on every field that feeds a trust decision: unknown + structural members reject (the official schema is + `additionalProperties: false` throughout), date-times must be the UTC `Z` + form, an unrecognised critical extension rejects the list, a malformed + certificate blob is skipped rather than trusted, and a **closed** list + (`NextUpdate` null) contributes zero anchors. The EU profiles enforce + Tables F.1–G.3 — registered URIs (including the spec's literal + `WRPRCrovidersList` StatusDetn typo, accepted alongside the corrected + spelling), territory `EU`, the exclusive service-type pairs, + `ServiceStatus`/`StatusStartingTime`/`HistoricalInformationPeriod` absent, + and the ≤ 6-month update window. Self-made signed vectors pin the behaviour; + the Commission's real EU lists become golden fixtures when published. + + The adversarial review hardened the lane before merge: a profiled + `walk_lote` now defaults its selection to the profile's **issuance** service + type (a provider's *revocation*-service certificates no longer anchor + credential verification unless explicitly selected — the review proved a + WRPRC signed under a revocation-service CA validated through the documented + flow), only follows pointers whose `LoTEType` matches the profile (and + consumes the pointed list under the same profile), fails closed instead of + raising an uncaught `ValueError` on a far-future `ListIssueDateTime`, + rejects a present-but-empty `ServiceStatus` under the profiles + (presence is the violation), and pins date-times to the exact + `YYYY-MM-DDThh:mm:ssZ` form clause 6.1.3 mandates. + ## [1.21.0] — 2026-07-19 ### Added diff --git a/src/openvc/trustlist/__init__.py b/src/openvc/trustlist/__init__.py index 57e8d48..3afb53d 100644 --- a/src/openvc/trustlist/__init__.py +++ b/src/openvc/trustlist/__init__.py @@ -1,11 +1,14 @@ """ -openvc.trustlist — consume EU Trusted Lists (LOTL → national TL) as a verifier -X.509 trust-anchor source (eIDAS 2.0 / EUDI, ETSI TS 119 612). +openvc.trustlist — consume EU trusted lists as a verifier X.509 trust-anchor +source (eIDAS 2.0 / EUDI): **one interface, two encodings**. -A Trusted List is, for a verifier, a **source of EU-recognised X.509 anchors**. -:func:`walk_lotl` turns the Commission's List of Trusted Lists and the national -lists it points at into a :class:`TrustAnchorSet`; its ``.certificates`` feed the -existing X.509 path directly: +* **ETSI TS 119 612** XML Trusted Lists (LOTL → national TL): :func:`walk_lotl`. +* **ETSI TS 119 602** JSON Lists of Trusted Entities (LoTE), the successor data + model whose EU profiles carry the EUDI wallet anchor lists — Annex F (WRPAC + providers) and Annex G (WRPRC providers): :func:`walk_lote`. + +Both distil into the same :class:`TrustAnchorSet`; its ``.certificates`` feed +the existing X.509 path directly: from openvc import verify_credential from openvc.trustlist import walk_lotl @@ -19,10 +22,12 @@ verify_credential(vc, x5c_trust_anchors=anchors.certificates) This adds no verification surface — :mod:`openvc.x5c` remains the path validator; -trust lists only tell it *which roots are EU-recognised*. Parsing is hardened -stdlib XML (no DTD/XXE, bounded); XML-signature verification is an **injected -callback** (the ``[trustlist]`` extra ships a reference XAdES one). See -``docs/adr/ADR-0003-eu-trusted-lists.md``. +trust lists only tell it *which roots are EU-recognised*. XML parsing is hardened +stdlib (no DTD/XXE, bounded) with XML-signature verification an **injected +callback** (the ``[trustlist]`` extra ships a reference XAdES one); the LoTE lane +is signed as compact JAdES and verifies on the library's own JOSE primitives, no +extra needed. See ``docs/adr/ADR-0003-eu-trusted-lists.md`` and +:mod:`openvc.trustlist.lote`. """ from __future__ import annotations @@ -40,10 +45,23 @@ from .errors import ( TrustListError, TrustListParseError, + TrustListProfileError, TrustListSignatureBackendUnavailable, TrustListSignatureError, TrustListSignatureUnavailable, ) +from .lote import ( + EU_WRPAC_PROVIDERS_PROFILE, + EU_WRPRC_PROVIDERS_PROFILE, + FetchLote, + LoteProfile, + LoteServiceType, + LoteType, + consume_lote, + default_lote_fetch, + parse_lote, + walk_lote, +) from .model import ( TrustAnchorSet, TrustList, @@ -57,7 +75,13 @@ __all__ = [ "DEFAULT_SELECT", + "EU_WRPAC_PROVIDERS_PROFILE", + "EU_WRPRC_PROVIDERS_PROFILE", + "FetchLote", "FetchTrustList", + "LoteProfile", + "LoteServiceType", + "LoteType", "Select", "ServiceStatus", "ServiceType", @@ -66,6 +90,7 @@ "TrustListError", "TrustListParseError", "TrustListProblem", + "TrustListProfileError", "TrustListSignatureBackendUnavailable", "TrustListSignatureError", "TrustListSignatureUnavailable", @@ -73,9 +98,13 @@ "TrustServiceProvider", "TslPointer", "VerifySignature", + "consume_lote", "consume_trust_list", + "default_lote_fetch", "default_trust_list_fetch", + "parse_lote", "parse_trust_list", "verify_xades_enveloped", + "walk_lote", "walk_lotl", ] diff --git a/src/openvc/trustlist/errors.py b/src/openvc/trustlist/errors.py index 36b9ca9..3d4ba4e 100644 --- a/src/openvc/trustlist/errors.py +++ b/src/openvc/trustlist/errors.py @@ -21,8 +21,17 @@ class TrustListSignatureUnavailable(TrustListError): class TrustListSignatureError(TrustListError): - """The Trusted List's XML signature did not verify against the expected signer - certificate(s) — the list is not authentic.""" + """The Trusted List's signature (XML XAdES, or a LoTE's compact JAdES) did + not verify against the expected signer certificate(s) — the list is not + authentic.""" + + +class TrustListProfileError(TrustListError): + """A verified, well-formed LoTE does not conform to the requested profile + (ETSI TS 119 602 clause 4.7 — e.g. the Annex F/G EU WRPAC/WRPRC providers + lists): wrong ``LoTEType``, a forbidden component present, a service type + outside the profile's exclusive set, or an update window over the ceiling. + Fail-closed: a non-conformant list contributes no anchors.""" class TrustListSignatureBackendUnavailable(TrustListSignatureUnavailable): @@ -36,6 +45,7 @@ class TrustListSignatureBackendUnavailable(TrustListSignatureUnavailable): __all__ = [ "TrustListError", "TrustListParseError", + "TrustListProfileError", "TrustListSignatureBackendUnavailable", "TrustListSignatureError", "TrustListSignatureUnavailable", diff --git a/src/openvc/trustlist/lote.py b/src/openvc/trustlist/lote.py new file mode 100644 index 0000000..18515b2 --- /dev/null +++ b/src/openvc/trustlist/lote.py @@ -0,0 +1,916 @@ +""" +openvc.trustlist.lote — consume ETSI TS 119 602 **Lists of Trusted Entities** +(LoTE, the JSON trusted-list binding) as a verifier X.509 trust-anchor source. + +TS 119 602 V1.1.1 is the format-agnostic successor data model to the TS 119 612 +XML Trusted Lists (its Table A.1 maps the two field-by-field), and its EU +profiles are the anchor sources for the EUDI wallet ecosystem: Annex F is the +**WRPAC providers list** (who may issue relying-party access certificates) and +Annex G the **WRPRC providers list** (who may issue relying-party registration +certificates — the registrar anchors +:func:`openvc.rp_registration.verify_rp_registration_certificate` consumes). +One interface, two encodings: a parsed LoTE lands in the same +:class:`~openvc.trustlist.model.TrustList` / :class:`~openvc.trustlist.model.TrustAnchorSet` +shapes the 119 612 lane produces, so ``.certificates`` feeds the existing X.509 +path unchanged. + +A JSON LoTE is distributed as a **compact JAdES baseline-B signature whose +payload is the list** (clause 6.8 + Annexes D.4–I.4: "compact JAdES Baseline B +… as specified in ETSI TS 119 182-1"), so verification runs on the library's +JOSE primitives: the ``{ES256, ES384, EdDSA, Ed25519}`` allow-list is applied +**before** any crypto, ``crit`` is allow-listed exactly like the WRPRC lane +(:mod:`openvc.rp_registration`), the signer comes from ``x5c`` and must +authenticate against **caller-pinned** certificates (no implicit root — the +same discipline as :func:`openvc.trustlist.walk_lotl`), and clause 6.8's DN +binding is enforced: the signing certificate's ``organizationName`` must match +a ``SchemeOperatorName`` value and its ``countryName`` the ``SchemeTerritory``. + +Parsing is **strict and fail-closed on every field that feeds a trust +decision**: unknown structural members are rejected (the official JSON schema +is ``additionalProperties: false`` throughout), a ``bool`` is never accepted +where an ``int`` is required, date-times must be the UTC ``Z`` form clause +6.1.3 mandates, and an unrecognised **critical** scheme/service extension +rejects the list (clause 6.3.17). Purely informational sub-structures +(addresses, policy notices) are container-type-checked and otherwise carried +opaquely. A certificate blob that does not load is skipped, never silently +trusted (the :mod:`openvc.trustlist.parse` convention). + +Two spec warts, handled deliberately: + +* Table G.1 / clause C.2.2 spell the WRPRC status-determination URI + ``…/WRPRCrovidersList/StatusDetn/EU`` (sic — the "P" of "Providers" is + missing) and the EUDI reference implementation carries the typo verbatim, so + :data:`EU_WRPRC_PROVIDERS_PROFILE` accepts **both** the literal and the + corrected spelling — scheme metadata, not an authorization grant, so the + tolerance cannot widen any privilege. +* The official JSON schema nests an ``additionalProperties: false`` *inside* + ``ServiceDigitalIdentity``'s ``properties`` (declaring a never-valid literal + property of that name); this parser is the authority instead, with the five + schema-defined members allowed. +""" +from __future__ import annotations + +import base64 +import calendar +import json +import re +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Callable, Mapping, NoReturn, Sequence + +from .consume import Select +from .errors import ( + TrustListError, + TrustListParseError, + TrustListProfileError, + TrustListSignatureError, +) +from .model import ( + TrustAnchorSet, + TrustList, + TrustListProblem, + TrustServiceAnchor, + TrustServiceProvider, + TslPointer, +) +from .parse import DEFAULT_MAX_BYTES + +# Fetch a LoTE URL -> its raw bytes (pass an SSRF-guarded fetch). +FetchLote = Callable[[str], bytes] + +_URI_19602 = "http://uri.etsi.org/19602" + +# JAdES header parameters this verifier processes — the same set as the WRPRC +# lane (openvc.rp_registration): a V1.1.1-era JAdES producer listed its +# non-registered parameters in ``crit``, so the check allow-lists exactly what +# is understood and fails closed on the rest. +_KNOWN_CRIT = frozenset({"alg", "typ", "x5c", "iat"}) + + +class LoteType: + """The EU ``LoTEType`` URIs of TS 119 602 clause C.2.1.""" + EU_PID_PROVIDERS = f"{_URI_19602}/LoTEType/EUPIDProvidersList" + EU_WALLET_PROVIDERS = f"{_URI_19602}/LoTEType/EUWalletProvidersList" + EU_WRPAC_PROVIDERS = f"{_URI_19602}/LoTEType/EUWRPACProvidersList" + EU_WRPRC_PROVIDERS = f"{_URI_19602}/LoTEType/EUWRPRCProvidersList" + EU_PUB_EAA_PROVIDERS = f"{_URI_19602}/LoTEType/EUPubEAAProvidersList" + EU_REGISTRARS_AND_REGISTERS = f"{_URI_19602}/LoTEType/EURegistrarsAndRegistersList" + + +class LoteServiceType: + """The ``ServiceTypeIdentifier`` URIs of the WRPAC / WRPRC providers-list + profiles (TS 119 602 Tables F.3 / G.3) — each profile uses its pair "to the + exclusion of any other".""" + WRPAC_ISSUANCE = f"{_URI_19602}/SvcType/WRPAC/Issuance" + WRPAC_REVOCATION = f"{_URI_19602}/SvcType/WRPAC/Revocation" + WRPRC_ISSUANCE = f"{_URI_19602}/SvcType/WRPRC/Issuance" + WRPRC_REVOCATION = f"{_URI_19602}/SvcType/WRPRC/Revocation" + + +@dataclass(frozen=True) +class LoteProfile: + """A LoTE profile (TS 119 602 clause 4.7): scheme-defined constraints a + specific list must satisfy on top of the general data model. Checking a list + against a profile is a conformance gate — every mismatch fails closed as + :class:`~openvc.trustlist.errors.TrustListProfileError`. + + ``service_types`` is the profile's *exclusive* set (every service in the + list must use one of them); ``anchor_service_types`` is the least-privilege + subset a profiled :func:`walk_lote` keeps by default — the **issuance** + services. Under Annex F/G both an issuance and a revocation service become + list entries, but only the issuance certificates should anchor credential + verification (the adversarial review's M1: without the split, a registrar's + *revocation*-service key would validate WRPRC chains).""" + name: str + lote_type: str # required LoTEType (Table x.1) + status_determination: tuple[str, ...] # accepted StatusDeterminationApproach spellings + scheme_rules: str # required SchemeTypeCommunityRules URI + territory: str # required SchemeTerritory + service_types: frozenset[str] # the exclusive ServiceTypeIdentifier set + anchor_service_types: frozenset[str] # the default anchors a profiled walk keeps + max_update_months: int = 6 # NextUpdate - ListIssueDateTime ceiling + + +EU_WRPAC_PROVIDERS_PROFILE = LoteProfile( + name="EU WRPAC providers list (TS 119 602 Annex F)", + lote_type=LoteType.EU_WRPAC_PROVIDERS, + status_determination=(f"{_URI_19602}/WRPACProvidersList/StatusDetn/EU",), + scheme_rules=f"{_URI_19602}/WRPACProvidersList/schemerules/EU", + territory="EU", + service_types=frozenset( + {LoteServiceType.WRPAC_ISSUANCE, LoteServiceType.WRPAC_REVOCATION}), + anchor_service_types=frozenset({LoteServiceType.WRPAC_ISSUANCE}), +) + +EU_WRPRC_PROVIDERS_PROFILE = LoteProfile( + name="EU WRPRC providers list (TS 119 602 Annex G)", + lote_type=LoteType.EU_WRPRC_PROVIDERS, + status_determination=( + # The literal Table G.1 / C.2.2 value — "WRPRCroviders" is a spec typo + # the EUDI reference implementation reproduces verbatim… + f"{_URI_19602}/WRPRCrovidersList/StatusDetn/EU", + # …and the corrected spelling, so a future erratum keeps verifying. + f"{_URI_19602}/WRPRCProvidersList/StatusDetn/EU", + ), + scheme_rules=f"{_URI_19602}/WRPRCProvidersList/schemerules/EU", + territory="EU", + service_types=frozenset( + {LoteServiceType.WRPRC_ISSUANCE, LoteServiceType.WRPRC_REVOCATION}), + anchor_service_types=frozenset({LoteServiceType.WRPRC_ISSUANCE}), +) + + +def default_lote_fetch(url: str) -> bytes: + """The blessed SSRF-guarded LoTE fetch: :func:`openvc.fetch.https_bytes_fetch` + with the trust-list byte cap.""" + from ..fetch import https_bytes_fetch + return https_bytes_fetch(url, max_bytes=DEFAULT_MAX_BYTES) + + +# --------------------------------------------------------------------------- # +# strict structural parsing (runs on attacker-influenced bytes) +# --------------------------------------------------------------------------- # + +def _require_mapping(value: Any, ctx: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise TrustListParseError(f"{ctx} must be a JSON object") + return value + + +def _require_list(value: Any, ctx: str) -> list[Any]: + if not isinstance(value, list): + raise TrustListParseError(f"{ctx} must be a JSON array") + return value + + +def _require_str(value: Any, ctx: str) -> str: + if not isinstance(value, str): + raise TrustListParseError(f"{ctx} must be a string") + return value + + +def _require_int(value: Any, ctx: str) -> int: + # bool is an int subclass — `true` is never a sequence number. + if not isinstance(value, int) or isinstance(value, bool): + raise TrustListParseError(f"{ctx} must be an integer") + return value + + +def _check_keys( + obj: Mapping[str, Any], *, allowed: frozenset[str], required: frozenset[str], ctx: str, +) -> None: + """Enforce the official schema's ``additionalProperties: false`` + required + members — an unknown structural member could carry semantics this verifier + does not understand, so it rejects rather than ignores.""" + unknown = [k for k in obj if k not in allowed] + if unknown: + raise TrustListParseError(f"{ctx} carries unknown member(s) {sorted(unknown)!r}") + missing = [k for k in required if k not in obj] + if missing: + raise TrustListParseError(f"{ctx} is missing required member(s) {sorted(missing)!r}") + + +def _ml_strings(value: Any, ctx: str) -> list[str]: + """A non-empty multilingual character-string sequence → its values, English + first (clause 6.1.4).""" + items = _require_list(value, ctx) + if not items: + raise TrustListParseError(f"{ctx} must not be empty") + english: list[str] = [] + other: list[str] = [] + for i, item in enumerate(items): + entry = _require_mapping(item, f"{ctx}[{i}]") + _check_keys(entry, allowed=frozenset({"lang", "value"}), + required=frozenset({"lang", "value"}), ctx=f"{ctx}[{i}]") + lang = _require_str(entry["lang"], f"{ctx}[{i}].lang") + text = _require_str(entry["value"], f"{ctx}[{i}].value") + (english if lang.lower() == "en" else other).append(text) + return english + other + + +def _ml_uris(value: Any, ctx: str) -> list[str]: + """A non-empty multilingual-pointer sequence → its ``uriValue`` strings.""" + items = _require_list(value, ctx) + if not items: + raise TrustListParseError(f"{ctx} must not be empty") + uris: list[str] = [] + for i, item in enumerate(items): + entry = _require_mapping(item, f"{ctx}[{i}]") + _check_keys(entry, allowed=frozenset({"lang", "uriValue"}), + required=frozenset({"lang", "uriValue"}), ctx=f"{ctx}[{i}]") + _require_str(entry["lang"], f"{ctx}[{i}].lang") + uris.append(_require_str(entry["uriValue"], f"{ctx}[{i}].uriValue")) + return uris + + +_DATETIME_Z = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z") + + +def _datetime_z(value: Any, ctx: str) -> datetime: + """Clause 6.1.3: ISO 8601, UTC, second precision — exactly + ``YYYY-MM-DDThh:mm:ssZ``, no decimal fraction, no alternate separators.""" + text = _require_str(value, ctx) + if not _DATETIME_Z.fullmatch(text): + raise TrustListParseError( + f"{ctx} must be a UTC date-time of the exact YYYY-MM-DDThh:mm:ssZ " + f"form (clause 6.1.3), got {text!r}") + try: + parsed = datetime.fromisoformat(text[:-1] + "+00:00") + except ValueError as exc: + raise TrustListParseError(f"{ctx} is not a valid date-time: {text!r}") from exc + return parsed + + +def _reject_critical_extensions(value: Any, ctx: str) -> None: + """Clause 6.3.17 / 6.6.9: extension formats are left open, but a critical + extension this verifier does not recognise **shall** reject the list. The + open format means criticality can only be detected best-effort — any member + whose common criticality spellings are ``true`` rejects; everything else is + carried opaquely.""" + items = _require_list(value, ctx) + if not items: + raise TrustListParseError(f"{ctx} must not be empty") + for i, item in enumerate(items): + if isinstance(item, Mapping) and any( + item.get(k) is True for k in ("critical", "Critical")): + raise TrustListParseError( + f"{ctx}[{i}] is a critical extension this verifier does not recognise " + f"(clause 6.3.17: a LoTE with an unrecognised critical extension is rejected)") + + +_PKIOB_KEYS = frozenset({"encoding", "specRef", "val"}) + + +def _certs_from_sdi(value: Any, ctx: str) -> list[Any]: + """One ``ServiceDigitalIdentity`` → its loaded ``X509Certificates``. A blob + that does not parse as DER is skipped (a malformed certificate never becomes + a silent anchor); structural violations reject.""" + sdi = _require_mapping(value, ctx) + _check_keys( + sdi, + allowed=frozenset( + {"X509Certificates", "X509SubjectNames", "PublicKeyValues", "X509SKIs", "OtherIds"}), + required=frozenset(), ctx=ctx) + from cryptography import x509 + certs: list[Any] = [] + if "X509Certificates" in sdi: + entries = _require_list(sdi["X509Certificates"], f"{ctx}.X509Certificates") + if not entries: + raise TrustListParseError(f"{ctx}.X509Certificates must not be empty") + for i, entry in enumerate(entries): + ob = _require_mapping(entry, f"{ctx}.X509Certificates[{i}]") + _check_keys(ob, allowed=_PKIOB_KEYS, required=frozenset({"val"}), + ctx=f"{ctx}.X509Certificates[{i}]") + val = _require_str(ob["val"], f"{ctx}.X509Certificates[{i}].val") + try: + certs.append(x509.load_der_x509_certificate(base64.b64decode(val, validate=True))) + except Exception: # a malformed cert is dropped, not trusted + continue + return certs + + +_QUALIFIER_KEYS = frozenset( + {"LoTEType", "SchemeOperatorName", "SchemeTypeCommunityRules", "SchemeTerritory", "MimeType"}) + + +def _parse_pointer(value: Any, ctx: str) -> TslPointer: + ptr = _require_mapping(value, ctx) + _pointer_keys = frozenset({"LoTELocation", "ServiceDigitalIdentities", "LoTEQualifiers"}) + _check_keys(ptr, allowed=_pointer_keys, required=_pointer_keys, ctx=ctx) + location = _require_str(ptr["LoTELocation"], f"{ctx}.LoTELocation") + sdis = _require_list(ptr["ServiceDigitalIdentities"], f"{ctx}.ServiceDigitalIdentities") + if not sdis: + raise TrustListParseError(f"{ctx}.ServiceDigitalIdentities must not be empty") + certs: list[Any] = [] + for i, sdi in enumerate(sdis): + certs.extend(_certs_from_sdi(sdi, f"{ctx}.ServiceDigitalIdentities[{i}]")) + qualifiers = _require_list(ptr["LoTEQualifiers"], f"{ctx}.LoTEQualifiers") + if not qualifiers: + raise TrustListParseError(f"{ctx}.LoTEQualifiers must not be empty") + territory = None + lote_type = None + mime_type = None + for i, q in enumerate(qualifiers): + qual = _require_mapping(q, f"{ctx}.LoTEQualifiers[{i}]") + _check_keys(qual, allowed=_QUALIFIER_KEYS, + required=frozenset({"LoTEType", "SchemeOperatorName", "MimeType"}), + ctx=f"{ctx}.LoTEQualifiers[{i}]") + if lote_type is None: + lote_type = _require_str(qual["LoTEType"], f"{ctx}.LoTEQualifiers[{i}].LoTEType") + if mime_type is None: + mime_type = _require_str(qual["MimeType"], f"{ctx}.LoTEQualifiers[{i}].MimeType") + if territory is None and "SchemeTerritory" in qual: + territory = _require_str( + qual["SchemeTerritory"], f"{ctx}.LoTEQualifiers[{i}].SchemeTerritory") + return TslPointer( + location=location, signer_certs=tuple(certs), + territory=territory, tsl_type=lote_type, mime_type=mime_type) + + +_SCHEME_KEYS = frozenset({ + "LoTEVersionIdentifier", "LoTESequenceNumber", "LoTEType", "SchemeOperatorName", + "SchemeOperatorAddress", "SchemeName", "SchemeInformationURI", "StatusDeterminationApproach", + "SchemeTypeCommunityRules", "SchemeTerritory", "PolicyOrLegalNotice", + "HistoricalInformationPeriod", "PointersToOtherLoTE", "ListIssueDateTime", "NextUpdate", + "DistributionPoints", "SchemeExtensions", +}) +_SCHEME_REQUIRED = frozenset({ + "LoTEVersionIdentifier", "LoTESequenceNumber", "SchemeOperatorName", + "ListIssueDateTime", "NextUpdate", +}) +_SERVICE_INFO_KEYS = frozenset({ + "ServiceName", "ServiceDigitalIdentity", "ServiceTypeIdentifier", "ServiceStatus", + "StatusStartingTime", "SchemeServiceDefinitionURI", "ServiceSupplyPoints", + "ServiceDefinitionURI", "ServiceInformationExtensions", +}) +_TE_INFO_KEYS = frozenset({ + "TEName", "TETradeName", "TEAddress", "TEInformationURI", "TEInformationExtensions", +}) + + +def parse_lote( + payload: Mapping[str, Any] | bytes, *, max_bytes: int = DEFAULT_MAX_BYTES, +) -> TrustList: + """Parse a TS 119 602 JSON LoTE document into a :class:`TrustList` — + strictly, fail-closed, WITHOUT any signature verification (use + :func:`consume_lote`, which verifies first). + + Accepts the decoded top-level object (``{"LoTE": …}``) or raw JSON bytes. + Raises :class:`TrustListParseError` on any structural violation.""" + if isinstance(payload, (bytes, bytearray)): + if len(payload) > max_bytes: + raise TrustListParseError( + f"LoTE is {len(payload)} bytes, over the {max_bytes}-byte cap") + try: + decoded = json.loads(payload) + except (ValueError, RecursionError) as exc: + raise TrustListParseError(f"LoTE is not valid JSON: {exc}") from exc + else: + decoded = payload + root = _require_mapping(decoded, "LoTE document") + _check_keys(root, allowed=frozenset({"LoTE"}), required=frozenset({"LoTE"}), + ctx="LoTE document") + lote = _require_mapping(root["LoTE"], "LoTE") + _check_keys(lote, allowed=frozenset({"ListAndSchemeInformation", "TrustedEntitiesList"}), + required=frozenset({"ListAndSchemeInformation"}), ctx="LoTE") + + scheme = _require_mapping(lote["ListAndSchemeInformation"], "ListAndSchemeInformation") + _check_keys(scheme, allowed=_SCHEME_KEYS, required=_SCHEME_REQUIRED, + ctx="ListAndSchemeInformation") + + version = _require_int(scheme["LoTEVersionIdentifier"], "LoTEVersionIdentifier") + sequence = _require_int(scheme["LoTESequenceNumber"], "LoTESequenceNumber") + operator_names = _ml_strings(scheme["SchemeOperatorName"], "SchemeOperatorName") + lote_type = (_require_str(scheme["LoTEType"], "LoTEType") + if "LoTEType" in scheme else None) + territory = (_require_str(scheme["SchemeTerritory"], "SchemeTerritory") + if "SchemeTerritory" in scheme else None) + if "HistoricalInformationPeriod" in scheme: + _require_int(scheme["HistoricalInformationPeriod"], "HistoricalInformationPeriod") + if "StatusDeterminationApproach" in scheme: + _require_str(scheme["StatusDeterminationApproach"], "StatusDeterminationApproach") + if "SchemeTypeCommunityRules" in scheme: + _ml_uris(scheme["SchemeTypeCommunityRules"], "SchemeTypeCommunityRules") + if "SchemeName" in scheme: + _ml_strings(scheme["SchemeName"], "SchemeName") + if "SchemeInformationURI" in scheme: + _ml_uris(scheme["SchemeInformationURI"], "SchemeInformationURI") + if "SchemeOperatorAddress" in scheme: + _require_mapping(scheme["SchemeOperatorAddress"], "SchemeOperatorAddress") + if "PolicyOrLegalNotice" in scheme: + _require_list(scheme["PolicyOrLegalNotice"], "PolicyOrLegalNotice") + if "DistributionPoints" in scheme: + points = _require_list(scheme["DistributionPoints"], "DistributionPoints") + if not points: + raise TrustListParseError("DistributionPoints must not be empty") + for i, p in enumerate(points): + _require_str(p, f"DistributionPoints[{i}]") + if "SchemeExtensions" in scheme: + _reject_critical_extensions(scheme["SchemeExtensions"], "SchemeExtensions") + + issue = _datetime_z(scheme["ListIssueDateTime"], "ListIssueDateTime") + # Clause 6.3.15: NextUpdate is null for a **closed** LoTE (scheme ceased); + # a closed list contributes no anchors — the walk stages it as expired. + next_update = (None if scheme["NextUpdate"] is None + else _datetime_z(scheme["NextUpdate"], "NextUpdate")) + + pointers: list[TslPointer] = [] + if "PointersToOtherLoTE" in scheme: + raw_ptrs = _require_list(scheme["PointersToOtherLoTE"], "PointersToOtherLoTE") + if not raw_ptrs: + raise TrustListParseError("PointersToOtherLoTE must not be empty") + for i, p in enumerate(raw_ptrs): + pointers.append(_parse_pointer(p, f"PointersToOtherLoTE[{i}]")) + + providers: list[TrustServiceProvider] = [] + if "TrustedEntitiesList" in lote: + entities = _require_list(lote["TrustedEntitiesList"], "TrustedEntitiesList") + if not entities: + raise TrustListParseError("TrustedEntitiesList must not be empty") + for i, e in enumerate(entities): + providers.append(_parse_entity(e, f"TrustedEntitiesList[{i}]", territory)) + + return TrustList( + tsl_type=lote_type, scheme_operator=operator_names[0], territory=territory, + sequence_number=sequence, issue_datetime=issue.isoformat(), next_update=next_update, + pointers=tuple(pointers), providers=tuple(providers), version=version) + + +def _parse_entity(value: Any, ctx: str, territory: str | None) -> TrustServiceProvider: + entity = _require_mapping(value, ctx) + _check_keys(entity, allowed=frozenset({"TrustedEntityInformation", "TrustedEntityServices"}), + required=frozenset({"TrustedEntityInformation", "TrustedEntityServices"}), ctx=ctx) + info = _require_mapping(entity["TrustedEntityInformation"], f"{ctx}.TrustedEntityInformation") + _check_keys(info, allowed=_TE_INFO_KEYS, + required=frozenset({"TEName", "TEAddress", "TEInformationURI"}), + ctx=f"{ctx}.TrustedEntityInformation") + names = _ml_strings(info["TEName"], f"{ctx}.TrustedEntityInformation.TEName") + _require_mapping(info["TEAddress"], f"{ctx}.TrustedEntityInformation.TEAddress") + _ml_uris(info["TEInformationURI"], f"{ctx}.TrustedEntityInformation.TEInformationURI") + if "TEInformationExtensions" in info: + _reject_critical_extensions( + info["TEInformationExtensions"], + f"{ctx}.TrustedEntityInformation.TEInformationExtensions") + + services: list[TrustServiceAnchor] = [] + raw_services = _require_list(entity["TrustedEntityServices"], f"{ctx}.TrustedEntityServices") + if not raw_services: + raise TrustListParseError(f"{ctx}.TrustedEntityServices must not be empty") + for i, s in enumerate(raw_services): + services.extend(_parse_service(s, f"{ctx}.TrustedEntityServices[{i}]", names[0], territory)) + return TrustServiceProvider(name=names[0], services=tuple(services)) + + +def _parse_service( + value: Any, ctx: str, te_name: str, territory: str | None, +) -> list[TrustServiceAnchor]: + svc = _require_mapping(value, ctx) + _check_keys(svc, allowed=frozenset({"ServiceInformation", "ServiceHistory"}), + required=frozenset({"ServiceInformation"}), ctx=ctx) + info = _require_mapping(svc["ServiceInformation"], f"{ctx}.ServiceInformation") + _check_keys(info, allowed=_SERVICE_INFO_KEYS, + required=frozenset({"ServiceName", "ServiceDigitalIdentity"}), + ctx=f"{ctx}.ServiceInformation") + names = _ml_strings(info["ServiceName"], f"{ctx}.ServiceInformation.ServiceName") + service_type = "" + if "ServiceTypeIdentifier" in info: + service_type = _require_str( + info["ServiceTypeIdentifier"], f"{ctx}.ServiceInformation.ServiceTypeIdentifier") + service_status = "" + if "ServiceStatus" in info: + service_status = _require_str( + info["ServiceStatus"], f"{ctx}.ServiceInformation.ServiceStatus") + if "StatusStartingTime" in info: + _datetime_z(info["StatusStartingTime"], f"{ctx}.ServiceInformation.StatusStartingTime") + if "ServiceInformationExtensions" in info: + _reject_critical_extensions( + info["ServiceInformationExtensions"], + f"{ctx}.ServiceInformation.ServiceInformationExtensions") + certs = _certs_from_sdi( + info["ServiceDigitalIdentity"], f"{ctx}.ServiceInformation.ServiceDigitalIdentity") + return [ + TrustServiceAnchor( + certificate=cert, service_type=service_type, service_status=service_status, + tsp_name=te_name, service_name=names[0], territory=territory) + for cert in certs + ] + + +# --------------------------------------------------------------------------- # +# signed consumption (JAdES B-B compact, clause 6.8) +# --------------------------------------------------------------------------- # + +def consume_lote( + token: str | bytes, *, + expected_signer_certs: Sequence[Any], + profile: LoteProfile | None = None, + now: datetime | None = None, + max_bytes: int = DEFAULT_MAX_BYTES, +) -> TrustList: + """Verify a signed LoTE (compact JAdES baseline B) **then** parse it. + + In order: the compact-JWS envelope (the ``{ES256, ES384, EdDSA, Ed25519}`` + allow-list applied before any crypto; a fail-closed allow-listed ``crit``; + ``x5c`` required); the signer authenticated against *expected_signer_certs* + — the leaf matching a pinned certificate byte-for-byte (within that + certificate's own validity window), or path-validating to the pinned set as + anchors; the **signature** against the leaf key; the strict payload parse; + clause 6.8's DN binding (signing-certificate ``organizationName`` must be a + ``SchemeOperatorName`` value, ``countryName`` the ``SchemeTerritory`` when + the list carries one); and, when *profile* is given, the profile's + conformance gate (:class:`TrustListProfileError` on any mismatch). + + There is no implicit trust root: *expected_signer_certs* are + ``cryptography`` ``x509.Certificate`` objects the caller pins (or, on a + pointer walk, the certificates the pointing list vouched for). + + ``consume_lote`` establishes authenticity and conformance — **not + freshness**: staging an expired or **closed** list (``NextUpdate`` in the + past, or null per clause 6.3.15) is :func:`walk_lote`'s job, mirroring the + 119 612 lane's ``consume_trust_list``/``walk_lotl`` split. A caller using + ``consume_lote`` directly must check ``next_update`` itself.""" + if isinstance(token, (bytes, bytearray)): + try: + token = bytes(token).decode("ascii") + except UnicodeDecodeError as exc: + raise TrustListParseError("a compact JAdES LoTE must be ASCII") from exc + if not isinstance(token, str): + raise TrustListParseError("LoTE token must be a compact-JWS string or bytes") + if len(token) > max_bytes: + raise TrustListParseError( + f"LoTE token is {len(token)} bytes, over the {max_bytes}-byte cap") + + from ..proof._jws import parse_compact + from ..proof.errors import ProofError + from ..proof.vc_jwt import ALLOWED_ALGS + + try: + header, payload, signing_input, signature = parse_compact(token) + except ProofError as exc: + raise TrustListParseError(f"LoTE is not a valid compact JWS: {exc}") from exc + + alg = header.get("alg") + if not isinstance(alg, str) or alg not in ALLOWED_ALGS: + raise TrustListSignatureError( + f"LoTE alg {alg!r} is not permitted (need one of {sorted(ALLOWED_ALGS)})") + _reject_unknown_crit(header) + + leaf, chain = _signer_chain(header) + _authenticate_signer(leaf, chain, expected_signer_certs, now=now) + + from ..keys import KeyBackendError, verify_signature + from ..x5c import X5cError, leaf_public_jwk + try: + public_jwk = leaf_public_jwk(leaf) + except X5cError as exc: + raise TrustListSignatureError(f"LoTE signing certificate: {exc}") from exc + try: + ok = verify_signature(alg=alg, public_jwk=public_jwk, + signing_input=signing_input, signature=signature) + except KeyBackendError as exc: + raise TrustListSignatureError(f"LoTE signature could not be checked: {exc}") from exc + if not ok: + raise TrustListSignatureError("LoTE signature verification failed") + + trust_list = parse_lote(payload, max_bytes=max_bytes) + lote_obj = payload["LoTE"] + _check_signer_dn(leaf, lote_obj["ListAndSchemeInformation"], territory=trust_list.territory) + if profile is not None: + _check_profile(trust_list, lote_obj, profile) + return trust_list + + +def _reject_unknown_crit(header: Mapping[str, Any]) -> None: + """RFC 7515 §4.1.11, with the WRPRC lane's allow-list stance: this verifier + processes the JAdES parameters in :data:`_KNOWN_CRIT` and fails closed on + anything else named critical.""" + if "crit" not in header: + return + crit = header["crit"] + if not isinstance(crit, list) or not crit or not all(isinstance(c, str) for c in crit): + raise TrustListSignatureError("LoTE 'crit' must be a non-empty array of strings") + unknown = [c for c in crit if c not in _KNOWN_CRIT] + if unknown: + raise TrustListSignatureError( + f"LoTE marks header parameter(s) {unknown!r} critical, which this verifier " + f"does not process (understood: {sorted(_KNOWN_CRIT)})") + missing = [c for c in crit if c not in header] + if missing: + raise TrustListSignatureError( + f"LoTE 'crit' names header parameter(s) {missing!r} that are not present") + + +def _signer_chain(header: Mapping[str, Any]) -> tuple[Any, list[Any]]: + from ..x5c import X5cError, load_x5c_chain + + x5c = header.get("x5c") + if x5c is None: + raise TrustListSignatureError( + "LoTE JWS header has no 'x5c' chain to identify the scheme operator " + "(TS 119 182-1 clause 5.1.7 requires a signing-certificate reference)") + try: + chain = load_x5c_chain(x5c) + except X5cError as exc: + raise TrustListSignatureError(f"LoTE 'x5c': {exc}") from exc + return chain[0], chain + + +def _authenticate_signer( + leaf: Any, chain: list[Any], expected: Sequence[Any], *, now: datetime | None, +) -> None: + """Authenticate the ``x5c`` leaf against the caller-pinned certificates: + byte-for-byte equality with a pinned certificate (checked inside its own + validity window), or a path validation treating the pinned set as anchors — + covering both the pointer model (the voucher pins the actual signer) and a + CA pin. No pin, no trust.""" + from cryptography import x509 + from cryptography.hazmat.primitives.serialization import Encoding + + from ..x5c import X5cError, validate_cert_chain + + pinned = [c for c in expected if isinstance(c, x509.Certificate)] + if not pinned: + raise TrustListSignatureError( + "no expected signer certificates given (a sequence of x509.Certificate " + "objects) — a LoTE is never trusted unverified") + leaf_der = leaf.public_bytes(Encoding.DER) + if any(leaf_der == c.public_bytes(Encoding.DER) for c in pinned): + instant = now if now is not None else datetime.now(timezone.utc) + if instant.tzinfo is None: + instant = instant.replace(tzinfo=timezone.utc) + if not (leaf.not_valid_before_utc <= instant <= leaf.not_valid_after_utc): + raise TrustListSignatureError( + "LoTE signing certificate matches a pinned certificate but is outside " + "its own validity window") + return + try: + validate_cert_chain(leaf, chain[1:], trust_anchors=pinned, now=now) + except X5cError as exc: + raise TrustListSignatureError( + f"LoTE signing certificate is not pinned and did not validate to the " + f"pinned set: {exc}") from exc + + +def _check_signer_dn(leaf: Any, scheme: Mapping[str, Any], *, territory: str | None) -> None: + """Clause 6.8: the signing certificate's subject ``countryName`` shall match + ``SchemeTerritory`` and its ``organizationName`` one of the + ``SchemeOperatorName`` values.""" + from cryptography.x509.oid import NameOID + + names = _ml_strings(scheme["SchemeOperatorName"], "SchemeOperatorName") + orgs = [a.value for a in leaf.subject.get_attributes_for_oid(NameOID.ORGANIZATION_NAME)] + if not orgs or not any(o in names for o in orgs): + raise TrustListSignatureError( + f"LoTE signing certificate organizationName {orgs!r} matches no " + f"SchemeOperatorName value (clause 6.8)") + if territory is not None: + countries = [a.value for a in leaf.subject.get_attributes_for_oid(NameOID.COUNTRY_NAME)] + if territory not in countries: + raise TrustListSignatureError( + f"LoTE signing certificate countryName {countries!r} does not match " + f"the SchemeTerritory {territory!r} (clause 6.8)") + + +def _plus_months(dt: datetime, months: int) -> datetime: + month = dt.month - 1 + months + year = dt.year + month // 12 + month = month % 12 + 1 + return dt.replace(year=year, month=month, + day=min(dt.day, calendar.monthrange(year, month)[1])) + + +def _check_profile( + trust_list: TrustList, lote_obj: Mapping[str, Any], profile: LoteProfile, +) -> None: + scheme = lote_obj["ListAndSchemeInformation"] + + def fail(detail: str) -> NoReturn: + raise TrustListProfileError(f"{profile.name}: {detail}") + + if trust_list.version != 1: + fail(f"LoTEVersionIdentifier must be 1, got {trust_list.version!r}") + if trust_list.tsl_type != profile.lote_type: + fail(f"LoTEType must be {profile.lote_type!r}, got {trust_list.tsl_type!r}") + detn = scheme.get("StatusDeterminationApproach") + if detn not in profile.status_determination: + fail(f"StatusDeterminationApproach must be one of " + f"{list(profile.status_determination)!r}, got {detn!r}") + rules = (_ml_uris(scheme["SchemeTypeCommunityRules"], "SchemeTypeCommunityRules") + if "SchemeTypeCommunityRules" in scheme else []) + if profile.scheme_rules not in rules: + fail(f"SchemeTypeCommunityRules must include {profile.scheme_rules!r}, got {rules!r}") + if trust_list.territory != profile.territory: + fail(f"SchemeTerritory must be {profile.territory!r}, got {trust_list.territory!r}") + if "HistoricalInformationPeriod" in scheme: + fail("HistoricalInformationPeriod shall not be present") + try: + issue = datetime.fromisoformat(trust_list.issue_datetime or "") + except ValueError: # unreachable from parse_lote, which always sets it + fail("ListIssueDateTime is missing or unparseable") + if trust_list.next_update is not None: + try: + ceiling = _plus_months(issue, profile.max_update_months) + except ValueError: # year overflow — cannot evaluate ⇒ fail closed + fail(f"ListIssueDateTime {issue.isoformat()} is too far in the future " + f"to evaluate the {profile.max_update_months}-month update window") + if trust_list.next_update > ceiling: + fail(f"NextUpdate {trust_list.next_update.isoformat()} is more than " + f"{profile.max_update_months} months after ListIssueDateTime " + f"{issue.isoformat()}") + for provider in trust_list.providers: + for svc in provider.services: + if svc.service_type not in profile.service_types: + fail(f"ServiceTypeIdentifier {svc.service_type!r} (service " + f"{svc.service_name!r} of {provider.name!r}) is outside the " + f"profile's exclusive set {sorted(profile.service_types)!r}") + # ServiceStatus / StatusStartingTime "shall not be used" (Tables F.3/G.3): + # *presence* is the violation, so both checks read the wire object — the + # typed model normalises absence and an empty string alike (the adversarial + # review's L2). ServiceHistory instances are exempt: Table G.3's "History + # information" row imposes no additional requirements, and clause 6.7 makes + # StatusStartingTime structurally required inside a history instance. + entities = lote_obj.get("TrustedEntitiesList") or [] + for entity in entities: + if not isinstance(entity, Mapping): + continue + for svc in entity.get("TrustedEntityServices") or []: + info = svc.get("ServiceInformation") if isinstance(svc, Mapping) else None + if not isinstance(info, Mapping): + continue + if "ServiceStatus" in info: + fail("ServiceStatus shall not be used") + if "StatusStartingTime" in info: + fail("StatusStartingTime shall not be used") + + +# --------------------------------------------------------------------------- # +# walk +# --------------------------------------------------------------------------- # + +# Sentinel: "derive the selection from the profile". Distinct from an explicit +# ``select=None``, which keeps every admitted anchor. +_DERIVED_SELECT = Select() + + +def walk_lote( + lote_url: str, *, + lote_signer_certs: Sequence[Any], + profile: LoteProfile | None = None, + fetch: FetchLote = default_lote_fetch, + select: Select | None = _DERIVED_SELECT, + now: datetime | None = None, + max_bytes: int = DEFAULT_MAX_BYTES, + max_lists: int = 8, +) -> TrustAnchorSet: + """Fetch, verify and distil the LoTE at *lote_url* (plus one hop of pointed + lists) into a :class:`TrustAnchorSet`. + + Trust is rooted in *lote_signer_certs* (caller-pinned — for the EU lists, + the Commission's published list-signing certificates). The root list is + verified with :func:`consume_lote` under *profile*. Pointed lists follow + the clause 6.3.13 vouching model (one hop, like + :func:`~openvc.trustlist.walk_lotl`): each is verified against the + certificates its pointer vouched for — and, in a **profiled** walk, a + pointer is only followed when its qualifier ``LoTEType`` matches the + profile, and the pointed list must conform to the **same** profile, so a + foreign list type can never leak anchors into a profiled walk. A list that + cannot be fetched, verified, parsed, is expired, or is **closed** + (``NextUpdate`` null, clause 6.3.15) contributes zero anchors and is + recorded in ``problems`` — never silently trusted, never aborting the walk. + + *select* defaults to the **profile's anchor service types** — the issuance + services only (least privilege: under Annex F/G a provider's *revocation* + service is also listed, but its certificates must not anchor credential + verification). With no profile, the default keeps everything (the EU + profiles forbid ``ServiceStatus``, so the 119 612 lane's granted-status + default would drop every anchor). Pass an explicit :class:`Select` to + filter differently, or ``select=None`` for every admitted anchor. + ``max_lists`` caps the total lists consumed (root + pointed).""" + if select is _DERIVED_SELECT: + select = (Select(service_types=profile.anchor_service_types) + if profile is not None else None) + instant = now if now is not None else datetime.now(timezone.utc) + if instant.tzinfo is None: + instant = instant.replace(tzinfo=timezone.utc) + problems: list[TrustListProblem] = [] + + try: + root_bytes = fetch(lote_url) + except Exception as exc: # root unreachable -> no anchors at all + return TrustAnchorSet( + anchors=(), problems=(TrustListProblem(lote_url, "fetch", str(exc)),)) + try: + root = consume_lote( + root_bytes if isinstance(root_bytes, str) else bytes(root_bytes), + expected_signer_certs=lote_signer_certs, profile=profile, + now=instant, max_bytes=max_bytes) + except TrustListError as exc: + return TrustAnchorSet( + anchors=(), problems=(TrustListProblem(lote_url, _stage(exc), str(exc)),)) + stale = _staleness(root, instant) + if stale: + return TrustAnchorSet( + anchors=(), problems=(TrustListProblem(lote_url, "expired", stale),)) + + anchors: list[TrustServiceAnchor] = [] + for provider in root.providers: + for svc in provider.services: + if select is None or select.matches(svc): + anchors.append(svc) + + visited = {lote_url} + consumed = 1 + for pointer in root.pointers: + if pointer.location in visited: + continue # the EU profiles' self-pointer, or a repeat + if profile is not None and pointer.tsl_type != profile.lote_type: + problems.append(TrustListProblem( + pointer.location, "profile", + f"not followed: pointer LoTEType {pointer.tsl_type!r} is outside " + f"this profiled walk ({profile.lote_type!r})")) + continue + if (select is not None and select.territories is not None + and (pointer.territory or "") not in select.territories): + continue + if consumed >= max_lists: + problems.append(TrustListProblem( + pointer.location, "consume", + f"not followed: the {max_lists}-list cap was reached")) + continue + visited.add(pointer.location) + consumed += 1 + try: + pointed_bytes = fetch(pointer.location) + except Exception as exc: + problems.append(TrustListProblem(pointer.location, "fetch", str(exc))) + continue + try: + pointed = consume_lote( + pointed_bytes if isinstance(pointed_bytes, str) else bytes(pointed_bytes), + expected_signer_certs=pointer.signer_certs, profile=profile, + now=instant, max_bytes=max_bytes) + except TrustListError as exc: + problems.append(TrustListProblem(pointer.location, _stage(exc), str(exc))) + continue + stale = _staleness(pointed, instant) + if stale: + problems.append(TrustListProblem(pointer.location, "expired", stale)) + continue + for provider in pointed.providers: + for svc in provider.services: + if select is None or select.matches(svc): + anchors.append(svc) + + return TrustAnchorSet(anchors=tuple(anchors), problems=tuple(problems)) + + +def _stage(exc: TrustListError) -> str: + if isinstance(exc, TrustListProfileError): + return "profile" + if isinstance(exc, TrustListSignatureError): + return "signature" + if isinstance(exc, TrustListParseError): + return "parse" + return "consume" + + +def _staleness(trust_list: TrustList, now: datetime) -> str | None: + if trust_list.next_update is None: + return "the LoTE is closed (NextUpdate null, clause 6.3.15) — its scheme ceased" + if trust_list.next_update < now: + return f"NextUpdate {trust_list.next_update.isoformat()} is in the past" + return None + + +__all__ = [ + "EU_WRPAC_PROVIDERS_PROFILE", + "EU_WRPRC_PROVIDERS_PROFILE", + "FetchLote", + "LoteProfile", + "LoteServiceType", + "LoteType", + "consume_lote", + "default_lote_fetch", + "parse_lote", + "walk_lote", +] diff --git a/tests/test_trustlist_lote.py b/tests/test_trustlist_lote.py new file mode 100644 index 0000000..5717bc7 --- /dev/null +++ b/tests/test_trustlist_lote.py @@ -0,0 +1,769 @@ +"""ETSI TS 119 602 LoTE lane — the JSON trusted-list codec, its JAdES-compact +verification, the EU WRPAC/WRPRC providers-list profiles, and the walk. + +Self-made signed vectors (no third-party LoTE artifact exists yet — the +Commission's lists are unpublished; the real-list goldens are gated on their +publication, tracked in the issue). The URIs are pinned byte-for-byte against +TS 119 602 V1.1.1 Annex C / Tables F.1–G.3 — including the ``WRPRCroviders`` +StatusDetn typo the spec (and the EUDI reference implementation) carry — so an +erratum shows up as a test failure, not silent drift. +""" +from __future__ import annotations + +import base64 +import copy +import datetime as dt +import json + +import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509.oid import NameOID, ObjectIdentifier + +from openvc.keys import P256SigningKey +from openvc.trustlist import ( + EU_WRPAC_PROVIDERS_PROFILE, + EU_WRPRC_PROVIDERS_PROFILE, + LoteServiceType, + LoteType, + Select, + TrustListParseError, + TrustListProfileError, + TrustListSignatureError, + consume_lote, + parse_lote, + walk_lote, +) + +_NOW = dt.datetime(2026, 7, 22, tzinfo=dt.timezone.utc) +_URI = "http://uri.etsi.org/19602" + + +# --------------------------------------------------------------------------- # +# builders — an operator (list signer), a registrar CA (the anchor), documents +# --------------------------------------------------------------------------- # + +def _cert(subject, issuer, subject_key, signer_key, exts, *, before=None, after=None): + b = (x509.CertificateBuilder().subject_name(subject).issuer_name(issuer) + .public_key(subject_key.public_key()).serial_number(x509.random_serial_number()) + .not_valid_before(before or _NOW - dt.timedelta(days=30)) + .not_valid_after(after or _NOW + dt.timedelta(days=3650))) + for ext, crit in exts: + b = b.add_extension(ext, crit) + return b.sign(signer_key, hashes.SHA256()) + + +def _operator(org="European Commission", country="EU", *, key=None, + before=None, after=None): + """A scheme-operator list-signing certificate whose DN satisfies clause 6.8.""" + key = key or ec.generate_private_key(ec.SECP256R1()) + name = x509.Name([ + x509.NameAttribute(NameOID.COUNTRY_NAME, country), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, org), + x509.NameAttribute(NameOID.COMMON_NAME, "LoTE signer")]) + cert = _cert(name, name, key, key, + [(x509.BasicConstraints(False, None), True)], + before=before, after=after) + return cert, key + + +def _ca(cn="Registrar Root CA"): + key = ec.generate_private_key(ec.SECP256R1()) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)]) + cert = _cert(name, name, key, key, + [(x509.BasicConstraints(True, None), True), + (x509.KeyUsage(False, False, False, False, False, True, True, + False, False), True)]) + return cert, key, name + + +def _b64(cert) -> str: + return base64.b64encode(cert.public_bytes(serialization.Encoding.DER)).decode() + + +def _b64url(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def _doc(anchor_b64, *, lote_type=LoteType.EU_WRPRC_PROVIDERS, + svc_type=LoteServiceType.WRPRC_ISSUANCE, + status_detn=f"{_URI}/WRPRCrovidersList/StatusDetn/EU", + rules=f"{_URI}/WRPRCProvidersList/schemerules/EU", + pointers=None) -> dict: + """A profile-conformant WRPRC providers LoTE document (Tables G.1–G.3).""" + scheme = { + "LoTEVersionIdentifier": 1, + "LoTESequenceNumber": 1, + "LoTEType": lote_type, + "SchemeOperatorName": [{"lang": "en", "value": "European Commission"}], + "StatusDeterminationApproach": status_detn, + "SchemeTypeCommunityRules": [{"lang": "en", "uriValue": rules}], + "SchemeTerritory": "EU", + "ListIssueDateTime": "2026-07-01T00:00:00Z", + "NextUpdate": "2026-12-01T00:00:00Z", + } + if pointers is not None: + scheme["PointersToOtherLoTE"] = pointers + return {"LoTE": { + "ListAndSchemeInformation": scheme, + "TrustedEntitiesList": [{ + "TrustedEntityInformation": { + "TEName": [{"lang": "en", "value": "Registrar"}], + "TEAddress": { + "TEPostalAddress": [ + {"lang": "en", "StreetAddress": "Calle Uno 1", "Country": "ES"}], + "TEElectronicAddress": [ + {"lang": "en", "uriValue": "mailto:registrar@example.es"}]}, + "TEInformationURI": [ + {"lang": "en", "uriValue": "https://registrar.example.es/info"}]}, + "TrustedEntityServices": [{ + "ServiceInformation": { + "ServiceName": [{"lang": "en", "value": "WRPRC issuance"}], + "ServiceTypeIdentifier": svc_type, + "ServiceDigitalIdentity": { + "X509Certificates": [{"val": anchor_b64}]}}}]}], + }} + + +def _pointer(location, voucher_cert, *, lote_type=LoteType.EU_WRPRC_PROVIDERS, + territory=None) -> dict: + qualifier = {"LoTEType": lote_type, + "SchemeOperatorName": [{"lang": "en", "value": "European Commission"}], + "MimeType": "application/jose"} + if territory is not None: + qualifier["SchemeTerritory"] = territory + return {"LoTELocation": location, + "ServiceDigitalIdentities": [ + {"X509Certificates": [{"val": _b64(voucher_cert)}]}], + "LoTEQualifiers": [qualifier]} + + +def _sign(doc, key, cert, *, alg="ES256", header=None) -> str: + head = {"alg": alg, "x5c": [_b64(cert)]} + head.update(header or {}) + signing_input = ( + f"{_b64url(json.dumps(head, separators=(',', ':')).encode())}." + f"{_b64url(json.dumps(doc, separators=(',', ':')).encode())}") + sig = P256SigningKey(key, "lote-signer").sign(signing_input.encode()) + return f"{signing_input}.{_b64url(sig)}" + + +@pytest.fixture() +def operator(): + return _operator() + + +@pytest.fixture() +def registrar_anchor(): + """(ca_cert, ca_key, ca_name) — the registrar CA the list will anchor.""" + return _ca() + + +@pytest.fixture() +def signed(operator, registrar_anchor): + """(token, doc, op_cert) — a conformant signed WRPRC providers list.""" + op_cert, op_key = operator + doc = _doc(_b64(registrar_anchor[0])) + return _sign(doc, op_key, op_cert), doc, op_cert + + +# --------------------------------------------------------------------------- # +# envelope + signature — reject before trusting +# --------------------------------------------------------------------------- # + +def test_consume_verifies_and_parses_under_the_wrprc_profile(signed, registrar_anchor): + token, _, op_cert = signed + tl = consume_lote(token, expected_signer_certs=[op_cert], + profile=EU_WRPRC_PROVIDERS_PROFILE, now=_NOW) + assert tl.tsl_type == LoteType.EU_WRPRC_PROVIDERS + assert tl.version == 1 and tl.sequence_number == 1 + assert tl.territory == "EU" and tl.scheme_operator == "European Commission" + anchors = [s for p in tl.providers for s in p.services] + assert len(anchors) == 1 + assert anchors[0].service_type == LoteServiceType.WRPRC_ISSUANCE + assert anchors[0].certificate.public_bytes(serialization.Encoding.DER) == \ + registrar_anchor[0].public_bytes(serialization.Encoding.DER) + + +@pytest.mark.parametrize("alg", ["none", "HS256", "RS256", "ES256K"]) +def test_rejects_algorithms_outside_the_allow_list(operator, alg): + op_cert, op_key = operator + doc = _doc(_b64(op_cert)) + good = _sign(doc, op_key, op_cert) + head = {"alg": alg, "x5c": [_b64(op_cert)]} + parts = good.split(".") + forged = (f"{_b64url(json.dumps(head, separators=(',', ':')).encode())}." + f"{parts[1]}.{parts[2]}") + with pytest.raises(TrustListSignatureError, match="not permitted"): + consume_lote(forged, expected_signer_certs=[op_cert], now=_NOW) + + +def test_crit_accepts_a_processed_jades_parameter(operator, registrar_anchor): + op_cert, op_key = operator + doc = _doc(_b64(registrar_anchor[0])) + token = _sign(doc, op_key, op_cert, + header={"iat": int(_NOW.timestamp()), "crit": ["iat"]}) + tl = consume_lote(token, expected_signer_certs=[op_cert], now=_NOW) + assert tl.sequence_number == 1 + + +@pytest.mark.parametrize("crit,detail", [ + (["exp"], "does not process"), # a parameter outside the allow-list + (["iat"], "not present"), # named but absent + ([], "non-empty"), # malformed shapes + ("iat", "non-empty"), + ([7], "non-empty"), +]) +def test_crit_fails_closed(operator, crit, detail): + op_cert, op_key = operator + token = _sign(_doc(_b64(op_cert)), op_key, op_cert, header={"crit": crit}) + with pytest.raises(TrustListSignatureError, match=detail): + consume_lote(token, expected_signer_certs=[op_cert], now=_NOW) + + +def test_rejects_a_header_without_x5c(operator): + op_cert, op_key = operator + doc = _doc(_b64(op_cert)) + head = {"alg": "ES256"} + signing_input = ( + f"{_b64url(json.dumps(head, separators=(',', ':')).encode())}." + f"{_b64url(json.dumps(doc, separators=(',', ':')).encode())}") + sig = P256SigningKey(op_key, "k").sign(signing_input.encode()) + with pytest.raises(TrustListSignatureError, match="x5c"): + consume_lote(f"{signing_input}.{_b64url(sig)}", + expected_signer_certs=[op_cert], now=_NOW) + + +@pytest.mark.parametrize("segment", [1, 2]) +def test_rejects_a_tampered_token(signed, segment): + token, _, op_cert = signed + parts = token.split(".") + # flip the FIRST character of the segment — fully significant bits + flipped = ("A" if parts[segment][0] != "A" else "B") + parts[segment][1:] + parts[segment] = flipped + with pytest.raises((TrustListSignatureError, TrustListParseError)): + consume_lote(".".join(parts), expected_signer_certs=[op_cert], now=_NOW) + + +def test_rejects_an_unpinned_signer(operator): + op_cert, op_key = operator + other_cert, _ = _operator() # same DN, different key — not pinned + token = _sign(_doc(_b64(op_cert)), op_key, op_cert) + with pytest.raises(TrustListSignatureError, match="not pinned"): + consume_lote(token, expected_signer_certs=[other_cert], now=_NOW) + + +def test_no_expected_signer_certs_fails_closed(signed): + token, _, _ = signed + with pytest.raises(TrustListSignatureError, match="never trusted unverified"): + consume_lote(token, expected_signer_certs=[], now=_NOW) + + +def test_accepts_a_leaf_chaining_to_a_pinned_ca(registrar_anchor): + ca_cert, ca_key, ca_name = _ca("Operator CA") + leaf_key = ec.generate_private_key(ec.SECP256R1()) + leaf_name = x509.Name([ + x509.NameAttribute(NameOID.COUNTRY_NAME, "EU"), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "European Commission"), + x509.NameAttribute(NameOID.COMMON_NAME, "LoTE signer 2026")]) + leaf = _cert(leaf_name, ca_name, leaf_key, ca_key, + [(x509.BasicConstraints(False, None), True)]) + doc = _doc(_b64(registrar_anchor[0])) + head = {"alg": "ES256", "x5c": [_b64(leaf), _b64(ca_cert)]} + signing_input = ( + f"{_b64url(json.dumps(head, separators=(',', ':')).encode())}." + f"{_b64url(json.dumps(doc, separators=(',', ':')).encode())}") + sig = P256SigningKey(leaf_key, "k").sign(signing_input.encode()) + tl = consume_lote(f"{signing_input}.{_b64url(sig)}", + expected_signer_certs=[ca_cert], now=_NOW) + assert tl.sequence_number == 1 + + +def test_rejects_a_pinned_leaf_outside_its_validity_window(registrar_anchor): + op_cert, op_key = _operator( + before=_NOW - dt.timedelta(days=730), after=_NOW - dt.timedelta(days=365)) + token = _sign(_doc(_b64(registrar_anchor[0])), op_key, op_cert) + with pytest.raises(TrustListSignatureError, match="validity window"): + consume_lote(token, expected_signer_certs=[op_cert], now=_NOW) + + +def test_rejects_a_signer_whose_organization_matches_no_operator_name(registrar_anchor): + op_cert, op_key = _operator(org="Mallory Scheme Op") + token = _sign(_doc(_b64(registrar_anchor[0])), op_key, op_cert) + with pytest.raises(TrustListSignatureError, match="organizationName"): + consume_lote(token, expected_signer_certs=[op_cert], now=_NOW) + + +def test_rejects_a_signer_country_that_is_not_the_scheme_territory(registrar_anchor): + op_cert, op_key = _operator(country="ES") # list says SchemeTerritory EU + token = _sign(_doc(_b64(registrar_anchor[0])), op_key, op_cert) + with pytest.raises(TrustListSignatureError, match="countryName"): + consume_lote(token, expected_signer_certs=[op_cert], now=_NOW) + + +# --------------------------------------------------------------------------- # +# strict structural parsing +# --------------------------------------------------------------------------- # + +def _mutated(signed_fixture, mutate): + token, doc, op_cert = signed_fixture + doc = copy.deepcopy(doc) + mutate(doc) + return doc + + +def test_parse_rejects_unknown_members(signed, operator): + op_cert, op_key = operator + for mutate, member in [ + (lambda d: d.update({"Extra": 1}), "Extra"), + (lambda d: d["LoTE"].update({"Extra": 1}), "Extra"), + (lambda d: d["LoTE"]["ListAndSchemeInformation"].update({"TSLType": "x"}), "TSLType"), + (lambda d: d["LoTE"]["TrustedEntitiesList"][0]["TrustedEntityServices"][0] + ["ServiceInformation"].update({"ServicePreviousStatus": "x"}), + "ServicePreviousStatus"), + ]: + doc = _mutated(signed, mutate) + with pytest.raises(TrustListParseError, match="unknown member"): + parse_lote(doc) + + +def test_parse_rejects_a_bool_where_an_integer_is_required(signed): + doc = _mutated(signed, lambda d: d["LoTE"]["ListAndSchemeInformation"] + .update({"LoTEVersionIdentifier": True})) + with pytest.raises(TrustListParseError, match="integer"): + parse_lote(doc) + + +def test_parse_requires_the_utc_z_datetime_form(signed): + doc = _mutated(signed, lambda d: d["LoTE"]["ListAndSchemeInformation"] + .update({"NextUpdate": "2026-12-01T00:00:00+00:00"})) + with pytest.raises(TrustListParseError, match="clause 6.1.3"): + parse_lote(doc) + + +def test_parse_rejects_missing_required_scheme_members(signed): + doc = _mutated(signed, lambda d: d["LoTE"]["ListAndSchemeInformation"] + .pop("SchemeOperatorName")) + with pytest.raises(TrustListParseError, match="missing required"): + parse_lote(doc) + + +def test_a_malformed_certificate_blob_is_skipped_never_trusted(signed, registrar_anchor): + good = _b64(registrar_anchor[0]) + doc = _mutated(signed, lambda d: d["LoTE"]["TrustedEntitiesList"][0] + ["TrustedEntityServices"][0]["ServiceInformation"] + ["ServiceDigitalIdentity"]["X509Certificates"] + .insert(0, {"val": base64.b64encode(b"not-a-cert").decode()})) + tl = parse_lote(doc) + anchors = [s for p in tl.providers for s in p.services] + assert len(anchors) == 1 # only the good blob became an anchor + assert anchors[0].certificate.public_bytes(serialization.Encoding.DER) == \ + base64.b64decode(good) + + +def test_an_unrecognised_critical_extension_rejects_the_list(signed): + doc = _mutated(signed, lambda d: d["LoTE"]["ListAndSchemeInformation"] + .update({"SchemeExtensions": [{"critical": True, "Weird": 1}]})) + with pytest.raises(TrustListParseError, match="critical extension"): + parse_lote(doc) + doc = _mutated(signed, lambda d: d["LoTE"]["ListAndSchemeInformation"] + .update({"SchemeExtensions": [{"critical": False, "Weird": 1}]})) + parse_lote(doc) # non-critical: carried opaquely + + +@pytest.mark.parametrize("mutate", [ + lambda d: d["LoTE"].update({"TrustedEntitiesList": []}), + lambda d: (d["LoTE"]["TrustedEntitiesList"][0]["TrustedEntityServices"][0] + ["ServiceInformation"]["ServiceDigitalIdentity"] + .update({"X509Certificates": []})), + lambda d: d["LoTE"]["ListAndSchemeInformation"].update({"PointersToOtherLoTE": []}), +]) +def test_parse_rejects_empty_containers(signed, mutate): + with pytest.raises(TrustListParseError, match="empty"): + parse_lote(_mutated(signed, mutate)) + + +def test_a_closed_lote_parses_with_a_null_next_update(signed): + doc = _mutated(signed, lambda d: d["LoTE"]["ListAndSchemeInformation"] + .update({"NextUpdate": None})) + assert parse_lote(doc).next_update is None + + +# --------------------------------------------------------------------------- # +# the EU profiles (Annex F / G) +# --------------------------------------------------------------------------- # + +def _consume_mutated(operator, registrar_anchor, mutate, **doc_kw): + op_cert, op_key = operator + doc = _doc(_b64(registrar_anchor[0]), **doc_kw) + mutate(doc) + token = _sign(doc, op_key, op_cert) + return consume_lote(token, expected_signer_certs=[op_cert], + profile=EU_WRPRC_PROVIDERS_PROFILE, now=_NOW) + + +def test_profile_constants_pin_the_spec_uris_byte_for_byte(): + """The drift alarm: Table G.1 spells the StatusDetn URI ``WRPRCroviders`` + (sic) and the reference implementation follows suit — an ETSI erratum (or a + corrected republication) must surface here, not silently.""" + assert EU_WRPRC_PROVIDERS_PROFILE.status_determination == ( + "http://uri.etsi.org/19602/WRPRCrovidersList/StatusDetn/EU", + "http://uri.etsi.org/19602/WRPRCProvidersList/StatusDetn/EU", + ) + assert EU_WRPRC_PROVIDERS_PROFILE.lote_type == \ + "http://uri.etsi.org/19602/LoTEType/EUWRPRCProvidersList" + assert EU_WRPRC_PROVIDERS_PROFILE.scheme_rules == \ + "http://uri.etsi.org/19602/WRPRCProvidersList/schemerules/EU" + assert EU_WRPAC_PROVIDERS_PROFILE.status_determination == ( + "http://uri.etsi.org/19602/WRPACProvidersList/StatusDetn/EU",) + assert LoteServiceType.WRPRC_ISSUANCE == \ + "http://uri.etsi.org/19602/SvcType/WRPRC/Issuance" + assert LoteServiceType.WRPAC_REVOCATION == \ + "http://uri.etsi.org/19602/SvcType/WRPAC/Revocation" + + +def test_profile_accepts_both_status_detn_spellings(operator, registrar_anchor): + for spelling in EU_WRPRC_PROVIDERS_PROFILE.status_determination: + tl = _consume_mutated(operator, registrar_anchor, lambda d: None, + status_detn=spelling) + assert tl.version == 1 + + +@pytest.mark.parametrize("mutate,detail", [ + (lambda d: d["LoTE"]["ListAndSchemeInformation"] + .update({"LoTEVersionIdentifier": 2}), "LoTEVersionIdentifier"), + (lambda d: d["LoTE"]["ListAndSchemeInformation"] + .update({"LoTEType": LoteType.EU_WRPAC_PROVIDERS}), "LoTEType"), + (lambda d: d["LoTE"]["ListAndSchemeInformation"] + .pop("StatusDeterminationApproach"), "StatusDeterminationApproach"), + (lambda d: d["LoTE"]["ListAndSchemeInformation"] + .update({"SchemeTypeCommunityRules": [ + {"lang": "en", "uriValue": "https://elsewhere.example/rules"}]}), + "SchemeTypeCommunityRules"), + (lambda d: d["LoTE"]["ListAndSchemeInformation"] + .update({"SchemeTerritory": "ES"}), "SchemeTerritory"), + (lambda d: d["LoTE"]["ListAndSchemeInformation"] + .update({"HistoricalInformationPeriod": 65535}), "HistoricalInformationPeriod"), + (lambda d: d["LoTE"]["ListAndSchemeInformation"] + .update({"NextUpdate": "2027-02-01T00:00:00Z"}), "months"), + (lambda d: d["LoTE"]["TrustedEntitiesList"][0]["TrustedEntityServices"][0] + ["ServiceInformation"] + .update({"ServiceStatus": "http://uri.etsi.org/TrstSvc/TrustedList/Svcstatus/granted"}), + "ServiceStatus"), + (lambda d: d["LoTE"]["TrustedEntitiesList"][0]["TrustedEntityServices"][0] + ["ServiceInformation"] + .update({"StatusStartingTime": "2026-07-01T00:00:00Z"}), "StatusStartingTime"), + (lambda d: d["LoTE"]["TrustedEntitiesList"][0]["TrustedEntityServices"][0] + ["ServiceInformation"] + .update({"ServiceTypeIdentifier": LoteServiceType.WRPAC_ISSUANCE}), + "exclusive"), +]) +def test_profile_fails_closed_on_every_annex_g_violation( + operator, registrar_anchor, mutate, detail): + # note: the ES-territory mutation also breaks the 6.8 country binding, which + # fires first — both are the same fail-closed outcome + with pytest.raises((TrustListProfileError, TrustListSignatureError), match=detail): + _consume_mutated(operator, registrar_anchor, mutate) + + +def test_the_signer_dn_territory_check_runs_before_the_profile(operator, registrar_anchor): + # a list whose territory changed no longer matches the signer C=EU: 6.8 fires + with pytest.raises(TrustListSignatureError, match="SchemeTerritory"): + _consume_mutated( + operator, registrar_anchor, + lambda d: d["LoTE"]["ListAndSchemeInformation"].update({"SchemeTerritory": "ES"})) + + +# --------------------------------------------------------------------------- # +# walk — fail-closed accounting, vouched pointers, caps +# --------------------------------------------------------------------------- # + +def test_walk_collects_anchors_and_follows_a_vouched_pointer(operator, registrar_anchor): + op_cert, op_key = operator + second_cert, second_key = _operator(org="Second Operator", country="EU") + second_anchor, _, _ = _ca("Second Registrar CA") + second_doc = _doc(_b64(second_anchor)) + second_doc["LoTE"]["ListAndSchemeInformation"]["SchemeOperatorName"] = [ + {"lang": "en", "value": "Second Operator"}] + second_tok = _sign(second_doc, second_key, second_cert) + + root_doc = _doc(_b64(registrar_anchor[0]), pointers=[ + _pointer("https://root.example/self.jwt", op_cert), # self — skipped + _pointer("https://second.example/lote.jwt", second_cert)]) # vouched hop + root_tok = _sign(root_doc, op_key, op_cert) + + fetched: list[str] = [] + + def fetch(url: str) -> bytes: + fetched.append(url) + return (root_tok if url == "https://root.example/self.jwt" else second_tok).encode() + + result = walk_lote( + "https://root.example/self.jwt", lote_signer_certs=[op_cert], + profile=EU_WRPRC_PROVIDERS_PROFILE, fetch=fetch, now=_NOW) + assert not result.problems + assert len(result.certificates) == 2 # both registrars, deduplicated + assert fetched == ["https://root.example/self.jwt", "https://second.example/lote.jwt"] + + +def test_walk_rejects_a_pointed_list_signed_outside_the_vouched_certs( + operator, registrar_anchor): + op_cert, op_key = operator + voucher_cert, _ = _operator(org="Second Operator") # whom the pointer vouches + mallory_cert, mallory_key = _operator(org="Second Operator") # same DN, other key + pointed = _sign(_doc(_b64(registrar_anchor[0])), mallory_key, mallory_cert) + root_doc = _doc(_b64(registrar_anchor[0]), pointers=[ + _pointer("https://second.example/lote.jwt", voucher_cert)]) + root_tok = _sign(root_doc, op_key, op_cert) + + result = walk_lote( + "https://root.example/lote.jwt", lote_signer_certs=[op_cert], + profile=EU_WRPRC_PROVIDERS_PROFILE, now=_NOW, + fetch=lambda u: (root_tok if "root" in u else pointed).encode()) + assert len(result.anchors) == 1 # the root's own anchor survives + assert [p.stage for p in result.problems] == ["signature"] + + +def test_walk_stages_an_expired_root_as_a_problem(operator, registrar_anchor): + op_cert, op_key = operator + token = _sign(_doc(_b64(registrar_anchor[0])), op_key, op_cert) + late = dt.datetime(2027, 1, 1, tzinfo=dt.timezone.utc) + result = walk_lote("https://root.example/lote.jwt", + lote_signer_certs=[op_cert], now=late, + fetch=lambda u: token.encode()) + assert result.anchors == () and [p.stage for p in result.problems] == ["expired"] + + +def test_walk_stages_a_closed_root_as_a_problem(operator, registrar_anchor): + op_cert, op_key = operator + doc = _doc(_b64(registrar_anchor[0])) + doc["LoTE"]["ListAndSchemeInformation"]["NextUpdate"] = None + token = _sign(doc, op_key, op_cert) + result = walk_lote("https://root.example/lote.jwt", + lote_signer_certs=[op_cert], now=_NOW, + fetch=lambda u: token.encode()) + assert result.anchors == () + assert [p.stage for p in result.problems] == ["expired"] + assert "closed" in result.problems[0].detail + + +def test_walk_stages_a_fetch_failure_without_aborting(operator, registrar_anchor): + op_cert, op_key = operator + voucher_cert, _ = _operator(org="Second Operator") + root_doc = _doc(_b64(registrar_anchor[0]), pointers=[ + _pointer("https://gone.example/lote.jwt", voucher_cert)]) + root_tok = _sign(root_doc, op_key, op_cert) + + def fetch(url: str) -> bytes: + if "gone" in url: + raise OSError("connection refused") + return root_tok.encode() + + result = walk_lote("https://root.example/lote.jwt", + lote_signer_certs=[op_cert], + profile=EU_WRPRC_PROVIDERS_PROFILE, fetch=fetch, now=_NOW) + assert len(result.anchors) == 1 + assert [p.stage for p in result.problems] == ["fetch"] + + +def test_walk_enforces_the_list_cap(operator, registrar_anchor): + op_cert, op_key = operator + voucher_cert, _ = _operator(org="Second Operator") + root_doc = _doc(_b64(registrar_anchor[0]), pointers=[ + _pointer("https://a.example/lote.jwt", voucher_cert), + _pointer("https://b.example/lote.jwt", voucher_cert)]) + root_tok = _sign(root_doc, op_key, op_cert) + result = walk_lote("https://root.example/lote.jwt", + lote_signer_certs=[op_cert], now=_NOW, max_lists=1, + fetch=lambda u: root_tok.encode()) + assert len(result.anchors) == 1 + assert {p.stage for p in result.problems} == {"consume"} + assert all("cap" in p.detail for p in result.problems) + + +def test_walk_applies_a_service_type_select(operator, registrar_anchor): + op_cert, op_key = operator + token = _sign(_doc(_b64(registrar_anchor[0])), op_key, op_cert) + result = walk_lote( + "https://root.example/lote.jwt", lote_signer_certs=[op_cert], now=_NOW, + fetch=lambda u: token.encode(), + select=Select(service_types=frozenset({LoteServiceType.WRPRC_REVOCATION}))) + assert result.anchors == () and result.problems == () + + +# --------------------------------------------------------------------------- # +# adversarial-review regressions (M1, L1, L2, I2) +# --------------------------------------------------------------------------- # + +def _two_service_doc(issuance_b64, revocation_b64): + doc = _doc(issuance_b64) + doc["LoTE"]["TrustedEntitiesList"][0]["TrustedEntityServices"].append({ + "ServiceInformation": { + "ServiceName": [{"lang": "en", "value": "WRPRC status"}], + "ServiceTypeIdentifier": LoteServiceType.WRPRC_REVOCATION, + "ServiceDigitalIdentity": {"X509Certificates": [{"val": revocation_b64}]}}}) + return doc + + +def test_default_profiled_walk_keeps_issuance_anchors_only(operator): + """M1: a provider's *revocation* service is a legitimate list entry, but its + certificates must not anchor credential verification by default.""" + op_cert, op_key = operator + iss_ca, _, _ = _ca("Issuance CA") + rev_ca, _, _ = _ca("Revocation CA") + token = _sign(_two_service_doc(_b64(iss_ca), _b64(rev_ca)), op_key, op_cert) + + result = walk_lote("https://ec.example/wrprc.jwt", lote_signer_certs=[op_cert], + profile=EU_WRPRC_PROVIDERS_PROFILE, now=_NOW, + fetch=lambda u: token.encode()) + assert [a.service_type for a in result.anchors] == [LoteServiceType.WRPRC_ISSUANCE] + + everything = walk_lote("https://ec.example/wrprc.jwt", lote_signer_certs=[op_cert], + profile=EU_WRPRC_PROVIDERS_PROFILE, now=_NOW, select=None, + fetch=lambda u: token.encode()) + assert len(everything.certificates) == 2 # the explicit escape hatch + + +def test_a_wrprc_signed_under_the_revocation_ca_is_rejected_by_the_default_flow(operator): + """M1 end to end: the documented walk → .certificates → verify flow must not + let a revocation-service key validate a WRPRC chain.""" + from openvc.rp_registration import ( + RpRegistrationError, verify_rp_registration_certificate) + + op_cert, op_key = operator + iss_ca, _, _ = _ca("Issuance CA") + rev_ca, rev_key, rev_name = _ca("Revocation CA") + leaf_key = ec.generate_private_key(ec.SECP256R1()) + leaf = _cert( + x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Revocation-side leaf")]), + rev_name, leaf_key, rev_key, + [(x509.BasicConstraints(False, None), True)]) + token = _sign(_two_service_doc(_b64(iss_ca), _b64(rev_ca)), op_key, op_cert) + anchors = walk_lote("https://ec.example/wrprc.jwt", lote_signer_certs=[op_cert], + profile=EU_WRPRC_PROVIDERS_PROFILE, now=_NOW, + fetch=lambda u: token.encode()) + + claims = {"sub": "VATES-B1", "iat": int(_NOW.timestamp()) - 60, + "entitlements": ["https://uri.etsi.org/19475/Entitlement/Service_Provider"]} + head = {"typ": "rc-wrp+jwt", "alg": "ES256", "x5c": [_b64(leaf)]} + signing_input = ( + f"{_b64url(json.dumps(head, separators=(',', ':')).encode())}." + f"{_b64url(json.dumps(claims, separators=(',', ':')).encode())}") + sig = P256SigningKey(leaf_key, "k").sign(signing_input.encode()) + wrprc = f"{signing_input}.{_b64url(sig)}" + + with pytest.raises(RpRegistrationError, match="did not validate"): + verify_rp_registration_certificate( + wrprc, trust_anchors=anchors.certificates, now=_NOW) + + +def test_a_profiled_walk_does_not_follow_foreign_type_pointers(operator, registrar_anchor): + """M1, pointer facet: a WRPAC-typed pointer cannot drag its list into a + WRPRC-profiled walk — not followed, staged as a problem, never fetched.""" + op_cert, op_key = operator + voucher_cert, _ = _operator(org="Second Operator") + root_doc = _doc(_b64(registrar_anchor[0]), pointers=[ + _pointer("https://wpac.example/lote.jwt", voucher_cert, + lote_type=LoteType.EU_WRPAC_PROVIDERS)]) + root_tok = _sign(root_doc, op_key, op_cert) + fetched: list[str] = [] + + def fetch(url: str) -> bytes: + fetched.append(url) + return root_tok.encode() + + result = walk_lote("https://root.example/lote.jwt", lote_signer_certs=[op_cert], + profile=EU_WRPRC_PROVIDERS_PROFILE, fetch=fetch, now=_NOW) + assert len(result.anchors) == 1 + assert [p.stage for p in result.problems] == ["profile"] + assert fetched == ["https://root.example/lote.jwt"] # the foreign hop never fetched + + +def test_a_far_future_issue_date_fails_closed_not_valueerror(operator, registrar_anchor): + """L1: a year-9999 issue date must land in the typed error family (and the + walk's problems), not crash with an uncaught ValueError.""" + op_cert, op_key = operator + doc = _doc(_b64(registrar_anchor[0])) + doc["LoTE"]["ListAndSchemeInformation"]["ListIssueDateTime"] = "9999-12-01T00:00:00Z" + doc["LoTE"]["ListAndSchemeInformation"]["NextUpdate"] = "9999-12-30T00:00:00Z" + token = _sign(doc, op_key, op_cert) + with pytest.raises(TrustListProfileError, match="update window"): + consume_lote(token, expected_signer_certs=[op_cert], + profile=EU_WRPRC_PROVIDERS_PROFILE, now=_NOW) + result = walk_lote("https://root.example/lote.jwt", lote_signer_certs=[op_cert], + profile=EU_WRPRC_PROVIDERS_PROFILE, now=_NOW, + fetch=lambda u: token.encode()) + assert result.anchors == () + assert [p.stage for p in result.problems] == ["profile"] + + +def test_profile_rejects_an_empty_string_service_status(operator, registrar_anchor): + """L2: ``"ServiceStatus": ""`` is *present* — presence is the violation.""" + with pytest.raises(TrustListProfileError, match="ServiceStatus"): + _consume_mutated( + operator, registrar_anchor, + lambda d: d["LoTE"]["TrustedEntitiesList"][0]["TrustedEntityServices"][0] + ["ServiceInformation"].update({"ServiceStatus": ""})) + + +@pytest.mark.parametrize("form", [ + "2026-07-01 00:00:00Z", # space separator + "2026-07-01T00:00:00.123Z", # decimal fraction + "2026-W27-1T00:00:00Z", # ISO week-date +]) +def test_datetime_fields_require_the_exact_clause_6_1_3_form(signed, form): + """I2: clause 6.1.3 mandates YYYY-MM-DDThh:mm:ssZ exactly.""" + doc = _mutated(signed, lambda d: d["LoTE"]["ListAndSchemeInformation"] + .update({"NextUpdate": form})) + with pytest.raises(TrustListParseError, match="clause 6.1.3"): + parse_lote(doc) + + +# --------------------------------------------------------------------------- # +# integration — LoTE anchors feed the WRPRC verify path +# --------------------------------------------------------------------------- # + +def test_lote_anchors_verify_a_wrprc_end_to_end(operator): + """The Annex G list anchors a registrar CA; a WRPRC signed under that CA + verifies with the walked ``.certificates`` — the whole point of the lane.""" + from openvc.rp_registration import verify_rp_registration_certificate + + op_cert, op_key = operator + ca_cert, ca_key, ca_name = _ca() + leaf_key = ec.generate_private_key(ec.SECP256R1()) + leaf = _cert( + x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Registrar Signing Key")]), + ca_name, leaf_key, ca_key, + [(x509.BasicConstraints(False, None), True), + (x509.ExtendedKeyUsage([ObjectIdentifier("0.4.0.19475.2.1")]), False)]) + + lote_token = _sign(_doc(_b64(ca_cert)), op_key, op_cert) + anchors = walk_lote("https://ec.example/wrprc-providers.jwt", + lote_signer_certs=[op_cert], + profile=EU_WRPRC_PROVIDERS_PROFILE, now=_NOW, + fetch=lambda u: lote_token.encode()) + assert not anchors.problems + + wrprc_claims = { + "sub": "VATES-B12345678", "name": "Acme Age Check", "country": "ES", + "registry_uri": "https://registrar.example/ES", + "iat": int(_NOW.timestamp()) - 3600, + "entitlements": ["https://uri.etsi.org/19475/Entitlement/Service_Provider"], + "intended_use_id": "age-verification", + "purpose": [{"lang": "en", "value": "Age check"}], + "credentials": [{"format": "dc+sd-jwt", + "meta": {"vct_values": ["urn:eudi:pid:1"]}, + "claims": [{"path": ["age_equal_or_over", "18"]}]}], + } + head = {"typ": "rc-wrp+jwt", "alg": "ES256", "x5c": [_b64(leaf)]} + signing_input = ( + f"{_b64url(json.dumps(head, separators=(',', ':')).encode())}." + f"{_b64url(json.dumps(wrprc_claims, separators=(',', ':')).encode())}") + sig = P256SigningKey(leaf_key, "registrar").sign(signing_input.encode()) + wrprc = f"{signing_input}.{_b64url(sig)}" + + reg = verify_rp_registration_certificate( + wrprc, trust_anchors=anchors.certificates, now=_NOW) + assert reg.subject_identifier == "VATES-B12345678" + assert reg.intended_use_id == "age-verification" diff --git a/wiki/Trust.md b/wiki/Trust.md index ce607d2..48dd01f 100644 --- a/wiki/Trust.md +++ b/wiki/Trust.md @@ -64,6 +64,75 @@ Properties worth knowing: Design rationale: [ADR-0003](https://github.com/luisgf/openvc/blob/main/docs/adr/ADR-0003-eu-trusted-lists.md). +## EU Lists of Trusted Entities (LoTE, TS 119 602) + +**ETSI TS 119 602** is the successor trusted-list data model: the same +information as a TS 119 612 list, serialized as **JSON** and signed as a +**compact JAdES baseline-B** JWS whose payload is the list. Its EU profiles are +the EUDI wallet anchor lists — **Annex F**: providers of relying-party *access* +certificates (WRPAC), **Annex G**: providers of relying-party *registration* +certificates (WRPRC, the registrar anchors +[`verify_rp_registration_certificate`](Relying-Party-Certificates) consumes). +One interface, two encodings — `walk_lote` distils into the same +`TrustAnchorSet` as `walk_lotl`: + + +```python +from openvc.rp_registration import verify_rp_registration_certificate +from openvc.trustlist import EU_WRPRC_PROVIDERS_PROFILE, walk_lote + +anchors = walk_lote( + WRPRC_PROVIDERS_LOTE_URL, # the Commission's Annex G list + lote_signer_certs=commission_certs, # caller-pinned: no implicit root + profile=EU_WRPRC_PROVIDERS_PROFILE, # Annex G conformance gate, fail-closed +) +reg = verify_rp_registration_certificate(wrprc, trust_anchors=anchors.certificates) +``` + +Properties worth knowing: + +- **JOSE, no extra needed** — a LoTE verifies on the library's own JWS + primitives: the `{ES256, ES384, EdDSA, Ed25519}` allow-list runs **before** + any crypto, `crit` is allow-listed (`{alg, typ, x5c, iat}`, the WRPRC lane's + JAdES stance), and the signer comes from `x5c`, authenticated against the + pinned certificates byte-for-byte or by path validation to them. +- **Clause 6.8 DN binding** — the signing certificate's `organizationName` + must match a `SchemeOperatorName` value and its `countryName` the + `SchemeTerritory`, so a certificate pinned for one scheme cannot vouch for a + list claiming another operator. +- **Strict, fail-closed parsing** — unknown structural members reject (the + official JSON schema is `additionalProperties: false` throughout), date-times + must be the UTC `Z` form, an unrecognised **critical** scheme/service + extension rejects the list (clause 6.3.17), and a certificate blob that does + not load is skipped, never silently trusted. +- **Profiles are conformance gates** — `EU_WRPAC_PROVIDERS_PROFILE` / + `EU_WRPRC_PROVIDERS_PROFILE` enforce Tables F.1–G.3: the registered + `LoTEType` / StatusDetn / schemerules URIs, territory `EU`, the exclusive + service-type pair, `ServiceStatus` / `StatusStartingTime` / + `HistoricalInformationPeriod` **absent** (under these profiles, *listing* is + the status — removal is revocation), and the ≤ 6-month update window. + Note: the WRPRC StatusDetn URI is spelled `…/WRPRCrovidersList/StatusDetn/EU` + (sic) in the spec and the EUDI reference implementation; the profile accepts + both that literal and the corrected spelling. +- **Same walk discipline, least privilege by default** — pointed lists verify + against the certificates their pointer vouched for (one hop, like LOTL→TL); + in a *profiled* walk a pointer is only followed when its qualifier `LoTEType` + matches the profile, and the pointed list must conform to the **same** + profile — a foreign list type can never leak anchors in. A list that cannot + be fetched, verified, or is expired — or is **closed** (`NextUpdate` null) — + contributes zero anchors and lands in `problems`. Under a profile, `select` + defaults to the profile's **issuance** service type only: a provider's + *revocation* service is a legitimate list entry, but its certificates must + not anchor credential verification (pass an explicit `Select`, or + `select=None` for every admitted anchor, to widen deliberately). With no + profile the default keeps everything — the EU profiles forbid + `ServiceStatus`, so the 119 612 lane's granted-status default would drop + every anchor. + +The Commission had not yet published the EU LoTE instances when this lane +shipped; conformance is pinned by self-made vectors against TS 119 602 V1.1.1, +and the real lists become golden fixtures when they go live. + ## EBSI (read-only plugin) `openvc_ebsi` resolves `did:ebsi` and reads the **Trusted Issuers Registry**,