Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 wiktr
|
0bb7ca21cf | ||
|
wiktr
|
cfa862d4e5 | ||
|
wiktr
|
8e143acb1e | ||
|
wiktr
|
1b5ea7be3a |
No files matched your search
+21
-399
@@ -1,413 +1,35 @@
|
||||
import sqlite3
|
||||
from os import urandom
|
||||
import db
|
||||
from hashlib import sha256
|
||||
|
||||
def init_db(path: str) -> None:
|
||||
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)
|
||||
|
||||
# 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()
|
||||
userid = db.Token.get_user(db_path, token)
|
||||
|
||||
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 == "":
|
||||
continue
|
||||
return domain in db.User.get_domains(db_path, userid)
|
||||
|
||||
cursor.execute('''
|
||||
SELECT COUNT(*) FROM Domains
|
||||
WHERE domain = ?
|
||||
''', (domain, ))
|
||||
# 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:
|
||||
try:
|
||||
userid = db.User.get_id(db_path, username)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
if cursor.fetchone()[0] != 0:
|
||||
continue
|
||||
pwsalt, pwhash = db.User.get_password_hash(db_path, userid)
|
||||
|
||||
cursor.execute('''
|
||||
INSERT INTO Domains (userid, domain) VALUES (?, ?)
|
||||
''', (userid, domain))
|
||||
password_hash = sha256( (pwsalt + password).encode('utf-8')).hexdigest()
|
||||
|
||||
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()
|
||||
if pwhash == password_hash:
|
||||
return userid
|
||||
|
||||
cursor.execute('''
|
||||
SELECT COUNT(*) FROM Users
|
||||
WHERE userid = ?
|
||||
''', (userid, ))
|
||||
return None
|
||||
|
||||
if cursor.fetchone()[0] != 1:
|
||||
return False, "Uerid not found"
|
||||
# Change password of user (by user)
|
||||
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:
|
||||
# crude check if domains is iterable
|
||||
for domain in domains:
|
||||
break
|
||||
if user_login(db_path, username, oldpass) is None:
|
||||
raise ValueError("Incorrect password")
|
||||
|
||||
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,))
|
||||
db.User.set_password(db_path, userid, newpass)
|
||||
@@ -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]
|
||||
+96
-63
@@ -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:
|
||||
return redirect("/"), 400
|
||||
userid = auth.user_login(DB_PATH, request.form.get("user"), request.form.get("pass"))
|
||||
|
||||
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):
|
||||
users = auth.get_users(DB_PATH)
|
||||
userid = session.get("USERID")
|
||||
|
||||
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")
|
||||
if request.form["token"].strip() == "":
|
||||
flash("Token name not supplied")
|
||||
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")
|
||||
if request.form["token"].strip() == "":
|
||||
flash("Token name not supplied")
|
||||
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)
|
||||
except ValueError:
|
||||
|
||||
if not db.User.exists(DB_PATH, userid):
|
||||
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")
|
||||
|
||||
auth.update_user(DB_PATH, userid, domains, password, is_admin, email)
|
||||
if email is not None:
|
||||
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)
|
||||
except ValueError:
|
||||
|
||||
if db.User.exists(DB_PATH, userid):
|
||||
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")
|
||||
return redirect("/dashboard")
|
||||
if request.form["pass"] == "" or request.form["pass-new"] == "" or request.form["pass-rep"] == "":
|
||||
flash("password change failed")
|
||||
flash("Required values not supplied")
|
||||
return redirect("/dashboard")
|
||||
|
||||
oldpass = request.form["pass"]
|
||||
newpass = request.form["pass-new"]
|
||||
reppass = request.form["pass-rep"]
|
||||
|
||||
if newpass != request.form["pass-rep"]:
|
||||
flash("passwords do not match")
|
||||
if newpass != reppass:
|
||||
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")
|
||||
try:
|
||||
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']):
|
||||
return jsonify({"status": "400", "code": "ip-error", "comment": "Invalid IP"}), 400
|
||||
else:
|
||||
elif validate_ip(request.args['ip']):
|
||||
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)
|
||||
@@ -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">
|
||||
|
||||
@@ -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 />
|
||||
|
||||
Reference in new issue
Block a user