"""Verifyum client, single file, standard library only. Creates a privacy-preserving proof that exact bytes existed at a point in time. The data never leaves the machine: only a domain-separated commitment is sent. No account, wallet or API key is required. from verifyum import anchor_bytes, anchor_file, anchor_record, verify proof = anchor_file("contract.pdf") print(proof["proof_url"]) Keep the returned draft. It holds the nonce, and without it nobody can link the original data to the public proof. The module also carries the Witness Layer primitives (canonical JSON, checkpoint hashes, Merkle path verification, service signature check) so a proof can be verified against the published witness checkpoints offline. Reference: https://verifyum.com/agents """ from __future__ import annotations import argparse import base64 import binascii import calendar import hashlib import json import os import re import secrets import sys import time import urllib.error import urllib.request __all__ = [ "commitment_for", "anchor_bytes", "anchor_file", "anchor_record", "verify", "VerifyumError", "digest", "digest_bytes", "is_digest", "jcs", "checkpoint_hash", "checkpoint_document", "checkpoint_document_digest", "sigsum_checksum", "proof_leaf_hash", "checkpoint_leaf_hash", "node_hash", "build_tree", "verify_path", "base64url_decode", "verify_service_signature", "verify_witness", "verify_witness_documents", "verify_witness_chain_documents", ] API_BASE = os.environ.get("VERIFYUM_API_BASE", "https://api.verifyum.com") PROOF_DOMAIN = os.environ.get("VERIFYUM_PROOF_DOMAIN", "verifyum.com") USER_AGENT = "verifyum-python/1.4.0" # What verify_witness checks, and what it does not. It once reported # `daily_witnessed` whenever a daily chain was present and said the checkpoint # was the one the external witnesses hold; a locally built daily checkpoint # that was never published passed as witnessed. Checkpoints are unsigned, so # only the channel receipts show external witnessing, and nothing here reads # them. WITNESS_CHAIN_BOUNDARY = ( "Shows that the signed proof metadata is included, by Merkle path, in the " "checkpoint these documents describe, and in the daily checkpoint when a " "chain is given. It does not show that any external witness observed a " "checkpoint: checkpoints are unsigned, and the channel receipts at the " "receipts URLs are what show that. Not authorship, ownership, or that the " "contents are true." ) WITNESS_OFFLINE_SCOPE = ( "Offline, the witness checks show that the receipt's checkpoints, Merkle " "paths and signed metadata agree with each other, that the service " "signature verifies against the registry, and that no checkpoint predates " "the proof. Checkpoints are unsigned, so whether a checkpoint was published " "and witnessed is shown by the published checkpoint document or a channel " "receipt at the URLs the receipt names, or by the ledger for the anchor; " "not by an offline check." ) # Ed25519 verification is built in, so "standard library only" is true of the # whole verifier and not only of the hashing. cryptography is used when it is # installed because it is far faster; the result is the same either way. An # earlier version treated the signature as unchecked when cryptography was # absent, and still reported the proof valid, which meant a bare install # printed success for a check that had not run. try: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey as _Ed25519PublicKey from cryptography.exceptions import InvalidSignature as _InvalidSignature except Exception: # pragma: no cover _Ed25519PublicKey = None _InvalidSignature = None # Ed25519 verification, RFC 8032, pure Python. Verification handles only # public values, so nothing here needs to be constant time; signing would be # a different matter and is deliberately absent. _P = 2 ** 255 - 19 _L = 2 ** 252 + 27742317777372353535851937790883648493 _D = -121665 * pow(121666, _P - 2, _P) % _P _I = pow(2, (_P - 1) // 4, _P) def _recover_x(y, sign): if y >= _P: return None xx = (y * y - 1) * pow(_D * y * y + 1, _P - 2, _P) % _P x = pow(xx, (_P + 3) // 8, _P) if (x * x - xx) % _P != 0: x = x * _I % _P if (x * x - xx) % _P != 0: return None if x == 0 and sign: return None return _P - x if x & 1 != sign else x def _add(p, q): x1, y1, z1, t1 = p x2, y2, z2, t2 = q a = (y1 - x1) * (y2 - x2) % _P b = (y1 + x1) * (y2 + x2) % _P c = 2 * t1 * t2 * _D % _P d = 2 * z1 * z2 % _P e, f, g, h = b - a, d - c, d + c, b + a return (e * f % _P, g * h % _P, f * g % _P, e * h % _P) def _mul(point, scalar): result = (0, 1, 1, 0) while scalar > 0: if scalar & 1: result = _add(result, point) point = _add(point, point) scalar >>= 1 return result _GY = 4 * pow(5, _P - 2, _P) % _P _GX = _recover_x(_GY, 0) _G = (_GX, _GY, 1, _GX * _GY % _P) def _decompress(data): if len(data) != 32: return None y = int.from_bytes(data, "little") sign = y >> 255 y &= (1 << 255) - 1 x = _recover_x(y, sign) return None if x is None else (x, y, 1, x * y % _P) def _equal(p, q): x1, y1, z1, _ = p x2, y2, z2, _ = q return (x1 * z2 - x2 * z1) % _P == 0 and (y1 * z2 - y2 * z1) % _P == 0 def ed25519_verify(public_key, signature, message): """RFC 8032 verification. Returns False on any malformed input.""" if len(public_key) != 32 or len(signature) != 64: return False point = _decompress(public_key) if point is None: return False r = _decompress(signature[:32]) if r is None: return False s = int.from_bytes(signature[32:], "little") if s >= _L: return False challenge = int.from_bytes( hashlib.sha512(signature[:32] + public_key + message).digest(), "little" ) % _L return _equal(_mul(_G, s), _add(r, _mul(point, challenge))) def ed25519_check(public_key: bytes, signature: bytes, message: bytes) -> bool: """Verifies an Ed25519 signature, by whichever implementation is present.""" if _Ed25519PublicKey is not None: try: _Ed25519PublicKey.from_public_bytes(public_key).verify(signature, message) return True except _InvalidSignature: return False except (ValueError, TypeError): return False return ed25519_verify(public_key, signature, message) # Always used with fullmatch(): with match() a trailing "$" still matches # before a final newline, so "sha256:\n" would pass as a digest string. DIGEST_RE = re.compile(r"sha256:[0-9a-f]{64}") PROOF_ID_RE = re.compile(r"[0-7][0-9a-hjkmnp-tv-z]{25}") KEY_RE = re.compile(r"[\x20-\x7e]+") BASE64URL_RE = re.compile(r"[A-Za-z0-9_-]*") BATCH_TIME_RE = re.compile(r"[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z") CHECKPOINT_PREFIX = b"verifyum:witness:checkpoint:v1\n" PROOF_LEAF_PREFIX = b"\x00verifyum:witness:proof-leaf:v1\n" CHECKPOINT_LEAF_PREFIX = b"\x00verifyum:witness:checkpoint-leaf:v1\n" NODE_PREFIX = b"\x01verifyum:witness:node:v1\n" CHECKPOINT_URL_PREFIX = f"https://{PROOF_DOMAIN}/witness/checkpoints/" RECEIPTS_URL_PREFIX = f"https://{PROOF_DOMAIN}/witness/receipts/" class VerifyumError(RuntimeError): """A Verifyum request failed. `retry_after` is set when it is worth retrying.""" def __init__(self, message: str, *, status: int | None = None, retry_after: int | None = None): super().__init__(message) self.status = status self.retry_after = retry_after # --- R1, R2: digest and canonical JSON ------------------------------------- def digest(payload: bytes) -> str: """R1: digest string over raw bytes.""" if not isinstance(payload, (bytes, bytearray)): raise TypeError("digest expects bytes") return "sha256:" + hashlib.sha256(bytes(payload)).hexdigest() def is_digest(value) -> bool: """True only for "sha256:" plus exactly 64 lowercase hex characters.""" return isinstance(value, str) and DIGEST_RE.fullmatch(value) is not None def digest_bytes(value: str) -> bytes: """The 32 raw bytes behind a digest string. Rejects anything else.""" if not is_digest(value): raise ValueError("not a digest string") return bytes.fromhex(value[7:]) def _check_jcs_value(value, path: str = "$") -> None: # Floats never appear in a Verifyum document; a float here means the # caller parsed something that was not canonical, so refuse to hash it. if value is None or isinstance(value, bool) or isinstance(value, str): return if isinstance(value, int): return if isinstance(value, float): raise ValueError(f"float at {path} is not allowed in canonical JSON") if isinstance(value, (list, tuple)): for index, item in enumerate(value): _check_jcs_value(item, f"{path}[{index}]") return if isinstance(value, dict): if not value: # The reference cannot tell {} from [] and no document has one. raise ValueError(f"empty object at {path} is not allowed in canonical JSON") for key, item in value.items(): if not isinstance(key, str) or not KEY_RE.fullmatch(key): raise ValueError(f"object key {key!r} at {path} is not printable ASCII") _check_jcs_value(item, f"{path}.{key}") return raise ValueError(f"unsupported value type {type(value).__name__} at {path}") def jcs(value) -> bytes: """R2: canonical JSON bytes, matching the PHP reference byte for byte.""" _check_jcs_value(value) text = json.dumps(value, separators=(",", ":"), ensure_ascii=False, sort_keys=True) # The reference escapes the two Unicode line terminators; keys are ASCII # only, so a plain replace can only touch string contents. text = text.replace("\u2028", "\\u2028").replace("\u2029", "\\u2029") return text.encode("utf-8") # --- R3 to R7: checkpoint and tree hashes ----------------------------------- def checkpoint_hash(checkpoint: dict) -> str: """R3: hash over the checkpoint with only `checkpoint_hash` removed.""" body = {key: value for key, value in checkpoint.items() if key != "checkpoint_hash"} return digest(CHECKPOINT_PREFIX + jcs(body)) def checkpoint_document(checkpoint: dict) -> bytes: """R4: the published checkpoint file, canonical JSON plus one newline.""" return jcs(checkpoint) + b"\n" def checkpoint_document_digest(checkpoint: dict) -> str: """R4: what OpenTimestamps, Sigsum, eIDAS and the others witness.""" return digest(checkpoint_document(checkpoint)) def sigsum_checksum(daily_checkpoint: dict) -> str: """Sigsum leaf checksum: sha256 over the 32 raw bytes of the document digest.""" return hashlib.sha256(digest_bytes(checkpoint_document_digest(daily_checkpoint))).hexdigest() def proof_leaf_hash(metadata: dict) -> str: """R5: hourly tree leaf over the full metadata, signature included.""" return digest(PROOF_LEAF_PREFIX + jcs(metadata)) def checkpoint_leaf_hash(hourly_checkpoint: dict) -> str: """R6: daily tree leaf over the raw bytes of an hourly checkpoint hash.""" if hourly_checkpoint.get("kind") != "hourly": raise ValueError("checkpoint leaves are only defined for hourly checkpoints") return digest(CHECKPOINT_LEAF_PREFIX + digest_bytes(hourly_checkpoint["checkpoint_hash"])) def node_hash(left: str, right: str) -> str: """R7: interior node over two digest strings.""" return digest(NODE_PREFIX + digest_bytes(left) + digest_bytes(right)) def build_tree(subjects) -> dict: """R8: Merkle tree over `[{id, leaf_hash}, ...]`, sorted by id. Returns `{"root", "leaves": [{"id", "leaf_hash", "leaf_index", "path"}]}` in sorted order. An odd trailing node is promoted unchanged to the next level (no duplication, no self-hash, no path step for the subjects beneath it at that level). """ if not isinstance(subjects, (list, tuple)) or not subjects: raise ValueError("tree needs at least one subject") entries = [] for subject in subjects: if not isinstance(subject, dict): raise ValueError("subject must be an object") subject_id = subject.get("id") leaf_hash = subject.get("leaf_hash") if not isinstance(subject_id, str) or subject_id == "": raise ValueError("subject id must be a non-empty string") if not is_digest(leaf_hash): raise ValueError(f"subject {subject_id!r} has no digest leaf_hash") entries.append((subject_id.encode("utf-8"), subject_id, leaf_hash)) entries.sort(key=lambda entry: entry[0]) for previous, current in zip(entries, entries[1:]): if previous[0] == current[0]: raise ValueError(f"duplicate subject id {current[1]!r}") leaves = [ {"id": subject_id, "leaf_hash": leaf_hash, "leaf_index": index, "path": []} for index, (_, subject_id, leaf_hash) in enumerate(entries) ] # Each level node carries the list of leaf indexes beneath it. level = [(leaf["leaf_hash"], [index]) for index, leaf in enumerate(leaves)] while len(level) > 1: next_level = [] for position in range(0, len(level) - 1, 2): left_hash, left_members = level[position] right_hash, right_members = level[position + 1] for index in left_members: leaves[index]["path"].append({"side": "right", "hash": right_hash}) for index in right_members: leaves[index]["path"].append({"side": "left", "hash": left_hash}) next_level.append((node_hash(left_hash, right_hash), left_members + right_members)) if len(level) % 2 == 1: next_level.append(level[-1]) level = next_level return {"root": level[0][0], "leaves": leaves} def verify_path(leaf_hash: str, path, root: str) -> bool: """R9: walk the sibling path from the leaf and compare with the root.""" if not is_digest(leaf_hash) or not is_digest(root): return False if not isinstance(path, list) or len(path) > 64: return False current = leaf_hash for step in path: if not isinstance(step, dict) or set(step.keys()) != {"side", "hash"}: return False side = step["side"] sibling = step["hash"] if side not in ("left", "right"): return False if not is_digest(sibling): return False current = node_hash(sibling, current) if side == "left" else node_hash(current, sibling) return current == root # --- R10: service signature -------------------------------------------------- def base64url_decode(value: str) -> bytes: """Strict base64url: URL alphabet, no padding, canonical trailing bits.""" if not isinstance(value, str) or not BASE64URL_RE.fullmatch(value): raise ValueError("base64url: invalid alphabet") if len(value) % 4 == 1: raise ValueError("base64url: invalid length") padded = value + "=" * (-len(value) % 4) try: raw = base64.b64decode(padded.replace("-", "+").replace("_", "/"), validate=True) except (binascii.Error, ValueError) as error: raise ValueError("base64url: decoding failed") from error # Python does not reject non-zero unused trailing bits; the round trip does. if base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") != value: raise ValueError("base64url: non-canonical encoding") return raw def _select_service_key(registry: dict, key_id: str, network: str) -> bytes: matches = [] for entry in registry.get("keys") or []: if not isinstance(entry, dict): continue if ( entry.get("key_id") == key_id and entry.get("network") == network and entry.get("status") in ("active", "retired") and entry.get("algorithm") == "ed25519" and isinstance(entry.get("public_key"), str) ): matches.append(entry) if len(matches) != 1: raise ValueError("service key unavailable" if not matches else "service key not unique") public_key = base64url_decode(matches[0]["public_key"]) if len(public_key) != 32: raise ValueError("service key has wrong length") return public_key def verify_service_signature(metadata: dict, registry: dict) -> bool | None: """R10: Ed25519 over JCS(metadata minus service_signature). Returns True or False. It no longer returns None for a missing Ed25519 implementation, because there is always one. """ if not isinstance(metadata, dict) or not isinstance(registry, dict): return False signature = metadata.get("service_signature") if not isinstance(signature, dict) or set(signature.keys()) != {"algorithm", "key_id", "value"}: return False if signature["algorithm"] != "ed25519" or not isinstance(signature["key_id"], str): return False if not isinstance(signature["value"], str): return False anchor = metadata.get("anchor") if not isinstance(anchor, dict) or not isinstance(anchor.get("network"), str): return False try: public_key = _select_service_key(registry, signature["key_id"], anchor["network"]) raw_signature = base64url_decode(signature["value"]) if len(raw_signature) != 64: return False message = jcs({key: value for key, value in metadata.items() if key != "service_signature"}) return ed25519_check(public_key, raw_signature, message) except (ValueError, TypeError): return False # --- HTTP helpers ------------------------------------------------------------ def _request(method: str, path: str, payload: dict | None = None) -> dict: url = path if path.startswith("https://") else API_BASE + path body = json.dumps(payload).encode("utf-8") if payload is not None else None headers = {"User-Agent": USER_AGENT, "Accept": "application/json"} if body is not None: headers["Content-Type"] = "application/json" request = urllib.request.Request(url, data=body, headers=headers, method=method) try: with urllib.request.urlopen(request, timeout=30) as response: return {"status": response.status, "body": json.loads(response.read() or b"{}")} except urllib.error.HTTPError as error: retry_after = error.headers.get("Retry-After") if error.headers else None detail = {} try: detail = json.loads(error.read() or b"{}") except Exception: pass raise VerifyumError( detail.get("message") or detail.get("error") or f"HTTP {error.code}", status=error.code, retry_after=int(retry_after) if retry_after and retry_after.isdigit() else None, ) from None # --- Protocol v2 client ------------------------------------------------------ # The recipe for the commitment, carried inside every draft so the file can # explain itself years from now with no network and no Verifyum software. # It sits beside the manifest, never inside it: the manifest is what gets # hashed, so one more field in there would change every commitment. # tools/check-code.php lifts the prefix back out of this text and recomputes # the protocol vector from it, so the prose cannot drift from the code. DERIVATION = { "summary": "commitment = 'sha256:' + hex(SHA-256('verifyum:commitment:v2' + LF + JCS(manifest)))", "steps": [ "file.hash.value is SHA-256 of the original bytes in lowercase hex. file.size is the length in bytes, written as a decimal string.", "nonce is 32 random bytes in base64url without padding. It is the secret half: the commitment discloses nothing without it, and proves the tie to the file with it.", "Canonicalize the manifest with RFC 8785 (JCS): object keys sorted by UTF-16 code unit, no insignificant whitespace, UTF-8 output.", "Prepend 'verifyum:commitment:v2' and one newline (0x0A) to those canonical bytes, then take SHA-256 of the whole.", "Write the result as 'sha256:' followed by that digest in lowercase hex.", ], "privacy": "Only the commitment was sent to Verifyum. The file, its hash, its name and the nonce stayed on this machine.", "independence": "Any SHA-256 and any RFC 8785 implementation reproduces this commitment from this file alone, with no network and no Verifyum software.", } def commitment_for(data: bytes) -> tuple[str, dict]: """Returns (commitment, private_draft). The draft must be kept locally.""" nonce = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode().rstrip("=") manifest = { "file": { "hash": {"algorithm": "sha256", "value": hashlib.sha256(data).hexdigest()}, "size": str(len(data)), }, "nonce": nonce, "protocol": "verifyum", "version": 2, } commitment = "sha256:" + hashlib.sha256(b"verifyum:commitment:v2\n" + jcs(manifest)).hexdigest() draft = { "format": "verifyum-private-draft", "version": 2, "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "manifest": manifest, "commitment": commitment, "derivation": DERIVATION, } return commitment, draft def anchor_bytes(data: bytes, *, idempotency_key: str | None = None, wait: bool = True, timeout: int = 180) -> dict: """Anchors a commitment over `data`. Returns the proof with its private draft.""" commitment, draft = commitment_for(data) key = idempotency_key or "py-" + hashlib.sha256(commitment.encode()).hexdigest()[:32] health = _request("GET", "/health")["body"] if health.get("anchoring") != "enabled" or health.get("access") != "public": raise VerifyumError("Verifyum is not accepting public anchors right now", retry_after=300) result = _request("POST", "/v2/anchor", {"commitment": commitment, "idempotency_key": key})["body"] proof_id = result.get("proof_id") if wait and proof_id: deadline = time.time() + timeout while result.get("status") not in ("finalized", "failed") and time.time() < deadline: time.sleep(3) result = _request("GET", f"/v2/proofs/{proof_id}")["body"] draft["anchor_result"] = {k: result.get(k) for k in ("proof_id", "status", "network", "proof_url", "status_url")} return {**result, "commitment": commitment, "private_draft": draft} def anchor_file(path: str, **kwargs) -> dict: """Anchors the exact bytes of a local file.""" with open(path, "rb") as handle: return anchor_bytes(handle.read(), **kwargs) def anchor_record(record: dict, **kwargs) -> dict: """Anchors a decision record or any JSON document, canonicalized first. See https://verifyum.com/schema/agent-decision-record-v1.json for the recommended shape. Store the returned private_draft with the record. """ return anchor_bytes(jcs(record), **kwargs) def verify(proof_id: str, data: bytes, draft: dict) -> dict: """Checks that `data` matches the draft and that the proof is finalized. This confirms the local binding and Verifyum's published metadata. For a fully independent result, also fetch the Solana transaction named in the metadata from any RPC endpoint and compare its memo. """ manifest = draft["manifest"] checks = { "hash_matches": hashlib.sha256(data).hexdigest() == manifest["file"]["hash"]["value"], "size_matches": str(len(data)) == manifest["file"]["size"], "commitment_matches": "sha256:" + hashlib.sha256(b"verifyum:commitment:v2\n" + jcs(manifest)).hexdigest() == draft["commitment"], } metadata = _request("GET", f"https://{proof_id}.{PROOF_DOMAIN}/.well-known/verifyum.json")["body"] checks["public_commitment_matches"] = metadata.get("commitment") == draft["commitment"] checks["anchor_finalized"] = metadata.get("anchor", {}).get("status") == "finalized" return { "valid": all(checks.values()), "checks": checks, "transaction_signature": metadata.get("anchor", {}).get("transaction_signature"), "network": metadata.get("anchor", {}).get("network"), "boundary": "Shows that these exact bytes existed no later than the block time. " "Not authorship, ownership, or that the contents are true.", } # --- Witness Layer verification ---------------------------------------------- def _batch_id(checkpoint: dict) -> str | None: period_start = checkpoint.get("period_start") if not isinstance(period_start, str) or not BATCH_TIME_RE.fullmatch(period_start): return None if not is_digest(checkpoint.get("checkpoint_hash")): return None return period_start.replace("-", "").replace(":", "") + "-" + checkpoint["checkpoint_hash"][7:] def _parse_time(value) -> int | None: if not isinstance(value, str) or not BATCH_TIME_RE.fullmatch(value): return None try: return calendar.timegm(time.strptime(value, "%Y-%m-%dT%H:%M:%SZ")) except (ValueError, OverflowError): return None def _checkpoint_after_proof(checkpoint: dict, metadata: dict) -> bool: """A checkpoint cannot predate the proof it contains. Nothing signs a checkpoint, so offline a receipt could carry one written by its holder: an internally consistent hourly checkpoint dated 2020 verified as valid in both clients. The one bound the proof's own signed metadata gives is time. The worker creates a checkpoint after collecting the proof, so created_at is at or after submitted_at and the anchor's block time; the period itself is no bound, since a proof finalized at five past the hour lands in the checkpoint for the hour just ended. """ created = _parse_time(checkpoint.get("created_at")) submitted = _parse_time(metadata.get("submitted_at")) block_time = _parse_time((metadata.get("anchor") or {}).get("block_time")) return ( created is not None and submitted is not None and created >= submitted and (block_time is None or created >= block_time) ) _CHECKPOINT_FIELDS = { "schema", "protocol", "version", "kind", "network", "period_start", "period_end", "created_at", "algorithm", "subject_type", "subject_count", "merkle_root", "previous_checkpoint_hash", "checkpoint_hash", } _MEMBERSHIP_FIELDS = { "schema", "protocol", "version", "checkpoint_kind", "checkpoint_hash", "subject_type", "subject_id", "leaf_hash", "leaf_index", "leaf_count", "path", } _CHAIN_FIELDS = { "schema", "protocol", "version", "network", "proof_id", "proof_membership", "hourly_receipts_url", "daily_checkpoint_url", "daily_receipts_url", "daily_checkpoint", "daily_membership", } def _checkpoint_valid(checkpoint: dict, kind: str) -> bool: if not isinstance(checkpoint, dict) or set(checkpoint) != _CHECKPOINT_FIELDS: return False expected_subject = "proof-v2" if kind == "hourly" else "hourly-checkpoint-v1" if ( checkpoint.get("schema") != "https://verifyum.com/schema/witness-checkpoint-v1.json" or checkpoint.get("protocol") != "verifyum" or checkpoint.get("version") != 1 or checkpoint.get("kind") != kind or checkpoint.get("network") not in ("devnet", "mainnet-beta") or checkpoint.get("algorithm") != "verifyum-sha256-merkle-v1" or checkpoint.get("subject_type") != expected_subject or not isinstance(checkpoint.get("subject_count"), int) or isinstance(checkpoint.get("subject_count"), bool) or checkpoint["subject_count"] < 1 or not is_digest(checkpoint.get("merkle_root")) or not is_digest(checkpoint.get("checkpoint_hash")) or checkpoint.get("previous_checkpoint_hash") is not None and not is_digest(checkpoint.get("previous_checkpoint_hash")) ): return False for field in ("period_start", "period_end", "created_at"): if not isinstance(checkpoint.get(field), str) or not BATCH_TIME_RE.fullmatch(checkpoint[field]): return False try: start = calendar.timegm(time.strptime(checkpoint["period_start"], "%Y-%m-%dT%H:%M:%SZ")) end = calendar.timegm(time.strptime(checkpoint["period_end"], "%Y-%m-%dT%H:%M:%SZ")) created = calendar.timegm(time.strptime(checkpoint["created_at"], "%Y-%m-%dT%H:%M:%SZ")) except (ValueError, OverflowError): return False duration = 3600 if kind == "hourly" else 86400 return ( start % duration == 0 and end - start == duration and created >= end and checkpoint_hash(checkpoint) == checkpoint["checkpoint_hash"] ) def _membership_valid(membership: dict, checkpoint: dict) -> bool: if not isinstance(membership, dict) or set(membership) != _MEMBERSHIP_FIELDS: return False count = checkpoint.get("subject_count") index = membership.get("leaf_index") subject_id = membership.get("subject_id") subject_type = checkpoint.get("subject_type") if subject_type == "proof-v2": subject_valid = isinstance(subject_id, str) and PROOF_ID_RE.fullmatch(subject_id) is not None else: subject_valid = is_digest(subject_id) return ( membership.get("schema") == "https://verifyum.com/schema/witness-membership-v1.json" and membership.get("protocol") == "verifyum" and membership.get("version") == 1 and membership.get("checkpoint_kind") == checkpoint.get("kind") and membership.get("checkpoint_hash") == checkpoint.get("checkpoint_hash") and membership.get("subject_type") == subject_type and subject_valid and is_digest(membership.get("leaf_hash")) and isinstance(count, int) and not isinstance(count, bool) and membership.get("leaf_count") == count and isinstance(index, int) and not isinstance(index, bool) and 0 <= index < count and verify_path(membership["leaf_hash"], membership.get("path"), checkpoint.get("merkle_root")) ) def verify_witness_chain_documents(bundle: dict, chain: dict) -> dict: """Verifies the additive proof -> hourly -> daily bridge offline. It binds the daily leaf to the exact hourly checkpoint hash, validates both Merkle paths and accepts only deterministic Verifyum URLs. No member list is needed or exposed. """ checks = { "chain_shape": isinstance(chain, dict) and set(chain) == _CHAIN_FIELDS, } if not checks["chain_shape"]: return {"valid": False, "checks": checks} proof_membership = chain.get("proof_membership") hourly = proof_membership.get("checkpoint") if isinstance(proof_membership, dict) else None proof_path = proof_membership.get("membership") if isinstance(proof_membership, dict) else None daily = chain.get("daily_checkpoint") daily_path = chain.get("daily_membership") try: embedded_bundle_matches = ( isinstance(bundle, dict) and isinstance(proof_membership, dict) and jcs(proof_membership) == jcs(bundle) ) except (TypeError, ValueError): embedded_bundle_matches = False checks["chain_identity"] = ( chain.get("schema") == "https://verifyum.com/schema/witness-proof-chain-v1.json" and chain.get("protocol") == "verifyum" and chain.get("version") == 1 and embedded_bundle_matches and chain.get("proof_id") == proof_membership.get("proof_id") ) checks["hourly_checkpoint_valid"] = _checkpoint_valid(hourly, "hourly") checks["hourly_membership_valid"] = ( checks["hourly_checkpoint_valid"] and _membership_valid(proof_path, hourly) ) checks["daily_checkpoint_valid"] = _checkpoint_valid(daily, "daily") checks["daily_membership_valid"] = ( checks["daily_checkpoint_valid"] and _membership_valid(daily_path, daily) ) if checks["hourly_checkpoint_valid"] and checks["daily_checkpoint_valid"]: hourly_batch = _batch_id(hourly) daily_batch = _batch_id(daily) checks["networks_match"] = ( chain.get("network") == proof_membership.get("network") == hourly.get("network") == daily.get("network") ) checks["daily_leaf_matches_hourly"] = ( isinstance(daily_path, dict) and daily_path.get("subject_type") == "hourly-checkpoint-v1" and daily_path.get("subject_id") == hourly.get("checkpoint_hash") and daily_path.get("leaf_hash") == checkpoint_leaf_hash(hourly) ) checks["urls_canonical"] = ( chain.get("hourly_receipts_url") == f"{RECEIPTS_URL_PREFIX}hourly/{hourly_batch}.json" and chain.get("daily_checkpoint_url") == f"{CHECKPOINT_URL_PREFIX}daily/{daily_batch}.json" and chain.get("daily_receipts_url") == f"{RECEIPTS_URL_PREFIX}daily/{daily_batch}.json" ) checks["period_contains_hourly"] = ( daily.get("period_start") <= hourly.get("period_start") and hourly.get("period_end") <= daily.get("period_end") ) else: checks.update({ "networks_match": False, "daily_leaf_matches_hourly": False, "urls_canonical": False, "period_contains_hourly": False, }) return { "valid": all(value is True for value in checks.values()), "checks": checks, "daily_checkpoint_hash": daily.get("checkpoint_hash") if isinstance(daily, dict) else None, "hourly_receipts_url": chain.get("hourly_receipts_url"), "daily_receipts_url": chain.get("daily_receipts_url"), } def verify_witness(proof_id: str) -> dict: """Fetches and verifies the proof, hourly membership and optional daily bridge. A missing daily bridge means the daily checkpoint is not due yet; it does not invalidate the already witnessed hourly proof. """ if not isinstance(proof_id, str) or not PROOF_ID_RE.fullmatch(proof_id): raise VerifyumError("invalid proof id") metadata = _request("GET", f"https://{proof_id}.{PROOF_DOMAIN}/.well-known/verifyum.json")["body"] bundle = _request("GET", f"/v2/proofs/{proof_id}/witnesses")["body"] chain = None try: chain = _request("GET", f"/v2/proofs/{proof_id}/witness-chain")["body"] except VerifyumError as error: if error.status != 404: raise registry = _request("GET", f"https://{PROOF_DOMAIN}/.well-known/verifyum-service-keys.json")["body"] result = verify_witness_documents(metadata, bundle, registry, chain) if bundle.get("proof_id") != proof_id or result["checks"].get("subject_id") != proof_id: result["checks"]["subject_matches"] = False result["valid"] = False result["checks"].pop("subject_id", None) return result def verify_witness_documents(metadata, bundle, registry, chain=None) -> dict: """Offline core of `verify_witness`: same checks over already-fetched documents. `bundle` is the witness-proof-membership-v1 document. """ if not isinstance(metadata, dict): metadata = {} if not isinstance(bundle, dict): bundle = {} checkpoint = bundle.get("checkpoint") membership = bundle.get("membership") if not isinstance(checkpoint, dict): checkpoint = {} if not isinstance(membership, dict): membership = {} checks: dict[str, bool | None] = {} proof_id = metadata.get("proof_id") try: leaf = proof_leaf_hash(metadata) except (ValueError, TypeError) as error: print(f"verify_witness: metadata not canonical: {error}", file=sys.stderr) leaf = None checks["leaf_hash_matches"] = leaf is not None and membership.get("leaf_hash") == leaf checks["subject_matches"] = ( isinstance(proof_id, str) and PROOF_ID_RE.fullmatch(proof_id) is not None and membership.get("subject_id") == proof_id and bundle.get("proof_id") == proof_id and membership.get("subject_type") == "proof-v2" and checkpoint.get("subject_type") == "proof-v2" and checkpoint.get("kind") == "hourly" ) checks["path_reaches_root"] = leaf is not None and verify_path( leaf, membership.get("path"), checkpoint.get("merkle_root") ) try: recomputed = checkpoint_hash(checkpoint) except (ValueError, TypeError) as error: print(f"verify_witness: checkpoint not canonical: {error}", file=sys.stderr) recomputed = None checks["checkpoint_hash_matches"] = ( recomputed is not None and checkpoint.get("checkpoint_hash") == recomputed and membership.get("checkpoint_hash") == recomputed ) subject_count = checkpoint.get("subject_count") leaf_index = membership.get("leaf_index") checks["membership_consistent"] = ( membership.get("checkpoint_kind") == checkpoint.get("kind") and isinstance(subject_count, int) and not isinstance(subject_count, bool) and membership.get("leaf_count") == subject_count and isinstance(leaf_index, int) and not isinstance(leaf_index, bool) and 0 <= leaf_index < subject_count ) batch_id = _batch_id(checkpoint) checks["bundle_consistent"] = ( bundle.get("network") == checkpoint.get("network") and (metadata.get("anchor") or {}).get("network") == checkpoint.get("network") and batch_id is not None and bundle.get("checkpoint_url") == f"https://verifyum.com/witness/checkpoints/{checkpoint.get('kind')}/{batch_id}.json" ) checks["checkpoint_after_proof"] = _checkpoint_after_proof(checkpoint, metadata) checks["service_signature_valid"] = verify_service_signature(metadata, registry) daily_result = None if chain is not None: daily_result = verify_witness_chain_documents(bundle, chain) checks["daily_chain_valid"] = daily_result["valid"] decided = [value for value in checks.values() if value is not None] return { "valid": all(decided), "checks": {**checks, "subject_id": proof_id}, "leaf_hash": leaf, "checkpoint_hash": recomputed, "checkpoint_url": bundle.get("checkpoint_url"), "network": checkpoint.get("network"), "period_end": checkpoint.get("period_end"), # Presence, not a verdict: whether a daily chain was given at all. "daily_chain_present": daily_result is not None, # Never checked here. None means "not verified", which is the truth; # the receipts at daily_receipts_url are where it can be checked. "external_witness_verified": None, "daily_checkpoint_hash": (daily_result or {}).get("daily_checkpoint_hash"), "hourly_receipts_url": (daily_result or {}).get("hourly_receipts_url"), "daily_receipts_url": (daily_result or {}).get("daily_receipts_url"), "boundary": WITNESS_CHAIN_BOUNDARY, } # -------------------------------------------------------------------------- # Self-bearing receipts # # A receipt used to hold the nonce and a proof id, and everything else lived # at verifyum.com. That made the holder depend on our uptime for the life of # the claim, which is the opposite of what the product promises and unlike # the .ots file it was modelled on. A receipt now carries the signed # metadata, the witness bundle with its Merkle path, the key registry and, # once the daily checkpoint exists, the additive bridge into its external # receipts. No request to us is then needed to check either Merkle path. # # What still cannot be carried is the ledger. `verify --independent` reads # the transaction from a public Solana RPC and compares the memo itself. # -------------------------------------------------------------------------- RECEIPT_VERSION = 4 SOLANA_RPC = os.environ.get("VERIFYUM_SOLANA_RPC", "https://api.mainnet-beta.solana.com") MEMO_PROGRAM = "MemoSq4gqABAXKb96qnH8TysNcWxMyWCqXgDLGmfcHr" def build_memo(proof_id: str, commitment: str) -> str: """The exact memo the anchor carries, so a reader can rebuild it.""" return "verifyum:v2:id={};alg=sha256;commitment={}".format( proof_id, commitment[len("sha256:"):] ) def fetch_evidence(proof_id: str) -> dict: """Collects everything a receipt needs to be checkable without us. Returns the parts it could get. The witness bundle only exists once the proof has been folded into an hourly checkpoint, so a receipt written immediately after anchoring is thin until `upgrade` completes it. """ evidence = {} try: evidence["metadata"] = _request( "GET", f"https://{proof_id}.{PROOF_DOMAIN}/.well-known/verifyum.json" )["body"] except VerifyumError: pass try: evidence["witness_bundle"] = _request("GET", f"/v2/proofs/{proof_id}/witnesses")["body"] except VerifyumError: pass try: evidence["witness_chain"] = _request( "GET", f"/v2/proofs/{proof_id}/witness-chain" )["body"] except VerifyumError: pass try: evidence["service_keys"] = _request( "GET", f"https://{PROOF_DOMAIN}/.well-known/verifyum-service-keys.json" )["body"] except VerifyumError: pass return evidence def receipt_is_complete(receipt: dict) -> bool: """True when the receipt can be checked with no network at all.""" return all( isinstance(receipt.get(part), dict) for part in ("metadata", "witness_bundle", "service_keys") ) def receipt_has_daily_chain(receipt: dict) -> bool: """True once the receipt also carries the hourly-to-daily bridge.""" return isinstance(receipt.get("witness_chain"), dict) def receipt_version(receipt: dict) -> int: """Preserves v3 until the additive daily witness document is present.""" return RECEIPT_VERSION if receipt_has_daily_chain(receipt) else 3 def verify_offline(data: bytes, receipt: dict) -> dict: """Checks a file against a complete receipt without contacting anyone. Covers the local binding, the signed metadata, the Merkle path into the checkpoint and the service signature. It cannot cover the ledger: for that the transaction has to be read from somewhere, which is what `verify_independent` does. """ if not receipt_is_complete(receipt): raise VerifyumError( "receipt does not carry its evidence yet; run `verifyum upgrade` while online" ) draft = receipt["private_draft"] manifest = draft["manifest"] metadata = receipt["metadata"] checks = { "hash_matches": hashlib.sha256(data).hexdigest() == manifest["file"]["hash"]["value"], "size_matches": str(len(data)) == manifest["file"]["size"], "commitment_matches": "sha256:" + hashlib.sha256(b"verifyum:commitment:v2\n" + jcs(manifest)).hexdigest() == draft["commitment"], "metadata_commitment_matches": metadata.get("commitment") == draft["commitment"], "anchor_finalized": metadata.get("anchor", {}).get("status") == "finalized", } witness = verify_witness_documents( metadata, receipt["witness_bundle"], receipt["service_keys"], receipt.get("witness_chain"), ) checks.update(witness["checks"]) checks.pop("subject_id", None) # A memo that does not rebuild from the receipt's own values would mean # the receipt and the anchor describe different things. memo = metadata.get("anchor", {}).get("memo") checks["memo_matches"] = memo == build_memo( metadata.get("proof_id", ""), draft["commitment"] ) return { "valid": all(value is True for value in checks.values()), "checks": checks, "transaction_signature": metadata.get("anchor", {}).get("transaction_signature"), "network": metadata.get("anchor", {}).get("network"), "offline": True, "witness_scope": WITNESS_OFFLINE_SCOPE, } def verify_independent(receipt: dict, rpc_url: str | None = None) -> dict: """Reads the anchor from a public Solana RPC and checks the memo itself. Nothing here touches verifyum.com. The transaction is fetched by its signature and the memo instruction is compared against the memo rebuilt from the receipt, so the ledger, not the operator, is what answers. """ metadata = receipt.get("metadata") if not isinstance(metadata, dict): raise VerifyumError( "receipt does not carry its metadata; run `verifyum upgrade` while online" ) anchor = metadata.get("anchor") if not isinstance(anchor, dict) or not isinstance(anchor.get("transaction_signature"), str): raise VerifyumError("receipt carries no transaction signature") signature = anchor["transaction_signature"] payload = json.dumps({ "jsonrpc": "2.0", "id": 1, "method": "getTransaction", "params": [signature, {"encoding": "jsonParsed", "maxSupportedTransactionVersion": 0}], }).encode() request = urllib.request.Request( rpc_url or SOLANA_RPC, data=payload, headers={"Content-Type": "application/json", "User-Agent": USER_AGENT}, method="POST", ) try: with urllib.request.urlopen(request, timeout=30) as response: answer = json.loads(response.read().decode("utf-8")) except (urllib.error.URLError, OSError, ValueError) as error: raise VerifyumError(f"could not reach the Solana RPC: {error}") from None if answer.get("error"): raise VerifyumError("Solana RPC refused: " + str(answer["error"].get("message"))) result = answer.get("result") checks = {"transaction_found": isinstance(result, dict)} observed = None if isinstance(result, dict): checks["transaction_succeeded"] = (result.get("meta") or {}).get("err") is None instructions = ( ((result.get("transaction") or {}).get("message") or {}).get("instructions") or [] ) for instruction in instructions: if not isinstance(instruction, dict): continue if instruction.get("program") == "spl-memo" or instruction.get("programId") == MEMO_PROGRAM: parsed = instruction.get("parsed") if isinstance(parsed, str): observed = parsed break checks["memo_present"] = isinstance(observed, str) expected = build_memo(metadata.get("proof_id", ""), metadata.get("commitment", "")) checks["memo_matches_receipt"] = observed == expected checks["anchor_address_matches"] = True address = anchor.get("anchor_address") if isinstance(address, str): signers = [ account.get("pubkey") for account in ( ((result.get("transaction") or {}).get("message") or {}).get("accountKeys") or [] ) if isinstance(account, dict) and account.get("signer") ] checks["anchor_address_matches"] = address in signers return { "valid": all(value is True for value in checks.values()), "checks": checks, "transaction_signature": signature, "memo": observed, "block_time": (result or {}).get("blockTime") if isinstance(result, dict) else None, "source": rpc_url or SOLANA_RPC, } # -------------------------------------------------------------------------- # Command line # # The receipt sits beside the file as .verifyum.json, the way an .ots # file does, so `verify` needs nothing but the original path. It holds the # nonce, which is the only thing that ties the file to its public proof, and # which Verifyum never has. Losing it loses the link for good. # # The mode is set to 0600, which POSIX honours and Windows does not: there a # receipt lands 0666 and inherits the directory's ACL. The tool therefore # does not promise the file is private. It says what the file is and leaves # the placing of it to someone who knows where their own secrets belong. # -------------------------------------------------------------------------- RECEIPT_SUFFIX = ".verifyum.json" EXIT_OK = 0 EXIT_FAILED = 1 EXIT_USAGE = 2 EXIT_UNAVAILABLE = 3 def receipt_path(path: str) -> str: """Where the receipt for a file lives.""" return path + RECEIPT_SUFFIX def write_receipt(path: str, proof: dict, evidence: dict | None = None) -> str: """Stores the proof and its draft beside the file. The mode request is honoured on POSIX and ignored on Windows, so this does not make the file private on its own. """ target = receipt_path(path) document = { "schema": "https://verifyum.com/schema/receipt-v1.json", "protocol": "verifyum", "version": receipt_version(evidence or {}), "proof_id": proof.get("proof_id"), "commitment": proof.get("commitment"), "status": proof.get("status"), "proof_url": proof.get("proof_url"), "private_draft": proof.get("private_draft"), } # Everything needed to check this without asking us anything. for name, part in (evidence or {}).items(): document[name] = part body = json.dumps(document, indent=2, sort_keys=True) + "\n" temporary = target + ".tmp" with open(temporary, "w", encoding="utf-8") as handle: handle.write(body) try: os.chmod(temporary, 0o600) except OSError: pass os.replace(temporary, target) return target def read_receipt(path: str) -> dict: with open(receipt_path(path), "r", encoding="utf-8") as handle: return json.load(handle) def _emit(payload: dict, as_json: bool, lines: list) -> None: if as_json: print(json.dumps(payload, sort_keys=True)) return for line in lines: print(line) def _cmd_stamp(args) -> int: worst = EXIT_OK for path in args.files: if not os.path.isfile(path): print("not a file: " + path, file=sys.stderr) worst = max(worst, EXIT_USAGE) continue if os.path.exists(receipt_path(path)) and not args.force: print("receipt already exists, use --force to replace: " + receipt_path(path), file=sys.stderr) worst = max(worst, EXIT_USAGE) continue try: proof = anchor_file(path, wait=not args.no_wait, timeout=args.timeout) except VerifyumError as error: print("verifyum: " + str(error), file=sys.stderr) worst = max(worst, EXIT_UNAVAILABLE) continue except OSError as error: print("verifyum: " + str(error), file=sys.stderr) worst = max(worst, EXIT_USAGE) continue evidence = fetch_evidence(proof["proof_id"]) if proof.get("proof_id") else {} target = write_receipt(path, proof, evidence) complete = receipt_is_complete(read_receipt(path)) _emit( {"file": path, "receipt": target, "proof_id": proof.get("proof_id"), "status": proof.get("status"), "proof_url": proof.get("proof_url"), "self_bearing": complete}, args.json, [ "stamped " + path, " proof " + str(proof.get("proof_id")), " status " + str(proof.get("status")), " url " + str(proof.get("proof_url")), " receipt " + target, " this holds the nonce. We do not have it and cannot", " replace it. Keep it, and do not commit or sync it", " anywhere you would not put a key.", ] + ([ " evidence carried: this receipt checks offline", ] if complete else [ " evidence incomplete: the witness bundle appears when the next", " hourly checkpoint is published. Run `verifyum upgrade` then,", " and the receipt stops needing us at all.", ]), ) return worst def _cmd_verify(args) -> int: worst = EXIT_OK for path in args.files: try: receipt = read_receipt(path) with open(path, "rb") as handle: data = handle.read() except (OSError, ValueError) as error: print("verifyum: " + str(error), file=sys.stderr) worst = max(worst, EXIT_USAGE) continue try: if args.offline: result = verify_offline(data, receipt) else: result = verify(receipt["proof_id"], data, receipt["private_draft"]) if args.independent: chain = verify_independent(receipt) result["checks"].update( {"chain_" + name: value for name, value in chain["checks"].items()} ) result["valid"] = result["valid"] and chain["valid"] result["chain_source"] = chain["source"] except VerifyumError as error: print("verifyum: " + str(error), file=sys.stderr) worst = max(worst, EXIT_UNAVAILABLE) continue except (KeyError, TypeError) as error: print("verifyum: receipt is not usable: " + str(error), file=sys.stderr) worst = max(worst, EXIT_USAGE) continue if not result["valid"]: worst = max(worst, EXIT_FAILED) failed = [name for name, passed in result["checks"].items() if not passed] _emit( {"file": path, "valid": result["valid"], "checks": result["checks"], "transaction_signature": result.get("transaction_signature")}, args.json, [ ("verified " if result["valid"] else "FAILED ") + path, " proof " + str(receipt.get("proof_id")), " chain " + str(result.get("transaction_signature")), " source " + ( "the receipt alone" if args.offline and not args.independent else "the receipt and " + str(result.get("chain_source")) if args.offline else "verifyum.com" + (" and " + str(result.get("chain_source")) if args.independent else "") ), ] + ([ " witness consistent offline; whether the checkpoint was published is shown", " at the URLs the receipt names, not here (checkpoints are unsigned)", ] if args.offline else []) + ([" failed " + ", ".join(failed)] if failed else []), ) return worst def _cmd_witness(args) -> int: worst = EXIT_OK for target in args.targets: proof_id = target if os.path.exists(receipt_path(target)): try: proof_id = read_receipt(target)["proof_id"] except (OSError, ValueError, KeyError) as error: print("verifyum: " + str(error), file=sys.stderr) worst = max(worst, EXIT_USAGE) continue try: result = verify_witness(proof_id) except VerifyumError as error: print("verifyum: " + str(error), file=sys.stderr) worst = max(worst, EXIT_UNAVAILABLE) continue # A check that did not run is not a check that passed. unchecked = [name for name, passed in (result.get("checks") or {}).items() if passed is None] ok = bool(result.get("valid")) and not unchecked if not ok: worst = max(worst, EXIT_FAILED) _emit( {"proof_id": proof_id, **result}, args.json, [ ("chain ok " if ok else "FAILED ") + proof_id, " checkpoint " + str(result.get("checkpoint_hash") or result.get("checkpoint", {}).get("checkpoint_hash")), " external witnesses not checked here: " + ( result.get("daily_receipts_url") or result.get("hourly_receipts_url") or str(result.get("checkpoint_url") or "").replace("/witness/checkpoints/", "/witness/receipts/") or "the channel receipts" ), ] + ([" not run " + ", ".join(unchecked)] if unchecked else []), ) return worst def _cmd_upgrade(args) -> int: """Fills in the evidence a receipt was written too early to carry.""" worst = EXIT_OK for path in args.files: try: receipt = read_receipt(path) except (OSError, ValueError) as error: print("verifyum: " + str(error), file=sys.stderr) worst = max(worst, EXIT_USAGE) continue if receipt_is_complete(receipt) and receipt_has_daily_chain(receipt) and not args.force: _emit({"file": path, "upgraded": False, "self_bearing": True, "daily_chain": True}, args.json, ["already complete " + path]) continue try: evidence = fetch_evidence(receipt["proof_id"]) except (VerifyumError, KeyError) as error: print("verifyum: " + str(error), file=sys.stderr) worst = max(worst, EXIT_UNAVAILABLE) continue merged = dict(receipt) merged.update(evidence) merged["version"] = receipt_version(merged) target = receipt_path(path) temporary = target + ".tmp" with open(temporary, "w", encoding="utf-8") as handle: handle.write(json.dumps(merged, indent=2, sort_keys=True) + "\n") try: os.chmod(temporary, 0o600) except OSError: pass os.replace(temporary, target) complete = receipt_is_complete(merged) if not complete: worst = max(worst, EXIT_UNAVAILABLE) _emit({"file": path, "upgraded": True, "self_bearing": complete, "daily_chain": receipt_has_daily_chain(merged)}, args.json, [("upgraded " if complete else "still incomplete ") + path] + ([" this receipt now links offline through the daily checkpoint"] if complete and receipt_has_daily_chain(merged) else [" hourly evidence complete; daily checkpoint is not published yet"] if complete else [" the witness bundle is not published yet; try again after the hour"])) return worst def main(argv=None) -> int: parser = argparse.ArgumentParser( prog="verifyum", description="Prove that a file existed, unchanged, at a point in time.", epilog="The receipt beside each file holds the nonce. Verifyum never has it, " "so a lost receipt cannot be recovered from us or from anyone. Treat it " "like a key: it is not made private by being written.", ) parser.add_argument("--version", action="version", version="verifyum " + USER_AGENT.split("/")[-1]) sub = parser.add_subparsers(dest="command", required=True) stamp = sub.add_parser("stamp", help="anchor one or more files and write their receipts") stamp.add_argument("files", nargs="+") stamp.add_argument("--force", action="store_true", help="replace an existing receipt") stamp.add_argument("--no-wait", action="store_true", help="return as soon as the anchor is queued") stamp.add_argument("--timeout", type=int, default=180, help="seconds to wait for finality") stamp.add_argument("--json", action="store_true", help="machine readable output") stamp.set_defaults(handler=_cmd_stamp) check = sub.add_parser("verify", help="check files against their receipts and the public record") check.add_argument("files", nargs="+") check.add_argument("--offline", action="store_true", help="check using the receipt alone, contacting no one") check.add_argument("--independent", action="store_true", help="also read the anchor from a public Solana RPC and compare the memo") check.add_argument("--json", action="store_true", help="machine readable output") check.set_defaults(handler=_cmd_verify) upgrade = sub.add_parser("upgrade", help="fetch the evidence a receipt needs to stand alone") upgrade.add_argument("files", nargs="+") upgrade.add_argument("--force", action="store_true", help="refetch even if already complete") upgrade.add_argument("--json", action="store_true", help="machine readable output") upgrade.set_defaults(handler=_cmd_upgrade) witness = sub.add_parser("witness", help="check a proof against the witness layer") witness.add_argument("targets", nargs="+", metavar="FILE_OR_PROOF_ID") witness.add_argument("--json", action="store_true", help="machine readable output") witness.set_defaults(handler=_cmd_witness) args = parser.parse_args(argv) return args.handler(args) if __name__ == "__main__": sys.exit(main())