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.9 KiB
106 lines
2.9 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, Optional, Union
|
|
|
|
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: Optional[str] = None
|
|
|
|
@classmethod
|
|
def from_string(cls, string: str) -> Config:
|
|
"""Reads 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:
|
|
"""Creates 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: Union[Path, Iterable[Path]] = CONFIG_FILES) -> None:
|
|
"""Reads 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)
|
|
|