implement admin panel

app/auth.py

@@ -1,34 +1,402 @@
 import sqlite3
+from os import urandom
+from hashlib import sha256
 
-#def __init__(self, path: str) -> None:
+def init_db(path: str) -> None:
-#    self.db_connection = sqlite3.connect(path)
+    with sqlite3.connect(path) as connection:
-#    cursor = self.db_connection.cursor()
+        cursor = connection.cursor()
-#    cursor.execute('''
+        cursor.execute('''
-#        CREATE TABLE IF NOT EXISTS Tokens (
+            CREATE TABLE IF NOT EXISTS Tokens (
-#            TOKEN TEXT PRIMARY KEY NOT NULL,
+                TKHASH TEXT PRIMARY KEY NOT NULL,
-#            USER TEXT NOT NULL,
+                USERID INTEGER NOT NULL,
-#            NAME TEXT NOT NULL
+                NAME TEXT NOT NULL
-#        )
+            )
-#    ''')
+        ''')
-#
+
-#    cursor.execute('''
+        cursor.execute('''
-#        CREATE TABLE IF NOT EXISTS Domains (
+            CREATE TABLE IF NOT EXISTS Domains (
-#            DOMAIN TEXT PRIMARY KEY NOT NULL,
+                DOMAIN TEXT PRIMARY KEY NOT NULL,
-#            USER TEXT NOT NULL
+                USERID INTEGER NOT NULL
-#        )
+            )
-#    ''')
+        ''')
-#
+
-#    self.db_connection.commit()
+        cursor.execute('''
-#    cursor.close()
+            CREATE TABLE IF NOT EXISTS Users (
+                USERID INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
+                USERNAME TEXT NOT NULL UNIQUE,
+                PWHASH TEXT NOT NULL,
+                PWSALT TEXT NOT NULL,
+                EMAIL TEXT
+                IS_ADMIN INTEGER NOT NULL DEFAULT 0
+            )
+        ''')
+
+        connection.commit()
+        cursor.close()
 
 def check_token_domain(db_path: str, token: str, domain: str) -> bool:
     with sqlite3.connect(db_path) as connection:
         cursor = connection.cursor()
+
+        token_hash = sha256(token.encode('utf-8')).hexdigest()
+
         cursor.execute('''
             SELECT COUNT(*) FROM Tokens
-            JOIN Domains ON Tokens.user = Domains.user
+            JOIN Domains ON Tokens.userid = Domains.userid
-            WHERE token = ?
+            WHERE tkhash = ?
                 AND domain = ?
-        ''', (token,domain))
+        ''', (token_hash, domain))
 
     return cursor.fetchone()[0] == 1
+
+def user_login(db_path: str, username: str, password: str) -> int | None:
+    with sqlite3.connect(db_path) as connection:
+        cursor = connection.cursor()
+
+        cursor.execute('''
+            SELECT pwsalt FROM Users
+            WHERE username = ?
+        ''', (username,))
+
+        password_salt = cursor.fetchone()
+        if password_salt is None:
+            return None
+        password_salt = password_salt[0]
+
+        password_hash = sha256( (password_salt + password).encode('utf-8') ).hexdigest()
+
+        cursor.execute('''
+            SELECT userid FROM Users
+            WHERE username = ?
+                AND pwhash = ?
+        ''', (username, password_hash))
+
+        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
+
+def is_admin(db_path: str, userid: int) -> bool:
+    with sqlite3.connect(db_path) as connection:
+        cursor = connection.cursor()
+
+        cursor.execute('''
+            SELECT COUNT(*) FROM Users
+            WHERE userid = ?
+                AND IS_ADMIN = 1
+        ''', (userid, ))
+
+        return cursor.fetchone()[0] == 1
+
+def get_users(db_path: str) -> dict:
+    with sqlite3.connect(db_path) as connection:
+        cursor = connection.cursor()
+
+        cursor.execute('''
+            SELECT userid, username, is_admin, email FROM Users
+        ''')
+
+        users = cursor.fetchall()
+
+        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('''
+                SELECT domain FROM Domains
+                WHERE userid = ?
+            ''', (user["userid"], ))
+
+            domains = cursor.fetchall()
+
+            domains = list(map(lambda x: x[0], domains))
+
+            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,))

app/main.py

@@ -1,4 +1,4 @@
-from flask import Flask, jsonify, request
+from flask import Flask, jsonify, request, render_template, redirect, session, flash
 import auth
 import re
 
@@ -8,7 +8,13 @@ HOSTS_PATH = "/app/hosts.d/izbi"
 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 = {}
 
