372 lines
12 KiB
Python
372 lines
12 KiB
Python
#!/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())
|
|
|
|
headers = {{"Authorization": f"Bearer {{token}}"}}
|
|
params = {{
|
|
"jq": args.jq,
|
|
"type": args.type,
|
|
}}
|
|
if args.title:
|
|
params["title"] = args.title
|
|
|
|
req = requests.post(args.url, headers=headers, params=params, data=data)
|
|
exit(not req.status_code == 200)
|
|
"""
|
|
|
|
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)
|
|
|
|
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]
|
|
|
|
# hack to make users confirm it
|
|
generate_token_name: str = "tmpaction"
|
|
generate_token_value: str = "Generate Token"
|
|
if request.form.get(generate_token_name):
|
|
generate_token_name = "action"
|
|
generate_token_value = "Generate Token. Are you sure?"
|
|
|
|
tmpl = f"""<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
|
<meta http-equiv="X-UA-Compatible" content="ie=edge">
|
|
<title>Notification Service</title>
|
|
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/css/bootstrap.min.css" rel="stylesheet" integrity="sha384-QWTKZyjpPEjISv5WaRU9OFeRpok6YctnYmDr5pNlyT2bRjXh0JMhjY6hW+ALEwIH" crossorigin="anonymous">
|
|
</head>
|
|
<body>
|
|
|
|
<div class="container my-5">
|
|
<h1>Hello {username}!</h1>
|
|
<div class="col-lg-8 px-0">
|
|
<form method="post">
|
|
<div class="mb-3">
|
|
<label class="form-label">Default Matrix Room ID</label>
|
|
<input type="text" name="room_id" value="{room_id}" placeholder="!yREJWHUMJhGROiHbtu:fricloud.dk" class="form-control">
|
|
</div>
|
|
<input type="submit" class="btn btn-primary" name="action" value="Set Default Room ID">
|
|
<hr>
|
|
<div class="mb-3">
|
|
<label class="form-label">Token</label>
|
|
<input type="text" value="{token}" placeholder="token-not-generated" readonly class="form-control" >
|
|
</div>
|
|
<input type="submit" class="btn btn-primary" name="{generate_token_name}" value="{generate_token_value}">
|
|
</form>
|
|
<hr>
|
|
<p>
|
|
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}.
|
|
</p>
|
|
<hr>
|
|
<table class="table">
|
|
<thead>
|
|
<tr>
|
|
<th scope="col">Param</th>
|
|
<th scope="col">Description</th>
|
|
<th scope="col">Default</th>
|
|
<th scope="col">Example</th>
|
|
</tr>
|
|
</thead>
|
|
<tbody>
|
|
<tr>
|
|
<th scope="row">title</th>
|
|
<td>Title of notification</td>
|
|
<td>Notification</td>
|
|
<td>Compilation Finished!</td>
|
|
</tr>
|
|
<tr>
|
|
<th scope="row">body</th>
|
|
<td>Body of notification</td>
|
|
<td>empty</td>
|
|
<td>Compilation result: success (default is nothing)</td>
|
|
</tr>
|
|
<tr>
|
|
<th scope="row">type</th>
|
|
<td>type of notification</td>
|
|
<td>matrix</td>
|
|
<td>matrix</td>
|
|
</tr>
|
|
<tr>
|
|
<th scope="row">room_id</th>
|
|
<td>Matrix room ID</td>
|
|
<td>default room</td>
|
|
<td>!yREJWHUMJhGROiHbtu:fricloud.dk or #na-offtopic:rend.al</td>
|
|
</tr>
|
|
</tbody>
|
|
</table>
|
|
<hr>
|
|
curl
|
|
<pre class="border"><code>curl "{CONFIG_URL}/notify" -H "Authorization: Bearer {token}"</code></pre>
|
|
curl w/ specific body/title
|
|
<pre class="border"><code>curl "{CONFIG_URL}/notify?title=MyTitle&body=MyBody" -H "Authorization: Bearer {token}"</code></pre>
|
|
Python
|
|
<pre class="border"><code>{script_example}</code></pre>
|
|
Python w/ <b>hardcoded token (DO NOT SHARE)</b>
|
|
<pre class="border"><code>{script_example_with_token}</code></pre>
|
|
|
|
Nix Python Script <b>HARDCODED TOKEN (DO NOT SHARE)</b>
|
|
<pre class="border"><code>pkgs.writers.writePython3Bin "notify" {{
|
|
libraries = with pkgs.python3Packages; [ requests ];
|
|
doCheck = false;
|
|
}} ''{script_example_with_token}'';
|
|
</code></pre>
|
|
|
|
<!--
|
|
<p class="fs-5">You've successfully loaded up the Bootstrap starter example. It includes <a href="https://getbootstrap.com/">Bootstrap 5</a> via the <a href="https://www.jsdelivr.com/package/npm/bootstrap">jsDelivr CDN</a> and includes an additional CSS and JS file for your own code.</p>
|
|
<p>Feel free to download or copy-and-paste any parts of this example.</p>
|
|
|
|
<hr class="col-1 my-4">
|
|
|
|
<a href="https://getbootstrap.com" class="btn btn-primary">Read the Bootstrap docs</a>
|
|
<a href="https://github.com/twbs/examples" class="btn btn-secondary">View on GitHub</a>
|
|
-->
|
|
</div>
|
|
</div>
|
|
</body>
|
|
</html>
|
|
""".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 <token> 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)
|