411 lines
16 KiB
Python
411 lines
16 KiB
Python
"""
|
|
config.py — Application configuration and DB connection manager.
|
|
Update DB_CONFIG with your remote MySQL server credentials.
|
|
"""
|
|
|
|
import mysql.connector
|
|
from mysql.connector import pooling
|
|
import logging
|
|
|
|
# ─── Logging Setup ────────────────────────────────────────────────────────────
|
|
import sys
|
|
|
|
_file_handler = logging.FileHandler("app.log", encoding="utf-8")
|
|
_stream_handler = logging.StreamHandler(stream=sys.stdout)
|
|
|
|
# Force UTF-8 on the stream so Windows cp1252 consoles don't choke on
|
|
# Unicode characters (arrows, em-dashes, etc.) in log messages.
|
|
if hasattr(_stream_handler.stream, "reconfigure"):
|
|
try:
|
|
_stream_handler.stream.reconfigure(encoding="utf-8", errors="replace")
|
|
except Exception:
|
|
pass
|
|
|
|
_formatter = logging.Formatter(
|
|
"%(asctime)s [%(levelname)s] %(name)s - %(message)s"
|
|
)
|
|
_file_handler.setFormatter(_formatter)
|
|
_stream_handler.setFormatter(_formatter)
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
handlers=[_file_handler, _stream_handler],
|
|
)
|
|
logger = logging.getLogger("config")
|
|
|
|
# ─── Config File Load / Save ──────────────────────────────────────────────────
|
|
import configparser as _cp
|
|
import os as _os
|
|
|
|
CONFIG_FILE = "config.ini"
|
|
APP_TITLE = "Website Checker"
|
|
APP_VERSION = "1.0.0"
|
|
|
|
|
|
# Sensitive fields that are DPAPI-encrypted in config.ini
|
|
_DB_SENSITIVE = {"user", "password"}
|
|
|
|
|
|
def load_config() -> dict:
|
|
"""
|
|
Load DB settings from config.ini.
|
|
Returns a dict with keys: host, port, database, user, password.
|
|
Sensitive fields (user, password) are decrypted transparently via DPAPI.
|
|
Returns empty dict if the file does not exist or is incomplete.
|
|
"""
|
|
from utils.config_crypto import decrypt_value
|
|
cfg = _cp.ConfigParser()
|
|
if not _os.path.exists(CONFIG_FILE):
|
|
return {}
|
|
cfg.read(CONFIG_FILE, encoding="utf-8")
|
|
if "database" not in cfg:
|
|
return {}
|
|
section = cfg["database"]
|
|
try:
|
|
return {
|
|
"host": section.get("host", ""),
|
|
"port": section.getint("port", 3306),
|
|
"database": section.get("database", ""),
|
|
"user": decrypt_value(section.get("user", "")),
|
|
"password": decrypt_value(section.get("password", "")),
|
|
}
|
|
except RuntimeError as exc:
|
|
logger.error(f"Failed to decrypt DB credentials: {exc}")
|
|
raise
|
|
|
|
|
|
def save_config(host: str, port: int, database: str, user: str, password: str):
|
|
"""Persist DB connection settings to config.ini (sensitive fields DPAPI-encrypted)."""
|
|
from utils.config_crypto import encrypt_value
|
|
# Preserve any existing non-database sections (email, crypto, groq)
|
|
cfg = _cp.ConfigParser()
|
|
if _os.path.exists(CONFIG_FILE):
|
|
cfg.read(CONFIG_FILE, encoding="utf-8")
|
|
cfg["database"] = {
|
|
"host": host,
|
|
"port": str(port),
|
|
"database": database,
|
|
"user": encrypt_value(user),
|
|
"password": encrypt_value(password),
|
|
}
|
|
with open(CONFIG_FILE, "w", encoding="utf-8") as fh:
|
|
cfg.write(fh)
|
|
logger.info(f"Configuration saved to {CONFIG_FILE} (credentials encrypted).")
|
|
|
|
|
|
def migrate_plaintext_config():
|
|
"""
|
|
One-time migration: if config.ini contains plain-text DB credentials
|
|
(no 'dpapi:' prefix) encrypt them in-place using Windows DPAPI.
|
|
Safe to call on every startup — is a no-op when already encrypted.
|
|
"""
|
|
from utils.config_crypto import encrypt_value, is_encrypted
|
|
if not _os.path.exists(CONFIG_FILE):
|
|
return
|
|
cfg = _cp.ConfigParser()
|
|
cfg.read(CONFIG_FILE, encoding="utf-8")
|
|
changed = False
|
|
|
|
# DB section
|
|
for key in ("user", "password"):
|
|
if cfg.has_option("database", key):
|
|
raw = cfg.get("database", key)
|
|
if raw and not is_encrypted(raw):
|
|
cfg.set("database", key, encrypt_value(raw))
|
|
changed = True
|
|
|
|
# Email section
|
|
if cfg.has_option("email", "smtp_password"):
|
|
raw = cfg.get("email", "smtp_password")
|
|
if raw and not is_encrypted(raw):
|
|
cfg.set("email", "smtp_password", encrypt_value(raw))
|
|
changed = True
|
|
|
|
# Groq section
|
|
if cfg.has_option("groq", "api_key"):
|
|
raw = cfg.get("groq", "api_key")
|
|
if raw and not is_encrypted(raw):
|
|
cfg.set("groq", "api_key", encrypt_value(raw))
|
|
changed = True
|
|
|
|
if changed:
|
|
with open(CONFIG_FILE, "w", encoding="utf-8") as fh:
|
|
cfg.write(fh)
|
|
logger.info("config.ini: plain-text credentials encrypted with Windows DPAPI.")
|
|
|
|
|
|
def config_exists() -> bool:
|
|
"""Return True if config.ini contains all required connection fields."""
|
|
ini = load_config()
|
|
return bool(ini.get("host") and ini.get("database") and ini.get("user"))
|
|
|
|
|
|
def reload_db_config():
|
|
"""
|
|
Re-read config.ini and update DB_CONFIG in place.
|
|
Also resets the connection pool so the next get_connection() uses new creds.
|
|
"""
|
|
global _pool, DB_CONFIG
|
|
ini = load_config()
|
|
DB_CONFIG.update({
|
|
"host": ini.get("host", DB_CONFIG["host"]),
|
|
"port": ini.get("port", DB_CONFIG["port"]),
|
|
"database": ini.get("database", DB_CONFIG["database"]),
|
|
"user": ini.get("user", DB_CONFIG["user"]),
|
|
"password": ini.get("password", DB_CONFIG["password"]),
|
|
})
|
|
_pool = None # force pool recreation on next connection
|
|
logger.info("DB_CONFIG reloaded from config.ini.")
|
|
|
|
|
|
# ─── Database Configuration ───────────────────────────────────────────────────
|
|
# Populated from config.ini at runtime; falls back to placeholder strings so
|
|
# the module is importable even before first-run setup has completed.
|
|
_ini = load_config()
|
|
DB_CONFIG = {
|
|
"host": _ini.get("host", "your-mysql-host"),
|
|
"port": _ini.get("port", 3306),
|
|
"database": _ini.get("database", "website_checker"),
|
|
"user": _ini.get("user", "your-db-user"),
|
|
"password": _ini.get("password", "your-db-password"),
|
|
"connection_timeout": 10,
|
|
}
|
|
|
|
# ─── Connection Pool ──────────────────────────────────────────────────────────
|
|
_pool = None
|
|
|
|
def get_connection_pool():
|
|
global _pool
|
|
if _pool is None:
|
|
try:
|
|
_pool = pooling.MySQLConnectionPool(
|
|
pool_name="app_pool",
|
|
pool_size=5,
|
|
**DB_CONFIG
|
|
)
|
|
logger.info("Database connection pool initialised.")
|
|
except mysql.connector.Error as e:
|
|
logger.error(f"Failed to create connection pool: {e}")
|
|
raise
|
|
return _pool
|
|
|
|
|
|
def get_connection():
|
|
"""Return a connection from the pool."""
|
|
return get_connection_pool().get_connection()
|
|
|
|
|
|
def initialize_database():
|
|
"""Create all required tables if they do not exist."""
|
|
ddl_statements = [
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
username VARCHAR(100) NOT NULL UNIQUE,
|
|
password VARCHAR(255) NOT NULL,
|
|
role ENUM('admin','user') NOT NULL DEFAULT 'user',
|
|
full_name VARCHAR(200),
|
|
is_active TINYINT(1) NOT NULL DEFAULT 1,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS websites (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
name VARCHAR(200) NOT NULL,
|
|
url TEXT NOT NULL,
|
|
check_type ENUM('daily','weekly') NOT NULL DEFAULT 'daily',
|
|
note TEXT,
|
|
is_active TINYINT(1) NOT NULL DEFAULT 1,
|
|
created_by INT,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (created_by) REFERENCES users(id) ON DELETE SET NULL
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS website_credentials (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
website_id INT NOT NULL,
|
|
username VARCHAR(200) NOT NULL,
|
|
password VARCHAR(255) NOT NULL,
|
|
label VARCHAR(100),
|
|
FOREIGN KEY (website_id) REFERENCES websites(id) ON DELETE CASCADE
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS shift_checks (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
website_id INT NOT NULL,
|
|
user_id INT NOT NULL,
|
|
checked_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
user_note TEXT,
|
|
FOREIGN KEY (website_id) REFERENCES websites(id) ON DELETE CASCADE,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS activity_log (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
user_id INT,
|
|
action VARCHAR(100) NOT NULL,
|
|
entity VARCHAR(100),
|
|
entity_id INT,
|
|
detail TEXT,
|
|
logged_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE SET NULL
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS shifts (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
name VARCHAR(200) NOT NULL,
|
|
days_of_week VARCHAR(20) NOT NULL DEFAULT '1234567',
|
|
start_time TIME NOT NULL DEFAULT '00:00:00',
|
|
end_time TIME NOT NULL DEFAULT '23:59:59',
|
|
is_active TINYINT(1) NOT NULL DEFAULT 1,
|
|
note TEXT,
|
|
created_by INT,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (created_by) REFERENCES users(id) ON DELETE SET NULL
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS shift_users (
|
|
shift_id INT NOT NULL,
|
|
user_id INT NOT NULL,
|
|
PRIMARY KEY (shift_id, user_id),
|
|
FOREIGN KEY (shift_id) REFERENCES shifts(id) ON DELETE CASCADE,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS shift_websites (
|
|
shift_id INT NOT NULL,
|
|
website_id INT NOT NULL,
|
|
sort_order INT NOT NULL DEFAULT 0,
|
|
PRIMARY KEY (shift_id, website_id),
|
|
FOREIGN KEY (shift_id) REFERENCES shifts(id) ON DELETE CASCADE,
|
|
FOREIGN KEY (website_id) REFERENCES websites(id) ON DELETE CASCADE
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS login_attempts (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
username VARCHAR(100) NOT NULL,
|
|
attempted_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
ip_address VARCHAR(45),
|
|
INDEX idx_username_time (username, attempted_at)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS website_users (
|
|
website_id INT NOT NULL,
|
|
user_id INT NOT NULL,
|
|
PRIMARY KEY (website_id, user_id),
|
|
FOREIGN KEY (website_id) REFERENCES websites(id) ON DELETE CASCADE,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS ai_criteria (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
title VARCHAR(200) NOT NULL,
|
|
description TEXT NOT NULL,
|
|
is_active TINYINT(1) NOT NULL DEFAULT 1,
|
|
sort_order INT NOT NULL DEFAULT 0,
|
|
created_by INT,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (created_by) REFERENCES users(id) ON DELETE SET NULL
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
""",
|
|
]
|
|
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cursor = conn.cursor()
|
|
for stmt in ddl_statements:
|
|
cursor.execute(stmt)
|
|
conn.commit()
|
|
logger.info("Database schema initialised successfully.")
|
|
|
|
# ── Safe migrations for existing deployments ───────────────────────
|
|
# Add check_type column to websites if it doesn't exist yet
|
|
cursor.execute(
|
|
"""
|
|
SELECT COUNT(*) FROM information_schema.COLUMNS
|
|
WHERE TABLE_SCHEMA = DATABASE()
|
|
AND TABLE_NAME = 'websites'
|
|
AND COLUMN_NAME = 'check_type'
|
|
"""
|
|
)
|
|
(has_col,) = cursor.fetchone()
|
|
if not has_col:
|
|
cursor.execute(
|
|
"ALTER TABLE websites ADD COLUMN check_type ENUM('daily','weekly') "
|
|
"NOT NULL DEFAULT 'daily' AFTER url"
|
|
)
|
|
conn.commit()
|
|
logger.info("Migration: added check_type column to websites table.")
|
|
|
|
# Add failed_attempts column to users if it doesn't exist
|
|
cursor.execute(
|
|
"""
|
|
SELECT COUNT(*) FROM information_schema.COLUMNS
|
|
WHERE TABLE_SCHEMA = DATABASE()
|
|
AND TABLE_NAME = 'users'
|
|
AND COLUMN_NAME = 'failed_attempts'
|
|
"""
|
|
)
|
|
(has_fa,) = cursor.fetchone()
|
|
if not has_fa:
|
|
cursor.execute(
|
|
"ALTER TABLE users "
|
|
"ADD COLUMN failed_attempts TINYINT UNSIGNED NOT NULL DEFAULT 0 AFTER is_active, "
|
|
"ADD COLUMN locked_until DATETIME NULL AFTER failed_attempts"
|
|
)
|
|
conn.commit()
|
|
logger.info("Migration: added failed_attempts and locked_until columns to users table.")
|
|
|
|
# Add visibility column to websites if it doesn't exist
|
|
cursor.execute(
|
|
"""
|
|
SELECT COUNT(*) FROM information_schema.COLUMNS
|
|
WHERE TABLE_SCHEMA = DATABASE()
|
|
AND TABLE_NAME = 'websites'
|
|
AND COLUMN_NAME = 'visibility'
|
|
"""
|
|
)
|
|
(has_vis,) = cursor.fetchone()
|
|
if not has_vis:
|
|
cursor.execute(
|
|
"ALTER TABLE websites ADD COLUMN visibility "
|
|
"ENUM('all','assigned') NOT NULL DEFAULT 'all' AFTER check_type"
|
|
)
|
|
conn.commit()
|
|
logger.info("Migration: added visibility column to websites table.")
|
|
|
|
# Seed default admin if users table is empty
|
|
cursor.execute("SELECT COUNT(*) FROM users")
|
|
(count,) = cursor.fetchone()
|
|
if count == 0:
|
|
import bcrypt as _bcrypt
|
|
default_pw = _bcrypt.hashpw(b"admin123", _bcrypt.gensalt(rounds=12)).decode("utf-8")
|
|
cursor.execute(
|
|
"INSERT INTO users (username, password, role, full_name) VALUES (%s,%s,'admin','System Admin')",
|
|
("admin", default_pw)
|
|
)
|
|
conn.commit()
|
|
logger.info("Default admin account seeded (username: admin / password: admin123).")
|
|
cursor.close()
|
|
except mysql.connector.Error as e:
|
|
logger.error(f"Database initialisation error: {e}")
|
|
raise
|
|
finally:
|
|
if conn:
|
|
conn.close()
|