|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Post messages to a Buzz relay as a dedicated bot identity. |
| 3 | +
|
| 4 | +Pure stdlib: implements BIP-340 Schnorr signing, Nostr event IDs, and NIP-98 |
| 5 | +HTTP auth, so it runs on a bare GitHub Actions runner with no pip install. |
| 6 | +
|
| 7 | +Usage: |
| 8 | + BUZZ_BOT_NSEC=nsec1... python3 buzz_bot.py post --channel <uuid> --file body.md |
| 9 | + BUZZ_BOT_NSEC=nsec1... python3 buzz_bot.py join --channel <uuid> |
| 10 | + BUZZ_BOT_NSEC=nsec1... python3 buzz_bot.py profile --name "Upstream Bot" --about "..." |
| 11 | + python3 buzz_bot.py selftest |
| 12 | +""" |
| 13 | + |
| 14 | +import base64 |
| 15 | +import hashlib |
| 16 | +import json |
| 17 | +import os |
| 18 | +import sys |
| 19 | +import time |
| 20 | +import urllib.request |
| 21 | + |
| 22 | +RELAY = os.environ.get("BUZZ_RELAY_HTTP", "https://chat.duvalsoftware.com") |
| 23 | + |
| 24 | +# --- secp256k1 --------------------------------------------------------------- |
| 25 | +P = 2**256 - 2**32 - 977 |
| 26 | +N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141 |
| 27 | +G = ( |
| 28 | + 0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798, |
| 29 | + 0x483ADA7726A3C4655DA4FBFC0E1108A8FD17B448A68554199C47D08FFB10D4B8, |
| 30 | +) |
| 31 | + |
| 32 | + |
| 33 | +def point_add(a, b): |
| 34 | + if a is None: |
| 35 | + return b |
| 36 | + if b is None: |
| 37 | + return a |
| 38 | + if a[0] == b[0] and (a[1] + b[1]) % P == 0: |
| 39 | + return None |
| 40 | + if a == b: |
| 41 | + lam = 3 * a[0] * a[0] * pow(2 * a[1], P - 2, P) % P |
| 42 | + else: |
| 43 | + lam = (b[1] - a[1]) * pow(b[0] - a[0], P - 2, P) % P |
| 44 | + x = (lam * lam - a[0] - b[0]) % P |
| 45 | + return (x, (lam * (a[0] - x) - a[1]) % P) |
| 46 | + |
| 47 | + |
| 48 | +def point_mul(k, point=None): |
| 49 | + point = G if point is None else point |
| 50 | + result = None |
| 51 | + while k: |
| 52 | + if k & 1: |
| 53 | + result = point_add(result, point) |
| 54 | + point = point_add(point, point) |
| 55 | + k >>= 1 |
| 56 | + return result |
| 57 | + |
| 58 | + |
| 59 | +def lift_x(x): |
| 60 | + """Recover the even-y point for an x-only pubkey.""" |
| 61 | + if x >= P: |
| 62 | + return None |
| 63 | + y_sq = (pow(x, 3, P) + 7) % P |
| 64 | + y = pow(y_sq, (P + 1) // 4, P) |
| 65 | + if pow(y, 2, P) != y_sq: |
| 66 | + return None |
| 67 | + return (x, y if y % 2 == 0 else P - y) |
| 68 | + |
| 69 | + |
| 70 | +def tagged_hash(tag, msg): |
| 71 | + t = hashlib.sha256(tag.encode()).digest() |
| 72 | + return hashlib.sha256(t + t + msg).digest() |
| 73 | + |
| 74 | + |
| 75 | +def schnorr_sign(msg32, sk_bytes, aux=b"\x00" * 32): |
| 76 | + d0 = int.from_bytes(sk_bytes, "big") |
| 77 | + if not (1 <= d0 < N): |
| 78 | + raise ValueError("private key out of range") |
| 79 | + point = point_mul(d0) |
| 80 | + d = d0 if point[1] % 2 == 0 else N - d0 |
| 81 | + px = point[0].to_bytes(32, "big") |
| 82 | + |
| 83 | + t = (d ^ int.from_bytes(tagged_hash("BIP0340/aux", aux), "big")).to_bytes(32, "big") |
| 84 | + k0 = int.from_bytes(tagged_hash("BIP0340/nonce", t + px + msg32), "big") % N |
| 85 | + if k0 == 0: |
| 86 | + raise ValueError("nonce is zero") |
| 87 | + r_point = point_mul(k0) |
| 88 | + k = k0 if r_point[1] % 2 == 0 else N - k0 |
| 89 | + rx = r_point[0].to_bytes(32, "big") |
| 90 | + |
| 91 | + e = int.from_bytes(tagged_hash("BIP0340/challenge", rx + px + msg32), "big") % N |
| 92 | + return rx + ((k + e * d) % N).to_bytes(32, "big") |
| 93 | + |
| 94 | + |
| 95 | +def schnorr_verify(msg32, pubkey32, sig): |
| 96 | + point = lift_x(int.from_bytes(pubkey32, "big")) |
| 97 | + if point is None or len(sig) != 64: |
| 98 | + return False |
| 99 | + r = int.from_bytes(sig[:32], "big") |
| 100 | + s = int.from_bytes(sig[32:], "big") |
| 101 | + if r >= P or s >= N: |
| 102 | + return False |
| 103 | + e = int.from_bytes(tagged_hash("BIP0340/challenge", sig[:32] + pubkey32 + msg32), "big") % N |
| 104 | + big_r = point_add(point_mul(s), point_mul(N - e, point)) |
| 105 | + return big_r is not None and big_r[1] % 2 == 0 and big_r[0] == r |
| 106 | + |
| 107 | + |
| 108 | +# --- bech32 ------------------------------------------------------------------ |
| 109 | +CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l" |
| 110 | + |
| 111 | + |
| 112 | +def _polymod(values): |
| 113 | + gen = [0x3B6A57B2, 0x26508E6D, 0x1EA119FA, 0x3D4233DD, 0x2A1462B3] |
| 114 | + chk = 1 |
| 115 | + for v in values: |
| 116 | + b = chk >> 25 |
| 117 | + chk = (chk & 0x1FFFFFF) << 5 ^ v |
| 118 | + for i in range(5): |
| 119 | + chk ^= gen[i] if ((b >> i) & 1) else 0 |
| 120 | + return chk |
| 121 | + |
| 122 | + |
| 123 | +def bech32_decode(s): |
| 124 | + hrp, data = s.rsplit("1", 1) |
| 125 | + vals = [CHARSET.index(c) for c in data][:-6] |
| 126 | + acc = bits = 0 |
| 127 | + out = bytearray() |
| 128 | + for v in vals: |
| 129 | + acc = (acc << 5) | v |
| 130 | + bits += 5 |
| 131 | + while bits >= 8: |
| 132 | + bits -= 8 |
| 133 | + out.append((acc >> bits) & 0xFF) |
| 134 | + return hrp, bytes(out) |
| 135 | + |
| 136 | + |
| 137 | +def bech32_encode(hrp, payload): |
| 138 | + acc = bits = 0 |
| 139 | + data = [] |
| 140 | + for b in payload: |
| 141 | + acc = (acc << 8) | b |
| 142 | + bits += 8 |
| 143 | + while bits >= 5: |
| 144 | + bits -= 5 |
| 145 | + data.append((acc >> bits) & 31) |
| 146 | + if bits: |
| 147 | + data.append((acc << (5 - bits)) & 31) |
| 148 | + vals = [ord(c) >> 5 for c in hrp] + [0] + [ord(c) & 31 for c in hrp] + data |
| 149 | + pm = _polymod(vals + [0] * 6) ^ 1 |
| 150 | + checksum = [(pm >> 5 * (5 - i)) & 31 for i in range(6)] |
| 151 | + return hrp + "1" + "".join(CHARSET[d] for d in data + checksum) |
| 152 | + |
| 153 | + |
| 154 | +# --- nostr ------------------------------------------------------------------- |
| 155 | +def load_key(): |
| 156 | + raw = os.environ.get("BUZZ_BOT_NSEC", "").strip() |
| 157 | + if not raw: |
| 158 | + sys.exit("BUZZ_BOT_NSEC is not set") |
| 159 | + if raw.startswith("nsec1"): |
| 160 | + _, sk = bech32_decode(raw) |
| 161 | + else: |
| 162 | + sk = bytes.fromhex(raw) |
| 163 | + pub = point_mul(int.from_bytes(sk, "big"))[0].to_bytes(32, "big") |
| 164 | + return sk, pub |
| 165 | + |
| 166 | + |
| 167 | +def build_event(sk, pub, kind, tags, content, created_at=None): |
| 168 | + created_at = created_at or int(time.time()) |
| 169 | + serialized = json.dumps( |
| 170 | + [0, pub.hex(), created_at, kind, tags, content], |
| 171 | + separators=(",", ":"), |
| 172 | + ensure_ascii=False, |
| 173 | + ) |
| 174 | + eid = hashlib.sha256(serialized.encode()).digest() |
| 175 | + return { |
| 176 | + "id": eid.hex(), |
| 177 | + "pubkey": pub.hex(), |
| 178 | + "created_at": created_at, |
| 179 | + "kind": kind, |
| 180 | + "tags": tags, |
| 181 | + "content": content, |
| 182 | + "sig": schnorr_sign(eid, sk).hex(), |
| 183 | + } |
| 184 | + |
| 185 | + |
| 186 | +def nip98_header(sk, pub, url, method, body_bytes): |
| 187 | + """NIP-98: a signed kind-27235 event proving this exact request.""" |
| 188 | + tags = [ |
| 189 | + ["u", url], |
| 190 | + ["method", method], |
| 191 | + ["payload", hashlib.sha256(body_bytes).hexdigest()], |
| 192 | + ] |
| 193 | + ev = build_event(sk, pub, 27235, tags, "") |
| 194 | + token = base64.b64encode(json.dumps(ev).encode()).decode() |
| 195 | + return f"Nostr {token}" |
| 196 | + |
| 197 | + |
| 198 | +def post_event(sk, pub, event): |
| 199 | + url = f"{RELAY}/events" |
| 200 | + body = json.dumps(event).encode() |
| 201 | + req = urllib.request.Request( |
| 202 | + url, |
| 203 | + data=body, |
| 204 | + method="POST", |
| 205 | + headers={ |
| 206 | + "Content-Type": "application/json", |
| 207 | + "Authorization": nip98_header(sk, pub, url, "POST", body), |
| 208 | + }, |
| 209 | + ) |
| 210 | + try: |
| 211 | + with urllib.request.urlopen(req, timeout=30) as resp: |
| 212 | + return resp.status, resp.read().decode() |
| 213 | + except urllib.error.HTTPError as e: |
| 214 | + return e.code, e.read().decode() |
| 215 | + |
| 216 | + |
| 217 | +# --- commands ---------------------------------------------------------------- |
| 218 | +def cmd_selftest(): |
| 219 | + sk = hashlib.sha256(b"buzz-bot-selftest").digest() |
| 220 | + pub = point_mul(int.from_bytes(sk, "big"))[0].to_bytes(32, "big") |
| 221 | + msg = hashlib.sha256(b"hello buzz").digest() |
| 222 | + sig = schnorr_sign(msg, sk) |
| 223 | + assert schnorr_verify(msg, pub, sig), "signature failed to verify" |
| 224 | + assert not schnorr_verify(hashlib.sha256(b"tampered").digest(), pub, sig), "verified a bad message" |
| 225 | + ev = build_event(sk, pub, 9, [["h", "test"]], "hi") |
| 226 | + assert schnorr_verify(bytes.fromhex(ev["id"]), pub, bytes.fromhex(ev["sig"])), "event sig invalid" |
| 227 | + # round-trip bech32 |
| 228 | + assert bech32_decode(bech32_encode("nsec", sk))[1] == sk, "bech32 round-trip failed" |
| 229 | + print("selftest OK — schnorr sign/verify, event id/sig, bech32 all consistent") |
| 230 | + |
| 231 | + |
| 232 | +def main(): |
| 233 | + cmd = sys.argv[1] if len(sys.argv) > 1 else "selftest" |
| 234 | + if cmd == "selftest": |
| 235 | + return cmd_selftest() |
| 236 | + |
| 237 | + args = dict(zip(sys.argv[2::2], sys.argv[3::2])) |
| 238 | + sk, pub = load_key() |
| 239 | + |
| 240 | + if cmd == "join": |
| 241 | + ev = build_event(sk, pub, 9021, [["h", args["--channel"]]], "") |
| 242 | + elif cmd == "profile": |
| 243 | + meta = {"display_name": args.get("--name", "Upstream Bot"), |
| 244 | + "about": args.get("--about", "")} |
| 245 | + ev = build_event(sk, pub, 0, [], json.dumps(meta)) |
| 246 | + elif cmd == "post": |
| 247 | + content = open(args["--file"], encoding="utf-8").read() if "--file" in args else args["--text"] |
| 248 | + ev = build_event(sk, pub, 9, [["h", args["--channel"]]], content) |
| 249 | + else: |
| 250 | + sys.exit(f"unknown command: {cmd}") |
| 251 | + |
| 252 | + status, resp = post_event(sk, pub, ev) |
| 253 | + print(f"{cmd}: HTTP {status} {resp[:400]}") |
| 254 | + sys.exit(0 if status < 300 else 1) |
| 255 | + |
| 256 | + |
| 257 | +if __name__ == "__main__": |
| 258 | + main() |
0 commit comments