diff --git a/.gitattributes b/.gitattributes index 30d92c07..ed2ee15f 100644 --- a/.gitattributes +++ b/.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 diff --git a/.github/workflows/fuzz.yml b/.github/workflows/fuzz.yml index 2708b36f..8619dc81 100644 --- a/.github/workflows/fuzz.yml +++ b/.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 diff --git a/models.py b/models.py index d0078f14..7ff4a5bf 100644 --- a/models.py +++ b/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) diff --git a/ssh_manager.py b/ssh_manager.py index 0870af82..58b51c7b 100644 --- a/ssh_manager.py +++ b/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] diff --git a/terminal.py b/terminal.py index 385c5c1c..d345daec 100644 --- a/terminal.py +++ b/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): diff --git a/tests/fuzz/corpus/console/bracket_paste b/tests/fuzz/corpus/console/bracket_paste new file mode 100644 index 00000000..0b779638 --- /dev/null +++ b/tests/fuzz/corpus/console/bracket_paste @@ -0,0 +1 @@ +[?2004h> status[?2004l diff --git a/tests/fuzz/corpus/console/cr_overwrite b/tests/fuzz/corpus/console/cr_overwrite new file mode 100644 index 00000000..318d7549 --- /dev/null +++ b/tests/fuzz/corpus/console/cr_overwrite @@ -0,0 +1 @@ +Downloading 100% Downloading 4% diff --git a/tests/fuzz/corpus/console/crlf_log b/tests/fuzz/corpus/console/crlf_log new file mode 100644 index 00000000..cf9b2a85 --- /dev/null +++ b/tests/fuzz/corpus/console/crlf_log @@ -0,0 +1,2 @@ +line one +line two diff --git a/tests/fuzz/corpus/console/dangling_esc b/tests/fuzz/corpus/console/dangling_esc new file mode 100644 index 00000000..11a10621 --- /dev/null +++ b/tests/fuzz/corpus/console/dangling_esc @@ -0,0 +1 @@ +player  \ No newline at end of file diff --git a/tests/fuzz/corpus/console/erase_line b/tests/fuzz/corpus/console/erase_line new file mode 100644 index 00000000..5e004adb --- /dev/null +++ b/tests/fuzz/corpus/console/erase_line @@ -0,0 +1 @@ +loading... done diff --git a/tests/fuzz/corpus/console/esc_control b/tests/fuzz/corpus/console/esc_control new file mode 100644 index 00000000..cfa00211 Binary files /dev/null and b/tests/fuzz/corpus/console/esc_control differ diff --git a/tests/fuzz/corpus/console/jline_echo b/tests/fuzz/corpus/console/jline_echo new file mode 100644 index 00000000..1422f4fd --- /dev/null +++ b/tests/fuzz/corpus/console/jline_echo @@ -0,0 +1 @@ +ssasay hi diff --git a/tests/fuzz/corpus/console/mixed b/tests/fuzz/corpus/console/mixed new file mode 100644 index 00000000..d82dfd76 --- /dev/null +++ b/tests/fuzz/corpus/console/mixed @@ -0,0 +1 @@ + hi XY]2;t\ diff --git a/tests/fuzz/corpus/console/osc_title b/tests/fuzz/corpus/console/osc_title new file mode 100644 index 00000000..2db17919 --- /dev/null +++ b/tests/fuzz/corpus/console/osc_title @@ -0,0 +1 @@ +]0;csgoserverserver ready diff --git a/tests/fuzz/corpus/console/sgr_colour b/tests/fuzz/corpus/console/sgr_colour new file mode 100644 index 00000000..6bd9643d --- /dev/null +++ b/tests/fuzz/corpus/console/sgr_colour @@ -0,0 +1 @@ +[INFO] player connected diff --git a/tests/fuzz/corpus/console/truncated_csi b/tests/fuzz/corpus/console/truncated_csi new file mode 100644 index 00000000..be16d00a --- /dev/null +++ b/tests/fuzz/corpus/console/truncated_csi @@ -0,0 +1 @@ +[38;5; \ No newline at end of file diff --git a/tests/fuzz/corpus/console/two_byte b/tests/fuzz/corpus/console/two_byte new file mode 100644 index 00000000..f3e55d9c --- /dev/null +++ b/tests/fuzz/corpus/console/two_byte @@ -0,0 +1 @@ +x>y=z diff --git a/tests/fuzz/fuzz_console.py b/tests/fuzz/fuzz_console.py new file mode 100644 index 00000000..b9414b90 --- /dev/null +++ b/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() diff --git a/tests/rbac_test.py b/tests/rbac_test.py index f433d953..15e85df6 100644 --- a/tests/rbac_test.py +++ b/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 / 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", diff --git a/tests/smoke_test.py b/tests/smoke_test.py index a50bf316..62f6947b 100644 --- a/tests/smoke_test.py +++ b/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: diff --git a/tests/unit_test.py b/tests/unit_test.py index 5426b9bb..df15adcb 100644 --- a/tests/unit_test.py +++ b/tests/unit_test.py @@ -3033,6 +3033,101 @@ def _boom(remote, cmd, timeout=None): finally: _mapp.run_command = _orig_rc +# ── API tokens: only a HASH is ever stored ──────────────────────────────────────────────────── +# Bearer tokens authenticate scripts as their owner and inherit that user's full RBAC, and app.py +# exempts Bearer requests from CSRF — yet nothing exercised any of it. These are the pure half; the +# lookup half (a disabled owner, a replayed hash) is asserted against the real DB in smoke_test. +import hashlib as _hl +from models import User as _U, RemoteServer as _RS + +_tu = _U(username="tok", password_hash="x") +_plain = _tu.generate_api_token() +check("api token: the minted token is handed back in plaintext, once", + isinstance(_plain, str) and _plain.startswith("lgsm_") and len(_plain) >= 32, _plain[:12]) +check("api token: the plaintext is NOT what gets stored", _tu.api_token != _plain) +check("api token: what IS stored is its sha256 — a leaked DB yields no usable token", + _tu.api_token == _hl.sha256(_plain.encode()).hexdigest()) +_plain2 = _tu.generate_api_token() +check("api token: minting again replaces the old one", _plain2 != _plain and + _tu.api_token == _hl.sha256(_plain2.encode()).hexdigest()) +check("api token: has_api_token reports the stored state", _tu.has_api_token is True) +_tu.revoke_api_token() +check("api token: revoking clears it", _tu.api_token is None and _tu.has_api_token is False) + +# ── Pinned SSH host key: the fingerprint an operator eyeballs before trusting a host ─────────── +import base64 as _b64 +_blob = b"\x00\x00\x00\x07ssh-rsa" + bytes(range(256)) * 2 +_key = "ssh-rsa " + _b64.b64encode(_blob).decode() +_want = "ssh-rsa SHA256:" + _b64.b64encode(_hl.sha256(_blob).digest()).decode().rstrip("=") +eq("host key: the fingerprint is sha256 of the DECODED key, in OpenSSH form", + _RS(host_key=_key).host_key_fingerprint, _want) +check("host key: it is the base64 digest, not hex, and unpadded", + ":" in _want and "=" not in _want.split(":")[1] and len(_want.split(":")[1]) == 43, _want) +eq("host key: nothing pinned yet reads as empty, not as an error", + _RS(host_key="").host_key_fingerprint, "") +for _bad in ("not base64 !!!", "ssh-rsa", "ssh-rsa ***", "\x00\xff"): + eq("host key: malformed pin %r degrades to empty rather than raising" % _bad[:14], + _RS(host_key=_bad).host_key_fingerprint, "") + +# ── File manager: the protected-path guard on `rm -rf` ──────────────────────────────────────── +# delete_path deletes the RESOLVED absolute path, so the guard has to judge the resolved path too. +# It used to read the raw string: "./lgsm" and "x/../serverfiles" looked like ordinary sub-paths +# while resolving onto the LinuxGSM control tree and the entire game install. +_SELF = "csgoserver" +for _p in ("lgsm", "lgsm/data", "/lgsm", "lgsm/", "serverfiles", "linuxgsm.sh", _SELF, + ".ssh", ".bashrc", "", "/", ".", "..", "../..", "a/../.."): + check("file guard: %r is protected from deletion" % _p, + sm._is_protected_path(_p, _SELF) is True) +# The bypasses. Each of these resolves onto something whose loss is unrecoverable. +for _p, _what in (("./lgsm", "the LinuxGSM control tree"), + (".//lgsm", "the LinuxGSM control tree"), + ("./serverfiles", "the game install"), + ("x/../serverfiles", "the game install"), + ("a/b/../../.ssh", "the host's SSH keys"), + ("./linuxgsm.sh", "the LinuxGSM launcher"), + ("./%s" % _SELF, "the server's own script")): + check("file guard: %r cannot sneak past and take %s" % (_p, _what), + sm._is_protected_path(_p, _SELF) is True) +# ...and ordinary content is still deletable, or the file manager is useless. +for _p in ("addons/mymap.bsp", "./addons/mymap.bsp", "cfg/server.cfg", "logs", "lgsm2", + "serverfiles-old", "my lgsm notes.txt"): + check("file guard: ordinary path %r stays deletable" % _p, + sm._is_protected_path(_p, _SELF) is False) + +# End-to-end through delete_path itself: the guard is worth nothing if the caller skips it, so +# assert no shell command is issued at all for a protected path. +_sent_rm = [] +_orig_rc2 = sm.run_command +try: + sm.run_command = lambda server, cmd, timeout=30, sudo=None: (_sent_rm.append(cmd), ("__OK__", "", 0))[1] + _ok, _msg = sm.delete_path(object(), _SELF, "./lgsm", selfname=_SELF) + check("file guard: delete_path refuses './lgsm' and runs NOTHING", + _ok is False and not _sent_rm, "ran: %s" % _sent_rm[:1]) + check("file guard: the refusal explains itself", "protected" in (_msg or "").lower(), _msg) + _sent_rm.clear() + _ok2, _ = sm.delete_path(object(), _SELF, "x/../serverfiles", selfname=_SELF) + check("file guard: delete_path refuses a traversal onto serverfiles", + _ok2 is False and not _sent_rm, "ran: %s" % _sent_rm[:1]) + _sent_rm.clear() + _ok3, _ = sm.delete_path(object(), _SELF, "addons/junk.txt", selfname=_SELF) + check("file guard: a real file still gets deleted, at its resolved path", + _ok3 is True and len(_sent_rm) == 1 + and "/home/%s/addons/junk.txt" % _SELF in _sent_rm[0], "ran: %s" % _sent_rm[:1]) +finally: + sm.run_command = _orig_rc2 + +# ── The renderer's one guarantee: no control byte reaches the page ──────────────────────────── +# Console text carries player names and chat, so these bytes are AUTHORED, not just accidental. +# A sequence that never completes matched none of the three grammars and used to survive. +import terminal as _term +for _src in ("player \x1b", "chat: \x1b\t hi", "x\x1b\x00y", "\x1b[38;5;", "\x1b\x9b"): + check("terminal: an incomplete escape %r leaves no ESC behind" % _src, + "\x1b" not in _term.render(_src), repr(_term.render(_src))) +for _src, _want in (("\x1b[32mgreen\x1b[0m", "green"), ("a\x1b[Kb", "ab"), ("x\x1b>y", "xy"), + ("\x1b]0;title\x07hi", "hi"), ("ssa\b\b\bsay hi", "say hi"), + ("abcdef\rXY", "XYcdef"), ("done\r\nnext", "done\nnext")): + eq("terminal: %r still renders as a terminal would" % _src, _term.render(_src), _want) + passed = sum(1 for ok, _, _ in results if ok) for ok, name, detail in results: line = ("PASS" if ok else "FAIL") + " " + name