ThomasandClaude Sonnet 5
315f64cf95
Traced how login placement actually works: CPlayer login placement (Src/World Server/worldpackets.cpp:27, pakPlayer) calls GetRespawnByID(Position->respawn), and Position->respawn is loaded straight from characters.respawnid (playerdata.cpp:110). So setting respawnid to a Src/World Server list_respawnzones.id is a real, live lever over where a character logs in next -- not a vestigial column (unlike characters.townid, which turned out to be dead: the only other code path that reads it is commented-out dead code in pakUserDied, superseded by CPlayer::RespawnPlayer which uses an in-memory Position->saved instead). Added to the character edit page: a teleport dropdown grouped by map, populated from list_respawnzones, wired straight to respawnid (takes effect on the character's next login same as other stat edits) and a delete-character action. Added to the account edit page: delete account (characters are kept) and a "reset online flag" action for accounts stuck thinking they're still logged in after a crash. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
356 lines
13 KiB
Python
356 lines
13 KiB
Python
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", "exp", "classid",
|
|
]
|
|
|
|
# Column edited via its own <select> (populated from list_respawnzones) rather
|
|
# than a bare number input, but it's a plain int column just like the ones
|
|
# above, so the generic CHAR_STAT_FIELDS UPDATE loop handles it unchanged.
|
|
CHAR_STAT_FIELDS.append("respawnid")
|
|
|
|
|
|
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()
|
|
cur.execute(
|
|
"SELECT id, map, x, y FROM list_respawnzones ORDER BY map, id"
|
|
)
|
|
respawn_zones = cur.fetchall()
|
|
finally:
|
|
db.close()
|
|
if not char:
|
|
flash("Charakter nicht gefunden.")
|
|
return redirect(url_for("characters"))
|
|
zones_by_map = {}
|
|
for z in respawn_zones:
|
|
zones_by_map.setdefault(z["map"], []).append(z)
|
|
return render_template(
|
|
"edit_character.html",
|
|
char=char,
|
|
stat_fields=[f for f in CHAR_STAT_FIELDS if f != "respawnid"],
|
|
zones_by_map=zones_by_map,
|
|
)
|
|
|
|
|
|
@app.route("/characters/<int:char_id>/delete", methods=["POST"])
|
|
@login_required
|
|
def delete_character(char_id):
|
|
db = get_db()
|
|
try:
|
|
with db.cursor() as cur:
|
|
cur.execute("SELECT char_name FROM characters WHERE id=%s", (char_id,))
|
|
row = cur.fetchone()
|
|
if row:
|
|
cur.execute("DELETE FROM characters WHERE id=%s", (char_id,))
|
|
finally:
|
|
db.close()
|
|
if row:
|
|
flash(f"Charakter '{row['char_name']}' gelöscht.")
|
|
return redirect(url_for("characters"))
|
|
|
|
|
|
@app.route("/accounts/<int:account_id>/delete", methods=["POST"])
|
|
@login_required
|
|
def delete_account(account_id):
|
|
db = get_db()
|
|
try:
|
|
with db.cursor() as cur:
|
|
cur.execute("SELECT username FROM accounts WHERE id=%s", (account_id,))
|
|
row = cur.fetchone()
|
|
if row:
|
|
cur.execute("DELETE FROM accounts WHERE id=%s", (account_id,))
|
|
finally:
|
|
db.close()
|
|
if row:
|
|
flash(f"Account '{row['username']}' gelöscht (Charaktere bleiben erhalten).")
|
|
return redirect(url_for("accounts"))
|
|
|
|
|
|
@app.route("/accounts/<int:account_id>/kick", methods=["POST"])
|
|
@login_required
|
|
def kick_account(account_id):
|
|
# Resets the DB's stale "online" flag -- useful after a crash/kill -9
|
|
# left an account stuck thinking it's still logged in and unable to
|
|
# log back in. Does not disconnect a currently-live session.
|
|
db = get_db()
|
|
try:
|
|
with db.cursor() as cur:
|
|
cur.execute("UPDATE accounts SET online=0 WHERE id=%s", (account_id,))
|
|
finally:
|
|
db.close()
|
|
flash("Online-Status zurückgesetzt.")
|
|
return redirect(url_for("edit_account", account_id=account_id))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=8080, debug=False)
|