squash bot api

Author: majestrate

sogs/events.py

@@ -0,0 +1,149 @@
+from collections import defaultdict
+
+from oxenmq import AuthLevel
+
+from . import config
+from . import model
+from .omq import omq
+
+from binascii import hexlify
+
+from oxenc import bt_serialize as serialize
+from oxenc import bt_deserialize as deserialize
+
+from requests import request as http_request
+
+# pools for event propagation
+_pools = defaultdict(list)
+
+status_OK = 'OK'
+status_ERR = 'ERROR'
+
+# the events we are able to subscribe to
+EVENTS = ('message', 'joined', 'parted', 'banned', 'unbanned', 'deleted', 'uploaded')
+
+
+def event_name_valid(eventname):
+    """ return True if this event name is something well formed """
+    return eventname in EVENTS
+
+
+def _user_from_conn(conn, automod=True):
+    """
+    make a model.User from a connection using it's curve pubkey as the session id, if automod is
+    True we auto promote the bot to a moderator if it isn't already
+    """
+    user = model.User(session_id='05' + hexlify(conn.pubkey).decode())
+    if automod and not user.global_moderator:
+        user.set_moderator()
+    return user
+
+
+def _sub_conn_to_event(name, conn):
+    """ subscribe a connection to an event type """
+    if not event_name_valid(name):
+        raise Exception('invalid event name: {}'.format(name))
+    global _pools
+    _pools[name].append(conn)
+
+
+def _unsub_conn_from_event(name, conn):
+    if not event_name_valid(name):
+        raise Exception(f"invalid event: {name}")
+    global _pool
+    _pools[name].remove(conn)
+
+
+def _handle_subscribe_request(msg):
+    """ handle a request to subscribe to events as a bot"""
+    parts = msg.data()
+    if len(parts) == 0:
+        raise Exception("no events selected")
+    for name in parts:
+        _sub_conn_to_event(name.decode(), msg.conn)
+
+
+def _handle_unsubscribe_request(msg):
+    """ unusb from events """
+    parts = msg.data()
+    if len(parts) == 0:
+        raise Exception("no events selected")
+    for name in parts:
+        _unsub_conn_from_event(name.decode(), msg.conn)
+
+
+def _propagate_event(eventname, *args):
+    """ propagate an event to everyone who cares about it """
+    assert event_name_valid(eventname)
+    global omq, _pools
+    ev = (eventname,) + args
+    for conn in _pools[eventname]:
+        omq.send(conn, 'sogs.event', *ev)
+
+
+def _handle_request(handler, msg):
+    """safely handle a request handler and catch any exceptions that happen and propagate them to
+    the remote connection"""
+    try:
+        retval = handler(msg)
+        if retval is None:
+            msg.reply(status_OK)
+        elif isinstance(retval, tuple):
+            msg.reply(status_OK, *retval)
+        else:
+            msg.reply(status_OK, serialize(retval))
+    except Exception as ex:
+        msg.reply(status_ERR, '{}'.format(ex))
+
+
+def _handle_api_request(msg):
+    """ do an http api request """
+    parts = msg.parts()
+    if len(parts) < 3:
+        raise Exception("invalid number of parts {}".format(len(parts)))
+    kwargs = dict()
+    if len(parts) >= 4:
+        kwargs['body'] = parts[3]
+
+    headers = deserialize(parts[2])
+    if not isinstance(headers, dict):
+        raise ValueError("headers is not a dict")
+
+    kwargs['headers'] = headers
+
+    path = parts[1].decode()
+    if not path.startswith("/"):
+        raise ValueError("invalid request path")
+    method = parts[0].decode()
+    resp = http_request(method, config.URL_BASE + path, **kwargs)
+
+    headers = dict()
+    for k, v in resp.headers.items():
+        headers[k] = v
+
+    return serialize(resp.status_code), serialize(headers), resp.content
+
+
+def setup_rpc():
+    """ set up bot api using an existing oxenmq instance """
+    _bot_category = omq.add_category('sogs', AuthLevel.none)
+    _bot_category.add_request_command(
+        'sub', lambda msg: _handle_request(_handle_subscribe_request, msg)
+    )
+    _bot_category.add_request_command(
+        'unsub', lambda msg: _handle_request(_handle_unsubscribe_request, msg)
+    )
+    _bot_category.add_request_command(
+        'request', lambda msg: _handle_request(_handle_api_request, msg)
+    )
+
+
+class _Notify:
+    """ Holder type for all event notification functions """
+
+
+notify = _Notify()
+
+# set up event notifiers
+for ev in EVENTS:
+    setattr(notify, ev, lambda *args: _propagate_event(ev, *args))

sogs/mule.py

@@ -1,6 +1,5 @@
 import traceback
 import oxenmq
-from oxenc import bt_deserialize
 import time
 from datetime import timedelta
 import functools
@@ -9,6 +8,7 @@
 from . import cleanup
 from . import config
 from . import omq as o
+from .events import setup_rpc, notify
 
 # This is the uwsgi "mule" that handles things not related to serving HTTP requests:
 # - it holds the oxenmq instance (with its own interface into sogs)
@@ -52,6 +52,10 @@ def setup_omq():
     for addr in listen:
         omq.listen(addr, curve=True, allow_connection=allow_conn)
         app.logger.info(f"OxenMQ listening on {addr}")
+    if not listen:
+        app.logger.warn(
+            "OxenMQ did not listen on any curve addresses which means the bot API is not accessable anywhere."
+        )
 
     # Internal socket for workers to talk to us:
     omq.listen(config.OMQ_INTERNAL, curve=False, allow_connection=admin_conn)
@@ -64,6 +68,13 @@ def setup_omq():
     worker.add_command("message_posted", message_posted)
     worker.add_command("messages_deleted", messages_deleted)
     worker.add_command("message_edited", message_edited)
+    worker.add_command("user_joined", user_joined)
+    worker.add_command("user_banned", user_banned)
+    worker.add_command("user_unbanned", user_unbanned)
+    worker.add_command("file_uploaded", file_uploaded)
+
+    # setup bot api endpoints
+    setup_rpc()
 
     app.logger.debug("Mule starting omq")
     omq.start()
@@ -88,14 +99,32 @@ def wrapper(*args, **kwargs):
 
 @log_exceptions
 def message_posted(m: oxenmq.Message):
-    id = bt_deserialize(m.data()[0])
-    app.logger.warning(f"FIXME: mule -- message posted stub, id={id}")
+    notify.message(*m.data())
 
 
 @log_exceptions
 def messages_deleted(m: oxenmq.Message):
-    ids = bt_deserialize(m.data()[0])
-    app.logger.warning(f"FIXME: mule -- message delete stub, deleted messages: {ids}")
+    notify.deleted(*m.data())
+
+
+@log_exceptions
+def user_banned(m: oxenmq.Message):
+    notify.banned(*m.data())
+
+
+@log_exceptions
+def user_unbanned(m: oxenmq.Message):
+    notify.unbannd(*m.data())
+
+
+@log_exceptions
+def user_joined(m: oxenmq.Message):
+    notify.joined(*m.data())
+
+
+@log_exceptions
+def file_uploaded(m: oxenmq.Message):
+    notify.uploaded(*m.data())
 
 
 @log_exceptions