Add oxenmq as a uwsgi mule

This adds an oxenmq as a mule that listens on :22028 by default (using
curve encryption with the same x25519 key as the server).  This oxenmq
now handles db cleanup as well (instead of scheduling it through a
regular front-end worker).

You can kick the mule by sending it signals through the
`signal.send_signal` function so that it can be triggered when things
happen from model; a couple stubs are included here for message creation
and message deletion.

(NB: requires pyoxenmq built against the last oxenmq (1.2.9), such as
the updated python3-oxenmq 1.0.0-3 package, or installed with pip3 with
an updated liboxenmq-dev installed).

Author: jagerman

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,12 @@
 # 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'
+
 # The log level.
 LOG_LEVEL = logging.WARNING
 

sogs/cleanup.py

@@ -2,14 +2,16 @@
 import os
 import time
 
-from .timer import timer
 from .web import app
 from . import db
 from . import config
+from .signal import Signal
 
+# 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/legacy_routes.py

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

sogs/model.py

@@ -3,6 +3,7 @@
 from . import config
 from . import db
 from . import utils
+from .signal import send_signal, Signal
 
 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_signal(Signal.MESSAGE_POSTED)
+
+    return msg
 
 
 def get_deletions_deprecated(room_id, since):

sogs/mule.py

@@ -0,0 +1,94 @@
+import uwsgi
+import traceback
+import oxenmq
+from datetime import timedelta
+
+from .web import app
+from .signal import Signal
+from . import cleanup
+from . import config
+from . import crypto
+
+# 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)
+
+omq = None
+
+
+def run():
+    if uwsgi.mule_id() != 1:
+        app.logger.critical(
+            "Error: sogs.mule must be the first uwsgi mule (mule1), not mule{}".format(
+                uwsgi.mule_id()
+            )
+        )
+        raise RuntimeError("Invalid uwsgi configuration")
+
+    try:
+        setup_omq()
+
+        run_loop()
+    except:
+        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 setup_omq():
+    global omq
+    omq = oxenmq.OxenMQ(
+        privkey=crypto._privkey.encode(),
+        pubkey=crypto.server_pubkey.encode(),
+        log_level=oxenmq.LogLevel.fatal,
+    )
+    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}")
+    omq.add_timer(cleanup.cleanup, timedelta(seconds=cleanup.INTERVAL))
+    omq.start()
+
+
+def run_loop():
+    app.logger.info("mule started!")
+
+    callbacks = {Signal.MESSAGE_POSTED: message_posted, Signal.MESSAGE_DELETED: message_deleted}
+
+    app.logger.info("mule started 2!")
+    while True:
+        app.logger.info("mule started looping")
+        msg = uwsgi.mule_get_msg()
+        app.logger.info("mule ax {}".format(msg))
+        try:
+            sig = Signal(int(msg.decode()))
+        except ValueError:
+            app.logger.error(f"mule received unregistered uwsgi mule message {msg}")
+            continue
+
+        if sig in callbacks:
+            try:
+                callbacks[sig]()
+            except Exception as e:
+                app.logger.error(
+                    f"An exception occured while mule was processing uwsgi signal {sig}:\n{e}"
+                )
+        else:
+            app.logger.error(f"mule received uwsgi signal {sig} but that signal has no handler!")
+    app.logger.info("mule done")
+
+
+def message_posted():
+    app.logger.warning("FIXME: mule -- message posted stub")
+
+
+def message_deleted():
+    app.logger.warning("FIXME: mule -- message delete stub")

sogs/signal.py

@@ -0,0 +1,34 @@
+from enum import Enum
+import logging
+
+
+class Signal(Enum):
+    """
+    UWSGI control signals we use to control the mule process.  These are used to wake up the mule to
+    respond to updates in the database that may need further action, for example to notify connected
+    oxenmq clients of an event.
+    """
+
+    # Signals for the mule.  The values don't matter, as long as they are unique integers
+
+    MESSAGE_POSTED = 1  # A new message has been inserted in the database
+    MESSAGE_DELETED = 2  # A message has been deleted from the database
+    MESSAGE_EDITED = 3  # A message has been edited
+
+
+try:
+    import uwsgi
+
+    _send_msg = uwsgi.mule_msg
+except ModuleNotFoundError:
+    logging.error("uwsgi not available; signals disabled")
+    _send_msg = None
+
+
+def send_signal(sig: Signal):
+    """
+    Sends the given signal to the mule.  If uwsgi is not available then this does nothing.
+    """
+
+    if _send_msg is not None:
+        _send_msg("{}".format(sig.value).encode(), 1)