#!/usr/bin/env python3 """ nukez_sign.py - canonical Nukez signed-envelope builder. Dependencies: - Python 3.8+ standard library - openssl CLI with Ed25519 support No pip packages. No gateway imports. The gateway currently expects: X-Nukez-Envelope: base64url(canonical_json(envelope)), no padding X-Nukez-Signature: base58(Ed25519 signature over canonical envelope bytes) This script reads the exact request body bytes from stdin. For JSON requests, send the same bytes to curl that you piped into this script. """ from __future__ import annotations import argparse import base64 import hashlib import json import os import subprocess import sys import tempfile import time from urllib.parse import parse_qsl, urlencode, urlsplit B58_ALPHABET = b"123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" EMPTY_SHA256 = hashlib.sha256(b"").hexdigest() def b58encode(data: bytes) -> str: n = int.from_bytes(data, "big") out: list[bytes] = [] while n: n, r = divmod(n, 58) out.append(B58_ALPHABET[r : r + 1]) for byte in data: if byte == 0: out.append(b"1") else: break return b"".join(reversed(out or [b"1"])).decode("ascii") def b64url(data: bytes) -> str: return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") def canonical_json_bytes(obj: object) -> bytes: return json.dumps( obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False, ).encode("utf-8") def canonical_query(query: str) -> str: query = (query or "").lstrip("?") if not query: return "" pairs = parse_qsl(query, keep_blank_values=True) pairs.sort() return urlencode(pairs, doseq=True) def split_path_and_query(path: str, query: str) -> tuple[str, str]: parsed = urlsplit(path) if parsed.scheme or parsed.netloc: raise SystemExit("--path must be a request path, not a full URL") clean_path = parsed.path or path merged_query = query or parsed.query return clean_path, canonical_query(merged_query) def stable_locker_id(receipt_id: str) -> str: return "locker_" + hashlib.sha256(receipt_id.encode("utf-8")).hexdigest()[:12] def load_solana_keypair(path: str) -> tuple[bytes, bytes]: with open(path, "r", encoding="utf-8") as f: raw = json.load(f) if not isinstance(raw, list) or len(raw) != 64: got = len(raw) if isinstance(raw, list) else type(raw).__name__ raise SystemExit(f"expected 64-byte Solana keypair array in {path}; got {got}") key = bytes(int(b) & 0xFF for b in raw) return key[:32], key[32:] def seed_to_pkcs8_pem(seed: bytes) -> str: if len(seed) != 32: raise ValueError("Ed25519 seed must be 32 bytes") der = bytes.fromhex("302e020100300506032b657004220420") + seed body = base64.b64encode(der).decode("ascii") wrapped = "\n".join(body[i : i + 64] for i in range(0, len(body), 64)) return f"-----BEGIN PRIVATE KEY-----\n{wrapped}\n-----END PRIVATE KEY-----\n" def ed25519_sign(seed: bytes, message: bytes) -> bytes: pem = seed_to_pkcs8_pem(seed) key_path = "" msg_path = "" try: with tempfile.NamedTemporaryFile("w", suffix=".pem", delete=False) as key_file: key_file.write(pem) key_path = key_file.name with tempfile.NamedTemporaryFile("wb", suffix=".bin", delete=False) as msg_file: msg_file.write(message) msg_path = msg_file.name result = subprocess.run( ["openssl", "pkeyutl", "-sign", "-inkey", key_path, "-rawin", "-in", msg_path], check=True, capture_output=True, ) if len(result.stdout) != 64: raise RuntimeError(f"expected 64-byte Ed25519 signature, got {len(result.stdout)}") return result.stdout finally: for path in (key_path, msg_path): if path: try: os.unlink(path) except FileNotFoundError: pass def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Build Nukez signed-envelope headers") parser.add_argument("--keypair", required=True, help="Solana CLI keypair JSON") parser.add_argument("--method", required=True, help="HTTP method, e.g. POST or GET") parser.add_argument("--path", required=True, help="Request path, e.g. /v1/lockers/{id}/files") parser.add_argument("--query", default="", help="Optional query string without leading ?") parser.add_argument("--receipt-id", required=True, help="Receipt id bound to the request") parser.add_argument("--ops", required=True, help='JSON array of required ops, e.g. ["locker:list"]') parser.add_argument("--ttl", type=int, default=60, help="Envelope TTL seconds; max gateway default is 300") return parser.parse_args() def main() -> None: args = parse_args() if args.ttl <= 0: raise SystemExit("--ttl must be positive") body = sys.stdin.buffer.read() if not sys.stdin.isatty() else b"" path, query = split_path_and_query(args.path, args.query) ops = json.loads(args.ops) if not isinstance(ops, list) or not ops or not all(isinstance(op, str) for op in ops): raise SystemExit("--ops must be a non-empty JSON string array") seed, pubkey = load_solana_keypair(args.keypair) now = int(time.time()) envelope = { "v": 1, "locker_id": stable_locker_id(args.receipt_id), "receipt_id": args.receipt_id, "nonce": os.urandom(16).hex(), "iat": now, "exp": now + args.ttl, "ops": ops, "method": args.method.upper(), "path": path, "body_sha256": hashlib.sha256(body).hexdigest(), } if query: envelope["query"] = query env_bytes = canonical_json_bytes(envelope) sig = ed25519_sign(seed, env_bytes) x_dl_envelope = b64url(env_bytes) x_dl_signature = b58encode(sig) print( json.dumps( { "x_dl_envelope": x_dl_envelope, "x_dl_signature": x_dl_signature, "headers": { "X-Nukez-Envelope": x_dl_envelope, "X-Nukez-Signature": x_dl_signature, }, "locker_id": envelope["locker_id"], "pubkey_b58": b58encode(pubkey), "body_sha256": envelope["body_sha256"], "body_bytes": len(body), "empty_body_sha256": EMPTY_SHA256, "envelope": envelope, }, indent=2, ensure_ascii=False, ) ) if __name__ == "__main__": main()