#!/usr/bin/env nix-shell #!nix-shell --pure -i python3 -p "python3.withPackages (ps: with ps; [ flask apprise mnemonic wtforms jq ])" from typing import Any import apprise from flask import Flask, request from mnemonic import Mnemonic import sqlite3 import os import jq import json app = Flask(__name__) import logging log = logging.getLogger("werkzeug") log.setLevel(logging.ERROR) ENV_PREFIX = "NOTIFIER_" def getenv(key: str, default: Any = None) -> Any: v = os.getenv(ENV_PREFIX + key, default) if not v: exit(f"{ENV_PREFIX+key} must be specified!") return v CONFIG_URL = getenv("URL", "http://127.0.0.1") CONFIG_PORT = int(getenv("PORT", 8080)) CONFIG_DATABASE_PATH = getenv("DATABASE_PATH", "notifications.db") CONFIG_MATRIX_BOT_NAME = getenv("MATRIX_BOT_NAME", "unset") CONFIG_MATRIX_BOT_TOKEN = getenv("MATRIX_BOT_TOKEN") CONFIG_MATRIX_HOST = getenv("MATRIX_HOST") CONFIG_PROXY_AUTH_USERNAME_HEADER = getenv("PROXY_AUTH_USERNAME_HEADER", "Remote-User") CONFIG_MAIL_USERNAME = getenv("MAIL_USERNAME") CONFIG_MAIL_PASSWORD = getenv("MAIL_PASSWORD") CONFIG_MAIL_DOMAIN = getenv("MAIL_DOMAIN") CONFIG_MAIL_HOST = getenv("MAIL_HOST") CONFIG_MAIL_PORT = int(getenv("MAIL_PORT", "465")) CONFIG_MAIL_MODE = getenv("MAIL_MODE", "ssl") script_example = rf"""#!/usr/bin/env bash #!/usr/bin/env nix-shell #!nix-shell --pure -i python3 -p "python3.withPackages (ps: with ps; [ requests ])" import requests import argparse import sys import os parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter) parser.add_argument("--title", default="", help="Subject/title of the notification") parser.add_argument("--body", default="", help="`-` for stdin") parser.add_argument("--jq", default=".", help="Jq filter to use") parser.add_argument("--type", default="matrix", help="mail or matrix") parser.add_argument("--token", help="token to use") parser.add_argument("--token-file", help="file to read token from") parser.add_argument("--url", default="{CONFIG_URL}/notify", help="Notify endpoint") args = parser.parse_args() token: str = args.token if args.token_file: token = open(args.token_file, "r").read().strip() if not token: exit("No token or tokenfile specified, or empty") data = args.body if data == "-" and not sys.stdin.isatty(): data = "\n".join(sys.stdin.readlines()) req = requests.post(args.url, headers={{"Authorization": f"Bearer {{token}}"}}, data=data) print(req.text) """ script_example_with_token = script_example.replace( '--token"', '--token", default="||TOKEN||"', ) def get_db(): con = sqlite3.connect(CONFIG_DATABASE_PATH) cur = con.cursor() cur.execute( "CREATE TABLE IF NOT EXISTS default_room(username TEXT PRIMARY KEY, room_id TEXT NOT NULL)" ) cur.execute( "CREATE TABLE IF NOT EXISTS tokens(username TEXT PRIMARY KEY, token TEXT NOT NULL)" ) return con @app.route("/", methods=["GET", "POST"]) def index(): username = request.headers.get(CONFIG_PROXY_AUTH_USERNAME_HEADER) if not username: return ("Not authenticated", 401) # handle post stuff if request.method == "POST": action = request.form.get("action", "").lower() if "token" in action: generate_token_for_user(username) elif "room id" in action: roomid = request.form.get("room_id") if not roomid: return ("Room Id cannot be empty", 400) set_user_default_matrix_room(username, roomid) else: return ("Unknown action", 400) con = get_db() cur = con.cursor() res = cur.execute( "SELECT token FROM tokens WHERE username = ?", (username,) ).fetchone() token: str = "" if res: token = res[0] res = cur.execute( "SELECT room_id FROM default_room WHERE username = ?", (username,) ).fetchone() room_id: str = "" if res: room_id = res[0] tmpl = f""" Notification Service

Hello {username}!

