mirror of https://github.com/conqp/rcon
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
106 lines
2.8 KiB
106 lines
2.8 KiB
"""RCON server configuration."""
|
|
|
|
from __future__ import annotations
|
|
from argparse import Namespace
|
|
from configparser import ConfigParser, SectionProxy
|
|
from getpass import getpass
|
|
from logging import getLogger
|
|
from os import getenv, name
|
|
from pathlib import Path
|
|
from typing import Iterable, NamedTuple
|
|
|
|
from rcon.exceptions import ConfigReadError, UserAbort
|
|
|
|
|
|
__all__ = ["CONFIG_FILES", "LOG_FORMAT", "SERVERS", "Config", "from_args"]
|
|
|
|
|
|
CONFIG = ConfigParser()
|
|
|
|
if name == "posix":
|
|
CONFIG_FILES = (
|
|
Path("/etc/rcon.conf"),
|
|
Path("/usr/local/etc/rcon.conf"),
|
|
Path.home().joinpath(".rcon.conf"),
|
|
)
|
|
elif name == "nt":
|
|
CONFIG_FILES = (
|
|
Path(getenv("LOCALAPPDATA")).joinpath("rcon.conf"),
|
|
Path.home().joinpath(".rcon.conf"),
|
|
)
|
|
else:
|
|
raise NotImplementedError(f"Unsupported operating system: {name}")
|
|
|
|
LOG_FORMAT = "[%(levelname)s] %(name)s: %(message)s"
|
|
LOGGER = getLogger("RCON Config")
|
|
SERVERS = {}
|
|
|
|
|
|
class Config(NamedTuple):
|
|
"""Represents server configuration."""
|
|
|
|
host: str
|
|
port: int
|
|
passwd: str | None = None
|
|
|
|
@classmethod
|
|
def from_string(cls, string: str) -> Config:
|
|
"""Read the credentials from the given string."""
|
|
try:
|
|
host, port = string.rsplit(":", maxsplit=1)
|
|
except ValueError:
|
|
raise ValueError(f"Invalid socket: {string}.") from None
|
|
|
|
port = int(port)
|
|
|
|
try:
|
|
passwd, host = host.rsplit("@", maxsplit=1)
|
|
except ValueError:
|
|
passwd = None
|
|
|
|
return cls(host, port, passwd)
|
|
|
|
@classmethod
|
|
def from_config_section(cls, section: SectionProxy) -> Config:
|
|
"""Create a credentials tuple from
|
|
the respective config section.
|
|
"""
|
|
host = section["host"]
|
|
port = section.getint("port")
|
|
passwd = section.get("passwd")
|
|
return cls(host, port, passwd)
|
|
|
|
|
|
def load(config_files: Path | Iterable[Path] = CONFIG_FILES) -> None:
|
|
"""Read the configuration files and populates SERVERS."""
|
|
|
|
SERVERS.clear()
|
|
CONFIG.read(config_files)
|
|
|
|
for section in CONFIG.sections():
|
|
SERVERS[section] = Config.from_config_section(CONFIG[section])
|
|
|
|
|
|
def from_args(args: Namespace) -> Config:
|
|
"""Get the credentials for a server from the respective arguments."""
|
|
|
|
try:
|
|
host, port, passwd = Config.from_string(args.server)
|
|
except ValueError:
|
|
load(args.config)
|
|
|
|
try:
|
|
host, port, passwd = SERVERS[args.server]
|
|
except KeyError:
|
|
LOGGER.error("No such server: %s.", args.server)
|
|
raise ConfigReadError() from None
|
|
|
|
if passwd is None:
|
|
try:
|
|
passwd = getpass("Password: ")
|
|
except (KeyboardInterrupt, EOFError):
|
|
print()
|
|
LOGGER.error("Aborted by user.")
|
|
raise UserAbort() from None
|
|
|
|
return Config(host, port, passwd)
|
|
|