infrastructure/modules/nixos/ntfy-sh/ntfy-acl.py

#!@python3@/bin/python
# SPDX-FileCopyrightText: 2024 Tom Hubrecht <tom.hubrecht@dgnum.eu>
#
# SPDX-License-Identifier: EUPL-1.2

import json
import sqlite3
import subprocess


def ntfy(*args: str, env=None):
    subprocess.run(["@ntfy@"] + list(args), env=env).check_returncode()


def create_user(u: str, role: str, passwordFile: str, hashedPassword: str):
    # Create the user with the required role and password
    if passwordFile != None:
        with open(passwordFile) as pwd_fp:
            env = {"NTFY_PASSWORD": pwd_fp.read().strip()}

        ntfy("user", "add", f"--role={role}", u, env=env)
    else:
        env = {"NTFY_PASSWORD": hashedPassword}

        ntfy("user", "add", f"--role={role}", u, env=env)
        # HACK: add does not supports hashedPassword entry
        ntfy("user", "change-pass-hash", u, env=env)

def update_user(u: str, role: str, passwordFile: str, hashedPassword: str):
    # Update the user with the required role and password
    if passwordFile != None:
        with open(passwordFile) as pwd_fp:
            env = {"NTFY_PASSWORD": pwd_fp.read().strip()}

        ntfy("user", "change-pass", u, env=env)
    else:
        env = {"NTFY_PASSWORD": hashedPassword}

        ntfy("user", "change-pass-hash", u, env=env)

    ntfy("user", "change-role", u, role)


# Compare the ACL file path to the one used to get the actual data
try:
    with open("/var/lib/ntfy-sh/.acl-path") as acl_path_fp:
        acl_path: str = acl_path_fp.read().strip()
except OSError:
    print("[!] Cannot open .acl-path")
    exit(1)

if acl_path == "@acl_file@":
    print("[-] Unchanged ACL file, exiting")
    exit(0)
else:
    print("[+] ACL file has changed, updating data")

# Get the wanted state
with open("@acl_file@") as acl_fp:
    acl_data = json.load(acl_fp)

# Connect to the db to recover the list of current users
with sqlite3.connect("@user_db@") as con:
    c = con.cursor()
    existing_users: set[str] = set(map(lambda e: e[0], c.execute("SELECT user FROM user"))) - {"*"}

wanted_users: set[str] = set(acl_data["users"].keys())

# Delete extraneous users
for user in existing_users - wanted_users:
    ntfy("user", "del", user)

# Create new users
for user in wanted_users - existing_users:
    create_user(user, **acl_data["users"][user])

# Update existing users
for user in existing_users & wanted_users:
    update_user(user, **acl_data["users"][user])

# Reset ACL rules
ntfy("access", "--reset")

for rule in acl_data["access"]:
    ntfy("access", rule["username"], rule["topic"], rule["permission"])

# Write the new ACL file path
with open("/var/lib/ntfy-sh/.acl-path", "w") as f:
    f.write("@acl_file@")