ecc4aa28c6
- ?limit clamp test now actually verifies the clamp instead of just passing through 5 rows. - Single is_error assignment per branch, single db.add path. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
104 lines
3.1 KiB
Python
104 lines
3.1 KiB
Python
from flask import Blueprint, Response, jsonify, render_template, request
|
|
from sqlalchemy import select
|
|
|
|
from l4d2web.auth import current_user, require_login
|
|
from l4d2web.db import session_scope
|
|
from l4d2web.models import CommandHistory, Server
|
|
from l4d2web.services import rcon
|
|
from l4d2web.services.rcon import RconAuthError, RconError
|
|
|
|
|
|
bp = Blueprint("console", __name__)
|
|
|
|
_HISTORY_DEFAULT_LIMIT = 50
|
|
_HISTORY_MAX_LIMIT = 200
|
|
_HISTORY_MIN_LIMIT = 1
|
|
|
|
|
|
@bp.post("/servers/<int:server_id>/console")
|
|
@require_login
|
|
def run_console_command(server_id: int) -> Response:
|
|
user = current_user()
|
|
assert user is not None
|
|
|
|
with session_scope() as db:
|
|
server = db.scalar(
|
|
select(Server).where(Server.id == server_id, Server.user_id == user.id)
|
|
)
|
|
if server is None:
|
|
return Response(status=404)
|
|
|
|
command_raw = request.form.get("command", "")
|
|
command = command_raw.strip()
|
|
|
|
try:
|
|
reply = rcon.execute_command("127.0.0.1", server.port, server.rcon_password, command)
|
|
is_error = False
|
|
except (RconAuthError, RconError) as exc:
|
|
reply = str(exc)
|
|
is_error = True
|
|
except ValueError:
|
|
# Input validation failure — command never reached the wire; no history row.
|
|
return render_template(
|
|
"_console_line.html",
|
|
command=command,
|
|
reply="invalid command",
|
|
is_error=True,
|
|
)
|
|
|
|
db.add(
|
|
CommandHistory(
|
|
user_id=user.id,
|
|
server_id=server_id,
|
|
command=command,
|
|
reply=reply,
|
|
is_error=is_error,
|
|
)
|
|
)
|
|
|
|
return render_template(
|
|
"_console_line.html",
|
|
command=command,
|
|
reply=reply,
|
|
is_error=is_error,
|
|
)
|
|
|
|
|
|
@bp.get("/servers/<int:server_id>/console/history")
|
|
@require_login
|
|
def console_history(server_id: int) -> Response:
|
|
user = current_user()
|
|
assert user is not None
|
|
|
|
with session_scope() as db:
|
|
server = db.scalar(
|
|
select(Server).where(Server.id == server_id, Server.user_id == user.id)
|
|
)
|
|
if server is None:
|
|
return Response(status=404)
|
|
|
|
try:
|
|
raw_before = request.args.get("before")
|
|
before = int(raw_before) if raw_before is not None else None
|
|
except (TypeError, ValueError):
|
|
before = None
|
|
|
|
try:
|
|
raw_limit = request.args.get("limit")
|
|
limit = int(raw_limit) if raw_limit is not None else _HISTORY_DEFAULT_LIMIT
|
|
except (TypeError, ValueError):
|
|
limit = _HISTORY_DEFAULT_LIMIT
|
|
|
|
limit = max(_HISTORY_MIN_LIMIT, min(_HISTORY_MAX_LIMIT, limit))
|
|
|
|
query = select(CommandHistory).where(
|
|
CommandHistory.user_id == user.id,
|
|
CommandHistory.server_id == server_id,
|
|
)
|
|
if before is not None:
|
|
query = query.where(CommandHistory.id < before)
|
|
query = query.order_by(CommandHistory.id.desc()).limit(limit)
|
|
|
|
rows = db.scalars(query).all()
|
|
|
|
return jsonify([{"id": row.id, "command": row.command} for row in rows])
|