nngo/WebChecker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

04/24 Migrated info which stored in .ini to database and keyring

Browse Source
This commit is contained in:
nngo
2026-04-24 15:37:58 -04:00
parent f2a8d3f15b
commit 5d6ca5a039
9 changed files with 479 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/crypto.py
+110 -22
View File
@@ -2,14 +2,18 @@
utils/crypto.py — Fernet symmetric encryption for website credentials.
Key derivation:
- A 32-byte random salt is generated on first use and stored in config.ini
under [crypto] / salt.
- A 32-byte random salt is generated on first use and stored in the
app_settings table (key: 'crypto.salt') instead of config.ini.
- The Fernet key is derived from the salt + a fixed application secret
using PBKDF2-HMAC-SHA256 (100,000 iterations).
- This means credentials are tied to the specific config.ini file on the
operator's machine; moving config.ini to another machine retains access.
- Because the salt lives in the shared MySQL database, any machine that
connects to the same DB can decrypt credentials without needing a local
config.ini — making the app fully portable across machines.
Migration:
- On first start after this change, if config.ini still contains a
[crypto]/salt entry it is automatically migrated to app_settings and
removed from the file.
- _decrypt() tries Fernet first; if that fails it returns the raw value
unchanged so that plaintext legacy credentials are still readable.
- Callers should re-encrypt on next write (update_website handles this).
@@ -18,7 +22,6 @@ Migration:
import base64
import logging
import os
import configparser
from cryptography.fernet import Fernet, InvalidToken
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
@@ -27,30 +30,116 @@ from cryptography.hazmat.primitives import hashes
logger = logging.getLogger("crypto")
_APP_SECRET = b"WebsiteChecker-v1-CredentialKey"
_CONFIG_FILE = "config.ini"
_ITERATIONS = 100_000
_SETTING_KEY = "crypto.salt"
_fernet: "Fernet | None" = None
# ─── Key bootstrap ────────────────────────────────────────────────────────────
def _ensure_app_settings_table() -> bool:
"""
Guarantee the app_settings table exists before we try to read/write it.
Returns True if the table is available, False if it could not be created
(e.g. the DB pool itself is not yet ready).
This guard is necessary because crypto.py can be called during login —
before initialize_database() has had a chance to run on a fresh install
or an upgraded database that doesn't yet have the app_settings table.
"""
try:
from config import get_connection
conn = get_connection()
cur = conn.cursor()
cur.execute(
"""
CREATE TABLE IF NOT EXISTS app_settings (
key_name VARCHAR(100) NOT NULL PRIMARY KEY,
value TEXT NULL,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
ON UPDATE CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
"""
)
conn.commit()
cur.close()
conn.close()
return True
except Exception as e:
logger.warning(f"Crypto: could not ensure app_settings table: {e}")
return False
def _get_or_create_salt() -> bytes:
"""Read salt from config.ini [crypto] section; create and persist if absent."""
cfg = configparser.ConfigParser()
cfg.read(_CONFIG_FILE, encoding="utf-8")
"""
Read the Fernet salt from app_settings, falling back to config.ini for
legacy installs, and generating a fresh salt for brand-new installs.
if "crypto" in cfg and cfg["crypto"].get("salt"):
return base64.b64decode(cfg["crypto"]["salt"])
Order of precedence:
1. app_settings table (primary — shared across machines via the DB)
2. config.ini [crypto]/salt (legacy migration path)
3. Generate a new random salt and persist it to app_settings
# Generate a fresh 32-byte salt
salt = os.urandom(32)
if "crypto" not in cfg:
cfg["crypto"] = {}
cfg["crypto"]["salt"] = base64.b64encode(salt).decode("ascii")
The table is created here if it doesn't yet exist, so this function is
safe to call before initialize_database() has run.
"""
# ── Step 1: try config.ini first (fastest, no DB needed yet) ─────────────
# Reading config.ini for the salt is always safe — it doesn't require the
# app_settings table to exist. If found, we attempt to also persist it to
# the DB (best-effort), but we use the value regardless.
legacy_b64 = None
try:
import configparser, os as _os
_cfg_file = "config.ini"
if _os.path.exists(_cfg_file):
cfg = configparser.ConfigParser()
cfg.read(_cfg_file, encoding="utf-8")
if cfg.has_option("crypto", "salt"):
legacy_b64 = cfg.get("crypto", "salt")
except Exception as e:
logger.warning(f"Crypto: could not read config.ini for salt: {e}")
with open(_CONFIG_FILE, "w", encoding="utf-8") as fh:
cfg.write(fh)
logger.info("Crypto: generated and persisted new credential encryption salt.")
# ── Step 2: ensure app_settings table exists ──────────────────────────────
table_ok = _ensure_app_settings_table()
# ── Step 3: try to read salt from DB ──────────────────────────────────────
if table_ok:
try:
from config import get_setting
raw = get_setting(_SETTING_KEY, "")
if raw:
return base64.b64decode(raw)
except Exception as e:
logger.warning(f"Crypto: could not read salt from app_settings: {e}")
# ── Step 4: migrate legacy salt from config.ini → DB ─────────────────────
if legacy_b64:
if table_ok:
try:
from config import set_setting
set_setting(_SETTING_KEY, legacy_b64)
logger.info("Crypto: migrated salt from config.ini to app_settings.")
except Exception as e:
logger.warning(f"Crypto: could not persist migrated salt to DB: {e}")
else:
logger.info("Crypto: using salt from config.ini (app_settings not available yet).")
return base64.b64decode(legacy_b64)
# ── Step 5: generate a fresh salt ────────────────────────────────────────
salt = os.urandom(32)
b64_salt = base64.b64encode(salt).decode("ascii")
if table_ok:
try:
from config import set_setting
set_setting(_SETTING_KEY, b64_salt)
logger.info("Crypto: generated and persisted new salt to app_settings.")
except Exception as e:
logger.warning(f"Crypto: could not persist new salt to DB: {e}")
else:
logger.warning(
"Crypto: generated a new salt but app_settings is not available — "
"salt will NOT persist across restarts until the table is created."
)
return salt
@@ -74,7 +163,7 @@ def _get_fernet() -> Fernet:
def reset_fernet():
"""Force key reload — call after config.ini is replaced (e.g. settings save)."""
"""Force key reload — call if the salt is ever rotated."""
global _fernet
_fernet = None
@@ -107,7 +196,6 @@ def decrypt(ciphertext: str) -> str:
if not ciphertext:
return ciphertext
if not ciphertext.startswith("enc:"):
# Legacy plaintext — return unchanged; will be re-encrypted on next save
return ciphertext
try:
token = ciphertext[4:].encode("ascii")
@@ -122,4 +210,4 @@ def decrypt(ciphertext: str) -> str:
def is_encrypted(value: str) -> bool:
"""Return True if the value was produced by encrypt()."""
return isinstance(value, str) and value.startswith("enc:")
return isinstance(value, str) and value.startswith("enc:")
utils/scheduler.py
+74 -125
View File
@@ -5,24 +5,26 @@ Runs a background daemon thread that wakes every minute, checks whether
the configured send_time (HH:MM) has been reached today, and sends the
summary report via SMTP if it hasn't been sent yet.
Configuration (config.ini [email] section):
enabled = true/false
smtp_host = smtp.example.com
smtp_port = 587
smtp_user = sender@example.com
smtp_password= secret
use_tls = true
recipients = admin@example.com, manager@example.com
send_time = 18:00 (24-hour HH:MM, local time)
Configuration is stored in the app_settings database table (not config.ini):
Key Description
─────────────────────── ────────────────────────────────────────────
email.enabled 'true' or 'false'
email.smtp_host SMTP server hostname
email.smtp_port SMTP port number (string)
email.smtp_user Sender email address / SMTP login
email.smtp_password SMTP password (Fernet-encrypted)
email.security 'starttls' | 'ssl' | 'none'
email.recipients Comma-separated recipient addresses
email.send_time HH:MM (24-hour local time)
email.last_sent_date ISO date of last successful send (YYYY-MM-DD)
Call start() once after login succeeds (admin only).
Call stop() on logout/shutdown.
"""
import configparser
import datetime
import logging
import os
import smtplib
import threading
from email.mime.multipart import MIMEMultipart
@@ -31,7 +33,6 @@ from email.utils import formatdate, make_msgid, formataddr
logger = logging.getLogger("scheduler")
CONFIG_FILE = "config.ini"
_scheduler_thread: "threading.Thread | None" = None
_stop_event = threading.Event()
@@ -39,56 +40,53 @@ _stop_event = threading.Event()
# ─── Config helpers ───────────────────────────────────────────────────────────
def load_email_config() -> dict:
"""Load email settings from app_settings table."""
from config import get_settings_dict
from utils.config_crypto import decrypt_value
cfg = configparser.ConfigParser()
if not os.path.exists(CONFIG_FILE):
return {}
cfg.read(CONFIG_FILE, encoding="utf-8")
if "email" not in cfg:
return {}
s = cfg["email"]
# security field: "starttls" | "ssl" | "none"
# Falls back from legacy use_tls boolean for existing configs.
security = s.get("security", "")
s = get_settings_dict("email.")
security = s.get("email.security", "")
if not security:
security = "starttls" if s.getboolean("use_tls", fallback=True) else "none"
# Legacy: fall back from boolean use_tls if security key absent
use_tls = s.get("email.use_tls", "true").lower() == "true"
security = "starttls" if use_tls else "none"
return {
"enabled": s.getboolean("enabled", fallback=False),
"smtp_host": s.get("smtp_host", ""),
"smtp_port": s.getint("smtp_port", fallback=587),
"smtp_user": s.get("smtp_user", ""),
"smtp_password": decrypt_value(s.get("smtp_password", "")),
"enabled": s.get("email.enabled", "false").lower() == "true",
"smtp_host": s.get("email.smtp_host", ""),
"smtp_port": int(s.get("email.smtp_port", "587") or "587"),
"smtp_user": s.get("email.smtp_user", ""),
"smtp_password": decrypt_value(s.get("email.smtp_password", "")),
"security": security,
"use_tls": security == "starttls", # kept for compatibility
"recipients": [r.strip() for r in s.get("recipients", "").split(",") if r.strip()],
"send_time": s.get("send_time", "18:00"),
"use_tls": security == "starttls",
"recipients": [r.strip() for r in
s.get("email.recipients", "").split(",") if r.strip()],
"send_time": s.get("email.send_time", "18:00"),
}
def save_email_config(enabled: bool, smtp_host: str, smtp_port: int,
smtp_user: str, smtp_password: str, security: str,
recipients: str, send_time: str):
"""Persist email settings to app_settings table."""
from config import get_setting, set_setting
from utils.config_crypto import encrypt_value
cfg = configparser.ConfigParser()
cfg.read(CONFIG_FILE, encoding="utf-8")
# Preserve last_sent_date if present
last_sent = cfg.get("email", "last_sent_date", fallback="")
cfg["email"] = {
"enabled": str(enabled).lower(),
"smtp_host": smtp_host,
"smtp_port": str(smtp_port),
"smtp_user": smtp_user,
"smtp_password": encrypt_value(smtp_password),
"security": security,
"use_tls": str(security == "starttls").lower(), # legacy compat
"recipients": recipients,
"send_time": send_time,
# Preserve last_sent_date — do not overwrite it on a normal save
last_sent = get_setting("email.last_sent_date", "")
pairs = {
"email.enabled": str(enabled).lower(),
"email.smtp_host": smtp_host,
"email.smtp_port": str(smtp_port),
"email.smtp_user": smtp_user,
"email.smtp_password": encrypt_value(smtp_password),
"email.security": security,
"email.use_tls": str(security == "starttls").lower(),
"email.recipients": recipients,
"email.send_time": send_time,
}
for k, v in pairs.items():
set_setting(k, v)
if last_sent:
cfg["email"]["last_sent_date"] = last_sent
with open(CONFIG_FILE, "w", encoding="utf-8") as fh:
cfg.write(fh)
logger.info(f"Email configuration saved (security={security}, encrypted).")
set_setting("email.last_sent_date", last_sent)
logger.info(f"Email configuration saved to app_settings (security={security}).")
def _make_smtp_server(smtp_host: str, smtp_port: int,
@@ -108,7 +106,7 @@ def _make_smtp_server(smtp_host: str, smtp_port: int,
server = smtplib.SMTP(smtp_host, smtp_port, timeout=15)
server.ehlo()
server.starttls()
server.ehlo() # re-identify after TLS upgrade — mandatory
server.ehlo()
else:
server = smtplib.SMTP(smtp_host, smtp_port, timeout=15)
server.ehlo()
@@ -120,16 +118,9 @@ def test_smtp_connection(smtp_host: str, smtp_port: int,
security: str) -> tuple:
"""
Step-by-step SMTP diagnostic. Returns (success: bool, message: str).
Each step is attempted independently so the error message tells the
admin exactly where the failure occurred:
Step 1 — DNS resolution
Step 2 — TCP connect
Step 3 — TLS handshake (if applicable)
Step 4 — Authentication
"""
import socket
# Step 1: DNS resolution
try:
addr = socket.getaddrinfo(smtp_host, smtp_port,
socket.AF_UNSPEC, socket.SOCK_STREAM)
@@ -143,7 +134,6 @@ def test_smtp_connection(smtp_host: str, smtp_port: int,
f"Check the hostname and your network connection.\n({e})"
)
# Step 2: TCP connect (raw socket, before any SMTP protocol)
try:
sock = socket.create_connection((smtp_host, smtp_port), timeout=8)
sock.close()
@@ -154,43 +144,30 @@ def test_smtp_connection(smtp_host: str, smtp_port: int,
f"The port may be blocked by a firewall or the server is down.\n({e})"
)
# Steps 3 + 4: SMTP protocol, TLS handshake, authentication
try:
server = _make_smtp_server(smtp_host, smtp_port, security)
logger.info(f"SMTP test: TLS/connection OK (security={security})")
except smtplib.SMTPConnectError as e:
return False, (
f"Step 3 FAILED — SMTP connect: {e}\n"
f"Try a different Security mode or port."
)
return False, (f"Step 3 FAILED — SMTP connect: {e}\nTry a different Security mode or port.")
except smtplib.SMTPException as e:
return False, (
f"Step 3 FAILED — TLS handshake: {e}\n"
f"Try switching Security mode (e.g. SSL/TLS on port 465)."
)
return False, (f"Step 3 FAILED — TLS handshake: {e}\nTry switching Security mode.")
except OSError as e:
return False, (
f"Step 3 FAILED — connection dropped: {e}\n"
f"Try switching Security mode or port."
)
return False, (f"Step 3 FAILED — connection dropped: {e}\nTry switching Security mode or port.")
try:
server.login(smtp_user, smtp_password)
server.quit()
logger.info("SMTP test: authentication OK")
return True, (
f"All steps passed.\n"
f"Connected to {smtp_host}:{smtp_port} "
f"All steps passed.\nConnected to {smtp_host}:{smtp_port} "
f"({security.upper()}) and authenticated successfully."
)
except smtplib.SMTPAuthenticationError as e:
try:
server.quit()
except Exception:
pass
try: server.quit()
except Exception: pass
return False, (
f"Step 4 FAILED — Authentication: username or password rejected.\n"
f"For Gmail/Google Workspace use an App Password, not your account password.\n({e})"
f"For Gmail/Google Workspace use an App Password.\n({e})"
)
except smtplib.SMTPException as e:
return False, f"Step 4 FAILED — SMTP error during login: {e}"
@@ -201,10 +178,7 @@ def test_smtp_connection(smtp_host: str, smtp_port: int,
def send_test_email(smtp_host: str, smtp_port: int,
smtp_user: str, smtp_password: str,
security: str, recipients: list) -> tuple:
"""
Send a real test email through the full pipeline.
Returns (success: bool, message: str).
"""
"""Send a real test email through the full pipeline."""
try:
subject = "Website Checker — SMTP Test"
body = (
@@ -226,10 +200,8 @@ def send_test_email(smtp_host: str, smtp_port: int,
msg["Date"] = formatdate(localtime=True)
msg["Message-ID"] = make_msgid(domain=smtp_user.split("@")[-1] if "@" in smtp_user else "webchecker")
msg["X-Mailer"] = "WebChecker"
# Plain-text part must come first; HTML second.
# Spam filters heavily penalise HTML-only messages with no text/plain alternative.
msg.attach(MIMEText(plain, "plain", "utf-8"))
msg.attach(MIMEText(body, "html", "utf-8"))
msg.attach(MIMEText(body, "html", "utf-8"))
server = _make_smtp_server(smtp_host, smtp_port, security)
server.login(smtp_user, smtp_password)
@@ -238,10 +210,8 @@ def send_test_email(smtp_host: str, smtp_port: int,
logger.info(f"Test email sent to {recipients} via {smtp_host}:{smtp_port}")
return True, f"Test email sent successfully to: {', '.join(recipients)}"
except smtplib.SMTPAuthenticationError as e:
return False, (
f"Authentication failed — check username and password.\n"
f"For Gmail/Google Workspace use an App Password.\n({e})"
)
return False, (f"Authentication failed — check username and password.\n"
f"For Gmail/Google Workspace use an App Password.\n({e})")
except smtplib.SMTPConnectError as e:
return False, f"Could not connect to {smtp_host}:{smtp_port} — {e}"
except smtplib.SMTPException as e:
@@ -269,8 +239,6 @@ def _build_html_report() -> str:
skipped_no_shift = 0
for r in rows:
total = int(r.get("total_sites") or 0)
# Skip users with no shifts scheduled today — they have no expected
# work for this day and showing them as "0 / 0 — 0%" is misleading.
if total == 0:
skipped_no_shift += 1
continue
@@ -324,11 +292,11 @@ def _build_html_report() -> str:
def _send_report(cfg: dict):
"""Build and send the daily report email."""
html = _build_html_report()
today = datetime.date.today().strftime("%d %b %Y")
subject = f"Website Checker — Daily Report {today}"
html = _build_html_report()
today = datetime.date.today().strftime("%d %b %Y")
subject = f"Website Checker — Daily Report {today}"
smtp_user = cfg["smtp_user"]
plain = (
f"Website Checker — Daily Report {today}\n\n"
"Please view this report in an HTML-capable email client for full formatting.\n"
@@ -341,17 +309,14 @@ def _send_report(cfg: dict):
msg["Date"] = formatdate(localtime=True)
msg["Message-ID"] = make_msgid(domain=smtp_user.split("@")[-1] if "@" in smtp_user else "webchecker")
msg["X-Mailer"] = "WebChecker"
# Plain-text part must come first; HTML second.
# Spam filters heavily penalise HTML-only messages with no text/plain alternative.
msg.attach(MIMEText(plain, "plain", "utf-8"))
msg.attach(MIMEText(html, "html", "utf-8"))
msg.attach(MIMEText(html, "html", "utf-8"))
try:
security = cfg.get("security", "starttls")
server = _make_smtp_server(
cfg["smtp_host"], cfg["smtp_port"], security)
server.login(cfg["smtp_user"], cfg["smtp_password"])
server.sendmail(cfg["smtp_user"], cfg["recipients"], msg.as_string())
server = _make_smtp_server(cfg["smtp_host"], cfg["smtp_port"], security)
server.login(smtp_user, cfg["smtp_password"])
server.sendmail(smtp_user, cfg["recipients"], msg.as_string())
server.quit()
logger.info(f"Daily report emailed to: {cfg['recipients']}")
except smtplib.SMTPAuthenticationError as e:
@@ -367,12 +332,9 @@ def _send_report(cfg: dict):
def _get_last_sent_date() -> "datetime.date | None":
"""Read the last-sent date from config.ini [email] last_sent_date key."""
cfg = configparser.ConfigParser()
if not os.path.exists(CONFIG_FILE):
return None
cfg.read(CONFIG_FILE, encoding="utf-8")
raw = cfg.get("email", "last_sent_date", fallback="")
"""Read the last-sent date from app_settings."""
from config import get_setting
raw = get_setting("email.last_sent_date", "")
if not raw:
return None
try:
@@ -382,25 +344,18 @@ def _get_last_sent_date() -> "datetime.date | None":
def _set_last_sent_date(d: "datetime.date"):
"""Persist the last-sent date to config.ini [email] last_sent_date key."""
cfg = configparser.ConfigParser()
if os.path.exists(CONFIG_FILE):
cfg.read(CONFIG_FILE, encoding="utf-8")
if "email" not in cfg:
cfg["email"] = {}
cfg["email"]["last_sent_date"] = d.isoformat()
with open(CONFIG_FILE, "w", encoding="utf-8") as fh:
cfg.write(fh)
"""Persist the last-sent date to app_settings."""
from config import set_setting
set_setting("email.last_sent_date", d.isoformat())
# ─── Scheduler loop ───────────────────────────────────────────────────────────
def _scheduler_loop():
# Seed from persisted value so a restart after send_time does not re-send.
last_sent_date = _get_last_sent_date()
while not _stop_event.is_set():
_stop_event.wait(60) # sleep 60 seconds between checks
_stop_event.wait(60)
if _stop_event.is_set():
break
@@ -427,16 +382,10 @@ def _scheduler_loop():
def start():
"""Start the background scheduler thread. Call once after successful login."""
global _scheduler_thread, _stop_event
# Guard against double-start: if a thread is already alive (e.g. admin
# logs out and back in), stop it cleanly before spawning a new one.
# Without this guard, _stop_event.clear() would unblock the sleeping
# thread while a second thread also starts, resulting in two scheduler
# threads firing simultaneously and potentially sending duplicate emails.
if _scheduler_thread is not None and _scheduler_thread.is_alive():
logger.info("Email scheduler already running - stopping before restart.")
_stop_event.set()
_scheduler_thread.join(timeout=5)
# Create a fresh Event so there is no residual set-state from a prior stop()
_stop_event = threading.Event()
_scheduler_thread = threading.Thread(target=_scheduler_loop,
name="EmailScheduler", daemon=True)