Merge pull request #17 from jagerman/oxenmq-mule

Add oxenmq as a uwsgi mule

Author: majestrate

contrib/uwsgi-vassal-sogs.ini

@@ -7,5 +7,8 @@ processes = 4
 enable-threads = true
 manage-script-name = true
 mount = /=sogs.web:app
+mule = sogs.mule:run
 
-logger = file:logfile=/home/USER/session-pysogs/sogs.log
+# By default logging goes to stderr (which is good; systemd redirects that to system logs), but for
+# debugging it can be useful to send it to a file:
+#logger = file:logfile=/home/USER/session-pysogs/sogs.log

sogs/base_config.py

@@ -16,6 +16,16 @@
 # a free Let's Encrypt certificate.
 URL_BASE = 'http://example.net'
 
+# Address (IP+port) where oxenmq listens for incoming connections.  Can also be specified as a list
+# to listen on multiple addresses.  Note that the connection is encrypted, using the same keys as
+# the main server.  Clients must use the key corresponding to their session_id (i.e. pubkey should
+# match, without the leading 05 byte).
+OMQ_LISTEN = 'tcp://*:22028'
+
+# Address (usually a unix socket) where the main oxenmq process listens for internal connections
+# from other uwsgi workers (to deliver event notifications).  This must not be publicly accessible!
+OMQ_INTERNAL = 'ipc://./omq.sock'
+
 # The log level.
 LOG_LEVEL = logging.WARNING
 

sogs/cleanup.py

@@ -2,14 +2,15 @@
 import os
 import time
 
-from .timer import timer
 from .web import app
 from . import db
 from . import config
 
+# Cleanup interval, in seconds.
+INTERVAL = 10
 
-@timer(10)
-def test_timer(signal):
+
+def cleanup():
     with app.app_context():
         app.logger.debug("Pruning expired items")
         files = prune_files()

sogs/filtration.py

@@ -4,9 +4,9 @@
 
 
 def should_drop_message_with_body(body):
-    body = message_body(body)
     """return true if we should drop a message given its body"""
     if os.path.exists(config.BAD_WORDS_FILE):
+        body = message_body(body)
         with open(config.BAD_WORDS_FILE, 'r') as f:
             for line in f:
                 word = line.rstrip()

sogs/legacy_routes.py

@@ -8,6 +8,7 @@
 from . import config
 from . import http
 from . import filtration
+from .omq import send_mule
 
 import os
 import time
@@ -402,6 +403,8 @@ def handle_legacy_delete_messages(ids=None):
             [room.id, *ids],
         )
 
+    send_mule("messages_deleted", ids)
+
     return jsonify({'status_code': 200})
 
 

sogs/model.py

@@ -3,6 +3,7 @@
 from . import config
 from . import db
 from . import utils
+from .omq import send_mule
 
 import time
 
