matrix/minecraft-bridge/lotus_mc_status.py

#!/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()