# ruff: noqa: T201 # usage of `print` from __future__ import annotations import argparse import json import socket import sys from typing import Any, TYPE_CHECKING, TypeAlias import dns.resolver from mcstatus import BedrockServer, JavaServer, LegacyServer from mcstatus.responses import JavaStatusResponse if TYPE_CHECKING: from mcstatus.motd import Motd SupportedServers: TypeAlias = JavaServer | LegacyServer | BedrockServer PING_PACKET_FAIL_WARNING = ( "warning: contacting {address} failed with a 'ping' packet but succeeded with a 'status' packet,\n" " this is likely a bug in the server-side implementation.\n" ' (note: ping packet failed due to "{ping_exc}")\n' " for more details, see: https://mcstatus.readthedocs.io/en/stable/pages/faq/\n" ) QUERY_FAIL_WARNING = ( "The server did not respond to the query protocol." "\nPlease ensure that the server has enable-query turned on," " and that the necessary port (same as server-port unless query-port is set) is open in any firewall(s)." "\nSee https://minecraft.wiki/w/Query for further information." ) def _motd(motd: Motd) -> str: """Format MOTD for human-readable output, with leading line break if multiline.""" s = motd.to_ansi() return f"\n{s}" if "\n" in s else f" {s}" def _kind(serv: SupportedServers) -> str: if isinstance(serv, JavaServer): return "Java" if isinstance(serv, LegacyServer): return "Java (pre-1.7)" if isinstance(serv, BedrockServer): # pyright: ignore[reportUnnecessaryIsInstance] # be explicit return "Bedrock" raise ValueError(f"unsupported server for kind: {serv}") def _ping_with_fallback(server: SupportedServers) -> float: # only Java has ping method if not isinstance(server, JavaServer): return server.status().latency # try faster ping packet first, falling back to status with a warning. try: return server.ping(tries=1) except Exception as e: # noqa: BLE001 # blindly catching Exception ping_exc = e latency = server.status().latency address = f"{server.address.host}:{server.address.port}" print( PING_PACKET_FAIL_WARNING.format(address=address, ping_exc=ping_exc), file=sys.stderr, ) return latency def ping_cmd(server: SupportedServers) -> int: print(_ping_with_fallback(server)) return 0 def status_cmd(server: SupportedServers) -> int: response = server.status() java_res = response if isinstance(response, JavaStatusResponse) else None if java_res and java_res.players.sample: player_sample = "\n " + "\n ".join(f"{player.name} ({player.id})" for player in java_res.players.sample) else: player_sample = "" print(f"version: {_kind(server)} {response.version.name} (protocol {response.version.protocol})") print(f"motd:{_motd(response.motd)}") print(f"players: {response.players.online}/{response.players.max}{player_sample}") print(f"ping: {response.latency:.2f} ms") return 0 def json_cmd(server: SupportedServers) -> int: data: dict[str, Any] = {"online": False, "kind": _kind(server)} status_res = query_res = exn = None try: status_res = server.status(tries=1) except Exception as e: # noqa: BLE001 # blindly catching Exception exn = exn or e try: if isinstance(server, JavaServer): query_res = server.query(tries=1) except Exception as e: # noqa: BLE001 # blindly catching Exception exn = exn or e # construct 'data' dict outside try/except to ensure data processing errors # are noticed. data["online"] = bool(status_res or query_res) if not data["online"]: assert exn, "server offline but no exception?" data["error"] = str(exn) if status_res is not None: data["status"] = status_res.as_dict() if query_res is not None: data["query"] = query_res.as_dict() json.dump(data, sys.stdout) return 0 def query_cmd(server: SupportedServers) -> int: if not isinstance(server, JavaServer): print("The 'query' protocol is only supported by Java servers.", file=sys.stderr) return 1 try: response = server.query() except TimeoutError: print(QUERY_FAIL_WARNING, file=sys.stderr) return 1 print(f"host: {response.raw['hostip']}:{response.raw['hostport']}") print(f"software: {_kind(server)} {response.software.version} {response.software.brand}") print(f"motd:{_motd(response.motd)}") print(f"plugins: {response.software.plugins}") print(f"players: {response.players.online}/{response.players.max} {response.players.list}") return 0 def main(argv: list[str] = sys.argv[1:]) -> int: parser = argparse.ArgumentParser( "mcstatus", description=""" mcstatus provides an easy way to query Minecraft servers for any information they can expose. It provides three modes of access: query, status, ping and json. """, ) _ = parser.add_argument("address", help="The address of the server.") group = parser.add_mutually_exclusive_group() _ = group.add_argument( "--bedrock", help="Specifies that 'address' is a Bedrock server (default: Java).", action="store_true" ) _ = group.add_argument( "--legacy", help="Specifies that 'address' is a pre-1.7 Java server (default: 1.7+).", action="store_true" ) subparsers = parser.add_subparsers(title="commands", description="Command to run, defaults to 'status'.") parser.set_defaults(func=status_cmd) subparsers.add_parser("ping", help="Ping server for latency.").set_defaults(func=ping_cmd) subparsers.add_parser("status", help="Prints server status.").set_defaults(func=status_cmd) subparsers.add_parser( "query", help="Prints detailed server information. Must be enabled in servers' server.properties file." ).set_defaults(func=query_cmd) subparsers.add_parser( "json", help="Prints server status and query in json.", ).set_defaults(func=json_cmd) args = parser.parse_args(argv) if args.bedrock: lookup = BedrockServer.lookup elif args.legacy: lookup = LegacyServer.lookup else: lookup = JavaServer.lookup try: server = lookup(args.address) return args.func(server) except (socket.gaierror, dns.resolver.NoNameservers, ConnectionError, TimeoutError) as e: # catch and hide traceback for expected user-facing errors print(f"Error: {e!r}", file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())