Add web admin panel for accounts/characters (login, GM rights, live stats)
New "webadmin" service (Flask + PyMySQL, port 8081) backing a small server-rendered admin UI: - Login gated on an existing game account: only accounts with accesslevel >= ADMIN_MIN_ACCESSLEVEL (default 900) get in. Password is MD5-hashed before comparison since the game client hashes it client-side before it ever hits the wire -- accounts.password is always an MD5 hex digest, confirmed against existing DB rows. - Accounts: list/search, create, edit accesslevel/active/password. - Characters: list/search, create (mirrors the real char-creation INSERT in Src/Char Server/charpackets.cpp -- only account_name, char_name, face, hairStyle, sex are set, everything else relies on the table's own defaults), edit core stats (level, zuly, attributes, hp/mp, stat/skill points) and isgm/isdev flags directly. All queries are parameterized (no string-formatted SQL). No CSRF protection or rate limiting yet -- acceptable for a small already-behind-a-high-accesslevel-gate private server, worth revisiting before ever exposing this port publicly. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
ThomasandClaude Sonnet 5 committed
1 parent
f053a06920
commit
9829688a85
14 files changed
+610
No files matched your search
+288
@@ -0,0 +1,288 @@
|
||||
import os
|
||||
import functools
|
||||
import hashlib
|
||||
|
||||
import pymysql
|
||||
import pymysql.cursors
|
||||
from flask import Flask, request, session, redirect, url_for, render_template, flash
|
||||
|
||||
app = Flask(__name__)
|
||||
app.secret_key = os.environ.get("ADMIN_SECRET_KEY", "dev-only-change-me")
|
||||
|
||||
|
||||
def hash_password(plaintext):
|
||||
# The game client MD5-hashes the password client-side before it ever
|
||||
# hits the wire, so accounts.password is always an MD5 hex digest --
|
||||
# match that here so admin-created/changed passwords work in-game.
|
||||
return hashlib.md5(plaintext.encode("utf-8")).hexdigest()
|
||||
|
||||
DB_HOST = os.environ.get("DB_HOST", "db")
|
||||
DB_PORT = int(os.environ.get("DB_PORT", "3306"))
|
||||
DB_USER = os.environ.get("DB_USER", "rose")
|
||||
DB_PASSWORD = os.environ.get("DB_PASSWORD", "change_me")
|
||||
DB_NAME = os.environ.get("DB_NAME", "ktrose")
|
||||
|
||||
# Only accounts with accesslevel >= this may log in to the admin panel.
|
||||
ADMIN_MIN_ACCESSLEVEL = int(os.environ.get("ADMIN_MIN_ACCESSLEVEL", "900"))
|
||||
|
||||
CHAR_STAT_FIELDS = [
|
||||
"level", "zuly", "str", "con", "dex", "_int", "cha", "sen",
|
||||
"curHp", "curMp", "statp", "skillp", "stamina",
|
||||
]
|
||||
|
||||
|
||||
def get_db():
|
||||
return pymysql.connect(
|
||||
host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD,
|
||||
database=DB_NAME, cursorclass=pymysql.cursors.DictCursor, autocommit=True,
|
||||
)
|
||||
|
||||
|
||||
def login_required(view):
|
||||
@functools.wraps(view)
|
||||
def wrapped(*args, **kwargs):
|
||||
if not session.get("username"):
|
||||
return redirect(url_for("login", next=request.path))
|
||||
return view(*args, **kwargs)
|
||||
return wrapped
|
||||
|
||||
|
||||
@app.route("/login", methods=["GET", "POST"])
|
||||
def login():
|
||||
if request.method == "POST":
|
||||
username = request.form.get("username", "").strip()
|
||||
password = request.form.get("password", "")
|
||||
db = get_db()
|
||||
try:
|
||||
with db.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT username, password, accesslevel FROM accounts WHERE username=%s",
|
||||
(username,),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
finally:
|
||||
db.close()
|
||||
if (
|
||||
row
|
||||
and row["password"].lower() == hash_password(password)
|
||||
and row["accesslevel"] >= ADMIN_MIN_ACCESSLEVEL
|
||||
):
|
||||
session.clear()
|
||||
session["username"] = row["username"]
|
||||
return redirect(request.args.get("next") or url_for("dashboard"))
|
||||
flash("Login fehlgeschlagen (falsches Passwort oder unzureichendes Accesslevel).")
|
||||
return render_template("login.html")
|
||||
|
||||
|
||||
@app.route("/logout")
|
||||
def logout():
|
||||
session.clear()
|
||||
return redirect(url_for("login"))
|
||||
|
||||
|
||||
@app.route("/")
|
||||
@login_required
|
||||
def dashboard():
|
||||
db = get_db()
|
||||
try:
|
||||
with db.cursor() as cur:
|
||||
cur.execute("SELECT COUNT(*) AS n FROM accounts")
|
||||
accounts_n = cur.fetchone()["n"]
|
||||
cur.execute("SELECT COUNT(*) AS n FROM characters")
|
||||
chars_n = cur.fetchone()["n"]
|
||||
finally:
|
||||
db.close()
|
||||
return render_template("dashboard.html", accounts_n=accounts_n, chars_n=chars_n)
|
||||
|
||||
|
||||
@app.route("/accounts")
|
||||
@login_required
|
||||
def accounts():
|
||||
q = request.args.get("q", "").strip()
|
||||
db = get_db()
|
||||
try:
|
||||
with db.cursor() as cur:
|
||||
if q:
|
||||
cur.execute(
|
||||
"SELECT id, username, accesslevel, active, online, lastip "
|
||||
"FROM accounts WHERE username LIKE %s ORDER BY id",
|
||||
(f"%{q}%",),
|
||||
)
|
||||
else:
|
||||
cur.execute(
|
||||
"SELECT id, username, accesslevel, active, online, lastip "
|
||||
"FROM accounts ORDER BY id"
|
||||
)
|
||||
rows = cur.fetchall()
|
||||
finally:
|
||||
db.close()
|
||||
return render_template("accounts.html", accounts=rows, q=q)
|
||||
|
||||
|
||||
@app.route("/accounts/new", methods=["GET", "POST"])
|
||||
@login_required
|
||||
def new_account():
|
||||
if request.method == "POST":
|
||||
username = request.form.get("username", "").strip()
|
||||
password = request.form.get("password", "").strip()
|
||||
accesslevel = int(request.form.get("accesslevel", "100") or "100")
|
||||
if not username or not password:
|
||||
flash("Username und Passwort sind Pflichtfelder.")
|
||||
return render_template("new_account.html")
|
||||
db = get_db()
|
||||
try:
|
||||
with db.cursor() as cur:
|
||||
cur.execute("SELECT id FROM accounts WHERE username=%s", (username,))
|
||||
if cur.fetchone():
|
||||
flash(f"Account '{username}' existiert bereits.")
|
||||
return render_template("new_account.html")
|
||||
cur.execute(
|
||||
"INSERT INTO accounts (username, password, accesslevel, lastlogin, logtime) "
|
||||
"VALUES (%s, %s, %s, '', 0)",
|
||||
(username, hash_password(password), accesslevel),
|
||||
)
|
||||
finally:
|
||||
db.close()
|
||||
flash(f"Account '{username}' angelegt.")
|
||||
return redirect(url_for("accounts"))
|
||||
return render_template("new_account.html")
|
||||
|
||||
|
||||
@app.route("/accounts/<int:account_id>/edit", methods=["GET", "POST"])
|
||||
@login_required
|
||||
def edit_account(account_id):
|
||||
db = get_db()
|
||||
try:
|
||||
if request.method == "POST":
|
||||
accesslevel = int(request.form.get("accesslevel", "100") or "100")
|
||||
active = 1 if request.form.get("active") else 0
|
||||
new_password = request.form.get("password", "").strip()
|
||||
with db.cursor() as cur:
|
||||
if new_password:
|
||||
cur.execute(
|
||||
"UPDATE accounts SET accesslevel=%s, active=%s, password=%s WHERE id=%s",
|
||||
(accesslevel, active, hash_password(new_password), account_id),
|
||||
)
|
||||
else:
|
||||
cur.execute(
|
||||
"UPDATE accounts SET accesslevel=%s, active=%s WHERE id=%s",
|
||||
(accesslevel, active, account_id),
|
||||
)
|
||||
flash("Account aktualisiert.")
|
||||
return redirect(url_for("accounts"))
|
||||
with db.cursor() as cur:
|
||||
cur.execute("SELECT * FROM accounts WHERE id=%s", (account_id,))
|
||||
account = cur.fetchone()
|
||||
cur.execute(
|
||||
"SELECT id, char_name, level FROM characters WHERE account_name=%s",
|
||||
(account["username"],),
|
||||
)
|
||||
chars = cur.fetchall()
|
||||
finally:
|
||||
db.close()
|
||||
if not account:
|
||||
flash("Account nicht gefunden.")
|
||||
return redirect(url_for("accounts"))
|
||||
return render_template("edit_account.html", account=account, chars=chars)
|
||||
|
||||
|
||||
@app.route("/characters")
|
||||
@login_required
|
||||
def characters():
|
||||
q = request.args.get("q", "").strip()
|
||||
db = get_db()
|
||||
try:
|
||||
with db.cursor() as cur:
|
||||
if q:
|
||||
cur.execute(
|
||||
"SELECT id, char_name, account_name, level, isgm, isdev "
|
||||
"FROM characters WHERE char_name LIKE %s OR account_name LIKE %s ORDER BY id",
|
||||
(f"%{q}%", f"%{q}%"),
|
||||
)
|
||||
else:
|
||||
cur.execute(
|
||||
"SELECT id, char_name, account_name, level, isgm, isdev "
|
||||
"FROM characters ORDER BY id"
|
||||
)
|
||||
rows = cur.fetchall()
|
||||
finally:
|
||||
db.close()
|
||||
return render_template("characters.html", chars=rows, q=q)
|
||||
|
||||
|
||||
@app.route("/characters/new", methods=["GET", "POST"])
|
||||
@login_required
|
||||
def new_character():
|
||||
db = get_db()
|
||||
try:
|
||||
with db.cursor() as cur:
|
||||
cur.execute("SELECT username FROM accounts ORDER BY username")
|
||||
account_names = [r["username"] for r in cur.fetchall()]
|
||||
if request.method == "POST":
|
||||
account_name = request.form.get("account_name", "").strip()
|
||||
char_name = request.form.get("char_name", "").strip()
|
||||
sex = int(request.form.get("sex", "0") or "0")
|
||||
face = int(request.form.get("face", "1") or "1")
|
||||
hairstyle = int(request.form.get("hairstyle", "1") or "1")
|
||||
if not account_name or not char_name:
|
||||
flash("Account und Charaktername sind Pflichtfelder.")
|
||||
return render_template("new_character.html", account_names=account_names)
|
||||
with db.cursor() as cur:
|
||||
cur.execute("SELECT id FROM characters WHERE char_name=%s", (char_name,))
|
||||
if cur.fetchone():
|
||||
flash(f"Charakter '{char_name}' existiert bereits.")
|
||||
return render_template("new_character.html", account_names=account_names)
|
||||
# Mirrors the real character-creation INSERT in
|
||||
# Src/Char Server/charpackets.cpp -- every other column
|
||||
# relies on its table default (level 1, zero stats, etc).
|
||||
cur.execute(
|
||||
"INSERT INTO characters (account_name, char_name, face, hairStyle, sex) "
|
||||
"VALUES (%s, %s, %s, %s, %s)",
|
||||
(account_name, char_name, face, hairstyle, sex),
|
||||
)
|
||||
flash(f"Charakter '{char_name}' angelegt.")
|
||||
return redirect(url_for("characters"))
|
||||
finally:
|
||||
db.close()
|
||||
return render_template("new_character.html", account_names=account_names)
|
||||
|
||||
|
||||
@app.route("/characters/<int:char_id>/edit", methods=["GET", "POST"])
|
||||
@login_required
|
||||
def edit_character(char_id):
|
||||
db = get_db()
|
||||
try:
|
||||
if request.method == "POST":
|
||||
values = []
|
||||
set_clauses = []
|
||||
for field in CHAR_STAT_FIELDS:
|
||||
raw = request.form.get(field, "").strip()
|
||||
if raw == "":
|
||||
continue
|
||||
set_clauses.append(f"`{field}`=%s")
|
||||
values.append(int(raw))
|
||||
isgm = 1 if request.form.get("isgm") else 0
|
||||
isdev = 1 if request.form.get("isdev") else 0
|
||||
set_clauses += ["isgm=%s", "isdev=%s"]
|
||||
values += [isgm, isdev]
|
||||
values.append(char_id)
|
||||
with db.cursor() as cur:
|
||||
cur.execute(
|
||||
f"UPDATE characters SET {', '.join(set_clauses)} WHERE id=%s",
|
||||
values,
|
||||
)
|
||||
flash("Charakter aktualisiert.")
|
||||
return redirect(url_for("characters"))
|
||||
with db.cursor() as cur:
|
||||
cur.execute("SELECT * FROM characters WHERE id=%s", (char_id,))
|
||||
char = cur.fetchone()
|
||||
finally:
|
||||
db.close()
|
||||
if not char:
|
||||
flash("Charakter nicht gefunden.")
|
||||
return redirect(url_for("characters"))
|
||||
return render_template("edit_character.html", char=char, stat_fields=CHAR_STAT_FIELDS)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run(host="0.0.0.0", port=8080, debug=False)
|
||||
Reference in new issue
Block a user