diff --git a/app/auth.py b/app/auth.py index b3e4584..669d5ec 100644 --- a/app/auth.py +++ b/app/auth.py @@ -2,9 +2,13 @@ import sqlite3 from os import urandom from hashlib import sha256 + +# Initialize database def init_db(path: str) -> None: with sqlite3.connect(path) as connection: cursor = connection.cursor() + + # Auth token table cursor.execute(''' CREATE TABLE IF NOT EXISTS Tokens ( TKHASH TEXT PRIMARY KEY NOT NULL, @@ -13,6 +17,7 @@ def init_db(path: str) -> None: ) ''') + # Registered domains table cursor.execute(''' CREATE TABLE IF NOT EXISTS Domains ( DOMAIN TEXT PRIMARY KEY NOT NULL, @@ -20,6 +25,7 @@ def init_db(path: str) -> None: ) ''') + # Users table cursor.execute(''' CREATE TABLE IF NOT EXISTS Users ( USERID INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, @@ -31,6 +37,7 @@ def init_db(path: str) -> None: ) ''') + # Create a new admin if there are no accounts cursor.execute(''' SELECT COUNT(*) FROM Users ''') @@ -43,6 +50,8 @@ def init_db(path: str) -> None: print(f"No users detected, created new admin\nUser: {username}\nPass: {password}", flush=True) + +# Check if access token can update domain's ip def check_token_domain(db_path: str, token: str, domain: str) -> bool: with sqlite3.connect(db_path) as connection: cursor = connection.cursor() @@ -58,10 +67,13 @@ def check_token_domain(db_path: str, token: str, domain: str) -> bool: return cursor.fetchone()[0] == 1 + +# Verify user's passowrd for login. Return UserID on success, None on failure def user_login(db_path: str, username: str, password: str) -> int | None: with sqlite3.connect(db_path) as connection: cursor = connection.cursor() + # Get salt cursor.execute(''' SELECT pwsalt FROM Users WHERE username = ? @@ -72,8 +84,10 @@ def user_login(db_path: str, username: str, password: str) -> int | None: return None password_salt = password_salt[0] + # Calculate hash password_hash = sha256( (password_salt + password).encode('utf-8') ).hexdigest() + # Check cursor.execute(''' SELECT userid FROM Users WHERE username = ? @@ -86,6 +100,8 @@ def user_login(db_path: str, username: str, password: str) -> int | None: return result[0] + +# Get username of UserID. Return username if user exists, None if not def get_username(db_path: str, userid: int) -> str | None: with sqlite3.connect(db_path) as connection: cursor = connection.cursor() @@ -101,6 +117,8 @@ def get_username(db_path: str, userid: int) -> str | None: return username[0] + +# Get list of domains assigned to UserID def get_user_domains(db_path: str, userid: int) -> list[str]: with sqlite3.connect(db_path) as connection: cursor = connection.cursor() @@ -116,6 +134,8 @@ def get_user_domains(db_path: str, userid: int) -> list[str]: return domains + +# Get list of tokens assigned to UserID def get_user_tokens(db_path: str, userid: int) -> list[str]: with sqlite3.connect(db_path) as connection: cursor = connection.cursor() @@ -131,6 +151,8 @@ def get_user_tokens(db_path: str, userid: int) -> list[str]: return tokens + +# Generate a new user token with a name, assigned to UserID def generate_user_token(db_path: str, userid: int, token_name: str) -> str | None: with sqlite3.connect(db_path) as connection: cursor = connection.cursor() @@ -182,6 +204,7 @@ def generate_user_token(db_path: str, userid: int, token_name: str) -> str | Non return None +# Revoke access token def revoke_user_token(db_path: str, userid: int, token_name: str) -> None: with sqlite3.connect(db_path) as connection: cursor = connection.cursor() @@ -192,6 +215,8 @@ def revoke_user_token(db_path: str, userid: int, token_name: str) -> None: AND name = ? ''', (userid, token_name)) + +# Change password of user (by user) def change_user_password(db_path: str, userid: int, oldpass: str, newpass: str) -> bool: with sqlite3.connect(db_path) as connection: cursor = connection.cursor() @@ -228,6 +253,8 @@ def change_user_password(db_path: str, userid: int, oldpass: str, newpass: str) return set_user_password(db_path, userid, newpass) + +# Set password of user (by admin) def set_user_password(db_path: str, userid: int, newpass: str) -> bool: with sqlite3.connect(db_path) as connection: cursor = connection.cursor() @@ -253,6 +280,8 @@ def set_user_password(db_path: str, userid: int, newpass: str) -> bool: return True + +# Check if user is admin def is_admin(db_path: str, userid: int) -> bool: with sqlite3.connect(db_path) as connection: cursor = connection.cursor() @@ -265,6 +294,8 @@ def is_admin(db_path: str, userid: int) -> bool: return cursor.fetchone()[0] == 1 + +# Get list of users as dictionaries def get_users(db_path: str) -> dict: with sqlite3.connect(db_path) as connection: cursor = connection.cursor() @@ -291,15 +322,20 @@ def get_users(db_path: str) -> dict: return users + +# Create a new user 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() + # Sanitize username ??? username.strip() + # Sanitize email ??? if email is not None: email.strip() + # Check if username is free cursor.execute(''' SELECT COUNT(*) FROM Users WHERE username = ? @@ -308,6 +344,7 @@ def create_user(db_path: str, username: str, password: str, domains: list = [], if cursor.fetchone()[0] != 0: return "Username exists" + # Sanitize 2 electric boogaloo if username == "": return "Username is empty" @@ -320,13 +357,16 @@ def create_user(db_path: str, username: str, password: str, domains: list = [], if len(password) < 8: return "Password is shorter than 8 characters" + # Generate salt salt = urandom(16).hex() password_hash = sha256( (salt + password).encode('utf-8')).hexdigest() + # Add user to the database cursor.execute(''' INSERT INTO Users (username, pwsalt, pwhash, email, is_admin) VALUES (?, ?, ?, ?, ?) ''', (username, salt, password_hash, email, is_admin)) + # Get new user's ID cursor.execute(''' SELECT userid FROM Users WHERE username = ? @@ -334,6 +374,7 @@ def create_user(db_path: str, username: str, password: str, domains: list = [], userid = cursor.fetchone()[0] + # Add domains for domain in domains: domain = domain.strip() @@ -352,10 +393,13 @@ def create_user(db_path: str, username: str, password: str, domains: list = [], INSERT INTO Domains (userid, domain) VALUES (?, ?) ''', (userid, domain)) + +# Update user's field 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() + # Check if user exists cursor.execute(''' SELECT COUNT(*) FROM Users WHERE userid = ? @@ -369,6 +413,7 @@ def update_user(db_path: str, userid: int, domains: list[str] = None, password: for domain in domains: break + # Delete all domains and add the fresh list cursor.execute(''' DELETE FROM Domains WHERE userid = ? @@ -379,9 +424,11 @@ def update_user(db_path: str, userid: int, domains: list[str] = None, password: INSERT INTO Domains (userid, domain) VALUES (?, ?) ''', (userid, domain)) + # Update passowrd if password is not None: set_user_password(db_path, userid, password) + # Update if user is admin if is_admin is not None: cursor.execute(''' UPDATE Users @@ -390,7 +437,7 @@ def update_user(db_path: str, userid: int, domains: list[str] = None, password: WHERE userid = ? ''', (is_admin, userid)) - + # Update email ??? if email is not None: if email == "": email = None @@ -403,7 +450,7 @@ def update_user(db_path: str, userid: int, domains: list[str] = None, password: ''', (email, userid)) - +# Delete user adn all data def delete_user(db_path: str, userid: int): with sqlite3.connect(db_path) as connection: cursor = connection.cursor() diff --git a/app/main.py b/app/main.py index d348f6d..ec3ba88 100644 --- a/app/main.py +++ b/app/main.py @@ -4,22 +4,19 @@ 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 +31,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,7 +45,7 @@ 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: @@ -65,13 +62,15 @@ def login(): return redirect("/dashboard", code=302) + +# 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: @@ -81,13 +80,15 @@ def dashboard(): domains = auth.get_user_domains(DB_PATH, userid) tokens = auth.get_user_tokens(DB_PATH, userid) + # If user is admin, display additional options 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) - +# Generate new access token with a name @app.route('/generate-token', methods=["POST"]) def generate_token(): if session.get("USERID") is None: @@ -106,6 +107,7 @@ 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: @@ -121,6 +123,7 @@ def revoke_token(): return redirect("/dashboard") +# Create new user (by admin) @app.route('/create-user', methods=["POST"]) def create_user(): if session.get("USERID") is None: @@ -138,6 +141,7 @@ 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(" ") @@ -149,6 +153,7 @@ def create_user(): return redirect("/dashboard") +# Update user's field (by admin) @app.route('/update-user', methods=["POST"]) def update_user(): if session.get("USERID") is None: @@ -186,7 +191,7 @@ def update_user(): return redirect("/dashboard") - +# Delete user (by admin) @app.route('/delete-user', methods=["POST"]) def delete_user(): if session.get("USERID") is None: @@ -207,8 +212,7 @@ def delete_user(): return redirect("/dashboard") - - +# Change password (by user) @app.route('/change-password', methods=["POST"]) def change_password(): if session.get("USERID") is None: @@ -235,12 +239,14 @@ def change_password(): 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: @@ -265,6 +271,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 +285,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) app.run(host="0.0.0.0", port=8080, debug=False)