finish rewrite

app/auth.py

@@ -1,413 +1,35 @@
-import sqlite3
+import db
-from os import urandom
 from hashlib import sha256
 
-def init_db(path: str) -> None:
+# Check if access token can update domain's ip
-    with sqlite3.connect(path) as connection:
-        cursor = connection.cursor()
-        cursor.execute('''
-            CREATE TABLE IF NOT EXISTS Tokens (
-                TKHASH TEXT PRIMARY KEY NOT NULL,
-                USERID INTEGER NOT NULL,
-                NAME TEXT NOT NULL
-            )
-        ''')
-
-        cursor.execute('''
-            CREATE TABLE IF NOT EXISTS Domains (
-                DOMAIN TEXT PRIMARY KEY NOT NULL,
-                USERID INTEGER NOT NULL
-            )
-        ''')
-
-        cursor.execute('''
-            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
-            )
-        ''')
-
-        cursor.execute('''
-            SELECT COUNT(*) FROM Users
-        ''')
-
-        if cursor.fetchone()[0] == 0:
-            username = "root"
-            password = urandom(8).hex()
-
-            create_user(path, username, password, is_admin=True)
-
-            print(f"No users detected, created new admin\nUser: {username}\nPass: {password}", flush=True)
-
 def check_token_domain(db_path: str, token: str, domain: str) -> bool:
-    with sqlite3.connect(db_path) as connection:
+    userid = db.Token.get_user(db_path, token)
-        cursor = connection.cursor()
 
-        token_hash = sha256(token.encode('utf-8')).hexdigest()
-
-        cursor.execute('''
-            SELECT COUNT(*) FROM Tokens
-            JOIN Domains ON Tokens.userid = Domains.userid
-            WHERE tkhash = ?
-                AND 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()
-
-        if email is not None:
-            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 == "":
+    return domain in db.User.get_domains(db_path, userid)
-                continue
 
-            cursor.execute('''
+# Verify user's passowrd for login. Return UserID on success, None on failure
-                SELECT COUNT(*) FROM Domains
+def user_login(db_path: str, username: str, password: str) -> int | None:
-                WHERE domain = ?
+    try:
-            ''', (domain, ))
+        userid = db.User.get_id(db_path, username)
+    except ValueError:
+        return None
 
-            if cursor.fetchone()[0] != 0:
+    pwsalt, pwhash = db.User.get_password_hash(db_path, userid)
-                continue
 
-            cursor.execute('''
+    password_hash = sha256( (pwsalt + password).encode('utf-8')).hexdigest()
-                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]:
+    if pwhash == password_hash:
-    with sqlite3.connect(db_path) as connection:
+        return userid
-        cursor = connection.cursor()
 
-        cursor.execute('''
+    return None
-            SELECT COUNT(*) FROM Users
-            WHERE userid = ?
-        ''', (userid, ))
 
-        if cursor.fetchone()[0] != 1:
+# Change password of user (by user)
-            return False, "Uerid not found"
+def change_user_password(db_path: str, userid: int, oldpass: str, newpass: str) -> None:
+    username = db.User.get_username(db_path, userid)
 
-        if domains is not None:
+    if user_login(db_path, username, oldpass) is None:
-            # crude check if domains is iterable
+        raise ValueError("Incorrect password")
-            for domain in domains:
-                break
 
