diff --git a/.github/workflows/xchk-registry-conformance.yml b/.github/workflows/xchk-registry-conformance.yml new file mode 100644 index 0000000..c43edec --- /dev/null +++ b/.github/workflows/xchk-registry-conformance.yml @@ -0,0 +1,55 @@ +# xChk registry integration workflow. +# +# floating installs the latest released agentrust-trace and agentrust-trace-tests, unpinned on purpose, as drift detection. +# fixed installs the versions named in integration.yaml tested_against. +# Both verify the committed Trust Record with the pinned key and run the Level 0 conformance suite on it. Neither +# reaches the registry: the one live check (the published key) is opt-in with REGISTRY_LIVE=1. +name: xchk-registry conformance +on: + push: + paths: + - "integrations/xchk-registry/**" + - ".github/workflows/xchk-registry-conformance.yml" + pull_request: + paths: + - "integrations/xchk-registry/**" + - ".github/workflows/xchk-registry-conformance.yml" + schedule: + - cron: "0 6 * * 1" + workflow_dispatch: + +permissions: + contents: read + +jobs: + check: + name: ${{ matrix.mode }} (py${{ matrix.python }}) + strategy: + fail-fast: false + matrix: + python: ["3.11", "3.12", "3.13"] + mode: [floating, fixed] + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: ${{ matrix.python }} + - name: Install agentrust-trace and agentrust-trace-tests (latest release) + if: matrix.mode == 'floating' + run: pip install agentrust-trace agentrust-trace-tests + - name: Install agentrust-trace and agentrust-trace-tests (tested_against) + if: matrix.mode == 'fixed' + run: | + v=$(sed -n 's/^ agentrust-trace: "\(.*\)"$/\1/p' integrations/xchk-registry/integration.yaml) + t=$(sed -n 's/^ agentrust-trace-tests: "\(.*\)"$/\1/p' integrations/xchk-registry/integration.yaml) + test -n "$v" && test -n "$t" + pip install "agentrust-trace==$v" "agentrust-trace-tests==$t" + - name: Install this integration + run: pip install -e "integrations/xchk-registry[test]" + - name: Integration tests (offline) + run: pytest integrations/xchk-registry/tests -q + - name: Level 0 conformance on the committed Trust Record + run: trace-tests verify --record integrations/xchk-registry/examples/01-trust-record.json --level 0 --max-age 315360000 diff --git a/README.md b/README.md index a39c9b3..ac2b46b 100644 --- a/README.md +++ b/README.md @@ -93,6 +93,7 @@ TRACE only works as a standard if it is genuinely neutral. Integrations are list | [WCM Key Release to TRACE](integrations/wcm-trace/) | agentrust-io | wcm, trace | community | | [WCM Triton Repository Staging](integrations/wcm-triton/) | agentrust-io | wcm | verified | | [WCM Serving Guard for vLLM](integrations/wcm-vllm/) | agentrust-io | wcm | verified | +| [xChk registry](integrations/xchk-registry/) | xChk | trace | community | | [agentrust-codex](plugins/agentrust-codex/) | agentrust-io | agent-manifest, trace | community | | [scheduled-agents](scheduled-agents/) | agentrust-io | trace | community | diff --git a/integrations/xchk-registry/LICENSE b/integrations/xchk-registry/LICENSE new file mode 100644 index 0000000..261eeb9 --- /dev/null +++ b/integrations/xchk-registry/LICENSE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/integrations/xchk-registry/NOTICE b/integrations/xchk-registry/NOTICE new file mode 100644 index 0000000..1c02147 --- /dev/null +++ b/integrations/xchk-registry/NOTICE @@ -0,0 +1,6 @@ +xchk-registry integration: Apache-2.0, copyright (c) 2026 Sean MacGuire. + +examples/01-trust-record.json: a Trust Record minted and signed by the registry at https://in.xchk.io for plate +ENG-WNP3, run 3057 (18 October 2026), cut to one approval-outcome reference. examples/expected.json pins the +registry's seal key, as published at https://in.xchk.io/.well-known/xchk-keys.json, so the test runs offline. The +record carries the plate owner's identifiers; he chose to publish it. diff --git a/integrations/xchk-registry/README.md b/integrations/xchk-registry/README.md new file mode 100644 index 0000000..d9867c4 --- /dev/null +++ b/integrations/xchk-registry/README.md @@ -0,0 +1,73 @@ +# xChk registry: a Trust Record per run + +An xChk registry licenses AI agents under a verified person and gates each act. This integration covers one thing: +the Trust Record the registry mints for each run, and how a relying party verifies it against the key the registry +publishes. + +## What the registry produces + +A software-only TRACE v0.2 Trust Record per run, Level 0: +- `subject`: the plate as a `did:web` identifier under the registry's origin. +- `origin` absent (the gate is in the execution path, not a copy of someone else's telemetry); + `runtime.platform: software-only`; `appraisal.status: none`. +- `tool_transcript`, `policy`, `model`, `data_class`, `build_provenance` as the registry recorded them for the run. +- `references`: one `approval-outcome` entry per act a person was asked about (approved or not), with `id` and + `digest` the registry's record hash and `resolver` the registry origin. +- `cnf.jwk`: the registry's seal key, Ed25519, with `kid` the key id the registry publishes. + +Route: `GET /api/agents//runs//trace-record` with the plate owner's token. The signing key is published +at `https://in.xchk.io/.well-known/xchk-keys.json`. + +## What this code does + +``` +python xchk_trace_record.py record.json --registry https://in.xchk.io # verifies with the key that registry publishes under the record's kid +python xchk_trace_record.py record.json --jwk examples/expected.json # offline, with a pinned key +``` + +It verifies the record with released `agentrust-trace` against a pinned key, or against the key published under the +record's `kid` at `/.well-known/xchk-keys.json`, where the origin is the registry the relying party trusts, +given out of band with `--registry`. Nothing in the record supplies the origin: not its `resolver` and not its +`did:web` subject, because a record signed under a publisher of its own choosing names that publisher in both, +and agreeing with itself is not a trust anchor. The record's subject and resolver must agree with the origin given. +Without `--registry` or `--jwk` the answer is `no-key`. It never verifies against the key the record itself carries, +for the same reason. `--registry` must be an https origin and nothing more (no credentials, path, query or +fragment; an explicit port 443 is the same origin as none). When `--registry` is given beside `--jwk`, both are enforced: the +pinned key verifies the signature, and the record's subject and every `approval-outcome` resolver must be under +the origin, compared as origins. The fetch is https only, follows no redirect, reads at most 256 KB, and a key document of the wrong +shape is no key, never a crash; the same holds for a record file or pinned-key file of the wrong shape, and for any +field of the record: every input ends in a JSON verdict and an exit code. A verified result carries the verified +`profile` and the complete `accepted_profiles` set (section 3.3), the thumbprint of the key actually used, `kid` +(the key the record names) and `pinned_kid` when a pinned key named one, the key source and the trusted origin, and +lists the `approval-outcome` references as they appear; it does not resolve them. + +The record's age is bounded: `iat` may be at most `--max-age` seconds old, default 86400 (24 hours, the library's +default and section 3.2.2's), and the bound cannot be switched off. A retained record is re-verified at its own time +with `--now ` rather than by widening the bound; every verdict carries `verified_at`, the instant it +was made against, and `verified_at_overridden`, so a historical verdict is never the same JSON as a current one. The output carries `revocation` as the library +reported it; this tool passes no revocation bundle or store, so it reports `no_check_performed`, which is not an +affirming appraisal (section 3.2.3). A stale or tampered record is `invalid`, with the reason. + +## Run it + +``` +pip install -e "integrations/xchk-registry[test]" agentrust-trace-tests +pytest integrations/xchk-registry/tests -q # offline; REGISTRY_LIVE=1 adds one live key fetch +trace-tests verify --record integrations/xchk-registry/examples/01-trust-record.json --level 0 --max-age 315360000 +``` + +The example record was minted on 17 Oct 2026, so the conformance command above carries an explicit `--max-age` +(ten years, the same as the workflow), as a historical fixture must; the verifier's own default stays 24 hours. + +## What it does not claim + +- **Nothing about what a reference resolves to.** An `approval-outcome` reference is a pointer the registry is + obliged to keep resolvable (spec section 3.1.2); what the referenced receipt establishes, and how to check it, is a + separate integration, not this one. +- **No hardware.** The record is software-only and unappraised, and says so in its own fields. +- **No ordering claim.** A Trust Record is issued per execution; it does not carry a commitment that an approval was + checked before the act ran. +- The registry's server code is not public. What is public is the record route, the key document and this + example. + +Maintained by the registry operator (see `integration.yaml`). Personal research project. diff --git a/integrations/xchk-registry/examples/01-trust-record.json b/integrations/xchk-registry/examples/01-trust-record.json new file mode 100644 index 0000000..ca85098 --- /dev/null +++ b/integrations/xchk-registry/examples/01-trust-record.json @@ -0,0 +1,47 @@ +{ + "eat_profile": "tag:agentrust-io.com,2026:trace-v0.2", + "iat": 1792252491, + "subject": "did:web:in.xchk.io:plate:ENG-WNP3", + "model": { + "provider": "anthropic", + "model_id": "claude-fable-5-1" + }, + "runtime": { + "platform": "software-only", + "measurement": "sha256:c60a771a8f6e4fda60fbb2b745397c800c38da090f153f72e302147a1851746a" + }, + "policy": { + "bundle_hash": "sha256:8e1e29b1e2b1470b57fba453991d3c9ef0d0ae19ba59927c014525177ee741ae", + "enforcement_mode": "enforce" + }, + "data_class": "internal", + "tool_transcript": { + "hash": "sha256:54ea01281f1acf13dec96c509cb243c5ee87b826524044ac8c2700a63cd9f2bb", + "call_count": 101 + }, + "references": [ + { + "rel": "approval-outcome", + "id": "aa2bfd5b1d9b6307ed22eb0daa4213383beecb57d7d31ec011060ac98f7fe0b5", + "resolver": "https://in.xchk.io", + "digest": "sha256:aa2bfd5b1d9b6307ed22eb0daa4213383beecb57d7d31ec011060ac98f7fe0b5" + } + ], + "build_provenance": { + "slsa_level": 0, + "digest": "sha256:c60a771a8f6e4fda60fbb2b745397c800c38da090f153f72e302147a1851746a" + }, + "appraisal": { + "status": "none", + "verifier": "https://in.xchk.io" + }, + "cnf": { + "jwk": { + "kty": "OKP", + "crv": "Ed25519", + "x": "mn02E1XyTkaVN6DEOemU5OADMUAUjPHw67oR5KKIM3k", + "kid": "5ecb3da505414c4f" + } + }, + "signature": "Vjgt2d7DVF_KmYdb3UXuOaL4YxRO1k46Uj91QkI5vc0EC5NraQ7U7S58DuD35d8GMy70wW-I08OtkRiCKTfPAQ" +} diff --git a/integrations/xchk-registry/examples/expected.json b/integrations/xchk-registry/examples/expected.json new file mode 100644 index 0000000..4293f2f --- /dev/null +++ b/integrations/xchk-registry/examples/expected.json @@ -0,0 +1,13 @@ +{ + "trace_signer_jwk": { + "kty": "OKP", + "crv": "Ed25519", + "x": "mn02E1XyTkaVN6DEOemU5OADMUAUjPHw67oR5KKIM3k", + "kid": "5ecb3da505414c4f" + }, + "resolver": "https://in.xchk.io", + "plate": "ENG-WNP3", + "runSeq": 3057, + "generated_at": "2026-10-17T15:54:52.151Z", + "note": "The signer key is the registry's seal key, published at /.well-known/xchk-keys.json; pinned here so the test runs offline." +} \ No newline at end of file diff --git a/integrations/xchk-registry/integration.yaml b/integrations/xchk-registry/integration.yaml new file mode 100644 index 0000000..fd0b1f5 --- /dev/null +++ b/integrations/xchk-registry/integration.yaml @@ -0,0 +1,26 @@ +name: xChk registry +vendor: xChk +integrates_with: + - trace +description: >- + A software-only Level 0 Trust Record per run from an agent-licensing registry, + with one approval-outcome reference per act a person was asked about. +maintainer: + github: minitru + email: sean@maclawran.ca +repository: https://github.com/agentrust-io/integrations +homepage: https://in.xchk.io +license: Apache-2.0 +tier: community +marketplace: + category: Evidence & receipts + mark: XC + keywords: [trust-record, approval-outcome, registry, level-0] +# Role. The registry mints and signs the Trust Record: record-producer, software-only, Level 0. What an +# approval-outcome reference resolves to, and what may be concluded from it, is a separate integration. +trace_roles: + - record-producer +trace_conformance_level: 0 +tested_against: + agentrust-trace: "0.11.0" + agentrust-trace-tests: "0.6.2" diff --git a/integrations/xchk-registry/pyproject.toml b/integrations/xchk-registry/pyproject.toml new file mode 100644 index 0000000..1581381 --- /dev/null +++ b/integrations/xchk-registry/pyproject.toml @@ -0,0 +1,23 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "xchk-registry-integration" +version = "0.2.0" +description = "Verify a Trust Record minted by an xChk registry against the key the registry publishes" +readme = "README.md" +requires-python = ">=3.11" +license = "Apache-2.0" +dependencies = [ + "agentrust-trace>=0.11.0", +] + +[project.optional-dependencies] +test = ["pytest"] + +[tool.setuptools] +py-modules = ["xchk_trace_record"] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/integrations/xchk-registry/tests/test_trust_record.py b/integrations/xchk-registry/tests/test_trust_record.py new file mode 100644 index 0000000..3494203 --- /dev/null +++ b/integrations/xchk-registry/tests/test_trust_record.py @@ -0,0 +1,631 @@ +"""Recomputed from the committed bytes with released agentrust-trace; nothing read from a generator's summary. +The live key fetch runs only with REGISTRY_LIVE=1, so CI never depends on the registry's uptime.""" +from __future__ import annotations + +import base64 +import copy +import json +import os +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +import xchk_trace_record as x # noqa: E402 + +EX = Path(__file__).resolve().parents[1] / "examples" +EXPECTED = json.loads((EX / "expected.json").read_text(encoding="utf-8")) +RECORD = json.loads((EX / "01-trust-record.json").read_text(encoding="utf-8")) +JWK = EXPECTED["trace_signer_jwk"] + + +def no_fetch(resolver, kid): + return None + + +AT_ISSUE = RECORD["iat"] + 60 # the fixture is historical: verified as of a minute after it was minted, under the default bound + + +def test_the_record_verifies_with_the_pinned_registry_key_and_a_tampered_one_does_not(): + assert x.record_verifies(RECORD, JWK, now=AT_ISSUE) + assert RECORD["cnf"]["jwk"]["kid"] == JWK["kid"] + tampered = json.loads(json.dumps(RECORD)) + tampered["iat"] += 1 + assert not x.record_verifies(tampered, JWK, now=AT_ISSUE) + + +def test_a_stale_record_is_not_verified_and_the_bound_cannot_be_switched_off(): + """Section 3.2.2: the verifier enforces a finite maximum age. The default is the library's 24 hours.""" + assert x.MAX_AGE_SECONDS == 86400 + fresh = x.verification(RECORD, JWK, now=RECORD["iat"] + x.MAX_AGE_SECONDS - 1) + assert fresh["status"] == "verified" + stale = x.verification(RECORD, JWK, now=RECORD["iat"] + x.MAX_AGE_SECONDS + 1) + assert stale["status"] == "invalid" and ("stale" in stale["reason"].lower() or "age" in stale["reason"].lower()) + assert x.verification(RECORD, JWK, max_age_seconds=3600, now=RECORD["iat"] + 3601)["status"] == "invalid" + assert x.verification(RECORD, JWK, max_age_seconds=315360000, now=RECORD["iat"] + 30 * 86400)["status"] == "verified" + for bad in (0, -1, None, True): + with pytest.raises(ValueError): + x.verification(RECORD, JWK, max_age_seconds=bad, now=AT_ISSUE) + with pytest.raises(SystemExit): + x.main([str(EX / "01-trust-record.json"), "--jwk", str(EX / "expected.json"), "--max-age", "0"]) + + +def test_the_revocation_outcome_is_reported_as_no_check_performed_when_nothing_was_checked(): + """Section 3.2.3: no bundle and no store were given, so the result says so rather than implying a clean key.""" + got = x.verify_with_key_source(RECORD, JWK, fetch=no_fetch, now=AT_ISSUE) + assert got["status"] == "verified" and got["revocation"] == "no_check_performed" + assert got["max_age_seconds"] == x.MAX_AGE_SECONDS + invalid = x.verify_with_key_source({**RECORD, "iat": RECORD["iat"] + 1}, JWK, fetch=no_fetch, now=AT_ISSUE) + assert invalid["status"] == "invalid" and invalid["revocation"] is None + + +def test_the_record_is_software_only_level_0_and_names_its_resolver(): + assert RECORD["eat_profile"].endswith("trace-v0.2") + assert "origin" not in RECORD + assert RECORD["runtime"]["platform"] == "software-only" + assert RECORD["appraisal"]["status"] == "none" + assert x.resolver_of(RECORD) == EXPECTED["resolver"] + refs = [r for r in RECORD["references"] if r["rel"] == "approval-outcome"] + assert len(refs) == 1 and len(refs[0]["id"]) == 64 and refs[0]["digest"] == "sha256:" + refs[0]["id"] + + +def test_the_embedded_key_is_never_used_without_a_pinned_or_published_key(): + assert x.verify_with_key_source(RECORD, None, fetch=no_fetch, now=AT_ISSUE)["status"] == "no-key" + got = x.verify_with_key_source(RECORD, JWK, fetch=no_fetch, now=AT_ISSUE) + assert got["status"] == "verified" and got["key_source"] == "pinned" + + +def signed_with(key, record_fields: dict, kid: str) -> dict: + """A record signed end to end by `key`, the way the library signs (Ed25519 over the JCS form, `signature` absent).""" + import agentrust_trace as t + jwk = {**t.key_to_jwk(key), "kid": kid} + payload = {**record_fields, "cnf": {"jwk": jwk}} + return {**payload, "signature": base64.urlsafe_b64encode(key.sign(t.sign._canonical_bytes(payload))).rstrip(b"=").decode()} + + +def doc_for(jwk: dict, kid: str) -> dict: + raw = base64.urlsafe_b64decode(jwk["x"] + "=" * (-len(jwk["x"]) % 4)) + return {"keys": [{"keyId": kid, "alg": "ed25519", "publicKeyRaw": base64.b64encode(raw).decode("ascii")}]} + + +def published_doc(): + raw = base64.urlsafe_b64decode(JWK["x"] + "=" * (-len(JWK["x"]) % 4)) + return {"keys": [{"keyId": JWK["kid"], "alg": "ed25519", "publicKeyRaw": base64.b64encode(raw).decode("ascii")}, + {"keyId": "other", "alg": "ml-dsa-65", "publicKeyRaw": "AAAA"}]} + + +REGISTRY = EXPECTED["resolver"] + + +def test_the_published_key_document_converts_to_the_pinned_jwk(): + doc = published_doc() + assert x.key_from_published(doc, JWK["kid"]) == {"kty": "OKP", "crv": "Ed25519", "x": JWK["x"], "kid": JWK["kid"]} + assert x.key_from_published(doc, "other") is None + got = x.verify_with_key_source(RECORD, None, fetch=lambda origin, kid: x.key_from_published(doc, kid), now=AT_ISSUE, registry=REGISTRY) + assert got["status"] == "verified" and got["key_source"] == "published" and got["trusted_origin"] == REGISTRY + + +def test_a_malformed_key_document_is_no_key_never_a_crash(): + """The fetch may answer anything. Every shape here used to raise out of key_from_published.""" + kid = JWK["kid"] + for doc in (None, [], "keys", 7, {"keys": None}, {"keys": "x"}, {"keys": {}}, {"keys": [None]}, {"keys": ["x"]}, {"keys": [7]}, + {"keys": [{"keyId": kid, "alg": "ed25519"}]}, {"keys": [{"keyId": kid, "alg": "ed25519", "publicKeyRaw": None}]}, + {"keys": [{"keyId": kid, "alg": "ed25519", "publicKeyRaw": 12}]}, {"keys": [{"keyId": kid, "alg": "ed25519", "publicKeyRaw": "not base64!"}]}, + {"keys": [{"keyId": kid, "alg": "ed25519", "publicKeyRaw": "AAAA"}]}, {"keys": [{"keyId": kid, "alg": "ed25519", "publicKeyRaw": ["x"]}]}): + assert x.key_from_published(doc, kid) is None, repr(doc) + assert x.key_from_published(published_doc(), None) is None + got = x.verify_with_key_source(RECORD, None, fetch=lambda origin, kid: x.key_from_published({"keys": [{"keyId": kid, "alg": "ed25519", "publicKeyRaw": 12}]}, kid), + now=AT_ISSUE, registry=REGISTRY) + assert got["status"] == "no-key" + + +def test_a_verified_result_carries_the_profile_fields_and_the_time_it_was_made_at(): + """Section 3.3: the profile and the complete accepted set; and the as-of instant, so a historical verdict is + never the same JSON as a current one.""" + got = x.verification(RECORD, JWK, now=AT_ISSUE) + assert got["profile"] == RECORD["eat_profile"] and got["accepted_profiles"] == [RECORD["eat_profile"]] + assert got["trusted_key_thumbprint"] + assert got["verified_at"] == AT_ISSUE and got["verified_at_overridden"] is True + later = RECORD["iat"] + 30 * 86400 + stale = x.verification(RECORD, JWK, now=later) + assert stale["status"] == "invalid" and stale["verified_at"] == later and stale["verified_at_overridden"] is True + assert "profile" not in stale + current = x.verification(RECORD, JWK) + assert current["verified_at_overridden"] is False and abs(current["verified_at"] - int(x.time.time())) < 5 + + +def test_one_instant_is_frozen_at_the_boundary_and_is_what_the_library_was_given(monkeypatch): + """A clock answering 100 then 101 (the review's two-value control): the library must be given the instant the + verdict reports, with no override and with --now, on every path (pinned, published, no-key).""" + given = [] + real = x.verify_record + + def spy(record, jwk, **kw): + given.append(kw["now"]) + return real(record, jwk, **kw) + + monkeypatch.setattr(x, "verify_record", spy) + base = RECORD["iat"] + ticks = iter([base + 100, base + 101, base + 102, base + 103, base + 104, base + 105]) + monkeypatch.setattr(x.time, "time", lambda: next(ticks)) + got = x.verify_with_key_source(RECORD, JWK, fetch=no_fetch) + assert given == [base + 100] and got["verified_at"] == base + 100 and got["verified_at_overridden"] is False, "pinned, no override" + doc = published_doc() + got = x.verify_with_key_source(RECORD, None, fetch=lambda o, k: x.key_from_published(doc, k), registry=REGISTRY) + assert given == [base + 100, base + 101] and got["verified_at"] == base + 101 and got["verified_at_overridden"] is False, "published, no override" + got = x.verify_with_key_source(RECORD, None, fetch=no_fetch) + assert given == [base + 100, base + 101] and got["verified_at"] == base + 102 and got["status"] == "no-key", "no-key: one clock read, no library call" + got = x.verify_with_key_source(RECORD, JWK, fetch=no_fetch, now=AT_ISSUE) + assert given[-1] == AT_ISSUE and got["verified_at"] == AT_ISSUE and got["verified_at_overridden"] is True, "explicit --now" + assert x.verification(RECORD, JWK)["verified_at"] == given[-1] == base + 103, "verification() alone reads the clock once" + + +def test_the_cli_freezes_one_instant_too(monkeypatch, capsys): + given = [] + real = x.verify_record + monkeypatch.setattr(x, "verify_record", lambda record, jwk, **kw: (given.append(kw["now"]), real(record, jwk, **kw))[1]) + ticks = iter([RECORD["iat"] + 7, RECORD["iat"] + 8]) + monkeypatch.setattr(x.time, "time", lambda: next(ticks)) + assert x.main([str(EX / "01-trust-record.json"), "--jwk", str(EX / "expected.json")]) == 0 + out = json.loads(capsys.readouterr().out) + assert given == [RECORD["iat"] + 7] and out["verified_at"] == RECORD["iat"] + 7 and out["verified_at_overridden"] is False + assert x.main([str(EX / "01-trust-record.json"), "--jwk", str(EX / "expected.json"), "--now", str(AT_ISSUE)]) == 0 + out = json.loads(capsys.readouterr().out) + assert given[-1] == AT_ISSUE and out["verified_at"] == AT_ISSUE and out["verified_at_overridden"] is True + + +def test_the_trust_origin_is_the_relying_partys_and_nothing_in_the_record_supplies_it(): + """A record names its subject and its resolver; a record signed under a publisher of its own choosing names that + publisher in both. So the origin comes from --registry alone, and the record must agree with it.""" + assert x.origin_of_did_web("did:web:in.xchk.io:plate:ENG-WNP3") == "https://in.xchk.io" + assert x.origin_of_did_web("did:web:example.com%3A8443:x") == "https://example.com:8443" + for bad in ("did:key:z6Mk", "did:web:", "did:web:a/b:c", "did:web:..:x", "", None, 7): + assert x.origin_of_did_web(bad) is None, repr(bad) + doc = published_doc() + asked = [] + + def fetch(origin, kid): + asked.append(origin) + return x.key_from_published(doc, kid) + + got = x.verify_with_key_source(RECORD, None, fetch=fetch, now=AT_ISSUE) + assert got["status"] == "no-key" and "--registry" in got["reason"] and asked == [], "no --registry: no trust anchor, nothing fetched" + assert got["verified_at"] == AT_ISSUE and got["verified_at_overridden"] is True, "the as-of time is on every verdict, no-key included" + elsewhere = json.loads(json.dumps(RECORD)) + elsewhere["references"][0]["resolver"] = "https://attacker.example" + got = x.verify_with_key_source(elsewhere, None, fetch=fetch, now=AT_ISSUE, registry=REGISTRY) + assert got["status"] == "no-key" and "attacker.example" in got["reason"] and asked == [], "nothing fetched from the record's server" + got = x.verify_with_key_source(RECORD, None, fetch=fetch, now=AT_ISSUE, registry="https://other.example") + assert got["status"] == "no-key" and asked == [], "the record's subject and resolver disagree with the relying party's origin" + foreign = json.loads(json.dumps(RECORD)) + foreign["subject"] = "did:web:attacker.example:plate:X" + got = x.verify_with_key_source(foreign, None, fetch=fetch, now=AT_ISSUE, registry=REGISTRY) + assert got["status"] == "no-key" and "subject" in got["reason"] and asked == [] + got = x.verify_with_key_source(RECORD, None, fetch=fetch, now=AT_ISSUE, registry="http://in.xchk.io") + assert got["status"] == "no-key" and "https" in got["reason"] and asked == [] + got = x.verify_with_key_source(RECORD, None, fetch=fetch, now=AT_ISSUE, registry="https://in.xchk.io/") + assert got["status"] == "verified" and asked == ["https://in.xchk.io"] + nokid = json.loads(json.dumps(RECORD)) + del nokid["cnf"]["jwk"]["kid"] + assert x.verify_with_key_source(nokid, None, fetch=fetch, now=AT_ISSUE, registry=REGISTRY)["status"] == "no-key" + + +def test_a_record_signed_under_its_own_chosen_publisher_is_never_verified(): + """The negative control the review asked for: a record signed, end to end, by a key the attacker made, with a + subject and resolver under a host the attacker runs and that host publishing the key. Without --registry + there is no anchor; with the relying party's registry the record disagrees with it; and if the attacker's host + is named as the registry, that is the relying party's own choice, which no verifier can second-guess.""" + import agentrust_trace as t + key = t.generate_key() + unsigned = {k: v for k, v in RECORD.items() if k not in ("signature", "cnf")} + unsigned["subject"] = "did:web:attacker.example:plate:EVIL" + unsigned["references"] = [{**unsigned["references"][0], "resolver": "https://attacker.example"}] + # signed the way the library signs (Ed25519 over the JCS form with `signature` absent), with a kid in cnf.jwk + attacker_jwk = {**t.key_to_jwk(key), "kid": "evil-kid"} + payload = {**unsigned, "cnf": {"jwk": attacker_jwk}} + forged = {**payload, "signature": base64.urlsafe_b64encode(key.sign(t.sign._canonical_bytes(payload))).rstrip(b"=").decode()} + assert x.record_verifies(forged, attacker_jwk, now=AT_ISSUE), "the forgery is internally consistent" + raw = base64.urlsafe_b64decode(attacker_jwk["x"] + "=" * (-len(attacker_jwk["x"]) % 4)) + attacker_doc = {"keys": [{"keyId": "evil-kid", "alg": "ed25519", "publicKeyRaw": base64.b64encode(raw).decode("ascii")}]} + served = {"https://attacker.example": attacker_doc, REGISTRY: published_doc()} + asked = [] + + def fetch(origin, kid): + asked.append(origin) + return x.key_from_published(served.get(origin, {}), kid) + + got = x.verify_with_key_source(forged, None, fetch=fetch, now=AT_ISSUE) + assert got["status"] == "no-key" and asked == [], "no anchor, no fetch" + got = x.verify_with_key_source(forged, None, fetch=fetch, now=AT_ISSUE, registry=REGISTRY) + assert got["status"] == "no-key" and asked == [], "the forgery does not agree with the registry the relying party trusts" + got = x.verify_with_key_source(forged, JWK, fetch=fetch, now=AT_ISSUE) + assert got["status"] == "invalid", "pinned to the real registry key, the forgery's signature fails" + + +def test_the_key_document_fetch_refuses_http_redirects_and_oversize(monkeypatch): + assert x.published_key("http://in.xchk.io", JWK["kid"]) is None + assert x.published_key("ftp://in.xchk.io", JWK["kid"]) is None + + class Resp: + def __init__(self, status, body): self.status, self._b = status, body + def __enter__(self): return self + def __exit__(self, *a): return False + def read(self, n=-1): return self._b if n < 0 else self._b[:n] + + class Opener: + def __init__(self, resp): self.resp = resp + def open(self, url, timeout=0): return self.resp + + monkeypatch.setattr(x.urllib.request, "build_opener", lambda *h: Opener(Resp(200, b"x" * (x.KEY_DOC_MAX_BYTES + 1)))) + assert x.published_key("https://in.xchk.io", JWK["kid"]) is None, "a document past the bound is not read" + monkeypatch.setattr(x.urllib.request, "build_opener", lambda *h: Opener(Resp(302, b"{}"))) + assert x.published_key("https://in.xchk.io", JWK["kid"]) is None + monkeypatch.setattr(x.urllib.request, "build_opener", lambda *h: Opener(Resp(200, json.dumps(published_doc()).encode()))) + assert x.published_key("https://in.xchk.io", JWK["kid"]) == {"kty": "OKP", "crv": "Ed25519", "x": JWK["x"], "kid": JWK["kid"]} + assert x._NoRedirect().redirect_request(None, None, 302, "", {}, "https://elsewhere") is None + + +def test_the_cli_runs_offline_with_a_pinned_key(capsys): + rc = x.main([str(EX / "01-trust-record.json"), "--jwk", str(EX / "expected.json"), "--now", str(AT_ISSUE)]) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["status"] == "verified" and out["key_source"] == "pinned" + assert out["revocation"] == "no_check_performed" and out["max_age_seconds"] == x.MAX_AGE_SECONDS + capsys.readouterr() + long_after = RECORD["iat"] + 30 * 86400 + assert x.main([str(EX / "01-trust-record.json"), "--jwk", str(EX / "expected.json"), "--now", str(long_after)]) == 1 + assert json.loads(capsys.readouterr().out)["status"] == "invalid", "a month on, under the default bound, the record is stale" + assert len(out["approval_outcome_references"]) == 1 + + +@pytest.mark.skipif(os.environ.get("REGISTRY_LIVE") != "1", reason="set REGISTRY_LIVE=1 to fetch the published key") +def test_the_published_key_at_the_registry_matches_the_pinned_one(): + assert x.published_key(EXPECTED["resolver"], JWK["kid"]) == {**JWK, "kid": JWK["kid"]} + + +# 18 Oct 2026, after four review rounds found input shapes one at a time: every top-level field, the key fields and the +# reference fields set to each odd value, through main() on all three paths. A verdict and an exit code, or it is a bug. +ODD = [None, [], {}, "", "x", 0, 1, -1, True, 1.5, [1], {"a": 1}, "did:web:in.xchk.io:plate:ENG-WNP3", "https://in.xchk.io"] + + +def _mutations(): + for k in list(RECORD) + ["cnf.jwk", "cnf.jwk.kid", "references.0", "references.0.resolver", "references.0.rel"]: + for v in ODD: + r = copy.deepcopy(RECORD) + parts = k.split(".") + tgt = r + for p in parts[:-1]: + tgt = tgt[int(p)] if p.isdigit() else tgt[p] + last = parts[-1] + if last.isdigit(): + tgt[int(last)] = v + else: + tgt[last] = v + yield f"{k}={v!r}", r + for whole in ODD: + yield f"={whole!r}", whole + + +def test_no_mutation_of_the_record_crashes_the_cli_on_any_path(tmp_path, capsys, monkeypatch): + doc = published_doc() + + class Resp: + status = 200 + def __enter__(self): return self + def __exit__(self, *a): return False + def read(self, n=-1): return json.dumps(doc).encode() + + class Opener: + def open(self, url, timeout=0): return Resp() + + monkeypatch.setattr(x.urllib.request, "build_opener", lambda *h: Opener()) + paths = (["--jwk", str(EX / "expected.json")], ["--registry", REGISTRY], []) + n = 0 + for label, record in _mutations(): + f = tmp_path / "r.json" + f.write_text(json.dumps(record), encoding="utf-8") + for extra in paths: + try: + rc = x.main([str(f), "--now", str(AT_ISSUE)] + extra) + except SystemExit as e: + rc = e.code + except Exception as e: # noqa: BLE001 + raise AssertionError(f"{label} with {extra[:1]} crashed: {type(e).__name__}: {e}") from e + out = json.loads(capsys.readouterr().out) + assert rc in (0, 1) and out["status"] in ("verified", "invalid", "no-key"), label + assert "verified_at" in out and out["verified_at"] == AT_ISSUE, label + n += 1 + assert n > 700 + + +def test_local_input_files_and_arguments_fail_closed_with_a_verdict(tmp_path, capsys): + rec = str(EX / "01-trust-record.json") + def cli(args): + try: + rc = x.main(args) + except SystemExit as e: + rc = e.code + return rc, json.loads(capsys.readouterr().out or "{}") + bad = tmp_path / "bad.json" + for body in ("not json", "[]", "null", "7", '"x"'): + bad.write_text(body, encoding="utf-8") + rc, out = cli([bad.as_posix(), "--jwk", str(EX / "expected.json"), "--now", str(AT_ISSUE)]) + assert rc == 1 and out["status"] == "invalid" and out["verified_at"] == AT_ISSUE, body + rc, out = cli([str(tmp_path / "missing.json"), "--jwk", str(EX / "expected.json")]) + assert rc == 1 and out["status"] == "invalid" and "record file" in out["reason"] + for body in ("not json", "[]", "null", "7", '{"trace_signer_jwk": []}', '{"kty":"RSA"}', '{"trace_signer_jwk": {"kty":"OKP","crv":"Ed25519","x":12}}', + '{"trace_signer_jwk": {"kty":"OKP","crv":"Ed25519","x":"!!"}}', '{"kty":"OKP","crv":"Ed25519","x":"AAAA"}'): + bad.write_text(body, encoding="utf-8") + rc, out = cli([rec, "--jwk", bad.as_posix(), "--now", str(AT_ISSUE)]) + assert rc == 1 and out["status"] == "no-key" and out["kid"] == RECORD["cnf"]["jwk"]["kid"], body + for reg in ("http://in.xchk.io", "https://user@in.xchk.io", "https://user:pw@in.xchk.io", "https://in.xchk.io/path", "https://in.xchk.io?x=1", "https://in.xchk.io#f", "https://", "in.xchk.io", ""): + rc, out = cli([rec, "--registry", reg, "--now", str(AT_ISSUE)]) + assert rc == 1 and out["status"] == "no-key", reg + assert x.https_origin("HTTPS://In.xchk.io/") == ("https://in.xchk.io", None) + assert x.https_origin("https://in.xchk.io:8443") == ("https://in.xchk.io:8443", None) + with pytest.raises(SystemExit): + x.main([rec, "--jwk", str(EX / "expected.json"), "--now", "-5"]) + capsys.readouterr() + + +def test_the_verdict_names_the_key_the_record_claims_and_the_key_actually_used(): + got = x.verify_with_key_source(RECORD, JWK, fetch=no_fetch, now=AT_ISSUE) + assert got["kid"] == RECORD["cnf"]["jwk"]["kid"] and got["pinned_kid"] == JWK["kid"] and got["trusted_key_thumbprint"] + unnamed = {k: v for k, v in JWK.items() if k != "kid"} + got = x.verify_with_key_source(RECORD, unnamed, fetch=no_fetch, now=AT_ISSUE) + assert got["status"] == "verified" and got["kid"] == RECORD["cnf"]["jwk"]["kid"] and got["pinned_kid"] is None + assert x.verify_with_key_source(RECORD, {"kty": "RSA"}, fetch=no_fetch, now=AT_ISSUE)["status"] == "no-key" + + +# Review of 2dcb159 (noah-ing, dinakarjs): three bounded cases, each with a freshly signed synthetic record and a valid twin. +def _fresh_signed(**changes): + import agentrust_trace as t + key = t.generate_key() + fields = {k: v for k, v in RECORD.items() if k not in ("signature", "cnf")} + fields.update(changes) + return signed_with(key, fields, "fresh-kid"), {**t.key_to_jwk(key), "kid": "fresh-kid"} + + +def test_malformed_registry_origins_are_refusals_with_a_verdict_and_no_fetch(capsys): + rec = str(EX / "01-trust-record.json") + asked = [] + for reg in ("https://in.xchk.io:bad", "https://in.xchk.io:65536", "https://in.xchk.io:-1", "https://[broken", "https://[::1", "https://in.xchk.io:", "https://:443", "https://in.xchk.io:80:80"): + origin, why = x.https_origin(reg) + assert origin is None and why, reg + got = x.verify_with_key_source(RECORD, None, fetch=lambda o, k: asked.append(o), now=AT_ISSUE, registry=reg) + assert got["status"] == "no-key" and got["reason"] and asked == [], reg + try: + rc = x.main([rec, "--registry", reg, "--now", str(AT_ISSUE)]) + except SystemExit as e: + rc = e.code + out = json.loads(capsys.readouterr().out) + assert rc == 1 and out["status"] == "no-key" and out["verified_at"] == AT_ISSUE, reg + assert x.https_origin("https://[::1]:8443") == ("https://[::1]:8443", None) + assert x.https_origin("https://in.xchk.io:443") == ("https://in.xchk.io", None), "443 is the same origin with or without the port (RFC 6454)" + assert x.origin_of_did_web("did:web:in.xchk.io%3A443:plate:X") == "https://in.xchk.io" + assert x.https_origin("https://in.xchk.io:8443") == ("https://in.xchk.io:8443", None) + # a malformed --registry is refused even beside a pinned key: an option that is given is checked + got = x.verify_with_key_source(RECORD, JWK, fetch=no_fetch, now=AT_ISSUE, registry="https://in.xchk.io:bad") + assert got["status"] == "no-key" and "parsed" in got["reason"] + assert x.verify_with_key_source(RECORD, JWK, fetch=no_fetch, now=AT_ISSUE, registry="https://in.xchk.io:443")["status"] == "verified" + # the valid twin: same record, well-formed origin, the key fetched once + doc = published_doc() + got = x.verify_with_key_source(RECORD, None, fetch=lambda o, k: (asked.append(o), x.key_from_published(doc, k))[1], now=AT_ISSUE, registry=REGISTRY) + assert got["status"] == "verified" and asked == [REGISTRY] + + +def test_the_subject_must_be_a_did_web_under_the_trusted_origin(): + asked = [] + for subject in ("did:key:z6Mkunbound", "", None, 7, "did:web:", "urn:x", "did:web:other.example:plate:X"): + rec, jwk = _fresh_signed(subject=subject) + if subject in ("did:key:z6Mkunbound", "did:web:other.example:plate:X"): + assert x.record_verifies(rec, jwk, now=AT_ISSUE), "the control is validly signed under its own key (the library's schema rejects the malformed ones on its own)" + doc = doc_for(jwk, "fresh-kid") + got = x.verify_with_key_source(rec, None, fetch=lambda o, k: (asked.append(o), x.key_from_published(doc, k))[1], now=AT_ISSUE, registry=REGISTRY) + assert got["status"] == "no-key" and "subject" in got["reason"] and asked == [], repr(subject) + rec, jwk = _fresh_signed(subject="did:web:in.xchk.io:plate:TWIN") + doc = doc_for(jwk, "fresh-kid") + got = x.verify_with_key_source(rec, None, fetch=lambda o, k: (asked.append(o), x.key_from_published(doc, k))[1], now=AT_ISSUE, registry=REGISTRY) + assert got["status"] == "verified" and asked == [REGISTRY], "the twin: a did:web under the origin, verified with the fetched key" + + +def test_every_approval_outcome_reference_must_resolve_at_the_trusted_origin_whatever_its_order(): + on = RECORD["references"][0] + off = {**on, "id": "b" * 64, "digest": "sha256:" + "b" * 64, "resolver": "https://attacker.example"} + unnamed = {k: v for k, v in on.items() if k != "resolver"} + other_kind = {"rel": "something-else", "id": "c" * 64, "digest": "sha256:" + "c" * 64, "resolver": "https://attacker.example"} + asked = [] + def check(refs, expect, why=None): + rec, jwk = _fresh_signed(references=refs) + doc = doc_for(jwk, "fresh-kid") + got = x.verify_with_key_source(rec, None, fetch=lambda o, k: (asked.append(o), x.key_from_published(doc, k))[1], now=AT_ISSUE, registry=REGISTRY) + assert got["status"] == expect, (refs, got.get("reason")) + if why: + assert why in got["reason"], got["reason"] + check([on, off], "no-key", "reference 1") + check([off, on], "no-key", "reference 0") + check([on, on, off], "no-key", "reference 2") + assert asked == [], "nothing fetched for any order" + check([on, unnamed], "no-key", "names no resolver") + check([unnamed], "no-key", "names no resolver") + check([on, other_kind], "verified") + check([on, on], "verified") + check([], "verified") + assert asked == [REGISTRY, REGISTRY, REGISTRY], "the twins fetched the key once each" + assert x.approval_resolvers({"references": [on, off]}) == ([on["resolver"], "https://attacker.example"], None) + + +# The sweep that should have come first: many origin strings, subjects and reference orderings, on signed synthetic +# records, through verify_with_key_source and main(). A verdict every time, and a fetch only for a well-formed origin +# that the record agrees with. +ORIGINS = ["https://in.xchk.io", "https://in.xchk.io/", "HTTPS://IN.XCHK.IO", "https://in.xchk.io:443", "https://in.xchk.io:8443", "https://[::1]:8443", + "https://in.xchk.io:", "https://in.xchk.io:bad", "https://in.xchk.io:65536", "https://in.xchk.io:-1", "https://in.xchk.io:80:80", "https://[broken", "https://[::1", + "https://", "https:///", "https://:443", "https://in.xchk.io/path", "https://in.xchk.io?x=1", "https://in.xchk.io#f", "https://u@in.xchk.io", "https://u:p@in.xchk.io", + "http://in.xchk.io", "ftp://in.xchk.io", "in.xchk.io", "//in.xchk.io", "", " ", " https://in.xchk.io", "https://in.xchk.io ", "https://in.xchk.io\n", "https://in.xchk.io.", + "https://xn--nxasmq6b.example", "https://ïn.xchk.io", "https://in.xchk.io\x00", "https://in.xchk.io:0", "https://in.xchk.io:00443", "https://127.0.0.1", "https://[::ffff:1.2.3.4]", + "https://in.xchk.io%2Fx", "https://in.xchk.io\\x", "https://in.xchk.io;x", "https://in.xchk.io:443/", "https://IN.xchk.IO:443"] +SUBJECTS = [RECORD["subject"], "did:web:IN.XCHK.IO:plate:X", "did:web:in.xchk.io", "did:web:in.xchk.io%3A443:plate:X", "did:web:in.xchk.io%3A8443:plate:X", "did:key:z6Mk", "did:web:", "did:web:other.example:x", + "", None, 7, [], {}, "did:web:in.xchk.io/plate", "did:web:in.xchk.io@x", "did:web:..", "did:web:in.xchk.io.", "DID:WEB:in.xchk.io:plate:X"] + + +def test_an_origin_given_beside_a_pinned_key_is_enforced_not_just_parsed(): + """Review of 09c436b: with --jwk the registry was only parsed. The caller named the registry the record must + belong to; the record must agree, pinned key or not.""" + ok = x.verify_with_key_source(RECORD, JWK, fetch=no_fetch, now=AT_ISSUE, registry=REGISTRY) + assert ok["status"] == "verified" and ok["key_source"] == "pinned" and ok["trusted_origin"] == REGISTRY + ok443 = x.verify_with_key_source(RECORD, JWK, fetch=no_fetch, now=AT_ISSUE, registry="https://in.xchk.io:443") + assert ok443["status"] == "verified" and ok443["trusted_origin"] == REGISTRY + bad = x.verify_with_key_source(RECORD, JWK, fetch=no_fetch, now=AT_ISSUE, registry="https://other.example") + assert bad["status"] == "no-key" and "subject lives under" in bad["reason"] + for subject in ("did:key:z6Mkunbound", None, "did:web:other.example:plate:X"): + rec, jwk = _fresh_signed(subject=subject) + got = x.verify_with_key_source(rec, jwk, fetch=no_fetch, now=AT_ISSUE, registry=REGISTRY) + assert got["status"] == "no-key" and "subject" in got["reason"], repr(subject) + assert x.verify_with_key_source(rec, jwk, fetch=no_fetch, now=AT_ISSUE)["status"] in ("verified", "invalid"), "pinned alone: the key decides, as before" + on = RECORD["references"][0] + off = {**on, "id": "b" * 64, "digest": "sha256:" + "b" * 64, "resolver": "https://attacker.example"} + rec, jwk = _fresh_signed(references=[on, off]) + got = x.verify_with_key_source(rec, jwk, fetch=no_fetch, now=AT_ISSUE, registry=REGISTRY) + assert got["status"] == "no-key" and "reference 1" in got["reason"] + assert x.verify_with_key_source(rec, jwk, fetch=no_fetch, now=AT_ISSUE)["status"] == "verified", "pinned alone verifies it; the origin is what refuses it" + + +def test_resolvers_are_compared_as_origins_so_an_explicit_443_is_the_same_origin(): + on = RECORD["references"][0] + rec, jwk = _fresh_signed(subject="did:web:in.xchk.io%3A443:plate:X", references=[{**on, "resolver": "https://in.xchk.io:443"}]) + doc = doc_for(jwk, "fresh-kid") + asked = [] + got = x.verify_with_key_source(rec, None, fetch=lambda o, k: (asked.append(o), x.key_from_published(doc, k))[1], now=AT_ISSUE, registry="https://in.xchk.io:443") + assert got["status"] == "verified" and asked == [REGISTRY], got.get("reason") + rec, jwk = _fresh_signed(references=[{**on, "resolver": "HTTPS://IN.XCHK.IO/"}]) + doc = doc_for(jwk, "fresh-kid") + assert x.verify_with_key_source(rec, None, fetch=lambda o, k: x.key_from_published(doc, k), now=AT_ISSUE, registry=REGISTRY)["status"] == "verified" + for resolver in ("https://in.xchk.io/api", "http://in.xchk.io", "https://in.xchk.io:8443", "https://u@in.xchk.io", "in.xchk.io"): + rec, jwk = _fresh_signed(references=[{**on, "resolver": resolver}]) + got = x.verify_with_key_source(rec, None, fetch=no_fetch, now=AT_ISSUE, registry=REGISTRY) + assert got["status"] == "no-key" and "reference 0" in got["reason"], resolver + + +def expected_origin(value): + """An oracle for the sweep that shares no code with the verifier: what a careful reader expects an https origin + string to mean, or None. Written from RFC 6454 and the README, not from https_origin.""" + import re + m = re.fullmatch(r"(?i)https://(\[[0-9a-f:.]+\]|[a-z0-9.-]+)(?::(\d{1,5}))?/?", value or "") + if not m or "@" in value or value != value.strip(): + return None + host, port = m.group(1).lower(), m.group(2) + if port is not None and (int(port) > 65535 or port != str(int(port))): + return None + if host.startswith(".") or host.endswith(".") or ".." in host: + return None + return f"https://{host}" + (f":{int(port)}" if port and int(port) != 443 else "") + + +def test_sweep_origins_subjects_and_references_on_signed_records(tmp_path, capsys, monkeypatch): + import agentrust_trace as t + key = t.generate_key() + jwk = {**t.key_to_jwk(key), "kid": "fresh-kid"} + doc = doc_for(jwk, "fresh-kid") + asked = [] + + def fetch(o, k): + asked.append(o) + return x.key_from_published(doc, k) + + base = {k: v for k, v in RECORD.items() if k not in ("signature", "cnf")} + on = RECORD["references"][0] + off = {**on, "id": "b" * 64, "digest": "sha256:" + "b" * 64, "resolver": "https://attacker.example"} + noresolver = {k: v for k, v in on.items() if k != "resolver"} + ref_sets = [[on], [on, on], [on, off], [off, on], [on, noresolver], [], [off], [on, {**on, "resolver": "https://IN.xchk.io/"}]] + def expected_subject_origin(subject): + import re + m = re.fullmatch(r"did:web:([^:/@]+?)(?:%3[aA](\d+))?(?::.*)?", subject or "") if isinstance(subject, str) else None + if not m: + return None + host = m.group(1).lower() + if host.startswith(".") or host.endswith(".") or ".." in host: + return None + return f"https://{host}" + (f":{m.group(2)}" if m.group(2) and m.group(2) != "443" else "") + + runs = 0 + for origin in ORIGINS: + want = expected_origin(origin) # the oracle, independent of https_origin + for subject in SUBJECTS: + for refs in ref_sets: + rec = signed_with(key, {**base, "subject": subject, "references": refs}, "fresh-kid") + agrees = bool(want) and expected_subject_origin(subject) == want and all( + expected_origin(r.get("resolver")) == want for r in refs if r.get("rel") == "approval-outcome") + for pinned in (None, jwk): + before = len(asked) + got = x.verify_with_key_source(rec, pinned, fetch=fetch, now=AT_ISSUE, registry=origin) + runs += 1 + assert got["status"] in ("verified", "invalid", "no-key") and got["verified_at"] == AT_ISSUE, (origin, subject, refs, pinned is not None) + fetched = len(asked) - before + if pinned is None: + assert fetched == (1 if agrees else 0), (origin, subject, refs, got.get("reason")) + else: + assert fetched == 0, "a pinned key never fetches" + if not agrees: + assert got["status"] == "no-key", (origin, subject, refs, pinned is not None, got.get("reason")) + elif pinned is not None: + assert got["status"] in ("verified", "invalid"), "agreeing record, pinned key: the key decides" + assert runs == 2 * len(ORIGINS) * len(SUBJECTS) * len(ref_sets) + # the same origins through the CLI, pinned and published, no crash + f = tmp_path / "r.json" + f.write_text(json.dumps(RECORD), encoding="utf-8") + + class Resp: + status = 200 + def __enter__(self): return self + def __exit__(self, *a): return False + def read(self, n=-1): return json.dumps(published_doc()).encode() + + class Opener: + def open(self, url, timeout=0): return Resp() + + monkeypatch.setattr(x.urllib.request, "build_opener", lambda *h: Opener()) + for origin in ORIGINS: + for extra in (["--registry", origin], ["--registry", origin, "--jwk", str(EX / "expected.json")]): + try: + rc = x.main([str(f), "--now", str(AT_ISSUE)] + extra) + except SystemExit as e: + rc = e.code + except Exception as e: # noqa: BLE001 + raise AssertionError(f"{origin!r} crashed the CLI: {type(e).__name__}: {e}") from e + out = json.loads(capsys.readouterr().out) + assert rc in (0, 1) and out["status"] in ("verified", "invalid", "no-key"), origin + + +def test_the_thumbprint_is_the_rfc7638_digest_of_the_public_key_the_library_verified_with(): + """The reported trusted_key_thumbprint is the library's own RFC 7638 function over the public key, and equals the + thumbprint on the library's verification result (19 Oct 2026: computed rather than read off the result).""" + from agentrust_trace import verify_record + got = x.verification(RECORD, JWK, now=AT_ISSUE) + lib = verify_record(RECORD, JWK, max_age_seconds=x.MAX_AGE_SECONDS, now=AT_ISSUE) + assert got["trusted_key_thumbprint"] == lib.trusted_key_thumbprint + assert got["trusted_key_thumbprint"] == x.verification(RECORD, {k: v for k, v in JWK.items() if k != "kid"}, now=AT_ISSUE)["trusted_key_thumbprint"] + + +def test_a_pinned_private_key_is_refused_and_not_echoed(tmp_path, capsys): + """A trust anchor is a public key. A key file that carries the private member d is refused with fixed text, + through the function and through the command line, and no part of the file reaches the output.""" + synthetic_d = "c3ludGhldGljLXByaXZhdGUtbWVtYmVyLW5vdC1hLWtleQ" + private = dict(JWK, d=synthetic_d) + got = x.verify_with_key_source(RECORD, private, fetch=no_fetch, now=AT_ISSUE) + assert got["status"] == "no-key" and "private key" in got["reason"] + assert synthetic_d not in json.dumps(got) + for doc in (private, {"trace_signer_jwk": private}): + p = tmp_path / "pinned.json" + p.write_text(json.dumps(doc), encoding="utf-8") + assert x.main([str(EX / "01-trust-record.json"), "--jwk", str(p), "--now", str(AT_ISSUE)]) == 1 + out = capsys.readouterr().out + assert '"no-key"' in out and "private key" in out and synthetic_d not in out + + +def test_a_key_file_that_does_not_decode_is_reported_by_error_class_only(tmp_path, capsys): + p = tmp_path / "pinned.json" + p.write_bytes(b'{"kty": "OKP", "x": "\xff\xfe-synthetic-bytes"}') + assert x.main([str(EX / "01-trust-record.json"), "--jwk", str(p), "--now", str(AT_ISSUE)]) == 1 + out = json.loads(capsys.readouterr().out) + assert out["status"] == "no-key" and out["reason"] == "pinned key file: UnicodeDecodeError" diff --git a/integrations/xchk-registry/xchk_trace_record.py b/integrations/xchk-registry/xchk_trace_record.py new file mode 100644 index 0000000..ea28599 --- /dev/null +++ b/integrations/xchk-registry/xchk_trace_record.py @@ -0,0 +1,363 @@ +"""Verify a Trust Record minted by an xChk registry against a key the relying party trusts. Apache-2.0. + + python xchk_trace_record.py record.json --registry https://in.xchk.io # the key that registry publishes under the record's kid + python xchk_trace_record.py record.json --jwk expected.json # offline: a pinned signer key + +The registry signs each record with its seal key. The record's `cnf.jwk` carries that key and its `kid`; the same +key is published at `/.well-known/xchk-keys.json`, where `` is the registry the relying party trusts, +given out of band with --registry; nothing in the record supplies it. The record's `subject` must be a `did:web` +under that origin, and every `approval-outcome` reference must resolve there. This tool verifies with the pinned key when given one, else with the published key under +that `kid`. It never verifies against the key the record itself carries, and never fetches a key from a server the +record names, in its references or in its subject: either way a self-made record would pass, which proves integrity +and not origin. Without --registry or --jwk there is no trust anchor and the answer is `no-key`. + +The record's age is bounded (default 24 hours, --max-age, never off; --now re-verifies a retained record at its own +time, and every verdict says which instant it was made against). A verified result reports the profile it verified +under and the whole accepted set (section 3.3), and the revocation outcome as the library reported it: with no +revocation bundle or store, `no_check_performed`, which is not an affirming appraisal (section 3.2.3). +""" +from __future__ import annotations + +import argparse +import base64 +import json +import sys +import time +import urllib.parse +import urllib.request +from pathlib import Path + +from agentrust_trace import jwk_thumbprint, verify_record +from cryptography.exceptions import InvalidSignature + + +# Section 3.2.2's default maximum record age, the library's own default. The age check is never disabled here: a +# record older than the bound is reported `invalid` with the reason, and a historical record is re-verified at its +# own time with --now, not by widening the bound to nothing. (18 Oct 2026, review of integrations#338.) +MAX_AGE_SECONDS = 86400 + + +class Instant: + """ONE effective verification instant, frozen where the caller's request enters and used for everything after: + the library call and every emitted verdict, no-key included. The clock is read at most once per verdict, so the + instant the library checked against and the instant the verdict reports can never differ (review of #338, + 18 Oct 2026: with a clock answering 100 then 101, the library had been given 101 while the verdict said 100). + `overridden` says whether the caller supplied it (--now, a retained record re-verified at its own time).""" + + __slots__ = ("at", "overridden") + + def __init__(self, now: int | None = None): + self.overridden = now is not None + self.at = int(now) if self.overridden else int(time.time()) + + def fields(self) -> dict: + return {"verified_at": self.at, "verified_at_overridden": self.overridden} + + +def _instant(now) -> Instant: + return now if isinstance(now, Instant) else Instant(now) + + +def verification(record: dict, jwk: dict, max_age_seconds: int = MAX_AGE_SECONDS, now: int | Instant | None = None) -> dict: + """The library's verdict on a record under a trusted key: status, the revocation outcome as the library + reported it (section 3.2.3: `no_check_performed` when no bundle or store was given, as here), and the reason + for a refusal. `max_age_seconds` must be a positive bound. `now` may be an already-frozen Instant.""" + if not isinstance(max_age_seconds, int) or isinstance(max_age_seconds, bool) or max_age_seconds <= 0: + raise ValueError("max_age_seconds must be a positive number of seconds; the age check cannot be disabled") + instant = _instant(now) + when = instant.fields() + if not isinstance(record, dict): + return {"status": "invalid", "reason": f"record is not a JSON object ({type(record).__name__})", "revocation": None, **when} + if not is_ed25519_jwk(jwk): + return {"status": "invalid", "reason": "trusted key is not an Ed25519 JWK", "revocation": None, **when} + try: + result = verify_record(record, jwk, max_age_seconds=max_age_seconds, now=instant.at) + except InvalidSignature: + return {"status": "invalid", "reason": "signature does not verify", "revocation": None, **when} + except ValueError as e: + return {"status": "invalid", "reason": str(e), "revocation": None, **when} + except Exception as e: # noqa: BLE001 a record shape the library did not foresee is not verified, and says why + return {"status": "invalid", "reason": f"{type(e).__name__}: {e}", "revocation": None, **when} + revocation = getattr(result, "revocation", None) + # section 3.3: a successful result reports the profile it verified under and the whole set the verifier accepts + return {"status": "verified", "revocation": getattr(revocation, "outcome", None) or "no_check_performed", + "profile": getattr(result, "profile", None), "accepted_profiles": list(getattr(result, "accepted_profiles", ()) or ()), + # RFC 7638 thumbprint of the PUBLIC key the record verified under, from the library's own function (19 Oct + # 2026: the attribute read was named like a secret to CodeQL's heuristics; the value is a public-key digest) + "trusted_key_thumbprint": jwk_thumbprint(jwk), **when} + + +def record_verifies(record: dict, jwk: dict, max_age_seconds: int = MAX_AGE_SECONDS, now: int | Instant | None = None) -> bool: + return verification(record, jwk, max_age_seconds, now)["status"] == "verified" + + +def is_ed25519_jwk(jwk) -> bool: + """An OKP/Ed25519 JWK with base64url key material of 32 bytes; nothing else is a key this tool verifies with.""" + if not isinstance(jwk, dict) or jwk.get("kty") != "OKP" or jwk.get("crv") != "Ed25519": + return False + x = jwk.get("x") + if not isinstance(x, str) or not x: + return False + try: + return len(base64.urlsafe_b64decode(x + "=" * (-len(x) % 4))) == 32 + except (ValueError, TypeError): + return False + + +def key_from_published(doc, kid: str) -> dict | None: + """The Ed25519 key with this id from the registry's key document, as a JWK; None when absent or when the + document is not the shape it should be. Nothing in here raises on a bad document: a key server that answers + nonsense is no key, not a crash (review of #338, 18 Oct 2026).""" + if not isinstance(doc, dict) or not isinstance(kid, str) or not kid: + return None + keys = doc.get("keys") + if not isinstance(keys, list): + return None + for k in keys: + if not isinstance(k, dict) or k.get("keyId") != kid or k.get("alg") != "ed25519": + continue + material = k.get("publicKeyRaw") + if not isinstance(material, str) or not material: + return None + try: + raw = base64.b64decode(material, validate=True) + except (ValueError, TypeError): + return None + if len(raw) != 32: + return None + return {"kty": "OKP", "crv": "Ed25519", "x": base64.urlsafe_b64encode(raw).decode("ascii").rstrip("="), "kid": kid} + return None + + +# The published-key path, bounded (18 Oct 2026). The key document is fetched from ONE origin, the one the relying +# party passed with --registry; nothing in the record supplies it (see trusted_origin). The record's `subject` and +# `resolver` are checked against it and never followed, because a record that names the server its key will be +# fetched from has chosen its own trust root, which is the embedded-key problem one hop away. Only https; no +# redirects; the document is read to a bound; a document of the wrong shape is no key. +KEY_DOC_MAX_BYTES = 256 * 1024 + + +class _NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D102 + return None + + +def origin_of_did_web(subject: str) -> str | None: + """`did:web:host[:port-encoded][:path...]` -> `https://host[:port]`; None for anything else (did-core, did:web).""" + if not isinstance(subject, str) or not subject.startswith("did:web:"): + return None + host = subject.split(":", 2)[2].split(":", 1)[0].replace("%3A", ":").replace("%3a", ":") + if not host or "/" in host or "@" in host or host.startswith(".") or host.endswith(".") or ".." in host: + return None + if host.endswith(":443"): # the same origin as without the port (RFC 6454), as https_origin also says + host = host[: -len(":443")] + return f"https://{host.lower()}" + + +def _same_origin(a, b) -> bool: + return isinstance(a, str) and isinstance(b, str) and a.rstrip("/").lower() == b.rstrip("/").lower() + + +def https_origin(value) -> tuple[str | None, str | None]: + """(origin, refusal): `https://host[:port]` and nothing else: no userinfo, path, query or fragment, because the + key document's location is derived from it and anything more is a different document.""" + if not isinstance(value, str) or not value: + return None, "the trusted origin must be an https origin" + if value != value.strip() or any(c.isspace() or ord(c) < 32 for c in value): + return None, "the trusted origin must not contain whitespace or control characters" # urlsplit would quietly strip them + try: + u = urllib.parse.urlsplit(value) + host, port = u.hostname, u.port # both raise ValueError on a malformed netloc (bad bracket, non-numeric or out-of-range port) + except ValueError as e: + return None, f"the trusted origin could not be parsed: {e}" + if u.scheme.lower() != "https": + return None, "the trusted origin must be https" + if not host or u.username is not None or u.password is not None: + return None, "the trusted origin must be a host, with no credentials" + if u.path not in ("", "/") or u.query or u.fragment: + return None, "the trusted origin must be an origin only: no path, query or fragment" + # the netloc must be exactly host or host:port: a trailing colon, an empty port or anything else in it is refused + shown = f"[{host.lower()}]" if ":" in host else host.lower() + if u.netloc.lower() != shown + (f":{port}" if port else ""): + return None, "the trusted origin must be host or host:port and nothing else" + # RFC 6454: https on 443 is the same origin with or without the port written + return f"https://{shown}" + (f":{port}" if port and port != 443 else ""), None + + +def published_key(origin: str, kid: str, timeout: float = 20.0) -> dict | None: + """The key with this id from `/.well-known/xchk-keys.json`; None when it cannot be had safely.""" + if not isinstance(origin, str) or not origin.lower().startswith("https://"): + return None + url = f"{origin.rstrip('/')}/.well-known/xchk-keys.json" + try: + opener = urllib.request.build_opener(_NoRedirect()) + with opener.open(url, timeout=timeout) as r: + if r.status != 200: + return None + raw = r.read(KEY_DOC_MAX_BYTES + 1) + if len(raw) > KEY_DOC_MAX_BYTES: + return None + doc = json.loads(raw) + except Exception: # noqa: BLE001 + return None + try: + return key_from_published(doc, kid) + except Exception: # noqa: BLE001 a document shape nobody foresaw is still no key, never a crash + return None + + +def references_of(record) -> list: + """The record's reference objects; anything that is not a list of objects reads as none.""" + refs = record.get("references") if isinstance(record, dict) else None + return [r for r in refs if isinstance(r, dict)] if isinstance(refs, list) else [] + + +def approval_resolvers(record) -> tuple[list, str | None]: + """(resolvers, refusal): the resolver of EVERY approval-outcome reference, in order. A reference of that rel with + no string resolver is a refusal: the registry is obliged to keep each one resolvable (section 3.1.2), so one that + names no resolver is not a reference this tool can check against the origin. References of other kinds are + ignored. (Review of #338, 18 Oct 2026: only the first resolver used to be checked, so the verdict depended on + the order of the references.)""" + out = [] + for i, r in enumerate(references_of(record)): + if r.get("rel") != "approval-outcome": + continue + resolver = r.get("resolver") + if not isinstance(resolver, str) or not resolver: + return [], f"approval-outcome reference {i} names no resolver" + out.append(resolver) + return out, None + + +def resolver_of(record) -> str | None: + """The first approval-outcome resolver, for display; trusted_origin checks all of them.""" + resolvers, _ = approval_resolvers(record) + return resolvers[0] if resolvers else None + + +def record_kid(record) -> str | None: + cnf = record.get("cnf") if isinstance(record, dict) else None + jwk = cnf.get("jwk") if isinstance(cnf, dict) else None + kid = jwk.get("kid") if isinstance(jwk, dict) else None + return kid if isinstance(kid, str) and kid else None + + +def anchor_origin(record: dict, registry: str | None) -> tuple[str | None, str | None]: + """(origin, refusal): the origin the published key may come from, or why there is none. The origin is the relying + party's, given out of band with --registry; nothing in the record supplies it. (An earlier head derived it from + the record's did:web subject, which only shows the record agrees with itself: a record signed under a publisher + of its own choosing names that publisher in its subject too. Review of #338, 18 Oct 2026.) The record's `subject` + and `resolver` must then agree with the origin, which catches a record that was minted for another registry.""" + origin, why = https_origin(registry) + if why: + return None, why + # the subject must be a did:web under the trusted origin: this registry mints no other kind, and a record whose + # subject is absent, unparseable or of another method cannot be bound to the origin at all (review of #338) + subject = record.get("subject") + subject_origin = origin_of_did_web(subject) + if not subject_origin: + return None, "the record's subject is not a did:web identifier, so it cannot be bound to the trusted origin" + if subject_origin != origin: + return None, f"the record's subject lives under {subject_origin}, not the trusted origin {origin}" + # every approval-outcome reference must resolve at the trusted origin, whatever its position; compared in canonical + # form, so an explicit :443 or a different case is the same origin (RFC 6454) and a resolver with a path is not one + resolvers, why = approval_resolvers(record) + if why: + return None, why + for i, resolver in enumerate(resolvers): + canonical, bad = https_origin(resolver) + if bad: + return None, f"approval-outcome reference {i} resolver {resolver!r} is not an https origin: {bad}" + if canonical != origin: + return None, f"approval-outcome reference {i} resolves at {canonical}, not the trusted origin {origin}" + return origin, None + + +def verify_with_key_source(record: dict, pinned: dict | None, fetch=published_key, + max_age_seconds: int = MAX_AGE_SECONDS, now: int | Instant | None = None, registry: str | None = None) -> dict: + instant = _instant(now) # frozen here, once; the same object reaches verification() and every verdict below + bounds = {"max_age_seconds": max_age_seconds, **instant.fields()} + if not isinstance(record, dict): + return {"status": "invalid", "reason": f"record is not a JSON object ({type(record).__name__})", "key_source": None, "kid": None, "revocation": None, **bounds} + kid = record_kid(record) # always the key the RECORD names; the key actually used is trusted_key_thumbprint, and pinned_kid when pinned + # an origin that is given is enforced, pinned key or not (review of 09c436b, 18 Oct 2026: with --jwk the + # registry was only parsed, so a record for another registry verified under the pinned key with --registry + # naming a third). The caller said which registry the record must belong to; the record must agree. + origin = None + if registry is not None: + origin, refusal = anchor_origin(record, registry) + if refusal: + return {"status": "no-key", "key_source": None, "kid": kid, "revocation": None, **bounds, "reason": refusal} + if pinned is not None: + if isinstance(pinned, dict) and "d" in pinned: + # 19 Oct 2026: a trust anchor is a PUBLIC key; a file carrying the private member is refused, and no part of it + # is echoed (the reason is fixed text) + return {"status": "no-key", "key_source": None, "kid": kid, "revocation": None, **bounds, "reason": "the pinned key file holds a private key (member d); pin the public key only"} + if not is_ed25519_jwk(pinned): + return {"status": "no-key", "key_source": None, "kid": kid, "revocation": None, **bounds, "reason": "the pinned key is not an Ed25519 JWK (kty OKP, crv Ed25519, x of 32 bytes)"} + pk = pinned.get("kid") + return {**verification(record, pinned, max_age_seconds, instant), "key_source": "pinned", "kid": kid, "pinned_kid": pk if isinstance(pk, str) else None, + **({"trusted_origin": origin} if origin else {}), **bounds} + if origin is None: + return {"status": "no-key", "key_source": None, "kid": kid, "revocation": None, **bounds, + "reason": "no trust anchor: pass --registry , or --jwk "} + if kid: + key = fetch(origin, kid) + if key: + return {**verification(record, key, max_age_seconds, instant), "key_source": "published", "kid": kid, "trusted_origin": origin, **bounds} + return {"status": "no-key", "key_source": None, "kid": kid, "revocation": None, "trusted_origin": origin, **bounds, + "reason": f"no Ed25519 key {kid!r} published at {origin}/.well-known/xchk-keys.json (or pass --jwk)"} + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description="Verify an xChk registry's Trust Record against a pinned or published key.") + ap.add_argument("record") + ap.add_argument("--jwk", help="a JSON file with trace_signer_jwk (or a bare JWK): the pinned key to verify with") + ap.add_argument("--max-age", type=int, default=MAX_AGE_SECONDS, metavar="SECONDS", + help=f"how old the record's iat may be (default {MAX_AGE_SECONDS}, section 3.2.2); must be positive") + ap.add_argument("--registry", metavar="ORIGIN", default=None, + help="the https origin of the registry you trust, whose published key verifies the record; required unless " + "--jwk pins the key; when given, with or without --jwk, the record's subject and every approval-outcome " + "resolver must be under it") + ap.add_argument("--now", type=int, default=None, metavar="EPOCH_SECONDS", + help="verify as of this moment instead of the clock: a retained record re-verified at its own time") + a = ap.parse_args(argv) + if a.max_age <= 0: + ap.error("--max-age must be positive; the age check cannot be disabled") + if a.now is not None and a.now < 0: + ap.error("--now must be a Unix time in seconds") + instant = Instant(a.now) # the one instant for everything below + record, why = load_json(a.record) + if why: + out = {"status": "invalid", "reason": f"record file: {why}", "key_source": None, "kid": None, "revocation": None, "max_age_seconds": a.max_age, **instant.fields()} + else: + jwk = None + if a.jwk: + j, why = load_json(a.jwk, detail=False) + if why: + out = {"status": "no-key", "reason": f"pinned key file: {why}", "key_source": None, "kid": record_kid(record), "revocation": None, "max_age_seconds": a.max_age, **instant.fields()} + print(json.dumps(out, indent=2)) + return 1 + jwk = j.get("trace_signer_jwk", j) if isinstance(j, dict) else j # a bare JWK, or {trace_signer_jwk: ...}; the shape is checked next + if not is_ed25519_jwk(jwk) and not (isinstance(jwk, dict) and "d" in jwk): + jwk = {"not": "a jwk"} # verify_with_key_source says so, with the reason; never None (that would mean "no pin given") + out = verify_with_key_source(record, jwk, max_age_seconds=a.max_age, now=instant, registry=a.registry) + out["approval_outcome_references"] = [ + {"id": r.get("id"), "resolver": r.get("resolver"), "digest": r.get("digest")} + for r in references_of(record) if r.get("rel") == "approval-outcome" + ] + print(json.dumps(out, indent=2)) + return 0 if out["status"] == "verified" else 1 + + +def load_json(path: str, detail: bool = True) -> tuple: + """(value, None) or (None, why): a local file that is not JSON is reported, not raised. detail=False reports the + error class only: a decode error's message can quote bytes of the file, and a key file's bytes are not echoed.""" + try: + return json.loads(Path(path).read_text(encoding="utf-8")), None + except (OSError, ValueError, UnicodeDecodeError) as e: + return None, f"{type(e).__name__}: {str(e)[:120]}" if detail else type(e).__name__ + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/marketplace/catalog.json b/marketplace/catalog.json index 07de0e0..a2bc5a8 100644 --- a/marketplace/catalog.json +++ b/marketplace/catalog.json @@ -1,6 +1,6 @@ { "catalog_version": 1, - "count": 47, + "count": 48, "integrations": [ { "name": "Claude Code", @@ -1075,6 +1075,29 @@ "key-release", "evidence" ] + }, + { + "name": "xChk registry", + "package_name": "xChk registry", + "vendor": "xChk", + "description": "A software-only Level 0 Trust Record per run from an agent-licensing registry, with one approval-outcome reference per act a person was asked about.", + "path": "integrations/xchk-registry", + "url": "https://github.com/agentrust-io/integrations/tree/main/integrations/xchk-registry", + "homepage": "https://in.xchk.io", + "repository": "https://github.com/agentrust-io/integrations", + "tier": "community", + "stack": [ + "TRACE" + ], + "category": "Evidence & receipts", + "mark": "XC", + "featured": null, + "keywords": [ + "trust-record", + "approval-outcome", + "registry", + "level-0" + ] } ] }