cayce
a282aedb94
All 32 findings resolved: regex literals kept single-line with noqa (pattern IS the reviewed artifact), HTML rows string-concatenated for line length, walrus tuple in BIP-340 verify split into named generator, unused pytest import dropped. No behavior changes; 44/44 tests pass. Full-gate discipline restored: ruff exit 0.
177 lines
6.3 KiB
Python
177 lines
6.3 KiB
Python
"""Minimal Nostr client: NIP-01 kind-1 notes (build, sign, verify, publish).
|
|
|
|
Implemented in-repo (DECISIONS D4) so every byte of the signing path is
|
|
auditable: event id = sha256 of the canonical NIP-01 serialization; signature =
|
|
BIP-340 Schnorr over the id. Keys are 32-byte hex. Relays are plain env config.
|
|
|
|
Publishing is fire-and-wait: open a websocket per relay, send the EVENT, wait
|
|
for the OK frame (or timeout), and return the per-relay results.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from typing import Any
|
|
|
|
from coincurve import PrivateKey, PublicKey
|
|
|
|
KIND_NOTE = 1
|
|
|
|
|
|
def normalize_key(private_key_hex: str) -> bytes:
|
|
"""Accept 64-char hex (with or without 0x). Anything else is a config error."""
|
|
key = private_key_hex.strip().lower().removeprefix("0x")
|
|
if len(key) != 64 or not all(c in "0123456789abcdef" for c in key):
|
|
raise ValueError("CAYCE_OPS_NOSTR_PRIVATE_KEY must be 64-char hex (32 bytes)")
|
|
return bytes.fromhex(key)
|
|
|
|
|
|
def pubkey_from_secret(secret: bytes) -> str:
|
|
return PublicKey.from_secret(secret).format().hex()[2:]
|
|
|
|
|
|
def build_event(
|
|
secret: bytes,
|
|
content: str,
|
|
tags: list[list[str]] | None = None,
|
|
created_at: int | None = None,
|
|
kind: int = KIND_NOTE,
|
|
) -> dict[str, Any]:
|
|
"""Build and sign a NIP-01 event. Deterministic given the same inputs."""
|
|
created_at = int(created_at if created_at is not None else __import__("time").time())
|
|
pubkey = pubkey_from_secret(secret)
|
|
serialized = json.dumps(
|
|
[0, pubkey, created_at, kind, tags or [], content],
|
|
separators=(",", ":"), # canonical NIP-01 form
|
|
ensure_ascii=False,
|
|
)
|
|
event_id = hashlib.sha256(serialized.encode("utf-8")).hexdigest()
|
|
# zero aux randomness -> fully deterministic signatures (BIP-340 permits it;
|
|
# verification is unaffected either way)
|
|
signature = (
|
|
PrivateKey(secret)
|
|
.sign_schnorr(bytes.fromhex(event_id), aux_randomness=b"\x00" * 32)
|
|
.hex()
|
|
)
|
|
return {
|
|
"pubkey": pubkey,
|
|
"created_at": created_at,
|
|
"kind": kind,
|
|
"tags": tags or [],
|
|
"content": content,
|
|
"id": event_id,
|
|
"sig": signature,
|
|
}
|
|
|
|
|
|
def _schnorr_verify(sig32: bytes, msg32: bytes, xonly_pub32: bytes) -> bool:
|
|
"""BIP-340 reference verification (pure Python, public domain spec code).
|
|
|
|
Signing runs in coincurve (libsecp256k1); verification is pure Python so
|
|
the auditor can read every step without CFFI indirection.
|
|
"""
|
|
P = 2**256 - 2**32 - 977
|
|
N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141
|
|
|
|
def tag_hash(tag: str, msg: bytes) -> bytes:
|
|
t = hashlib.sha256(tag.encode()).digest()
|
|
return hashlib.sha256(t + t + msg).digest()
|
|
|
|
def lift_x(b: int):
|
|
if b >= P:
|
|
return None
|
|
y_sq = (pow(b, 3, P) + 7) % P
|
|
y = pow(y_sq, (P + 1) // 4, P)
|
|
if pow(y, 2, P) != y_sq:
|
|
return None
|
|
return (b, y if y % 2 == 0 else P - y)
|
|
|
|
def point_add(p1, p2):
|
|
if p1 is None:
|
|
return p2
|
|
if p2 is None:
|
|
return p1
|
|
if p1[0] == p2[0] and p1[1] != p2[1]:
|
|
return None
|
|
if p1 == p2:
|
|
lam = (3 * p1[0] * p1[0] * pow(2 * p1[1], P - 2, P)) % P
|
|
else:
|
|
lam = ((p2[1] - p1[1]) * pow(p2[0] - p1[0], P - 2, P)) % P
|
|
x3 = (lam * lam - p1[0] - p2[0]) % P
|
|
return (x3, (lam * (p1[0] - x3) - p1[1]) % P)
|
|
|
|
def point_mul(p, scalar):
|
|
acc = None
|
|
for i in range(256):
|
|
if (scalar >> i) & 1:
|
|
acc = point_add(acc, p)
|
|
p = point_add(p, p)
|
|
return acc
|
|
|
|
if len(sig32) != 64 or len(msg32) != 32 or len(xonly_pub32) != 32:
|
|
return False
|
|
r = int.from_bytes(sig32[:32], "big")
|
|
s = int.from_bytes(sig32[32:], "big")
|
|
if not (0 <= r < P and 0 <= s < N):
|
|
return False
|
|
pub_point = lift_x(int.from_bytes(xonly_pub32, "big"))
|
|
if pub_point is None:
|
|
return False
|
|
e = int.from_bytes(tag_hash("BIP0340/challenge", sig32[:32] + xonly_pub32 + msg32), "big") % N
|
|
generator = (
|
|
0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798, # Gx
|
|
0x483ADA7726A3C4655DA4FBFC0E1108A8FD17B448A68554199C47D08FFB10D4B8, # Gy
|
|
)
|
|
r_point = point_add(
|
|
point_mul(generator, s),
|
|
point_mul(pub_point, (N - e) % N),
|
|
)
|
|
if r_point is None or r_point[1] % 2 != 0:
|
|
return False
|
|
return r_point[0] == r
|
|
|
|
|
|
def verify_event(event: dict[str, Any]) -> bool:
|
|
"""Recompute the id and verify the Schnorr signature against the pubkey."""
|
|
serialized = json.dumps(
|
|
[0, event["pubkey"], event["created_at"], event["kind"], event["tags"], event["content"]],
|
|
separators=(",", ":"),
|
|
ensure_ascii=False,
|
|
)
|
|
if hashlib.sha256(serialized.encode("utf-8")).hexdigest() != event["id"]:
|
|
return False
|
|
# 33-byte compressed point (02-prefixed x-only)
|
|
compressed = PublicKey(bytes.fromhex("02" + event["pubkey"])).format()
|
|
return _schnorr_verify(
|
|
bytes.fromhex(event["sig"]), bytes.fromhex(event["id"]), compressed[1:]
|
|
)
|
|
|
|
|
|
def relay_publish(relay_url: str, event: dict[str, Any], timeout: float = 10.0) -> str:
|
|
"""Publish one event to one relay; return the relay's OK message."""
|
|
from websockets.sync.client import connect
|
|
|
|
with connect(relay_url, close_timeout=timeout) as ws:
|
|
ws.send(json.dumps(["EVENT", event]))
|
|
reply = json.loads(ws.recv(timeout=timeout))
|
|
if not (isinstance(reply, list) and len(reply) >= 3 and reply[0] == "OK" and reply[2] is True):
|
|
raise RuntimeError(f"relay {relay_url} rejected event: {str(reply)[:200]}")
|
|
return str(reply[1])
|
|
|
|
|
|
def publish(
|
|
relays: list[str], event: dict[str, Any], timeout: float = 10.0
|
|
) -> list[dict[str, str]]:
|
|
"""Publish to every relay; report per-relay outcome (never raises unless all fail)."""
|
|
results = []
|
|
for relay in relays:
|
|
try:
|
|
relay_publish(relay, event, timeout)
|
|
results.append({"relay": relay, "ok": "true"})
|
|
except Exception as exc: # noqa: BLE001 — per-relay isolation
|
|
results.append({"relay": relay, "ok": "false", "error": str(exc)[:200]})
|
|
if relays and all(r["ok"] == "false" for r in results):
|
|
raise RuntimeError(f"all relays failed: {results}")
|
|
return results
|