minecraft: show players online as the bridge bot's status

A small service on CT 154 pings the Lotus Minecraft server and sets the result
as the Matrix status of @lotus-mc-bridge, re-sent every 20 s because the mod
syncs with set_presence=offline and Synapse would otherwise drop it to offline.
Uses the mod's own token, so no new account or secret.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPmy3tPq869XDW4njjVaKA

README.md

@@ -513,6 +513,7 @@ Every bot account has a proper name and an avatar in one style (white glyph on a
 | `@draupnir` | Draupnir | Profile API with Draupnir's token (Draupnir doesn't manage its own profile) |
 | `@lotusbot` | Lotus Bot | Already set |
 | `@notices` | Server Notices | `server_notices.system_mxid_avatar_url` in `homeserver.yaml` (Lotus logo); needs a Synapse restart |
+| `@lotus-mc-bridge` | Lotus Minecraft | Profile API with its own token, from the Minecraft mod's config (voxel block with a lotus, not the white-glyph style; image not in `avatars/`). Its status is the player count: `minecraft-bridge/lotus_mc_status.py`, which runs on CT 154 as `lotus-mc-status.service` (not on LXC 151; installed by hand) |
 
 ---
 

minecraft-bridge/lotus_mc_status.py

@@ -0,0 +1,177 @@
+#!/usr/bin/env python3
+"""Player count -> status message on the Lotus Minecraft bot.
+
+Pings the Minecraft server the way the in-game server list does (no RCON, no
+password) and sets the result as the Matrix status of the account the chat
+bridge mod logs in as, for example "3 online: Daisybell, Honnehh, Wolf".
+
+- Credentials: the homeserver and token come from the mod's own config, so there
+  is no extra account or secret. systemd hands this service a private copy.
+- The mod syncs with set_presence=offline, so it doesn't keep the account online,
+  and Synapse drops an unrefreshed user to offline. So state is re-sent on a timer.
+- This runs beside the game, not in it, so it can report the server being down.
+  One failed ping is ignored: a busy modded server can miss one.
+"""
+import json
+import logging
+import os
+import re
+import socket
+import struct
+import time
+import tomllib
+import urllib.error
+import urllib.parse
+import urllib.request
+
+log = logging.getLogger("lotus-mc-status")
+
+INTERVAL = 20  # seconds; Synapse's default limit for this endpoint is one per 10 s
+PING_TIMEOUT = 8
+MAX_LEN = 64  # what Cinny's own status field allows
+FAILS_BEFORE_DOWN = 2
+OFFLINE = "Server offline"
+PLAYER_NAME = re.compile(r"[A-Za-z0-9_]{1,16}")
+
+
+def varint(n: int) -> bytes:
+    out = b""
+    while True:
+        byte = n & 0x7F
+        n >>= 7
+        out += bytes([byte | (0x80 if n else 0)])
+        if not n:
+            return out
+
+
+def read_varint(read) -> int:
+    result = 0
+    for shift in range(0, 35, 7):
+        byte = read(1)[0]
+        result |= (byte & 0x7F) << shift
+        if not byte & 0x80:
+            return result
+    raise ValueError("varint too long")
+
+
+def parse_status(packet: bytes) -> tuple[int, list[str]]:
+    """Return (players online, names) from a status response packet."""
+    pos = [0]
+
+    def read(n: int) -> bytes:
+        chunk = packet[pos[0] : pos[0] + n]
+        if len(chunk) < n:
+            raise ValueError("truncated status packet")
+        pos[0] += n
+        return chunk
+
+    if read_varint(read) != 0:
+        raise ValueError("not a status response")
+    players = json.loads(read(read_varint(read)))["players"]
+    return int(players["online"]), [str(p.get("name", "")) for p in players.get("sample") or []]
+
+
+def ping(host: str, port: int) -> tuple[int, list[str]]:
+    with socket.create_connection((host, port), timeout=PING_TIMEOUT) as sock:
+        addr = host.encode()
+        hello = b"\0" + varint(767) + varint(len(addr)) + addr + struct.pack(">H", port) + varint(1)
+        sock.sendall(varint(len(hello)) + hello + b"\x01\0")
+
+        def read(n: int) -> bytes:
+            buf = b""
+            while len(buf) < n:
+                chunk = sock.recv(n - len(buf))
+                if not chunk:
+                    raise ConnectionError("server closed the connection")
+                buf += chunk
+            return buf
+
+        return parse_status(read(read_varint(read)))
+
+
+def format_status(online: int, names: list[str], max_len: int = MAX_LEN) -> str:
+    """Count plus as many names as fit. Names are sorted so a re-ordered list
+    doesn't change the status, and anything that isn't a real player name
+    (servers can send placeholders) is dropped and counted under "more"."""
+    if online <= 0:
+        return "Nobody online"
+    head = f"{online} online"
+    real = sorted({n for n in names if PLAYER_NAME.fullmatch(n)}, key=str.casefold)[:online]
+    shown: list[str] = []
+    for name in real:
+        rest = online - len(shown) - 1
+        candidate = f"{head}: {', '.join([*shown, name])}" + (f" +{rest} more" if rest else "")
+        if len(candidate) > max_len:
+            break
+        shown.append(name)
+    if not shown:
+        return head
+    rest = online - len(shown)
+    return f"{head}: {', '.join(shown)}" + (f" +{rest} more" if rest else "")
+
+
+def read_config(path: str) -> tuple[str, str]:
+    with open(path, "rb") as f:
+        matrix = tomllib.load(f)["matrix"]
+    return matrix["homeserver"].rstrip("/"), matrix["accessToken"]
+
+
+def api(method: str, url: str, token: str, body: dict | None = None) -> dict:
+    data = json.dumps(body).encode() if body is not None else None
+    req = urllib.request.Request(url, data=data, method=method, headers={
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json",
+        "User-Agent": "lotus-mc-status",
+    })
+    with urllib.request.urlopen(req, timeout=15) as resp:
+        return json.load(resp)
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+    creds = os.environ.get("CREDENTIALS_DIRECTORY")
+    path = os.path.join(creds, "bridge-config") if creds else os.environ["LOTUS_MC_CONFIG"]
+    host = os.environ["LOTUS_MC_HOST"]
+    port = int(os.environ.get("LOTUS_MC_PORT", "25565"))
+    homeserver, token = read_config(path)
+
+    user_id = api("GET", f"{homeserver}/_matrix/client/v3/account/whoami", token)["user_id"]
+    url = f"{homeserver}/_matrix/client/v3/presence/{urllib.parse.quote(user_id, safe='')}/status"
+    log.info("watching %s:%d for %s", host, port, user_id)
+
+    fails = 0
+    body = None
+    last_error = None
+    last_count = None
+    while True:
+        try:
+            online, names = ping(host, port)
+            fails = 0
+            body = {"presence": "online", "status_msg": format_status(online, names)}
+            if online != last_count:
+                log.info("%d online", online)
+                last_count = online
+        except (OSError, ValueError, KeyError, TypeError) as exc:
+            fails += 1
+            if fails == FAILS_BEFORE_DOWN:
+                log.warning("server not answering: %s", type(exc).__name__)
+                last_count = None
+            if fails >= FAILS_BEFORE_DOWN:
+                body = {"presence": "unavailable", "status_msg": OFFLINE}
+        if body:
+            try:
+                api("PUT", url, token, body)
+                if last_error:
+                    log.info("Matrix reachable again")
+                last_error = None
+            except (OSError, ValueError) as exc:
+                # HTTPError and URLError are OSErrors. Their text has no token in it.
+                error = f"HTTP {exc.code}" if isinstance(exc, urllib.error.HTTPError) else type(exc).__name__
+                if error != last_error:
+                    log.warning("could not set status: %s", error)
+                last_error = error
+        time.sleep(INTERVAL)
+
+
+if __name__ == "__main__":
+    main()