@@ -36,6 +42,200 @@ 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")
+
+
+
+@app.route('/login', methods=["POST"])
+def login():
+    if "user" not in request.form or "pass" not in request.form:
+        return redirect("/"), 400
+
+    if request.form["user"].strip() == "" or request.form["pass"].strip() == "":
+        return redirect("/"), 400
+
+    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("/")
+    userid = session.get("USERID")
+    username = auth.get_username(DB_PATH, userid)
+    domains = auth.get_user_domains(DB_PATH, userid)
+    tokens = auth.get_user_tokens(DB_PATH, userid)
+
+    if auth.is_admin(DB_PATH, userid):
+        users = auth.get_users(DB_PATH)
+        return render_template("dashboard.html", username=username, domains=domains, tokens=tokens, is_admin=True, users=users)
+    return render_template("dashboard.html", username=username, domains=domains, tokens=tokens, is_admin=False)
+
+
+
+@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('/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:
+        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": ""})
 
@@ -73,10 +273,12 @@ def update_addr():
             return jsonify({"status": "500", "code": "internal-server-error", "comment": "An error occurred while processing request"}), 500
 
         return jsonify({"status": "200", "code": "ok", "comment": "Updated successfully"})
-    except:
+    except Exception as e:
+        print(e, flush=True)
         return jsonify({"status": "500", "code": "internal-server-error", "comment": "An error occurred while processing request"}), 500
 
 
 
 if __name__ == '__main__':
+    auth.init_db(DB_PATH)
     app.run(host="0.0.0.0", port=8080, debug=True)

app/templates/admin.html

@@ -0,0 +1,49 @@
+
+
+	<h2>Administration</h2>
+	
+	<p>Users:
+		<table>
+			<tr><th>Username</th><th>Domains</th><th>Password</th><th>Admin</th><th>Email</th><th>Delete</th></tr>
+			
+			<form method="POST" action="/create-user"><tr>
+				<td><input type="text" name="username" placeholder="*user name" required /></td>
+				<td><input type="text" name="domains" placeholder="space separated domains" /></td>
+				<td><input type="password" name="password" placeholder="*password" required /></td>
+				<td><input type="checkbox" name="is_admin" /></td>
+				<td><input type="email" name="email" placeholder="email" /></td>
+				<td><button type="submit">Add</button></td>
+			</tr></form>
+
+			{% for user in users %}
+			<tr>
+				<td>{{ user["username"] }}</td>
+				<td><form method="POST" action="/update-user">
+					<input type="hidden" name="userid" value="{{ user["userid"] }}" />
+					<input type="text" name="domains" placeholder="space separated domains" value="{{ user["domains"] | join(' ') }}" />
+					<button type="submit">update domains</button>
+				</form></td>
+				<td><form method="POST" action="/update-user">
+					<input type="hidden" name="userid" value="{{ user["userid"] }}" />
+					<input type="password" name="password" placeholder="*password" required />
+					<button type="submit">update password</button>
+				</form></td>
+				<td><form method="POST" action="/update-user">
+					<input type="hidden" name="userid" value="{{ user["userid"] }}" />
+					<input type="hidden" name="is_admin" value="{% if user["is_admin"] %}false{% else %}true{% endif %}" />
+					<input type="checkbox" disabled {% if user["is_admin"] %}checked{% endif %} />
+					<button type="submit">toggle admin</button>
+				</form></td>
+				<td><form method="POST" action="/update-user">
+					<input type="hidden" name="userid" value="{{ user["userid"] }}" />
+					<input type="email" name="email" placeholder="email" value="{{ user["email"] }}"/>
+					<button type="submit">update email</button>
+				</form></td>
+				<td><form method="POST" action="/delete-user">
+					<input type="hidden" name="userid" value="{{ user["userid"] }}" />
+					<button type="submit">Delete</button>
+				</form></td>
+			</tr>
+			{% endfor %}
+		</table>
+	</p>

app/templates/dashboard.html

@@ -0,0 +1,49 @@
+<!DOCTYPE html>
+<html>
+<head>
+	<title>Dashboard - IZBI DNS</title>
+</head>
+<body>
+	<h2>Hello {{ username }} <a href="/logout">log&nbsp;out</a></h2>
+
+{% with messages = get_flashed_messages()%}
+{% if messages%}
+{% for message in messages%}
+	<h3>{{message}}</h3>
+{%endfor%}
+{%endif%}
+{%endwith%}
+
+	<p>Your domains:
+		<ul>{% for domain in domains %}
+			<li>{{ domain }}</li>
+{% endfor %}
+		</ul>
+	</p>
+
+	<p>Your tokens: 
+		<form method="POST" action="/generate-token">
+			<input type="text" name="token" placeholder="token name" />
+			<button type="submit">new&nbsp;token</button>
+		</form>
+		<ul>{% for token in tokens %}
+			<li>
+				<form method="POST" action="/revoke-token">
+					<input type="text" name="token" readonly value="{{ token }}" />
+					<button type="submit">revoke</button>
+				</form>
+			</li>
+{% endfor %}
+		</ul>
+	</p>
+
+	<p>Password change
+		<form method="POST" action="/change-password">
+			<input type="password" name="pass" placeholder="old password" /><br />
+			<input type="password" name="pass-new" placeholder="new password" /><br />
+			<input type="password" name="pass-rep" placeholder="repeat password" /><br />
+			<button type="submit">change password</button>
+		</form>
+	</p>{% if is_admin %}{% include 'admin.html' %}{% endif %}
+</body>
+</html>

app/templates/index.html

@@ -0,0 +1,15 @@
+<!DOCTYPE html>
+<html>
+<head>
+	<title>IZBI DNS</title>
+</head>
+<body>
+	<p>Dynamic DNS service for <span style="font-family: monospace">.izbi</span> domains <span style="font-family: monospace">:3</span></p>
+
+	<form method="POST" action="/login">
+		<input type="text" name="user" placeholder="username" required autofocus /><br />
+		<input type="password" name="pass" placeholder="password" required /><br />
+		<button type="submit">log in</button>
+	</form>
+</body>
+</html>