Simon/MClogger
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modified: web/app.py

Browse Source 
modified:   web/blueprints/auth.py
	modified:   web/blueprints/group_admin.py
	modified:   web/blueprints/site_admin.py
	new file:   web/limiter.py
	modified:   web/panel_db.py
	modified:   web/requirements.txt
	new file:   web/templates/429.html
	new file:   web/templates/admin/audit_log.html
	modified:   web/templates/admin/base.html
This commit is contained in:
simon
2026-04-14 13:02:41 +02:00
parent 452d50e5b5
commit 3b78f5dfb1
10 changed files with 564 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
web/app.py
View File

@@ -9,6 +9,7 @@ from flask import Flask, abort, render_template, request, session, url_for
from config import Config
from panel_db import init_databases, get_user_groups
from roles import can_manage_group
from limiter import limiter
from blueprints.auth import auth
from blueprints.site_admin import site_admin
@@ -33,6 +34,17 @@ def create_app() -> Flask:
app.register_blueprint(group_admin)
app.register_blueprint(panel)
# Rate limiter
limiter.init_app(app)
@app.errorhandler(429)
def rate_limit_exceeded(e):
retry_after = getattr(e, "retry_after", None)
return render_template(
"429.html",
retry_after=int(retry_after) if retry_after else 60,
), 429
# Panel-Datenbank-Tabellen anlegen
try:
init_databases()

45
web/blueprints/auth.py
View File

@@ -5,18 +5,21 @@ Getrennte Login-Seiten für Site-Admins und normale Nutzer/Gruppen-Admins.
import json
from datetime import datetime
from flask import Blueprint, render_template, request, redirect, url_for, session, flash
from panel_db import accept_group_invite, check_login, get_invite_by_token, get_user_groups
from panel_db import accept_group_invite, check_login, get_invite_by_token, get_user_groups, log_audit_event
from limiter import limiter
auth = Blueprint("auth", __name__)
@auth.route("/login", methods=["GET", "POST"])
@limiter.limit("15 per minute", methods=["POST"])
def login():
if session.get("user_id"):
return redirect(url_for("panel.dashboard"))
error = None
if request.method == "POST":
user = check_login(request.form.get("username", ""), request.form.get("password", ""))
username = request.form.get("username", "")
user = check_login(username, request.form.get("password", ""))
if user and user["is_site_admin"]:
flash("Please use the Site Admin login.", "warning")
return redirect(url_for("auth.admin_login"))
@@ -26,29 +29,56 @@ def login():
error = "You are not assigned to any group. Please contact an admin."
else:
_set_user_session(user, groups)
log_audit_event(
user["id"], user["username"], "user.login",
entity_type="user", entity_id=user["id"],
ip_address=request.remote_addr,
)
return redirect(url_for("panel.dashboard"))
else:
log_audit_event(
None, None, "user.login_failed",
details={"username": username},
ip_address=request.remote_addr,
)
error = "Incorrect username or password."
return render_template("auth/login.html", error=error)
@auth.route("/admin/login", methods=["GET", "POST"])
@limiter.limit("10 per minute", methods=["POST"])
def admin_login():
if session.get("is_site_admin"):
return redirect(url_for("site_admin.dashboard"))
error = None
if request.method == "POST":
user = check_login(request.form.get("username", ""), request.form.get("password", ""))
username = request.form.get("username", "")
user = check_login(username, request.form.get("password", ""))
if user and user["is_site_admin"]:
session["user_id"] = user["id"]
session["username"] = user["username"]
session["is_site_admin"] = True
session["group_id"] = None
session["permissions"] = {}
log_audit_event(
user["id"], user["username"], "admin.login",
entity_type="user", entity_id=user["id"],
ip_address=request.remote_addr,
)
return redirect(url_for("site_admin.dashboard"))
elif user:
log_audit_event(
user["id"], user["username"], "admin.login_failed",
details={"reason": "no_admin_privileges"},
ip_address=request.remote_addr,
)
error = "No Site Admin privileges."
else:
log_audit_event(
None, None, "admin.login_failed",
details={"username": username},
ip_address=request.remote_addr,
)
error = "Incorrect username or password."
return render_template("auth/admin_login.html", error=error)
@@ -74,6 +104,7 @@ def switch_group(group_id):
@auth.route("/invite/<token>", methods=["GET", "POST"])
@limiter.limit("20 per minute", methods=["POST"])
def accept_invite(token):
if session.get("user_id"):
return redirect(url_for("panel.dashboard"))
@@ -103,6 +134,14 @@ def accept_invite(token):
if result.get("error") == "username_or_email_taken":
error = "The invited username or email is already in use. Please contact your administrator."
else:
log_audit_event(
result.get("user_id"), invite["invited_username"],
"invite.accepted",
entity_type="invite", entity_id=invite["id"],
details={"group_id": invite.get("group_id"), "role": invite.get("role")},
group_id=invite.get("group_id"),
ip_address=request.remote_addr,
)
flash("Your account has been created. You can now sign in.", "success")
return redirect(url_for("auth.login"))

