fonstr.py

fonstr.py

import json
import argparse
from typing import List, Dict, Optional
from fastapi import FastAPI, WebSocket
from pydantic import BaseModel
from nostr import tools
import ssl
import uvicorn

parser = argparse.ArgumentParser()
parser.add_argument("-h", "--https", action="store_true", help="Use HTTPS")
parser.add_argument("port", type=int, nargs="?", default=4444, help="Port number")
args = parser.parse_args()

app = FastAPI()
events = []
subscribers = {}


class Filter(BaseModel):
    types: Optional[List[str]] = None
    kind: Optional[str] = None
    from_: Optional[str] = None
    to: Optional[str] = None
    subscription_id: str


def event_passes_filter(event, filter_: Filter) -> bool:
    if filter_.types and event["type"] not in filter_.types:
        return False

    if filter_.kind and event["kind"] != filter_.kind:
        return False

    if filter_.from_ and event["from"] != filter_.from_:
        return False

    if filter_.to and event["to"] != filter_.to:
        return False

    return True


@app.websocket("/")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    print("ws connection started")
    print("ws connection established")

    while True:
        message = await websocket.receive_text()
        print("received message", message)
        data = json.loads(message)
        action_type, value, *rest = data

        if action_type == "EVENT":
            ok = tools.validate_event(value)
            very_ok = tools.verify_signature(value)
            print("ok", ok, very_ok)

            if ok and very_ok:
                events.append(value)
                print("event ok")

                for filters in subscribers.values():
                    for filter_ in filters:
                        if event_passes_filter(value, filter_):
                            await websocket.send_json(["EVENT", filter_.subscription_id, value])

                await websocket.send_json(["OK", value["id"], True, ""])
            else:
                await websocket.send_json(["NOTICE", "Invalid event"])

        elif action_type == "REQ":
            print("REQ")
            subscription_id = value
            filters = [Filter(**filter_, subscription_id=subscription_id) for filter_ in rest]
            subscribers[websocket] = filters

            for filter_ in filters:
                filtered_events = [event for event in events if event_passes_filter(event, filter_)]
                for event in filtered_events:
                    await websocket.send_json(["EVENT", filter_.subscription_id, event])

            await websocket.send_json(["EOSE", subscription_id])

        elif action_type == "CLOSE":
            sub_id = value
            if websocket in subscribers:
                updated_filters = [filter_ for filter_ in subscribers[websocket] if filter_.subscription_id != sub_id]
                if not updated_filters:
                    del subscribers[websocket]
                else:
                    subscribers[websocket] = updated_filters

        else:
            await websocket.send_json(["NOTICE", "Unrecognized event"])
            print("Unrecognized event")

if args.https:
    ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    ssl_context.load_cert_chain(certfile="fullchain.pem", keyfile="privkey.pem")
    uvicorn.run(app, host="0.0.0.0", port=args.port, ssl_context=ssl_context)
else:
    uvicorn.run(app, host="0.0.0.0", port=args.port)

Hi Melvin , where did this code come from?

just made it up, its a port from https://fonstr.com to python, but i have no idea if it works

ah that makes sense. I was scratching my head about the nostr library having a tools module