Merge branch 'dev'

.github/workflows/main.yaml

@@ -29,10 +29,18 @@ jobs:
           username: ${{ secrets.REGISTRY_USER }}
           password: ${{ secrets.REGISTRY_PASS }}
       -
-        name: Build and push signal-event-dispatcher
+        name: Build and push event-dispatcher
         uses: docker/build-push-action@v3
         with:
-          context: .
+          context: ./server/
           platforms: linux/amd64
           push: true
           tags: "${{ secrets.REGISTRY }}/athq/event-dispatcher:latest"
+      -
+        name: Build and push event-dispatcher
+        uses: docker/build-push-action@v3
+        with:
+          context: ./client/
+          platforms: linux/amd64
+          push: true
+          tags: "${{ secrets.REGISTRY }}/athq/event-dispatcher-worker:latest"

client/Dockerfile

@@ -0,0 +1,9 @@
+FROM alpine
+
+WORKDIR /app/
+RUN apk --update --no-cache add python3 py3-requests
+
+COPY ./*.py ./
+
+ENTRYPOINT ["python"]
+CMD ["dispatch-query.py"]

client/dispatch-query.py

@@ -1,36 +1,91 @@
 #!/usr/bin/python3
 
 import sys
+import time
+import argparse
 import subprocess
 import os
 import requests
-from functools import wraps
+import smtphelper
+import json
 
 HTTP_NOT_FOUND = 404
+
+DISPATCH_SERVER = None
 AUTH = None
 
-def email_address(dispatch_uuid, user_topic, message, smtp_target, smtp_user, smtp_pass):
+def debug_send(uuid, data, fail_it=False):
+    '''Dummy function to print and ack a dispatch for debugging'''
+
+    print(json.dumps(data, indent=2))
+    if fail_it:
+        report_failed_dispatch(uuid, "Dummy Error for Debugging")
+    else:
+        confirm_dispatch(uuid)
+
+
+def email_send(dispatch_uuid, email_address, message, smtp_target, smtp_user, smtp_pass):
     '''Send message via email'''
    
+    subject = "Atlantis Dispatch"
+    smtphelper.smtp_send(smtp_target, smtp_user, smtp_pass, email_address, subject, message)
     report_failed_dispatch(uuid, "Email dispatch not yet implemented")
 
-def ntfy_send(dispatch_uuid, user_topic, message, ntfy_push_target, ntfy_user, ntfy_pass):
+def ntfy_api_get_topic(ntfy_api_server, ntfy_api_token, username):
+    '''Get the topic of the user'''
+
+    params = {
+        "user" : username,
+        "token" : ntfy_api_token,
+    }
+
+    r = requests.get(ntfy_api_server + "/topic", params=params)
+    if r.status_code != 200:
+        print(r.text)
+        return None
+    else:
+        print(r.text)
+        return r.json().get("topic")
+
+def ntfy_send(dispatch_uuid, user_topic, title, message, ntfy_push_target, ntfy_user, ntfy_pass):
     '''Send message via NTFY topic'''
 
+    if not user_topic:
+        report_failed_dispatch(dispatch_uuid, "No user topic")
+        return
+
     try:
-        r = requests.post(ntfy_push_target, auth=(ntfy_user, ntfy_pass) , json=payload)
+
+        # build message #
+        payload = {
+            "topic" : user_topic,
+            "message" : message,
+            "title" : title or "Atlantis Notify",
+            #"tags" : [],
+            "priority" : 4,
+            #"attach" : None,
+            "click" : "https://vid.pr0gramm.com/2022/11/06/ed66c8c5a9cd1a3b.mp4",
+            #"actions" : []
+        }
+
+        # send #
+        r = requests.post(ntfy_push_target, auth=(ntfy_user, ntfy_pass), json=payload)
+        print(r.status_code, r.text, payload)
         r.raise_for_status()
-        confirm_dispatch(uuid)
+
+        # talk to dispatch #
+        confirm_dispatch(dispatch_uuid)
+
     except requests.exceptions.HTTPError as e:
-        report_failed_dispatch(uuid, str(e))
+        report_failed_dispatch(dispatch_uuid, str(e))
     except requests.exceptions.ConnectionError as e:
-        report_failed_dispatch(uuid, str(e))
+        report_failed_dispatch(dispatch_uuid, str(e))
 
 def report_failed_dispatch(uuid, error):
     '''Inform the server that the dispatch has failed'''
 
-    response = requests.post(args.dispatch_target + "/report-dispatch-failed",
-                                json={ "uuid" : uuid, "error" : error })
+    payload = [{ "uuid" : uuid, "error" : error }]
+    response = requests.post(DISPATCH_SERVER + "/report-dispatch-failed", json=payload, auth=AUTH)
 
     if response.status_code not in [200, 204]:
         print("Failed to report back failed dispatch for {} ({})".format(
@@ -39,8 +94,8 @@ def report_failed_dispatch(uuid, error):
 def confirm_dispatch(uuid):
     '''Confirm to server that message has been dispatched and can be removed'''
 
-    response = requests.post(target + "/confirm-dispatch", json=[{ "uuid" : uuid }],
-                                auth=(args.user, args.password))
+    payload = [{ "uuid" : uuid }]
+    response = requests.post(DISPATCH_SERVER + "/confirm-dispatch", json=payload, auth=AUTH)
 
     if response.status_code not in [200, 204]:
         print("Failed to confirm dispatch with server for {} ({})".format(
@@ -56,6 +111,9 @@ if __name__ == "__main__":
     parser.add_argument('--dispatch-user')
     parser.add_argument('--dispatch-password')
 
+    parser.add_argument('--ntfy-api-server')
+    parser.add_argument('--ntfy-api-token')
+
     parser.add_argument('--ntfy-push-target')
     parser.add_argument('--ntfy-user')
     parser.add_argument('--ntfy-pass')
@@ -64,15 +122,21 @@ if __name__ == "__main__":
     parser.add_argument('--smtp-user')
     parser.add_argument('--smtp-pass')
 
+    parser.add_argument('--loop', default=True, action=argparse.BooleanOptionalAction)
+
     args = parser.parse_args() 
 
-    # set authentication #
+    # set dispatch server & authentication #
+    DISPATCH_SERVER = args.dispatch_server
     AUTH = (args.dispatch_user, args.dispatch_password)
 
     dispatch_server = args.dispatch_server or os.environ.get("DISPATCH_SERVER")
     dispatch_user = args.dispatch_user or os.environ.get("DISPATCH_USER")
     dispatch_password = args.dispatch_password or os.environ.get("DISPATCH_PASSWORD")
 
+    ntfy_api_server = args.ntfy_api_server or os.environ.get("NTFY_API_SERVER")
+    ntfy_api_token = args.ntfy_api_token or os.environ.get("NTFY_API_TOKEN")
+
     ntfy_push_target = args.ntfy_push_target or os.environ.get("NTFY_PUSH_TARGET")
     ntfy_user = args.ntfy_user or os.environ.get("NTFY_USER")
     ntfy_pass = args.ntfy_pass or os.environ.get("NTFY_PASS")
@@ -81,45 +145,58 @@ if __name__ == "__main__":
     smtp_user = args.smtp_user or os.environ.get("SMTP_USER")
     smtp_pass = args.smtp_pass or os.environ.get("SMTP_PASS")
 
-    # request dispatches #
-    response = requests.get(args.target + "/get-dispatch".format(args.method),
-                            auth=(args.user, args.password))
+    first_run = True
+    while args.loop or first_run:
 
-    # check status #
-    if response.status_code == HTTP_NOT_FOUND:
-        sys.exit(0)
+        # request dispatches #
+        response = requests.get(args.dispatch_server + "/get-dispatch?method=all&timeout=0", auth=AUTH)
 
-    # fallback check for status #
-    response.raise_for_status()
+        # check status #
+        if response.status_code == HTTP_NOT_FOUND:
+            sys.exit(0)
 
-    # track dispatches that were confirmed to avoid duplicate confirmation #
-    dispatch_confirmed = []
+        # fallback check for status #
+        response.raise_for_status()
 
-    # track failed dispatches #
-    errors = dict()
+        # track dispatches that were confirmed to avoid duplicate confirmation #
+        dispatch_confirmed = []
 
-    # iterate over dispatch requests #
-    for entry in response.json():
+        # track failed dispatches #
+        errors = dict()
 
-        user = entry["person"]
-        dispatch_uuid = entry["uid"]
-        method = entry["method"]
-        message = entry["message"]
+        # iterate over dispatch requests #
+        for entry in response.json():
 
-        # method dependent fields #
-        user_topic = entry.get("topic")
-        phone = entry.get("phone")
-        email_address = entry.get("email")
+            user = entry["username"]
+            dispatch_uuid = entry["uuid"]
+            method = entry["method"]
+            message = entry["message"]
+            title = entry.get("title")
 
-        # send message #
-        if method == "signal":
-            pass
-        elif method == "ntfy":
-            ntfy_send(dispatch_uuid, user_topic, message, ntfy_push_target, ntfy_user, ntfy_pass)
-        elif method == "email":
-            email_send(email_address, message)
-        else:
-            print("Unsupported dispatch method {}".format(entry["method"]), sys=sys.stderr)
-            continue
+            # method dependent fields #
+            phone = entry.get("phone")
+            email_address = entry.get("email")
 
-    sys.exit(0)
+            # send message #
+            if method == "signal":
+                pass
+            elif method == "ntfy":
+                user_topic = ntfy_api_get_topic(ntfy_api_server, ntfy_api_token, user)
+                ntfy_send(dispatch_uuid, user_topic, title, message,
+                                ntfy_push_target, ntfy_user, ntfy_pass)
+            elif method == "email":
+                email_send(dispatch_uuid, email_address, message, smtp_target, smtp_user, smtp_pass)
+            elif method == "debug":
+                debug_send(dispatch_uuid, entry)
+            elif method == "debug-fail":
+                debug_send(dispatch_uuid, entry, fail_it=True)
+            else:
+                print("Unsupported dispatch method {}".format(entry["method"]), sys=sys.stderr)
+                continue
+
+        # wait a moment #
+        if args.loop:
+            time.sleep(5)
+        
+        # handle non-loop runs #
+        first_run = False

client/smtphelper.py

@@ -0,0 +1,37 @@
+import smtplib
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+
+def smtp_send(server, user, password, recipient, subject, body):
+
+    # Email and password for authentication
+    sender_email = f'{user}@{server}'
+    sender_password = password
+    
+    # Recipient email address
+    recipient_email = recipient
+    
+    # SMTP server details
+    smtp_server = server
+    smtp_port = 587  # Default port for TLS connection
+    
+    # Create a message
+    message = MIMEMultipart()
+    message['From'] = sender_email
+    message['To'] = recipient_email
+    message['Subject'] = subject
+    
+    # Add body to email
+    body = body
+    message.attach(MIMEText(body, 'plain'))
+    
+    # Establish a connection to the SMTP server
+    server = smtplib.SMTP(smtp_server, smtp_port)
+    server.starttls()  # Secure the connection
+    server.login(sender_email, sender_password)
+    
+    # Send the email
+    server.sendmail(sender_email, recipient_email, message.as_string())
+    
+    # Close the connection
+    server.quit()

server/Dockerfile

@@ -1,19 +1,16 @@
-FROM python:3.9-slim-buster
+FROM alpine
 
-RUN apt update
-RUN apt install python3-pip -y
-RUN apt install libsasl2-dev python-dev libldap2-dev libssl-dev -y
-RUN python3 -m pip install --upgrade pip
-RUN apt install curl -y
-RUN apt autoremove -y
-RUN apt clean
+RUN apk add --update --no-cache python3 py3-pip py3-ldap
 
 WORKDIR /app
 
-RUN python3 -m pip install waitress
+RUN python3 -m pip install --no-cache-dir --break-system-packages waitress
 
 COPY req.txt .
-RUN python3 -m pip install --no-cache-dir -r req.txt
+
+# remove python-ldap (installed via apk) #
+RUN sed -i '/^python-ldap.*$/d' req.txt
+RUN python3 -m pip install --no-cache-dir --break-system-packages -r req.txt
 
 # precreate database directory for mount (will otherwise be created at before_first_request)
 COPY ./ .

server/interface.py

@@ -33,11 +33,35 @@ class DispatchObject(db.Model):
     timestamp = Column(Integer, primary_key=True)
     phone = Column(String)
     email = Column(String)
+
+    title = Column(String)
     message = Column(String, primary_key=True)
     method = Column(String)
+
     dispatch_secret = Column(String)
     dispatch_error = Column(String)
 
+    def serialize(self):
+        ret = {
+            "person" : self.username, # legacy field TODO remove at some point
+            "username" : self.username,
+            "timestamp" : self.timestamp,
+            "phone" : self.phone,
+            "email" : self.email,
+            "title" : self.title,
+            "message" : self.message,
+            "uuid" : self.dispatch_secret,
+            "method" : self.method,
+            "error" : self.dispatch_error,
+        }
+        
+        # fix bytes => string from LDAP #
+        for key, value in ret.items():
+            if type(value) == bytes:
+                ret[key] = value.decode("utf-8")
+
+        return ret
+
 @app.route('/get-dispatch-status')
 def get_dispatch_status():
     '''Retrive the status of a specific dispatch by it's secret'''
@@ -56,6 +80,7 @@ def get_dispatch():
 
     method = flask.request.args.get("method")
     timeout = flask.request.args.get("timeout") or 5 # timeout in seconds
+    timeout = int(timeout)
 
     if not method:
         return (500, "Missing Dispatch Target (signal|email|phone|ntfy|all)")
@@ -72,34 +97,61 @@ def get_dispatch():
     else:
         dispatch_objects = lines_timeout.all()
 
-    # accumulate messages by person #
-    dispatch_by_person = dict()
-    dispatch_secrets = []
-    for dobj in dispatch_objects:
-        if dobj.username not in dispatch_by_person:
-            dispatch_by_person.update({ dobj.username : dobj.message })
-            dispatch_secrets.append(dobj.dispatch_secret)
-        else:
-            dispatch_by_person[dobj.username] += "\n{}".format(dobj.message)
-            dispatch_secrets.append(dobj.dispatch_secret)
+    # TODO THIS IS THE NEW MASTER PART
+    if method and method != "signal":
+        print([ d.serialize() for d in dispatch_objects])
+        return flask.jsonify([ d.serialize() for d in dispatch_objects])
+    else:
+        # TODO THIS PART WILL BE REMOVED ##
+        # accumulate messages by person #
+        dispatch_by_person = dict()
+        dispatch_secrets = []
+        for dobj in dispatch_objects:
+            if dobj.username not in dispatch_by_person:
+                dispatch_by_person.update({ dobj.username : dobj.message })
+                dispatch_secrets.append(dobj.dispatch_secret)
+            else:
+                dispatch_by_person[dobj.username] += "\n{}".format(dobj.message)
+                dispatch_secrets.append(dobj.dispatch_secret)
 
-    response = [ { "person" : tupel[0].decode("utf-8"),
-                    "message" : tupel[1],
-                    "method" : method,
-                    "uids" : dispatch_secrets 
-                  } for tupel in dispatch_by_person.items() ]
+        response = [ { "person" : tupel[0].decode("utf-8"),
+                        "message" : tupel[1],
+                        "method" : method,
+                        "uids" : dispatch_secrets 
+                      } for tupel in dispatch_by_person.items() ]
 
-    # add phone numbers and emails #
-    for obj in response:
-        for person in dispatch_objects:
-            if obj["person"] == person.username.decode("utf-8"):
-                if person.email:
-                    obj.update({ "email" : person.email.decode("utf-8") })
-                if person.phone:
-                    obj.update({ "phone" : person.phone.decode("utf-8") })
+        # add phone numbers and emails #
+        for obj in response:
+            for person in dispatch_objects:
+                if obj["person"] == person.username.decode("utf-8"):
+                    if person.email:
+                        obj.update({ "email" : person.email.decode("utf-8") })
+                    if person.phone:
+                        obj.update({ "phone" : person.phone.decode("utf-8") })
 
-    return flask.jsonify(response)
+        return flask.jsonify(response)
 
+@app.route('/report-dispatch-failed', methods=["POST"])
+def reject_dispatch():
+    '''Inform the server that a dispatch has failed'''
+
+    rejects = flask.request.json
+
+    for r in rejects:
+
+        uuid = r["uuid"]
+        error = r["error"]
+        dpo = db.session.query(DispatchObject).filter(
+                        DispatchObject.dispatch_secret == uuid).first()
+
+        if not dpo:
+            return ("No pending dispatch for this UID/Secret", 404)
+
+        dpo.dispatch_error = error
+        db.session.merge(dpo)
+        db.session.commit()
+
+    return ("", 204)
 
 @app.route('/confirm-dispatch', methods=["POST"])
 def confirm_dispatch():
@@ -109,8 +161,9 @@ def confirm_dispatch():
 
     for c in confirms:
 
-        uid = c["uid"]
-        dpo = db.session.query(DispatchObject).filter(DispatchObject.dispatch_secret == uid).first()
+        uuid = c["uuid"]
+        dpo = db.session.query(DispatchObject).filter(
+                        DispatchObject.dispatch_secret == uuid).first()
 
         if not dpo:
             return ("No pending dispatch for this UID/Secret", 404)
@@ -139,6 +192,12 @@ def smart_send_to_clients():
     users = instructions.get("users")
     groups = instructions.get("groups")
     message = instructions.get("msg")
+    title = instructions.get("title")
+    method = instructions.get("method")
+
+    # allow single use string instead of array #
+    if type(users) == str:
+        users = [users]
 
     struct = instructions.get("data")
     if struct:
@@ -148,13 +207,17 @@ def smart_send_to_clients():
             print(str(e), file=sys.stderr)
             return (e.response(), 408)
 
+    if method in ["debug", "debug-fail"]:
+        persons = [ldaptools.Person(cn="none", username=users[0], name="Mr. Debug",
+                        email="invalid@nope.notld", phone="0")]
+    else:
+        persons = ldaptools.select_targets(users, groups, app.config["LDAP_ARGS"])
 
-    persons = ldaptools.select_targets(users, groups, app.config["LDAP_ARGS"])
-    dispatch_secrets = save_in_dispatch_queue(persons, message)
+    dispatch_secrets = save_in_dispatch_queue(persons, title, message, method)
     return flask.jsonify(dispatch_secrets)
 
 
-def save_in_dispatch_queue(persons, message):
+def save_in_dispatch_queue(persons, title, message, method):
 
 
     dispatch_secrets = []
@@ -166,12 +229,16 @@ def save_in_dispatch_queue(persons, message):
         # this secret will be needed to confirm the message as dispatched #
         dispatch_secret = secrets.token_urlsafe(32)
 
+        # TODO fix this
+        master_method = "signal"
+
         obj = DispatchObject(username=p.username,
                         phone=p.phone,
                         email=p.email,
-                        method="signal",
+                        method=method or master_method,
                         timestamp=datetime.datetime.now().timestamp(),
                         dispatch_secret=dispatch_secret,
+                        title=title,
                         message=message)
 
         db.session.merge(obj)