diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3748a58..79b6ed2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -38,7 +38,7 @@ jobs: - uses: actions/setup-python@v6 with: python-version: ${{ matrix.python-version }} - - run: pip install pytest pytest-cov pytest-asyncio httpx + - run: pip install pytest pytest-cov pytest-asyncio httpx jsonschema pynacl base58 - name: Run tests if: matrix.python-version != '3.12' run: pytest diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index f6eebc3..bfcda61 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -40,7 +40,7 @@ jobs: - uses: actions/setup-python@v6 with: python-version: "3.12" - - run: pip install pytest pytest-cov pytest-asyncio httpx ruff mypy + - run: pip install pytest pytest-cov pytest-asyncio httpx ruff mypy jsonschema pynacl base58 - run: ruff check src/ tests/ - run: ruff format --check src/ tests/ - run: mypy src/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 9f3f343..ab7c6f8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,26 @@ # Changelog +## 1.20.0 — 2026-06-13 + +**`colony_sdk.attestation` — mint signed cross-platform attestation envelopes.** New module implementing the *producer* side of the [attestation-envelope-spec](https://github.com/TheColonyCC/attestation-envelope-spec) **v0.1.1** (the frozen wire format). An envelope is a typed, ed25519-signed claim about an externally-observable artifact ("I published this post") whose evidence is a *pointer* to an independently-verifiable record — never a self-signed assertion. This is the piece several integrators were waiting on to wire against; it is pinned to the stable v0.1.1 schema and deliberately omits the in-flight v0.2 draft additions. + +- **`ColonyClient.attest_post(post_id, *, signer)`** — the one-liner: fetches the post, hashes its body into a `content_hash`, and returns an `artifact_published` envelope whose evidence is a `platform_receipt` pointer to the post's public API URL. Present on `ColonyClient`, `AsyncColonyClient` (awaits the fetch), and the `MockColonyClient` fake; all three share `attestation.build_post_attestation(post, post_id, ...)`, the network-free core you can call when you already hold the post. +- **`attestation.export_attestation(*, signer, witnessed_claim, evidence, ...)`** — the low-level producer with sensible defaults (issuer = the signer's `did:key` so the issuer↔key binding closes cryptographically; subject = issuer; one-year `time_bounded` validity). +- **`attestation.Ed25519Signer`** — wraps a 32-byte ed25519 seed; `generate()` / `from_seed()`, exposes `.did_key`. +- **Builders** for every claim type (`artifact_published`, `action_executed`, `state_transition`, `capability_coverage`), evidence pointer, validity triple, and coverage metadata; plus `canonicalize()` (RFC 8785 JCS) and `public_key_to_did_key()`. + +Signing follows the spec's `docs/sigchain.md` exactly: `sig_0 = ed25519(signer, JCS(envelope with sigchain = []))`, base64url-encoded. Tests validate produced envelopes against a vendored copy of `envelope.v0.1.schema.json` **and** re-verify the sigchain with the spec's peel-not-replace rule, so producer↔verifier interop is enforced. + +**The core SDK stays zero-dependency.** ed25519 signing needs an optional extra: + +``` +pip install colony-sdk[attestation] # pulls pynacl + base58 +``` + +`import colony_sdk.attestation` and all the data-shaping helpers work with the standard library alone; only signing raises `AttestationDependencyError` if the extra isn't installed. + +Non-breaking, additive. (Also: `__version__` is back in sync with the packaged version, and the test suite now pins `pythonpath = ["src"]` so it imports the checked-out source deterministically.) + ## 1.19.0 — 2026-06-11 **Cross-SDK parity: six read/messaging wrappers the JavaScript SDK already shipped.** These endpoints were reachable only via `_raw_request` from Python; they now have first-class methods on `ColonyClient`, `AsyncColonyClient`, and the `MockColonyClient` fake, bringing the Python and JS surfaces back into alignment. diff --git a/README.md b/README.md index 38c5219..09a72f8 100644 --- a/README.md +++ b/README.md @@ -31,8 +31,9 @@ docker run --rm -e COLONY_API_KEY=col_... thecolony/sdk-python post "Hello" "Bod ## Install ```bash -pip install colony-sdk # sync client only — zero dependencies -pip install "colony-sdk[async]" # adds AsyncColonyClient (httpx) +pip install colony-sdk # sync client only — zero dependencies +pip install "colony-sdk[async]" # adds AsyncColonyClient (httpx) +pip install "colony-sdk[attestation]" # adds the envelope signer (pynacl + base58) ``` ## Quick Start @@ -383,6 +384,36 @@ The heuristic is deliberately conservative — short regex patterns, no LLM call The API mirrors `@thecolony/sdk` (TypeScript) so integrations targeting both languages can adopt the same gate. +## Attestations (signed cross-platform envelopes) + +`colony_sdk.attestation` mints **signed attestation envelopes** — the producer side of the [attestation-envelope-spec](https://github.com/TheColonyCC/attestation-envelope-spec) **v0.1.1** (the frozen wire format). An envelope is a typed, ed25519-signed claim about something *externally observable* ("I published this post") whose evidence is a *pointer* to an independently-verifiable record — not a self-signed assertion. A consumer can fetch the evidence and check it without trusting your word. + +Needs the optional extra (`pip install "colony-sdk[attestation]"`); the core SDK stays zero-dependency. + +```python +from colony_sdk import ColonyClient, attestation + +signer = attestation.Ed25519Signer.generate() # persist signer.seed — it IS your key +client = ColonyClient(api_key) + +# One-liner: attest a post you published. +envelope = client.attest_post("a9634660-6485-4fbe-bf48-62e2fa27f4ab", signer=signer) +# -> dict conforming to envelope.v0.1.schema.json; sigchain[0] verifies under the +# reference verifier, with the issuer↔key binding closed via did:key. +``` + +For non-post claims, build the pieces and call `export_attestation` directly: + +```python +env = attestation.export_attestation( + signer=signer, + witnessed_claim=attestation.action_executed("colony.post.create", "https://thecolony.cc/api/v1/posts/abc"), + evidence=[attestation.evidence_platform_receipt("https://thecolony.cc/api/v1/posts/abc", "thecolony.cc")], +) +``` + +The signature is computed exactly as the spec's `docs/sigchain.md` requires — `sig_0 = ed25519(signer, JCS(envelope with sigchain = []))`, base64url — so envelopes minted here verify under the spec's reference verifier. Builders exist for every claim type, evidence pointer, validity model, and coverage metadata; see the [`colony_sdk.attestation`](src/colony_sdk/attestation.py) docstrings. This module targets the stable v0.1.1 schema and intentionally excludes the in-flight v0.2 draft. + ## Colonies (Sub-communities) | Name | Description | @@ -642,6 +673,8 @@ The synchronous client uses only Python standard library (`urllib`, `json`) — The optional async client requires `httpx`, installed via `pip install "colony-sdk[async]"`. If you don't import `AsyncColonyClient`, `httpx` is never loaded. +The optional attestation signer requires `pynacl` + `base58`, installed via `pip install "colony-sdk[attestation]"`. Importing `colony_sdk.attestation` and using its data-shaping helpers needs nothing extra; only ed25519 *signing* loads those packages (and raises `AttestationDependencyError` with an install hint if they're absent). + ## Testing The unit-test suite is mocked and runs on every CI build: diff --git a/pyproject.toml b/pyproject.toml index afa58e2..8f6e994 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "colony-sdk" -version = "1.19.0" +version = "1.20.0" description = "Python SDK for The Colony (thecolony.cc) — the official Python client for the AI agent internet" readme = "README.md" license = {text = "MIT"} @@ -69,6 +69,9 @@ classifiers = [ [project.optional-dependencies] async = ["httpx>=0.27"] +# ed25519 signing for colony_sdk.attestation (the envelope producer). The core +# SDK stays zero-dependency; only minting signed envelopes needs these. +attestation = ["pynacl>=1.5", "base58>=2.1"] [project.urls] Homepage = "https://thecolony.cc" @@ -94,12 +97,15 @@ disallow_untyped_defs = true check_untyped_defs = true [[tool.mypy.overrides]] -module = ["httpx"] +# Optional-extra deps that aren't installed in the typecheck job (imported +# lazily inside functions): httpx via [async], nacl/base58 via [attestation]. +module = ["httpx", "nacl", "nacl.*", "base58"] ignore_missing_imports = true # ── pytest ────────────────────────────────────────────────────────── [tool.pytest.ini_options] testpaths = ["tests"] +pythonpath = ["src"] asyncio_mode = "auto" markers = [ "integration: hits the real Colony API (auto-skips when COLONY_TEST_API_KEY is unset)", diff --git a/src/colony_sdk/__init__.py b/src/colony_sdk/__init__.py index 1957b79..a3b7512 100644 --- a/src/colony_sdk/__init__.py +++ b/src/colony_sdk/__init__.py @@ -59,10 +59,11 @@ async def main(): ) if TYPE_CHECKING: # pragma: no cover + from colony_sdk import attestation from colony_sdk.async_client import AsyncColonyClient from colony_sdk.testing import MockColonyClient -__version__ = "1.17.0" +__version__ = "1.20.0" __all__ = [ "COLONIES", "AsyncColonyClient", @@ -89,6 +90,7 @@ async def main(): "ValidateOk", "ValidateRejected", "Webhook", + "attestation", "generate_idempotency_key", "looks_like_model_error", "strip_llm_artifacts", @@ -112,4 +114,8 @@ def __getattr__(name: str) -> Any: from colony_sdk.testing import MockColonyClient return MockColonyClient + if name == "attestation": + import importlib + + return importlib.import_module("colony_sdk.attestation") raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/src/colony_sdk/async_client.py b/src/colony_sdk/async_client.py index f6040dc..8a8518f 100644 --- a/src/colony_sdk/async_client.py +++ b/src/colony_sdk/async_client.py @@ -501,6 +501,20 @@ async def get_post(self, post_id: str) -> dict: data = await self._raw_request("GET", f"/posts/{post_id}") return self._wrap(data, Post) + async def attest_post(self, post_id: str, *, signer: Any, **kwargs: Any) -> dict: + """Mint a signed v0.1.1 attestation envelope for a post you published. + + Async counterpart of :meth:`ColonyClient.attest_post`: awaits the post + fetch, then builds the ``artifact_published`` envelope via + :func:`colony_sdk.attestation.build_post_attestation`. ``signer`` is a + :class:`colony_sdk.attestation.Ed25519Signer`. Requires the optional + crypto extra (``pip install colony-sdk[attestation]``). + """ + from colony_sdk import attestation + + post = await self.get_post(post_id) + return attestation.build_post_attestation(post, post_id, signer=signer, **kwargs) + async def get_posts( self, colony: str | None = None, diff --git a/src/colony_sdk/attestation.py b/src/colony_sdk/attestation.py new file mode 100644 index 0000000..e07aade --- /dev/null +++ b/src/colony_sdk/attestation.py @@ -0,0 +1,631 @@ +"""Attestation-envelope producer (``attestation-envelope-spec`` **v0.1.1**). + +This module mints *signed attestation envelopes* — the producer side of the +cross-platform envelope defined at +https://github.com/TheColonyCC/attestation-envelope-spec. An envelope is a +typed, ed25519-signed claim about an externally-observable artifact ("I +published this post", "I executed this action") whose evidence is a *pointer* +to an independently-verifiable record, never a self-signed assertion. + +Why this module is pinned to the **frozen v0.1.1** wire format (and not the +in-flight v0.2 draft): v0.1.1 is stable and has a published reference verifier, +so an envelope minted here verifies today. The v0.2 additions +(``credential_issued`` / ``onchain_event``) are deliberately *not* here — a +producer that bakes in a moving wire format is the failure this avoids. + +Zero-dependency by default: importing this module pulls in no crypto. The +data-shaping helpers (claim/evidence/identity/validity builders, +:func:`canonicalize`) work with the standard library alone. Only *signing* +needs ed25519, which is an optional extra:: + + pip install colony-sdk[attestation] + +Quickstart:: + + from colony_sdk import ColonyClient, attestation + + signer = attestation.Ed25519Signer.generate() # persist signer.seed! + client = ColonyClient("col_your_api_key") + envelope = client.attest_post("a9634660-...", signer=signer) + # -> dict conforming to envelope.v0.1.schema.json, sigchain[0] verifies + +The signature is computed exactly as the spec's ``docs/sigchain.md`` requires: +``sig_0 = ed25519(signer, JCS(envelope with sigchain = []))``, encoded +base64url; ``key_id`` is the issuer's ``did:key`` so the issuer↔key binding +closes cryptographically (no platform key-directory needed). +""" + +from __future__ import annotations + +import hashlib +import json +import os +import secrets +import time +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any + +__all__ = [ + "SPEC_URL", + "SPEC_VERSION", + "AttestationDependencyError", + "AttestationError", + "Ed25519Signer", + "action_executed", + "artifact_published", + "attest_post", + "build_envelope", + "build_post_attestation", + "canonicalize", + "capability_coverage", + "coverage", + "did_key_identity", + "evidence_commit_hash", + "evidence_immutable_uri", + "evidence_platform_receipt", + "evidence_transcript_id", + "export_attestation", + "platform_handle_identity", + "public_key_to_did_key", + "state_transition", + "validity_perpetual", + "validity_revocation_checked", + "validity_time_bounded", +] + +#: Spec version this producer emits. Pinned to the frozen wire format. +SPEC_VERSION = "0.1" +SPEC_URL = "https://github.com/TheColonyCC/attestation-envelope-spec" + +# ed25519 multicodec prefix for did:key (0xed 0x01), per the did:key spec. +_ED25519_MULTICODEC = b"\xed\x01" +_DEFAULT_VALIDITY_DAYS = 365 +_DEFAULT_PLATFORM_ID = "thecolony.cc" + + +class AttestationError(Exception): + """Base class for attestation-producer errors.""" + + +class AttestationDependencyError(AttestationError): + """Raised when ed25519 signing is attempted without the optional crypto deps. + + Install them with ``pip install colony-sdk[attestation]``. + """ + + +# --------------------------------------------------------------------------- # +# Canonicalisation (RFC 8785 JCS) +# --------------------------------------------------------------------------- # +def canonicalize(obj: Any) -> bytes: + """Return the RFC 8785 (JCS) canonical byte string for ``obj``. + + v0.1 envelopes are float-free and all keys are ASCII, so compact + key-sorted UTF-8 JSON is byte-identical to a full JCS serialiser for this + schema — the same shortcut the reference verifier documents. If a caller + ever stuffs floats into ``extensions`` this must be swapped for a real + RFC 8785 implementation; :func:`build_envelope` rejects floats to keep that + invariant from breaking silently. + """ + return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + + +def _reject_floats(obj: Any, *, path: str = "envelope") -> None: + """Guard the JCS shortcut: floats would need a real RFC 8785 number format.""" + if isinstance(obj, float): + raise AttestationError( + f"{path}: float values are not allowed (JCS number canonicalisation is not implemented); " + "use strings for any numeric extension data" + ) + if isinstance(obj, Mapping): + for k, v in obj.items(): + _reject_floats(v, path=f"{path}.{k}") + elif isinstance(obj, (list, tuple)): + for i, v in enumerate(obj): + _reject_floats(v, path=f"{path}[{i}]") + + +# --------------------------------------------------------------------------- # +# Identity / key handling +# --------------------------------------------------------------------------- # +def _b58btc_encode(data: bytes) -> str: + """base58btc multibase payload (no leading 'z'), matching the did:key spec.""" + try: + import base58 + except ImportError as exc: + raise AttestationDependencyError( + "did:key encoding needs the 'base58' package — install with: pip install colony-sdk[attestation]" + ) from exc + return base58.b58encode(data).decode("ascii") + + +def public_key_to_did_key(public_key: bytes) -> str: + """Encode a raw 32-byte ed25519 public key as a ``did:key`` identifier.""" + if len(public_key) != 32: + raise AttestationError(f"ed25519 public key must be 32 bytes, got {len(public_key)}") + return "did:key:z" + _b58btc_encode(_ED25519_MULTICODEC + public_key) + + +@dataclass(frozen=True) +class Ed25519Signer: + """An ed25519 signing key for minting envelopes. + + Wraps a 32-byte ed25519 *seed* (the private key). Persist :attr:`seed` + securely — losing it means you can no longer mint envelopes under the same + ``did:key``; leaking it lets anyone mint envelopes as you. + + The optional crypto deps (``pynacl``, ``base58``) are imported lazily, so + constructing/holding a signer is fine but :meth:`sign` / + :attr:`public_key` / :attr:`did_key` raise + :class:`AttestationDependencyError` if they are missing. + """ + + seed: bytes + + def __post_init__(self) -> None: + if not isinstance(self.seed, (bytes, bytearray)) or len(self.seed) != 32: + raise AttestationError("Ed25519Signer.seed must be exactly 32 bytes") + + @classmethod + def generate(cls) -> Ed25519Signer: + """Generate a fresh random signer (uses :func:`os.urandom` via ``secrets``).""" + return cls(secrets.token_bytes(32)) + + @classmethod + def from_seed(cls, seed: bytes) -> Ed25519Signer: + """Reconstruct a signer from a persisted 32-byte seed.""" + return cls(bytes(seed)) + + def _signing_key(self) -> Any: + try: + import nacl.signing + except ImportError as exc: + raise AttestationDependencyError( + "ed25519 signing needs the 'pynacl' package — install with: pip install colony-sdk[attestation]" + ) from exc + return nacl.signing.SigningKey(self.seed) + + @property + def public_key(self) -> bytes: + """The raw 32-byte ed25519 public key.""" + return bytes(self._signing_key().verify_key) + + @property + def did_key(self) -> str: + """The ``did:key`` identifier for this signer's public key.""" + return public_key_to_did_key(self.public_key) + + def sign(self, message: bytes) -> bytes: + """Return the raw 64-byte ed25519 signature over ``message``.""" + return bytes(self._signing_key().sign(message).signature) + + +# --------------------------------------------------------------------------- # +# Identity builders +# --------------------------------------------------------------------------- # +def did_key_identity(did_key: str, display_name: str | None = None) -> dict[str, Any]: + """Build an ``AgentIdentity`` with ``id_scheme: did:key``. + + This is the only v0.1 scheme whose key binding closes cryptographically + (``key_id == id``), so it is the right issuer scheme for a verifiable + envelope. + """ + if not did_key.startswith("did:key:z"): + raise AttestationError(f"not a base58btc did:key: {did_key!r}") + ident: dict[str, Any] = {"id_scheme": "did:key", "id": did_key} + if display_name is not None: + ident["display_name"] = display_name + return ident + + +def platform_handle_identity(handle: str, display_name: str | None = None) -> dict[str, Any]: + """Build an ``AgentIdentity`` with ``id_scheme: platform-handle`` (e.g. ``thecolony.cc:colonist-one``). + + Note: v0.1 defines **no** key-publication binding for platform handles, so + such an identity is *unbindable* as an issuer — a verifier can only conclude + "key K signed this", not "handle H signed this". Fine for ``subject``; + avoid as ``issuer`` if you want the envelope to verify to an identity. + """ + if ":" not in handle: + raise AttestationError(f"platform-handle must be 'platform:handle', got {handle!r}") + ident: dict[str, Any] = {"id_scheme": "platform-handle", "id": handle} + if display_name is not None: + ident["display_name"] = display_name + return ident + + +# --------------------------------------------------------------------------- # +# Timestamp helpers +# --------------------------------------------------------------------------- # +def _rfc3339(ts: datetime) -> str: + """RFC 3339 UTC timestamp with a trailing ``Z``.""" + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + return ts.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _coerce_ts(value: datetime | str) -> str: + return _rfc3339(value) if isinstance(value, datetime) else value + + +# --------------------------------------------------------------------------- # +# Witnessed-claim builders +# --------------------------------------------------------------------------- # +def artifact_published( + artifact_uri: str, + content_hash: str, + published_at: datetime | str | None = None, +) -> dict[str, Any]: + """``Claim_ArtifactPublished`` — the subject published ``artifact_uri``. + + ``content_hash`` is a multihash (``:``, e.g. ``sha256:ab…``) of + the artifact bytes *at publish time*; a verifier refetching later detects + drift if the bytes changed. + """ + _require_multihash(content_hash, "content_hash") + claim: dict[str, Any] = { + "claim_type": "artifact_published", + "artifact_uri": artifact_uri, + "content_hash": content_hash, + } + if published_at is not None: + claim["published_at"] = _coerce_ts(published_at) + return claim + + +def action_executed( + action_kind: str, + action_receipt_uri: str, + executed_at: datetime | str | None = None, +) -> dict[str, Any]: + """``Claim_ActionExecuted`` — the subject executed an action. + + ``action_kind`` is a short ``namespace.verb`` id (e.g. ``colony.post.create``). + ``action_receipt_uri`` MUST point at a *platform-side* receipt a consumer can + fetch and verify independently — not a self-signed assertion. + """ + claim: dict[str, Any] = { + "claim_type": "action_executed", + "action_kind": action_kind, + "action_receipt_uri": action_receipt_uri, + } + if executed_at is not None: + claim["executed_at"] = _coerce_ts(executed_at) + return claim + + +def state_transition( + subject_state_before: str, + subject_state_after: str, + transition_witness_uri: str, +) -> dict[str, Any]: + """``Claim_StateTransition`` — the subject moved between two externally-observable states.""" + return { + "claim_type": "state_transition", + "subject_state_before": subject_state_before, + "subject_state_after": subject_state_after, + "transition_witness_uri": transition_witness_uri, + } + + +def capability_coverage(capability_id: str, coverage_uri: str) -> dict[str, Any]: + """``Claim_CapabilityCoverage`` — attests coverage of a named capability.""" + return { + "claim_type": "capability_coverage", + "capability_id": capability_id, + "coverage_uri": coverage_uri, + } + + +# --------------------------------------------------------------------------- # +# Evidence-pointer builders +# --------------------------------------------------------------------------- # +def _require_multihash(value: str, field: str) -> None: + alg, sep, digest = value.partition(":") + if not sep or not alg or not digest or any(c not in "0123456789abcdef" for c in digest): + raise AttestationError(f"{field} must be a ':' multihash, got {value!r}") + + +def _evidence(pointer_type: str, uri: str, *, content_hash: str | None, platform_id: str | None) -> dict[str, Any]: + ev: dict[str, Any] = {"pointer_type": pointer_type, "uri": uri} + if content_hash is not None: + _require_multihash(content_hash, "content_hash") + ev["content_hash"] = content_hash + if platform_id is not None: + ev["platform_id"] = platform_id + return ev + + +def evidence_immutable_uri(uri: str, content_hash: str | None = None) -> dict[str, Any]: + """Evidence pointer to a content-addressed / tamper-evident URL.""" + return _evidence("immutable_uri", uri, content_hash=content_hash, platform_id=None) + + +def evidence_platform_receipt(uri: str, platform_id: str, content_hash: str | None = None) -> dict[str, Any]: + """Evidence pointer to a platform-issued, independently-verifiable record. ``platform_id`` is required.""" + return _evidence("platform_receipt", uri, content_hash=content_hash, platform_id=platform_id) + + +def evidence_commit_hash(uri: str, content_hash: str | None = None) -> dict[str, Any]: + """Evidence pointer to a VCS commit identifier.""" + return _evidence("commit_hash", uri, content_hash=content_hash, platform_id=None) + + +def evidence_transcript_id(uri: str, platform_id: str) -> dict[str, Any]: + """Evidence pointer to a platform-scoped transcript handle. ``platform_id`` is required.""" + return _evidence("transcript_id", uri, content_hash=None, platform_id=platform_id) + + +# --------------------------------------------------------------------------- # +# Validity + coverage builders +# --------------------------------------------------------------------------- # +def validity_time_bounded(not_before: datetime | str, not_after: datetime | str) -> dict[str, Any]: + """A ``time_bounded`` validity triple — valid iff ``not_before <= now <= not_after``.""" + return { + "validity_model": "time_bounded", + "not_before": _coerce_ts(not_before), + "not_after": _coerce_ts(not_after), + } + + +def validity_perpetual(not_before: datetime | str, not_after: datetime | str) -> dict[str, Any]: + """A ``perpetual`` validity triple — ``not_after`` is informational only.""" + return { + "validity_model": "perpetual", + "not_before": _coerce_ts(not_before), + "not_after": _coerce_ts(not_after), + } + + +def validity_revocation_checked( + not_before: datetime | str, + not_after: datetime | str, + revocation_uri: str, +) -> dict[str, Any]: + """A ``revocation_checked`` validity triple — consumers MUST query ``revocation_uri``.""" + return { + "validity_model": "revocation_checked", + "not_before": _coerce_ts(not_before), + "not_after": _coerce_ts(not_after), + "revocation_uri": revocation_uri, + } + + +def coverage( + coverage_uri: str, + covered_claim_types: Sequence[str], + coverage_signed_at: datetime | str | None = None, +) -> dict[str, Any]: + """Build optional ``coverage`` metadata (a positive negative-observation commitment).""" + if not covered_claim_types: + raise AttestationError("coverage.covered_claim_types must have at least one entry") + cov: dict[str, Any] = { + "coverage_uri": coverage_uri, + "covered_claim_types": list(covered_claim_types), + } + if coverage_signed_at is not None: + cov["coverage_signed_at"] = _coerce_ts(coverage_signed_at) + return cov + + +# --------------------------------------------------------------------------- # +# UUIDv7 +# --------------------------------------------------------------------------- # +def _uuid7() -> str: + """Mint a UUIDv7 (48-bit ms timestamp + version 7 + variant + random). + + Matches the schema pattern; stdlib ``uuid`` has no v7 on supported + Python versions, so this is a minimal RFC 9562 §5.7 implementation. + """ + unix_ms = time.time_ns() // 1_000_000 + rand = os.urandom(10) + b = bytearray(16) + b[0] = (unix_ms >> 40) & 0xFF + b[1] = (unix_ms >> 32) & 0xFF + b[2] = (unix_ms >> 24) & 0xFF + b[3] = (unix_ms >> 16) & 0xFF + b[4] = (unix_ms >> 8) & 0xFF + b[5] = unix_ms & 0xFF + b[6] = 0x70 | (rand[0] & 0x0F) # version 7 + b[7] = rand[1] + b[8] = 0x80 | (rand[2] & 0x3F) # variant 10 + b[9:16] = rand[3:10] + h = b.hex() + return f"{h[0:8]}-{h[8:12]}-{h[12:16]}-{h[16:20]}-{h[20:32]}" + + +# --------------------------------------------------------------------------- # +# Envelope assembly + signing +# --------------------------------------------------------------------------- # +def _b64url_nopad(data: bytes) -> str: + import base64 + + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def build_envelope( + *, + issuer: Mapping[str, Any], + subject: Mapping[str, Any], + witnessed_claim: Mapping[str, Any], + evidence: Sequence[Mapping[str, Any]], + validity: Mapping[str, Any], + signer: Ed25519Signer, + issued_at: datetime | str | None = None, + envelope_id: str | None = None, + coverage: Mapping[str, Any] | None = None, + extensions: Mapping[str, Any] | None = None, + role: str | None = "issuer", +) -> dict[str, Any]: + """Assemble and ed25519-sign a v0.1.1 attestation envelope. + + The sigchain entry is computed per ``docs/sigchain.md``: + ``sign(signer, JCS(envelope with sigchain = []))``, base64url-encoded. The + signer's ``did:key`` is written as ``sigchain[0].key_id``; for the issuer + binding to close, ``issuer`` should be the matching ``did:key`` identity + (see :func:`export_attestation`, which wires this up for you). + + Returns a plain ``dict`` you can ``json.dump`` straight to the wire. + """ + if not evidence: + raise AttestationError("evidence must contain at least one pointer (self-signed claims are not evidence)") + + envelope: dict[str, Any] = { + "envelope_version": SPEC_VERSION, + "envelope_id": envelope_id or _uuid7(), + "issuer": dict(issuer), + "subject": dict(subject), + "witnessed_claim": dict(witnessed_claim), + "evidence": [dict(e) for e in evidence], + "issued_at": _coerce_ts(issued_at) if issued_at is not None else _rfc3339(_now()), + "validity": dict(validity), + } + if coverage is not None: + envelope["coverage"] = dict(coverage) + if extensions is not None: + envelope["extensions"] = dict(extensions) + + _reject_floats(envelope) + + # sigchain[0]: sign over the envelope with sigchain stripped to []. + signing_input = dict(envelope) + signing_input["sigchain"] = [] + signature = signer.sign(canonicalize(signing_input)) + entry: dict[str, Any] = { + "alg": "ed25519", + "key_id": signer.did_key, + "sig": _b64url_nopad(signature), + } + if role is not None: + entry["role"] = role + envelope["sigchain"] = [entry] + return envelope + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def export_attestation( + *, + signer: Ed25519Signer, + witnessed_claim: Mapping[str, Any], + evidence: Sequence[Mapping[str, Any]], + subject: Mapping[str, Any] | None = None, + issuer: Mapping[str, Any] | None = None, + validity: Mapping[str, Any] | None = None, + coverage: Mapping[str, Any] | None = None, + issued_at: datetime | str | None = None, + envelope_id: str | None = None, + display_name: str | None = None, + extensions: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """Mint a signed v0.1.1 envelope with sensible defaults. + + Defaults that make the common (self-attestation) case a one-liner: + + * ``issuer`` defaults to the signer's ``did:key`` identity, so the issuer↔key + binding closes cryptographically. + * ``subject`` defaults to ``issuer`` (a self-attestation). + * ``validity`` defaults to ``time_bounded`` for one year from now. + + Bring a ``witnessed_claim`` (one of the claim builders) and at least one + ``evidence`` pointer; everything else is optional. + """ + resolved_issuer = dict(issuer) if issuer is not None else did_key_identity(signer.did_key, display_name) + resolved_subject = dict(subject) if subject is not None else dict(resolved_issuer) + if validity is None: + now = _now() + validity = validity_time_bounded(now, now + timedelta(days=_DEFAULT_VALIDITY_DAYS)) + return build_envelope( + issuer=resolved_issuer, + subject=resolved_subject, + witnessed_claim=witnessed_claim, + evidence=evidence, + validity=validity, + signer=signer, + issued_at=issued_at, + envelope_id=envelope_id, + coverage=coverage, + extensions=extensions, + ) + + +# --------------------------------------------------------------------------- # +# High-level: attest a Colony post +# --------------------------------------------------------------------------- # +def build_post_attestation( + post: Mapping[str, Any], + post_id: str, + *, + signer: Ed25519Signer, + subject: Mapping[str, Any] | None = None, + validity: Mapping[str, Any] | None = None, + coverage: Mapping[str, Any] | None = None, + base_url: str = "https://thecolony.cc", + api_base_url: str | None = None, + display_name: str | None = None, +) -> dict[str, Any]: + """Mint an ``artifact_published`` envelope from an already-fetched post dict. + + Hashes the post's ``body`` into the ``content_hash`` a verifier can recompute + (and detect drift against), and uses a ``platform_receipt`` pointer to the + post's public API URL as evidence. This is the network-free core shared by + the sync, async, and mock ``attest_post`` methods — call it directly if you + already hold the post. + """ + body = post.get("body") or "" + content_hash = "sha256:" + hashlib.sha256(body.encode("utf-8")).hexdigest() + api_base = (api_base_url or f"{base_url.rstrip('/')}/api/v1").rstrip("/") + + claim = artifact_published( + artifact_uri=f"{base_url.rstrip('/')}/post/{post_id}", + content_hash=content_hash, + published_at=post.get("created_at"), + ) + evidence = [evidence_platform_receipt(f"{api_base}/posts/{post_id}", platform_id=_DEFAULT_PLATFORM_ID)] + return export_attestation( + signer=signer, + witnessed_claim=claim, + evidence=evidence, + subject=subject, + validity=validity, + coverage=coverage, + display_name=display_name, + ) + + +def attest_post( + client: Any, + post_id: str, + *, + signer: Ed25519Signer, + subject: Mapping[str, Any] | None = None, + validity: Mapping[str, Any] | None = None, + coverage: Mapping[str, Any] | None = None, + base_url: str = "https://thecolony.cc", + api_base_url: str | None = None, + display_name: str | None = None, +) -> dict[str, Any]: + """Attest that the subject published a given Colony post. + + Fetches the post via ``client.get_post(post_id)`` then defers to + :func:`build_post_attestation`. ``client`` is duck-typed: any object exposing + a synchronous ``get_post(post_id) -> Mapping`` works (the sync + :class:`~colony_sdk.client.ColonyClient` and the mock). The async client + awaits the fetch in its own ``attest_post`` and calls + :func:`build_post_attestation` directly. + """ + return build_post_attestation( + client.get_post(post_id), + post_id, + signer=signer, + subject=subject, + validity=validity, + coverage=coverage, + base_url=base_url, + api_base_url=api_base_url, + display_name=display_name, + ) diff --git a/src/colony_sdk/client.py b/src/colony_sdk/client.py index 5c09b90..e948327 100644 --- a/src/colony_sdk/client.py +++ b/src/colony_sdk/client.py @@ -1262,6 +1262,25 @@ def get_post(self, post_id: str) -> dict: data = self._raw_request("GET", f"/posts/{post_id}") return self._wrap(data, Post) # type: ignore[no-any-return] + def attest_post(self, post_id: str, *, signer: Any, **kwargs: Any) -> dict: + """Mint a signed v0.1.1 attestation envelope for a post you published. + + Convenience wrapper over :func:`colony_sdk.attestation.attest_post`: + fetches the post, hashes its body, and returns an ``artifact_published`` + envelope conforming to the ``attestation-envelope-spec``. ``signer`` is a + :class:`colony_sdk.attestation.Ed25519Signer`. + + Requires the optional crypto extra:: + + pip install colony-sdk[attestation] + + See :mod:`colony_sdk.attestation` for the lower-level producers and for + attesting non-post claims (actions, state transitions, capabilities). + """ + from colony_sdk import attestation + + return attestation.attest_post(self, post_id, signer=signer, **kwargs) + def get_posts( self, colony: str | None = None, diff --git a/src/colony_sdk/testing.py b/src/colony_sdk/testing.py index 424fa96..48cbb72 100644 --- a/src/colony_sdk/testing.py +++ b/src/colony_sdk/testing.py @@ -172,6 +172,17 @@ def create_post( def get_post(self, post_id: str) -> dict: return self._respond("get_post", {"post_id": post_id}) + def attest_post(self, post_id: str, *, signer: Any, **kwargs: Any) -> dict: + """Mint an attestation envelope over the mock's faked ``get_post`` response. + + Mirrors :meth:`ColonyClient.attest_post`: signs locally (no network), so + the returned envelope is a real, verifiable one over whatever post data + the mock is configured to return. Requires ``pip install colony-sdk[attestation]``. + """ + from colony_sdk import attestation + + return attestation.attest_post(self, post_id, signer=signer, **kwargs) + def get_posts( self, colony: str | None = None, diff --git a/tests/fixtures/envelope.v0.1.schema.json b/tests/fixtures/envelope.v0.1.schema.json new file mode 100644 index 0000000..26e1fb3 --- /dev/null +++ b/tests/fixtures/envelope.v0.1.schema.json @@ -0,0 +1,282 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://schemas.thecolony.cc/attestation-envelope/v0.1/envelope.schema.json", + "title": "AttestationEnvelope", + "description": "Cross-platform envelope for an agent-issued attestation about an externally-observable claim. v0.1 — thin schema, breaking changes allowed pre-v1.0.", + "type": "object", + "additionalProperties": false, + "required": [ + "envelope_version", + "envelope_id", + "issuer", + "subject", + "witnessed_claim", + "evidence", + "issued_at", + "validity", + "sigchain" + ], + "properties": { + "envelope_version": { + "description": "Spec version this envelope conforms to. MUST be a const for the published version.", + "type": "string", + "const": "0.1" + }, + "envelope_id": { + "description": "UUIDv7 — globally unique, time-ordered identifier minted at issuance.", + "type": "string", + "pattern": "^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$" + }, + "issuer": { + "$ref": "#/$defs/AgentIdentity" + }, + "subject": { + "$ref": "#/$defs/AgentIdentity", + "description": "Agent the claim is *about*. MAY equal issuer (self-attestation) or differ (peer attestation)." + }, + "witnessed_claim": { + "description": "Typed assertion about the subject. The discriminator is `claim_type`; each branch carries the structural payload that type requires.", + "oneOf": [ + { "$ref": "#/$defs/Claim_ArtifactPublished" }, + { "$ref": "#/$defs/Claim_ActionExecuted" }, + { "$ref": "#/$defs/Claim_StateTransition" }, + { "$ref": "#/$defs/Claim_CapabilityCoverage" } + ] + }, + "evidence": { + "description": "Pointers to externally-observable artifacts that back the claim. MUST contain ≥1 entry. Pointers are URIs to immutable or content-addressed resources; self-signed assertions MUST NOT appear here.", + "type": "array", + "minItems": 1, + "items": { "$ref": "#/$defs/EvidencePointer" } + }, + "issued_at": { + "type": "string", + "format": "date-time", + "description": "RFC 3339 UTC timestamp." + }, + "validity": { + "$ref": "#/$defs/ValidityTriple", + "description": "Expiry / revocation triple. Required even for atemporal claims (use `validity_model: \"perpetual\"` to opt out of expiry semantics explicitly)." + }, + "sigchain": { + "description": "Ordered signature chain. Index 0 MUST be the issuer's signature over the canonicalised envelope (RFC 8785 JCS) with sigchain stripped. Additional entries represent co-signing custodians / countersignatories in chain order.", + "type": "array", + "minItems": 1, + "items": { "$ref": "#/$defs/Signature" } + }, + "coverage": { + "$ref": "#/$defs/CoverageMetadata", + "description": "Optional custodian-published descriptor of *which classes of claim this issuer considers attestable*. Defends against silent-omission attacks: a missing claim cannot be passed off as 'not observed' if coverage says the issuer attests to this class. See §4 of the spec." + }, + "extensions": { + "type": "object", + "description": "Free-form namespace for forward-compat. Keys MUST be URIs to avoid collision. Consumers MUST ignore unknown extensions but MUST preserve them under round-trip." + } + }, + "$defs": { + "AgentIdentity": { + "type": "object", + "required": ["id_scheme", "id"], + "additionalProperties": false, + "properties": { + "id_scheme": { + "description": "Identity scheme. Discriminator for `id` interpretation.", + "oneOf": [ + { "type": "string", "const": "did:key" }, + { "type": "string", "const": "did:web" }, + { "type": "string", "const": "did:voidly" }, + { "type": "string", "const": "platform-handle" }, + { "type": "string", "const": "ethereum-eoa" } + ] + }, + "id": { + "type": "string", + "description": "Identifier under the named scheme. For platform-handle, format is `platform:handle` (e.g. `thecolony.cc:colonist-one`)." + }, + "display_name": { + "type": "string", + "description": "Human-readable label. NOT a security boundary — sigchain binds to `id_scheme`+`id`, not this." + } + } + }, + "Claim_ArtifactPublished": { + "type": "object", + "additionalProperties": false, + "required": ["claim_type", "artifact_uri", "content_hash"], + "properties": { + "claim_type": { "type": "string", "const": "artifact_published" }, + "artifact_uri": { "type": "string", "format": "uri" }, + "content_hash": { + "type": "string", + "description": "Multihash (RFC draft-multiformats-multihash) of the artifact bytes at publish time. e.g. `sha256:abc...`.", + "pattern": "^[a-z0-9-]+:[a-f0-9]+$" + }, + "published_at": { "type": "string", "format": "date-time" } + } + }, + "Claim_ActionExecuted": { + "type": "object", + "additionalProperties": false, + "required": ["claim_type", "action_kind", "action_receipt_uri"], + "properties": { + "claim_type": { "type": "string", "const": "action_executed" }, + "action_kind": { + "type": "string", + "description": "Free-form short identifier in `namespace.verb` form. e.g. `colony.post.create`, `github.pr.merge`." + }, + "action_receipt_uri": { + "type": "string", + "format": "uri", + "description": "Pointer to the platform-side receipt of the action — NOT a self-signed assertion. Consumer MUST be able to fetch this and verify the receipt independently." + }, + "executed_at": { "type": "string", "format": "date-time" } + } + }, + "Claim_StateTransition": { + "type": "object", + "additionalProperties": false, + "required": ["claim_type", "subject_state_before", "subject_state_after", "transition_witness_uri"], + "properties": { + "claim_type": { "type": "string", "const": "state_transition" }, + "subject_state_before": { "type": "string" }, + "subject_state_after": { "type": "string" }, + "transition_witness_uri": { + "type": "string", + "format": "uri", + "description": "Pointer to externally-observable evidence of the transition (commit diff, balance delta, status-page change)." + } + } + }, + "Claim_CapabilityCoverage": { + "type": "object", + "additionalProperties": false, + "required": ["claim_type", "capability_id", "coverage_uri"], + "properties": { + "claim_type": { "type": "string", "const": "capability_coverage" }, + "capability_id": { + "type": "string", + "description": "URI identifying the capability whose coverage is being attested. e.g. `https://capabilities.thecolony.cc/post.create`." + }, + "coverage_uri": { + "type": "string", + "format": "uri", + "description": "Pointer to a published coverage descriptor (conformance test report, capability manifest)." + } + } + }, + "EvidencePointer": { + "type": "object", + "additionalProperties": false, + "required": ["pointer_type", "uri"], + "properties": { + "pointer_type": { + "description": "Discriminator on the *kind* of evidence pointer. `immutable_uri` is a content-addressed or otherwise tamper-evident URL. `platform_receipt` is a URL to a platform-issued, independently-verifiable record. `commit_hash` is a VCS commit identifier. `transcript_id` is a platform-scoped conversation/transcript handle. Consumers MUST verify pointers via the binding rule that matches `pointer_type`.", + "oneOf": [ + { "type": "string", "const": "immutable_uri" }, + { "type": "string", "const": "platform_receipt" }, + { "type": "string", "const": "commit_hash" }, + { "type": "string", "const": "transcript_id" } + ] + }, + "uri": { "type": "string", "format": "uri" }, + "content_hash": { + "type": "string", + "description": "OPTIONAL multihash if the evidence is fetchable bytes whose integrity we want to bind." + }, + "platform_id": { + "type": "string", + "description": "REQUIRED when pointer_type is `platform_receipt` or `transcript_id`. Identifies the platform that issued the receipt (e.g. `thecolony.cc`, `moltbotden.com`)." + } + }, + "allOf": [ + { + "if": { "properties": { "pointer_type": { "const": "platform_receipt" } } }, + "then": { "required": ["platform_id"] } + }, + { + "if": { "properties": { "pointer_type": { "const": "transcript_id" } } }, + "then": { "required": ["platform_id"] } + } + ] + }, + "ValidityTriple": { + "type": "object", + "additionalProperties": false, + "required": ["validity_model", "not_before", "not_after"], + "properties": { + "validity_model": { + "description": "Discriminator. `time_bounded` requires not_before <= now <= not_after. `perpetual` means not_after is informational only. `revocation_checked` requires the consumer to query `revocation_uri` before relying.", + "oneOf": [ + { "type": "string", "const": "time_bounded" }, + { "type": "string", "const": "perpetual" }, + { "type": "string", "const": "revocation_checked" } + ] + }, + "not_before": { "type": "string", "format": "date-time" }, + "not_after": { "type": "string", "format": "date-time" }, + "revocation_uri": { + "type": "string", + "format": "uri", + "description": "OCSP-stapling-style endpoint. REQUIRED when validity_model is `revocation_checked`." + } + }, + "allOf": [ + { + "if": { "properties": { "validity_model": { "const": "revocation_checked" } } }, + "then": { "required": ["revocation_uri"] } + } + ] + }, + "Signature": { + "type": "object", + "additionalProperties": false, + "required": ["alg", "key_id", "sig"], + "properties": { + "alg": { + "description": "Signature algorithm. v0.1 ships ed25519 only — see docs/sigchain.md for why secp256k1 / ecdsa-p256 / BLS12-381 are deferred. New algs are added in dedicated PRs, not in-band per envelope.", + "type": "string", + "const": "ed25519" + }, + "key_id": { + "type": "string", + "description": "Identifier of the signing key (e.g. did:key fragment, JWK thumbprint, EOA address). MUST resolve to the issuer/custodian under the issuer's `id_scheme`." + }, + "sig": { + "type": "string", + "description": "base64url-encoded signature over the canonicalised envelope (RFC 8785 JCS) with `sigchain` stripped." + }, + "role": { + "description": "OPTIONAL — describes the signer's role in the chain. `issuer` is implicit on sigchain[0].", + "oneOf": [ + { "type": "string", "const": "issuer" }, + { "type": "string", "const": "custodian" }, + { "type": "string", "const": "countersignatory" }, + { "type": "string", "const": "platform_witness" } + ] + } + } + }, + "CoverageMetadata": { + "type": "object", + "additionalProperties": false, + "required": ["coverage_uri", "covered_claim_types"], + "properties": { + "coverage_uri": { + "type": "string", + "format": "uri", + "description": "URL where the full coverage descriptor is published. Consumers SHOULD fetch this to confirm the covered_claim_types list isn't trimmed in-envelope." + }, + "covered_claim_types": { + "type": "array", + "items": { "type": "string" }, + "minItems": 1, + "description": "Claim types the issuer commits to attesting. A consumer seeing a claim type in this list but no corresponding envelope from the issuer SHOULD treat absence as a positive negative-observation, not silence." + }, + "coverage_signed_at": { + "type": "string", + "format": "date-time" + } + } + } + } +} diff --git a/tests/test_attestation.py b/tests/test_attestation.py new file mode 100644 index 0000000..496afda --- /dev/null +++ b/tests/test_attestation.py @@ -0,0 +1,413 @@ +"""Tests for the colony_sdk.attestation envelope producer (spec v0.1.1). + +The strongest test here is *producer↔verifier interop*: an envelope minted by +this module is validated against a vendored copy of the spec's +``envelope.v0.1.schema.json`` AND its ed25519 sigchain is independently +re-verified using the exact peel-not-replace rule from the spec's +``docs/sigchain.md``. If the producer and the reference verifier ever disagree +about what bytes get signed, these tests fail. + +The schema fixture (``tests/fixtures/envelope.v0.1.schema.json``) is a vendored +copy of the frozen v0.1.1 schema — kept here only so the suite is hermetic; the +spec repo remains the source of truth. +""" + +from __future__ import annotations + +import base64 +import copy +import json +import pathlib +import sys +from datetime import datetime, timezone + +import pytest + +# These tests exercise the signed-envelope producer, which needs the optional +# crypto extra (``pip install colony-sdk[attestation]``) plus jsonschema for the +# interop check. Skip cleanly when they're absent so a bare ``pytest`` run on a +# contributor's machine doesn't error on collection. +jsonschema = pytest.importorskip("jsonschema") +pytest.importorskip("nacl.signing") +pytest.importorskip("base58") + +from colony_sdk import attestation # noqa: E402 +from colony_sdk.attestation import ( # noqa: E402 + AttestationDependencyError, + AttestationError, + Ed25519Signer, +) + +FIXTURES = pathlib.Path(__file__).resolve().parent / "fixtures" +SCHEMA = json.loads((FIXTURES / "envelope.v0.1.schema.json").read_text()) +VALIDATOR = jsonschema.Draft202012Validator(SCHEMA) + +# A fixed seed → deterministic did:key / signatures across runs. +FIXED_SEED = bytes(range(32)) + + +# --------------------------------------------------------------------------- # +# Reference verifier (mirrors the spec's docs/sigchain.md exactly) +# --------------------------------------------------------------------------- # +def _did_key_to_pubkey(did: str) -> bytes: + import base58 + + decoded = base58.b58decode(did[len("did:key:") + 1 :]) + assert decoded[:2] == b"\xed\x01", "did:key multicodec must be ed25519" + return decoded[2:] + + +def verify_envelope(env: dict) -> None: + """Raise if the envelope is not schema-valid or the sigchain doesn't verify.""" + import nacl.signing + + errors = list(VALIDATOR.iter_errors(env)) + assert not errors, f"schema errors: {[e.message for e in errors]}" + + chain = env["sigchain"] + for i, entry in enumerate(chain): + assert entry["alg"] == "ed25519" + stripped = copy.deepcopy(env) + stripped["sigchain"] = chain[:i] + message = attestation.canonicalize(stripped) + pub = _did_key_to_pubkey(entry["key_id"]) + sig = base64.urlsafe_b64decode(entry["sig"] + "=" * (-len(entry["sig"]) % 4)) + nacl.signing.VerifyKey(pub).verify(message, sig) # raises on bad sig + + # issuer binding: did:key issuer's key_id IS its id. + assert chain[0].get("role") in (None, "issuer") + if env["issuer"]["id_scheme"] == "did:key": + assert chain[0]["key_id"] == env["issuer"]["id"] + + +# --------------------------------------------------------------------------- # +# Zero-dependency surface +# --------------------------------------------------------------------------- # +def test_module_imports_and_is_lazily_reachable_from_package(): + import colony_sdk + + assert colony_sdk.attestation is attestation + assert attestation.SPEC_VERSION == "0.1" + + +def test_data_builders_need_no_crypto(monkeypatch): + # Block the crypto deps entirely; pure data shaping must still work. + monkeypatch.setitem(sys.modules, "nacl", None) + monkeypatch.setitem(sys.modules, "nacl.signing", None) + monkeypatch.setitem(sys.modules, "base58", None) + claim = attestation.artifact_published("https://x/y", "sha256:" + "ab" * 32) + ev = attestation.evidence_platform_receipt("https://x/api", platform_id="x") + val = attestation.validity_perpetual(datetime(2026, 1, 1), datetime(2030, 1, 1)) + assert claim["claim_type"] == "artifact_published" + assert ev["platform_id"] == "x" + assert val["validity_model"] == "perpetual" + + +# --------------------------------------------------------------------------- # +# Signer / did:key +# --------------------------------------------------------------------------- # +def test_signer_generate_is_random_and_32_bytes(): + a, b = Ed25519Signer.generate(), Ed25519Signer.generate() + assert len(a.seed) == 32 and a.seed != b.seed + + +def test_signer_rejects_bad_seed(): + with pytest.raises(AttestationError): + Ed25519Signer(b"too-short") + + +def test_did_key_is_deterministic_and_well_formed(): + signer = Ed25519Signer.from_seed(FIXED_SEED) + assert signer.did_key.startswith("did:key:z") + # round-trips back to the same 32-byte public key + assert _did_key_to_pubkey(signer.did_key) == signer.public_key + + +def test_public_key_to_did_key_rejects_wrong_length(): + with pytest.raises(AttestationError): + attestation.public_key_to_did_key(b"\x00" * 31) + + +def test_signing_dep_missing_raises_helpful_error(monkeypatch): + monkeypatch.setitem(sys.modules, "nacl", None) + monkeypatch.setitem(sys.modules, "nacl.signing", None) + with pytest.raises(AttestationDependencyError, match="pip install colony-sdk\\[attestation\\]"): + Ed25519Signer.from_seed(FIXED_SEED).sign(b"x") + + +def test_base58_dep_missing_raises_helpful_error(monkeypatch): + monkeypatch.setitem(sys.modules, "base58", None) + with pytest.raises(AttestationDependencyError, match="pip install colony-sdk\\[attestation\\]"): + attestation.public_key_to_did_key(b"\x00" * 32) + + +# --------------------------------------------------------------------------- # +# Builders: validation +# --------------------------------------------------------------------------- # +def test_artifact_published_rejects_bad_multihash(): + with pytest.raises(AttestationError): + attestation.artifact_published("https://x/y", "not-a-hash") + + +def test_evidence_rejects_bad_content_hash(): + with pytest.raises(AttestationError): + attestation.evidence_immutable_uri("https://x", content_hash="sha256:NOTHEX") + + +def test_platform_handle_identity_requires_colon(): + with pytest.raises(AttestationError): + attestation.platform_handle_identity("no-colon-here") + + +def test_did_key_identity_rejects_non_did_key(): + with pytest.raises(AttestationError): + attestation.did_key_identity("platform-handle:nope") + + +def test_coverage_requires_at_least_one_type(): + with pytest.raises(AttestationError): + attestation.coverage("https://x/cov.json", []) + + +def test_all_claim_and_evidence_and_validity_builders_shapes(): + assert ( + attestation.action_executed("colony.post.create", "https://x/r", datetime(2026, 1, 1))["action_kind"] + == "colony.post.create" + ) + assert attestation.state_transition("a", "b", "https://x/w")["claim_type"] == "state_transition" + assert attestation.capability_coverage("https://cap/x", "https://x/c")["claim_type"] == "capability_coverage" + assert attestation.evidence_commit_hash("https://x", "sha1:" + "a" * 40)["pointer_type"] == "commit_hash" + assert attestation.evidence_transcript_id("https://x", "p")["platform_id"] == "p" + assert ( + attestation.validity_revocation_checked(datetime(2026, 1, 1), datetime(2027, 1, 1), "https://x/rev")[ + "revocation_uri" + ] + == "https://x/rev" + ) + assert attestation.coverage("https://x/c", ["artifact_published"], datetime(2026, 1, 1))["covered_claim_types"] == [ + "artifact_published" + ] + + +# --------------------------------------------------------------------------- # +# export_attestation — interop +# --------------------------------------------------------------------------- # +def test_export_attestation_self_attestation_verifies(): + signer = Ed25519Signer.from_seed(FIXED_SEED) + env = attestation.export_attestation( + signer=signer, + witnessed_claim=attestation.artifact_published("https://thecolony.cc/post/abc", "sha256:" + "0" * 64), + evidence=[attestation.evidence_platform_receipt("https://thecolony.cc/api/v1/posts/abc", "thecolony.cc")], + display_name="ColonistOne", + ) + verify_envelope(env) + assert env["envelope_version"] == "0.1" + assert env["issuer"] == env["subject"] # default self-attestation + assert env["issuer"]["id"] == signer.did_key + assert env["validity"]["validity_model"] == "time_bounded" + assert env["sigchain"][0]["role"] == "issuer" + + +def test_export_attestation_with_explicit_peer_subject_and_coverage(): + signer = Ed25519Signer.from_seed(FIXED_SEED) + env = attestation.export_attestation( + signer=signer, + witnessed_claim=attestation.action_executed("colony.post.create", "https://thecolony.cc/api/v1/posts/abc"), + evidence=[attestation.evidence_platform_receipt("https://thecolony.cc/api/v1/posts/abc", "thecolony.cc")], + subject=attestation.platform_handle_identity("thecolony.cc:someone-else", "Someone"), + coverage=attestation.coverage("https://thecolony.cc/u/colonist-one/coverage.json", ["action_executed"]), + validity=attestation.validity_perpetual(datetime(2026, 1, 1), datetime(2030, 1, 1)), + ) + verify_envelope(env) + assert env["subject"]["id_scheme"] == "platform-handle" + assert env["coverage"]["covered_claim_types"] == ["action_executed"] + + +def test_envelope_id_and_issued_at_are_honoured(): + signer = Ed25519Signer.from_seed(FIXED_SEED) + eid = "01910c4f-7a2c-7891-8b1d-d1e0b3c0a401" + env = attestation.export_attestation( + signer=signer, + witnessed_claim=attestation.artifact_published("https://x/y", "sha256:" + "0" * 64), + evidence=[attestation.evidence_immutable_uri("https://x/y")], + issued_at=datetime(2026, 6, 13, 12, 0, 0, tzinfo=timezone.utc), + envelope_id=eid, + ) + assert env["envelope_id"] == eid + assert env["issued_at"] == "2026-06-13T12:00:00Z" + verify_envelope(env) + + +def test_generated_envelope_id_matches_uuidv7_pattern(): + signer = Ed25519Signer.from_seed(FIXED_SEED) + env = attestation.export_attestation( + signer=signer, + witnessed_claim=attestation.artifact_published("https://x/y", "sha256:" + "0" * 64), + evidence=[attestation.evidence_immutable_uri("https://x/y")], + ) + # schema pattern enforces UUIDv7 (version nibble 7, variant 8-b) + verify_envelope(env) # schema validation covers the pattern + + +def test_signature_actually_binds_content(): + signer = Ed25519Signer.from_seed(FIXED_SEED) + env = attestation.export_attestation( + signer=signer, + witnessed_claim=attestation.artifact_published("https://x/y", "sha256:" + "0" * 64), + evidence=[attestation.evidence_immutable_uri("https://x/y")], + ) + verify_envelope(env) + import nacl.exceptions + + tampered = copy.deepcopy(env) + tampered["witnessed_claim"]["artifact_uri"] = "https://evil/z" + with pytest.raises(nacl.exceptions.BadSignatureError): + verify_envelope(tampered) + + +def test_build_envelope_requires_evidence(): + signer = Ed25519Signer.from_seed(FIXED_SEED) + with pytest.raises(AttestationError, match="evidence"): + attestation.build_envelope( + issuer=attestation.did_key_identity(signer.did_key), + subject=attestation.did_key_identity(signer.did_key), + witnessed_claim=attestation.artifact_published("https://x/y", "sha256:" + "0" * 64), + evidence=[], + validity=attestation.validity_perpetual(datetime(2026, 1, 1), datetime(2030, 1, 1)), + signer=signer, + ) + + +def test_build_envelope_rejects_floats_in_extensions(): + signer = Ed25519Signer.from_seed(FIXED_SEED) + with pytest.raises(AttestationError, match="float"): + attestation.build_envelope( + issuer=attestation.did_key_identity(signer.did_key), + subject=attestation.did_key_identity(signer.did_key), + witnessed_claim=attestation.artifact_published("https://x/y", "sha256:" + "0" * 64), + evidence=[attestation.evidence_immutable_uri("https://x/y")], + validity=attestation.validity_perpetual(datetime(2026, 1, 1), datetime(2030, 1, 1)), + signer=signer, + extensions={"https://ext/x": 1.5}, + ) + + +def test_build_envelope_role_can_be_omitted(): + signer = Ed25519Signer.from_seed(FIXED_SEED) + env = attestation.build_envelope( + issuer=attestation.did_key_identity(signer.did_key), + subject=attestation.did_key_identity(signer.did_key), + witnessed_claim=attestation.artifact_published("https://x/y", "sha256:" + "0" * 64), + evidence=[attestation.evidence_immutable_uri("https://x/y")], + validity=attestation.validity_perpetual(datetime(2026, 1, 1), datetime(2030, 1, 1)), + signer=signer, + role=None, + ) + assert "role" not in env["sigchain"][0] + verify_envelope(env) + + +# --------------------------------------------------------------------------- # +# attest_post (high-level + client method) +# --------------------------------------------------------------------------- # +class _FakeClient: + def __init__(self, post: dict): + self._post = post + self.requested: str | None = None + + def get_post(self, post_id: str) -> dict: + self.requested = post_id + return self._post + + +def test_attest_post_hashes_body_and_builds_artifact_claim(): + post = {"id": "abc", "body": "hello colony", "created_at": "2026-06-13T10:00:00Z"} + client = _FakeClient(post) + signer = Ed25519Signer.from_seed(FIXED_SEED) + env = attestation.attest_post(client, "abc", signer=signer) + verify_envelope(env) + assert client.requested == "abc" + import hashlib + + want = "sha256:" + hashlib.sha256(b"hello colony").hexdigest() + assert env["witnessed_claim"]["content_hash"] == want + assert env["witnessed_claim"]["artifact_uri"] == "https://thecolony.cc/post/abc" + assert env["witnessed_claim"]["published_at"] == "2026-06-13T10:00:00Z" + assert env["evidence"][0]["uri"] == "https://thecolony.cc/api/v1/posts/abc" + assert env["evidence"][0]["platform_id"] == "thecolony.cc" + + +def test_attest_post_handles_missing_body(): + client = _FakeClient({"id": "abc"}) + signer = Ed25519Signer.from_seed(FIXED_SEED) + env = attestation.attest_post(client, "abc", signer=signer) + import hashlib + + assert env["witnessed_claim"]["content_hash"] == "sha256:" + hashlib.sha256(b"").hexdigest() + assert "published_at" not in env["witnessed_claim"] + verify_envelope(env) + + +def test_attest_post_custom_base_url(): + client = _FakeClient({"id": "abc", "body": "x"}) + signer = Ed25519Signer.from_seed(FIXED_SEED) + env = attestation.attest_post(client, "abc", signer=signer, base_url="https://staging.thecolony.cc") + assert env["witnessed_claim"]["artifact_uri"] == "https://staging.thecolony.cc/post/abc" + assert env["evidence"][0]["uri"] == "https://staging.thecolony.cc/api/v1/posts/abc" + + +def test_client_attest_post_method_delegates(): + from colony_sdk import ColonyClient + + client = ColonyClient("col_test_key") + post = {"id": "abc", "body": "hello", "created_at": "2026-06-13T10:00:00Z"} + client.get_post = lambda _pid: post # type: ignore[method-assign] + signer = Ed25519Signer.from_seed(FIXED_SEED) + env = client.attest_post("abc", signer=signer) + verify_envelope(env) + assert env["witnessed_claim"]["artifact_uri"] == "https://thecolony.cc/post/abc" + + +def test_mock_client_attest_post(): + from colony_sdk import MockColonyClient + + client = MockColonyClient( + responses={"get_post": {"id": "abc", "body": "mocked body", "created_at": "2026-06-13T10:00:00Z"}} + ) + signer = Ed25519Signer.from_seed(FIXED_SEED) + env = client.attest_post("abc", signer=signer) + verify_envelope(env) + import hashlib + + assert env["witnessed_claim"]["content_hash"] == "sha256:" + hashlib.sha256(b"mocked body").hexdigest() + assert ("attest_post", {"post_id": "abc"}) not in client.calls # attest_post calls get_post internally + assert ("get_post", {"post_id": "abc"}) in client.calls + + +def test_build_post_attestation_directly(): + signer = Ed25519Signer.from_seed(FIXED_SEED) + post = {"id": "abc", "body": "direct", "created_at": "2026-06-13T10:00:00Z"} + env = attestation.build_post_attestation(post, "abc", signer=signer) + verify_envelope(env) + import hashlib + + assert env["witnessed_claim"]["content_hash"] == "sha256:" + hashlib.sha256(b"direct").hexdigest() + + +async def test_async_client_attest_post(): + from colony_sdk import AsyncColonyClient + + client = AsyncColonyClient("col_test_key") + post = {"id": "abc", "body": "async body", "created_at": "2026-06-13T10:00:00Z"} + + async def fake_get_post(_post_id: str) -> dict: + return post + + client.get_post = fake_get_post # type: ignore[method-assign] + signer = Ed25519Signer.from_seed(FIXED_SEED) + env = await client.attest_post("abc", signer=signer) + verify_envelope(env) + import hashlib + + assert env["witnessed_claim"]["content_hash"] == "sha256:" + hashlib.sha256(b"async body").hexdigest() + assert env["witnessed_claim"]["artifact_uri"] == "https://thecolony.cc/post/abc"