"""muse.core.msign — canonical MSign signing primitives. Single source of truth for MSign authentication across the muse ecosystem: MuseHub (repo VCS), Stori (DAW session auth), Maestro (NL→MIDI pipeline), and MPay (real-time crypto micropayments). Wire format ----------- Every authenticated HTTP request carries:: Authorization: MSign handle="" alg="ed25519" ts= sig="" Canonical message (the bytes Ed25519 signs):: {algorithm}\\n{METHOD}\\n{host}\\n{path_with_query}\\n{ts}\\n{body_hash} where: - ``algorithm`` is the signing algorithm identifier (``"ed25519"``). - ``METHOD`` is the HTTP verb in uppercase (``"POST"``, ``"GET"``, …). - ``host`` is the lowercased hostname with non-standard port kept (``"staging.musehub.ai"``, ``"localhost:1337"``); standard ports (80 for http, 443 for https) are stripped. - ``path_with_query`` is the URL path including ``?query`` if present. - ``ts`` is a decimal integer Unix timestamp (seconds since epoch). - ``body_hash`` is ``"sha256:" + sha256(body).hexdigest()``; empty body → ``"sha256:e3b0c442…"``. Algorithm: Ed25519 (RFC 8032). Signatures are base64url-encoded with no padding (``=`` stripped). Replay window: ±30 seconds from server time. MPay extension -------------- ``build_payment_claim()`` uses the same Ed25519 key with a domain-separated canonical message:: MPAY\\nFROM\\nTO\\nAMOUNT_NANO\\nCURRENCY\\nNONCE_HEX\\nMEMO The ``MPAY`` domain separator prevents cross-protocol signature confusion. Payment claims can be chained (each ``nonce_hex`` is the SHA-256 of the previous claim's signature), enabling real-time streaming micropayments that batch-settle on Avalanche L1. Public API ---------- :func:`canonical_message` — build the bytes to sign for an HTTP request :func:`build_msign_header` — produce ``Authorization: MSign …`` value :func:`parse_msign_header` — parse a header value into its components :func:`verify_msign_header` — verify a header against a known public key :func:`build_payment_claim` — sign an MPay micropayment attestation """ import re import time import urllib.parse from typing import TYPE_CHECKING, Any, TypedDict import hashlib as _hashlib from muse.core.types import DEFAULT_SIGN_ALGO, b64url_decode, b64url_encode if TYPE_CHECKING: from muse.core.transport import SigningIdentity # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- REPLAY_WINDOW_SECONDS: int = 30 """Maximum age (in seconds) of an MSign timestamp accepted by the server.""" # Standard ports whose inclusion in the host header is redundant. _STANDARD_PORTS: dict[str, int] = {"https": 443, "http": 80} # Regex for parsing Authorization header values. _MSIGN_RE = re.compile( r'MSign\s+' r'handle="(?P[^"]+)"\s+' r'alg="(?P[^"]+)"\s+' r'ts=(?P\d+)\s+' r'sig="(?P[A-Za-z0-9_=-]+)"' ) # sha256:-prefixed raw hash of empty bytes — the "no body" sentinel. EMPTY_BODY_HASH: str = "sha256:" + _hashlib.sha256(b"").hexdigest() # --------------------------------------------------------------------------- # TypedDicts for structured return values # --------------------------------------------------------------------------- class ParsedMSign(TypedDict): """Parsed components of an MSign Authorization header value.""" handle: str alg: str ts: int sig: str class _PaymentClaimRequired(TypedDict): """Required fields present on every MPay claim.""" from_handle: str to_handle: str amount_nano: int currency: str nonce_hex: str memo: str ts: int signature_b64: str canonical_message: str class PaymentClaim(_PaymentClaimRequired, total=False): """Signed MPay micropayment attestation. The required fields (inherited from :class:`_PaymentClaimRequired`) are always present. The optional AVAX dual-signature fields are populated when a secp256k1 key is supplied to :func:`build_payment_claim`. A claim that carries both an Ed25519 MSign signature **and** an EIP-191 secp256k1 signature can be verified entirely on Avalanche C-Chain via ``ecrecover`` — enabling trust-minimised on-chain settlement without any off-chain oracle. Optional fields --------------- payer_avax_address: EIP-55 checksummed AVAX C-Chain address of the payer (``0x…``), derived from the secp256k1 key used to produce ``eth_sig``. recipient_avax_address: EIP-55 checksummed AVAX C-Chain address of the recipient (``0x…``), supplied by the caller as the on-chain settlement target. eth_sig: Hex-encoded 65-byte EIP-191 ``personal_sign`` signature over the same canonical MPAY message, produced with the payer's secp256k1 key. ``v`` is 27 or 28 (legacy Ethereum recovery ID convention). Verifiable on EVM chains with ``ecrecover(keccak256(prefixed_msg), sig)``. """ payer_avax_address: str recipient_avax_address: str eth_sig: str # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _normalise_host(parsed_url: urllib.parse.ParseResult) -> str: """Return the canonical host string for inclusion in the signed payload. Lowercased; standard ports stripped (:443 for https, :80 for http); non-standard ports kept (localhost:1337). """ host = (parsed_url.hostname or "").lower() port = parsed_url.port scheme = parsed_url.scheme.lower() if port is not None and port != _STANDARD_PORTS.get(scheme): return f"{host}:{port}" return host # --------------------------------------------------------------------------- # Core primitives # --------------------------------------------------------------------------- def canonical_message( method: str, path_with_query: str, ts: int, body_bytes: bytes, host: str, algorithm: str = DEFAULT_SIGN_ALGO, ) -> bytes: """Return the bytes that Ed25519 signs for an MSign HTTP request. Args: method: HTTP verb in uppercase (``"POST"``, ``"GET"``, …). path_with_query: URL path including ``?query`` string if present. ts: Unix timestamp (integer seconds). body_bytes: Raw request body (``b""`` for requests with no body). host: Canonical host string (lowercased; standard ports stripped). E.g. ``"staging.musehub.ai"`` or ``"localhost:1337"``. algorithm: Signing algorithm identifier (default ``"ed25519"``). Returns: UTF-8 encoded canonical message ready for signing. Example:: msg = canonical_message( "POST", "/gabriel/muse/push", 1744000000, b"", host="staging.musehub.ai", ) # b"ed25519\\nPOST\\nstaging.musehub.ai\\n/gabriel/muse/push\\n1744000000\\ne3b0c44..." """ body_hash = "sha256:" + _hashlib.sha256(body_bytes).hexdigest() return f"{algorithm}\n{method.upper()}\n{host}\n{path_with_query}\n{ts}\n{body_hash}".encode() def build_msign_header( signing: "SigningIdentity", method: str, url: str, body_bytes: bytes | None = None, *, ts: int | None = None, ) -> str: """Return an ``Authorization: MSign …`` header value for a request. Args: signing: Ed25519 signing identity (handle + private key). method: HTTP verb (``"POST"``, ``"GET"``, …). url: Full request URL including scheme and host. body_bytes: Raw request body bytes, or ``None`` for an empty body. ts: Unix timestamp override. **Use only in tests** to produce deterministic output. Defaults to ``int(time.time())``. Returns: Complete ``Authorization`` header value, e.g.:: MSign handle="gabriel" alg="ed25519" ts=1744000000 sig="aBcDeFg..." Raises: AssertionError: If ``signing.private_key`` is not an :class:`~cryptography.hazmat.primitives.asymmetric.ed25519.Ed25519PrivateKey`. """ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey parsed = urllib.parse.urlparse(url) path_with_query = parsed.path if parsed.query: path_with_query = f"{path_with_query}?{parsed.query}" host = _normalise_host(parsed) if ts is None: ts = int(time.time()) body = body_bytes or b"" msg = canonical_message(method, path_with_query, ts, body, host=host) private_key = signing.private_key assert isinstance(private_key, Ed25519PrivateKey) sig_bytes = private_key.sign(msg) sig_b64 = b64url_encode(sig_bytes) return f'MSign handle="{signing.handle}" alg="{DEFAULT_SIGN_ALGO}" ts={ts} sig="{sig_b64}"' def parse_msign_header(header_value: str) -> ParsedMSign: """Parse an MSign Authorization header value into its components. Args: header_value: Full ``Authorization`` header value, e.g. ``MSign handle="gabriel" alg="ed25519" ts=1744000000 sig="aBcD..."`` Returns: :class:`ParsedMSign` with ``handle`` (str), ``alg`` (str), ``ts`` (int), ``sig`` (str, base64url). Raises: ValueError: If *header_value* does not match the MSign format. """ m = _MSIGN_RE.search(header_value) if not m: raise ValueError( f"Not a valid MSign header: {header_value!r}. " 'Expected: MSign handle="" alg="" ts= sig=""' ) return ParsedMSign( handle=m.group("handle"), alg=m.group("alg"), ts=int(m.group("ts")), sig=m.group("sig"), ) def verify_msign_header( header_value: str, method: str, url: str, body_bytes: bytes | None, public_key_b64: str, *, max_age: int = REPLAY_WINDOW_SECONDS, now: int | None = None, ) -> tuple[bool, str]: """Verify an MSign Authorization header against a known Ed25519 public key. Args: header_value: The ``Authorization`` header value to verify. method: HTTP verb used for the request. url: Full request URL. body_bytes: Raw request body bytes (``None`` for empty body). public_key_b64: URL-safe base64 Ed25519 public key (no padding). max_age: Maximum timestamp age in seconds (default :data:`REPLAY_WINDOW_SECONDS` = 30). now: Override current time (Unix seconds) for testing. Returns: ``(True, "ok")`` on success. ``(False, reason)`` on failure, where *reason* is a human-readable explanation of why verification failed. Example:: ok, reason = verify_msign_header( header, "POST", url, body, public_key_b64 ) if not ok: raise HTTPException(401, reason) """ from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey # 1. Parse header. try: parsed = parse_msign_header(header_value) except ValueError as exc: return False, str(exc) ts = parsed["ts"] alg = parsed["alg"] # 2. Replay-window check. if now is None: now = int(time.time()) age = abs(now - ts) if age > max_age: return False, ( f"timestamp {ts} is {age}s outside the ±{max_age}s replay window " f"(server time={now})" ) # 3. Reconstruct canonical message. parsed_url = urllib.parse.urlparse(url) path_with_query = parsed_url.path if parsed_url.query: path_with_query = f"{path_with_query}?{parsed_url.query}" host = _normalise_host(parsed_url) body = body_bytes or b"" msg = canonical_message(method, path_with_query, ts, body, host=host, algorithm=alg) # 4. Decode public key. try: pub_bytes = b64url_decode(public_key_b64) pub_key = Ed25519PublicKey.from_public_bytes(pub_bytes) except Exception as exc: return False, f"invalid public key: {exc}" # 5. Decode and verify signature. try: sig_b64 = parsed["sig"] sig_bytes = b64url_decode(sig_b64) pub_key.verify(sig_bytes, msg) except InvalidSignature: return False, "Ed25519 signature verification failed" except Exception as exc: return False, f"signature decode error: {exc}" return True, "ok" # --------------------------------------------------------------------------- # MPay micropayment primitives # --------------------------------------------------------------------------- def build_payment_claim( signing: "SigningIdentity", from_handle: str, to_handle: str, amount_nano: int, currency: str, nonce_hex: str, memo: str, *, ts: int | None = None, avax_private_key: "Any | None" = None, recipient_avax_address: str | None = None, ) -> PaymentClaim: """Sign an MPay micropayment claim, optionally with AVAX dual-signature. Uses the same Ed25519 key as MSign HTTP auth with a ``MPAY`` domain separator to prevent cross-protocol signature confusion. Canonical message:: MPAY\\nFROM\\nTO\\nAMOUNT_NANO\\nCURRENCY\\nNONCE_HEX\\nMEMO\\nTIMESTAMP Chain linkage: set ``nonce_hex`` to ``sha256(prev_claim["signature_b64"])`` to create a tamper-evident payment chain settleable on Avalanche L1. AVAX dual-signature (``avax_private_key``) ------------------------------------------- When *avax_private_key* is supplied, the **same canonical MPAY message** is also signed with EIP-191 ``personal_sign`` using the caller's secp256k1 key. The resulting 65-byte signature (``r || s || v``, ``v`` = 27 or 28) is stored in ``eth_sig`` as a lowercase hex string. The payer's AVAX C-Chain address is derived from the public key and stored in ``payer_avax_address``. On-chain verification (Solidity sketch):: bytes32 digest = keccak256(abi.encodePacked( "\\x19Ethereum Signed Message:\\n", Strings.toString(canonicalBytes.length), canonicalBytes )); address recovered = ecrecover(digest, v, r, s); require(recovered == claim.payerAvaxAddress, "bad sig"); Args: signing: Ed25519 signing identity. from_handle: Sender's hub handle. to_handle: Recipient's hub handle. amount_nano: Payment amount in nanoMUSE (1 MUSE = 1_000_000_000 nanoMUSE). currency: Currency identifier (``"nanoMUSE"``, ``"nanoETH"``, …). nonce_hex: Hex nonce — use ``sha256(prev_sig_b64)`` for chain linkage, or a fresh random 32-byte hex for the first payment. memo: Free-form memo string (e.g. ``"stem:sha256:abc123"``). ts: Unix timestamp override (testing only). avax_private_key: Optional ``eth_keys.PrivateKey`` for AVAX C-Chain dual-signing. Obtain via :func:`~muse.core.secp256k1_sign.derive_avax_key`. When ``None``, the AVAX fields are omitted. recipient_avax_address: EIP-55 checksummed AVAX C-Chain address of the recipient. Stored verbatim — not derived from any key. Pass when the recipient's on-chain address is known at call time. Returns: :class:`PaymentClaim` with all required fields. Optional AVAX fields (``payer_avax_address``, ``eth_sig``, ``recipient_avax_address``) are present only when the corresponding arguments are supplied. """ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey if ts is None: ts = int(time.time()) canonical = ( f"MPAY\n{from_handle}\n{to_handle}\n{amount_nano}\n" f"{currency}\n{nonce_hex}\n{memo}\n{ts}" ).encode() private_key = signing.private_key assert isinstance(private_key, Ed25519PrivateKey) sig_bytes = private_key.sign(canonical) sig_b64 = b64url_encode(sig_bytes) claim = PaymentClaim( from_handle=from_handle, to_handle=to_handle, amount_nano=amount_nano, currency=currency, nonce_hex=nonce_hex, memo=memo, ts=ts, signature_b64=sig_b64, canonical_message=canonical.decode(), ) if avax_private_key is not None: from muse.core.secp256k1_sign import avax_c_chain_address, eip191_sign eth_sig_bytes = eip191_sign(avax_private_key, canonical) claim["payer_avax_address"] = avax_c_chain_address(avax_private_key.public_key) claim["eth_sig"] = eth_sig_bytes.hex() if recipient_avax_address is not None: claim["recipient_avax_address"] = recipient_avax_address return claim