import json from threading import Lock from websocket import WebSocketApp from .event import Event from .filter import Filters from .message_pool import MessagePool from .message_type import RelayMessageType from .subscription import Subscription class RelayPolicy: def __init__(self, should_read: bool=True, should_write: bool=True) -> None: self.should_read = should_read self.should_write = should_write def to_json_object(self) -> dict[str, bool]: return { "read": self.should_read, "write": self.should_write } class Relay: def __init__( self, url: str, policy: RelayPolicy, message_pool: MessagePool, subscriptions: dict[str, Subscription]={}) -> None: self.url = url self.policy = policy self.message_pool = message_pool self.subscriptions = subscriptions self.lock = Lock() self.ws = WebSocketApp( url, on_open=self._on_open, on_message=self._on_message, on_error=self._on_error, on_close=self._on_close) def connect(self, ssl_options: dict=None, proxy: dict=None): self.ws.run_forever( sslopt=ssl_options, http_proxy_host=None if proxy is None else proxy.get('host'), http_proxy_port=None if proxy is None else proxy.get('port') ) def close(self): self.ws.close() def publish(self, message: str): self.ws.send(message) def add_subscription(self, id, filters: Filters): with self.lock: self.subscriptions[id] = Subscription(id, filters) def close_subscription(self, id: str) -> None: with self.lock: self.subscriptions.pop(id) def update_subscription(self, id: str, filters: Filters) -> None: with self.lock: subscription = self.subscriptions[id] subscription.filters = filters def to_json_object(self) -> dict: return { "url": self.url, "policy": self.policy.to_json_object(), "subscriptions": [subscription.to_json_object() for subscription in self.subscriptions.values()] } def _on_open(self, class_obj): pass def _on_close(self, class_obj, status_code, message): pass def _on_message(self, class_obj, message: str): if self._is_valid_message(message): self.message_pool.add_message(message, self.url) def _on_error(self, class_obj, error): pass def _is_valid_message(self, message: str) -> bool: message = message.strip("\n") if not message or message[0] != '[' or message[-1] != ']': return False message_json = json.loads(message) message_type = message_json[0] if not RelayMessageType.is_valid(message_type): return False if message_type == RelayMessageType.EVENT: if not len(message_json) == 3: return False subscription_id = message_json[1] with self.lock: if subscription_id not in self.subscriptions: return False e = message_json[2] event = Event(e['pubkey'], e['content'], e['created_at'], e['kind'], e['tags'], e['id'], e['sig']) if not event.verify(): return False with self.lock: subscription = self.subscriptions[subscription_id] if not subscription.filters.match(event): return False return True