minecraft-bridge/test_lotus_mc_status.py

@@ -0,0 +1,77 @@
+import json
+import unittest
+
+import lotus_mc_status as s
+
+
+def packet(players: dict) -> bytes:
+    body = json.dumps({"version": {"name": "1.21.1"}, "players": players}).encode()
+    return s.varint(0) + s.varint(len(body)) + body
+
+
+class FormatStatus(unittest.TestCase):
+    def test_nobody(self):
+        self.assertEqual(s.format_status(0, []), "Nobody online")
+
+    def test_one(self):
+        self.assertEqual(s.format_status(1, ["Honnehh"]), "1 online: Honnehh")
+
+    def test_sorted_regardless_of_server_order(self):
+        self.assertEqual(s.format_status(3, ["b_x", "C", "a"]), "3 online: a, b_x, C")
+
+    def test_long_names_collapse_and_stay_within_the_limit(self):
+        names = [f"Player{i}_abcdefghi"[:16] for i in range(10)]
+        got = s.format_status(10, names)
+        self.assertLessEqual(len(got), s.MAX_LEN)
+        self.assertRegex(got, r"^10 online: .+ \+\d+ more$")
+        shown = got.split(": ")[1].split(" +")[0].split(", ")
+        self.assertEqual(int(got.split("+")[1].split()[0]), 10 - len(shown))
+
+    def test_exact_fit_has_no_more_suffix(self):
+        got = s.format_status(1, ["A" * 16], max_len=len("1 online: ") + 16)
+        self.assertEqual(got, "1 online: " + "A" * 16)
+
+    def test_more_players_than_the_sample(self):
+        self.assertEqual(s.format_status(14, ["a", "b"]), "14 online: a, b +12 more")
+
+    def test_placeholders_are_not_shown_as_players(self):
+        self.assertEqual(s.format_status(2, ["Anonymous Player", "a"]), "2 online: a +1 more")
+        self.assertEqual(s.format_status(1, ["§cinjected name"]), "1 online")
+
+    def test_count_only_when_nothing_fits(self):
+        self.assertEqual(s.format_status(2, ["abcdefgh"], max_len=9), "2 online")
+
+    def test_duplicates_and_stale_extras_do_not_overcount(self):
+        self.assertEqual(s.format_status(1, ["a", "a", "b"]), "1 online: a")
+
+
+class ParseStatus(unittest.TestCase):
+    def test_with_sample(self):
+        p = packet({"max": 20, "online": 2, "sample": [{"name": "a", "id": "x"}, {"name": "b", "id": "y"}]})
+        self.assertEqual(s.parse_status(p), (2, ["a", "b"]))
+
+    def test_without_sample(self):
+        self.assertEqual(s.parse_status(packet({"max": 20, "online": 0})), (0, []))
+
+    def test_rejects_other_packets_and_truncation(self):
+        with self.assertRaises(ValueError):
+            s.parse_status(s.varint(1) + b"xx")
+        with self.assertRaises(ValueError):
+            s.parse_status(packet({"max": 20, "online": 0})[:-3])
+
+    def test_varint_round_trip(self):
+        for n in (0, 1, 127, 128, 300, 6087, 2097151):
+            data = iter([s.varint(n)])
+            buf = next(data)
+            pos = [0]
+
+            def read(k, buf=buf, pos=pos):
+                out = buf[pos[0] : pos[0] + k]
+                pos[0] += k
+                return out
+
+            self.assertEqual(s.read_varint(read), n)
+
+
+if __name__ == "__main__":
+    unittest.main()

systemd/lotus-mc-status.service

@@ -0,0 +1,21 @@
+[Unit]
+Description=Lotus Minecraft player count as the bridge bot's Matrix status
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+ExecStart=/usr/bin/python3 -I /opt/lotus-mc-status/lotus_mc_status.py
+Environment=LOTUS_MC_HOST=10.10.10.78
+Environment=LOTUS_MC_PORT=25565
+LoadCredential=bridge-config:/var/lib/pelican/volumes/5476addf-384e-4f49-baa6-303e05719e9f/config/minecraftmatrixbridge.toml
+DynamicUser=yes
+NoNewPrivileges=yes
+ProtectSystem=strict
+ProtectHome=yes
+PrivateTmp=yes
+Restart=always
+RestartSec=30
+
+[Install]
+WantedBy=multi-user.target