test: cover the untested auth paths — and fix the three bugs that found

.gitattributes

@@ -3,3 +3,7 @@
 * text=auto eol=lf
 *.sh text eol=lf
 *.py text eol=lf
+
+# Fuzzing corpora are exact byte sequences, not text. The `* text=auto eol=lf` rule above rewrote
+# CRLF out of the console seeds — which is the very thing those seeds exist to feed the renderer.
+tests/fuzz/corpus/** -text

.github/workflows/fuzz.yml

@@ -43,7 +43,7 @@ jobs:
     strategy:
       fail-fast: false   # fuzz every target even if one finds a crash, so all findings surface
       matrix:
-        target: [ game_status, firewall, config, fail2ban ]
+        target: [ game_status, firewall, config, fail2ban, console ]
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
 

models.py

@@ -331,7 +331,13 @@ def host_key_fingerprint(self):
             import base64
             import hashlib
             parts = self.host_key.split()
-            raw = base64.b64decode(parts[1] if len(parts) > 1 else parts[0])
+            # validate=True: the lenient default DISCARDS characters outside the base64 alphabet, so
+            # a corrupted pin like "ssh-rsa ***" decoded to b"" and printed the sha256 of nothing —
+            # a confident-looking fingerprint for a key that isn't there, which is the opposite of
+            # what an operator eyeballing this needs.
+            raw = base64.b64decode(parts[1] if len(parts) > 1 else parts[0], validate=True)
+            if not raw:
+                return ""
             digest = base64.b64encode(hashlib.sha256(raw).digest()).decode().rstrip("=")
             keytype = parts[0] if len(parts) > 1 else "key"
             return "%s SHA256:%s" % (keytype, digest)

ssh_manager.py

@@ -4596,8 +4596,12 @@ def mods_action(server, user, selfname, which, mod_id, timeout=600):
 # from the file browser's delete (they can still be edited where that makes sense).
 def _is_protected_path(relpath, selfname):
     """True if `relpath` (relative to the game user's home) must not be deleted."""
-    r = (relpath or "").strip("/")
-    if not r:
+    # Normalise BEFORE inspecting. delete_path deletes the resolved absolute path, so a guard that
+    # reads the raw string is judging a different path than the one `rm -rf` receives: "./lgsm" and
+    # "x/../serverfiles" look like ordinary sub-paths here while resolving onto the LinuxGSM control
+    # tree and the whole game install. Anything that climbs out collapses to "" and is refused.
+    r = _pp.normpath("/" + str(relpath or "")).strip("/")
+    if not r or r == ".":
         return True  # the home dir itself
     parts = r.split("/")
     top = parts[0]

terminal.py

@@ -29,7 +29,13 @@ def strip_escapes(text):
     left alone — they carry rendering meaning; see apply_carriage_returns / apply_backspaces."""
     if not text:
         return text
-    return ESC2_RE.sub("", CSI_RE.sub("", OSC_RE.sub("", text)))
+    text = ESC2_RE.sub("", CSI_RE.sub("", OSC_RE.sub("", text)))
+    # Any ESC still standing is a sequence that never completed: a log read mid-write, or ESC
+    # followed by something none of the three grammars accept (\x1b\t, \x1b\x00, a trailing \x1b).
+    # Player names and chat reach this text, so those bytes are authored, not just accidental. They
+    # carry no rendering meaning on their own, and the one guarantee this module owes its callers is
+    # that control bytes do not reach the page.
+    return text.replace("\x1b", "")
 
 
 def apply_carriage_returns(line):

tests/fuzz/corpus/console/bracket_paste

@@ -0,0 +1 @@
+[?2004h> status[?2004l

tests/fuzz/corpus/console/cr_overwrite

@@ -0,0 +1 @@
+Downloading 100%Downloading   4%

tests/fuzz/corpus/console/crlf_log

@@ -0,0 +1,2 @@
+line one
+line two

tests/fuzz/corpus/console/dangling_esc

@@ -0,0 +1 @@
+player 

tests/fuzz/corpus/console/erase_line

@@ -0,0 +1 @@
+loading... done

tests/fuzz/corpus/console/jline_echo

@@ -0,0 +1 @@
+ssasay hi

tests/fuzz/corpus/console/mixed

@@ -0,0 +1 @@
+<Player> hiXY]2;t\

tests/fuzz/corpus/console/osc_title

@@ -0,0 +1 @@
+]0;csgoserverserver ready

tests/fuzz/corpus/console/sgr_colour

@@ -0,0 +1 @@
+[INFO] player connected

tests/fuzz/corpus/console/truncated_csi

@@ -0,0 +1 @@
+[38;5;

tests/fuzz/corpus/console/two_byte

@@ -0,0 +1 @@
+x>y=z

tests/fuzz/fuzz_console.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+"""Atheris fuzz harness: the terminal renderer that turns raw console output into panel text.
+
+Everything terminal.py sees is UNTRUSTED and, unlike the other targets, partly attacker-*authored*:
+a game console echoes player names, chat and RCON replies, so a player picks the bytes — escape
+sequences, lone carriage returns, backspaces, half-formed CSI/OSC introducers, invalid UTF-8. The
+rendered result is then shown in the panel's console view.
+
+Two properties under test:
+  1. No input may make a renderer raise (the console poller and the cron-error path both call these,
+     and a raise there breaks a page rather than one line).
+  2. render() must not emit an ESC (0x1b) or a bare CR — those are exactly what it exists to strip,
+     and leaking one through means raw control bytes reach the browser.
+
+Run locally (from anywhere):
+    pip install atheris
+    python tests/fuzz/fuzz_console.py -max_total_time=60 tests/fuzz/corpus/console
+"""
+import os
+import sys
+
+import atheris
+
+# Running `python tests/fuzz/fuzz_x.py` puts tests/fuzz (not the project root) on sys.path, so make
+# the project root importable before importing the panel's modules.
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+
+with atheris.instrument_imports():
+    import terminal
+
+
+def TestOneInput(data):
+    fdp = atheris.FuzzedDataProvider(data)
+    text = fdp.ConsumeUnicodeNoSurrogates(fdp.remaining_bytes())
+
+    terminal.strip_escapes(text)
+    terminal.apply_carriage_returns(text)
+    terminal.apply_backspaces(text)
+    terminal.render_line(text)
+
+    out = terminal.render(text)
+    # The whole point of the renderer: control bytes must not survive into the panel.
+    if "\x1b" in out:
+        raise AssertionError("render() leaked an ESC byte")
+    if "\r" in out:
+        raise AssertionError("render() leaked a carriage return")
+
+
+def main():
+    atheris.Setup(sys.argv, TestOneInput)
+    atheris.Fuzz()
+
+
+if __name__ == "__main__":
+    main()

tests/rbac_test.py

@@ -148,6 +148,30 @@ def check(name, cond, detail=""):
     db.session.commit()
     uid3 = u3.id
 
+    # Fourth fixture: a DELEGATED group admin — MANAGE_GROUPS and nothing else. They can edit a
+    # group they belong to, so _grantable_perms is the only thing between them and self-promotion.
+    tag4 = tag + "_grpadm"
+    grp4 = Group(name=tag4, description="RBAC test group-admin (auto)", is_default=False)
+    grp4.set_permissions([auth.VIEW_SERVERS, auth.MANAGE_GROUPS])
+    grp4.servers.append(RemoteServer.query.get(granted_remote))
+    db.session.add(grp4)
+    db.session.flush()
+    u4 = User(username=tag4, password_hash=auth.hash_password(secrets.token_hex(16)),
+              display_name=tag4, is_superadmin=False, is_active=True)
+    u4.groups.append(grp4)
+    db.session.add(u4)
+    db.session.commit()
+    uid4 = u4.id
+
+    # A group that already holds a permission the delegated admin CANNOT grant, to prove an edit
+    # by them preserves it instead of silently stripping it.
+    tag5 = tag + "_holds_mu"
+    grp5 = Group(name=tag5, description="RBAC test preserve (auto)", is_default=False)
+    grp5.set_permissions([auth.MANAGE_USERS])
+    db.session.add(grp5)
+    db.session.commit()
+    gid5 = grp5.id
+
 print("Fixtures: limited user id=%d, group grants remote %d only." % (uid, granted_remote))
 print("Accessible server id=%d (remote %d); non-granted server id=%s (remote %s)\n"
       % (accessible_id, granted_remote, other_id, other_remote))
@@ -294,6 +318,73 @@ def client_as(user_id=None):
           "ACCOUNT WAS CREATED (status %d)" % r.status_code if was_created else "blocked")
     check("unauth GET /setup -> redirect", cu.get("/setup").status_code == 302)
 
+    # ── Privilege escalation: a delegated group admin cannot grant what they don't hold ──────────
+    # They have MANAGE_GROUPS, so they may create and edit groups — including groups they are in.
+    # _grantable_perms is the whole defence, and nothing exercised it for a non-superadmin
+    # permission. The existing "can't grant super_admin" check passes even with the guard removed,
+    # because super_admin was dropped from ALL_PERMISSIONS and is filtered separately.
+    c4 = client_as(uid4)
+    esc_name = tag + "_escalation"
+    c4.post("/groups/add", data={"name": esc_name, "description": "",
+                                 "permissions": [auth.MANAGE_USERS, auth.UNINSTALL_SERVER,
+                                                 auth.VIEW_SERVERS]})
+    with app.app_context():
+        made = Group.query.filter_by(name=esc_name).first()
+        got = set(made.get_permissions()) if made else None
+    check("escalation: a MANAGE_GROUPS admin cannot grant permissions they lack",
+          made is not None and auth.MANAGE_USERS not in got and auth.UNINSTALL_SERVER not in got,
+          "granted: %s" % sorted(got or []))
+    check("escalation: they CAN grant a permission they do hold",
+          made is not None and auth.VIEW_SERVERS in (got or set()), "granted: %s" % sorted(got or []))
+
+    # ...and editing a group must not silently strip a permission they cannot grant.
+    c4.post("/groups/%d/edit" % gid5, data={"name": tag5, "description": "",
+                                            "permissions": [auth.VIEW_SERVERS]})
+    with app.app_context():
+        kept = set(Group.query.get(gid5).get_permissions())
+    check("escalation: an edit PRESERVES a permission the editor cannot grant",
+          auth.MANAGE_USERS in kept, "after edit: %s" % sorted(kept))
+
+    # ── Bulk actions are access-checked per id ────────────────────────────────────────────────────
+    # /api/servers/bulk-action is not an /<int:server_id> route, so the structural sweep below
+    # never sees it, and it carries no @server_access_required — the check is hand-written in the
+    # loop. A break here fans a power action out over SSH to every server in the install.
+    if other_id:
+        import app as _appmod
+        _ran = []
+        _sv_rag = _appmod.run_as_game_user
+        try:
+            _appmod.run_as_game_user = lambda *a, **k: (_ran.append(a), ("", "", 0))[1]
+            grpb = None
+            with app.app_context():
+                grpb = Group(name=tag + "_bulk", description="RBAC bulk (auto)", is_default=False)
+                grpb.set_permissions([auth.VIEW_SERVERS, auth.START_SERVER])
+                grpb.servers.append(RemoteServer.query.get(granted_remote))
+                db.session.add(grpb)
+                db.session.flush()
+                ub = User(username=tag + "_bulk", password_hash=auth.hash_password(secrets.token_hex(16)),
+                          display_name=tag + "_bulk", is_superadmin=False, is_active=True)
+                ub.groups.append(grpb)
+                db.session.add(ub)
+                db.session.commit()
+                uidb, gidb = ub.id, grpb.id
+            rb = client_as(uidb).post("/api/servers/bulk-action",
+                                      json={"action": "start", "server_ids": [other_id]})
+            jb = rb.get_json() or {}
+            check("bulk-action: a server on a non-granted host is refused, not queued",
+                  not jb.get("queued")
+                  and any(sk.get("reason") == "no access" for sk in jb.get("skipped") or []),
+                  "queued=%s skipped=%s" % (jb.get("queued"), jb.get("skipped")))
+            import time as _t
+            _t.sleep(0.3)   # the action runs in a background thread; give it time to have fired
+            check("bulk-action: and nothing was actually run on it", not _ran, "ran: %s" % _ran[:1])
+            with app.app_context():
+                db.session.delete(User.query.get(uidb))
+                db.session.delete(Group.query.get(gidb))
+                db.session.commit()
+        finally:
+            _appmod.run_as_game_user = _sv_rag
+
     # ── Superadmin sanity: still full access ──
     ca = client_as(admin_id)
     for p in ["/users", "/groups", "/logs", "/remotes", "/server-management", "/tailscale",

tests/smoke_test.py

@@ -33,7 +33,7 @@
 save_config(_cfg)
 
 from app import create_app
-from models import db, User, Group, RemoteServer, GameServer, SetupState
+from models import db, User, Group, RemoteServer, GameServer, SetupState, CustomCommand
 import auth
 
 app = create_app()
@@ -1649,6 +1649,107 @@ def _unreadable(gs):
         _g3, _e3 = _tg_find_server("no-such-server-xyz")   # unknown
         check("telegram: an unknown server name returns a helpful error", _g3 is None and "No server" in (_e3 or ""))
 
+    # ── Bearer API tokens: the other way into every route ─────────────────────────────────────────
+    # A token authenticates AS its owner and inherits exactly that user's RBAC, and app.py exempts
+    # Bearer requests from CSRF — so this is a full authentication path that had no test at all.
+    def _bearer(tok):
+        return app.test_client().get("/api/servers", headers={"Authorization": "Bearer %s" % tok})
+
+    with app.app_context():
+        _au = db.session.get(User, admin_id)
+        _admin_tok = _au.generate_api_token()
+        _stored = _au.api_token
+        db.session.commit()
+    _ok = _bearer(_admin_tok)
+    check("api token: a valid token authenticates with no session cookie",
+          _ok.status_code == 200, "got %d" % _ok.status_code)
+    check("api token: an unknown token does not authenticate",
+          _bearer("lgsm_" + "0" * 48).status_code != 200)
+    # The property that makes storing only a hash worth anything: whoever reads the DB holds the
+    # hash, and replaying it must NOT authenticate.
+    check("api token: replaying the STORED hash does not authenticate",
+          _bearer(_stored).status_code != 200, "the stored value logged in")
+    check("api token: an empty bearer does not authenticate", _bearer("").status_code != 200)
+
+    # A token inherits its owner's scope — no more. mru can see host #1 only.
+    with app.app_context():
+        _mu = db.session.get(User, mru_id)
+        _mru_tok = _mu.generate_api_token()
+        db.session.commit()
+    _mine = _bearer(_mru_tok)
+    check("api token: a restricted user's token sees only that user's servers",
+          _mine.status_code == 200
+          and 0 < len(_mine.get_json() or []) < len(_ok.get_json() or []),
+          "restricted=%s admin=%s" % (len(_mine.get_json() or []), len(_ok.get_json() or [])))
+
+    # Deactivating the owner must kill the token — by_api_token filters on is_active, and an
+    # offboarded account keeping API access is exactly the failure nobody would notice.
+    with app.app_context():
+        db.session.get(User, mru_id).is_active = False
+        db.session.commit()
+    check("api token: deactivating the owner kills their token",
+          _bearer(_mru_tok).status_code != 200, "a disabled user's token still worked")
+    with app.app_context():
+        _mu2 = db.session.get(User, mru_id)
+        _mu2.is_active = True
+        _mu2.revoke_api_token()
+        db.session.commit()
+    check("api token: a revoked token stops working", _bearer(_mru_tok).status_code != 200)
+
+    # ── can_run_custom_command: who may press a superadmin-authored console button ────────────────
+    # Every branch of this decides whether a non-superadmin gets to run a console command on a
+    # server, and none of it was asserted.
+    from auth import can_run_custom_command as _crcc
+    with app.app_context():
+        _gs = db.session.get(GameServer, gs_id)          # game_type "csgo" on host #1
+        _cmd = CustomCommand(name="Say", command_template="say {}", scope_type="all", enabled=True)
+        db.session.add(_cmd)
+        _grp = Group(name="smoke_cc", description="", is_default=False)
+        _grp.set_permissions([auth.VIEW_SERVERS])
+        _grp.servers.append(db.session.get(RemoteServer, remote_id))
+        db.session.add(_grp)
+        db.session.flush()
+        _cu = User(username="smoke_cc_user", password_hash=auth.hash_password("Str0ng!passw0rd"),
+                   display_name="CC", is_superadmin=False, is_active=True)
+        _cu.groups.append(_grp)
+        db.session.add(_cu)
+        db.session.commit()
+        _adm = db.session.get(User, admin_id)
+
+        check("custom command: a superadmin may run an enabled, in-scope command",
+              _crcc(_adm, _cmd, _gs) is True)
+        # The core rule: having ACCESS to the server is not having the COMMAND.
+        check("custom command: a user whose groups lack the command may NOT run it",
+              _crcc(_cu, _cmd, _gs) is False, "access to the server leaked the command")
+        _grp.custom_commands.append(_cmd)
+        db.session.commit()
+        check("custom command: granting it to the user's group lets them run it",
+              _crcc(_cu, _cmd, _gs) is True)
+
+        _cmd.enabled = False
+        db.session.commit()
+        check("custom command: a disabled command is refused even to a superadmin",
+              _crcc(_adm, _cmd, _gs) is False)
+        _cmd.enabled = True
+        # Scope: this command is for a different game, so it must not appear on this server.
+        _cmd.scope_type, _cmd.scope_value = "game", "minecraft"
+        db.session.commit()
+        check("custom command: a command scoped to another game is refused",
+              _crcc(_adm, _cmd, _gs) is False)
+        _cmd.scope_value = _gs.game_type
+        db.session.commit()
+        check("custom command: scoping it to THIS game allows it again",
+              _crcc(_adm, _cmd, _gs) is True)
+
+        # A user with no groups has no access to the host, so the access check must refuse first.
+        _nu = User(username="smoke_cc_none", password_hash=auth.hash_password("Str0ng!passw0rd"),
+                   display_name="None", is_superadmin=False, is_active=True)
+        db.session.add(_nu)
+        db.session.commit()
+        check("custom command: a user with no groups is refused", _crcc(_nu, _cmd, _gs) is False)
+        check("custom command: a missing command is refused, not an exception",
+              _crcc(_adm, None, _gs) is False)
+
 finally:
     passed = sum(1 for ok, _, _ in results if ok)
     for ok, name, detail in results: