2ro
65ec8fcef2
Public-note kinds are now accepted only from an operator-configured list of
author pubkeys (FLOONET_AUTHORIZED_AUTHORS, hex or npub). Closed by default:
with no authors set, kinds 1 and 30023 are rejected for everyone, so random
notes cannot be spammed to the relay. Every other kind is unaffected and
kind 0 profiles stay open.
- add 30023 (long-form article) to the default kind whitelist
- check_authorized_authors runs right after check_kind; LOCKED_KINDS {1,30023}
- pure-python bech32 npub decoder (no new deps); invalid entries logged/skipped
- load_config also reads a KEY=VALUE floonet.env next to the plugin
(FLOONET_ENV_FILE), env vars win, so config changes need no container
recreate; strfry reloads the plugin on mtime change (no restart)
- extend test_policy.py (32 tests green)
- scripts/purge_public_notes.sh (dry-run default) to clean pre-existing notes
- scripts/smoke_test_lockdown.py post-deploy check
- README + .env.example config docs
269 lines
8.3 KiB
Python
269 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Post-deploy smoke test for the public-note lockdown.
|
|
|
|
Signs two 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)
|
|
|
|
Exit 0 only if both 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)
|
|
note = make_signed_event(seckey, 1, "floonet lockdown smoke test; please ignore")
|
|
profile = make_signed_event(seckey, 0, json.dumps({"name": "floonet-smoke"}))
|
|
|
|
print("relay: %s" % url)
|
|
print("throwaway: %s (unauthorized by design)" % pubkey_xonly(seckey).hex())
|
|
|
|
ws = WS(url, insecure=insecure)
|
|
ok = True
|
|
try:
|
|
accepted, why = publish_expect(ws, note, want_accept=False)
|
|
good = not accepted
|
|
ok = ok and good
|
|
print("kind 1 note: %s (expected BLOCKED) msg=%r"
|
|
% ("BLOCKED" if not accepted else "ACCEPTED", why))
|
|
if accepted:
|
|
print(" FAIL: an unauthorized author's text note was accepted")
|
|
|
|
accepted, why = publish_expect(ws, profile, want_accept=True)
|
|
ok = ok and accepted
|
|
print("kind 0 profile: %s (expected ACCEPTED) msg=%r"
|
|
% ("ACCEPTED" if accepted else "BLOCKED", why))
|
|
if not accepted:
|
|
print(" FAIL: a profile was blocked; kind 0 must stay open")
|
|
finally:
|
|
ws.close()
|
|
|
|
print("RESULT: %s" % ("PASS" if ok else "FAIL"))
|
|
return 0 if ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|