78 lines
2.7 KiB
Python
78 lines
2.7 KiB
Python
import sqlite3, datetime, secrets, bcrypt, os
|
|
from config import DB_PATH, ADMIN_USERNAME
|
|
|
|
def get_db():
|
|
conn = sqlite3.connect(DB_PATH, timeout=30)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.execute("PRAGMA busy_timeout=30000")
|
|
conn.execute("PRAGMA foreign_keys=ON")
|
|
return conn
|
|
|
|
def init_db():
|
|
conn = get_db()
|
|
conn.executescript("""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
username TEXT UNIQUE NOT NULL,
|
|
password_hash TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
CREATE TABLE IF NOT EXISTS domains (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT UNIQUE NOT NULL,
|
|
description TEXT,
|
|
status TEXT DEFAULT 'active',
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
created_by INTEGER REFERENCES users(id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS certificates (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
domain_id INTEGER REFERENCES domains(id),
|
|
subject TEXT NOT NULL,
|
|
san TEXT,
|
|
serial TEXT UNIQUE,
|
|
status TEXT DEFAULT 'pending',
|
|
cert_path TEXT,
|
|
issued_at TIMESTAMP,
|
|
expires_at TIMESTAMP,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
created_by INTEGER REFERENCES users(id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS api_keys (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL,
|
|
key_hash TEXT UNIQUE NOT NULL,
|
|
prefix TEXT NOT NULL,
|
|
permissions TEXT DEFAULT 'read',
|
|
active BOOLEAN DEFAULT 1,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
created_by INTEGER REFERENCES users(id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS audit_log (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
action TEXT NOT NULL,
|
|
details TEXT,
|
|
user_id INTEGER REFERENCES users(id),
|
|
ip_address TEXT,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
""")
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT id FROM users WHERE username = ?", (ADMIN_USERNAME,))
|
|
if not cur.fetchone():
|
|
pw_hash = bcrypt.hashpw(b"CHANGE_ME_ADMIN_PASS", bcrypt.gensalt())
|
|
if isinstance(pw_hash, bytes): pw_hash = pw_hash.decode()
|
|
cur.execute("INSERT INTO users (username, password_hash) VALUES (?, ?)",
|
|
(ADMIN_USERNAME, pw_hash))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def hash_password(password):
|
|
h = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
|
|
return h.decode() if isinstance(h, bytes) else h
|
|
|
|
def verify_password(password, hash_):
|
|
if isinstance(hash_, str): hash_ = hash_.encode()
|
|
return bcrypt.checkpw(password.encode(), hash_)
|