2ro
b35ddb29d3
Extend smoke_test_lockdown.py from two checks to four, all signed by one
throwaway unauthorized key:
kind 1 text note -> BLOCKED (lockdown)
kind 0 profile -> ACCEPTED (profiles stay open)
kind 1059 gift wrap -> ACCEPTED (money path ungated; p-tag a
random recipient, opaque base64 content)
kind 30023 long-form article -> BLOCKED (lockdown)
Prints all four results; RESULT: PASS and exit 0 only if every expectation
holds, nonzero otherwise. Verified offline end to end against a local mock
relay driven by the real plugin decide() (PASS) and against an accept-all
relay (FAIL, exit 1).
305 lines
9.6 KiB
Python
305 lines
9.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Post-deploy smoke test for the public-note lockdown.
|
|
|
|
Signs four throwaway events with a fresh (unauthorized) key and publishes
|
|
them to a live relay, then checks the relay's OK responses:
|
|
|
|
* kind 1 (text note) -> expected BLOCKED (author not authorized)
|
|
* kind 0 (profile) -> expected ACCEPTED (profiles stay open)
|
|
* kind 1059 (gift wrap) -> expected ACCEPTED (money path ungated)
|
|
* kind 30023 (long-form article) -> expected BLOCKED (author not authorized)
|
|
|
|
Exit 0 only if all four expectations hold. Zero third-party dependencies: a
|
|
compact BIP-340 Schnorr signer and a minimal RFC-6455 client, both stdlib.
|
|
|
|
Usage:
|
|
./smoke_test_lockdown.py wss://relay.example.com
|
|
./smoke_test_lockdown.py wss://relay.example.com --insecure # skip TLS verify
|
|
"""
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import socket
|
|
import ssl
|
|
import sys
|
|
import time
|
|
from urllib.parse import urlparse
|
|
|
|
# --- BIP-340 Schnorr (secp256k1), reference-style, stdlib only ---------------
|
|
|
|
_P = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEFFFFFC2F
|
|
_N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141
|
|
_G = (
|
|
0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798,
|
|
0x483ADA7726A3C4655DA4FBFC0E1108A8FD17B448A68554199C47D08FFB10D4B8,
|
|
)
|
|
|
|
|
|
def _tagged_hash(tag, msg):
|
|
t = hashlib.sha256(tag.encode()).digest()
|
|
return hashlib.sha256(t + t + msg).digest()
|
|
|
|
|
|
def _inv(x):
|
|
return pow(x, _P - 2, _P)
|
|
|
|
|
|
def _point_add(a, b):
|
|
if a is None:
|
|
return b
|
|
if b is None:
|
|
return a
|
|
if a[0] == b[0] and (a[1] != b[1]):
|
|
return None
|
|
if a == b:
|
|
lam = (3 * a[0] * a[0] * _inv(2 * a[1])) % _P
|
|
else:
|
|
lam = ((b[1] - a[1]) * _inv(b[0] - a[0])) % _P
|
|
x = (lam * lam - a[0] - b[0]) % _P
|
|
y = (lam * (a[0] - x) - a[1]) % _P
|
|
return (x, y)
|
|
|
|
|
|
def _point_mul(point, k):
|
|
r = None
|
|
while k:
|
|
if k & 1:
|
|
r = _point_add(r, point)
|
|
point = _point_add(point, point)
|
|
k >>= 1
|
|
return r
|
|
|
|
|
|
def _has_even_y(point):
|
|
return point[1] % 2 == 0
|
|
|
|
|
|
def _lift_x(x):
|
|
y_sq = (pow(x, 3, _P) + 7) % _P
|
|
y = pow(y_sq, (_P + 1) // 4, _P)
|
|
if pow(y, 2, _P) != y_sq:
|
|
return None
|
|
return (x, y if y % 2 == 0 else _P - y)
|
|
|
|
|
|
def pubkey_xonly(seckey):
|
|
d0 = int.from_bytes(seckey, "big")
|
|
p = _point_mul(_G, d0)
|
|
return p[0].to_bytes(32, "big")
|
|
|
|
|
|
def schnorr_sign(msg32, seckey, aux=b"\x00" * 32):
|
|
d0 = int.from_bytes(seckey, "big")
|
|
p = _point_mul(_G, d0)
|
|
d = d0 if _has_even_y(p) else _N - d0
|
|
t = (d ^ int.from_bytes(_tagged_hash("BIP0340/aux", aux), "big")).to_bytes(32, "big")
|
|
px = p[0].to_bytes(32, "big")
|
|
rand = _tagged_hash("BIP0340/nonce", t + px + msg32)
|
|
k0 = int.from_bytes(rand, "big") % _N
|
|
r = _point_mul(_G, k0)
|
|
k = k0 if _has_even_y(r) else _N - k0
|
|
rx = r[0].to_bytes(32, "big")
|
|
e = int.from_bytes(_tagged_hash("BIP0340/challenge", rx + px + msg32), "big") % _N
|
|
return rx + ((k + e * d) % _N).to_bytes(32, "big")
|
|
|
|
|
|
# --- Nostr event ------------------------------------------------------------
|
|
|
|
|
|
def make_signed_event(seckey, kind, content, tags=None):
|
|
tags = tags or []
|
|
pubkey = pubkey_xonly(seckey).hex()
|
|
created_at = int(time.time())
|
|
serial = json.dumps(
|
|
[0, pubkey, created_at, kind, tags, content],
|
|
separators=(",", ":"),
|
|
ensure_ascii=False,
|
|
)
|
|
eid = hashlib.sha256(serial.encode()).digest()
|
|
sig = schnorr_sign(eid, seckey)
|
|
return {
|
|
"id": eid.hex(),
|
|
"pubkey": pubkey,
|
|
"created_at": created_at,
|
|
"kind": kind,
|
|
"tags": tags,
|
|
"content": content,
|
|
"sig": sig.hex(),
|
|
}
|
|
|
|
|
|
# --- Minimal RFC-6455 client (text frames only) -----------------------------
|
|
|
|
|
|
class WS:
|
|
def __init__(self, url, insecure=False, timeout=15):
|
|
u = urlparse(url)
|
|
secure = u.scheme == "wss"
|
|
host = u.hostname
|
|
port = u.port or (443 if secure else 80)
|
|
path = u.path or "/"
|
|
raw = socket.create_connection((host, port), timeout=timeout)
|
|
if secure:
|
|
ctx = ssl.create_default_context()
|
|
if insecure:
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
raw = ctx.wrap_socket(raw, server_hostname=host)
|
|
self.sock = raw
|
|
key = base64.b64encode(os.urandom(16)).decode()
|
|
handshake = (
|
|
"GET %s HTTP/1.1\r\nHost: %s\r\nUpgrade: websocket\r\n"
|
|
"Connection: Upgrade\r\nSec-WebSocket-Key: %s\r\n"
|
|
"Sec-WebSocket-Version: 13\r\n\r\n" % (path, host, key)
|
|
)
|
|
self.sock.sendall(handshake.encode())
|
|
resp = self._read_until(b"\r\n\r\n")
|
|
if b"101" not in resp.split(b"\r\n", 1)[0]:
|
|
raise RuntimeError("websocket upgrade failed: %r" % resp[:120])
|
|
self._buf = b""
|
|
|
|
def _read_until(self, marker):
|
|
data = b""
|
|
while marker not in data:
|
|
chunk = self.sock.recv(4096)
|
|
if not chunk:
|
|
break
|
|
data += chunk
|
|
return data
|
|
|
|
def send_text(self, text):
|
|
payload = text.encode()
|
|
header = bytearray([0x81]) # FIN + text opcode
|
|
mask = os.urandom(4)
|
|
n = len(payload)
|
|
if n < 126:
|
|
header.append(0x80 | n)
|
|
elif n < 65536:
|
|
header.append(0x80 | 126)
|
|
header += n.to_bytes(2, "big")
|
|
else:
|
|
header.append(0x80 | 127)
|
|
header += n.to_bytes(8, "big")
|
|
header += mask
|
|
masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
|
|
self.sock.sendall(bytes(header) + masked)
|
|
|
|
def _recv_exact(self, n):
|
|
while len(self._buf) < n:
|
|
chunk = self.sock.recv(4096)
|
|
if not chunk:
|
|
raise RuntimeError("connection closed")
|
|
self._buf += chunk
|
|
out, self._buf = self._buf[:n], self._buf[n:]
|
|
return out
|
|
|
|
def recv_text(self):
|
|
b0, b1 = self._recv_exact(2)
|
|
opcode = b0 & 0x0F
|
|
length = b1 & 0x7F
|
|
if length == 126:
|
|
length = int.from_bytes(self._recv_exact(2), "big")
|
|
elif length == 127:
|
|
length = int.from_bytes(self._recv_exact(8), "big")
|
|
payload = self._recv_exact(length) if length else b""
|
|
if opcode == 0x8: # close
|
|
raise RuntimeError("server closed the connection")
|
|
if opcode in (0x9, 0xA): # ping/pong: ignore, read next
|
|
return self.recv_text()
|
|
return payload.decode("utf-8", "replace")
|
|
|
|
def close(self):
|
|
try:
|
|
self.sock.close()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def publish_expect(ws, event, want_accept, timeout=15):
|
|
"""Send one event, wait for its OK frame, return (accepted, message)."""
|
|
ws.send_text(json.dumps(["EVENT", event]))
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
msg = json.loads(ws.recv_text())
|
|
if msg[0] == "OK" and msg[1] == event["id"]:
|
|
return bool(msg[2]), (msg[3] if len(msg) > 3 else "")
|
|
raise RuntimeError("no OK response for event %s" % event["id"])
|
|
|
|
|
|
def main():
|
|
args = [a for a in sys.argv[1:] if not a.startswith("--")]
|
|
insecure = "--insecure" in sys.argv
|
|
if not args:
|
|
sys.stderr.write("usage: smoke_test_lockdown.py wss://relay [--insecure]\n")
|
|
return 2
|
|
url = args[0]
|
|
seckey = os.urandom(32)
|
|
|
|
# A gift-wrap-shaped event: gift wraps are signed by throwaway keys
|
|
# anyway (NIP-59), p-tag the recipient, content is an opaque ciphertext
|
|
# blob (here random base64, ephemeral-ish; nobody will ever unwrap it).
|
|
recipient = pubkey_xonly(os.urandom(32)).hex()
|
|
wrap_blob = base64.b64encode(os.urandom(192)).decode()
|
|
|
|
checks = [
|
|
# (label, event, want_accept, failure explanation)
|
|
(
|
|
"kind 1 note",
|
|
make_signed_event(seckey, 1, "floonet lockdown smoke test; please ignore"),
|
|
False,
|
|
"an unauthorized author's text note was accepted",
|
|
),
|
|
(
|
|
"kind 0 profile",
|
|
make_signed_event(seckey, 0, json.dumps({"name": "floonet-smoke"})),
|
|
True,
|
|
"a profile was blocked; kind 0 must stay open",
|
|
),
|
|
(
|
|
"kind 1059 gift wrap",
|
|
make_signed_event(seckey, 1059, wrap_blob, tags=[["p", recipient]]),
|
|
True,
|
|
"a gift wrap was blocked; the money path must stay ungated",
|
|
),
|
|
(
|
|
"kind 30023 long-form",
|
|
make_signed_event(
|
|
seckey,
|
|
30023,
|
|
"floonet lockdown smoke test article; please ignore",
|
|
tags=[["d", "floonet-smoke-test"], ["title", "smoke test"]],
|
|
),
|
|
False,
|
|
"an unauthorized author's long-form article was accepted",
|
|
),
|
|
]
|
|
|
|
print("relay: %s" % url)
|
|
print("throwaway: %s (unauthorized by design)" % pubkey_xonly(seckey).hex())
|
|
|
|
ws = WS(url, insecure=insecure)
|
|
ok = True
|
|
try:
|
|
for label, event, want_accept, fail_msg in checks:
|
|
accepted, why = publish_expect(ws, event, want_accept=want_accept)
|
|
good = accepted == want_accept
|
|
ok = ok and good
|
|
print("%-22s %s (expected %s) msg=%r"
|
|
% (label + ":",
|
|
"ACCEPTED" if accepted else "BLOCKED",
|
|
"ACCEPTED" if want_accept else "BLOCKED",
|
|
why))
|
|
if not good:
|
|
print(" FAIL: %s" % fail_msg)
|
|
finally:
|
|
ws.close()
|
|
|
|
print("RESULT: %s" % ("PASS" if ok else "FAIL"))
|
|
return 0 if ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|