This notification service has support for both matrix and email. Matrix notifications will be sent from {CONFIG_MATRIX_BOT_NAME}, be sure to invite them to the room/space if it is private. If using email, they will only be sent to your member email, and can't be sent anywhere else. You'll receive them from {CONFIG_MAIL_USERNAME}@{CONFIG_MAIL_DOMAIN}.

Param	Description	Default	Example
title	Title of notification	Notification	Compilation Finished!
body	Body of notification	empty	Compilation result: success (default is nothing)
type	type of notification	matrix	matrix
room_id	Matrix room ID	default room	!yREJWHUMJhGROiHbtu:fricloud.dk or #na-offtopic:rend.al

curl

curl "{CONFIG_URL}/notify" -H "Authorization: Bearer {token}"

curl w/ specific body/title

curl "{CONFIG_URL}/notify?title=MyTitle&body=MyBody" -H "Authorization: Bearer {token}"

Python

{script_example}

Python w/ hardcoded token (DO NOT SHARE)

{script_example_with_token}

Nix Python Script HARDCODED TOKEN (DO NOT SHARE)

pkgs.writers.writePython3Bin "notify" {{
    libraries = with pkgs.python3Packages; [ requests ];
    doCheck = false;
  }} ''{script_example_with_token}'';

""".replace( r"||TOKEN||", token ) return tmpl @app.route("/notify", methods=["GET", "POST"]) def send_notification(): # default to this ntype = request.args.get("type", "matrix") title = request.args.get("title", "Notification") body = request.args.get("body", request.get_data().decode("utf-8")) if not body: body = " " token = request.authorization.token or request.authorization.password if not token: return ( "No token found, please either specify with Authorization: Bearer or HTTPBasic", 401, ) if ( request.method == "POST" and "json" in request.headers.get("content-type", "") or is_json(body) ): try: body = jq.compile(request.args.get("jq", ".")).input_text(body).first() except Exception: return ("Unable to compile JQ, please ensure it is correct", 400) if not isinstance(body, str): body = json.dumps(body, indent=4) con = get_db() cur = con.cursor() res = cur.execute( "SELECT username FROM tokens WHERE token = ?", (token,) ).fetchone() if not res: return ("Access denied - invalid token", 401) username = res[0] if ntype not in ["matrix", "mail"]: return ("Invalid type, only matrix or mail allowed", 400) apobj = apprise.Apprise() if ntype == "matrix": # try to get a room_id room_id = request.args.get("room_id") if not room_id: res = cur.execute( "SELECT room_id FROM default_room WHERE username = ?", (username,) ).fetchone() if res: room_id = res[0] if not room_id: return ("No room_id specified, and no default saved", 400) apobj.add(f"matrixs://{CONFIG_MATRIX_BOT_TOKEN}@{CONFIG_MATRIX_HOST}/{room_id}") else: apobj.add( f"mailto://{CONFIG_MAIL_USERNAME}:{CONFIG_MAIL_PASSWORD}@{CONFIG_MAIL_HOST}:{CONFIG_MAIL_PORT}?mode={CONFIG_MAIL_MODE}&from={CONFIG_MAIL_USERNAME}@{CONFIG_MAIL_DOMAIN}&to={username}@{CONFIG_MAIL_DOMAIN}" ) sent_notification = apobj.notify( title=title, body=body, ) if not send_notification: return ("Unable to send notification", 500) return ("Notification sent!", 200) def generate_token_for_user(username: str): token = generate_token() con = get_db() cur = con.cursor() t = cur.execute( "INSERT INTO tokens (username, token) VALUES (?, ?) ON CONFLICT (username) DO UPDATE SET token = ?", ( username, token, token, ), ) con.commit() def set_user_default_matrix_room(username: str, roomid: str): con = get_db() cur = con.cursor() t = cur.execute( "INSERT INTO default_room (username, room_id) VALUES (?, ?) ON CONFLICT (username) DO UPDATE SET room_id = ?", ( username, roomid, roomid, ), ) con.commit() def generate_token(num_words: int = 5) -> str: mnemo = Mnemonic("english") words = mnemo.generate(strength=256) return "-".join(words.split(" ")[0:num_words]) def is_json(potential_json: str) -> bool: try: json.loads(potential_json) return True except Exception: return False if __name__ == "__main__": app.run(host="127.0.0.1", port=CONFIG_PORT, debug=True)