#!/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
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_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"))
CONFIG_MAIL_MODE = getenv("MAIL_MODE", "ssl")
script_example = r"""#!/usr/bin/env bash
BODY="$1"
TITLE=${2:-Notification}
JQ_EXPR=${3:-.}
TYPE=${4:-matrix}
TOKEN="$(cat ~/.config/notify/token)"
# TOKEN="$(cat /run/agenix/notify-token)"
URL="||URL||/notify"
# get stdin if needed
if [ "$BODY" = "-" ]; then
BODY="$(cat -)"
fi
# make request
curl -H "Authorization: Bearer $TOKEN" "$URL" \
--get \
--data-urlencode "title=$TITLE" \
--data-urlencode "body=$BODY" \
--data-urlencode "jq=$JQ_EXPR" \
--data-urlencode "type=$TYPE"
""".replace(
"||URL||", "https://notify.fricloud.dk"
)
app = Flask(__name__)
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()
print("Action", action)
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}!
{script_example}
"""
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 = " "
print("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)
con = get_db()
cur = con.cursor()
print(token)
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)
print(ntype, username, token)
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]
print(room_id)
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)