Files
Infosec/Alert Hub System/app/routes_backoffice.py
T

427 lines
18 KiB
Python
Raw Normal View History

2026-09-13 20:09:20 +01:00
"""Backoffice routes — admins only.
Sections:
3. Manage Users & Permissions
- Mail / SMTP configuration
- Global settings (default reminder days, scheduler, run-now)
- Audit log viewer
"""
import csv
import io
from datetime import date, datetime
from flask import Blueprint, render_template, redirect, url_for, flash, request, abort, current_app, Response
from flask_login import current_user
from .models import db, User, Alert, AppSetting, AuditLog, MailLog
from .forms import UserForm, MailConfigForm, TestMailForm, GlobalSettingsForm, LogEmailForm
from .auth_utils import admin_required, get_remote_ip
from .mailer import send_mail, get_mail_config
from .scheduler import trigger_now
backoffice_bp = Blueprint("backoffice", __name__)
def _csv_response(filename: str, header: list, rows) -> Response:
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow(header)
for r in rows:
writer.writerow(r)
data = buf.getvalue()
return Response(
data,
mimetype="text/csv",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
def _audit_rows():
for e in AuditLog.query.order_by(AuditLog.timestamp.desc()).all():
yield [
e.timestamp.strftime("%Y-%m-%d %H:%M:%S") if e.timestamp else "",
e.username or "",
e.action or "",
e.details or "",
e.ip_address or "",
]
def _maillog_rows():
for m in MailLog.query.order_by(MailLog.timestamp.desc()).all():
yield [
m.timestamp.strftime("%Y-%m-%d %H:%M:%S") if m.timestamp else "",
m.kind or "",
m.recipients or "",
m.subject or "",
m.status or "",
m.sent_copy or "",
m.triggered_by or "",
m.info or "",
]
AUDIT_CSV_HEADER = ["timestamp_utc", "user", "action", "details", "ip"]
MAILLOG_CSV_HEADER = ["timestamp_utc", "kind", "recipients", "subject", "status", "sent_copy", "triggered_by", "info"]
def _build_csv_string(header, rows) -> str:
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow(header)
for r in rows:
writer.writerow(r)
return buf.getvalue()
# --------------------------------------------------------------------------- #
# Backoffice home — admin dashboard
# --------------------------------------------------------------------------- #
@backoffice_bp.route("/")
@admin_required
def dashboard():
user_count = User.query.count()
admin_count = User.query.filter_by(role="admin").count()
active_alerts = Alert.query.filter_by(is_active=True).count()
today = date.today()
expiring_soon = Alert.query.filter(
Alert.is_active.is_(True),
Alert.expiration_date >= today,
).order_by(Alert.expiration_date.asc()).limit(5).all()
recent_audit = AuditLog.query.order_by(AuditLog.timestamp.desc()).limit(10).all()
return render_template(
"backoffice/dashboard.html",
user_count=user_count,
admin_count=admin_count,
active_alerts=active_alerts,
expiring_soon=expiring_soon,
recent_audit=recent_audit,
today=today,
)
# --------------------------------------------------------------------------- #
# User management
# --------------------------------------------------------------------------- #
@backoffice_bp.route("/users")
@admin_required
def users_list():
users = User.query.order_by(User.username.asc()).all()
return render_template("backoffice/users_list.html", users=users)
@backoffice_bp.route("/users/new", methods=["GET", "POST"])
@admin_required
def user_new():
form = UserForm()
if form.validate_on_submit():
if not form.password.data:
flash("Password is required for a new user.", "error")
elif User.query.filter((User.username == form.username.data) | (User.email == form.email.data)).first():
flash("Username or email already exists.", "error")
elif len(form.password.data) < current_app.config["PASSWORD_MIN_LENGTH"]:
flash(f"Password must be at least {current_app.config['PASSWORD_MIN_LENGTH']} characters.", "error")
else:
user = User(
username=form.username.data.strip(),
email=form.email.data.strip().lower(),
role=form.role.data,
is_active=form.is_active.data,
)
user.set_password(form.password.data)
db.session.add(user)
db.session.commit()
AuditLog.record("user.create", user=current_user, details=f"new={user.username} role={user.role}", ip=get_remote_ip())
flash(f"User '{user.username}' created.", "success")
return redirect(url_for("backoffice.users_list"))
return render_template("backoffice/user_form.html", form=form, user=None)
@backoffice_bp.route("/users/<int:user_id>/edit", methods=["GET", "POST"])
@admin_required
def user_edit(user_id: int):
user = db.session.get(User, user_id)
if user is None:
abort(404)
form = UserForm(obj=user)
if form.validate_on_submit():
# Username / email uniqueness check (excluding self)
clash = User.query.filter(
User.id != user.id,
(User.username == form.username.data) | (User.email == form.email.data.lower())
).first()
if clash:
flash("Another user already has that username or email.", "error")
else:
# Prevent self-demotion of last admin / self-deactivation lock-out
if user.id == current_user.id:
if form.role.data != "admin":
flash("You cannot demote yourself.", "error")
return render_template("backoffice/user_form.html", form=form, user=user)
if not form.is_active.data:
flash("You cannot deactivate your own account.", "error")
return render_template("backoffice/user_form.html", form=form, user=user)
# Prevent removing the last active admin
if user.role == "admin" and form.role.data != "admin":
other_admins = User.query.filter(User.id != user.id, User.role == "admin", User.is_active.is_(True)).count()
if other_admins == 0:
flash("Cannot demote the last administrator.", "error")
return render_template("backoffice/user_form.html", form=form, user=user)
user.username = form.username.data.strip()
user.email = form.email.data.strip().lower()
user.role = form.role.data
user.is_active = form.is_active.data
if form.password.data:
if len(form.password.data) < current_app.config["PASSWORD_MIN_LENGTH"]:
flash(f"Password must be at least {current_app.config['PASSWORD_MIN_LENGTH']} characters.", "error")
return render_template("backoffice/user_form.html", form=form, user=user)
user.set_password(form.password.data)
db.session.commit()
AuditLog.record("user.update", user=current_user, details=f"id={user.id} username={user.username}", ip=get_remote_ip())
flash("User updated.", "success")
return redirect(url_for("backoffice.users_list"))
return render_template("backoffice/user_form.html", form=form, user=user)
@backoffice_bp.route("/users/<int:user_id>/delete", methods=["POST"])
@admin_required
def user_delete(user_id: int):
user = db.session.get(User, user_id)
if user is None:
abort(404)
if user.id == current_user.id:
flash("You cannot delete your own account.", "error")
return redirect(url_for("backoffice.users_list"))
if user.role == "admin":
other_admins = User.query.filter(User.id != user.id, User.role == "admin", User.is_active.is_(True)).count()
if other_admins == 0:
flash("Cannot delete the last administrator.", "error")
return redirect(url_for("backoffice.users_list"))
username = user.username
db.session.delete(user)
db.session.commit()
AuditLog.record("user.delete", user=current_user, details=f"deleted={username}", ip=get_remote_ip())
flash(f"User '{username}' deleted.", "success")
return redirect(url_for("backoffice.users_list"))
@backoffice_bp.route("/users/<int:user_id>/reset-mfa", methods=["POST"])
@admin_required
def user_reset_mfa(user_id: int):
user = db.session.get(User, user_id)
if user is None:
abort(404)
user.mfa_enabled = False
user.mfa_secret = None
db.session.commit()
AuditLog.record("user.mfa_reset", user=current_user, details=f"target={user.username}", ip=get_remote_ip())
flash(f"MFA reset for '{user.username}'. They can re-enrol on next login.", "info")
return redirect(url_for("backoffice.users_list"))
@backoffice_bp.route("/users/<int:user_id>/unlock", methods=["POST"])
@admin_required
def user_unlock(user_id: int):
user = db.session.get(User, user_id)
if user is None:
abort(404)
user.locked_until = None
user.failed_login_count = 0
db.session.commit()
AuditLog.record("user.unlock", user=current_user, details=f"target={user.username}", ip=get_remote_ip())
flash(f"Unlocked '{user.username}'.", "success")
return redirect(url_for("backoffice.users_list"))
# --------------------------------------------------------------------------- #
# Mail configuration
# --------------------------------------------------------------------------- #
@backoffice_bp.route("/mail-config", methods=["GET", "POST"])
@admin_required
def mail_config():
form = MailConfigForm()
test_form = TestMailForm()
if request.method == "GET":
cfg = get_mail_config()
form.smtp_host.data = cfg["host"]
form.smtp_port.data = cfg["port"]
form.smtp_username.data = cfg["username"]
form.smtp_encryption.data = cfg["encryption"]
form.smtp_from_address.data = cfg["from_address"]
form.smtp_from_name.data = cfg["from_name"]
form.imap_enabled.data = cfg["imap_enabled"]
form.imap_host.data = cfg["imap_host"]
form.imap_port.data = cfg["imap_port"]
form.imap_username.data = cfg["imap_username"]
form.imap_sent_folder.data = cfg["imap_sent_folder"]
if form.submit.data and form.validate_on_submit():
AppSetting.set("smtp_host", form.smtp_host.data.strip())
AppSetting.set("smtp_port", str(form.smtp_port.data))
AppSetting.set("smtp_username", form.smtp_username.data.strip())
if form.smtp_password.data:
AppSetting.set("smtp_password", form.smtp_password.data)
AppSetting.set("smtp_encryption", form.smtp_encryption.data)
AppSetting.set("smtp_from_address", form.smtp_from_address.data.strip().lower())
AppSetting.set("smtp_from_name", (form.smtp_from_name.data or "").strip() or "AlertHub")
# ---- IMAP (Sent-folder copy) ----
AppSetting.set("imap_enabled", "true" if form.imap_enabled.data else "false")
AppSetting.set("imap_host", (form.imap_host.data or "").strip())
AppSetting.set("imap_port", str(form.imap_port.data or 993))
AppSetting.set("imap_username", (form.imap_username.data or "").strip())
if form.imap_password.data:
AppSetting.set("imap_password", form.imap_password.data)
AppSetting.set("imap_sent_folder", (form.imap_sent_folder.data or "").strip() or "Sent")
AuditLog.record("mailconfig.update", user=current_user, ip=get_remote_ip())
flash("Mail configuration saved.", "success")
return redirect(url_for("backoffice.mail_config"))
return render_template("backoffice/mail_config.html", form=form, test_form=test_form)
@backoffice_bp.route("/mail-config/test", methods=["POST"])
@admin_required
def mail_test():
test_form = TestMailForm()
if test_form.validate_on_submit():
ok, info = send_mail(
test_form.test_recipient.data.strip(),
"[AlertHub] Test email",
"This is a test email from AlertHub. If you received this, SMTP is configured correctly.",
"<p>This is a <strong>test email</strong> from AlertHub. If you received this, SMTP is configured correctly.</p>",
kind="test",
triggered_by=current_user.username,
)
AuditLog.record("mailconfig.test", user=current_user, details=f"to={test_form.test_recipient.data} ok={ok}", ip=get_remote_ip())
flash(("Test email sent. " if ok else "Test failed: ") + info, "success" if ok else "error")
else:
flash("Provide a valid recipient email address.", "error")
return redirect(url_for("backoffice.mail_config"))
# --------------------------------------------------------------------------- #
# Global settings + scheduler trigger
# --------------------------------------------------------------------------- #
@backoffice_bp.route("/settings", methods=["GET", "POST"])
@admin_required
def settings():
form = GlobalSettingsForm()
if request.method == "GET":
form.default_reminder_days.data = AppSetting.get("default_reminder_days", "30,14,7,1")
if form.validate_on_submit():
cleaned = ",".join(
str(x) for x in sorted({int(p.strip()) for p in form.default_reminder_days.data.split(",") if p.strip().isdigit()}, reverse=True)
)
if not cleaned:
flash("Please enter at least one valid day value.", "error")
else:
AppSetting.set("default_reminder_days", cleaned)
AuditLog.record("settings.update", user=current_user, details=f"default_reminder_days={cleaned}", ip=get_remote_ip())
flash("Settings saved.", "success")
return redirect(url_for("backoffice.settings"))
scheduler_interval = current_app.config.get("SCHEDULER_INTERVAL_MINUTES", 60)
return render_template("backoffice/settings.html", form=form, scheduler_interval=scheduler_interval)
@backoffice_bp.route("/settings/run-check", methods=["POST"])
@admin_required
def run_check_now():
"""Manually run the reminder scan once."""
try:
trigger_now(current_app._get_current_object())
AuditLog.record("scheduler.manual_run", user=current_user, ip=get_remote_ip())
flash("Reminder check executed.", "success")
except Exception as e:
current_app.logger.exception("Manual scheduler run failed")
flash(f"Reminder check failed: {e}", "error")
return redirect(url_for("backoffice.settings"))
# --------------------------------------------------------------------------- #
# Audit log viewer
# --------------------------------------------------------------------------- #
@backoffice_bp.route("/audit-log")
@admin_required
def audit_log():
page = max(1, int(request.args.get("page", 1)))
per_page = 50
pagination = AuditLog.query.order_by(AuditLog.timestamp.desc()).paginate(page=page, per_page=per_page, error_out=False)
return render_template("backoffice/audit_log.html", pagination=pagination, log_email_form=LogEmailForm())
@backoffice_bp.route("/audit-log/export")
@admin_required
def audit_log_export():
stamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
AuditLog.record("auditlog.export", user=current_user, ip=get_remote_ip())
return _csv_response(f"alerthub-audit-log-{stamp}.csv", AUDIT_CSV_HEADER, _audit_rows())
@backoffice_bp.route("/audit-log/email", methods=["POST"])
@admin_required
def audit_log_email():
form = LogEmailForm()
if not form.validate_on_submit():
flash("Provide a valid destination email address.", "error")
return redirect(url_for("backoffice.audit_log"))
recipient = form.log_recipient.data.strip()
stamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
csv_str = _build_csv_string(AUDIT_CSV_HEADER, _audit_rows())
ok, info = send_mail(
recipient,
f"[AlertHub] Audit log export ({stamp} UTC)",
"Attached is the AlertHub audit log export requested from the backoffice.",
"<p>Attached is the <strong>AlertHub audit log</strong> export requested from the backoffice.</p>",
attachments=[(f"alerthub-audit-log-{stamp}.csv", "text", "csv", csv_str.encode("utf-8"))],
kind="log-export",
triggered_by=current_user.username,
)
AuditLog.record("auditlog.email", user=current_user, details=f"to={recipient} ok={ok}", ip=get_remote_ip())
flash(("Audit log emailed. " if ok else "Failed to email log: ") + info, "success" if ok else "error")
return redirect(url_for("backoffice.audit_log"))
# --------------------------------------------------------------------------- #
# Mail log viewer (every message the platform sends)
# --------------------------------------------------------------------------- #
@backoffice_bp.route("/mail-log")
@admin_required
def mail_log():
page = max(1, int(request.args.get("page", 1)))
per_page = 50
pagination = MailLog.query.order_by(MailLog.timestamp.desc()).paginate(page=page, per_page=per_page, error_out=False)
return render_template("backoffice/mail_log.html", pagination=pagination, log_email_form=LogEmailForm())
@backoffice_bp.route("/mail-log/export")
@admin_required
def mail_log_export():
stamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
AuditLog.record("maillog.export", user=current_user, ip=get_remote_ip())
return _csv_response(f"alerthub-mail-log-{stamp}.csv", MAILLOG_CSV_HEADER, _maillog_rows())
@backoffice_bp.route("/mail-log/email", methods=["POST"])
@admin_required
def mail_log_email():
form = LogEmailForm()
if not form.validate_on_submit():
flash("Provide a valid destination email address.", "error")
return redirect(url_for("backoffice.mail_log"))
recipient = form.log_recipient.data.strip()
stamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
csv_str = _build_csv_string(MAILLOG_CSV_HEADER, _maillog_rows())
ok, info = send_mail(
recipient,
f"[AlertHub] Mail log export ({stamp} UTC)",
"Attached is the AlertHub mail log export requested from the backoffice.",
"<p>Attached is the <strong>AlertHub mail log</strong> export requested from the backoffice.</p>",
attachments=[(f"alerthub-mail-log-{stamp}.csv", "text", "csv", csv_str.encode("utf-8"))],
kind="log-export",
triggered_by=current_user.username,
)
AuditLog.record("maillog.email", user=current_user, details=f"to={recipient} ok={ok}", ip=get_remote_ip())
flash(("Mail log emailed. " if ok else "Failed to email log: ") + info, "success" if ok else "error")
return redirect(url_for("backoffice.mail_log"))