"""
Functions for launching and controlling a Minecraft server in a Podman container.
mciwb talks to the Podman service over its REST API socket, see
`podman_client` for how the socket is found.
"""
import os
import shutil
from pathlib import Path
from time import monotonic, sleep
from typing import Any
from mcwb import Vec3
from podman import PodmanClient
from podman.domain.containers import Container
from rcon.exceptions import WrongPassword
from mciwb.logging import log
from mciwb.threads import Client
HOST = "localhost"
# the default folder for backups of the server's world
backup_folder_default = Path.home() / "mciwb-backups"
server_name = "mciwb-server"
default_server_folder = Path.home() / server_name
# The Minecraft Java Edition release that mciwb servers run. Keep this in step
# with mcwb.MINECRAFT_VERSION, the release that mcwb.Item is generated from.
MINECRAFT_VERSION = "26.3"
# itzg/minecraft-server image, pinned for reproducibility. Minecraft 26.x
# needs Java 25, hence the java25 variant.
SERVER_IMAGE = "docker.io/itzg/minecraft-server:2026.10.0-java25"
# the socket of a rootful podman service, used when there is no rootless one
ROOTFUL_SOCKET = Path("/run/podman/podman.sock")
# the uid and gid that the itzg/minecraft-server image runs the server as
IMAGE_UID = 1000
# container states that mean the server may still come up
STARTING_STATES = ("configured", "created", "initialized", "running", "restarting")
def_pass = "default_pass"
def_port = 20100
def_world_type = "normal"
# seconds to wait for a server to save and shut down before it is killed
stop_timeout = 60
class PodmanNotAvailableError(RuntimeError):
"""
No Podman service is reachable to run the Minecraft server in
"""
def container_host() -> str:
"""
Find the URL of the Podman service socket. In order of preference:
- the ``CONTAINER_HOST`` environment variable
- the rootless socket ``$XDG_RUNTIME_DIR/podman/podman.sock``
- the rootful socket ``/run/podman/podman.sock``
:raises PodmanNotAvailableError: if no socket is found
"""
host = os.environ.get("CONTAINER_HOST")
if host:
return host
sockets = []
runtime_dir = os.environ.get("XDG_RUNTIME_DIR")
if runtime_dir:
sockets.append(Path(runtime_dir) / "podman" / "podman.sock")
sockets.append(ROOTFUL_SOCKET)
for socket in sockets:
if socket.is_socket():
return f"unix://{socket}"
raise PodmanNotAvailableError(
"No Podman socket found (looked for CONTAINER_HOST, then "
f"{', '.join(str(s) for s in sockets)}).\n"
"Start the Podman service with:\n"
" systemctl --user enable --now podman.socket"
)
def podman_client() -> PodmanClient:
"""
Connect to the Podman service, see `container_host`
:raises PodmanNotAvailableError: if the service does not respond
"""
url = container_host()
client = PodmanClient(base_url=url)
try:
client.ping()
except Exception as e:
client.close()
raise PodmanNotAvailableError(
f"Podman service at {url} is not responding ({e}).\n"
"Start the Podman service with:\n"
" systemctl --user enable --now podman.socket"
) from e
return client
def _is_rootless(client: PodmanClient) -> bool:
return bool(client.info()["host"]["security"]["rootless"])
[docs]
class MinecraftServer:
"""
Create and monitor Minecraft servers on the local machine using Podman
:param name: the name of the server
:param rcon: the rcon port for the server
:param password: the rcon password for the server
:param server_folder: the folder to store the server files in
:param world_type: the type of world to create (a world preset such as
"normal" or "flat")
:param keep: keep the server running: the container has the "always"
restart policy, so podman-restart.service restarts it after a reboot,
and `remove` does nothing unless forced
:param test: run the server in test mode (a flat world without structures
or mobs, regenerated by each `create`)
:param version: the Minecraft Java Edition release to run. Note that
an existing world is upgraded if it is opened with a newer release
"""
def __init__(
self,
name: str,
rcon: int,
password: str,
server_folder: Path,
world_type: str,
keep: bool = True,
test: bool = False,
version: str = MINECRAFT_VERSION,
) -> None:
"""
Create a `MinecraftServer` object only. Use `create` to
spin up the container.
"""
self.rcon = rcon
self.port = rcon + 1
self.name = name
self.password = password
self.server_folder = server_folder
self.world = self.server_folder / "world"
self.world_type = world_type
self.container: Container | None = None
self.keep = keep
self.test = test
self.version = version
# how long to wait for a server to start: a new world or a world upgrade
# to a new Minecraft release can take minutes
start_timeout = 300.0
def _logs_tail(self, lines: int = 50) -> str:
assert isinstance(self.container, Container)
logs = self.container.logs(tail=lines)
if not isinstance(logs, bytes):
# without stream=True podman returns an iterator of log lines
logs = b"".join(logs)
return logs.decode("utf-8", errors="replace")
[docs]
def wait_server(self):
"""
Wait until the server is ready to accept rcon connections
:raises RuntimeError: if the container stops, or the server does not
accept connections within ``start_timeout`` seconds. The error
includes the end of the server log.
"""
assert isinstance(self.container, Container)
log.info("waiting for server to come online ...")
deadline = monotonic() + self.start_timeout
while True:
self.container.reload()
if self.container.status not in STARTING_STATES:
raise RuntimeError(
f"minecraft server {self.name} failed to start "
f"(status {self.container.status})\n\n{self._logs_tail()}"
)
# RCON is the last thing to come up, so wait until it connects
client = Client(HOST, self.rcon, passwd=self.password, timeout=10)
try:
client.connect(True)
break
except WrongPassword:
raise
except Exception as e:
# e.g. refused, or reset by the port forwarder until RCON listens
log.debug(f"waiting for RCON: {e!r}")
finally:
client.close()
if monotonic() > deadline:
raise RuntimeError(
f"Timeout starting minecraft server {self.name}\n\n"
f"{self._logs_tail()}"
)
sleep(2)
log.info(f"Server {self.name} is online on port {self.port}")
[docs]
def stop(self):
"""
Stop the minecraft server
"""
assert isinstance(self.container, Container)
log.info(f"Stopping Minecraft Server {self.name} ...")
# allow time for the server to save the world before it is killed
# (ignore a container that is already stopped)
self.container.stop(timeout=stop_timeout, ignore=True)
self.container.wait()
log.info(f"Stopped Minecraft Server {self.name} ...")
[docs]
def start(self):
"""
Start the minecraft server
"""
assert isinstance(self.container, Container)
log.info(f"Starting Minecraft Server {self.name} on port {self.port}...")
self.container.start()
self.wait_server()
log.info(f"Started Minecraft Server {self.name} ...")
[docs]
def remove(self, force=False):
"""
Remove a minecraft server container
:param force: force the removal of the container
"""
if self.container and (not self.keep or force):
log.info(f"Removing Minecraft Server {self.name} ...")
self.stop()
self.container.remove() # noqa
[docs]
def create(self, force: bool = False) -> None:
"""
Spin up a minecraft server in a container, creating a new world if
the server folder has none.
If the container already exists it is reused (and started if it is
stopped) with the settings that it was created with. Use *force*, or
``mciwb stop``, to recreate it with new settings.
:param force: force the server to be removed if it already exists
"""
with podman_client() as client:
if client.containers.exists(self.name):
container = client.containers.get(self.name)
self.container = container
if force:
self.remove(force=True)
else:
self._reuse_settings(container)
if container.status == "running":
log.info(
f"Minecraft Server '{self.name}' "
f"already running on port {self.port}"
)
# it may still be booting, e.g. just after a reboot
self.wait_server()
else:
self.start()
return
self._run(client)
self.wait_server()
self._settings()
def _reuse_settings(self, container: Container) -> None:
"""
Adopt the RCON port and password of an existing container, warning
about any settings that differ from those requested
"""
env = dict(
e.split("=", 1) for e in container.attrs.get("Config", {}).get("Env") or []
)
mounts = container.attrs.get("Mounts") or []
data = [m.get("Source") for m in mounts if m.get("Destination") == "/data"]
ignored = []
rcon = env.get("RCON_PORT")
if rcon and int(rcon) != self.rcon:
ignored.append(f"port {self.rcon} (using {rcon})")
self.rcon = int(rcon)
self.port = self.rcon + 1
password = env.get("RCON_PASSWORD")
if password and password != self.password:
ignored.append("password (using the existing one)")
self.password = password
version = env.get("VERSION")
if version and version != self.version:
ignored.append(f"minecraft version {self.version} (using {version})")
self.version = version
if data and data[0] and Path(data[0]) != Path(self.server_folder).resolve():
ignored.append(f"folder {self.server_folder} (using {data[0]})")
self.server_folder = Path(data[0])
self.world = self.server_folder / "world"
if ignored:
log.warning(
f"Minecraft Server '{self.name}' already exists, so its settings "
f"are reused, ignoring: {'; '.join(ignored)}. "
f"Run 'mciwb stop --server-name {self.name}' first to "
"recreate it with new settings."
)
def _run(self, client: PodmanClient) -> None:
"""
Create and start the server container
"""
log.info(f"Launching Minecraft Server '{self.name}' on port {self.port} ...")
env = {
"EULA": "TRUE",
"VERSION": self.version,
"SERVER_PORT": self.port,
"RCON_PORT": self.rcon,
"ENABLE_RCON": "true",
"RCON_PASSWORD": self.password,
"SEED": 0,
"LEVEL_TYPE": self.world_type,
"MODE": "creative",
"SPAWN_PROTECTION": 0,
# keep ticking with no players online so that scripted builds
# behave (e.g. falling blocks, flowing water) and the items
# dropped by mcwb.get_block fall out of the world
"PAUSE_WHEN_EMPTY_SECONDS": 0,
}
# files in the server folder must belong to the current user, so that
# they can be backed up, restored and deleted without root
if _is_rootless(client):
# map the current user to the uid that the image runs the server
# as (the image's default UID and GID)
# (the equivalent of --userns=keep-id:uid=1000,gid=1000)
keep_id = {"nsmode": "keep-id", "value": f"uid={IMAGE_UID},gid={IMAGE_UID}"}
userns: dict[str, Any] = {"userns_mode": keep_id}
else:
# no user namespace: ask the image to run the server as the
# current user
userns = {}
env["UID"] = os.getuid()
env["GID"] = os.getgid()
if self.test:
env.update(
{
"GENERATE_STRUCTURES": "false",
"VIEW_DISTANCE": "5",
"LEVEL_TYPE": "flat",
"FORCE_WORLD_COPY": "TRUE",
}
)
# offline mode disables OPS so dont use it if we are keeping the server
# for local testing. But normally for running CI we want this option.
if not self.keep:
env["ONLINE_MODE"] = "FALSE"
if not self.server_folder.exists():
self.server_folder.mkdir(parents=True)
elif self.test:
shutil.rmtree(self.server_folder)
self.server_folder.mkdir(parents=True)
container = client.containers.run(
SERVER_IMAGE,
detach=True,
# the podman API only accepts string values
environment={k: str(v) for k, v in env.items()},
ports={f"{self.rcon}/tcp": self.rcon, f"{self.port}/tcp": self.port},
restart_policy={"Name": "always" if self.keep else "no"},
volumes={
str(self.server_folder): {"bind": "/data", "mode": "rw"},
},
name=self.name,
# let the server use the bind mounts on SELinux hosts
security_opt=["disable"],
**userns,
)
assert isinstance(container, Container)
self.container = container
def _settings(self):
"""
Some default settings for the server that this class creates
"""
with Client(HOST, self.rcon, passwd=self.password) as client:
if self.test:
# keep the chunks around the world origin loaded, so that the
# system tests can build there with no players online
client.forceload.add((-48, -48), (47, 47))
# the spawn-animals/monsters/npcs server properties no longer
# exist, this game rule controls natural mob spawning instead
client.gamerule("spawn_mobs", False)
else:
# keep the chunk at the world origin loaded
client.forceload.add((0, 0), (0, 0))
# a nice starting point for the tutorials in seed 0 world
client.setworldspawn(Vec3(632, 73, -1658))
[docs]
@classmethod
def stop_named(cls, name: str):
"""
Stop a minecraft server by name
:param name: the name of the server to stop
"""
with podman_client() as client:
if not client.containers.exists(name):
log.warning(f"Minecraft Server '{name}' not found")
return
container = client.containers.get(name)
log.info(f"Stopping Minecraft Server {name} ...")
container.stop(timeout=stop_timeout, ignore=True)
container.wait()
container.remove()
