udp_nostr/src/udp_nostr_recv.py

#!/usr/bin/env python3
"""UDP Nostr Receiver — listens for Nostr events as single UDP datagrams.

Usage:
  # Plaintext mode (unchanged):
  python3 udp_nostr_recv.py [bind_host] [port]

  # Encrypted mode (also accepts plaintext on the same port):
  python3 udp_nostr_recv.py [bind_host] [port] --relay-secret <relay_sec_hex>

  # The relay secret may also be supplied via the UDP_NOSTR_RELAY_SECRET
  # environment variable (the flag takes precedence).

Both plaintext and encrypted events are accepted on the same port. Detection:

  1. If the datagram starts with '{', attempt strict plaintext JSON/event
     validation (JSON object with the required Nostr event fields).
  2. If that fails (or the datagram does not start with '{'), fall back to
     encrypted framing detection: length >= 131, bytes 0..31 a valid x-only
     secp256k1 pubkey, byte 32 == NIP-44 version 0x02. This handles the
     leading-'{' ephemeral-pubkey collision.

Encrypted datagrams are decrypted with the relay secret (NIP-44 v2, wire
format: ephemeral_xonly_pubkey[32] || native payload). Any framing, key,
version, padding, or MAC failure drops the datagram silently. Without
--relay-secret, encrypted datagrams cannot be processed and are dropped.
"""
import argparse
import json
import os
import socket
import sys

try:
    import nip44
except ImportError:  # running from another directory
    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    import nip44

MAX_SINGLE_PACKET = 1472  # Ethernet MTU budget for one unfragmented datagram
EVENT_REQUIRED_FIELDS = ("id", "pubkey", "created_at", "kind", "tags", "content", "sig")


def parse_args(argv):
    parser = argparse.ArgumentParser(
        description="Receive Nostr events as single UDP datagrams (plaintext and/or encrypted)."
    )
    parser.add_argument("bind_host", nargs="?", default="0.0.0.0", help="bind host (default: 0.0.0.0)")
    parser.add_argument("port", nargs="?", type=int, default=8888, help="bind port (default: 8888)")
    parser.add_argument("--relay-secret", default=None,
                        help="relay secret key (64 hex chars) enabling NIP-44 v2 decryption; "
                             "also readable from UDP_NOSTR_RELAY_SECRET")
    return parser.parse_args(argv)


def parse_plaintext_event(data: bytes):
    """Strictly validate a plaintext Nostr event datagram.

    Returns the parsed event dict, or None if data is not a valid plaintext
    event (bad JSON, not an object, or missing required fields).
    """
    try:
        event = json.loads(data.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None
    if not isinstance(event, dict):
        return None
    if any(field not in event for field in EVENT_REQUIRED_FIELDS):
        return None
    return event


def _log(message: str):
    print(message, flush=True)


def handle_datagram(data: bytes, addr, relay_secret=None, log=_log):
    """Process one datagram; returns 'plaintext', 'encrypted', or None (dropped)."""
    if len(data) > MAX_SINGLE_PACKET:
        log(f"⚠ {len(data)} bytes from {addr[0]}:{addr[1]} exceeds the "
            f"{MAX_SINGLE_PACKET}-byte single-datagram budget (fragmented on path)")

    # Stage 1: strict plaintext validation (only when it plausibly is JSON).
    if data[:1] == b"{":
        event = parse_plaintext_event(data)
        if event is not None:
            log(f"{len(data)} bytes from {addr[0]}:{addr[1]} (plaintext)")
            log(json.dumps(event, separators=(",", ":")))
            return "plaintext"
        # Fall through: a leading '{' can also be the first byte of an
        # ephemeral pubkey — try encrypted framing below.

    # Stage 2: encrypted framing detection and decryption.
    if nip44.looks_encrypted(data):
        if relay_secret is None:
            return None  # encrypted traffic not enabled; drop silently
        try:
            plaintext = nip44.decrypt(data, relay_secret)
        except nip44.NIP44Error:
            return None  # wrong key, tampering, or malformed framing; drop silently
        event = parse_plaintext_event(plaintext)
        if event is None:
            return None  # decrypted bytes are not a valid event; drop silently
        log(f"{len(data)} bytes from {addr[0]}:{addr[1]} (encrypted, "
            f"{len(plaintext)}-byte inner event)")
        log(json.dumps(event, separators=(",", ":")))
        return "encrypted"

    return None  # neither valid plaintext nor plausible encrypted framing


def main(argv=None):
    args = parse_args(argv)

    relay_secret = args.relay_secret or os.environ.get("UDP_NOSTR_RELAY_SECRET")
    if relay_secret:
        try:
            relay_secret = nip44.validate_secret(relay_secret)
        except nip44.NIP44Error as e:
            print(f"error: invalid --relay-secret: {e}", file=sys.stderr)
            return 2

    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind((args.bind_host, args.port))
    mode = "plaintext + encrypted" if relay_secret else "plaintext only"
    print(f"Listening for Nostr events on UDP {args.bind_host}:{args.port} ({mode})",
          flush=True)

    try:
        while True:
            data, addr = sock.recvfrom(65535)
            handle_datagram(data, addr, relay_secret)
    except KeyboardInterrupt:
        return 0


if __name__ == "__main__":
    sys.exit(main())