Add comments

This commit is contained in:
2026-08-25 01:48:03 +02:00
parent 07ff654baa
commit 1b5ea7be3a
2 changed files with 69 additions and 16 deletions
+49 -2
View File
@@ -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()