diff --git a/CHANGELOG.md b/CHANGELOG.md index 7baa894d..ab42317b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,8 +6,53 @@ follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +### Added + +- **`pyrxd glyph inspect --verify-wave` names the signer.** For a VERIFIED v2 + signature, resolves the WAVE names the signing key owns: hash160 -> address -> + `wave.reverse_lookup`. A recipient can then confirm both that a file matches the + recorded digest AND that it was recorded by the holder of `company.rxd`. + `pyrxd.glyph.wave.wave_names_for_hash160` is the reusable half and is not + HashMark-specific. + + **Gated on verification.** An unverified or absent signature resolves nothing and + says why — presenting a name for an unproven signer would dress a claim up as an + identity, which is the failure the signature check exists to prevent. + +- **HashMark v2 signatures are VERIFIED, not just reported.** `verify_attestation` + rebuilds the canonical signed statement (fixed key order, label omitted when absent, + raw UTF-8 rather than `\uXXXX` escapes), recovers the public key, and requires + `hash160(recovered) == signer`. The commitment is what makes recovery non-circular: + without a value fixed in advance, an attacker writes whatever hash their chosen + signature recovers to. Low-S is mandatory and the header is range-checked (27..34). + + Attestation is kept SEPARATE from decoding, as the spec requires — an invalid + signature means the bytes were fine and the claim does not hold, which is a different + problem from a malformed record. v1 reports `not_attested` rather than a failure, + because v1 never claimed to say who. + + The chain's genesis hash is part of the signed statement and is NOT carried by the + record, so the same bytes do not verify on another chain. `inspect` assumes Radiant + mainnet for a pasted script and says so. + +- **`pyrxd inspect` decodes the Photonic `msg` data carrier.** `OP_RETURN PUSH3 "msg" + ` — the only OP_RETURN format with real volume on Radiant: measured + across 20 consecutive mainnet blocks, **73 of 73** data outputs carried this marker and + nothing else did. pyrxd already WROTE these and could not read one back, so the + commonest data output on the chain rendered as opaque hex. Non-UTF-8 bytes are reported + rather than refused — they are already on chain — and display sanitisation happens at + the render boundary so the raw bytes stay recoverable for a caller verifying them. + ### Fixed +- **Containers classified as `nft`/`mut` — the container branch was dead code (#578).** + `GlyphProtocol.CONTAINER` (7) is the spec'd marker and no mainnet token uses it: all + four containers on Radiant mainnet declare `type: "container"` on an ordinary NFT/MUT + protocol set. Verified on chain — the "BTC" container (reveal `57c4d660…dfb1`) decodes + to `p = (2,)` with `type = 'container'`. Three sites recognised only the protocol form: + `GlyphMetadata.is_container`, the inspect classifier, and its deliberate mirror in + `wave.py`. Both declarations now count at all three. + - **A multi-glyph reveal reported one glyph's metadata as the transaction's (#577).** `find_reveal_metadata` returns the FIRST input carrying a decodable `gly` payload, and that single payload was surfaced as the whole transaction's metadata. Multi-glyph diff --git a/src/pyrxd/cli/glyph_inspect.py b/src/pyrxd/cli/glyph_inspect.py index 70f00d21..d0e2e82e 100644 --- a/src/pyrxd/cli/glyph_inspect.py +++ b/src/pyrxd/cli/glyph_inspect.py @@ -365,6 +365,56 @@ def _render_script_human(payload: dict) -> str: type_ = payload.get("type", "?") head = f"type: {type_} length: {payload['length']} bytes" body: list[str] = [] + msg = payload.get("message") + if msg: + if msg["outcome"] == "ok": + if msg["is_utf8"]: + body.append(f" message ({msg['byte_length']} bytes): {_truncate_for_human(msg['text'])}") + else: + # Say WHY there is no text rather than printing nothing, or a caller + # assumes the field is empty when the bytes simply are not text. + body.append(f" message: {msg['byte_length']} bytes, not valid UTF-8 (see data_hex)") + else: + body.append(f" message: {msg['outcome']}" + (f" — {msg['detail']}" if msg.get("detail") else "")) + + hm = payload.get("hashmark") + if hm: + # HashMark is a third-party OP_RETURN format (MIT, github.com/cdonnachie/hashmark.rxd). + # Classifying it in JSON and not printing it here would leave the feature + # invisible to the person actually reading a terminal. + if hm["outcome"] == "ok": + body.append(f" HashMark v{hm['version']} ({hm['algorithm']})") + body.append(f" digest: {hm['digest']}") + if hm.get("label"): + body.append(f" label: {_truncate_for_human(hm['label'])}") + if hm.get("signer_hash160"): + body.append(f" signer: {hm['signer_hash160']}") + att = hm.get("attestation") or {} + if att.get("outcome") == "valid": + body.append(" signature VERIFIED — recovers to the committed signer") + body.append(f" (assuming {att.get('assumed_network')}; the chain is part of") + body.append(" the signed statement and a pasted script carries no context)") + wi = hm.get("wave_identity") + if wi and wi.get("resolved"): + names = wi.get("names") or [] + if names: + body.append(f" WAVE identity: {', '.join(names)}") + body.append(" (the signing key owns these names — file matches the") + body.append(" digest AND was recorded by that name's holder)") + else: + body.append(" WAVE identity: none — the signing key owns no WAVE name") + elif wi: + body.append(f" WAVE identity: not resolved ({wi.get('reason')})") + elif att.get("outcome") == "invalid_signature": + # The bytes decoded; the CLAIM does not hold. Saying "malformed" + # here would send whoever is debugging it after the wrong problem. + body.append(f" signature DOES NOT VERIFY — {att.get('detail', 'no detail')}") + body.append(" (the record is well-formed; its claim is not supported)") + body.append(" (proves someone knew this digest no later than the confirming") + body.append(" block — not authorship, ownership, originality or contents)") + else: + body.append(f" HashMark: {hm['outcome']}" + (f" — {hm['detail']}" if hm.get("detail") else "")) + if type_ == "p2pkh": body.append(f" owner_pkh: {payload['owner_pkh']}") elif type_ in ("nft", "ft"): @@ -561,8 +611,18 @@ def _render_ref_summary_body(payload: dict) -> list[str]: default=False, help="For an outpoint, fetch its source tx and classify the named vout.", ) +@click.option( + "--verify-wave", + "verify_wave", + is_flag=True, + default=False, + help=( + "For a VERIFIED HashMark v2 signature, look up the WAVE names the signing " + "key owns. Needs the network. Never runs on an unverified signature." + ), +) @click.pass_obj -def inspect_cmd(ctx: CliContext, inspect_input: str, fetch: bool, resolve: bool) -> None: +def inspect_cmd(ctx: CliContext, inspect_input: str, fetch: bool, resolve: bool, verify_wave: bool) -> None: """Classify a Glyph input. INPUT can be: @@ -689,6 +749,9 @@ def inspect_cmd(ctx: CliContext, inspect_input: str, fetch: bool, resolve: bool) else: # pragma: no cover — _classify_input never returns other values raise UserError(f"internal: unknown form {form!r}") + if verify_wave: + _attach_wave_identity(ctx, payload) + mode = ctx.output_mode if mode == "json": click.echo(emit(payload, mode="json")) @@ -734,3 +797,44 @@ async def _do() -> dict: cause=str(exc), fix=f"check that {ctx.electrumx_url} is reachable", ) from exc + + +def _attach_wave_identity(ctx: CliContext, payload: dict) -> None: + """Resolve the WAVE names a VERIFIED HashMark signer owns, and attach them. + + ONLY runs on a signature that actually verified. Resolving an unverified + signer would dress a claim up as an identity — the exact failure the + signature check exists to prevent — so an unverified or absent attestation + attaches nothing and says why. + + Errors are attached rather than raised: a name lookup failing is not a reason + to lose the classification the user asked for. + """ + hm = payload.get("hashmark") + if not hm: + return + att = hm.get("attestation") or {} + if att.get("outcome") != "valid": + hm["wave_identity"] = { + "resolved": False, + "reason": ( + "signature did not verify; refusing to resolve an unproven signer" + if att.get("outcome") == "invalid_signature" + else "no verified v2 signature on this record" + ), + } + return + + from ..glyph.wave import wave_names_for_hash160 + + async def _do() -> list[str]: + client = ctx.make_client() + async with client: + return await wave_names_for_hash160(client, bytes.fromhex(att["recovered_hash160"])) + + try: + names = asyncio.run(_do()) + except Exception as exc: + hm["wave_identity"] = {"resolved": False, "reason": f"lookup failed: {exc}"} + return + hm["wave_identity"] = {"resolved": True, "names": names} diff --git a/src/pyrxd/glyph/_inspect_core.py b/src/pyrxd/glyph/_inspect_core.py index 9dba31b1..61cbc7f0 100644 --- a/src/pyrxd/glyph/_inspect_core.py +++ b/src/pyrxd/glyph/_inspect_core.py @@ -36,7 +36,13 @@ import unicodedata from ..hash import hash256 -from ..script.hashmark import HashMarkOutcome, decode_hashmark +from ..script.hashmark import ( + RADIANT_MAINNET_GENESIS, + HashMarkOutcome, + decode_hashmark, + verify_attestation, +) +from ..script.message import MessageOutcome, decode_message from ..security.errors import ValidationError from ..security.types import Txid from ..transaction.transaction import Transaction @@ -369,6 +375,27 @@ def _inspect_script(script_hex: str) -> dict: # additive: anything that is not one stays plain `op_return`, because a # scanner meets thousands of other protocols' data outputs and treating # them as errors buries the real ones. + # The Photonic `msg` convention: OP_RETURN PUSH3 "msg" . + # Measured on 20 consecutive mainnet blocks, 73 of 73 OP_RETURN outputs + # carried this marker and nothing else did — it is the whole observed + # population, and pyrxd already WRITES it. Reading it back turns the + # commonest data output on the chain from an opaque blob into its text. + msg = decode_message(script) + if msg.outcome is not MessageOutcome.NOT_MESSAGE: + out["message"] = { + "outcome": msg.outcome.value, + # SANITISED here, at the display boundary. The message is arbitrary + # operator bytes and `repr` does not escape U+202E and friends; the + # decoder deliberately returns it unmangled so the raw bytes stay + # recoverable, and mangling belongs where it is shown. + "text": _sanitize_display_string(msg.text) if msg.text else None, + "is_utf8": msg.is_utf8, + "byte_length": len(msg.raw) if msg.raw else 0, + "detail": msg.detail, + } + if msg.ok: + out["type"] = "op_return-msg" + mark = decode_hashmark(script) if mark.outcome is not HashMarkOutcome.NOT_HASHMARK: out["hashmark"] = { @@ -385,6 +412,18 @@ def _inspect_script(script_hex: str) -> dict: } if mark.ok: out["type"] = f"op_return-hashmark-v{mark.version}" + # ATTEST, now that we can. The signed statement includes the + # chain's genesis hash — the verified context the tx was found + # in — and a pasted script carries no such context, so mainnet is + # ASSUMED and the assumption is reported rather than hidden. The + # same bytes on another chain are a different statement. + att = verify_attestation(mark, network_genesis=RADIANT_MAINNET_GENESIS) + out["hashmark"]["attestation"] = { + "outcome": att.outcome.value, + "recovered_hash160": att.recovered_hash160_hex, + "assumed_network": "radiant-mainnet", + "detail": att.detail, + } return out if is_nft_script(script_hex): @@ -640,6 +679,24 @@ def _classify_metadata_protocol(metadata) -> str: return "wave" if GlyphProtocol.CONTAINER in p: return "container" + # ...OR the `type` STRING, which is what the chain actually carries (#578). + # + # GlyphProtocol.CONTAINER (7) is the spec'd form and no mainnet token uses it. + # All four containers on Radiant mainnet declare themselves with `type: + # "container"` on an ordinary NFT/MUT protocol set, so the branch above was + # dead code and every container classified as "nft" or "mut". + # + # Verified against the chain, not inferred: the "BTC" container + # (ref 5558395540...c2ab:0, reveal 57c4d660...dfb1) decodes to `p = (2,)` with + # `type = 'container'`. The indexer agrees — it reports token_type CONTAINER + # for exactly these four and exposes no protocol field, so its label is derived + # from the same string. + # + # This is a DECLARATION, like the protocol array itself: `type` is operator CBOR + # and nothing on chain enforces it. Both forms are claims about what a token is; + # neither is a proof, and the ecosystem treats this one as the classification. + if (metadata.token_type or "").strip().lower() == "container": + return "container" if GlyphProtocol.AUTHORITY in p: return "authority" if GlyphProtocol.TIMELOCK in p: diff --git a/src/pyrxd/glyph/types.py b/src/pyrxd/glyph/types.py index 530825d6..6a9d548b 100644 --- a/src/pyrxd/glyph/types.py +++ b/src/pyrxd/glyph/types.py @@ -358,8 +358,20 @@ class GlyphMetadata: @property def is_container(self) -> bool: - """True when this envelope marks the token itself as a CONTAINER.""" - return GlyphProtocol.CONTAINER in self.protocol + """True when this envelope marks the token itself as a CONTAINER. + + Either declaration counts. `GlyphProtocol.CONTAINER` (7) is the spec'd + form and NO mainnet token uses it — all four containers on Radiant + mainnet declare `type: "container"` on an ordinary NFT/MUT protocol set, + so a protocol-only test was False for every real container (#578). + + Verified on chain: the "BTC" container (reveal 57c4d660...dfb1) decodes + to `p = (2,)` with `type = 'container'`. + + Both are DECLARATIONS — `type` is operator CBOR and nothing on chain + enforces it, exactly as nothing enforces the protocol array. + """ + return GlyphProtocol.CONTAINER in self.protocol or (self.token_type or "").strip().lower() == "container" def __post_init__(self) -> None: import re diff --git a/src/pyrxd/glyph/wave.py b/src/pyrxd/glyph/wave.py index 43176f30..cb9ab000 100644 --- a/src/pyrxd/glyph/wave.py +++ b/src/pyrxd/glyph/wave.py @@ -31,6 +31,9 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Final +if TYPE_CHECKING: + from ..constants import Network + from ..security.errors import ValidationError from .types import GlyphMetadata, GlyphProtocol @@ -303,6 +306,31 @@ def from_indexer_response(cls, data: dict[str, Any]) -> WaveRecord: raise WaveResolverError(f"could not parse indexer response: {exc}") from exc +async def wave_names_for_hash160(client, pubkey_hash160: bytes, *, network: Network | None = None) -> list[str]: + """WAVE names owned by the address a public-key hash encodes. + + The bridge between a KEY and a NAME: a hash160 is what a P2PKH address + encodes, and WAVE resolves names to addresses, so the reverse lookup answers + "which names does the holder of this key own". + + Written for HashMark v2 attestation — a verified signer is a hash160, and the + question a recipient actually has is "was this recorded by company.rxd?" — but + nothing here is HashMark-specific. + + CALLERS MUST ONLY PASS A KEY THEY HAVE VERIFIED. Resolving an unproven signer + would dress a claim up as an identity, which is precisely the failure the + signature check exists to prevent. See + :func:`pyrxd.script.hashmark.verify_attestation`. + """ + from ..base58 import base58check_encode + from ..constants import NETWORK_ADDRESS_PREFIX_DICT, Network + + if len(pubkey_hash160) != 20: + raise ValidationError(f"pubkey_hash160 must be 20 bytes, got {len(pubkey_hash160)}") + address = base58check_encode(NETWORK_ADDRESS_PREFIX_DICT[network or Network.MAINNET] + pubkey_hash160) + return await WaveResolver(client).reverse_lookup(address) + + def classify_glyph_metadata(metadata: GlyphMetadata) -> str: """Return the highest-specificity protocol classification for a metadata payload. @@ -333,6 +361,11 @@ def classify_glyph_metadata(metadata: GlyphMetadata) -> str: return "wave" if GlyphProtocol.CONTAINER in p: return "container" + # ...or the `type` STRING, which is the form the chain actually carries (#578). + # Kept in step with the deliberate mirror of this function in + # `_inspect_core._classify_metadata_protocol`; see the note there. + if (metadata.token_type or "").strip().lower() == "container": + return "container" if GlyphProtocol.AUTHORITY in p: return "authority" if GlyphProtocol.TIMELOCK in p: diff --git a/src/pyrxd/script/hashmark.py b/src/pyrxd/script/hashmark.py index 4e266bd9..10b0c7c6 100644 --- a/src/pyrxd/script/hashmark.py +++ b/src/pyrxd/script/hashmark.py @@ -33,13 +33,19 @@ from enum import Enum from ..constants import OpCode -from .script import Script +from ..security.errors import ValidationError +from .script import data_pushes_after_op_return __all__ = [ "HASHMARK_MAGIC", + "RADIANT_MAINNET_GENESIS", + "AttestationOutcome", + "AttestationResult", "HashMarkOutcome", "HashMarkRecord", + "canonical_statement", "decode_hashmark", + "verify_attestation", ] #: The exact ASCII bytes every record opens with. Compared as BYTES, never as a @@ -100,30 +106,6 @@ def ok(self) -> bool: return self.outcome is HashMarkOutcome.OK -def _pushes(script: bytes) -> list[bytes] | None: - """Every push after the leading ``OP_RETURN``, or None if the script is not - cleanly push-only from offset 1. - - Returns None rather than raising: at this point no magic has been seen, so - the output has not yet claimed to be a HashMark and a parse failure means - "some other protocol", not "a broken HashMark". - """ - # allow_malformed: a truncated push must set truncated_at rather than raise. - # The spec is explicit that a bad push BEFORE the magic is NOT_HASHMARK — no - # HashMark claim has been made yet — and the inspector's contract allows only - # ValidationError to escape. Raising here did both wrong; the fuzzer caught it. - parsed = Script(script[1:], allow_malformed=True) - if parsed.truncated_at is not None: - return None - out: list[bytes] = [] - for chunk in parsed.chunks: - op = chunk.op[0] if isinstance(chunk.op, bytes) else chunk.op - if op > 0x4B: # not a direct/PUSHDATA data push - return None - out.append(chunk.data or b"") - return out - - def decode_hashmark(script: bytes) -> HashMarkRecord: """Decode a ``scriptPubKey`` as a HashMark record. @@ -133,7 +115,7 @@ def decode_hashmark(script: bytes) -> HashMarkRecord: if not script or script[0] != _OP_RETURN: return HashMarkRecord(HashMarkOutcome.NOT_HASHMARK) - pushes = _pushes(script) + pushes = data_pushes_after_op_return(script) if pushes is None or not pushes or pushes[0] != HASHMARK_MAGIC: return HashMarkRecord(HashMarkOutcome.NOT_HASHMARK) @@ -210,3 +192,135 @@ def decode_hashmark(script: bytes) -> HashMarkRecord: signer_hash160_hex=signer_hex, signature_hex=signature_hex, ) + + +# --------------------------------------------------------------------------- +# Attestation — SEPARATE from decoding, deliberately. +# +# The spec keeps these apart and says why: "a record that decodes is well-formed, +# not yet believed". An invalid signature is not a malformed record — the bytes +# were fine and the CLAIM does not hold — and reporting it as malformed sends +# whoever is debugging it after the wrong problem. +# --------------------------------------------------------------------------- + +#: Radiant mainnet genesis, RPC/display byte order. Part of the SIGNED statement, +#: and NOT carried by the record — it is the verified context the transaction was +#: found in. The same record bytes on another chain make a different statement and +#: will not verify there, which is intended. +RADIANT_MAINNET_GENESIS = "0000000065d8ed5d8be28d6876b3ffb660ac2a6c0ca59e437e1f7a6f4e003fb4" + +#: secp256k1 group order, for the range checks in §5.6. +_SECP256K1_N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141 + + +class AttestationOutcome(Enum): + """Whether a decoded v2 record's signature actually holds.""" + + VALID = "valid" + INVALID_SIGNATURE = "invalid_signature" + #: v1 carries no signer, so there is nothing to attest. Not a failure — v1 + #: never claimed to say WHO, only WHEN. + NOT_ATTESTED = "not_attested" + + +@dataclass(frozen=True) +class AttestationResult: + outcome: AttestationOutcome + #: hash160 of the key recovered from the signature, when recovery succeeded. + recovered_hash160_hex: str | None = None + detail: str | None = None + + @property + def valid(self) -> bool: + return self.outcome is AttestationOutcome.VALID + + +def _json_string(value: str) -> str: + r"""Escape per §5.6: a quote becomes \\" and a backslash \\\\; everything else is + emitted as raw UTF-8, never as a \\uXXXX escape. + + Hand-rolled rather than ``json.dumps`` on purpose — the stdlib escapes + non-ASCII to ``\\uXXXX`` by default, which would change the signed bytes for + any label containing an accent or an emoji. + """ + return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"' + + +def canonical_statement(record: HashMarkRecord, *, network_genesis: str = RADIANT_MAINNET_GENESIS) -> str: + """The exact single-line JSON a v2 signature covers (§5.6). + + Fixed key order, no insignificant whitespace, and ``label`` OMITTED ENTIRELY + when absent rather than included as an empty string — a different statement, + and therefore a different signature. + """ + if record.version != 2: + raise ValidationError("only a v2 record carries a signed statement") + parts = [ + f'"v":{_json_string("HashMark/v2")}', + f'"network":{_json_string(network_genesis)}', + f'"signerHash160":{_json_string(record.signer_hash160_hex or "")}', + # The header byte as two lowercase hex digits, never a name: names acquire + # aliases (sha256 / SHA-256 / sha-256) and a signature must not depend on + # which spelling was in fashion. + f'"algorithmId":{_json_string(f"{record.algorithm_id:02x}")}', + f'"digest":{_json_string(record.digest_hex or "")}', + ] + if record.label is not None: + parts.append(f'"label":{_json_string(record.label)}') + return "{" + ",".join(parts) + "}" + + +def verify_attestation(record: HashMarkRecord, *, network_genesis: str = RADIANT_MAINNET_GENESIS) -> AttestationResult: + """Recover the signer from a v2 signature and require it to match the commitment. + + The signer hash160 is committed TWICE — in the record and inside the signed + statement — and both are required. Without a value fixed in advance to compare + against, recovery is circular and proves nothing: an attacker would simply + write whatever hash their chosen signature recovers to. + + Needs the chain's genesis hash, which is why this is not part of decoding: a + dependency-free decoder does not have it, and the same bytes on another chain + are a different statement. + """ + from ..hash import hash160, hash256 + from ..keys import recover_public_key + from ..utils import text_digest + + if not record.ok: + return AttestationResult(AttestationOutcome.INVALID_SIGNATURE, detail="record did not decode") + if record.version != 2 or not record.signature_hex or not record.signer_hash160_hex: + return AttestationResult(AttestationOutcome.NOT_ATTESTED, detail="v1 record carries no signer") + + sig = bytes.fromhex(record.signature_hex) + header, r_bytes, s_bytes = sig[0], sig[1:33], sig[33:65] + + # §5.6: header is 27 + recoveryId, +4 when the key is compressed; 27..34. + if not 27 <= header <= 34: + return AttestationResult(AttestationOutcome.INVALID_SIGNATURE, detail=f"header {header} outside 27..34") + rec_id = (header - 27) & 3 + + r, s_val = int.from_bytes(r_bytes, "big"), int.from_bytes(s_bytes, "big") + if not 1 <= r < _SECP256K1_N: + return AttestationResult(AttestationOutcome.INVALID_SIGNATURE, detail="r out of range") + # LOW-S IS MANDATORY. It removes the s versus n-s malleability so a verifier + # has one accepted form. It does NOT make signatures unique — a different + # nonce yields different bytes for the same key and message — so an + # attestation is identified by its statement and recovered signer, never by + # these bytes. + if not 1 <= s_val <= _SECP256K1_N // 2: + return AttestationResult(AttestationOutcome.INVALID_SIGNATURE, detail="s is not low-S") + + statement = canonical_statement(record, network_genesis=network_genesis) + try: + pub = recover_public_key(r_bytes + s_bytes + bytes([rec_id]), text_digest(statement), hasher=hash256) + recovered = hash160(pub.serialize(compressed=header >= 31)).hex() + except Exception as exc: + return AttestationResult(AttestationOutcome.INVALID_SIGNATURE, detail=f"recovery failed: {exc}") + + if recovered != record.signer_hash160_hex: + return AttestationResult( + AttestationOutcome.INVALID_SIGNATURE, + recovered_hash160_hex=recovered, + detail="recovered key does not match the committed signer", + ) + return AttestationResult(AttestationOutcome.VALID, recovered_hash160_hex=recovered) diff --git a/src/pyrxd/script/message.py b/src/pyrxd/script/message.py new file mode 100644 index 00000000..249d3a15 --- /dev/null +++ b/src/pyrxd/script/message.py @@ -0,0 +1,81 @@ +"""Decode the Photonic ``msg`` convention — the only OP_RETURN format with real +volume on Radiant mainnet. + +``OP_RETURN PUSH3 "msg" ``. The 3-byte ``msg`` marker is what +wallet and explorer parsers key on. pyrxd already WRITES this (see +:data:`pyrxd.constants.MAX_OP_RETURN_MSG_BYTES` and the dMint miner's +``op_return_msg``); until now it could not read one back, so every such output +rendered as an opaque hex blob. + +Measured on 20 consecutive mainnet blocks: **73 of 73** ``OP_RETURN`` outputs +carried this marker, and nothing else did. + +The message is arbitrary operator bytes. This module returns it as data and does +NOT sanitise — display sanitisation belongs at the render boundary, which already +has ``_sanitize_display_string`` for bidi overrides and friends. Returning +pre-mangled text would make the raw bytes unrecoverable for a caller that wants +to verify them. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + +from .script import data_pushes_after_op_return + +__all__ = ["MSG_MARKER", "MessageOutcome", "MessageRecord", "decode_message"] + +#: The marker push, compared as bytes. +MSG_MARKER = b"msg" + + +class MessageOutcome(Enum): + OK = "ok" + #: Not a `msg` output. Skip silently — most data carriers belong to something else. + NOT_MESSAGE = "not_message" + #: Claims to be one (carries the marker) and is malformed. + INVALID = "invalid" + + +@dataclass(frozen=True) +class MessageRecord: + outcome: MessageOutcome + #: The message decoded as UTF-8, or None when the bytes are not valid UTF-8. + #: NOT sanitised for display; the renderer does that. + text: str | None = None + #: The raw message bytes, always present on OK — a caller verifying a + #: commitment needs the bytes, not a lossy decode of them. + raw: bytes | None = None + is_utf8: bool = False + detail: str | None = None + + @property + def ok(self) -> bool: + return self.outcome is MessageOutcome.OK + + +def decode_message(script: bytes) -> MessageRecord: + """Decode a ``scriptPubKey`` as a Photonic ``msg`` data carrier.""" + pushes = data_pushes_after_op_return(script) + if pushes is None or not pushes or pushes[0] != MSG_MARKER: + return MessageRecord(MessageOutcome.NOT_MESSAGE) + + # Past the marker the output claims to be a message, so a bad shape is a + # genuine defect rather than another protocol's data. + if len(pushes) != 2: + return MessageRecord( + MessageOutcome.INVALID, + detail=f"expected exactly 2 pushes (marker + message), found {len(pushes)}", + ) + raw = pushes[1] + if not raw: + return MessageRecord(MessageOutcome.INVALID, detail="message push is empty") + + # NON-UTF-8 IS NOT AN ERROR. The bytes are already on chain and nothing + # constrains them to text; refusing would lose a record we can otherwise + # describe exactly. Report the bytes and say the decode failed. + try: + return MessageRecord(MessageOutcome.OK, text=raw.decode("utf-8"), raw=raw, is_utf8=True) + except UnicodeDecodeError: + return MessageRecord(MessageOutcome.OK, text=None, raw=raw, is_utf8=False) diff --git a/src/pyrxd/script/script.py b/src/pyrxd/script/script.py index 3c7f8562..42f000e8 100644 --- a/src/pyrxd/script/script.py +++ b/src/pyrxd/script/script.py @@ -305,3 +305,27 @@ def find_and_delete(cls, source: Script, pattern: Script) -> Script: @classmethod def write_bin(cls, octets: bytes) -> Script: return Script(encode_pushdata(octets)) + + +def data_pushes_after_op_return(script: bytes) -> list[bytes] | None: + """Every data push following a leading ``OP_RETURN``, or None if not push-only. + + Shared by the ``OP_RETURN`` payload decoders (HashMark, the Photonic ``msg`` + convention) so the walk exists once. Returns None rather than raising: at the + point a caller uses this, no protocol marker has been seen yet, so a parse + failure means "some other protocol", not "a broken record". Radiant Core + classifies a data carrier as ``TX_NULL_DATA`` only when the remainder after + ``OP_RETURN`` is push-only, so a non-push chunk means this is not one. + """ + if not script or script[0] != 0x6A: + return None + parsed = Script(script[1:], allow_malformed=True) + if parsed.truncated_at is not None: + return None + out: list[bytes] = [] + for chunk in parsed.chunks: + op = chunk.op[0] if isinstance(chunk.op, bytes) else chunk.op + if op > 0x4B: # not a direct or OP_PUSHDATA data push + return None + out.append(chunk.data or b"") + return out diff --git a/tests/test_container_declared_by_type_string.py b/tests/test_container_declared_by_type_string.py new file mode 100644 index 00000000..d2efb829 --- /dev/null +++ b/tests/test_container_declared_by_type_string.py @@ -0,0 +1,71 @@ +"""A container declares itself with `type: "container"`, not protocol 7. + +`GlyphProtocol.CONTAINER` (7) is the spec'd marker and **no mainnet token uses +it**. All four containers on Radiant mainnet carry `type: "container"` on an +ordinary NFT/MUT protocol set, so every protocol-only test was False for every +real container and they classified as "nft" or "mut" (#578). + +Verified against the chain rather than inferred: the "BTC" container +(ref ``5558395540…c2ab:0``, reveal ``57c4d660…dfb1``) decodes to ``p = (2,)`` +with ``type = 'container'``. The indexer agrees — it labels exactly these four +CONTAINER and exposes no protocol field, so its label comes from the same string. + +Both forms are DECLARATIONS. `type` is operator-supplied CBOR and nothing on +chain enforces it, exactly as nothing enforces the protocol array; the ecosystem +treats this one as the classification. + +Three sites had to change together — the property, the inspect classifier, and +the deliberate mirror of that classifier in `wave.py`. This suite pins all three, +because a copy left behind is how the branch stayed dead in the first place. +""" + +from __future__ import annotations + +import pytest + +from pyrxd.glyph._inspect_core import _classify_metadata_protocol +from pyrxd.glyph.types import GlyphMetadata, GlyphProtocol +from pyrxd.glyph.wave import classify_glyph_metadata + +#: The shape every mainnet container actually has. +_AS_ON_CHAIN = {"protocol": (GlyphProtocol.NFT.value,), "token_type": "container", "name": "BTC"} +#: The spec'd shape, which nothing on mainnet uses. +_AS_SPECD = {"protocol": (GlyphProtocol.NFT.value, GlyphProtocol.CONTAINER.value), "name": "BTC"} + + +def _meta(**kw) -> GlyphMetadata: + return GlyphMetadata(**kw) + + +@pytest.mark.parametrize(("label", "kw"), [("as on chain", _AS_ON_CHAIN), ("as spec'd", _AS_SPECD)]) +class TestBothDeclarationsAreRecognisedEverywhere: + """One test class per site, so a fix applied to only one copy fails here.""" + + def test_the_is_container_property(self, label: str, kw: dict) -> None: + assert _meta(**kw).is_container is True, label + + def test_the_inspect_classifier(self, label: str, kw: dict) -> None: + assert _classify_metadata_protocol(_meta(**kw)) == "container", label + + def test_the_wave_mirror(self, label: str, kw: dict) -> None: + assert classify_glyph_metadata(_meta(**kw)) == "container", label + + +class TestItDoesNotOverReach: + """A guard that refuses valid work is a bug, and so is one that claims too much.""" + + def test_a_plain_nft_is_not_a_container(self) -> None: + m = _meta(protocol=(GlyphProtocol.NFT.value,), name="Just an NFT") + assert m.is_container is False + assert _classify_metadata_protocol(m) == "nft" + + def test_a_different_type_string_is_not_a_container(self) -> None: + m = _meta(protocol=(GlyphProtocol.NFT.value,), token_type="object", name="Not one") + assert m.is_container is False + assert _classify_metadata_protocol(m) == "nft" + + @pytest.mark.parametrize("spelling", ["Container", "CONTAINER", " container "]) + def test_case_and_whitespace_do_not_change_the_declaration(self, spelling: str) -> None: + """`type` is free CBOR text written by whatever minted the token; the four + mainnet containers all use lowercase, but nothing forces that.""" + assert _meta(protocol=(GlyphProtocol.NFT.value,), token_type=spelling).is_container is True diff --git a/tests/test_hashmark_attestation.py b/tests/test_hashmark_attestation.py new file mode 100644 index 00000000..740c8c7f --- /dev/null +++ b/tests/test_hashmark_attestation.py @@ -0,0 +1,181 @@ +"""v2 attestation — recovering the signer and requiring it to match the commitment. + +Decoding and attestation are SEPARATE steps with separate outcomes, and the spec +argues the separation earns its keep: "an invalid-signature record must never be +shown as a valid mark, but calling it malformed sends whoever is debugging it +after the wrong problem." A record that decodes is well-formed, not yet believed. + +The signer hash160 is committed TWICE — in the record and inside the signed +statement — and both are required. Without a value fixed in advance, recovery is +circular: an attacker would write whatever hash their chosen signature recovers to. +""" + +from __future__ import annotations + +import pytest + +from pyrxd.hash import hash256 +from pyrxd.keys import PrivateKey +from pyrxd.script.hashmark import ( + RADIANT_MAINNET_GENESIS, + AttestationOutcome, + HashMarkOutcome, + HashMarkRecord, + canonical_statement, + decode_hashmark, + verify_attestation, +) +from pyrxd.utils import text_digest + +_LABEL = "invoice-2026-09.pdf" +_DIGEST = hash256(b"the document bytes") + + +def _push(b: bytes) -> bytes: + return (bytes([len(b)]) + b) if len(b) <= 0x4B else (b"\x4c" + bytes([len(b)]) + b) + + +def _signed_record( + key: PrivateKey, + *, + digest: bytes = _DIGEST, + label: str | None = _LABEL, + signer: bytes | None = None, + genesis: str = RADIANT_MAINNET_GENESIS, +) -> bytes: + """Build a genuinely signed v2 record, the way a writer would.""" + signer_h = signer if signer is not None else key.public_key().hash160() + draft = HashMarkRecord( + HashMarkOutcome.OK, + version=2, + algorithm_id=1, + algorithm="sha256", + digest_hex=digest.hex(), + label=label, + signer_hash160_hex=signer_h.hex(), + signature_hex="00" * 65, + ) + sig = key.sign_recoverable(text_digest(canonical_statement(draft, network_genesis=genesis)), hasher=hash256) + header = 27 + sig[64] + 4 # compressed + pushes = [ + _push(b"HASHMARK"), + _push(bytes([2, 1])), + _push(digest), + _push(signer_h), + _push(bytes([header]) + sig[:64]), + ] + if label is not None: + pushes.append(_push(label.encode())) + return b"\x6a" + b"".join(pushes) + + +class TestAnHonestRecordVerifies: + """Paired with every refusal below — a guard that refuses valid work is a bug.""" + + def test_a_genuinely_signed_record_is_valid(self) -> None: + rec = decode_hashmark(_signed_record(PrivateKey())) + res = verify_attestation(rec) + assert res.valid + assert res.recovered_hash160_hex == rec.signer_hash160_hex + + def test_a_record_with_no_label_verifies(self) -> None: + """`label` is OMITTED from the statement when absent, never sent as an + empty string — a different statement, and so a different signature.""" + assert verify_attestation(decode_hashmark(_signed_record(PrivateKey(), label=None))).valid + + def test_a_non_ascii_label_verifies(self) -> None: + """The escaping rule matters here: `json.dumps` would emit \\uXXXX and + change the signed bytes for any label with an accent or an emoji.""" + assert verify_attestation(decode_hashmark(_signed_record(PrivateKey(), label="facture-café-☀"))).valid + + +class TestTamperingIsCaught: + @pytest.mark.parametrize("field", ["digest", "label", "signer"]) + def test_editing_a_committed_field_invalidates_the_signature(self, field: str) -> None: + """Every field in the statement is covered, so changing any of them in the + record makes the recovered key disagree with the commitment.""" + key = PrivateKey() + raw = bytearray(_signed_record(key)) + rec = decode_hashmark(bytes(raw)) + assert verify_attestation(rec).valid, "the untampered record must verify first" + + if field == "digest": + tampered = _signed_record(key) + tampered = tampered.replace(_DIGEST, hash256(b"a different document"), 1) + elif field == "label": + tampered = _signed_record(key).replace(_LABEL.encode(), b"paid-in-full.pdf!!!", 1) + else: + tampered = _signed_record(key, signer=PrivateKey().public_key().hash160()) + + res = verify_attestation(decode_hashmark(tampered)) + assert res.outcome is AttestationOutcome.INVALID_SIGNATURE, field + + def test_a_signature_from_a_DIFFERENT_key_does_not_attest(self) -> None: + """The commitment is what stops recovery being circular: an attacker who + signs the statement themselves still cannot match someone else's hash160.""" + victim, attacker = PrivateKey(), PrivateKey() + forged = _signed_record(attacker, signer=victim.public_key().hash160()) + assert verify_attestation(decode_hashmark(forged)).outcome is AttestationOutcome.INVALID_SIGNATURE + + +class TestTheStatementIsChainScoped: + def test_the_same_bytes_do_not_verify_against_another_genesis(self) -> None: + """`network` is in the signed statement and NOT in the record — it is the + verified context the transaction was found in. The same bytes on another + chain are a different statement, deliberately.""" + rec = decode_hashmark(_signed_record(PrivateKey())) + assert verify_attestation(rec).valid + other = verify_attestation(rec, network_genesis="00" * 32) + assert other.outcome is AttestationOutcome.INVALID_SIGNATURE + + +class TestMalformedSignatures: + def test_a_high_s_signature_is_refused(self) -> None: + """Low-S is mandatory: it removes the s versus n-s malleability so a + verifier has exactly one accepted form.""" + n = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141 + key = PrivateKey() + raw = _signed_record(key) + rec = decode_hashmark(raw) + sig = bytearray(bytes.fromhex(rec.signature_hex)) + s_val = int.from_bytes(sig[33:65], "big") + sig[33:65] = (n - s_val).to_bytes(32, "big") # the malleated twin + flipped = HashMarkRecord( + HashMarkOutcome.OK, + version=2, + algorithm_id=1, + algorithm="sha256", + digest_hex=rec.digest_hex, + label=rec.label, + signer_hash160_hex=rec.signer_hash160_hex, + signature_hex=bytes(sig).hex(), + ) + res = verify_attestation(flipped) + assert res.outcome is AttestationOutcome.INVALID_SIGNATURE + assert "low-S" in (res.detail or "") + + @pytest.mark.parametrize("header", [0, 26, 35, 255]) + def test_a_header_outside_27_34_is_refused(self, header: int) -> None: + rec = decode_hashmark(_signed_record(PrivateKey())) + sig = bytearray(bytes.fromhex(rec.signature_hex)) + sig[0] = header + bad = HashMarkRecord( + HashMarkOutcome.OK, + version=2, + algorithm_id=1, + algorithm="sha256", + digest_hex=rec.digest_hex, + label=rec.label, + signer_hash160_hex=rec.signer_hash160_hex, + signature_hex=bytes(sig).hex(), + ) + assert verify_attestation(bad).outcome is AttestationOutcome.INVALID_SIGNATURE + + +class TestV1IsNotAFailure: + def test_a_v1_record_is_NOT_ATTESTED_rather_than_invalid(self) -> None: + """v1 never claimed to say WHO — only WHEN. Reporting it as an invalid + signature would be reporting a claim it does not make.""" + v1 = b"\x6a" + _push(b"HASHMARK") + _push(bytes([1, 1])) + _push(_DIGEST) + res = verify_attestation(decode_hashmark(v1)) + assert res.outcome is AttestationOutcome.NOT_ATTESTED diff --git a/tests/test_hashmark_mainnet_vectors.py b/tests/test_hashmark_mainnet_vectors.py new file mode 100644 index 00000000..f8979efa --- /dev/null +++ b/tests/test_hashmark_mainnet_vectors.py @@ -0,0 +1,110 @@ +"""Real HashMark records from Radiant mainnet, produced by the REFERENCE writer. + +This is the thing pyrxd's own conformance vectors cannot be. Ours are generated +from our own builders, so they can only ever prove pyrxd agrees with pyrxd — +which is how we once published vectors that accepted an exploitable HTLC timelock +ordering and rejected the correct one. These bytes were written by a DIFFERENT +implementation, by a different author, from the same specification. + +Found by scanning 10,000 mainnet blocks (heights 451,021-461,021, roughly +2026-07-30 to 2026-09-03) for the `HASHMARK` magic. Three records, all from the +protocol's author testing ahead of launch. + +The v2 record is the valuable one: its signature was produced by his signer and +is verified here by a decoder written from `HASHMARK_PROTOCOL.md` alone, without +reading his source. Two independent implementations agreeing on a real +cryptographic attestation. + +If one of these ever stops decoding or verifying, either we broke something or +our reading of the spec was wrong — and both are worth being told about loudly. +""" + +from __future__ import annotations + +import pytest + +from pyrxd.base58 import base58check_encode +from pyrxd.constants import NETWORK_ADDRESS_PREFIX_DICT, Network +from pyrxd.script.hashmark import AttestationOutcome, decode_hashmark, verify_attestation + +#: Height 460,572. v2, signed. The signature verifies and the recovered key is the +#: creator address on the author's own Canon profile glyph — independently +#: confirmed, so these records are his rather than an unknown third party's. +_V2_SIGNED = bytes.fromhex( + "6a08484153484d41524b02020120e2c55efb34b6e9d6db008ee72d56bf86456ab3f55ae76488ff677fda88df1f1e" + "1426ba056431ec69cf27eabeaab250d99ddbd895d2411f750d18df9ab44ba66ced01285a5a067b9ebf7c8ff6b32d" + "ddb40cc276c5e98d4c2054937e44a40d7628d80cafdd6a372b0aae8f8bb31dbb4d975273a23e8c9771" +) +#: Height 459,905. v1, and it marks the SAME digest as the v2 above — the same +#: file marked under v1 and then again under v2, which is what testing an upgrade +#: looks like. +_V1_SAME_DIGEST = bytes.fromhex( + "6a08484153484d41524b02010120e2c55efb34b6e9d6db008ee72d56bf86456ab3f55ae76488ff677fda88df1f1e" +) +#: Height 460,364. v1, a different digest. +_V1_OTHER = bytes.fromhex( + "6a08484153484d41524b0201012049f82c41b6d6c78dbffe0df9014177b1b423171b5b6b7e09cccb68e4746dbc05" +) + +_SIGNER_HASH160 = "26ba056431ec69cf27eabeaab250d99ddbd895d2" +_SIGNER_ADDRESS = "14XmXG3dSBWZUukGT3xzS9zxpiZ53vgx1i" + + +class TestTheSignedMainnetRecord: + def test_it_decodes(self) -> None: + rec = decode_hashmark(_V2_SIGNED) + assert rec.ok and rec.version == 2 and rec.algorithm == "sha256" + assert rec.digest_hex == "e2c55efb34b6e9d6db008ee72d56bf86456ab3f55ae76488ff677fda88df1f1e" + assert rec.signer_hash160_hex == _SIGNER_HASH160 + assert rec.label is None + + def test_A_REAL_SIGNATURE_FROM_ANOTHER_IMPLEMENTATION_VERIFIES(self) -> None: + """The cross-implementation result, and the reason this file exists. + + His signer produced these bytes; our verifier — canonical statement, + magic hash, key recovery, commitment check — was written from the spec + without reading his code. It recovers the committed signer exactly. + """ + res = verify_attestation(decode_hashmark(_V2_SIGNED)) + assert res.outcome is AttestationOutcome.VALID + assert res.recovered_hash160_hex == _SIGNER_HASH160 + + def test_the_recovered_key_is_the_authors_known_address(self) -> None: + """Confirms whose records these are, from chain data rather than context: + the recovered key encodes the creator address on the author's own profile + glyph.""" + rec = decode_hashmark(_V2_SIGNED) + addr = base58check_encode( + NETWORK_ADDRESS_PREFIX_DICT[Network.MAINNET] + bytes.fromhex(rec.signer_hash160_hex or "") + ) + assert addr == _SIGNER_ADDRESS + + def test_it_does_not_verify_against_another_chain(self) -> None: + """`network` is in the signed statement and not in the record, so real + mainnet bytes must fail elsewhere. Pinned on real data because the + synthetic version of this test could pass with a broken statement builder + that happened to be broken consistently.""" + res = verify_attestation(decode_hashmark(_V2_SIGNED), network_genesis="00" * 32) + assert res.outcome is AttestationOutcome.INVALID_SIGNATURE + + +class TestTheUnsignedMainnetRecords: + @pytest.mark.parametrize( + ("raw", "digest"), + [ + (_V1_SAME_DIGEST, "e2c55efb34b6e9d6db008ee72d56bf86456ab3f55ae76488ff677fda88df1f1e"), + (_V1_OTHER, "49f82c41b6d6c78dbffe0df9014177b1b423171b5b6b7e09cccb68e4746dbc05"), + ], + ) + def test_v1_records_decode(self, raw: bytes, digest: str) -> None: + rec = decode_hashmark(raw) + assert rec.ok and rec.version == 1 and rec.digest_hex == digest + + def test_v1_is_not_attested_rather_than_invalid(self) -> None: + res = verify_attestation(decode_hashmark(_V1_SAME_DIGEST)) + assert res.outcome is AttestationOutcome.NOT_ATTESTED + + def test_the_v1_and_v2_records_mark_the_SAME_digest(self) -> None: + """Not an assertion about our code — a note about the corpus, kept as a + test so it cannot drift from the bytes above.""" + assert decode_hashmark(_V1_SAME_DIGEST).digest_hex == decode_hashmark(_V2_SIGNED).digest_hex diff --git a/tests/test_hashmark_wave_identity.py b/tests/test_hashmark_wave_identity.py new file mode 100644 index 00000000..27f4bc90 --- /dev/null +++ b/tests/test_hashmark_wave_identity.py @@ -0,0 +1,102 @@ +"""The key -> name bridge, and the gate in front of it. + +A HashMark v2 signer is a hash160, which is what a P2PKH address encodes, and +WAVE resolves names to addresses. So a verified signature can answer the question +a recipient actually has: *was this recorded by the holder of `company.rxd`?* + +The gate matters more than the lookup. Resolving an UNVERIFIED signer would dress +a claim up as an identity — anyone can put someone else's hash160 in a record — +which is the exact failure the signature check exists to prevent. +""" + +from __future__ import annotations + +import pytest + +from pyrxd.glyph.wave import wave_names_for_hash160 +from pyrxd.keys import PrivateKey +from pyrxd.security.errors import ValidationError + + +class _FakeIndexer: + """A fake at the REAL seam — the transport `RxinDexerClient` calls. + + `WaveResolver` wraps anything that is not already an `RxinDexerClient` in one, + so faking `wave_reverse_lookup` would bypass the wrapper the production path + actually goes through. Faking `call_extension` exercises it. + """ + + def __init__(self, names: dict[str, list[str]] | None = None) -> None: + self.names = names or {} + self.asked: list[str] = [] + + async def call_extension(self, method: str, params: list): + assert method == "wave.reverse_lookup", method + address = params[0] + self.asked.append(address) + return self.names.get(address, []) + + +class TestTheKeyToNameBridge: + @pytest.mark.asyncio + async def test_it_looks_up_the_signers_own_address(self) -> None: + """The join is hash160 -> address -> names, and the address must be the + one the signing key actually encodes.""" + key = PrivateKey() + addr = key.public_key().address() + idx = _FakeIndexer({addr: ["company.rxd", "invoices.company.rxd"]}) + + names = await wave_names_for_hash160(idx, key.public_key().hash160()) + + assert idx.asked == [addr], "must derive the signer's own address" + assert names == ["company.rxd", "invoices.company.rxd"] + + @pytest.mark.asyncio + async def test_a_key_owning_no_name_returns_empty_not_an_error(self) -> None: + """Most keys own no WAVE name. That is an answer, not a failure.""" + idx = _FakeIndexer() + assert await wave_names_for_hash160(idx, PrivateKey().public_key().hash160()) == [] + + @pytest.mark.asyncio + @pytest.mark.parametrize("bad", [b"", b"\x11" * 19, b"\x11" * 21]) + async def test_a_wrong_width_hash_is_refused(self, bad: bytes) -> None: + with pytest.raises(ValidationError, match="20 bytes"): + await wave_names_for_hash160(_FakeIndexer(), bad) + + +class TestTheGateInFrontOfIt: + """`_attach_wave_identity` must never resolve an unproven signer.""" + + @staticmethod + def _payload(outcome: str | None) -> dict: + hm: dict = {"version": 2, "signer_hash160": "aa" * 20} + if outcome is not None: + hm["attestation"] = {"outcome": outcome, "recovered_hash160": "aa" * 20} + return {"hashmark": hm} + + @pytest.mark.parametrize( + ("outcome", "expect"), + [ + ("invalid_signature", "unproven signer"), + ("not_attested", "no verified v2 signature"), + (None, "no verified v2 signature"), + ], + ) + def test_an_unverified_signature_is_NOT_resolved(self, outcome, expect: str) -> None: + """No network call, and a reason the operator can read.""" + from pyrxd.cli.glyph_inspect import _attach_wave_identity + + payload = self._payload(outcome) + # ctx is never used on this path — reaching for it would mean the gate failed. + _attach_wave_identity(None, payload) # type: ignore[arg-type] + + wi = payload["hashmark"]["wave_identity"] + assert wi["resolved"] is False + assert expect in wi["reason"] + + def test_a_payload_with_no_hashmark_is_untouched(self) -> None: + from pyrxd.cli.glyph_inspect import _attach_wave_identity + + payload: dict = {"type": "op_return"} + _attach_wave_identity(None, payload) # type: ignore[arg-type] + assert payload == {"type": "op_return"} diff --git a/tests/test_message_data_carrier.py b/tests/test_message_data_carrier.py new file mode 100644 index 00000000..50b639a9 --- /dev/null +++ b/tests/test_message_data_carrier.py @@ -0,0 +1,100 @@ +"""The Photonic ``msg`` convention — the only OP_RETURN format with real volume. + +``OP_RETURN PUSH3 "msg" ``. pyrxd already WROTE these (the dMint +miner's ``op_return_msg``, bounded by ``MAX_OP_RETURN_MSG_BYTES``) and could not +read one back, so the commonest data output on the chain rendered as opaque hex. + +Measured on 20 consecutive mainnet blocks: **73 of 73** ``OP_RETURN`` outputs +carried this marker, and nothing else did. +""" + +from __future__ import annotations + +import pytest + +from pyrxd.glyph._inspect_core import _inspect_script +from pyrxd.script.message import MSG_MARKER, MessageOutcome, decode_message + + +def _push(b: bytes) -> bytes: + return (bytes([len(b)]) + b) if len(b) <= 0x4B else (b"\x4c" + bytes([len(b)]) + b) + + +def _msg(body: bytes) -> bytes: + return b"\x6a" + _push(MSG_MARKER) + _push(body) + + +#: A REAL output, copied byte-for-byte off Radiant mainnet. Chosen because it is +#: not the clean case anyone would invent: "Radiate " + U+1F31E + a NUL + "33". +#: Real data carries control characters, and a fixture that only ever holds tidy +#: ASCII cannot show whether the display path handles what the chain contains. +_REAL_MAINNET = bytes.fromhex("6a036d73670f5261646961746520f09f8c9e003333") + + +class TestRealMainnetOutput: + def test_it_decodes(self) -> None: + r = decode_message(_REAL_MAINNET) + assert r.ok and r.is_utf8 + assert r.raw == b"Radiate \xf0\x9f\x8c\x9e\x0033" + assert r.text is not None and r.text.startswith("Radiate \U0001f31e") + + def test_the_classifier_sanitises_the_control_byte_for_display(self) -> None: + """The decoder returns the bytes unmangled — a caller verifying a + commitment needs them — and the DISPLAY boundary sanitises. The NUL in + this real message is what proves the two are separate.""" + row = _inspect_script(_REAL_MAINNET.hex()) + assert row["type"] == "op_return-msg" + assert "\x00" not in row["message"]["text"] + assert row["message"]["byte_length"] == 15 + + +class TestDecoding: + def test_utf8_text(self) -> None: + r = decode_message(_msg(b"gm radiant")) + assert r.ok and r.text == "gm radiant" and r.is_utf8 + + def test_non_utf8_is_REPORTED_not_refused(self) -> None: + """The bytes are already on chain and nothing constrains them to text. + Refusing would lose a record we can otherwise describe exactly.""" + r = decode_message(_msg(b"\xff\xfe\x00")) + assert r.ok and r.is_utf8 is False and r.text is None + assert r.raw == b"\xff\xfe\x00" + + @pytest.mark.parametrize( + ("script", "why"), + [ + (b"\x76\xa9\x14" + b"\x11" * 20 + b"\x88\xac", "a P2PKH"), + (b"\x6a" + _push(b"OTHER") + _push(b"x"), "another protocol's marker"), + (b"\x6a", "a bare OP_RETURN"), + (b"", "empty"), + ], + ) + def test_anything_else_is_skipped_silently(self, script: bytes, why: str) -> None: + assert decode_message(script).outcome is MessageOutcome.NOT_MESSAGE, why + + @pytest.mark.parametrize( + ("script", "why"), + [ + (b"\x6a" + _push(MSG_MARKER), "marker with no message"), + (b"\x6a" + _push(MSG_MARKER) + _push(b"a") + _push(b"b"), "a third push"), + (b"\x6a" + _push(MSG_MARKER) + b"\x00", "empty message push"), + ], + ) + def test_a_claimed_message_that_is_malformed_is_reported(self, script: bytes, why: str) -> None: + """Past the marker the output claims to be a message, so a bad shape is a + defect rather than another protocol.""" + assert decode_message(script).outcome is MessageOutcome.INVALID, why + + +class TestItStaysAdditive: + def test_a_non_message_data_output_is_unchanged(self) -> None: + row = _inspect_script((b"\x6a" + _push(b"OTHERPROTO") + _push(b"payload here, long enough")).hex()) + assert row["type"] == "op_return" + assert "message" not in row + + def test_a_hashmark_is_not_claimed_as_a_message(self) -> None: + """Two decoders on the same branch must not fight over one script.""" + hm = b"\x6a" + _push(b"HASHMARK") + _push(bytes([1, 1])) + _push(bytes(32)) + row = _inspect_script(hm.hex()) + assert row["type"] == "op_return-hashmark-v1" + assert "message" not in row