43
web/blueprints/group_admin.py
View File

@@ -10,6 +10,7 @@ from config import Config
from mailer import send_mail, build_invite_email, force_https_url
import panel_db as db
from roles import GROUP_MANAGEMENT_ROLES, GROUP_ROLE_OPTIONS, GROUP_ROLE_SET, OWNER_ONLY_ROLES, role_label
from limiter import limiter
group_admin = Blueprint("group_admin", __name__, url_prefix="/group-admin")
@@ -95,12 +96,20 @@ def member_add():
return redirect(url_for("group_admin.members"))
if user_id:
db.add_group_member(user_id, group_id, role)
target_user = db.get_user_by_id(user_id)
db.log_audit_event(
session["user_id"], session["username"], "member.added",
entity_type="user", entity_id=user_id,
details={"role": role, "target": target_user["username"] if target_user else str(user_id)},
group_id=group_id, ip_address=request.remote_addr,
)
flash("Member added.", "success")
return redirect(url_for("group_admin.members"))
@group_admin.route("/members/invite", methods=["POST"])
@group_admin_required
@limiter.limit("30 per hour", methods=["POST"])
def member_invite():
group_id = session["group_id"]
username = request.form.get("username", "").strip()
@@ -146,6 +155,12 @@ def member_invite():
token = db.create_group_invite(group_id, username, email, role, session["user_id"])
invite = db.get_invite_by_token(token)
invite_url = force_https_url(url_for("auth.accept_invite", token=token, _external=True))
db.log_audit_event(
session["user_id"], session["username"], "invite.created",
entity_type="invite", entity_id=invite["id"] if invite else None,
details={"username": username, "email": email, "role": role},
group_id=group_id, ip_address=request.remote_addr,
)
mail_settings = db.get_site_mail_settings()
if mail_settings:
@@ -171,6 +186,7 @@ def member_invite():
@group_admin.route("/invites/<int:invite_id>/resend", methods=["POST"])
@group_admin_required
@limiter.limit("20 per hour", methods=["POST"])
def resend_invite(invite_id):
group_id = session["group_id"]
invite = db.get_group_invite_by_id(invite_id, group_id)
@@ -204,6 +220,12 @@ def resend_invite(invite_id):
try:
send_mail(mail_settings, invite["invited_email"], subject, text_body, html_body=html_body)
db.mark_group_invite_sent(invite_id, group_id)
db.log_audit_event(
session["user_id"], session["username"], "invite.resent",
entity_type="invite", entity_id=invite_id,
details={"to": invite["invited_email"], "username": invite["invited_username"]},
group_id=group_id, ip_address=request.remote_addr,
)
flash("Invitation email resent.", "success")
except Exception:
flash("Resend failed. Please verify SMTP settings and try again.", "danger")
@@ -213,7 +235,14 @@ def resend_invite(invite_id):
@group_admin.route("/invites/<int:invite_id>/revoke", methods=["POST"])
@group_admin_required
def revoke_invite(invite_id):
invite = db.get_group_invite_by_id(invite_id, session["group_id"])
db.revoke_group_invite(invite_id, session["group_id"])
db.log_audit_event(
session["user_id"], session["username"], "invite.revoked",
entity_type="invite", entity_id=invite_id,
details={"username": invite["invited_username"] if invite else None},
group_id=session["group_id"], ip_address=request.remote_addr,
)
flash("Invitation revoked.", "success")
return redirect(url_for("group_admin.members"))
@@ -241,7 +270,14 @@ def member_edit(user_id):
flash("Invalid role selected.", "danger")
return redirect(url_for("group_admin.members"))
new_perms = {key: bool(request.form.get(f"perm_{key}")) for key, _ in ALL_PERMISSIONS}
old_role = member.get("role")
db.update_member(user_id, group_id, role, new_perms)
db.log_audit_event(
session["user_id"], session["username"], "member.updated",
entity_type="user", entity_id=user_id,
details={"target": user["username"], "old_role": old_role, "new_role": role},
group_id=group_id, ip_address=request.remote_addr,
)
flash("Permissions updated.", "success")
return redirect(url_for("group_admin.members"))
@@ -258,7 +294,14 @@ def member_remove(user_id):
if user_id == session["user_id"]:
flash("You cannot remove yourself.", "danger")
else:
target_user = db.get_user_by_id(user_id)
db.remove_group_member(user_id, session["group_id"])
db.log_audit_event(
session["user_id"], session["username"], "member.removed",
entity_type="user", entity_id=user_id,
details={"target": target_user["username"] if target_user else str(user_id)},
group_id=session["group_id"], ip_address=request.remote_addr,
)
flash("Member removed.", "success")
return redirect(url_for("group_admin.members"))