@@ -530,7 +531,10 @@ def add_post_to_room(user_id, room_id, data, sig, rate_limit_size=5, rate_limit_
         result = conn.execute("SELECT posted, id FROM messages WHERE id = ?", [lastid])
         row = result.fetchone()
         msg = {'timestamp': utils.convert_time(row['posted']), 'server_id': row['id']}
-        return msg
+
+    send_mule("message_posted", msg["server_id"])
+
+    return msg
 
 
 def get_deletions_deprecated(room_id, since):

sogs/mule.py

@@ -0,0 +1,105 @@
+import uwsgi
+import traceback
+import oxenmq
+from oxenc import bt_deserialize
+import time
+from datetime import timedelta
+import functools
+
+from .web import app
+from . import cleanup
+from . import config
+from . import crypto
+from .omq import omq
+
+# 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)
+# - it handles cleanup jobs (e.g. periodic deletions)
+
+
+def run():
+    try:
+        app.logger.info("OxenMQ mule started.")
+
+        while True:
+            time.sleep(1)
+
+    except Exception:
+        app.logger.error("mule died via exception:\n{}".format(traceback.format_exc()))
+
+
+def allow_conn(addr, pk, sn):
+    # TODO: user recognition auth
+    return oxenmq.AuthLevel.basic
+
+
+def admin_conn(addr, pk, sn):
+    return oxenmq.AuthLevel.admin
+
+
+def inproc_fail(connid, reason):
+    raise RuntimeError(f"Couldn't connect mule to itself: {reason}")
+
+
+def setup_omq():
+    global omq, mule_conn
+
+    app.logger.debug("Mule setting up omq")
+    if isinstance(config.OMQ_LISTEN, list):
+        listen = config.OMQ_LISTEN
+    elif config.OMQ_LISTEN is None:
+        listen = []
+    else:
+        listen = [config.OMQ_LISTEN]
+    for addr in listen:
+        omq.listen(addr, curve=True, allow_connection=allow_conn)
+        app.logger.info(f"OxenMQ listening on {addr}")
+
+    # Internal socket for workers to talk to us:
+    omq.listen(config.OMQ_INTERNAL, curve=False, allow_connection=admin_conn)
+
+    # Periodic database cleanup timer:
+    omq.add_timer(cleanup.cleanup, timedelta(seconds=cleanup.INTERVAL))
+
+    # Commands other workers can send to us, e.g. for notifications of activity for us to know about
+    worker = omq.add_category("worker", access_level=oxenmq.AuthLevel.admin)
+    worker.add_command("message_posted", message_posted)
+    worker.add_command("messages_deleted", messages_deleted)
+    worker.add_command("message_edited", message_edited)
+
+    app.logger.debug("Mule starting omq")
+    omq.start()
+
+    # Connect mule to itself so that if something the mule does wants to send something to the mule
+    # it will work.  (And so be careful not to recurse!)
+    app.logger.debug("Mule connecting to self")
+    mule_conn = omq.connect_inproc(on_success=None, on_failure=inproc_fail)
+
+
+def log_exceptions(f):
+    @functools.wraps(f)
+    def wrapper(*args, **kwargs):
+        try:
+            return f(*args, **kwargs)
+        except Exception as e:
+            app.logger.error(f"{f.__name__} raised exception: {e}")
+            raise
+
+    return wrapper
+
+
+@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}")
+
+
+@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}")
+
+
+@log_exceptions
+def message_edited(m: oxenmq.Message):
+    app.logger.warning("FIXME: mule -- message edited stub")

sogs/omq.py

@@ -0,0 +1,63 @@
+# Common oxenmq object; this is used by workers and the oxenmq mule.  We create, but do not start,
+# this pre-forking.
+
+import oxenmq
+from oxenc import bt_serialize
+
+from . import crypto, config
+from .postfork import postfork
+
+omq = None
+mule_conn = None
+
+
+def make_omq():
+    omq = oxenmq.OxenMQ(privkey=crypto._privkey.encode(), pubkey=crypto.server_pubkey.encode())
+
+    # We have multiple workers talking to the mule, so we *must* use ephemeral ids to not replace each
+    # others' connections.
+    omq.ephemeral_routing_id = True
+
+    return omq
+
+
+# Postfork for workers: we start oxenmq and connect to the mule process
+@postfork
+def start_oxenmq():
+    try:
+        import uwsgi
+    except ModuleNotFoundError:
+        return
+
+    global omq, mule_conn
+
+    omq = make_omq()
+
+    if uwsgi.mule_id() != 0:
+        from . import mule
+
+        mule.setup_omq()
+        return
+
+    from .web import app  # Imported here to avoid circular import
+
+    app.logger.debug(f"Starting oxenmq connection to mule in worker {uwsgi.worker_id()}")
+
+    omq.start()
+    app.logger.debug(f"Started, connecting to mule")
+    mule_conn = omq.connect_remote(oxenmq.Address(config.OMQ_INTERNAL))
+
+    app.logger.debug(f"worker {uwsgi.worker_id()} connected to mule OMQ")
+
+
+def send_mule(command, *args, prefix="worker."):
+    """
+    Sends a command to the mule from a worker (or possibly from the mule itself).  The command will
+    be prefixed with "worker." (unless overridden).
+
+    Any args will be bt-serialized and send as message parts.
+    """
+    if prefix:
+        command = prefix + command
+
+    omq.send(mule_conn, command, *(bt_serialize(data) for data in args))