From 42fe8d9a767ecf5547e0f685f45763d0d50795e1 Mon Sep 17 00:00:00 2001 From: wiktr Date: Mon, 24 Aug 2026 21:49:17 +0200 Subject: [PATCH] implement admin panel --- app/auth.py | 123 ++++++++++++++++++++++++++++++++++++++- app/main.py | 88 ++++++++++++++++++++++++++++ app/templates/admin.html | 7 ++- 3 files changed, 213 insertions(+), 5 deletions(-) diff --git a/app/auth.py b/app/auth.py index accb18c..e017f79 100644 --- a/app/auth.py +++ b/app/auth.py @@ -261,12 +261,12 @@ def get_users(db_path: str) -> dict: cursor = connection.cursor() cursor.execute(''' - SELECT userid, username, is_admin FROM Users + SELECT userid, username, is_admin, email FROM Users ''') users = cursor.fetchall() - users = list(map(lambda x: {"userid": x[0], "username": x[1], "domains": [], "is_admin": bool(x[2])}, users)) + users = list(map(lambda x: {"userid": x[0], "username": x[1], "domains": [], "email": x[3], "is_admin": bool(x[2])}, users)) for user in users: cursor.execute(''' @@ -281,3 +281,122 @@ def get_users(db_path: str) -> dict: user['domains'] = domains return users + +def create_user(db_path: str, username: str, password: str, domains: list = [], is_admin: bool = False, email: str = None): + with sqlite3.connect(db_path) as connection: + cursor = connection.cursor() + + username.strip() + email.strip() + + cursor.execute(''' + SELECT COUNT(*) FROM Users + WHERE username = ? + ''', (username, )) + + if cursor.fetchone()[0] != 0: + return "Username exists" + + if username == "": + return "Username is empty" + + if not username.isalnum(): + return "Username is not alphanumeric" + + if password == "": + return "Password empty" + + if len(password) < 8: + return "Password is shorter than 8 characters" + + salt = urandom(16).hex() + password_hash = sha256( (salt + password).encode('utf-8')).hexdigest() + + cursor.execute(''' + INSERT INTO Users (username, pwsalt, pwhash, email, is_admin) VALUES (?, ?, ?, ?, ?) + ''', (username, salt, password_hash, email, is_admin)) + + cursor.execute(''' + SELECT userid FROM Users + WHERE username = ? + ''', (username, )) + + userid = cursor.fetchone()[0] + + for domain in domains: + domain = domain.strip() + + if domain == "": + continue + + cursor.execute(''' + SELECT COUNT(*) FROM Domains + WHERE domain = ? + ''', (domain, )) + + if cursor.fetchone()[0] != 0: + continue + + cursor.execute(''' + INSERT INTO Domains (userid, domain) VALUES (?, ?) + ''', (userid, domain)) + +def update_user(db_path: str, userid: int, domains: list[str] = None, password: str = None, is_admin: bool = None, email: str = None) -> tuple[bool, str]: + with sqlite3.connect(db_path) as connection: + cursor = connection.cursor() + + cursor.execute(''' + SELECT COUNT(*) FROM Users + WHERE userid = ? + ''', (userid, )) + + if cursor.fetchone()[0] != 1: + return False, "Uerid not found" + + if domains is not None: + # crude check if domains is iterable + for domain in domains: + break + + cursor.execute(''' + DELETE FROM Domains + WHERE userid = ? + ''', (userid, )) + + for domain in domains: + cursor.execute(''' + INSERT INTO Domains (userid, domain) VALUES (?, ?) + ''', (userid, domain)) + + if password is not None: + set_user_password(db_path, userid, password) + + if is_admin is not None: + cursor.execute(''' + UPDATE Users + SET + is_admin = ? + WHERE userid = ? + ''', (is_admin, userid)) + + + if email is not None: + if email == "": + email = None + + cursor.execute(''' + UPDATE Users + SET + email = ? + WHERE userid = ? + ''', (email, userid)) + + + +def delete_user(db_path: str, userid: int): + with sqlite3.connect(db_path) as connection: + cursor = connection.cursor() + + cursor.execute('DELETE FROM Domains WHERE userid = ?', (userid,)) + cursor.execute('DELETE FROM Tokens WHERE userid = ?', (userid,)) + cursor.execute('DELETE FROM Users WHERE userid = ?', (userid,)) diff --git a/app/main.py b/app/main.py index 56816c2..6384cfb 100644 --- a/app/main.py +++ b/app/main.py @@ -121,6 +121,94 @@ def revoke_token(): return redirect("/dashboard") +@app.route('/create-user', methods=["POST"]) +def create_user(): + if session.get("USERID") is None: + return redirect("/") + if not auth.is_admin(DB_PATH, session.get("USERID")): + return redirect("/dashboard") + + if 'username' not in request.form or 'password' not in request.form: + return redirect("/dashboard") + + username = request.form.get("username").strip() + password = request.form.get("password") + + domains = request.form.get('domains') + if domains is None: + domains = [] + else: + collapse_exp = re.compile(r'\s+') + domains = collapse_exp.sub(' ', domains.strip()) + domains = domains.split(" ") + + is_admin = request.form.get("is_admin") is not None + + flash(auth.create_user(DB_PATH, username, password, domains, is_admin, request.form.get('email'))) + + return redirect("/dashboard") + + +@app.route('/update-user', methods=["POST"]) +def update_user(): + if session.get("USERID") is None: + return redirect("/") + if not auth.is_admin(DB_PATH, session.get("USERID")): + return redirect("/dashboard") + + userid = request.form.get("userid") + if userid is None: + return redirect("/dashboard") + try: + userid = int(userid) + except ValueError: + return redirect("/dashboard") + + domains = request.form.get('domains') + + if domains is not None: + collapse_exp = re.compile(r'\s+') + domains = collapse_exp.sub(' ', domains.strip()) + domains = domains.split(" ") + + password = request.form.get("password") + + is_admin = request.form.get("is_admin") + print(is_admin, flush=True) + if is_admin is not None: + is_admin = is_admin == "true" + print(is_admin, flush=True) + + email = request.form.get("email") + + auth.update_user(DB_PATH, userid, domains, password, is_admin, email) + + return redirect("/dashboard") + + + +@app.route('/delete-user', methods=["POST"]) +def delete_user(): + if session.get("USERID") is None: + return redirect("/") + if not auth.is_admin(DB_PATH, session.get("USERID")): + return redirect("/dashboard") + + userid = request.form.get("userid") + if userid is None: + return redirect("/dashboard") + try: + userid = int(userid) + except ValueError: + return redirect("/dashboard") + + auth.delete_user(DB_PATH, userid) + + return redirect("/dashboard") + + + + @app.route('/change-password', methods=["POST"]) def change_password(): if session.get("USERID") is None: diff --git a/app/templates/admin.html b/app/templates/admin.html index bf7d090..f2c3d91 100644 --- a/app/templates/admin.html +++ b/app/templates/admin.html @@ -30,12 +30,13 @@
- - + + +
- +