From abdfcde1c935a6cd0174ee6e956c84475d130981 Mon Sep 17 00:00:00 2001 From: wiktr Date: Mon, 24 Aug 2026 17:24:08 +0200 Subject: [PATCH] add user dashboard --- app/auth.py | 180 ++++++++++++++++++++++++++++++++++- app/main.py | 90 +++++++++++++++++- app/templates/dashboard.html | 49 ++++++++++ app/templates/index.html | 4 +- 4 files changed, 315 insertions(+), 8 deletions(-) create mode 100644 app/templates/dashboard.html diff --git a/app/auth.py b/app/auth.py index e7b06b1..d5a8b18 100644 --- a/app/auth.py +++ b/app/auth.py @@ -1,4 +1,5 @@ import sqlite3 +from os import urandom from hashlib import sha256 def init_db(path: str) -> None: @@ -48,7 +49,7 @@ def check_token_domain(db_path: str, token: str, domain: str) -> bool: return cursor.fetchone()[0] == 1 -def user_login(db_path: str, username: str, password: str) -> bool: +def user_login(db_path: str, username: str, password: str) -> int | None: with sqlite3.connect(db_path) as connection: cursor = connection.cursor() @@ -59,15 +60,186 @@ def user_login(db_path: str, username: str, password: str) -> bool: password_salt = cursor.fetchone() if password_salt is None: - return False + return None password_salt = password_salt[0] password_hash = sha256( (password_salt + password).encode('utf-8') ).hexdigest() cursor.execute(''' - SELECT COUNT(*) FROM Users + SELECT userid FROM Users WHERE username = ? AND pwhash = ? ''', (username, password_hash)) - return cursor.fetchone()[0] == 1 + result = cursor.fetchone() + if result is None: + return None + + return result[0] + +def get_username(db_path: str, userid: int) -> str | None: + with sqlite3.connect(db_path) as connection: + cursor = connection.cursor() + + cursor.execute(''' + SELECT username FROM Users + WHERE userid = ? + ''', (userid, )) + + username = cursor.fetchone() + if username is None: + return None + + return username[0] + +def get_user_domains(db_path: str, userid: int) -> list[str]: + with sqlite3.connect(db_path) as connection: + cursor = connection.cursor() + + cursor.execute(''' + SELECT domain FROM Domains + WHERE userid = ? + ''', (userid, )) + + domains = cursor.fetchall() + + domains = map(lambda x: x[0], domains) + + return domains + +def get_user_tokens(db_path: str, userid: int) -> list[str]: + with sqlite3.connect(db_path) as connection: + cursor = connection.cursor() + + cursor.execute(''' + SELECT name FROM Tokens + WHERE userid = ? + ''', (userid, )) + + tokens = cursor.fetchall() + + tokens = map(lambda x: x[0], tokens) + + return tokens + +def generate_user_token(db_path: str, userid: int, token_name: str) -> str | None: + 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 None + + cursor.execute(''' + SELECT COUNT(*) FROM Tokens + WHERE userid = ? + ''', (userid, )) + + if cursor.fetchone()[0] >= 5: + return None + + cursor.execute(''' + SELECT COUNT(*) FROM Tokens + WHERE userid = ? + AND name = ? + ''', (userid, token_name)) + + if cursor.fetchone()[0] != 0: + return None + + if not token_name.isalnum(): + return None + + if len(token_name) >= 16: + return None + + for _ in range(10): + try: + token = urandom(32).hex() + token_hash = sha256(token.encode("utf-8")).hexdigest() + + cursor.execute(''' + INSERT INTO Tokens (tkhash, userid, name) VALUES (?, ?, ?) + ''', (token_hash, userid, token_name)) + + return token + except sqlite3.IntegrityError: + pass + + return None + + +def revoke_user_token(db_path: str, userid: int, token_name: str) -> None: + with sqlite3.connect(db_path) as connection: + cursor = connection.cursor() + + cursor.execute(''' + DELETE FROM Tokens + WHERE userid = ? + AND name = ? + ''', (userid, token_name)) + +def change_user_password(db_path: str, userid: int, oldpass: str, newpass: str) -> bool: + 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 + + cursor.execute(''' + SELECT pwsalt FROM Users + WHERE userid = ? + ''', (userid, )) + + salt = cursor.fetchone() + if salt is None: + return False + salt = salt[0] + + + oldpass_hash = sha256( (salt + oldpass).encode('utf-8')).hexdigest() + + cursor.execute(''' + SELECT COUNT(*) FROM Users + WHERE userid = ? + AND pwhash = ? + ''', (userid, oldpass_hash)) + + if cursor.fetchone()[0] != 1: + return False + + return set_user_password(db_path, userid, newpass) + +def set_user_password(db_path: str, userid: int, newpass: str) -> bool: + 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 + + salt = urandom(16).hex() + newpass_hash = sha256( (salt + newpass).encode('utf-8')).hexdigest() + + cursor.execute(''' + UPDATE Users + SET + pwsalt = ?, + pwhash = ? + WHERE userid = ? + ''', (salt, newpass_hash, userid)) + + return True diff --git a/app/main.py b/app/main.py index e446192..c00115c 100644 --- a/app/main.py +++ b/app/main.py @@ -1,4 +1,4 @@ -from flask import Flask, jsonify, request, render_template, redirect +from flask import Flask, jsonify, request, render_template, redirect, session, flash import auth import re @@ -10,6 +10,11 @@ DB_PATH = "/app/sqlite.db" app = Flask(__name__) +app.config["SESSION_PERMANENT"] = True +app.config["SESSION_TYPE"] = "memcached" +app.config["PERMANENT_SESSION_LIFETIME"] = 3600 + +app.config["SECRET_KEY"] = "dsadasda" hosts = {} @@ -38,6 +43,8 @@ def validate_ip(ip: str) -> bool: @app.route('/', methods=["GET"]) def homepage(): + if session.get("USERID") is not None: + return redirect("/dashboard") return render_template("index.html") @@ -50,13 +57,92 @@ def login(): if request.form["user"].strip() == "" or request.form["pass"].strip() == "": return redirect("/"), 400 - if not auth.user_login(DB_PATH, request.form["user"], request.form["pass"]): + userid = auth.user_login(DB_PATH, request.form["user"], request.form["pass"]) + if userid is None: return redirect("/") + session["USERID"] = userid + return redirect("/dashboard", code=302) +@app.route('/logout', methods=["GET"]) +def logout(): + session.clear() + return redirect("/") + +@app.route('/dashboard', methods=["GET"]) +def dashboard(): + if session.get("USERID") is None: + return redirect("/") + username = auth.get_username(DB_PATH, session.get("USERID")) + domains = auth.get_user_domains(DB_PATH, session.get("USERID")) + tokens = auth.get_user_tokens(DB_PATH, session.get("USERID")) + + return render_template("dashboard.html", username=username, domains=domains, tokens=tokens) + + + +@app.route('/generate-token', methods=["POST"]) +def generate_token(): + if session.get("USERID") is None: + return redirect("/") + + if "token" not in request.form: + return redirect("/dashboard") + if request.form["token"].strip() == "": + return redirect("/dashboard") + + token = auth.generate_user_token(DB_PATH, session.get("USERID"), request.form["token"]) + + if token is not None: + flash(token) + + return redirect("/dashboard") + + +@app.route('/revoke-token', methods=["POST"]) +def revoke_token(): + if session.get("USERID") is None: + return redirect("/") + + if "token" not in request.form: + return redirect("/dashboard") + if request.form["token"].strip() == "": + return redirect("/dashboard") + + auth.revoke_user_token(DB_PATH, session.get("USERID"), request.form["token"]) + + return redirect("/dashboard") + + +@app.route('/change-password', methods=["POST"]) +def change_password(): + if session.get("USERID") is None: + return redirect("/") + + if "pass" not in request.form or "pass-new" not in request.form or "pass-rep" not in request.form: + flash("password change failed") + return reidrect("/dashboard") + if request.form["pass"] == "" or request.form["pass-new"] == "" or request.form["pass-rep"] == "": + flash("password change failed") + return redirect("/dashboard") + + oldpass = request.form["pass"] + newpass = request.form["pass-new"] + + if newpass != request.form["pass-rep"]: + flash("passwords do not match") + return redirect("/dashboard") + + if not auth.change_user_password(DB_PATH, session.get("USERID"), oldpass, newpass): + flash("password change failed") + return redirect("/dashboard") + + flash("password changed successfully") + return redirect("/dashboard") + @app.route('/health', methods=["GET"]) def health(): return jsonify({"status": "healthy", "comment": ""}) diff --git a/app/templates/dashboard.html b/app/templates/dashboard.html new file mode 100644 index 0000000..8e16c53 --- /dev/null +++ b/app/templates/dashboard.html @@ -0,0 +1,49 @@ + + + + Dashboard - IZBI DNS + + +

Hello {{ username }} log out

+ +{% with messages = get_flashed_messages()%} +{% if messages%} +{% for message in messages%} +

{{message}}

+{%endfor%} +{%endif%} +{%endwith%} + +

Your domains: +

+

+ +

Your tokens: +

+ + +
+ +

+ +

Password change +

+
+
+
+ +
+

+ + diff --git a/app/templates/index.html b/app/templates/index.html index 0cc30fe..ee06efc 100644 --- a/app/templates/index.html +++ b/app/templates/index.html @@ -7,9 +7,9 @@

Dynamic DNS service for .izbi domains :3

-
+

- +