diff --git a/acestep/provenance/__init__.py b/acestep/provenance/__init__.py new file mode 100644 index 00000000..92e10fe6 --- /dev/null +++ b/acestep/provenance/__init__.py @@ -0,0 +1,55 @@ +"""Local provenance: session logs, C2PA manifests, local signing keys. + +Phase-1 scope of the DEMON provenance architecture (see the +demon-provenance spec, 02-architecture.md §2/§3/§5/§7): + +- :mod:`~acestep.provenance.session_log` — local JSONL session logs + using the same event schema as the (future) cloud ledger, tapped off + the streaming event bus. +- :mod:`~acestep.provenance.keys` — self-signed local ES256 signing + material, claim generator "DEMON (local, self-signed)". +- :mod:`~acestep.provenance.manifest` — C2PA manifest construction + + embedding on the track-asset WAV writers via ``c2pa-python``. +- :mod:`~acestep.provenance.ledger_client` — thin event/slice-hash + POST client, a complete no-op until ``DEMON_LEDGER_URL`` is set. +- :mod:`~acestep.provenance.receipts` — client-side verification of + the ledger's signed receipts (spec 06 §2.4: key-cert chain, JCS + + chain-head recomputation, Ed25519 framing checks). + +Everything here degrades to a no-op (one logged warning, never an +exception on a write path) when the optional ``provenance`` extra +(``c2pa-python`` + ``cryptography``) is not installed. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +_ENV_PROVENANCE_DIR = "ACESTEP_PROVENANCE_DIR" +_DEFAULT_PROVENANCE_DIR = os.path.join( + os.path.expanduser("~"), ".daydream-scope", "provenance", +) + + +def provenance_dir() -> Path: + """Root directory for local provenance state (keys, session logs). + + Resolution order mirrors :func:`acestep.paths.models_dir`: + 1. ACESTEP_PROVENANCE_DIR environment variable + 2. ~/.daydream-scope/provenance + + Deliberately NOT memoized: tests point it at a tmp dir per case, + and it is read only on session start / file sign, never per tick. + """ + return Path(os.environ.get(_ENV_PROVENANCE_DIR, _DEFAULT_PROVENANCE_DIR)) + + +def session_logs_dir() -> Path: + """Directory holding per-session JSONL event logs.""" + return provenance_dir() / "sessions" + + +def keys_dir() -> Path: + """Directory holding the local self-signed signing material.""" + return provenance_dir() / "keys" diff --git a/acestep/provenance/keys.py b/acestep/provenance/keys.py new file mode 100644 index 00000000..d9c2cb45 --- /dev/null +++ b/acestep/provenance/keys.py @@ -0,0 +1,216 @@ +"""Local signing key management for self-signed C2PA manifests. + +On first use, generates an ES256 (P-256) key plus a certificate chain +into :func:`acestep.provenance.keys_dir` and reuses it afterwards. The +claim generator identity is deliberately distinct ("DEMON (local, +self-signed)") so locally signed content can never be confused with +Daydream-signed content (spec 02 §8). + +The chain is a throwaway local mini-CA plus one leaf rather than a +single self-issued cert because c2pa-rs enforces the C2PA claim-signing +certificate profile at sign time and rejects self-issued leafs. The CA +private key is discarded after issuing the leaf, so the "CA" can never +sign anything else; trust-wise this is still a self-signed identity. + +Requires the optional ``cryptography`` dependency (``provenance`` +extra); without it every entry point returns ``None`` after a single +logged warning. +""" + +from __future__ import annotations + +import datetime +import os +from dataclasses import dataclass +from pathlib import Path + +from loguru import logger + +from acestep.provenance import keys_dir + +__all__ = [ + "CLAIM_GENERATOR_NAME", + "SigningMaterial", + "signing_material", +] + +CLAIM_GENERATOR_NAME = "DEMON (local, self-signed)" + +_KEY_FILENAME = "demon-local-es256.key.pem" +_CERTS_FILENAME = "demon-local-es256.certs.pem" +_VALIDITY_DAYS = 3650 + +_warned_unavailable = False + + +@dataclass(frozen=True) +class SigningMaterial: + """PEM material ready to hand to a C2PA signer.""" + + key_pem: bytes + cert_chain_pem: bytes + key_path: Path + cert_path: Path + + +def _cryptography_available() -> bool: + global _warned_unavailable + try: + import cryptography # noqa: F401 + return True + except ImportError: + if not _warned_unavailable: + _warned_unavailable = True + logger.warning( + "provenance signing disabled: `cryptography` is not " + "installed (install the `provenance` extra to enable " + "local C2PA signing)" + ) + return False + + +def _generate(directory: Path) -> SigningMaterial: + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import ec + from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID + + now = datetime.datetime.now(datetime.timezone.utc) + not_before = now - datetime.timedelta(days=1) + not_after = now + datetime.timedelta(days=_VALIDITY_DAYS) + + ca_key = ec.generate_private_key(ec.SECP256R1()) + ca_name = x509.Name([ + x509.NameAttribute(NameOID.COMMON_NAME, "DEMON local provenance CA"), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "DEMON local provenance"), + ]) + ca_cert = ( + x509.CertificateBuilder() + .subject_name(ca_name) + .issuer_name(ca_name) + .public_key(ca_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(not_before) + .not_valid_after(not_after) + .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) + .add_extension( + x509.KeyUsage( + digital_signature=True, content_commitment=False, + key_encipherment=False, data_encipherment=False, + key_agreement=False, key_cert_sign=True, crl_sign=True, + encipher_only=False, decipher_only=False, + ), + critical=True, + ) + .add_extension( + x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()), + critical=False, + ) + .sign(ca_key, hashes.SHA256()) + ) + + leaf_key = ec.generate_private_key(ec.SECP256R1()) + leaf_name = x509.Name([ + x509.NameAttribute(NameOID.COMMON_NAME, CLAIM_GENERATOR_NAME), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "DEMON local provenance"), + ]) + leaf_cert = ( + x509.CertificateBuilder() + .subject_name(leaf_name) + .issuer_name(ca_name) + .public_key(leaf_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(not_before) + .not_valid_after(not_after) + .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) + .add_extension( + x509.KeyUsage( + digital_signature=True, content_commitment=False, + key_encipherment=False, data_encipherment=False, + key_agreement=False, key_cert_sign=False, crl_sign=False, + encipher_only=False, decipher_only=False, + ), + critical=True, + ) + .add_extension( + # The cert profile wants a non-empty EKU; emailProtection is + # what the c2pa-rs test fixtures use for document signing. + x509.ExtendedKeyUsage([ExtendedKeyUsageOID.EMAIL_PROTECTION]), + critical=False, + ) + .add_extension( + x509.SubjectKeyIdentifier.from_public_key(leaf_key.public_key()), + critical=False, + ) + .add_extension( + x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), + critical=False, + ) + .sign(ca_key, hashes.SHA256()) + ) + + key_pem = leaf_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + chain_pem = ( + leaf_cert.public_bytes(serialization.Encoding.PEM) + + ca_cert.public_bytes(serialization.Encoding.PEM) + ) + + directory.mkdir(parents=True, exist_ok=True) + key_path = directory / _KEY_FILENAME + cert_path = directory / _CERTS_FILENAME + key_path.write_bytes(key_pem) + os.chmod(key_path, 0o600) + cert_path.write_bytes(chain_pem) + logger.info( + "provenance signing material generated key={} certs={}", + key_path, cert_path, + ) + return SigningMaterial( + key_pem=key_pem, cert_chain_pem=chain_pem, + key_path=key_path, cert_path=cert_path, + ) + + +def _still_valid(cert_chain_pem: bytes) -> bool: + from cryptography import x509 + + try: + leaf = x509.load_pem_x509_certificate(cert_chain_pem) + except Exception: + return False + now = datetime.datetime.now(datetime.timezone.utc) + return leaf.not_valid_before_utc <= now < leaf.not_valid_after_utc + + +def signing_material(directory: Path | None = None) -> SigningMaterial | None: + """Return the local signing material, generating it on first use. + + ``None`` when ``cryptography`` is unavailable or generation failed; + callers must treat that as "provenance off", never as an error. + """ + if not _cryptography_available(): + return None + d = Path(directory) if directory is not None else keys_dir() + key_path = d / _KEY_FILENAME + cert_path = d / _CERTS_FILENAME + try: + if key_path.is_file() and cert_path.is_file(): + key_pem = key_path.read_bytes() + chain_pem = cert_path.read_bytes() + if _still_valid(chain_pem): + return SigningMaterial( + key_pem=key_pem, cert_chain_pem=chain_pem, + key_path=key_path, cert_path=cert_path, + ) + logger.warning( + "provenance cert at {} invalid or expired; regenerating", + cert_path, + ) + return _generate(d) + except Exception as exc: # noqa: BLE001 — never break a caller's write path + logger.warning("provenance signing material unavailable: {}", exc) + return None diff --git a/acestep/provenance/ledger_client.py b/acestep/provenance/ledger_client.py new file mode 100644 index 00000000..29eedaab --- /dev/null +++ b/acestep/provenance/ledger_client.py @@ -0,0 +1,430 @@ +"""Thin client for the Provenance Ledger service (spec 06 §2.3). + +Best-effort background reporter for one session's **pod** event stream. +It batches events and POSTs them to:: + + POST {DEMON_LEDGER_URL}/sessions/{sessionId}/events + Authorization: Bearer {DEMON_LEDGER_TOKEN} + { "events": [ {seq, type, ts, ppq?, payload}, ... ] } + +and parses the signed per-slice receipts the ledger returns +(``{stream, chain_head, receipts:[{seq, chain_head, ts, receipt_kid, +sig}]}``, spec 06 §2.3). + +Receipts are **verified as they arrive** per the full 06 §2.4 +procedure (key-cert chain, validity window, Ed25519 over the +``ddp-receipt:v1`` framing, and — step 5 — equality with the chain +head this client recomputes from its *own* submitted events via +:class:`~acestep.provenance.receipts.ChainMirror`). Verification is +fail-open: a failure logs one WARNING, bumps +:attr:`~LedgerClient.receipts_unverified` and records +:attr:`~LedgerClient.last_verification_failure`; it never interrupts +reporting or streaming. + +``DEMON_LEDGER_URL`` already includes the ``/v1`` prefix (it is the +``ledgerBaseUrl`` the queue broker hands the pod at session bootstrap, +spec 06 §1). The bearer token is the per-session **pod ledger token** +(``dlt_pod_…``), delivered to the pod in the broker→pod session +assignment; it is supplied here via ``DEMON_LEDGER_TOKEN`` and/or a +constructor param. With **either** the URL or the token unset the whole +client is a **complete no-op**: construction is free, every method +returns immediately, and no thread is started. + +Async-safe by construction: callers (the bus drainer thread, the WS +dispatch thread, async handlers) only ever enqueue onto a bounded +in-memory queue; one daemon worker owns all network I/O and is the sole +assigner of the per-stream ``seq``. Overflow drops the *oldest* +un-flushed event — the local session log, not this client, is the +durable record, and provenance reporting is fail-open for playback +(spec 06 §7). Because ``seq`` is assigned by the worker at flush time +(never at enqueue time), a dropped event never punches a hole in the +sequence: the surviving events stay contiguous from 0 and the ledger's +seq-gap check (spec 06 §2.3) is not tripped by best-effort drops. +""" + +from __future__ import annotations + +import json +import os +import queue +import threading +import time +import urllib.error +import urllib.request +from dataclasses import dataclass, field +from typing import Optional + +from loguru import logger + +from acestep.provenance.receipts import ChainMirror, ReceiptVerifier + +__all__ = [ + "LEDGER_URL_ENV", + "LEDGER_TOKEN_ENV", + "SLICE_POD_HASH_TYPE", + "LedgerReceipt", + "LedgerClient", +] + +LEDGER_URL_ENV = "DEMON_LEDGER_URL" +LEDGER_TOKEN_ENV = "DEMON_LEDGER_TOKEN" + +# This client reports the pod stream (its token is dlt_pod_…, spec 06 §2.1). +_STREAM = "pod" + +# Event type whose ingestion returns a signed receipt (spec 06 §2.3). +SLICE_POD_HASH_TYPE = "slice.pod_hash" + +_QUEUE_MAX = 1024 +_REQUEST_TIMEOUT_S = 5.0 +# Batch caps (spec 06 §2.3: max 500 events / 1 MiB per request). +_BATCH_MAX = 500 +# Flush cadence for non-receipted events: whichever comes first, a +# slice-hash report (flushed immediately for its receipt) or this +# interval (spec 06 §2.3 recommends 2 s). +_FLUSH_INTERVAL_S = 2.0 + + +def _now_ms() -> int: + """Wall-clock ms since epoch — the on-the-wire timestamp form + (spec 06 §0: ISO-8601 is never used on the wire).""" + return int(time.time() * 1000) + + +@dataclass(frozen=True) +class LedgerReceipt: + """Signed per-slice acknowledgement (spec 06 §2.3): an Ed25519 + signature over the ``ddp-receipt:v1`` framing, committing to the + chain head at a given ``seq``. Field names track the wire response + exactly (``sig`` not ``signature``, ``receipt_kid`` not ``key_id``).""" + + seq: Optional[int] = None + chain_head: Optional[str] = None + ts: Optional[int] = None + receipt_kid: Optional[str] = None + sig: Optional[str] = None + raw: dict = field(default_factory=dict) + + @classmethod + def from_response(cls, data: dict) -> Optional["LedgerReceipt"]: + """Parse the latest receipt out of an ingestion response + ``{stream, chain_head, receipts:[...]}``. Returns ``None`` when + the batch produced no receipts (e.g. a batch of only non-slice + events, which are chained but not individually receipted).""" + if not isinstance(data, dict): + return None + receipts = data.get("receipts") + if not isinstance(receipts, list) or not receipts: + return None + last = receipts[-1] + if not isinstance(last, dict): + return None + return cls( + seq=last.get("seq"), + chain_head=last.get("chain_head") or data.get("chain_head"), + ts=last.get("ts"), + receipt_kid=last.get("receipt_kid"), + sig=last.get("sig"), + raw=last, + ) + + +class LedgerClient: + """Fire-and-forget batched event reporter for one session's pod + stream. A no-op unless both a base URL and a bearer token are + configured.""" + + def __init__( + self, + base_url: str | None = None, + *, + session_id: str = "", + token: str | None = None, + ) -> None: + self.base_url = ( + base_url if base_url is not None + else os.environ.get(LEDGER_URL_ENV, "") + ).rstrip("/") + self.token = ( + token if token is not None + else os.environ.get(LEDGER_TOKEN_ENV, "") + ) + self.session_id = session_id + self._last_receipt: Optional[LedgerReceipt] = None + self._chain_head: Optional[str] = None + self._warned = False + self._queue: Optional[queue.Queue] = None + self._worker: Optional[threading.Thread] = None + # Assigned by the worker thread only (single writer): the next + # per-stream seq. Contiguous from 0 by construction. + self._next_seq = 0 + # Receipt verification (spec 06 §2.4), owned by the worker + # thread. The mirror replays our own submitted events through + # the §2.1 chain rules so step 5 checks receipts against *our* + # history, not just their signatures. + self._mirror: Optional[ChainMirror] = None + self._verifier: Optional[ReceiptVerifier] = None + self._mirror_broken: Optional[str] = None + self._verify_warned = False + self.receipts_verified = 0 + self.receipts_unverified = 0 + self.last_verification_failure: Optional[str] = None + + @property + def enabled(self) -> bool: + """Reporting is live only when the pod holds both the ledger URL + and its write-scoped pod token (spec 06 §1).""" + return bool(self.base_url) and bool(self.token) + + @property + def last_receipt(self) -> Optional[LedgerReceipt]: + """Most recent parsed receipt (spec 06 §2.3: the reporter keeps + the latest signed commitment).""" + return self._last_receipt + + @property + def chain_head(self) -> Optional[str]: + """Latest server-reported chain head for this stream.""" + return self._chain_head + + # ---- reporting ------------------------------------------------------- + + def post_event( + self, + type: str, + payload: dict | None = None, + *, + ts: int | None = None, + ppq: float | None = None, + ) -> None: + """Enqueue one authorship / config event (spec 06 §2.2). ``seq`` + is assigned later, by the worker, so best-effort drops never + create gaps.""" + if not self.enabled: + return + self._enqueue(type, payload or {}, ts=ts, ppq=ppq, flush_now=False) + + def post_slice_hash( + self, + *, + sha256: str, + start_sample: int, + num_samples: int, + channels: int, + slice_seq: int | None = None, + mac_verified: bool | None = None, + ts: int | None = None, + ) -> None: + """Enqueue a ``slice.pod_hash`` event (spec 06 §2.3): SHA-256 over + the uncompressed interleaved float16 slice payload bytes, plus the + geometry copied from the WS slice header. Flushed promptly because + every slice-hash event gets a signed receipt.""" + if not self.enabled: + return + payload: dict = { + "start_sample": int(start_sample), + "num_samples": int(num_samples), + "channels": int(channels), + "sha256": sha256, + } + # slice_seq is the pod's monotonic per-session slice counter and + # the join key for pod/client cross-checking (spec 06 §2.3 / §3). + if slice_seq is not None: + payload["slice_seq"] = int(slice_seq) + if mac_verified is not None: + payload["mac_verified"] = bool(mac_verified) + self._enqueue( + SLICE_POD_HASH_TYPE, payload, ts=ts, ppq=None, flush_now=True, + ) + + def close(self, timeout: float = 2.0) -> None: + if self._queue is None: + return + try: + self._queue.put_nowait(None) + except queue.Full: + # Make room for the sentinel so the worker can drain + exit. + try: + self._queue.get_nowait() + self._queue.put_nowait(None) + except (queue.Empty, queue.Full): + pass + if self._worker is not None: + self._worker.join(timeout=timeout) + + # ---- worker ---------------------------------------------------------- + + def _enqueue( + self, + type: str, + payload: dict, + *, + ts: int | None, + ppq: float | None, + flush_now: bool, + ) -> None: + if self._queue is None: + self._queue = queue.Queue(maxsize=_QUEUE_MAX) + self._worker = threading.Thread( + target=self._drain, name="ledger-client", daemon=True, + ) + self._worker.start() + item = { + "type": type, + "ts": _now_ms() if ts is None else int(ts), + "payload": payload, + "_flush": flush_now, + } + if ppq is not None: + item["ppq"] = float(ppq) + try: + self._queue.put_nowait(item) + except queue.Full: + # Drop the oldest un-flushed item to make room (fail-open): + # the local log is the durable record. seq is assigned at + # flush, so the drop leaves the surviving stream contiguous. + try: + self._queue.get_nowait() + self._queue.put_nowait(item) + except (queue.Empty, queue.Full): + pass + + def _drain(self) -> None: + assert self._queue is not None + batch: list[dict] = [] + while True: + timeout = _FLUSH_INTERVAL_S if batch else None + try: + item = self._queue.get(timeout=timeout) + except queue.Empty: + # Idle flush interval elapsed with events pending. + self._flush(batch) + batch = [] + continue + if item is None: + self._flush(batch) + return + flush_now = item.pop("_flush", False) + batch.append(item) + if flush_now or len(batch) >= _BATCH_MAX: + self._flush(batch) + batch = [] + + def _flush(self, batch: list[dict]) -> None: + if not batch: + return + # Assign contiguous seq now (worker is the single seq writer). + events = [] + for item in batch: + event = {"seq": self._next_seq, "type": item["type"], + "ts": item["ts"], "payload": item["payload"]} + if "ppq" in item: + event["ppq"] = item["ppq"] + events.append(event) + self._next_seq += 1 + # Extend the local chain mirror before the POST: it models the + # history we *submit* (spec 06 §2.1: "recompute the chain head + # ... from your own submitted events"). + self._mirror_extend(events) + body = json.dumps({"events": events}, default=str).encode("utf-8") + url = f"{self.base_url}/sessions/{self.session_id}/events" + try: + req = urllib.request.Request( + url, + data=body, + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {self.token}", + }, + method="POST", + ) + with urllib.request.urlopen(req, timeout=_REQUEST_TIMEOUT_S) as resp: + raw = resp.read() + try: + data = json.loads(raw) + except (json.JSONDecodeError, UnicodeDecodeError): + return + if isinstance(data, dict): + head = data.get("chain_head") + if isinstance(head, str): + self._chain_head = head + receipt = LedgerReceipt.from_response(data) + if receipt is not None: + self._last_receipt = receipt + receipts = data.get("receipts") + if isinstance(receipts, list): + for r in receipts: + self._verify_receipt(r) + except (urllib.error.URLError, OSError, ValueError) as exc: + if not self._warned: + self._warned = True + logger.warning( + "ledger post failed url={} error={} " + "(further failures silenced)", + url, exc, + ) + + # ---- receipt verification (spec 06 §2.4) ----------------------------- + + def _mirror_extend(self, events: list[dict]) -> None: + """Replay the outgoing batch through the local chain mirror. + Fail-open: a payload the mirror cannot canonicalize (non-JSON + type, JS-unsafe integer) permanently marks the mirror broken — + subsequent receipts count as unverified with that reason, and + reporting itself is untouched.""" + if self._mirror_broken is not None: + return + if self._mirror is None: + self._mirror = ChainMirror(self.session_id, _STREAM) + try: + for event in events: + self._mirror.append(event) + except Exception as exc: # noqa: BLE001 — never break reporting + self._mirror_broken = f"local chain mirror failed: {exc}" + logger.warning( + "ledger receipt verification degraded for session={}: {}", + self.session_id, self._mirror_broken, + ) + + def _verify_receipt(self, receipt: object) -> None: + """Run the full §2.4 procedure on one arriving receipt and keep + the verified/unverified tallies. Never raises.""" + try: + if self._mirror_broken is not None: + ok, reason = False, self._mirror_broken + elif not isinstance(receipt, dict): + ok, reason = False, "malformed receipt (not an object)" + else: + if self._verifier is None: + self._verifier = ReceiptVerifier(f"{self.base_url}/keys") + seq = receipt.get("seq") + expected = ( + self._mirror.head_at(seq) + if self._mirror is not None and isinstance(seq, int) + else None + ) + ok, reason = self._verifier.verify( + session_id=self.session_id, + stream=_STREAM, + seq=seq, + chain_head=receipt.get("chain_head"), + ts=receipt.get("ts"), + receipt_kid=receipt.get("receipt_kid"), + sig=receipt.get("sig"), + expected_head=expected, + ) + except Exception as exc: # noqa: BLE001 — fail-open by contract + ok, reason = False, f"receipt verification error: {exc}" + if ok: + self.receipts_verified += 1 + return + self.receipts_unverified += 1 + self.last_verification_failure = reason + if not self._verify_warned: + self._verify_warned = True + logger.warning( + "ledger receipt FAILED verification session={}: {} " + "(further failures silenced; see receipts_unverified / " + "last_verification_failure)", + self.session_id, reason, + ) diff --git a/acestep/provenance/manifest.py b/acestep/provenance/manifest.py new file mode 100644 index 00000000..7180262b --- /dev/null +++ b/acestep/provenance/manifest.py @@ -0,0 +1,251 @@ +"""C2PA manifest construction + embedding for local file outputs. + +Implements the manifest schema of spec 02 §3 for the DEMON local +file-writer path (§2 row 3): ``c2pa.created`` action with the IPTC +``digitalSourceType``, the CAWG training-and-data-mining assertion +(``notAllowed`` for all categories — the do-not-train default), and the +custom ``com.daydream.session`` assertion carrying session id, +model/LoRA identifiers, timeline summary counts and the session-log +hash. Signed with the local self-signed material from +:mod:`acestep.provenance.keys`. + +Import-guarded: when ``c2pa-python`` (import name ``c2pa``) or +``cryptography`` is missing, :func:`embed_wav_manifest` degrades to a +no-op with a single logged warning. It also never raises — asset +writing must succeed whether or not it could be signed. +""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any, Optional + +from loguru import logger + +from acestep.provenance import keys as provenance_keys + +__all__ = [ + "TRAINED_ALGORITHMIC_MEDIA", + "COMPOSITE_WITH_TRAINED_ALGORITHMIC_MEDIA", + "DIGITAL_CAPTURE", + "provenance_available", + "build_manifest_definition", + "embed_wav_manifest", +] + +TRAINED_ALGORITHMIC_MEDIA = ( + "http://cv.iptc.org/newscodes/digitalsourcetype/trainedAlgorithmicMedia" +) +COMPOSITE_WITH_TRAINED_ALGORITHMIC_MEDIA = ( + "http://cv.iptc.org/newscodes/digitalsourcetype/" + "compositeWithTrainedAlgorithmicMedia" +) +# Human-origin content (recorded / uploaded, not synthesized). Used for +# the user's own uploaded source track and separated stems: those are +# the dry input, and marking them trained/composite would be an +# affirmatively false synthetic claim (spec 02-architecture §4). +DIGITAL_CAPTURE = ( + "http://cv.iptc.org/newscodes/digitalsourcetype/digitalCapture" +) + +_CAWG_CATEGORIES = ( + "cawg.ai_generative_training", + "cawg.ai_inference", + "cawg.ai_training", + "cawg.data_mining", +) + +_warned_unavailable = False + + +def _import_c2pa() -> Any: + global _warned_unavailable + try: + import c2pa + return c2pa + except ImportError: + if not _warned_unavailable: + _warned_unavailable = True + logger.warning( + "provenance manifests disabled: `c2pa` is not installed " + "(install the `provenance` extra to embed Content " + "Credentials in written audio)" + ) + return None + + +def provenance_available() -> bool: + """True when both the c2pa SDK and local signing material are + usable. Cheap after the first call (imports are cached).""" + return _import_c2pa() is not None and provenance_keys.signing_material() is not None + + +def _generator_version() -> str | None: + try: + from importlib.metadata import version + return version("demon") + except Exception: # noqa: BLE001 — running from a source tree + return None + + +def build_manifest_definition( + *, + title: str, + ingredient_fingerprint: str | None = None, + session: dict | None = None, + source_type: str | None = None, +) -> dict: + """Pure manifest-definition builder (JSON dict for ``c2pa.Builder``). + + ``ingredient_fingerprint`` is the seed audio's ``waveform_sha256`` + fingerprint (:func:`acestep.track_assets.waveform_fingerprint`); + when present the action's ``digitalSourceType`` switches to + ``compositeWithTrainedAlgorithmicMedia`` per spec 02 §3. The seed + reference lives inside ``com.daydream.session`` rather than a + ``c2pa.ingredient`` because a hard ingredient assertion requires + hashing the original asset stream, which these writers no longer + have — the fingerprint is the identity the stem cache already uses. + + ``session`` is a :meth:`SessionLogTap.manifest_summary` dict; when + ``None`` (no live session — offline precompute) the session + assertion still ships with null identity fields so the do-not-train + and AI-disclosure posture is never conditional on a session. + + ``source_type`` overrides the derived digitalSourceType for callers + that know seed audio exists without holding its fingerprint (e.g. + stems separated from a not-yet-persisted source). + """ + if source_type is None: + source_type = ( + COMPOSITE_WITH_TRAINED_ALGORITHMIC_MEDIA + if ingredient_fingerprint + else TRAINED_ALGORITHMIC_MEDIA + ) + generator: dict = {"name": provenance_keys.CLAIM_GENERATOR_NAME} + version = _generator_version() + if version: + generator["version"] = version + session = session or {} + session_data = { + "session_id": session.get("session_id"), + "model": session.get("model"), + "loras": session.get("loras") or [], + "timeline_summary": session.get("timeline_summary") or {}, + "session_log_sha256": session.get("session_log_sha256"), + } + if ingredient_fingerprint: + session_data["seed_waveform_sha256"] = ingredient_fingerprint + return { + "claim_generator_info": [generator], + "title": title, + "assertions": [ + { + "label": "c2pa.actions", + "data": { + "actions": [ + { + "action": "c2pa.created", + "digitalSourceType": source_type, + "softwareAgent": generator, + }, + ], + }, + }, + { + # CAWG Training & Data Mining: notAllowed for every + # category, on by default (spec 02 §3). + "label": "cawg.training-mining", + "data": { + "entries": { + c: {"use": "notAllowed"} for c in _CAWG_CATEGORIES + }, + }, + }, + { + "label": "com.daydream.session", + "data": session_data, + }, + ], + } + + +def _session_summary_for(session_id: str) -> Optional[dict]: + try: + from acestep.provenance.session_log import session_summary_for + return session_summary_for(session_id) + except Exception: # noqa: BLE001 + return None + + +def embed_wav_manifest( + path: Path, + *, + ingredient_fingerprint: str | None = None, + session: dict | None = None, + session_id: str | None = None, + title: str | None = None, + source_type: str | None = None, +) -> bool: + """Embed a signed C2PA manifest into the WAV at ``path``, in place + (write-temp-then-replace, so a failure leaves the unsigned WAV + intact). Returns True on success; NEVER raises. + + The session assertion is bound to the SPECIFIC session that produced + the asset (spec 06 §2.5): pass an explicit ``session`` summary dict, + or a ``session_id`` to resolve the summary of that session's live tap. + When neither is given (pure-local write, or the producing session is + no longer live) the assertion ships with null session fields — the + manifest never borrows an arbitrary "latest" session's identity + (audit F7/G5). + """ + c2pa = _import_c2pa() + if c2pa is None: + return False + material = provenance_keys.signing_material() + if material is None: + return False + path = Path(path) + tmp = path.with_name(path.name + ".c2pa.tmp") + try: + if session is not None: + resolved_session = session + elif session_id is not None: + resolved_session = _session_summary_for(session_id) + else: + resolved_session = None + manifest = build_manifest_definition( + title=title or path.name, + ingredient_fingerprint=ingredient_fingerprint, + session=resolved_session, + source_type=source_type, + ) + # The wrapper's __init__ rejects ta_url=None but the native + # field is nullable; a NULL ta_url means "no timestamp + # authority", which is right for offline local signing (an + # empty-string URL makes the TSA fetch fail the whole sign). + info = c2pa.C2paSignerInfo( + alg=b"es256", + sign_cert=material.cert_chain_pem, + private_key=material.key_pem, + ta_url=b"", + ) + info.ta_url = None + with c2pa.Signer.from_info(info) as signer: + builder = c2pa.Builder(manifest) + # Positional on purpose: Builder.sign's parameters are the + # overload-style ``signer_or_format`` / ``format_or_source`` + # / ``source_or_dest`` names, so keyword calls TypeError. + with open(path, "rb") as src, open(tmp, "wb") as dst: + builder.sign(signer, "audio/wav", src, dst) + os.replace(tmp, path) + return True + except Exception as exc: # noqa: BLE001 — asset write must survive + logger.warning( + "c2pa manifest embed failed path={} error={}", path, exc, + ) + try: + tmp.unlink(missing_ok=True) + except OSError: + pass + return False diff --git a/acestep/provenance/receipts.py b/acestep/provenance/receipts.py new file mode 100644 index 00000000..99f53c71 --- /dev/null +++ b/acestep/provenance/receipts.py @@ -0,0 +1,472 @@ +"""Client-side ledger receipt verification (spec 06 §2.4). + +The Provenance Ledger acknowledges every slice-hash event with a signed +receipt (spec 06 §2.3). A receipt is only worth something if the client +runs the full §2.4 verification procedure — in particular **step 5**: +recompute the chain head from *your own* submitted events and require +the receipt to commit to exactly that head. Signature checks alone only +prove the ledger signed *something*; step 5 proves it signed *your +history*. + +Three pieces, mirroring the server implementation byte-for-byte +(``provenance.keys.ts`` / ``chain.ts`` / ``jcs.ts``): + +- :func:`jcs` — RFC 8785 canonical JSON with ECMAScript + ``JSON.stringify`` number/string formatting, the byte layer under + every ledger hash (spec 06 §0). +- :class:`ChainMirror` — the client's local replica of the §2.1 chain + rules (``event_hash = SHA-256(JCS(event))``, genesis domain-separation + head, ``head_n = SHA-256(head_{n-1} || event_hash_n)``), tracking the + head after every submitted seq. +- :class:`ReceiptVerifier` — fetches + caches ``GET {base}/keys`` + (spec 06 §2.4), validates the key-cert chain (sealing pubkey → + ``ddp-keycert:v1`` cert over the receipt key), and verifies each + receipt's Ed25519 signature over the ``ddp-receipt:v1`` framing plus + the step-5 chain-head equality. + +Number-formatting caveats (JCS = ECMAScript formatting, not Python's): + +- Integers are serialized as plain digits; magnitudes ``>= 2**53`` + raise ``TypeError`` because a JS peer cannot represent them exactly + (the server would hash a rounded value and heads would diverge). +- Floats follow the ECMAScript Number-to-string algorithm: integral + floats drop the ``.0`` (``1.0`` → ``"1"``), ``-0.0`` → ``"0"``, + fixed notation for decimal exponents in ``(-7, 21]`` and exponential + (``1e-7``, ``1e+21``) outside it. Python and JS both emit + shortest-round-trip digits for IEEE-754 doubles, so digit sequences + match; only the notation had to be ported. +- Non-finite numbers and non-JSON types raise ``TypeError``. Callers + are fail-open: a payload the mirror cannot canonicalize marks the + receipt unverifiable, it never breaks streaming. + +Requires the optional ``cryptography`` dependency (``provenance`` +extra) for signature checks; without it verification degrades to a +logged "unverified" outcome, never an exception. +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import math +import time +import urllib.error +import urllib.request +from dataclasses import dataclass +from typing import Optional + +from loguru import logger + +__all__ = [ + "jcs", + "event_hash", + "genesis_head", + "advance_head", + "receipt_message", + "key_cert_message", + "ChainMirror", + "ReceiptKey", + "ReceiptVerifier", +] + +_REQUEST_TIMEOUT_S = 5.0 +# Floor between key-document fetch attempts (receipt keys rotate daily, +# spec 06 §2.4; a fetch happens only when a receipt names an unknown +# kid) so a broken endpoint is not hammered from the flush path. +_KEYS_RETRY_MIN_S = 60.0 + +_MAX_SAFE_INT = 2 ** 53 + + +# --------------------------------------------------------------------------- +# RFC 8785 (JCS) canonicalization — port of the server's jcs.ts +# --------------------------------------------------------------------------- + + +def _format_float(x: float) -> str: + """ECMAScript Number-to-string (ECMA-262 §6.1.6.1.20) for a nonzero, + finite, non-integral-shortcut float. Python's ``repr`` supplies the + same shortest-round-trip digit sequence ES computes; this reshapes + it into ES notation (fixed for decimal exponent in (-7, 21], + ``d[.ddd]e±N`` outside).""" + sign = "-" if x < 0 else "" + r = repr(abs(x)) + if "e" in r: + mantissa, _, exp_s = r.partition("e") + whole, _, frac = mantissa.partition(".") + digits = (whole + frac).rstrip("0") + n = int(exp_s) + len(whole) + else: + whole, _, frac = r.partition(".") + combined = whole + frac + stripped = combined.lstrip("0") + n = len(whole) - (len(combined) - len(stripped)) + digits = stripped.rstrip("0") + k = len(digits) + if k <= n <= 21: + body = digits + "0" * (n - k) + elif 0 < n <= 21: + body = digits[:n] + "." + digits[n:] + elif -6 < n <= 0: + body = "0." + "0" * (-n) + digits + else: + e = n - 1 + exp_part = f"e+{e}" if e >= 0 else f"e-{-e}" + body = digits[0] + ("." + digits[1:] if k > 1 else "") + exp_part + return sign + body + + +def _format_number(value: int | float) -> str: + if isinstance(value, int): + if abs(value) >= _MAX_SAFE_INT: + raise TypeError( + f"JCS: integer {value} exceeds the JS safe-integer range " + "(a JS peer would hash a rounded value)" + ) + return str(value) + if math.isnan(value) or math.isinf(value): + raise TypeError("JCS: non-finite numbers are not serializable") + if value == 0.0: + return "0" # covers -0.0: JSON.stringify(-0) === "0" + if value.is_integer() and abs(value) < _MAX_SAFE_INT: + return str(int(value)) # exact double → same digits, no ".0" + return _format_float(value) + + +def jcs(value: object) -> str: + """RFC 8785 canonical JSON: keys sorted by UTF-16 code units, + numbers/strings serialized exactly as ECMAScript ``JSON.stringify`` + (spec 06 §0). Byte-matches the server's ``jcs.ts``.""" + if value is None: + return "null" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, (int, float)): + return _format_number(value) + if isinstance(value, str): + # json.dumps escapes exactly the JSON.stringify set: `"`, `\`, + # and control chars (short escapes for \b \t \n \f \r). + return json.dumps(value, ensure_ascii=False) + if isinstance(value, (list, tuple)): + return "[" + ",".join(jcs(v) for v in value) + "]" + if isinstance(value, dict): + parts = [] + # UTF-16BE byte order == UTF-16 code-unit order (JCS sort rule; + # differs from code-point order only above the BMP). + for key in sorted(value, key=lambda s: s.encode("utf-16-be")): + if not isinstance(key, str): + raise TypeError("JCS: object keys must be strings") + parts.append( + json.dumps(key, ensure_ascii=False) + ":" + jcs(value[key]) + ) + return "{" + ",".join(parts) + "}" + raise TypeError(f"JCS: cannot serialize a {type(value).__name__}") + + +# --------------------------------------------------------------------------- +# Chain rules (spec 06 §2.1) — port of the server's chain.ts +# --------------------------------------------------------------------------- + + +def event_hash(event: dict) -> bytes: + """``event_hash = SHA-256(JCS(event))`` over the client-supplied + fields only — ``ppq`` enters the canonical form only when the + reporter sent it, matching the server's ``eventHash``.""" + canonical: dict = { + "seq": event["seq"], + "type": event["type"], + "ts": event["ts"], + "payload": event["payload"], + } + if event.get("ppq") is not None: + canonical["ppq"] = event["ppq"] + return hashlib.sha256(jcs(canonical).encode("utf-8")).digest() + + +def genesis_head(session_id: str, stream: str) -> bytes: + """``head_0 = SHA-256(UTF8("ddp:v1:" + sessionId + ":" + stream))``.""" + return hashlib.sha256( + f"ddp:v1:{session_id}:{stream}".encode("utf-8") + ).digest() + + +def advance_head(prev_head: bytes, ev_hash: bytes) -> bytes: + """``head_n = SHA-256(head_{n-1} || event_hash_n)`` — raw digests.""" + return hashlib.sha256(prev_head + ev_hash).digest() + + +class ChainMirror: + """Local replica of one stream's hash chain, fed with the exact + event objects the reporter submits. ``head_at(seq)`` is the step-5 + reference value a receipt for that seq must commit to.""" + + # Receipts arrive in the same response as the flush that produced + # them, so only a recent window of per-seq heads is ever consulted. + _HEAD_WINDOW = 8192 + + def __init__(self, session_id: str, stream: str) -> None: + self.session_id = session_id + self.stream = stream + self._head = genesis_head(session_id, stream) + self._heads: dict[int, str] = {} + + @property + def head_hex(self) -> str: + """Current chain head (lowercase hex64).""" + return self._head.hex() + + def append(self, event: dict) -> str: + """Advance the chain with one submitted event; returns the new + head hex. Raises ``TypeError`` on payloads a JS peer could not + hash identically (caller treats that as fail-open).""" + self._head = advance_head(self._head, event_hash(event)) + head_hex = self._head.hex() + seq = event["seq"] + self._heads[seq] = head_hex + stale = seq - self._HEAD_WINDOW + if stale in self._heads: + del self._heads[stale] + return head_hex + + def head_at(self, seq: int) -> Optional[str]: + """Chain head after the event at ``seq``, if still windowed.""" + return self._heads.get(seq) + + +# --------------------------------------------------------------------------- +# Signed-message framings (exact bytes per 06 §2.3/§2.4) +# --------------------------------------------------------------------------- + + +def receipt_message( + session_id: str, stream: str, seq: int, chain_head_hex: str, ts: int, +) -> bytes: + """06 §2.3 — per-slice receipt framing (receipt key signs this).""" + return ( + f"ddp-receipt:v1\n{session_id}\n{stream}\n{seq}" + f"\n{chain_head_hex}\n{ts}" + ).encode("utf-8") + + +def key_cert_message( + kid: str, public_key_b64url: str, not_before: int, not_after: int, +) -> bytes: + """06 §2.4 — receipt-key certification framing (sealing key signs + this).""" + return ( + f"ddp-keycert:v1\n{kid}\n{public_key_b64url}" + f"\n{not_before}\n{not_after}" + ).encode("utf-8") + + +# --------------------------------------------------------------------------- +# Key document + receipt verification (spec 06 §2.4) +# --------------------------------------------------------------------------- + + +def _b64url_decode(s: str) -> bytes: + return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) + + +def _verify_ed25519(public_key_raw: bytes, sig: bytes, message: bytes) -> bool: + """Raw Ed25519 verify; ``ImportError`` propagates so the caller can + report "cryptography unavailable" distinctly.""" + from cryptography.exceptions import InvalidSignature + from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PublicKey, + ) + + try: + Ed25519PublicKey.from_public_bytes(public_key_raw).verify(sig, message) + return True + except InvalidSignature: + return False + + +@dataclass(frozen=True) +class ReceiptKey: + """One certified entry of the ``receipt_keys`` list, post key-cert + validation (steps 1–2 of the §2.4 procedure).""" + + kid: str + public_key_raw: bytes + not_before: int + not_after: int + + +class ReceiptVerifier: + """Runs the 06 §2.4 verification procedure for one ledger base URL. + + Fail-open by contract: :meth:`verify` never raises — every failure + mode (network, missing ``cryptography``, malformed documents, bad + signatures) comes back as ``(False, reason)``. + """ + + def __init__(self, keys_url: str) -> None: + self.keys_url = keys_url + self._keys: dict[str, ReceiptKey] = {} + self._last_attempt: float = 0.0 + self._fetch_warned = False + + # ---- key document ---------------------------------------------------- + + def install_keys_document(self, doc: dict) -> int: + """Validate a ``GET /v1/keys`` document (steps 1–2): every + receipt key must carry a ``cert_sig`` by the sealing key over + the ``ddp-keycert:v1`` framing. Keys failing certification are + dropped with a warning. Returns the number installed.""" + keys: dict[str, ReceiptKey] = {} + try: + sealing = doc["sealing_key"] + sealing_raw = _b64url_decode(sealing["public_key"]) + entries = doc.get("receipt_keys") or [] + except (KeyError, TypeError, ValueError) as exc: + logger.warning("ledger keys document malformed: {}", exc) + self._keys = {} + return 0 + for entry in entries: + try: + kid = entry["kid"] + public_key = entry["public_key"] + not_before = int(entry["not_before"]) + not_after = int(entry["not_after"]) + cert_sig = _b64url_decode(entry["cert_sig"]) + certified = _verify_ed25519( + sealing_raw, + cert_sig, + key_cert_message(kid, public_key, not_before, not_after), + ) + except ImportError: + logger.warning( + "ledger receipt verification disabled: `cryptography` " + "is not installed (install the `provenance` extra)" + ) + self._keys = {} + return 0 + except (KeyError, TypeError, ValueError) as exc: + logger.warning("ledger receipt-key entry malformed: {}", exc) + continue + if not certified: + logger.warning( + "ledger receipt key kid={} failed key-cert validation " + "(cert_sig does not verify with the sealing key) — " + "dropping it", + entry.get("kid"), + ) + continue + keys[kid] = ReceiptKey( + kid=kid, + public_key_raw=_b64url_decode(public_key), + not_before=not_before, + not_after=not_after, + ) + self._keys = keys + return len(keys) + + def _fetch_keys(self) -> None: + now = time.monotonic() + if now - self._last_attempt < _KEYS_RETRY_MIN_S: + return + self._last_attempt = now + try: + req = urllib.request.Request(self.keys_url, method="GET") + with urllib.request.urlopen( + req, timeout=_REQUEST_TIMEOUT_S, + ) as resp: + doc = json.loads(resp.read()) + except (urllib.error.URLError, OSError, ValueError) as exc: + if not self._fetch_warned: + self._fetch_warned = True + logger.warning( + "ledger keys fetch failed url={} error={} " + "(receipts will count as unverified until it succeeds)", + self.keys_url, exc, + ) + return + if isinstance(doc, dict): + self.install_keys_document(doc) + + def _receipt_key(self, kid: str) -> Optional[ReceiptKey]: + if kid not in self._keys: + # Unknown kid → first receipt or a daily rotation; refetch + # (rate-floored inside _fetch_keys). + self._fetch_keys() + return self._keys.get(kid) + + # ---- the §2.4 procedure ---------------------------------------------- + + def verify( + self, + *, + session_id: str, + stream: str, + seq: object, + chain_head: object, + ts: object, + receipt_kid: object, + sig: object, + expected_head: Optional[str], + ) -> tuple[bool, Optional[str]]: + """Verify one §2.3 receipt against the local chain mirror. + + ``expected_head`` is the caller's own recomputed chain head at + ``seq`` (step 5). Returns ``(True, None)`` or + ``(False, reason)``; never raises.""" + try: + if ( + not isinstance(seq, int) + or not isinstance(chain_head, str) + or not isinstance(ts, int) + or not isinstance(receipt_kid, str) + or not isinstance(sig, str) + ): + return False, "malformed receipt (missing/mistyped fields)" + # Step 5 first: it needs no keys and is the check that makes + # the receipt a commitment to *our* history. + if expected_head is None: + return False, ( + f"no local chain head for seq={seq} " + "(mirror gap or out-of-window receipt)" + ) + if chain_head != expected_head: + return False, ( + f"chain_head mismatch at seq={seq}: receipt commits to " + f"{chain_head[:16]}… but local history gives " + f"{expected_head[:16]}…" + ) + # Steps 1–2: certified receipt key from the key document. + key = self._receipt_key(receipt_kid) + if key is None: + return False, ( + f"no certified receipt key for kid={receipt_kid}" + ) + # Step 3: receipt ts inside the key's validity window. + if not (key.not_before <= ts <= key.not_after): + return False, ( + f"receipt ts={ts} outside key validity " + f"[{key.not_before}, {key.not_after}] for " + f"kid={receipt_kid}" + ) + # Step 4: Ed25519 over the exact ddp-receipt:v1 framing. + try: + sig_raw = _b64url_decode(sig) + except (ValueError, TypeError): + return False, "receipt sig is not valid base64url" + if not _verify_ed25519( + key.public_key_raw, + sig_raw, + receipt_message(session_id, stream, seq, chain_head, ts), + ): + return False, ( + f"receipt signature invalid for seq={seq} " + f"kid={receipt_kid}" + ) + return True, None + except ImportError: + return False, ( + "cryptography is not installed (provenance extra) — " + "cannot verify receipts" + ) + except Exception as exc: # noqa: BLE001 — fail-open by contract + return False, f"receipt verification error: {exc}" diff --git a/acestep/provenance/session_log.py b/acestep/provenance/session_log.py new file mode 100644 index 00000000..d84854b3 --- /dev/null +++ b/acestep/provenance/session_log.py @@ -0,0 +1,675 @@ +"""Local session log: a JSONL event stream mirroring the ledger schema. + +One file per streaming session under +``/sessions/.jsonl``. Every line is one +event envelope in the **shared** schema of spec 06 §2.2, so "upload my +local history" can replay these files into the cloud ledger unchanged +(spec 02 §7):: + + {"stream": "local", "seq": 0, "type": "session.config", + "ts": 1751871234567, "payload": {...}} + +| Field | Meaning | +|---|---| +| ``stream`` | Always ``"local"`` for this file (spec 06 §2.2 file envelope). | +| ``seq`` | Per-file, contiguous from 0 (single local stream). | +| ``type`` | Namespaced event type (``session.config``, ``action.prompt``, ``action.param``, ``action.lora``, ``action.seed_audio``, ``action.transport``, ``session.note`` …), matching what :mod:`ledger_client` sends. | +| ``ts`` | Wall-clock **milliseconds since epoch** (spec 06 §0: ISO-8601 is never on the wire). | +| ``ppq`` | Optional DAW playhead in PPQ; omitted when unknown. | +| ``payload`` | Type-specific object. Prompt text is allowed here (local-only, spec 02 §5); only counts ever enter a manifest. | + +Slice hashing: this tap counts decoded slices seen on the bus, but does +**not** hash them. The §2.3 slice hash is SHA-256 over the *uncompressed +interleaved float16 downlink payload bytes* — which are produced in the +per-subscriber transport codec, not on this bus (the bus carries fully +reconstructed float32 audio). Pod-side slice hashing therefore lives in +the codec and is reported here via :meth:`SessionLogTap.record_pod_slice_hash` +→ ``slice.pod_hash`` ledger events, so the pod and client hash the same +bytes and can cross-check (spec 06 §2.3). + +Wiring: :func:`attach_session` subscribes a :class:`SessionLogTap` to +the session's typed event bus (:mod:`acestep.streaming.events`). The +session registry (:mod:`acestep.streaming.registry`) calls attach/detach +when a handle carrying a ``bus`` is registered/unregistered, so transport +adapters only have to hand the bus over. Torch-free, like the registry. +""" + +from __future__ import annotations + +import hashlib +import json +import threading +import time +from pathlib import Path +from typing import Any, Optional + +import numpy as np +from loguru import logger + +from acestep.provenance import session_logs_dir +from acestep.provenance.ledger_client import LedgerClient +from acestep.streaming.events import ( + AudioReady, + AudioWriteFailed, + AudioWritten, + CommandFailed, + DepthApplied, + LoraCatalogUpdate, + ParamsEcho, + ParamsUpdate, + PromptApplied, + PromptBlendEcho, + SessionError, + SessionReady, + StemAssets, + StemFailed, + StructureCleared, + StructureFailed, + StructureSet, + SubscriberDropped, + SwapFailed, + SwapReady, + TimbreCleared, + TimbreFailed, + TimbreSet, +) + +__all__ = [ + "LOCAL_STREAM", + "SessionLogWriter", + "SessionLogTap", + "attach_session", + "detach_session", + "get_tap", + "record_user_action", + "record_pod_slice_hash", + "session_summary_for", + "latest_session_summary", +] + +# File-envelope stream label for local logs (spec 06 §2.2). +LOCAL_STREAM = "local" + +# Params messages arrive at ~125 Hz; action.param records for them are +# diffed against the last logged values and rate-limited. +_PARAMS_ACTION_MIN_INTERVAL_S = 1.0 + +# Telemetry fields of the wire "params" message that are not user +# actions (playhead/flow-control reporting). +_PARAMS_TELEMETRY_KEYS = frozenset( + {"type", "playback_pos", "client_time", "slice_lead_s", "slice_bytes_rx"}, +) + +_MAX_SUMMARY_STR = 300 +_MAX_SUMMARY_LIST = 24 + + +def _now_ms() -> int: + """Wall-clock ms since epoch (spec 06 §0).""" + return int(time.time() * 1000) + + +def buffer_fingerprint(arr: np.ndarray) -> str: + """Numpy analogue of :func:`acestep.track_assets.waveform_fingerprint` + for ``[N, C]`` / ``[N]`` float buffers carried on bus events: mono + mix, fixed 4096-point decimation grid, quantize, sha256. Same + robustness intent (stable across benign decode round-trips); kept + torch-free because the tap runs on pods and local CPUs alike. + """ + a = np.asarray(arr, dtype=np.float32) + mono = a.mean(axis=1) if a.ndim == 2 else a.reshape(-1) + n = int(mono.shape[-1]) + if n == 0: + return "empty" + grid = 4096 + if n > grid: + idx = np.round(np.linspace(0, n - 1, grid)).astype(np.int64) + mono = mono[idx] + quantized = np.round(np.ascontiguousarray(mono) * 10_000.0).astype(np.int32) + return hashlib.sha256(quantized.tobytes()).hexdigest() + + +def _summarize(value: Any) -> Any: + """Shrink a payload value: bounded strings and lists, no binary. + Prompt text stays intact up to the cap — the log is local-only, so + prompt content is allowed here (spec 02 §5) even though only counts + ever enter a manifest.""" + if isinstance(value, (bytes, bytearray, memoryview)): + return {"bytes": len(value)} + if isinstance(value, str): + return value if len(value) <= _MAX_SUMMARY_STR else value[:_MAX_SUMMARY_STR] + "…" + if isinstance(value, dict): + return {str(k): _summarize(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + if len(value) > _MAX_SUMMARY_LIST: + return {"items": len(value)} + return [_summarize(v) for v in value] + if isinstance(value, (int, float, bool)) or value is None: + return value + return repr(value)[:_MAX_SUMMARY_STR] + + +class SessionLogWriter: + """Append-only JSONL writer for the spec 06 §2.2 envelope. Owns the + local stream's contiguous ``seq`` counter. Thread-safe; one flush per + event so a crash loses at most the in-flight line. Never raises out + of :meth:`record` — a broken log must not take down the session.""" + + def __init__(self, path: Path, session_id: str) -> None: + self.path = path + self.session_id = session_id + self._lock = threading.Lock() + self._seq = 0 + self._failed = False + path.parent.mkdir(parents=True, exist_ok=True) + self._fh = open(path, "a", encoding="utf-8") + + def record( + self, + type: str, + payload: dict, + *, + ts: int | None = None, + ppq: float | None = None, + ) -> None: + try: + with self._lock: + if self._fh.closed: + return + line = { + "stream": LOCAL_STREAM, + "seq": self._seq, + "type": type, + "ts": _now_ms() if ts is None else int(ts), + "payload": payload, + } + if ppq is not None: + line["ppq"] = float(ppq) + payload_str = json.dumps( + line, default=str, separators=(",", ":"), + ) + self._fh.write(payload_str + "\n") + self._fh.flush() + self._seq += 1 + except Exception as exc: # noqa: BLE001 + if not self._failed: + self._failed = True + logger.warning( + "session log write failed path={} error={}", + self.path, exc, + ) + + def file_sha256(self) -> str | None: + try: + with self._lock: + if not self._fh.closed: + self._fh.flush() + return hashlib.sha256(self.path.read_bytes()).hexdigest() + except Exception: # noqa: BLE001 + return None + + def close(self) -> None: + with self._lock: + try: + self._fh.close() + except Exception: # noqa: BLE001 + pass + + +class SessionLogTap: + """Event-bus subscriber that serializes session events to the local + log in the shared §2.2 schema (and forwards them to the ledger client + when one is configured). + + Counts decoded output slices for the summary but does not hash them + (see the module docstring): the §2.3 slice hash is over the transport + codec's float16 downlink bytes, reported via + :meth:`record_pod_slice_hash`. Runs on the subscription's drainer + thread plus whichever thread calls :meth:`record_user_action` / + :meth:`record_pod_slice_hash`; shared counters sit behind one lock. + """ + + def __init__( + self, + bus: Any, + session_id: str, + *, + meta: dict | None = None, + snapshot: Any = None, + log_dir: Path | None = None, + ) -> None: + self.session_id = session_id + d = Path(log_dir) if log_dir is not None else session_logs_dir() + self.writer = SessionLogWriter(d / f"{session_id}.jsonl", session_id) + self.ledger = LedgerClient(session_id=session_id) + + self._lock = threading.Lock() + self._started_wall = time.time() + self._slices = 0 # decoded slices seen on the bus + self._slice_hashes = 0 # slice.pod_hash reports forwarded + self._counts = { + "events": 0, + "prompt_changes": 0, + "param_changes": 0, + "user_actions": 0, + } + self._prompts_seen: set[str] = set() + self._last_params: dict = {} + self._last_params_wall = 0.0 + self._closed = False + + snap: dict = {} + if callable(snapshot): + try: + snap = snapshot() or {} + except Exception as exc: # noqa: BLE001 + logger.warning( + "session snapshot failed for provenance log: {}", exc, + ) + meta = dict(meta or {}) + loras = [ + e.get("id") + for e in snap.get("lora_catalog") or [] + if isinstance(e, dict) and e.get("state") == "enabled" + ] + self._model = meta.get("checkpoint") or snap.get("checkpoint") + self._loras = loras + # session.config: model + LoRA identifiers + initial config, + # mirroring the WS config handshake (spec 06 §2.2). + self._record( + "session.config", + { + "model": self._model, + "loras": loras, + "fixture_name": meta.get("fixture_name") + or snap.get("fixture_name"), + "prompt": snap.get("prompt"), + "bpm": snap.get("bpm"), + "key": snap.get("key"), + "time_signature": snap.get("time_signature"), + "extra": { + k: v for k, v in meta.items() + if k not in ("checkpoint", "fixture_name") + }, + }, + ) + if isinstance(snap.get("prompt"), str) and snap["prompt"]: + self._prompts_seen.add(snap["prompt"]) + + self._sub = bus.subscribe(self._on_event, name="provenance") + self._bus = bus + + # ---- recording ------------------------------------------------------- + + def _record(self, type: str, payload: dict, *, ppq: float | None = None) -> None: + ts = _now_ms() + with self._lock: + self._counts["events"] += 1 + self.writer.record(type, payload, ts=ts, ppq=ppq) + self.ledger.post_event(type, payload, ts=ts, ppq=ppq) + + def record_user_action( + self, action: str, payload: dict, *, source: str = "ws", + ) -> None: + """One authorship-action record in the §2.2 schema. ``params`` + actions map to ``action.param`` and are diffed against the last + logged knob values and rate-limited so the ~125 Hz knob channel + doesn't flood the log.""" + if self._closed: + return + if action == "params": + values = { + k: v for k, v in payload.items() + if k not in _PARAMS_TELEMETRY_KEYS + } + with self._lock: + changed = { + k: v for k, v in values.items() + if self._last_params.get(k) != v + } + now = time.monotonic() + if not changed or ( + now - self._last_params_wall < _PARAMS_ACTION_MIN_INTERVAL_S + ): + return + self._last_params.update(values) + self._last_params_wall = now + self._counts["user_actions"] += 1 + self._counts["param_changes"] += 1 + self._record( + "action.param", + {"source": source, "changed": _summarize(changed)}, + ) + return + with self._lock: + self._counts["user_actions"] += 1 + if action == "prompt": + self._counts["prompt_changes"] += 1 + tags = payload.get("tags") + if isinstance(tags, str) and tags: + self._prompts_seen.add(tags) + summary = _summarize({ + k: v for k, v in payload.items() if k != "type" + }) + body = {"source": source} + if isinstance(summary, dict): + body.update(summary) + else: + body["value"] = summary + if action == "prompt": + self._record("action.prompt", body) + else: + body["action"] = action + self._record("session.note", {"note": "user_action", **body}) + + def record_pod_slice_hash( + self, + *, + sha256: str, + start_sample: int, + num_samples: int, + channels: int, + slice_seq: int | None = None, + mac_verified: bool | None = None, + ) -> None: + """Report a pod-side output-slice hash (spec 06 §2.3): SHA-256 + over the uncompressed interleaved float16 downlink payload bytes, + computed in the transport codec where those exact bytes exist. + Emits a ``slice.pod_hash`` ledger event through this session's + ledger client so it shares the pod stream's contiguous seq.""" + if self._closed: + return + with self._lock: + self._slice_hashes += 1 + self.ledger.post_slice_hash( + sha256=sha256, + start_sample=start_sample, + num_samples=num_samples, + channels=channels, + slice_seq=slice_seq, + mac_verified=mac_verified, + ) + + # ---- bus tap --------------------------------------------------------- + + def _on_event(self, event: Any) -> None: + try: + self._dispatch_event(event) + except Exception as exc: # noqa: BLE001 — tap must never wedge the drainer + logger.warning("provenance tap event failed: {}", exc) + + def _dispatch_event(self, event: Any) -> None: + if isinstance(event, AudioReady): + self._on_slice(event) + elif isinstance(event, ParamsUpdate): + # Per-slice telemetry snapshot; authorship-irrelevant. + return + elif isinstance(event, PromptApplied): + with self._lock: + self._counts["prompt_changes"] += 1 + if event.tags: + self._prompts_seen.add(event.tags) + self._record("action.prompt", {"prompt": _summarize(event.tags)}) + elif isinstance(event, PromptBlendEcho): + self._record( + "action.param", {"name": "prompt_blend", "value": event.value}, + ) + elif isinstance(event, ParamsEcho): + with self._lock: + self._counts["param_changes"] += 1 + self._record("action.param", {"raw": _summarize(event.raw)}) + elif isinstance(event, DepthApplied): + self._record( + "action.param", {"name": "depth", "value": event.value}, + ) + elif isinstance(event, LoraCatalogUpdate): + loras = [ + e.get("id") + for e in event.catalog or [] + if isinstance(e, dict) and e.get("state") == "enabled" + ] + with self._lock: + self._loras = loras + self._record("action.lora", {"loras": loras}) + elif isinstance(event, SessionReady): + self._record( + "action.seed_audio", + { + "sha256": buffer_fingerprint(event.initial_buffer), + "label": "initial_source", + "duration_sec": event.duration, + "sample_rate": event.sample_rate, + "bpm": event.bpm, + "key": event.key, + "time_signature": event.time_signature, + "pipeline_depth": event.pipeline_depth, + }, + ) + elif isinstance(event, SwapReady): + self._record( + "action.seed_audio", + { + "sha256": buffer_fingerprint(event.initial_buffer), + "label": "source_swap", + "fixture_name": event.fixture_name, + "duration_sec": event.duration, + "bpm": event.bpm, + "key": event.key, + "source_epoch": event.source_epoch, + }, + ) + elif isinstance(event, TimbreSet): + self._record( + "action.param", + {"name": "timbre", "op": "set", + "timbre": event.name, "duration": event.duration}, + ) + elif isinstance(event, TimbreCleared): + self._record("action.param", {"name": "timbre", "op": "clear"}) + elif isinstance(event, StructureSet): + self._record( + "action.param", + {"name": "structure", "op": "set", + "structure": event.name, "duration": event.duration}, + ) + elif isinstance(event, StructureCleared): + self._record("action.param", {"name": "structure", "op": "clear"}) + elif isinstance(event, AudioWritten): + self._record( + "action.transport", + {"op": "write_audio", "start_s": event.start_s, + "end_s": event.end_s, "source_epoch": event.source_epoch}, + ) + elif isinstance(event, StemAssets): + self._record( + "session.note", + {"note": "stem_assets", "fixture_name": event.fixture_name, + "source_mode": event.source_mode, "frames": event.frames}, + ) + elif isinstance(event, ( + CommandFailed, SwapFailed, StemFailed, TimbreFailed, + StructureFailed, AudioWriteFailed, SessionError, + )): + self._record( + "session.note", + {"note": "error", "kind": type(event).__name__, + "error": _summarize(getattr(event, "error", None) + or getattr(event, "message", ""))}, + ) + elif isinstance(event, SubscriberDropped): + # Our own queue overflowed: the log has a gap from here on. + self._record( + "session.note", + {"note": "log_tap_dropped", "reason": event.reason}, + ) + + def _on_slice(self, event: AudioReady) -> None: + # Count decoded slices for the summary. The bus carries fully + # reconstructed float32 audio, NOT the float16 downlink bytes the + # client hashes, so hashing here could never cross-check — that + # hash is produced in the transport codec and reported via + # record_pod_slice_hash (spec 06 §2.3). + with self._lock: + self._slices += 1 + + # ---- summary / lifecycle --------------------------------------------- + + def timeline_summary(self) -> dict: + """Counts-only view of the timeline for the manifest's + ``com.daydream.session`` assertion — never prompt content.""" + with self._lock: + elapsed = time.time() - self._started_wall + return { + **self._counts, + "distinct_prompts": len(self._prompts_seen), + "slices": self._slices, + "slice_hashes": self._slice_hashes, + "duration_s": round(elapsed, 3), + } + + def manifest_summary(self) -> dict: + with self._lock: + model, loras = self._model, list(self._loras) + return { + "session_id": self.session_id, + "model": model, + "loras": loras, + "timeline_summary": self.timeline_summary(), + "session_log_sha256": self.writer.file_sha256(), + } + + def close(self) -> None: + if self._closed: + return + self._closed = True + try: + self._bus.unsubscribe(self._sub) + self._sub.join(timeout=5) + except Exception: # noqa: BLE001 + pass + with self._lock: + n, h = self._slices, self._slice_hashes + self._record( + "session.note", + { + "note": "session_end", + "slices": n, + "slice_hashes": h, + "timeline_summary": self.timeline_summary(), + }, + ) + self.ledger.close() + self.writer.close() + + +# --------------------------------------------------------------------------- +# Process-global tap registry, driven by acestep.streaming.registry +# --------------------------------------------------------------------------- + +_taps: dict[str, SessionLogTap] = {} +_taps_order: list[str] = [] +_taps_lock = threading.Lock() + + +def attach_session(handle: Any, *, log_dir: Path | None = None) -> Optional[SessionLogTap]: + """Attach a log tap for a registered session handle (must carry a + ``bus``; ``snapshot`` and ``provenance_meta`` are optional). Returns + ``None`` — after one warning — if the tap could not be created.""" + bus = getattr(handle, "bus", None) + if bus is None: + return None + try: + tap = SessionLogTap( + bus, + handle.id, + meta=getattr(handle, "provenance_meta", None), + snapshot=getattr(handle, "snapshot", None), + log_dir=log_dir, + ) + except Exception as exc: # noqa: BLE001 — never block session registration + logger.warning( + "session log attach failed session_id={} error={}", + getattr(handle, "id", "?"), exc, + ) + return None + with _taps_lock: + _taps[handle.id] = tap + _taps_order.append(handle.id) + logger.info("session_log_attached path={}", tap.writer.path) + return tap + + +def detach_session(session_id: str) -> None: + with _taps_lock: + tap = _taps.pop(session_id, None) + if session_id in _taps_order: + _taps_order.remove(session_id) + if tap is not None: + tap.close() + + +def get_tap(session_id: str) -> Optional[SessionLogTap]: + with _taps_lock: + return _taps.get(session_id) + + +def record_user_action( + session_id: str, action: str, payload: dict, *, source: str = "ws", +) -> None: + """Module-level convenience for transport adapters: no-op when the + session has no attached tap (e.g. the brief window before the + registry registers the handle).""" + tap = get_tap(session_id) + if tap is not None: + tap.record_user_action(action, payload, source=source) + + +def record_pod_slice_hash( + session_id: str, + *, + sha256: str, + start_sample: int, + num_samples: int, + channels: int, + slice_seq: int | None = None, + mac_verified: bool | None = None, +) -> None: + """Module-level convenience for the transport codec: forward a + pod-side slice hash (spec 06 §2.3) to the session's tap, no-op when + there is none.""" + tap = get_tap(session_id) + if tap is not None: + tap.record_pod_slice_hash( + sha256=sha256, + start_sample=start_sample, + num_samples=num_samples, + channels=channels, + slice_seq=slice_seq, + mac_verified=mac_verified, + ) + + +def session_summary_for(session_id: str) -> Optional[dict]: + """Manifest-ready summary of a **specific** session (spec 06 §2.5 + binds a record to the session that produced the asset). ``None`` when + that session has no active tap — callers must then emit null session + fields rather than borrow another session's identity.""" + tap = get_tap(session_id) + return tap.manifest_summary() if tap is not None else None + + +def latest_session_summary() -> Optional[dict]: + """Summary of the most recently attached live session, or ``None``. + + Do NOT use this to bind a manifest to an asset: under concurrent + sessions the newest tap is not necessarily the one that produced a + given asset (audit F7/G5). Use :func:`session_summary_for` with the + id of the session that produced the asset instead. Retained only for + diagnostics / single-session callers that explicitly want "whatever + is live now". + """ + with _taps_lock: + if not _taps_order: + return None + tap = _taps.get(_taps_order[-1]) + return tap.manifest_summary() if tap is not None else None diff --git a/acestep/streaming/registry.py b/acestep/streaming/registry.py index 7c12a530..fddce3a6 100644 --- a/acestep/streaming/registry.py +++ b/acestep/streaming/registry.py @@ -39,6 +39,13 @@ class SessionHandle: started_at: float inject: InjectFn snapshot: SnapshotFn + # Optional provenance surface. When ``bus`` (the session's typed + # EventBus) is present, register() attaches a local session-log tap + # (acestep.provenance.session_log) and unregister() closes it. + # ``provenance_meta`` carries adapter-known identifiers the snapshot + # doesn't expose (checkpoint, fixture name). + bus: Optional[Any] = None + provenance_meta: Optional[dict] = None _sessions: dict[str, SessionHandle] = {} @@ -53,11 +60,25 @@ def new_session_id() -> str: def register(handle: SessionHandle) -> None: with _lock: _sessions[handle.id] = handle + if handle.bus is not None: + # Lazy + guarded: provenance logging must never block session + # registration, and this module stays importable (torch-free, + # provenance-free) under --no-backend. + try: + from acestep.provenance.session_log import attach_session + attach_session(handle) + except Exception: + pass def unregister(session_id: str) -> None: with _lock: _sessions.pop(session_id, None) + try: + from acestep.provenance.session_log import detach_session + detach_session(session_id) + except Exception: + pass def get(session_id: str) -> Optional[SessionHandle]: diff --git a/acestep/track_assets.py b/acestep/track_assets.py index 985d076e..4b248154 100644 --- a/acestep/track_assets.py +++ b/acestep/track_assets.py @@ -188,6 +188,33 @@ def save_track_metadata( os.replace(tmp, p) +def _embed_upload_provenance(path: Path) -> None: + """Best-effort C2PA manifest embed after a HUMAN-ORIGIN WAV lands on + disk (uploaded source track / separated stems). + + Import is lazy and the embed itself never raises (it degrades to a + single logged warning without the ``provenance`` extra), so asset + persistence is byte-identical in behaviour when provenance is off. + + These writers are called only from the user-upload path, i.e. on the + user's own uploaded/seed audio BEFORE any generation. That audio is + the dry input, so the manifest asserts ``digitalCapture`` (human + origin) and NOT any trained-algorithmic / composite source type — + marking the dry input as synthetic would be an affirmatively false + claim (spec 02-architecture §4, audit G5). The CAWG do-not-train + default still rides along (protecting the user's own content), and no + session identity is bound (these writes precede any session output). + """ + try: + from acestep.provenance.manifest import ( + DIGITAL_CAPTURE, + embed_wav_manifest, + ) + except Exception: + return + embed_wav_manifest(path, source_type=DIGITAL_CAPTURE) + + def write_stem_wavs( root: Path, name: str, @@ -197,6 +224,9 @@ def write_stem_wavs( ) -> None: import soundfile as sf + # Stems are model-separated from the user's UPLOADED source track — + # human-origin audio, produced here before any generation. They are + # not synthetic, so their manifest asserts digitalCapture (audit G5). for mode in STEM_MODES: if mode not in stems: raise ValueError(f"missing stem: {mode}") @@ -206,6 +236,9 @@ def write_stem_wavs( wav = stems[mode].detach().cpu().float() sf.write(str(tmp), wav.numpy().T, int(sample_rate), format="WAV", subtype="FLOAT") os.replace(tmp, p) + # Stems are separated from the user's uploaded source — human + # origin, not synthetic (audit G5). digitalCapture, no AI claim. + _embed_upload_provenance(p) def write_track_wav( @@ -223,6 +256,10 @@ def write_track_wav( wav = waveform.detach().cpu().float() sf.write(str(tmp), wav.numpy().T, int(sample_rate), format="WAV", subtype="FLOAT") os.replace(tmp, p) + # The written track is the user's UPLOADED source (human origin, dry + # input before any generation), so the manifest asserts digitalCapture + # — never a trained-algorithmic / composite synthetic claim (audit G5). + _embed_upload_provenance(p) def read_stem_wavs( diff --git a/demos/realtime_motion_graph_web/audio_codec.py b/demos/realtime_motion_graph_web/audio_codec.py index 1e1c46f7..f150b0da 100644 --- a/demos/realtime_motion_graph_web/audio_codec.py +++ b/demos/realtime_motion_graph_web/audio_codec.py @@ -19,6 +19,7 @@ from __future__ import annotations +import hashlib import json import struct @@ -49,6 +50,12 @@ def __init__(self, initial_mirror: np.ndarray, zstd_level: int = 1): # place on every encode. self._mirror = initial_mirror.copy() self._zctx = zstd.ZstdCompressor(level=zstd_level) + # Pod-side monotonic slice counter (spec 06 §3) and the last + # encoded frame's slice-hash report (spec 06 §2.3), consumed by + # the transport to emit a ``slice.pod_hash`` ledger event. ``None`` + # until the first non-empty frame is encoded. + self._slice_seq = 0 + self.last_slice_hash: dict | None = None @property def mirror(self) -> np.ndarray: @@ -78,12 +85,26 @@ def encode( ss = int(start_sample) se = min(ss + len(audio), len(self._mirror)) if se <= ss: + self.last_slice_hash = None return None region = audio[: se - ss] mirror_region = self._mirror[ss:se] # Delta = what server has now minus what client has delta = (region - mirror_region).astype(np.float16) - compressed = self._zctx.compress(delta.tobytes()) + delta_bytes = delta.tobytes() + # Pod-side slice hash (spec 06 §2.3): SHA-256 over the uncompressed + # interleaved float16 payload bytes — the exact bytes the client + # gets back after zstd-decompressing this frame, so the pod and + # client hashes compare directly for cross-checking. + self.last_slice_hash = { + "sha256": hashlib.sha256(delta_bytes).hexdigest(), + "start_sample": ss, + "num_samples": se - ss, + "channels": int(channels), + "slice_seq": self._slice_seq, + } + self._slice_seq += 1 + compressed = self._zctx.compress(delta_bytes) # Mirror our copy to the *reconstruction the client will hold*, not # the exact ``region``. The client applies ``mirror += float32(delta)`` # with ``delta`` quantized to float16, so storing the exact region diff --git a/demos/realtime_motion_graph_web/ws_adapter.py b/demos/realtime_motion_graph_web/ws_adapter.py index 7756578b..0c3bdab2 100644 --- a/demos/realtime_motion_graph_web/ws_adapter.py +++ b/demos/realtime_motion_graph_web/ws_adapter.py @@ -1366,6 +1366,19 @@ def _serialize_audio_ready(event: AudioReady) -> None: # the in-flight window permanently negative so the load-bearing # flow-control layer stops engaging. _slice_flow["sent"] += len(frame) + # Report the pod-side slice hash over the exact float16 bytes + # just sent (spec 06 §2.3), for pod/client cross-checking. + # Fully guarded + fail-open: provenance must never wedge the + # downlink (spec 06 §7). + slice_hash = codec.last_slice_hash + if slice_hash is not None: + try: + from acestep.provenance.session_log import ( + record_pod_slice_hash, + ) + record_pod_slice_hash(session_id, **slice_hash) + except Exception: # noqa: BLE001 + pass except ConnectionClosed: state.running = False diff --git a/pyproject.toml b/pyproject.toml index 6f7bf967..b5fb850b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -81,6 +81,15 @@ dependencies = [ "mcp>=1.0", ] +[project.optional-dependencies] +# Local C2PA provenance (acestep/provenance): signed Content Credentials +# on written WAVs + self-signed local signing material. Everything in +# acestep.provenance degrades to a logged no-op without this extra. +provenance = [ + "c2pa-python>=0.36", + "cryptography>=42", +] + [[tool.uv.index]] name = "pytorch-cu128" url = "https://download.pytorch.org/whl/cu128" diff --git a/tests/unit/test_provenance_manifest.py b/tests/unit/test_provenance_manifest.py new file mode 100644 index 00000000..d6ecb888 --- /dev/null +++ b/tests/unit/test_provenance_manifest.py @@ -0,0 +1,320 @@ +"""C2PA manifest round-trip through the track-asset WAV writers. + +Pure CPU — no GPU, no model load. Exercises the real ``c2pa-python`` +SDK (the ``provenance`` extra): :func:`acestep.track_assets. +write_track_wav` / :func:`write_stem_wavs` embed a signed manifest, +and we read it back with ``c2pa.Reader`` to assert the spec-02 §3 +posture survives signing: the IPTC ``digitalSourceType``, the CAWG +training-and-data-mining assertion (``notAllowed`` everywhere), and +the custom ``com.daydream.session`` assertion. + +The no-SDK degrade path is covered too (simulated by forcing the +import guard), because asset writing must behave identically when the +``provenance`` extra is absent. +""" + +from __future__ import annotations + +import json +import sys +import time +from pathlib import Path + +import pytest +import soundfile as sf +import torch + +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +c2pa = pytest.importorskip("c2pa", reason="provenance extra not installed") + +import acestep.provenance.manifest as manifest_mod +from acestep.provenance.manifest import ( + COMPOSITE_WITH_TRAINED_ALGORITHMIC_MEDIA, + DIGITAL_CAPTURE, + TRAINED_ALGORITHMIC_MEDIA, + build_manifest_definition, + embed_wav_manifest, +) +from acestep.track_assets import ( + save_track_metadata, + source_audio_path, + stem_audio_path, + waveform_fingerprint, + write_stem_wavs, + write_track_wav, +) + +SAMPLE_RATE = 48_000 + + +@pytest.fixture(autouse=True) +def _isolated_provenance_dir(tmp_path, monkeypatch): + """Every test signs with throwaway keys under its own tmp dir.""" + monkeypatch.setenv("ACESTEP_PROVENANCE_DIR", str(tmp_path / "provenance")) + + +def _waveform(seed: int = 0) -> torch.Tensor: + g = torch.Generator().manual_seed(seed) + return torch.rand((2, 4800), generator=g) * 0.5 + + +def _read_active_manifest(path: Path) -> dict: + with c2pa.Reader(str(path)) as reader: + data = json.loads(reader.json()) + return data["manifests"][data["active_manifest"]] + + +def _assertion(active: dict, label: str) -> dict: + matches = [a for a in active["assertions"] if a["label"].startswith(label)] + assert matches, f"no {label!r} assertion in {active['assertions']}" + assert len(matches) == 1 + return matches[0]["data"] + + +# --------------------------------------------------------------------------- +# Round trip through the writers +# --------------------------------------------------------------------------- + + +def test_write_track_wav_labels_upload_as_digital_capture(tmp_path): + """The written track is the user's UPLOADED source — human origin, dry + input before any generation. It must NOT be marked trained/composite + (audit G5); digitalCapture is the honest label.""" + root = tmp_path / "uploads" + waveform = _waveform() + + write_track_wav(root, "song.wav", waveform=waveform, sample_rate=SAMPLE_RATE) + + p = source_audio_path(root, "song.wav") + active = _read_active_manifest(p) + + # Claim generator identity: local self-signed, never Daydream. + assert active["claim_generator_info"][0]["name"] == "DEMON (local, self-signed)" + + # c2pa.created with digitalCapture — never a synthetic source type. + actions = _assertion(active, "c2pa.actions")["actions"] + created = [a for a in actions if a["action"] == "c2pa.created"] + assert len(created) == 1 + assert created[0]["digitalSourceType"] == DIGITAL_CAPTURE + assert created[0]["digitalSourceType"] != COMPOSITE_WITH_TRAINED_ALGORITHMIC_MEDIA + assert created[0]["digitalSourceType"] != TRAINED_ALGORITHMIC_MEDIA + + # CAWG do-not-train still rides along (protecting the user's content). + entries = _assertion(active, "cawg.training-mining")["entries"] + assert set(entries) == { + "cawg.ai_generative_training", + "cawg.ai_inference", + "cawg.ai_training", + "cawg.data_mining", + } + assert all(v == {"use": "notAllowed"} for v in entries.values()) + + # com.daydream.session: no session bound, and no AI "seed" claim on + # human input. + session = _assertion(active, "com.daydream.session") + assert session["session_id"] is None + assert session["model"] is None + assert session["loras"] == [] + assert "seed_waveform_sha256" not in session + + # The signed WAV must still be a decodable WAV. + data, sr = sf.read(str(p), dtype="float32", always_2d=True) + assert sr == SAMPLE_RATE + assert data.shape == (4800, 2) + + +def test_write_stem_wavs_labels_stems_as_digital_capture(tmp_path): + """Separated stems are derived from the user's uploaded source — + still human origin, not synthetic. digitalCapture, no AI claim.""" + root = tmp_path / "uploads" + waveform = _waveform() + stems = {"vocals": waveform + 0.1, "instruments": waveform + 0.2} + + save_track_metadata(root, "song.wav", waveform=waveform, sample_rate=SAMPLE_RATE) + write_stem_wavs(root, "song.wav", stems=stems, sample_rate=SAMPLE_RATE) + + for mode in ("vocals", "instruments"): + active = _read_active_manifest(stem_audio_path(root, "song.wav", mode)) + actions = _assertion(active, "c2pa.actions")["actions"] + assert actions[0]["digitalSourceType"] == DIGITAL_CAPTURE + assert actions[0]["digitalSourceType"] != COMPOSITE_WITH_TRAINED_ALGORITHMIC_MEDIA + session = _assertion(active, "com.daydream.session") + assert "seed_waveform_sha256" not in session + + +def test_uploads_never_borrow_a_live_session_identity(tmp_path): + """Even with a live session attached, an upload-path write must NOT be + stamped with that session's identity or an AI source type (audit + F7/G5): the upload is human input, produced before any generation.""" + from acestep.streaming.events import EventBus, PromptApplied + from acestep.streaming import registry + + bus = EventBus() + handle = registry.SessionHandle( + id="prov-test-1", + started_at=time.time(), + inject=lambda data, audio: None, + snapshot=lambda: {"lora_catalog": [{"id": "lead", "state": "enabled"}]}, + bus=bus, + provenance_meta={"checkpoint": "ace_step_v1"}, + ) + registry.register(handle) + try: + bus.publish(PromptApplied(tags="dark techno")) + root = tmp_path / "uploads" + waveform = _waveform() + write_track_wav(root, "live.wav", waveform=waveform, sample_rate=SAMPLE_RATE) + finally: + registry.unregister(handle.id) + + active = _read_active_manifest(source_audio_path(root, "live.wav")) + actions = _assertion(active, "c2pa.actions")["actions"] + assert actions[0]["digitalSourceType"] == DIGITAL_CAPTURE + session = _assertion(active, "com.daydream.session") + # The upload did NOT grab the live session's identity. + assert session["session_id"] is None + assert session["model"] is None + assert session["loras"] == [] + + +def test_embed_binds_to_the_specific_session_not_latest(tmp_path): + """embed_wav_manifest(session_id=...) binds a genuinely-generated + asset to THAT session, even when a newer session is also live (audit + F7/G5: never the arbitrary 'latest' tap).""" + from acestep.streaming.events import EventBus + from acestep.streaming import registry + + bus_a, bus_b = EventBus(), EventBus() + ha = registry.SessionHandle( + id="sess-A", started_at=time.time(), + inject=lambda d, a: None, + snapshot=lambda: {"lora_catalog": [{"id": "lead", "state": "enabled"}]}, + bus=bus_a, provenance_meta={"checkpoint": "model-A"}, + ) + hb = registry.SessionHandle( + id="sess-B", started_at=time.time(), + inject=lambda d, a: None, snapshot=lambda: {}, + bus=bus_b, provenance_meta={"checkpoint": "model-B"}, + ) + registry.register(ha) + registry.register(hb) # B is newest → the old "latest" bug's target + try: + root = tmp_path / "gen" + write_track_wav(root, "seed.wav", waveform=_waveform(), sample_rate=SAMPLE_RATE) + p = source_audio_path(root, "seed.wav") + # Simulate a generated-output writer re-signing the asset bound to + # session A (the specific producing session), not the latest tap. + ok = embed_wav_manifest( + p, session_id="sess-A", + source_type=COMPOSITE_WITH_TRAINED_ALGORITHMIC_MEDIA, + ) + assert ok + finally: + registry.unregister("sess-A") + registry.unregister("sess-B") + + active = _read_active_manifest(p) + session = _assertion(active, "com.daydream.session") + assert session["session_id"] == "sess-A" + assert session["model"] == "model-A" + assert session["loras"] == ["lead"] + assert session["session_log_sha256"] + # The generated-output path keeps its synthetic source type. + actions = _assertion(active, "c2pa.actions")["actions"] + assert actions[0]["digitalSourceType"] == COMPOSITE_WITH_TRAINED_ALGORITHMIC_MEDIA + + +def test_embed_without_session_context_emits_null_session(tmp_path): + """No session dict and no session_id → null session fields, never a + borrowed 'latest' identity.""" + from acestep.streaming.events import EventBus + from acestep.streaming import registry + + bus = EventBus() + h = registry.SessionHandle( + id="sess-live", started_at=time.time(), + inject=lambda d, a: None, snapshot=lambda: {}, + bus=bus, provenance_meta={"checkpoint": "model-live"}, + ) + registry.register(h) + try: + root = tmp_path / "gen" + write_track_wav(root, "x.wav", waveform=_waveform(), sample_rate=SAMPLE_RATE) + p = source_audio_path(root, "x.wav") + # Re-embed with NO session context at all. + assert embed_wav_manifest( + p, source_type=COMPOSITE_WITH_TRAINED_ALGORITHMIC_MEDIA, + ) + finally: + registry.unregister("sess-live") + + active = _read_active_manifest(source_audio_path(root, "x.wav")) + session = _assertion(active, "com.daydream.session") + assert session["session_id"] is None + assert session["model"] is None + + +# --------------------------------------------------------------------------- +# Manifest definition builder (pure, no SDK involvement) +# --------------------------------------------------------------------------- + + +def test_build_manifest_definition_defaults_to_trained_media(): + manifest = build_manifest_definition(title="out.wav") + actions = manifest["assertions"][0]["data"]["actions"] + assert actions[0]["digitalSourceType"] == TRAINED_ALGORITHMIC_MEDIA + + +def test_build_manifest_definition_fingerprint_implies_composite(): + manifest = build_manifest_definition( + title="out.wav", ingredient_fingerprint="ab" * 32, + ) + actions = manifest["assertions"][0]["data"]["actions"] + assert actions[0]["digitalSourceType"] == COMPOSITE_WITH_TRAINED_ALGORITHMIC_MEDIA + session = manifest["assertions"][2]["data"] + assert session["seed_waveform_sha256"] == "ab" * 32 + + +def test_build_manifest_definition_explicit_source_type_wins(): + manifest = build_manifest_definition( + title="out.wav", + source_type=COMPOSITE_WITH_TRAINED_ALGORITHMIC_MEDIA, + ) + actions = manifest["assertions"][0]["data"]["actions"] + assert actions[0]["digitalSourceType"] == COMPOSITE_WITH_TRAINED_ALGORITHMIC_MEDIA + + +# --------------------------------------------------------------------------- +# Degrade path: writers must behave identically without the SDK +# --------------------------------------------------------------------------- + + +def test_writers_survive_missing_c2pa(tmp_path, monkeypatch): + monkeypatch.setattr(manifest_mod, "_import_c2pa", lambda: None) + root = tmp_path / "uploads" + waveform = _waveform() + + write_track_wav(root, "plain.wav", waveform=waveform, sample_rate=SAMPLE_RATE) + + p = source_audio_path(root, "plain.wav") + data, sr = sf.read(str(p), dtype="float32", always_2d=True) + assert sr == SAMPLE_RATE and data.shape == (4800, 2) + assert c2pa.Reader.try_create(str(p)) is None # no manifest embedded + assert not list(p.parent.glob("*.c2pa.tmp")) + + +def test_embed_failure_leaves_unsigned_wav_intact(tmp_path, monkeypatch): + root = tmp_path / "uploads" + waveform = _waveform() + write_track_wav(root, "keep.wav", waveform=waveform, sample_rate=SAMPLE_RATE) + p = source_audio_path(root, "keep.wav") + before = p.read_bytes() + + def _boom(*args, **kwargs): + raise RuntimeError("signer exploded") + + monkeypatch.setattr(manifest_mod, "build_manifest_definition", _boom) + assert embed_wav_manifest(p) is False + assert p.read_bytes() == before + assert not list(p.parent.glob("*.c2pa.tmp")) diff --git a/tests/unit/test_provenance_receipts.py b/tests/unit/test_provenance_receipts.py new file mode 100644 index 00000000..e48127f7 --- /dev/null +++ b/tests/unit/test_provenance_receipts.py @@ -0,0 +1,504 @@ +"""Client-side ledger receipt verification (spec 06 §2.4). + +Pure Python — no GPU, no c2pa. Covers the three layers of +:mod:`acestep.provenance.receipts` and their wiring into +:class:`~acestep.provenance.ledger_client.LedgerClient`: + +- **JCS + chain golden vector**: the fixed events, canonical strings + and chain heads below are copied verbatim from the server's own + regression fixtures (pipelines-provenance ``test/provenance/ + chain.test.ts`` / ``jcs.test.ts``). Both implementations hashing the + same inputs to the same heads is the whole point of §2.4 step 5 — + these values must NEVER be regenerated from this codebase. +- **§2.4 procedure**: server-shaped key documents and receipts built + with ephemeral Ed25519 keys and the exact ``ddp-keycert:v1`` / + ``ddp-receipt:v1`` framings; every element (sig, chain_head, ts, + key-cert) is tampered in turn and must fail with a telling reason. +- **LedgerClient integration**: a stub ledger that really chains and + really signs; receipts verify as they arrive, tampering bumps the + unverified counter, and nothing ever raises into the reporter + (fail-open). +""" + +from __future__ import annotations + +import base64 +import json +import math +import sys +import threading +import time +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat + +import pytest + +from acestep.provenance.ledger_client import LedgerClient +from acestep.provenance.receipts import ( + ChainMirror, + ReceiptVerifier, + advance_head, + event_hash, + genesis_head, + jcs, + key_cert_message, + receipt_message, +) + +# --------------------------------------------------------------------------- +# Golden fixture — copied VERBATIM from the server's chain.test.ts. +# --------------------------------------------------------------------------- + +GOLDEN_SESSION = "sess_test01" +GOLDEN_STREAM = "client" +GOLDEN_EVENTS = [ + { + "seq": 0, + "type": "action.prompt", + "ts": 1751871234567, + "payload": {"prompt": "warm analog keys", "slot": "A"}, + }, + { + "seq": 1, + "type": "action.param", + "ts": 1751871234890, + "ppq": 3840.25, + "payload": {"name": "cutoff", "value": 0.42}, + }, + { + "seq": 2, + "type": "slice.client_hash", + "ts": 1751871235000, + "payload": { + "slice_seq": 812, + "start_sample": 4177920, + "num_samples": 96000, + "channels": 2, + "sha256": "ab34cd56ab34cd56ab34cd56ab34cd56" + "ab34cd56ab34cd56ab34cd56ab34cd56", + "mac_verified": True, + }, + }, +] +GOLDEN = { + "genesis": "5dbdea7c3069a7a5af26257c7a8a9899d5a6ace861e7173a0a45d8ed5f9ea4d6", + "event0Hash": "311d3b371d059b44fa137d8922b822c24f05d8b881ec78144ac390d09a5e7ed9", + "head1": "84c9291fd02dee9a3bca470469c0edce22cee49b837d53cfdb2ec049ac542905", + "head2": "12b769bb9d8eda564f5a81c279d2f563d60220042a3e6f9b9447b34acebb11f9", + "head3": "03cbf0fde0a7b6238600859273b5c7f6088f8701d20206715c9eabe3ce63de9b", +} + + +# --------------------------------------------------------------------------- +# JCS canonicalization (RFC 8785, ECMAScript number/string formatting) +# --------------------------------------------------------------------------- + + +def test_jcs_matches_server_golden_canonical_form(): + # Same assertion as the server's jcs.test.ts: the exact bytes a + # reporter must produce locally for receipt verification to work. + assert jcs(GOLDEN_EVENTS[0]) == ( + '{"payload":{"prompt":"warm analog keys","slot":"A"},' + '"seq":0,"ts":1751871234567,"type":"action.prompt"}' + ) + + +def test_jcs_sorts_keys_recursively_and_deterministically(): + assert jcs({"b": 1, "a": {"d": 2, "c": 3}}) == '{"a":{"c":3,"d":2},"b":1}' + x = {} + x["z"] = 1 + x["a"] = 2 + y = {} + y["a"] = 2 + y["z"] = 1 + assert jcs(x) == jcs(y) + + +def test_jcs_numbers_serialize_like_ecmascript_json_stringify(): + # The jcs.test.ts vector plus the notation boundaries JS uses. + assert jcs(3840.25) == "3840.25" + assert jcs(1e21) == "1e+21" # not Python's "1e+21" by luck: e+ form + assert jcs(0.000001) == "0.000001" # Python repr says "1e-06" + assert jcs(1e-7) == "1e-7" # exponential below 1e-6, JS "e-" form + assert jcs(1751871234567) == "1751871234567" + assert jcs(-0.0) == "0" # JSON.stringify(-0) === "0" + assert jcs(1.0) == "1" # integral floats drop the ".0" + assert jcs(0.42) == "0.42" + + +def test_jcs_rejects_values_a_js_peer_cannot_hash_identically(): + with pytest.raises(TypeError): + jcs(float("nan")) + with pytest.raises(TypeError): + jcs(math.inf) + with pytest.raises(TypeError): + jcs(2 ** 53) # beyond Number.MAX_SAFE_INTEGER: JS would round + with pytest.raises(TypeError): + jcs({"payload": object()}) + + +def test_jcs_escapes_strings_like_json_stringify(): + assert jcs('a"b\\c\n') == json.dumps('a"b\\c\n', ensure_ascii=False) + assert jcs("\x07") == '"\\u0007"' + assert jcs("é中") == '"é中"' # non-ASCII stays literal + + +# --------------------------------------------------------------------------- +# Chain rules — cross-implementation golden vector (the §2.4 step-5 core) +# --------------------------------------------------------------------------- + + +def test_chain_golden_vector_byte_matches_the_server(): + # genesis = SHA-256 of the ddp:v1 domain-separation string. + assert genesis_head(GOLDEN_SESSION, GOLDEN_STREAM).hex() == GOLDEN["genesis"] + # First event hash, then every intermediate head, must equal the + # values the server's independent implementation pins. + assert event_hash(GOLDEN_EVENTS[0]).hex() == GOLDEN["event0Hash"] + head = genesis_head(GOLDEN_SESSION, GOLDEN_STREAM) + heads = [] + for event in GOLDEN_EVENTS: + head = advance_head(head, event_hash(event)) + heads.append(head.hex()) + assert heads == [GOLDEN["head1"], GOLDEN["head2"], GOLDEN["head3"]] + + +def test_chain_mirror_tracks_per_seq_heads(): + mirror = ChainMirror(GOLDEN_SESSION, GOLDEN_STREAM) + for event in GOLDEN_EVENTS: + mirror.append(event) + assert mirror.head_hex == GOLDEN["head3"] + assert mirror.head_at(0) == GOLDEN["head1"] + assert mirror.head_at(1) == GOLDEN["head2"] + assert mirror.head_at(2) == GOLDEN["head3"] + assert mirror.head_at(3) is None + + +def test_ppq_changes_the_event_hash_only_when_present(): + with_ppq = event_hash(GOLDEN_EVENTS[1]) + without = event_hash({**GOLDEN_EVENTS[1], "ppq": None}) + assert with_ppq != without + + +# --------------------------------------------------------------------------- +# Framings — exact bytes per 06 §2.3/§2.4 +# --------------------------------------------------------------------------- + + +def test_framing_bytes_are_newline_delimited_utf8(): + assert receipt_message("sess_9f2c", "pod", 42, "9c" * 32, 1751871234890) == ( + b"ddp-receipt:v1\nsess_9f2c\npod\n42\n" + b"9c" * 32 + b"\n1751871234890" + ) + assert key_cert_message("rk-1", "PUBB64", 100, 200) == ( + b"ddp-keycert:v1\nrk-1\nPUBB64\n100\n200" + ) + + +# --------------------------------------------------------------------------- +# §2.4 verification procedure with server-shaped key material +# --------------------------------------------------------------------------- + + +def _b64url(raw: bytes) -> str: + return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") + + +def _keypair() -> tuple[Ed25519PrivateKey, str]: + priv = Ed25519PrivateKey.generate() + pub = priv.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw) + return priv, _b64url(pub) + + +class _Ledger: + """Ephemeral key material + the server's exact signing behaviour + (provenance.keys.ts keysDocument / signReceipt).""" + + def __init__(self, *, now_ms: int | None = None): + self.now_ms = now_ms if now_ms is not None else int(time.time() * 1000) + self.sealing_priv, self.sealing_pub = _keypair() + self.receipt_priv, self.receipt_pub = _keypair() + self.kid = "rk-testcafe" + self.not_before = self.now_ms - 60_000 + self.not_after = self.now_ms + 48 * 3600_000 + + def keys_document(self, *, tamper_cert: bool = False) -> dict: + cert_sig = self.sealing_priv.sign( + key_cert_message( + self.kid, self.receipt_pub, self.not_before, self.not_after, + ) + ) + if tamper_cert: + cert_sig = bytes([cert_sig[0] ^ 0xFF]) + cert_sig[1:] + return { + "sealing_key": { + "kid": "sk-testcafe", "alg": "Ed25519", + "public_key": self.sealing_pub, + "not_before": self.not_before, + "not_after": self.not_after + 300 * 24 * 3600_000, + }, + "receipt_keys": [{ + "kid": self.kid, "alg": "Ed25519", + "public_key": self.receipt_pub, + "not_before": self.not_before, + "not_after": self.not_after, + "cert_sig": _b64url(cert_sig), + }], + } + + def sign_receipt( + self, session_id: str, stream: str, seq: int, + chain_head: str, ts: int, + ) -> dict: + sig = self.receipt_priv.sign( + receipt_message(session_id, stream, seq, chain_head, ts) + ) + return { + "seq": seq, "chain_head": chain_head, "ts": ts, + "receipt_kid": self.kid, "sig": _b64url(sig), + } + + +def _mirror_and_head() -> tuple[ChainMirror, str]: + mirror = ChainMirror(GOLDEN_SESSION, GOLDEN_STREAM) + for event in GOLDEN_EVENTS: + mirror.append(event) + return mirror, mirror.head_at(2) + + +def test_verify_accepts_a_server_shaped_receipt(): + ledger = _Ledger() + verifier = ReceiptVerifier("http://unused.invalid/v1/keys") + assert verifier.install_keys_document(ledger.keys_document()) == 1 + + mirror, head2 = _mirror_and_head() + receipt = ledger.sign_receipt( + GOLDEN_SESSION, GOLDEN_STREAM, 2, head2, ledger.now_ms, + ) + ok, reason = verifier.verify( + session_id=GOLDEN_SESSION, stream=GOLDEN_STREAM, + expected_head=mirror.head_at(receipt["seq"]), **receipt, + ) + assert (ok, reason) == (True, None) + + +def test_verify_rejects_every_tampered_element(): + ledger = _Ledger() + verifier = ReceiptVerifier("http://unused.invalid/v1/keys") + verifier.install_keys_document(ledger.keys_document()) + mirror, head2 = _mirror_and_head() + good = ledger.sign_receipt( + GOLDEN_SESSION, GOLDEN_STREAM, 2, head2, ledger.now_ms, + ) + + _MIRROR = object() # sentinel: "look the head up in the mirror" + + def check(receipt: dict, expected_head: object = _MIRROR) -> str: + ok, reason = verifier.verify( + session_id=GOLDEN_SESSION, stream=GOLDEN_STREAM, + expected_head=( + mirror.head_at(receipt["seq"]) + if expected_head is _MIRROR else expected_head + ), + **receipt, + ) + assert not ok, f"tampered receipt must not verify: {receipt}" + return reason + + # Tampered signature. + bad_sig = _b64url(bytes(64)) + assert "signature" in check({**good, "sig": bad_sig}) + # Tampered chain_head: fails step 5 BEFORE any signature check — + # a ledger claiming a different history is caught even if it signs it. + forged_head = "0" * 64 + forged = ledger.sign_receipt( + GOLDEN_SESSION, GOLDEN_STREAM, 2, forged_head, ledger.now_ms, + ) + assert "chain_head mismatch" in check(forged) + # Tampered ts: the framing no longer matches what was signed. + assert "signature" in check({**good, "ts": good["ts"] + 1}) + # ts outside the receipt key's validity window (step 3). + late_ts = ledger.not_after + 1000 + late = ledger.sign_receipt( + GOLDEN_SESSION, GOLDEN_STREAM, 2, head2, late_ts, + ) + assert "validity" in check(late) + # Unknown kid. + assert "kid" in check({**good, "receipt_kid": "rk-nope"}) + # A receipt for a seq we never submitted (no local head). + assert "no local chain head" in check(good, expected_head=None) + # Malformed fields never raise. + assert "malformed" in check({**good, "seq": "2"}) + + +def test_tampered_key_cert_disables_the_receipt_key(): + ledger = _Ledger() + verifier = ReceiptVerifier("http://unused.invalid/v1/keys") + # A cert_sig that does not verify with the sealing key must drop the + # receipt key entirely (steps 1–2): receipts then fail on kid lookup. + assert verifier.install_keys_document( + ledger.keys_document(tamper_cert=True) + ) == 0 + mirror, head2 = _mirror_and_head() + receipt = ledger.sign_receipt( + GOLDEN_SESSION, GOLDEN_STREAM, 2, head2, ledger.now_ms, + ) + ok, reason = verifier.verify( + session_id=GOLDEN_SESSION, stream=GOLDEN_STREAM, + expected_head=mirror.head_at(2), **receipt, + ) + assert not ok + assert "kid" in reason + + +# --------------------------------------------------------------------------- +# LedgerClient wiring: verify receipts as they arrive, fail-open +# --------------------------------------------------------------------------- + + +class _SigningLedgerServer: + """A stub ledger that really chains (§2.1 rules) and really signs + (§2.3 receipts, §2.4 key document). ``tamper`` switches it to + returning receipts for a forged history.""" + + def __init__(self, *, tamper: str | None = None): + self.ledger = _Ledger() + self.tamper = tamper + outer = self + heads: dict[str, bytes] = {} + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *a): # silence + pass + + def _send(self, payload: dict): + raw = json.dumps(payload).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + def do_GET(self): + if not self.path.endswith("/keys"): + self.send_error(404) + return + self._send(outer.ledger.keys_document()) + + def do_POST(self): + session_id = self.path.split("/sessions/")[1].split("/")[0] + length = int(self.headers.get("Content-Length", 0)) + body = json.loads(self.rfile.read(length)) + head = heads.get(session_id) or genesis_head(session_id, "pod") + receipts = [] + for event in body["events"]: + head = advance_head(head, event_hash(event)) + if event["type"].startswith("slice."): + chain_head = head.hex() + if outer.tamper == "chain_head": + chain_head = "f" * 64 + receipt = outer.ledger.sign_receipt( + session_id, "pod", event["seq"], chain_head, + outer.ledger.now_ms, + ) + if outer.tamper == "sig": + receipt["sig"] = _b64url(bytes(64)) + receipts.append(receipt) + heads[session_id] = head + self._send({ + "stream": "pod", "chain_head": head.hex(), + "receipts": receipts, + }) + + self._server = HTTPServer(("127.0.0.1", 0), Handler) + self._thread = threading.Thread( + target=self._server.serve_forever, daemon=True, + ) + self._thread.start() + + @property + def base_url(self) -> str: + return f"http://127.0.0.1:{self._server.server_address[1]}/v1" + + def close(self): + self._server.shutdown() + self._server.server_close() + + +def _run_session(server: _SigningLedgerServer, session_id: str) -> LedgerClient: + client = LedgerClient( + base_url=server.base_url, session_id=session_id, token="dlt_pod_x", + ) + try: + client.post_event("session.config", {"model": "acestep-1.5"}, ts=1000) + client.post_event( + "action.param", {"name": "cutoff", "value": 0.42}, + ts=1001, ppq=3840.25, + ) + client.post_slice_hash( + sha256="cd" * 32, start_sample=0, num_samples=96000, + channels=2, slice_seq=0, ts=1002, + ) + client.post_slice_hash( + sha256="ef" * 32, start_sample=96000, num_samples=96000, + channels=2, slice_seq=1, ts=1003, + ) + client.close(timeout=5.0) + finally: + server.close() + return client + + +def test_ledger_client_verifies_receipts_end_to_end(): + server = _SigningLedgerServer() + client = _run_session(server, "sess_e2e_ok") + # Both slice receipts arrived and passed the full §2.4 procedure — + # including step 5 against the client's own chain mirror. + assert client.receipts_verified == 2 + assert client.receipts_unverified == 0 + assert client.last_verification_failure is None + + +def test_ledger_client_counts_forged_chain_head_receipts(): + server = _SigningLedgerServer(tamper="chain_head") + client = _run_session(server, "sess_e2e_forged") + # The ledger signed a history that is not ours: step 5 must catch it, + # fail-open (reporting kept running; nothing raised). + assert client.receipts_verified == 0 + assert client.receipts_unverified == 2 + assert "chain_head mismatch" in client.last_verification_failure + + +def test_ledger_client_counts_bad_signature_receipts(): + server = _SigningLedgerServer(tamper="sig") + client = _run_session(server, "sess_e2e_badsig") + assert client.receipts_verified == 0 + assert client.receipts_unverified == 2 + assert "signature" in client.last_verification_failure + + +def test_ledger_client_fails_open_when_keys_endpoint_is_down(): + # Server chains + signs correctly but the key document is + # unreachable (404) → the verifier cannot certify any receipt key; + # receipts count as unverified and the reporter keeps working. + server = _SigningLedgerServer() + client = LedgerClient( + base_url=server.base_url, session_id="sess_e2e_nokeys", + token="dlt_pod_x", + ) + # Pre-seed the lazily built verifier with a keys URL that 404s. + client._verifier = ReceiptVerifier(f"{server.base_url}/nope-keys") + try: + client.post_slice_hash( + sha256="ab" * 32, start_sample=0, num_samples=96000, + channels=2, slice_seq=0, ts=1002, + ) + client.close(timeout=5.0) + finally: + server.close() + assert client.receipts_verified == 0 + assert client.receipts_unverified == 1 + assert "kid" in client.last_verification_failure diff --git a/tests/unit/test_provenance_session_log.py b/tests/unit/test_provenance_session_log.py new file mode 100644 index 00000000..130d4bec --- /dev/null +++ b/tests/unit/test_provenance_session_log.py @@ -0,0 +1,387 @@ +"""Local JSONL session log: bus tap wiring, §2.2 envelope schema, slice +counting, and the batched ledger client (spec 06 §2.3). + +Pure Python — no GPU, no c2pa. Drives a real +:class:`acestep.streaming.events.EventBus` through the registry hook +(:func:`acestep.streaming.registry.register` attaches the tap when the +handle carries a ``bus``) and asserts the on-disk JSONL record uses the +shared ledger schema: ``{stream:"local", seq, type, ts(ms), payload}`` +envelopes with namespaced type names, a ``session.config`` opener, and a +``session.note`` seal. + +Determinism: :meth:`SessionLogTap.close` (via ``unregister``) +unsubscribes and joins the drainer, and the drainer delivers all +already-queued events before exiting — so once ``unregister`` returns, +every published event is on disk and the seal is the last line. +""" + +from __future__ import annotations + +import json +import sys +import threading +import time +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path + +import numpy as np + +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +import acestep.provenance.session_log as session_log_mod +from acestep.provenance.ledger_client import LedgerClient +from acestep.provenance.session_log import ( + LOCAL_STREAM, + get_tap, + latest_session_summary, + record_pod_slice_hash, + record_user_action, + session_summary_for, +) +from acestep.streaming import registry +from acestep.streaming.events import ( + AudioReady, + EventBus, + PromptApplied, + SwapFailed, +) + + +def _slice(tag: int) -> AudioReady: + audio = np.full((16, 2), float(tag), dtype=np.float32) + return AudioReady( + audio=audio, start_sample=tag * 16, num_samples=16, channels=2, + tick_ms=0.0, dec_ms=0.0, num_gens=tag, params={}, + ) + + +def _register(bus: EventBus, session_id: str, **kwargs) -> registry.SessionHandle: + handle = registry.SessionHandle( + id=session_id, + started_at=time.time(), + inject=lambda data, audio: None, + snapshot=kwargs.pop("snapshot", lambda: {}), + bus=bus, + provenance_meta=kwargs.pop("provenance_meta", None), + ) + assert not kwargs + registry.register(handle) + return handle + + +def _read_log(tmp_path: Path, session_id: str) -> list[dict]: + p = tmp_path / "provenance" / "sessions" / f"{session_id}.jsonl" + assert p.is_file(), f"missing session log at {p}" + return [json.loads(line) for line in p.read_text(encoding="utf-8").splitlines()] + + +def _by_type(lines: list[dict], type_: str) -> list[dict]: + return [l for l in lines if l["type"] == type_] + + +def _notes(lines: list[dict], note: str) -> list[dict]: + return [l for l in lines + if l["type"] == "session.note" and l["payload"].get("note") == note] + + +def test_session_log_schema_and_lifecycle(tmp_path, monkeypatch): + monkeypatch.setenv("ACESTEP_PROVENANCE_DIR", str(tmp_path / "provenance")) + bus = EventBus() + snapshot = lambda: { # noqa: E731 — mirrors adapter snapshot shape + "lora_catalog": [ + {"id": "lead-guitar", "state": "enabled"}, + {"id": "muted", "state": "disabled"}, + ], + "prompt": "warm analog dub", + "bpm": 120, + "key": "C minor", + "time_signature": "4", + } + _register( + bus, "sess-schema", + snapshot=snapshot, + provenance_meta={"checkpoint": "ace_step_v1", "fixture_name": "loop60"}, + ) + + bus.publish(PromptApplied(tags="dark techno")) + bus.publish(SwapFailed(error="no such fixture")) + record_user_action("sess-schema", "prompt", {"type": "prompt", "tags": "dub 2"}) + registry.unregister("sess-schema") + + lines = _read_log(tmp_path, "sess-schema") + + # Every line carries the shared §2.2 envelope: a "local" stream label, + # a contiguous-from-0 seq, a namespaced string type, an integer + # epoch-ms ts (never an ISO string), and a payload object. + for i, line in enumerate(lines): + assert line["stream"] == LOCAL_STREAM + assert line["seq"] == i + assert isinstance(line["type"], str) and "." in line["type"] + assert isinstance(line["ts"], int) + assert line["ts"] > 1_600_000_000_000 # plausibly ms, not seconds + assert isinstance(line["payload"], dict) + + # session.config is first and carries the identity fields in payload. + start = lines[0] + assert start["type"] == "session.config" + cfg = start["payload"] + assert cfg["model"] == "ace_step_v1" + assert cfg["loras"] == ["lead-guitar"] + assert cfg["fixture_name"] == "loop60" + assert cfg["prompt"] == "warm analog dub" + assert cfg["bpm"] == 120 + + # Prompts (bus PromptApplied + user action) both map to action.prompt. + prompts = _by_type(lines, "action.prompt") + assert {p["payload"].get("prompt") for p in prompts} >= {"dark techno"} + user_prompt = [p for p in prompts if p["payload"].get("tags") == "dub 2"] + assert len(user_prompt) == 1 + assert user_prompt[0]["payload"]["source"] == "ws" + + # A failure lands as a session.note error with the kind + message. + errors = _notes(lines, "error") + assert len(errors) == 1 + assert errors[0]["payload"]["kind"] == "SwapFailed" + assert errors[0]["payload"]["error"] == "no such fixture" + + # session_end seals the log with counts (never prompt content). + end = lines[-1] + assert end["type"] == "session.note" + assert end["payload"]["note"] == "session_end" + assert end["payload"]["slices"] == 0 + assert end["payload"]["slice_hashes"] == 0 + summary = end["payload"]["timeline_summary"] + assert summary["prompt_changes"] == 2 # bus PromptApplied + user action + assert summary["user_actions"] == 1 + assert summary["distinct_prompts"] == 3 # snapshot + applied + user action + # The seal's summary is computed before the session_end line itself + # is counted. + assert summary["events"] == len(lines) - 1 + assert summary["duration_s"] >= 0.0 + + # No masquerading float32 slice-hash chain remains (audit F3). + assert not _by_type(lines, "output_chain_checkpoint") + assert all("chain_head" not in l["payload"] for l in lines) + + # Detach removed the tap from the process-global registry. + assert get_tap("sess-schema") is None + assert latest_session_summary() is None + assert session_summary_for("sess-schema") is None + + +def test_slice_counting_and_pod_slice_hash(tmp_path, monkeypatch): + monkeypatch.setenv("ACESTEP_PROVENANCE_DIR", str(tmp_path / "provenance")) + bus = EventBus() + _register(bus, "sess-slice") + + # Decoded slices on the bus are counted, not hashed (the §2.3 hash is + # over the transport codec's float16 downlink bytes, reported + # separately via record_pod_slice_hash). + for i in range(6): + bus.publish(_slice(i)) + for seq in range(3): + record_pod_slice_hash( + "sess-slice", + sha256="ab" * 32, + start_sample=seq * 16, + num_samples=16, + channels=2, + slice_seq=seq, + ) + registry.unregister("sess-slice") + + lines = _read_log(tmp_path, "sess-slice") + # Per-slice hashes go to the ledger, not the local JSONL (anti-bloat). + assert not _by_type(lines, "slice.pod_hash") + + end = lines[-1] + assert end["payload"]["note"] == "session_end" + assert end["payload"]["slices"] == 6 + assert end["payload"]["slice_hashes"] == 3 + assert end["payload"]["timeline_summary"]["slices"] == 6 + assert end["payload"]["timeline_summary"]["slice_hashes"] == 3 + + +def test_params_actions_are_diffed_and_rate_limited(tmp_path, monkeypatch): + monkeypatch.setenv("ACESTEP_PROVENANCE_DIR", str(tmp_path / "provenance")) + bus = EventBus() + _register(bus, "sess-params") + tap = get_tap("sess-params") + assert tap is not None + + # First params action logs only the changed knobs; telemetry keys + # are stripped. + record_user_action( + "sess-params", "params", + {"type": "params", "steer": 0.5, "flow": 0.1, "playback_pos": 12.5}, + ) + # Inside the rate-limit window: suppressed even though a knob moved. + record_user_action("sess-params", "params", {"steer": 0.9}) + # Identical values are suppressed regardless of the window. + monkeypatch.setattr(session_log_mod.time, "monotonic", lambda: 1e9) + record_user_action( + "sess-params", "params", {"steer": 0.5, "flow": 0.1}, + ) + # Outside the window with a real change: logs the diff only. + record_user_action("sess-params", "params", {"steer": 0.7, "flow": 0.1}) + + registry.unregister("sess-params") + lines = _read_log(tmp_path, "sess-params") + actions = _by_type(lines, "action.param") + assert [a["payload"]["changed"] for a in actions] == [ + {"steer": 0.5, "flow": 0.1}, + {"steer": 0.7}, + ] + assert all(a["payload"]["source"] == "ws" for a in actions) + assert lines[-1]["payload"]["timeline_summary"]["param_changes"] == 2 + + +def test_registry_hook_is_inert_without_bus(tmp_path, monkeypatch): + monkeypatch.setenv("ACESTEP_PROVENANCE_DIR", str(tmp_path / "provenance")) + handle = registry.SessionHandle( + id="sess-no-bus", + started_at=time.time(), + inject=lambda data, audio: None, + snapshot=lambda: {}, + ) + registry.register(handle) + registry.unregister("sess-no-bus") + + assert get_tap("sess-no-bus") is None + assert not (tmp_path / "provenance" / "sessions").exists() + + +# --------------------------------------------------------------------------- +# Batched ledger client (spec 06 §2.3): wire shape, auth, seq, receipts +# --------------------------------------------------------------------------- + + +class _CapturingLedger: + """A throwaway HTTP server that captures ingestion requests and + answers with a §2.3-shaped receipt response.""" + + def __init__(self): + self.requests: list[dict] = [] + outer = self + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *a): # silence + pass + + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + body = json.loads(self.rfile.read(length) or b"{}") + outer.requests.append({ + "path": self.path, + "auth": self.headers.get("Authorization"), + "content_type": self.headers.get("Content-Type"), + "body": body, + }) + last_seq = body["events"][-1]["seq"] + resp = json.dumps({ + "stream": "pod", + "chain_head": "9c" * 32, + "receipts": [{ + "seq": last_seq, + "chain_head": "9c" * 32, + "ts": 1751871234890, + "receipt_kid": "rk-2026-07-07", + "sig": "base64url-signature", + }], + }).encode("utf-8") + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(resp))) + self.end_headers() + self.wfile.write(resp) + + self._server = HTTPServer(("127.0.0.1", 0), Handler) + self.port = self._server.server_address[1] + self._thread = threading.Thread( + target=self._server.serve_forever, daemon=True, + ) + self._thread.start() + + @property + def base_url(self) -> str: + return f"http://127.0.0.1:{self.port}/v1" + + def close(self): + self._server.shutdown() + self._server.server_close() + + +def test_ledger_client_noop_without_url_or_token(): + # No config at all → disabled, no thread, no crash. + c = LedgerClient(base_url="", session_id="s", token="") + assert not c.enabled + c.post_event("action.prompt", {"prompt": "x"}) + c.post_slice_hash(sha256="ab" * 32, start_sample=0, num_samples=16, channels=2) + c.close() + assert c._worker is None + # URL but no token is still a no-op (the pod token is mandatory). + c2 = LedgerClient(base_url="http://x/v1", session_id="s", token=None) + assert not c2.enabled + + +def test_ledger_client_batches_events_with_auth_and_contiguous_seq(): + server = _CapturingLedger() + try: + client = LedgerClient( + base_url=server.base_url, session_id="sess_9f2c", token="dlt_pod_abc", + ) + assert client.enabled + client.post_event("session.config", {"model": "acestep-1.5"}, ts=1000) + client.post_event("action.prompt", {"prompt": "warm keys"}, ts=1001) + # A slice-hash report forces a prompt flush and yields a receipt. + client.post_slice_hash( + sha256="cd" * 32, start_sample=4177920, num_samples=96000, + channels=2, slice_seq=812, ts=1002, + ) + client.close(timeout=5.0) + finally: + server.close() + + assert len(server.requests) >= 1 + req = server.requests[0] + # Path + auth per §2.3. + assert req["path"] == "/v1/sessions/sess_9f2c/events" + assert req["auth"] == "Bearer dlt_pod_abc" + assert req["content_type"] == "application/json" + # Batched envelope with contiguous seq from 0. + events = req["body"]["events"] + assert [e["seq"] for e in events] == list(range(len(events))) + types = [e["type"] for e in events] + assert types[:2] == ["session.config", "action.prompt"] + # The slice.pod_hash event carries the §2.3 payload. + slice_ev = [e for e in events if e["type"] == "slice.pod_hash"][-1] + assert slice_ev["payload"] == { + "start_sample": 4177920, "num_samples": 96000, "channels": 2, + "sha256": "cd" * 32, "slice_seq": 812, + } + # Every event uses integer epoch-ms ts. + assert all(isinstance(e["ts"], int) for e in events) + + +def test_ledger_client_parses_receipt_field_names(): + server = _CapturingLedger() + try: + client = LedgerClient( + base_url=server.base_url, session_id="s", token="dlt_pod_x", + ) + client.post_slice_hash( + sha256="ef" * 32, start_sample=0, num_samples=16, channels=2, + slice_seq=0, + ) + client.close(timeout=5.0) + finally: + server.close() + + r = client.last_receipt + assert r is not None + # Correct wire field names (sig / receipt_kid), not the old ones. + assert r.sig == "base64url-signature" + assert r.receipt_kid == "rk-2026-07-07" + assert r.chain_head == "9c" * 32 + assert r.ts == 1751871234890 + assert client.chain_head == "9c" * 32