-            cursor.execute('''
+    db.User.set_password(db_path, userid, newpass)
-                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/db.py

@@ -0,0 +1,432 @@
+import sqlite3
+from os import urandom
+from hashlib import sha256
+
+# Initialize database
+def init(db_path: str) -> None:
+    with sqlite3.connect(db_path) as connection:
+        cursor = connection.cursor()
+
+        # Auth token table
+        cursor.execute('''
+            CREATE TABLE IF NOT EXISTS Tokens (
+                TKHASH TEXT PRIMARY KEY NOT NULL,
+                USERID INTEGER NOT NULL,
+                NAME TEXT NOT NULL
+            )
+        ''')
+
+        # Registered domains table
+        cursor.execute('''
+            CREATE TABLE IF NOT EXISTS Domains (
+                DOMAIN TEXT PRIMARY KEY NOT NULL,
+                USERID INTEGER NOT NULL
+            )
+        ''')
+
+        # Users table
+        cursor.execute('''
+            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
+            )
+        ''')
+
+        # Create a new admin if there are no accounts
+        cursor.execute('''
+            SELECT COUNT(*) FROM Users
+        ''')
+
+        if cursor.fetchone()[0] == 0:
+            username = "root"
+            password = urandom(8).hex()
+
+            User.new(db_path, username, password, is_admin=True)
+
+            print(f"No users detected, created new admin\nUser: {username}\nPass: {password}", flush=True)
+
+
+def list_users(db_path: str) -> list[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": 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
+
+
+# Collection of getters and setters
+class User:
+    def exists(db_path: str, userid: int) -> bool:
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            cursor.execute('''
+                SELECT COUNT(*) FROM Users
+                WHERE userid = ?
+            ''', (userid, ))
+
+            return cursor.fetchone()[0] == 1
+
+
+    def new(db_path: str, username: str, password: str, domains: list = [], is_admin: bool = False, email: str = None) -> int:
+        username = username.strip()
+
+        if not username.isalnum():
+            raise ValueError("Username has to be alphanumeric")
+
+        if len(username) < 1:
+            raise ValueError("Username has to be at least 1 character long")
+
+        if len(password) > 16:
+            raise ValueError("Username has to be at most 16 characters long")
+
+        if len(password) < 8:
+            raise ValueError("Password has to be at least 8 characters long")
+
+        if len(password) > 128:
+            raise ValueError("Password has to be at most 128 characters long")
+
+        if email is not None:
+            email = email.strip()
+
+            if len(email) > 128:
+                raise ValueError("Email has to be at most 128 characters long, be reasonable")
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            pwsalt = urandom(16).hex()
+            pwhash = sha256( (pwsalt + password).encode('utf-8') ).hexdigest()
+
+            cursor.execute('''
+                INSERT INTO Users (username, pwsalt, pwhash, email, is_admin) VALUES (?, ?, ?, ?, ?)
+            ''', (username, pwsalt, pwhash, email, is_admin))
+
+            # Get new user's ID
+            cursor.execute('''
+                SELECT userid FROM Users
+                WHERE username = ?
+            ''', (username, ))
+
+            userid = cursor.fetchone()[0]
+
+        User.set_domains(db_path, userid, domains)
+
+        return userid
+
+
+    def delete(db_path: str, userid: int):
+        if not User.exists(db_path, userid):
+            raise ValueError(f"User {userid} does not exist")
+
+        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, ))
+
+
+    def is_admin(db_path: str, userid: int) -> bool:
+        if not User.exists(db_path, userid):
+            raise ValueError(f"User {userid} does not exist")
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            cursor.execute('''
+                SELECT is_admin FROM Users
+                WHERE userid = ?
+            ''', (userid, ))
+
+            return cursor.fetchone()[0]
+
+    def set_admin(db_path: str, userid: int, is_admin: bool):
+        if not User.exists(db_path, userid):
+            raise ValueError(f"User {userid} does not exist")
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            cursor.execute('''
+                UPDATE Users
+                SET is_admin = ?
+                WHERE userid = ?
+            ''', (is_admin, userid))
+
+    def get_username(db_path: str, userid: int) -> str:
+        if not User.exists(db_path, userid):
+            raise ValueError(f"User {userid} does not exist")
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            cursor.execute('''
+                SELECT username FROM Users
+                WHERE userid = ?
+            ''', (userid, ))
+
+            return cursor.fetchone()[0]
+
+
+    def get_id(db_path: str, username: str) -> int:
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            cursor.execute('''
+                SELECT userid FROM Users
+                WHERE username = ?
+            ''', (username, ))
+
+            userid = cursor.fetchone()
+
+            if userid is None:
+                raise ValueError(f"User {username} does not exist")
+
+            return userid[0]
+
+    def get_domains(db_path: str, userid: int) -> list[str]:
+        if not User.exists(db_path, userid):
+            raise ValueError(f"User {userid} does not exist")
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            cursor.execute('''
+                SELECT domain FROM Domains
+                WHERE userid = ?
+            ''', (userid, ))
+
+            domains = cursor.fetchall()
+            domains = list(map(lambda x: x[0], domains))
+
+            return domains
+
+
+    def set_domains(db_path: str, userid: int, domains: list[str]) -> None:
+        if not User.exists(db_path, userid):
+            raise ValueError(f"User {userid} does not exist")
+
+        #TODO domain regex
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            cursor.execute('''
+                DELETE FROM Domains
+                WHERE userid = ?
+            ''', (userid, ))
+
+            for domain in domains:
+                domain = domain.strip()
+
+                if len(domain) < 1:
+                    continue
+
+                cursor.execute('''
+                    INSERT INTO Domains (domain, userid) VALUES (?, ?)
+                ''', (domain, userid))
+
+
+    def get_token_names(db_path: str, userid: int) -> list[str]:
+        if not User.exists(db_path, userid):
+            raise ValueError(f"User {userid} does not exist")
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            cursor.execute('''
+                SELECT name FROM Tokens
+                WHERE userid = ?
+            ''', (userid, ))
+
+            token_names = cursor.fetchall()
+            token_names = list(map(lambda x: x[0], token_names))
+
+            return token_names
+
+
+    def generate_token(db_path: str, userid: int, token_name: str) -> str | None:
+        if not User.exists(db_path, userid):
+            raise ValueError(f"User {userid} does not exist")
+
+        if len(token_name) < 1:
+            raise ValueError("Token name has to be at least 1 character long")
+
+        if len(token_name) > 16:
+            raise ValueError("Token name has to be at most 16 characters long")
+
+        if not token_name.isalnum():
+            raise ValueError("Token name has to be alphanumeric")
+
+        token_name_list = User.get_token_names(db_path, userid)
+
+        if len(token_name_list) >= 5:
+            raise ValueError("One user can have at most 5 tokens")
+
+        if token_name in token_name_list:
+            raise ValueError("Token name is not unique")
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            # Try 10 times in case of collisions
+            for _ in range(10):
+                token = urandom(32).hex()
+                token_hash = sha256(token.encode("utf-8")).hexdigest()
+
+                try:
+                    cursor.execute('''
+                        INSERT INTO Tokens (tkhash, userid, name) VALUES (?, ?, ?)
+                    ''', (token_hash, userid, token_name))
+                except sqlite3.IntegrityError:
+                    continue
+
+                return token
+
+            return None
+
+
+    def revoke_token(db_path: str, userid: int, token_name: str) -> None:
+        if not User.exists(db_path, userid):
+            raise ValueError(f"User {userid} does not exist")
+
+        if token_name not in User.get_token_names(db_path, userid):
+            raise ValueError("Token name not found")
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            cursor.execute('''
+                DELETE FROM Tokens
+                WHERE userid = ?
+                    AND name = ?
+            ''', (userid, token_name))
+
+
+    def get_password_hash(db_path: str, userid: int) -> tuple[str, str]:
+        if not User.exists(db_path, userid):
+            raise ValueError(f"User {userid} does not exist")
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            cursor.execute('''
+                SELECT pwsalt, pwhash FROM Users
+                WHERE userid = ?
+            ''', (userid, ))
+
+            return cursor.fetchone()
+
+
+    def set_password(db_path: str, userid: int, password: str) -> None:
+        if not User.exists(db_path, userid):
+            raise ValueError(f"User {userid} does not exist")
+
+        if len(password) < 8:
+            raise ValueError("Password has to be at least 8 characters long")
+
+        if len(password) > 128:
+            raise ValueError("Password has to be at most 128 characters long")
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            pwsalt = urandom(16).hex()
+            pwhash = sha256( (pwsalt + password).encode('utf-8') ).hexdigest()
+
+            cursor.execute('''
+                UPdATE Users
+                SET pwsalt = ?,
+                    pwhash = ?
+                WHERE userid = ?
+            ''', (pwsalt, pwhash, userid))
+
+    def get_email(db_path: str, userid: int) -> str:
+        if not User.exists(db_path, userid):
+            raise ValueError(f"User {userid} does not exist")
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            cursor.execute('''
+                SELECT email FROM Users
+                WHERE userid = ?
+            ''', (userid, ))
+
+            return cursor.fetchone()[0]
+
+    def set_email(db_path: str, userid: int, email: str | None) -> None:
+        if not User.exists(db_path, userid):
+            raise ValueError(f"User {userid} does not exist")
+
+        if email is not None:
+            email = email.strip()
+
+            if len(email) > 128:
+                raise ValueError("Email has to be at most 128 characters long, be reasonable")
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            cursor.execute('''
+                UPDATE Users
+                SET email = ?
+                WHERE userid = ?
+            ''', (email, userid, ))
+
+
+# Collection of getters
+class Token:
+    def exists(db_path: str, token: str) -> bool:
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            tkhash = sha256(token.encode('utf-8')).hexdigest()
+
+            cursor.execute('''
+                SELECT COUNT(*) FROM Tokens
+                WHERE tkhash = ?
+            ''', (tkhash, ))
+
+            return cursor.fetchone()[0] == 1
+
+
+    def get_user(db_path: str, token: str) -> int:
+        if not Token.exists(db_path, userid):
+            raise ValueError("Token does not exist")
+
+        with sqlite3.connect(db_path) as connection:
+            cursor = connection.cursor()
+
+            tkhash = sha256(token.encode('utf-8')).hexdigest()
+
+            cursor.execute('''
+                SELECT userid FROM Tokens
+                WHERE tkhash = ?
+            ''', (tkhash, ))
+
+            return cursor.fetcone()[0]

app/main.py

@@ -1,25 +1,23 @@
 from flask import Flask, jsonify, request, render_template, redirect, session, flash
 import auth
+import db
 import re
 from os import environ, urandom
 
 
-
 HOSTS_PATH = ( environ["HOSTS"] if "HOSTS" in environ else"/app/data/hosts.d/" ) + "/izbi"
 DB_PATH = environ["DB_PATH"] if "DB_PATH" in environ else "/app/data/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"] = environ["SECRET_KEY"] if "SECRET_KEY" in environ else urandom(32).hex()
 
 hosts = {}
 
 
+# Update hosts file
 def update_hosts(path: str, hosts: dict = {}) -> bool:
     try:
         hosts_str = ""
@@ -34,13 +32,13 @@ def update_hosts(path: str, hosts: dict = {}) -> bool:
         return False
 
 
-
+# check if string is a valid IP
 def validate_ip(ip: str) -> bool:
     exp = r"^((25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])\.){3}(25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])$"
     return re.search(exp, ip)
 
 
-
+# Serve homepage
 @app.route('/', methods=["GET"])
 def homepage():
     if session.get("USERID") is not None:
@@ -48,57 +46,62 @@ def homepage():
     return render_template("index.html")
 
 
-
+# Handle login
 @app.route('/login', methods=["POST"])
 def login():
-    if "user" not in request.form or "pass" not in request.form:
+    userid = auth.user_login(DB_PATH, request.form.get("user"), request.form.get("pass"))
-        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:
+        flash("Invalid user or password")
         return redirect("/")
 
     session["USERID"] = userid
 
-    return redirect("/dashboard", code=302)
+    return redirect("/dashboard")
 
+
+# Log out by clearing session data
 @app.route('/logout', methods=["GET"])
 def logout():
     session.clear()
     return redirect("/")
 
 
-
+# Display dashboard for logged in users
 @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):
+    userid = session.get("USERID")
-        users = auth.get_users(DB_PATH)
+
+    username = db.User.get_username(DB_PATH, userid)
+    domains = db.User.get_domains(DB_PATH, userid)
+    tokens = db.User.get_token_names(DB_PATH, userid)
+
+    # If user is admin, display additional options
+    if db.User.is_admin(DB_PATH, userid):
+        users = db.list_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)
 
 
-
+# Generate new access token with a name
 @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")
+        flash("Token name not supplied")
-    if request.form["token"].strip() == "":
         return redirect("/dashboard")
 
-    token = auth.generate_user_token(DB_PATH, session.get("USERID"), request.form["token"])
+    try:
+        token = db.User.generate_token(DB_PATH, session.get("USERID"), request.form["token"])
+    except ValueError as e:
+        flash(str(e))
+        token = None
 
     if token is not None:
         flash(token)
@@ -106,29 +109,34 @@ def generate_token():
     return redirect("/dashboard")
 
 
+# Revoke token by name
 @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")
+        flash("Token name not supplied")
-    if request.form["token"].strip() == "":
         return redirect("/dashboard")
 
-    auth.revoke_user_token(DB_PATH, session.get("USERID"), request.form["token"])
+    try:
+        db.User.revoke_token(DB_PATH, session.get("USERID"), request.form["token"])
+    except ValueError as e:
+        flash(str(e))
 
     return redirect("/dashboard")
 
 
+# Create new user (by admin)
 @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")):
+    if not db.User.is_admin(DB_PATH, session.get("USERID")):
         return redirect("/dashboard")
 
     if 'username' not in request.form or 'password' not in request.form:
+        flash('Required fields not supplied')
         return redirect("/dashboard")
 
     username = request.form.get("username").strip()
@@ -138,109 +146,134 @@ def create_user():
     if domains is None:
         domains = []
     else:
+        # Collapse whitespaces
         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
+    is_admin = request.form.get("is_admin")
+    if is_admin is not None:
+        is_admin = is_admin == "true"
+    else:
+        is_admin = False
 
-    flash(auth.create_user(DB_PATH, username, password, domains, is_admin, request.form.get('email')))
+    email = request.form.get("email")
+    if email is not None:
+        if email == "":
+            email = None
+
+    try:
+        db.User.new(DB_PATH, username, password, domains, is_admin, email)
+    except ValueError as e:
+        flash(str(e))
 
     return redirect("/dashboard")
 
 
+# Update user's field (by admin)
 @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")):
+    if not db.User.is_admin(DB_PATH, session.get("USERID")):
         return redirect("/dashboard")
 
     userid = request.form.get("userid")
     if userid is None:
+        flash("UserID not supplied")
         return redirect("/dashboard")
-    try:
+
-        userid = int(userid)
+    if not db.User.exists(DB_PATH, userid):
-    except ValueError:
+        flash("User does not exist")
         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(" ")
+        try:
+            db.User.set_domains(DB_PATH, userid, domains)
+        except ValueError as e:
+            flash(str(e))
 
     password = request.form.get("password")
+    if password is not None:
+        try:
+            db.User.set_password(DB_PATH, userid, password)
+        except ValueError as e:
+            flash(str(e))
 
     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)
+        db.User.set_admin(DB_PATH, userid, is_admin)
 
     email = request.form.get("email")
-
+    if email is not None:
-    auth.update_user(DB_PATH, userid, domains, password, is_admin, email)
+        if email == "":
+            email = None
+        db.User.set_email(DB_PATH, userid, email)
 
     return redirect("/dashboard")
 
 
-
+# Delete user (by admin)
 @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")):
+    if not db.User.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)
+    if db.User.exists(DB_PATH, userid):
-    except ValueError:
+        flash("User does not exst")
         return redirect("/dashboard")
 
-    auth.delete_user(DB_PATH, userid)
+    db.User.delete(DB_PATH, userid)
 
     return redirect("/dashboard")
 
 
-
+# Change password (by user)
-
 @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")
+        flash("Required values not supplied")
-        return redirect("/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"]
+    reppass = request.form["pass-rep"]
 
-    if newpass != request.form["pass-rep"]:
+    if newpass != reppass:
-        flash("passwords do not match")
+        flash("Passwords do not match")
         return redirect("/dashboard")
 
-    if not auth.change_user_password(DB_PATH, session.get("USERID"), oldpass, newpass):
+    try:
-        flash("password change failed")
+        auth.change_user_password(DB_PATH, session.get("USERID"), oldpass, newpass)
+    except ValueError as e:
+        flash(str(e))
         return redirect("/dashboard")
 
-    flash("password changed successfully")
+    flash("Password changed successfully")
     return redirect("/dashboard")
 
+
+# API health endpoint ???
 @app.route('/health', methods=["GET"])
 def health():
     return jsonify({"status": "healthy", "comment": ""})
 
 
-
+# API for updating domain's IP
 @app.route('/update', methods=["GET"])
 def update_addr():
     try:
@@ -250,10 +283,10 @@ def update_addr():
 
         if 'ip' not in request.args:
             ip = request.remote_addr
-        elif not validate_ip(request.args['ip']):
+        elif validate_ip(request.args['ip']):
-            return jsonify({"status": "400", "code": "ip-error", "comment": "Invalid IP"}), 400
-        else:
             ip = request.args['ip']
+        else:
+            return jsonify({"status": "400", "code": "ip-error", "comment": "Invalid IP"}), 400
 
         if 'token' not in request.args:
             return jsonify({"status": "400", "code": "token-error", "comment": "Missing or invalid token"}), 400
@@ -265,6 +298,7 @@ def update_addr():
 
         success = True
 
+        # Update hosts file if something changed
         if domain not in hosts or ip != hosts[domain]:
             hosts[domain] = ip
             success = update_hosts(HOSTS_PATH, hosts)
@@ -278,7 +312,6 @@ def update_addr():
         return jsonify({"status": "500", "code": "internal-server-error", "comment": "An error occurred while processing request"}), 500
 
 
-
 if __name__ == '__main__':
-    auth.init_db(DB_PATH)
+    db.init(DB_PATH)
     app.run(host="0.0.0.0", port=8080, debug=False)

app/templates/admin.html

@@ -36,7 +36,7 @@
 				</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"] }}"/>
+					<input type="email" name="email" placeholder="email" value="{{ user["email"] if user["email"] }}"/>
 					<button type="submit">update email</button>
 				</form></td>
 				<td><form method="POST" action="/delete-user">

app/templates/index.html

@@ -6,6 +6,14 @@
 <body>
 	<p>Dynamic DNS service for <span style="font-family: monospace">.izbi</span> domains <span style="font-family: monospace">:3</span></p>
 
+{% with messages = get_flashed_messages()%}
+{% if messages%}
+{% for message in messages%}
+        <h3>{{message}}</h3>
+{%endfor%}
+{%endif%}
+{%endwith%}
+
 	<form method="POST" action="/login">
 		<input type="text" name="user" placeholder="username" required autofocus /><br />
 		<input type="password" name="pass" placeholder="password" required /><br />