ROSS/webadmin/app.py

import os
import functools
import hashlib

import pymysql
import pymysql.cursors
from flask import Flask, request, session, redirect, url_for, render_template, flash
from werkzeug.middleware.proxy_fix import ProxyFix

app = Flask(__name__)
app.secret_key = os.environ.get("ADMIN_SECRET_KEY", "dev-only-change-me")
# Sits behind the webtools nginx container at /admin/ (see webtools/nginx.conf);
# this reads X-Forwarded-Prefix/-Proto/-For so url_for()/redirect() produce
# correct /admin/-prefixed URLs instead of root-relative ones.
app.wsgi_app = ProxyFix(app.wsgi_app, x_prefix=1, x_proto=1, x_for=1, x_host=1)


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":
            with db.cursor() as cur:
                cur.execute("SELECT char_name FROM characters WHERE id=%s", (char_id,))
                old = cur.fetchone()
            if not old:
                flash("Charakter nicht gefunden.")
                return redirect(url_for("characters"))

            new_name = request.form.get("char_name", "").strip()
            if new_name and new_name != old["char_name"]:
                with db.cursor() as cur:
                    cur.execute(
                        "SELECT id FROM characters WHERE char_name=%s AND id<>%s",
                        (new_name, char_id),
                    )
                    if cur.fetchone():
                        flash(f"Name '{new_name}' ist bereits vergeben.")
                        return redirect(url_for("edit_character", char_id=char_id))
                    cur.execute(
                        "UPDATE characters SET char_name=%s WHERE id=%s",
                        (new_name, char_id),
                    )
                    # namefriend is a denormalized display-name cache in other
                    # players' friend lists (list_friend.idfriend references
                    # this character's id) -- keep it in sync so renames
                    # don't leave stale names in friend lists.
                    cur.execute(
                        "UPDATE list_friend SET namefriend=%s WHERE idfriend=%s",
                        (new_name, char_id),
                    )
                flash(f"Charakter '{old['char_name']}' umbenannt in '{new_name}'.")

            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)