104
web/blueprints/site_admin.py
View File

@@ -9,6 +9,7 @@ from config import Config
from mailer import send_mail, build_invite_email, force_https_url
import panel_db as db
from roles import GROUP_MANAGEMENT_ROLES, GROUP_ROLE_OPTIONS, GROUP_ROLE_SET, role_label
from limiter import limiter
site_admin = Blueprint("site_admin", __name__, url_prefix="/admin")
@@ -52,6 +53,7 @@ def dashboard():
@site_admin.route("/mail", methods=["GET", "POST"])
@admin_required
@limiter.limit("20 per hour", methods=["POST"])
def mail_settings():
settings = db.get_site_mail_settings()
error = None
@@ -101,6 +103,10 @@ def mail_settings():
"Your SMTP settings were verified successfully and have been saved.",
)
db.set_site_mail_settings(host, port, username, password, from_email, from_name, use_tls)
db.log_audit_event(
session["user_id"], session["username"], "mail.settings_saved",
ip_address=request.remote_addr,
)
flash("Mail settings saved and verified.", "success")
return redirect(url_for("site_admin.mail_settings"))
except Exception as exc:
@@ -123,6 +129,10 @@ def mail_settings():
@admin_required
def mail_settings_delete():
db.delete_site_mail_settings()
db.log_audit_event(
session["user_id"], session["username"], "mail.settings_deleted",
ip_address=request.remote_addr,
)
flash("Mail settings removed.", "success")
return redirect(url_for("site_admin.mail_settings"))
@@ -151,7 +161,13 @@ def group_new():
elif db.get_group_by_name(name):
flash("A group with that name already exists.", "danger")
else:
db.create_group(name, desc)
gid = db.create_group(name, desc)
db.log_audit_event(
session["user_id"], session["username"], "group.created",
entity_type="group", entity_id=gid,
details={"name": name},
ip_address=request.remote_addr,
)
flash(f"Group '{name}' created.", "success")
return redirect(url_for("site_admin.groups"))
return render_template("admin/group_edit.html", group=None)
@@ -171,6 +187,12 @@ def group_edit(group_id):
flash("Group name must not be empty.", "danger")
else:
db.update_group(group_id, name, desc)
db.log_audit_event(
session["user_id"], session["username"], "group.updated",
entity_type="group", entity_id=group_id,
details={"name": name},
ip_address=request.remote_addr,
)
flash("Group updated.", "success")
return redirect(url_for("site_admin.groups"))
return render_template("admin/group_edit.html", group=group)
@@ -179,7 +201,14 @@ def group_edit(group_id):
@site_admin.route("/groups/<int:group_id>/delete", methods=["POST"])
@admin_required
def group_delete(group_id):
group = db.get_group_by_id(group_id)
db.delete_group(group_id)
db.log_audit_event(
session["user_id"], session["username"], "group.deleted",
entity_type="group", entity_id=group_id,
details={"name": group["name"] if group else None},
ip_address=request.remote_addr,
)
flash("Group deleted.", "success")
return redirect(url_for("site_admin.groups"))
@@ -211,6 +240,13 @@ def group_member_add(group_id):
return redirect(url_for("site_admin.group_members", group_id=group_id))
if user_id:
db.add_group_member(user_id, group_id, role)
target = db.get_user_by_id(user_id)
db.log_audit_event(
session["user_id"], session["username"], "member.added",
entity_type="user", entity_id=user_id,
details={"target": target["username"] if target else str(user_id), "role": role},
group_id=group_id, ip_address=request.remote_addr,
)
flash("Member added.", "success")
return redirect(url_for("site_admin.group_members", group_id=group_id))
@@ -218,7 +254,14 @@ def group_member_add(group_id):
@site_admin.route("/groups/<int:group_id>/members/<int:user_id>/remove", methods=["POST"])
@admin_required
def group_member_remove(group_id, user_id):
target = db.get_user_by_id(user_id)
db.remove_group_member(user_id, group_id)
db.log_audit_event(
session["user_id"], session["username"], "member.removed",
entity_type="user", entity_id=user_id,
details={"target": target["username"] if target else str(user_id)},
group_id=group_id, ip_address=request.remote_addr,
)
flash("Member removed.", "success")
return redirect(url_for("site_admin.group_members", group_id=group_id))
@@ -234,7 +277,15 @@ def group_member_set_role(group_id, user_id):
flash("Invalid role selected.", "danger")
return redirect(url_for("site_admin.group_members", group_id=group_id))
perms = member["permissions"] if isinstance(member["permissions"], dict) else (_json.loads(member["permissions"]) if member["permissions"] else {})
old_role = member.get("role")
db.update_member(user_id, group_id, new_role, perms)
target = db.get_user_by_id(user_id)
db.log_audit_event(
session["user_id"], session["username"], "member.role_changed",
entity_type="user", entity_id=user_id,
details={"target": target["username"] if target else str(user_id), "old_role": old_role, "new_role": new_role},
group_id=group_id, ip_address=request.remote_addr,
)
flash(f"Role changed to '{new_role}'.", "success")
return redirect(url_for("site_admin.group_members", group_id=group_id))
@@ -507,6 +558,12 @@ def user_edit(user_id):
if new_password:
db.change_password(user_id, new_password)
flash("Password changed.", "info")
db.log_audit_event(
session["user_id"], session["username"], "user.updated",
entity_type="user", entity_id=user_id,
details={"username": username, "is_site_admin": is_site_admin},
ip_address=request.remote_addr,
)
flash("User updated.", "success")
return redirect(url_for("site_admin.users"))
return render_template("admin/user_edit.html", user=user)
@@ -518,7 +575,14 @@ def user_delete(user_id):
if user_id == session.get("user_id"):
flash("You cannot delete yourself.", "danger")
else:
target = db.get_user_by_id(user_id)
db.delete_user(user_id)
db.log_audit_event(
session["user_id"], session["username"], "user.deleted",
entity_type="user", entity_id=user_id,
details={"username": target["username"] if target else str(user_id)},
ip_address=request.remote_addr,
)
flash("User deleted.", "success")
return redirect(url_for("site_admin.users"))
@@ -568,3 +632,41 @@ def stop_view():
session.pop("permissions", None)
session.pop("admin_viewing", None)
return redirect(url_for("site_admin.dashboard"))
# ──────────────────────────────────────────────────────────────
# Audit-Log
# ──────────────────────────────────────────────────────────────
@site_admin.route("/audit")
@admin_required
def audit_log():
page = request.args.get("page", 1, type=int)
action_f = request.args.get("action", "").strip() or None
group_f = request.args.get("group_id", None, type=int)
actor_f = request.args.get("actor", "").strip() or None
per_page = 50
rows, total = db.get_audit_log(
page=page, per_page=per_page,
action_filter=action_f,
group_id_filter=group_f,
actor_filter=actor_f,
)
total_pages = max(1, (total + per_page - 1) // per_page)
all_groups = db.list_all_groups() or []
actions = db.get_audit_log_distinct_actions()
return render_template(
"admin/audit_log.html",
rows=rows,
total=total,
page=page,
total_pages=total_pages,
per_page=per_page,
action_filter=action_f or "",
group_filter=group_f,
actor_filter=actor_f or "",
all_groups=all_groups,
actions=actions,
)

14
web/limiter.py Normal file
View File

@@ -0,0 +1,14 @@
"""
MCLogger – Rate-Limiter Singleton
Shared across app.py and all blueprints to avoid circular imports.
"""
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
# In-memory storage is fine for single-process / single-worker deployments.
# For multi-worker gunicorn, set RATELIMIT_STORAGE_URI=redis://... in ENV.
limiter = Limiter(
key_func=get_remote_address,
storage_uri="memory://",
default_limits=[],
)

187
web/panel_db.py
View File

@@ -119,6 +119,56 @@ PANEL_SCHEMA = [
FOREIGN KEY (group_id) REFERENCES user_groups(id) ON DELETE CASCADE,
FOREIGN KEY (created_by_user_id) REFERENCES users(id) ON DELETE CASCADE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""",
"""CREATE TABLE IF NOT EXISTS audit_log (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
actor_user_id INT NULL,
actor_username VARCHAR(50) NULL,
action VARCHAR(100) NOT NULL,
entity_type VARCHAR(50) NULL,
entity_id VARCHAR(100) NULL,
details JSON NULL,
group_id INT NULL,
ip_address VARCHAR(45) NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_audit_actor (actor_user_id),
INDEX idx_audit_group (group_id),
INDEX idx_audit_action (action),
INDEX idx_audit_ts (created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""",
"""CREATE TABLE IF NOT EXISTS schema_migrations (
version INT PRIMARY KEY,
applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
note VARCHAR(255) NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""",
]
# ─────────────────────────────────────────────────────────────
# Versioned migrations (applied once, tracked in schema_migrations)
# Each entry: (version_int, sql_statement, human_readable_note)
# ─────────────────────────────────────────────────────────────
PANEL_MIGRATIONS = [
(1,
"ALTER TABLE group_members MODIFY COLUMN role "
"ENUM('group_owner','group_admin','moderator','viewer','auditor','admin','member') DEFAULT 'viewer'",
"Extend group_members.role ENUM"),
(2,
"ALTER TABLE group_invites MODIFY COLUMN role "
"ENUM('group_owner','group_admin','moderator','viewer','auditor','admin','member') DEFAULT 'viewer'",
"Extend group_invites.role ENUM"),
(3,
"ALTER TABLE group_invites ADD COLUMN IF NOT EXISTS last_sent_at DATETIME NULL",
"Add group_invites.last_sent_at"),
(4,
"ALTER TABLE group_invites ADD COLUMN IF NOT EXISTS send_count INT NOT NULL DEFAULT 0",
"Add group_invites.send_count"),
(5,
"ALTER TABLE group_invites MODIFY COLUMN group_id INT NULL",
"Allow group_invites.group_id to be NULL"),
(6,
"ALTER TABLE group_invites ADD COLUMN IF NOT EXISTS is_site_admin TINYINT(1) NOT NULL DEFAULT 0",
"Add group_invites.is_site_admin"),
]
CREDS_SCHEMA = [
@@ -149,43 +199,33 @@ CREDS_SCHEMA = [
def init_databases():
"""Erstellt alle benötigten Tabellen falls nicht vorhanden."""
"""Creates all required tables and applies pending schema migrations."""
import logging
_log = logging.getLogger(__name__)
panel = get_panel_db()
try:
with panel.cursor() as cur:
# Create tables (idempotent)
for stmt in PANEL_SCHEMA:
cur.execute(stmt)
# Best-effort migrations for existing installs.
# Determine already-applied migration versions
cur.execute("SELECT version FROM schema_migrations")
applied = {row["version"] for row in cur.fetchall()}
for version, sql, note in PANEL_MIGRATIONS:
if version in applied:
continue
try:
cur.execute(sql)
cur.execute(
"ALTER TABLE group_members MODIFY role ENUM('group_owner','group_admin','moderator','viewer','auditor','admin','member') DEFAULT 'viewer'"
"INSERT IGNORE INTO schema_migrations (version, note) VALUES (%s, %s)",
(version, note),
)
except Exception:
pass
try:
cur.execute(
"ALTER TABLE group_invites MODIFY role ENUM('group_owner','group_admin','moderator','viewer','auditor','admin','member') DEFAULT 'viewer'"
)
except Exception:
pass
try:
cur.execute("ALTER TABLE group_invites ADD COLUMN last_sent_at DATETIME NULL")
except Exception:
pass
try:
cur.execute("ALTER TABLE group_invites ADD COLUMN send_count INT NOT NULL DEFAULT 0")
except Exception:
pass
try:
cur.execute("SET foreign_key_checks=0")
cur.execute("ALTER TABLE group_invites MODIFY group_id INT NULL")
cur.execute("SET foreign_key_checks=1")
except Exception:
pass
try:
cur.execute("ALTER TABLE group_invites ADD COLUMN is_site_admin TINYINT(1) NOT NULL DEFAULT 0")
except Exception:
pass
_log.info("Migration %d applied: %s", version, note)
except Exception as exc:
_log.warning("Migration %d skipped (%s): %s", version, note, exc)
finally:
panel.close()
@@ -683,3 +723,92 @@ def delete_site_mail_settings():
def has_site_mail_settings() -> bool:
row = _creds_query("SELECT id FROM site_mail_settings WHERE config_key=%s", ("primary",), fetchone=True)
return row is not None
# ─────────────────────────────────────────────────────────────
# Audit-Log
# ─────────────────────────────────────────────────────────────
def log_audit_event(
actor_user_id,
actor_username: str | None,
action: str,
entity_type: str | None = None,
entity_id: str | None = None,
details: dict | None = None,
group_id: int | None = None,
ip_address: str | None = None,
):
"""Records an audit event. Never raises — audit log must not break the main flow."""
try:
_panel_query(
"INSERT INTO audit_log "
"(actor_user_id, actor_username, action, entity_type, entity_id, details, group_id, ip_address) "
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s)",
(
actor_user_id,
actor_username,
action,
entity_type,
str(entity_id) if entity_id is not None else None,
json.dumps(details) if details else None,
group_id,
ip_address,
),
write=True,
)
except Exception:
import logging
logging.getLogger(__name__).warning("Failed to write audit event: %s", action)
def get_audit_log(
page: int = 1,
per_page: int = 50,
action_filter: str | None = None,
group_id_filter: int | None = None,
actor_filter: str | None = None,
):
offset = (page - 1) * per_page
conditions: list[str] = []
args: list = []
if action_filter:
conditions.append("al.action LIKE %s")
args.append(f"%{action_filter}%")
if group_id_filter:
conditions.append("al.group_id = %s")
args.append(group_id_filter)
if actor_filter:
conditions.append("al.actor_username LIKE %s")
args.append(f"%{actor_filter}%")
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
count_row = _panel_query(
f"SELECT COUNT(*) AS c FROM audit_log al {where}", args, fetchone=True
)
total = int(count_row["c"]) if count_row else 0
rows = _panel_query(
f"SELECT al.*, g.name AS group_name "
f"FROM audit_log al "
f"LEFT JOIN user_groups g ON g.id = al.group_id "
f"{where} ORDER BY al.created_at DESC LIMIT %s OFFSET %s",
args + [per_page, offset],
)
# Ensure details is always a dict (pymysql may return JSON as string)
for row in (rows or []):
d = row.get("details")
if isinstance(d, str):
try:
row["details"] = json.loads(d)
except Exception:
row["details"] = {}
return rows, total
def get_audit_log_distinct_actions() -> list[str]:
rows = _panel_query("SELECT DISTINCT action FROM audit_log ORDER BY action")
return [r["action"] for r in rows] if rows else []

1
web/requirements.txt
View File

@@ -2,3 +2,4 @@ Flask==3.1.0
PyMySQL==1.1.1
cryptography==42.0.8
gunicorn==22.0.0
flask-limiter==3.9.0

31
web/templates/429.html Normal file
View File

@@ -0,0 +1,31 @@
<!DOCTYPE html>
<html lang="en" data-bs-theme="dark">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Too Many Requests — MCLogger</title>
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/css/bootstrap.min.css">
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.3/font/bootstrap-icons.min.css">
<style>
body { display: flex; align-items: center; justify-content: center; min-height: 100vh; background: #0d1117; }
</style>
</head>
<body>
<div class="text-center p-4" style="max-width:420px">
<i class="bi bi-shield-exclamation text-warning" style="font-size:3rem"></i>
<h2 class="fw-bold mt-3">Too Many Requests</h2>
<p class="text-muted">You have submitted this form too frequently. Please wait
{% if retry_after %}
<strong>{{ retry_after }} second{{ 's' if retry_after != 1 }}</strong>
{% else %}
a moment
{% endif %}
before trying again.
</p>
<a href="javascript:history.back()" class="btn btn-outline-secondary mt-2">
<i class="bi bi-arrow-left me-1"></i>Go back
</a>
</div>
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/js/bootstrap.bundle.min.js"></script>
</body>
</html>

157
web/templates/admin/audit_log.html Normal file
View File

@@ -0,0 +1,157 @@
{% extends "admin/base.html" %}
{% block title %}Audit Log{% endblock %}
{% block content %}
<div class="d-flex align-items-center justify-content-between mb-3">
<h4 class="mb-0"><i class="bi bi-journal-text me-2"></i>Audit Log</h4>
<span class="text-muted small">{{ total }} event{{ 's' if total != 1 }}</span>
</div>
<!-- Filters -->
<form method="get" action="{{ url_for('site_admin.audit_log') }}" class="card border-secondary mb-4">
<div class="card-body py-2">
<div class="row g-2 align-items-end">
<div class="col-md-3">
<label class="form-label form-label-sm mb-1">Action</label>
<select name="action" class="form-select form-select-sm bg-dark text-white border-secondary">
<option value="">— All actions —</option>
{% for a in actions %}
<option value="{{ a }}" {{ 'selected' if action_filter == a }}>{{ a }}</option>
{% endfor %}
</select>
</div>
<div class="col-md-3">
<label class="form-label form-label-sm mb-1">Group</label>
<select name="group_id" class="form-select form-select-sm bg-dark text-white border-secondary">
<option value="">— All groups —</option>
{% for g in all_groups %}
<option value="{{ g.id }}" {{ 'selected' if group_filter == g.id }}>{{ g.name }}</option>
{% endfor %}
</select>
</div>
<div class="col-md-3">
<label class="form-label form-label-sm mb-1">Actor</label>
<input type="text" name="actor" class="form-control form-control-sm bg-dark text-white border-secondary"
placeholder="Username…" value="{{ actor_filter }}">
</div>
<div class="col-md-3 d-flex gap-2">
<button type="submit" class="btn btn-sm btn-primary w-100">
<i class="bi bi-funnel-fill me-1"></i>Filter
</button>
<a href="{{ url_for('site_admin.audit_log') }}" class="btn btn-sm btn-outline-secondary w-100">
<i class="bi bi-x-lg"></i>
</a>
</div>
</div>
</div>
</form>
<!-- Table -->
<div class="card border-secondary">
<div class="table-responsive">
<table class="table table-dark table-hover table-sm align-middle mb-0">
<thead class="table-secondary text-dark">
<tr>
<th style="width:155px">Timestamp (UTC)</th>
<th style="width:130px">Actor</th>
<th style="width:180px">Action</th>
<th style="width:90px">Entity</th>
<th style="width:80px">Entity ID</th>
<th style="width:120px">Group</th>
<th>Details</th>
<th style="width:110px">IP Address</th>
</tr>
</thead>
<tbody>
{% for row in rows %}
<tr>
<td class="text-muted small">{{ row.created_at | fmt_dt }}</td>
<td>
{% if row.actor_username %}
<span class="text-info">{{ row.actor_username }}</span>
{% else %}
<span class="text-muted">—</span>
{% endif %}
</td>
<td>
{% set action_class = {
'user.login': 'badge bg-success',
'user.login_failed': 'badge bg-danger',
'admin.login': 'badge bg-warning text-dark',
'admin.login_failed': 'badge bg-danger',
'invite.created': 'badge bg-primary',
'invite.accepted': 'badge bg-success',
'invite.revoked': 'badge bg-secondary',
'invite.resent': 'badge bg-info text-dark',
'member.added': 'badge bg-primary',
'member.removed': 'badge bg-danger',
'member.role_changed': 'badge bg-warning text-dark',
'member.updated': 'badge bg-warning text-dark',
'group.created': 'badge bg-success',
'group.updated': 'badge bg-secondary',
'group.deleted': 'badge bg-danger',
'user.updated': 'badge bg-secondary',
'user.deleted': 'badge bg-danger',
'mail.settings_saved': 'badge bg-info text-dark',
'mail.settings_deleted':'badge bg-danger',
} %}
<span class="{{ action_class.get(row.action, 'badge bg-secondary') }} font-monospace" style="font-size:.75em">
{{ row.action }}
</span>
</td>
<td class="text-muted small">{{ row.entity_type or '—' }}</td>
<td class="text-muted small font-monospace">{{ row.entity_id or '—' }}</td>
<td class="small">
{% if row.group_name %}
<span class="badge bg-dark border border-secondary">{{ row.group_name }}</span>
{% else %}
<span class="text-muted">—</span>
{% endif %}
</td>
<td class="small text-muted font-monospace">
{% if row.details %}
{% set d = row.details if row.details is mapping else {} %}
{% for k, v in d.items() %}
<span class="me-2"><strong>{{ k }}:</strong> {{ v }}</span>
{% endfor %}
{% else %}
—
{% endif %}
</td>
<td class="text-muted small font-monospace">{{ row.ip_address or '—' }}</td>
</tr>
{% else %}
<tr>
<td colspan="8" class="text-center text-muted py-4">
<i class="bi bi-journal-x me-2"></i>No audit events found.
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
<!-- Pagination -->
{% if total_pages > 1 %}
<nav class="mt-3">
<ul class="pagination pagination-sm justify-content-center mb-0">
<li class="page-item {{ 'disabled' if page <= 1 }}">
<a class="page-link" href="{{ url_for('site_admin.audit_log', page=page-1, action=action_filter, group_id=group_filter, actor=actor_filter) }}">
<i class="bi bi-chevron-left"></i>
</a>
</li>
{% for p in range([1, page-2]|max, [total_pages+1, page+3]|min) %}
<li class="page-item {{ 'active' if p == page }}">
<a class="page-link" href="{{ url_for('site_admin.audit_log', page=p, action=action_filter, group_id=group_filter, actor=actor_filter) }}">{{ p }}</a>
</li>
{% endfor %}
<li class="page-item {{ 'disabled' if page >= total_pages }}">
<a class="page-link" href="{{ url_for('site_admin.audit_log', page=page+1, action=action_filter, group_id=group_filter, actor=actor_filter) }}">
<i class="bi bi-chevron-right"></i>
</a>
</li>
</ul>
</nav>
{% endif %}
{% endblock %}

1
web/templates/admin/base.html
View File

@@ -19,6 +19,7 @@
<a href="{{ url_for('site_admin.groups') }}" class="nav-link text-white {{ 'fw-bold' if request.endpoint == 'site_admin.groups' }}">Groups</a>
<a href="{{ url_for('site_admin.users') }}" class="nav-link text-white {{ 'fw-bold' if request.endpoint == 'site_admin.users' }}">Users</a>
<a href="{{ url_for('site_admin.mail_settings') }}" class="nav-link text-white {{ 'fw-bold' if request.endpoint == 'site_admin.mail_settings' }}">Mail</a>
<a href="{{ url_for('site_admin.audit_log') }}" class="nav-link text-white {{ 'fw-bold' if request.endpoint == 'site_admin.audit_log' }}">Audit Log</a>
<form method="post" action="{{ url_for('auth.logout') }}" class="d-inline">
<input type="hidden" name="_csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="btn btn-outline-light btn-sm">