syncplay-docker/boot.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Syncplay Bootstrap using to convert redesigned parameter fields into arguments
that are non-intrusive to Syncplay Server.

The official arguments of Syncplay are not convenient for container startup,
especially for file specification scenarios, which can easily confuse people
who use docker.

Document: https://man.archlinux.org/man/extra/syncplay/syncplay-server.1

┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃ Docker Arguments              ┃ Official Arguments                    ┃
┠───────────────────────────────╂───────────────────────────────────────┨
┃ --motd [MESSAGE]              ┃ --motd-file [FILE]                    ┃
┃ --salt [SALT] & --random-salt ┃ --salt [SALT]                         ┃
┃ --enable-stats                ┃ --stats-db-file [FILE]                ┃
┃ --enable-tls                  ┃ --tls [PATH]                          ┃
┃ --persistent                  ┃ --permanent-rooms-file [FILE]         ┃
┃ --max-username [NUM]          ┃ --max-username-length [NUM]           ┃
┃ --max-chat-message [NUM]      ┃ --max-chat-message-length [NUM]       ┃
┃ --permanent-rooms [ROOM ...]  ┃ --permanent-rooms-file [FILE]         ┃
┃ --listen-ipv4 [IPv4]          ┃ --ipv4-only & --interface-ipv4 [IPv4] ┃
┃ --listen-ipv6 [IPv6]          ┃ --ipv6-only & --interface-ipv6 [IPv6] ┃
┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┻━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛

Through this adapter, you no longer need to create files and specify paths, but
directly configure arguments through the command line or other methods.
"""

import os
import sys
import yaml
import argparse
from typing import Any
from typing import Generator
from syncplay import ep_server


class SyncplayBoot:
    """ Handle Syncplay bootstrap arguments. """
    def __debug(self, prefix: str, message: Any) -> None:
        """ Print out debug information. """
        if self.__debug_mode:
            print(f'\033[33m{prefix}\033[0m -> \033[90m{message}\033[0m', file=sys.stderr)

    def __temp_file(self, file: str, content: str) -> str:
        """ Create and save content to temporary files. """
        file = os.path.join(self.__temp_dir, file)
        with open(file, 'w', encoding='utf-8') as fp:
            fp.write(content)
        return file

    def __build_parser(self) -> Generator:
        """ Build arguments parser for Syncplay bootstrap. """
        parser = argparse.ArgumentParser(description='Syncplay Docker Bootstrap')
        yield parser.add_argument('-p', '--port', type=int, help='listen port of syncplay server')
        yield parser.add_argument('--password', metavar='PASSWD', type=str, help='authentication of syncplay server')
        yield parser.add_argument('--motd', metavar='MESSAGE', type=str, help='welcome text after the user enters the room')
        yield parser.add_argument('--salt', metavar='TEXT', type=str, help='string used to secure passwords')
        yield parser.add_argument('--random-salt', action='store_true', help='use a randomly generated salt value')
        yield parser.add_argument('--isolate-rooms', action='store_true', help='room isolation enabled')
        yield parser.add_argument('--disable-chat', action='store_true', help='disables the chat feature')
        yield parser.add_argument('--disable-ready', action='store_true', help='disables the readiness indicator feature')
        yield parser.add_argument('--enable-stats', action='store_true', help='enable syncplay server statistics')
        yield parser.add_argument('--enable-tls', action='store_true', help='enable tls support of syncplay server')
        yield parser.add_argument('--persistent', action='store_true', help='enables room persistence')
        yield parser.add_argument('--max-username', metavar='NUM', type=int, help='maximum length of usernames')
        yield parser.add_argument('--max-chat-message', metavar='NUM', type=int, help='maximum length of chat messages')
        yield parser.add_argument('--permanent-rooms', metavar='ROOM', type=str, nargs='*', help='permanent rooms of syncplay server')
        yield parser.add_argument('--listen-ipv4', metavar='INTERFACE', type=str, help='listening address of ipv4')
        yield parser.add_argument('--listen-ipv6', metavar='INTERFACE', type=str, help='listening address of ipv6')
        self.__parser = parser

    def __build_options(self) -> Generator:
        """ Build options list for Syncplay bootstrap. """
        for action in [x for x in self.__build_parser()]:
            is_list = type(action.nargs) is str
            opt_type = bool if action.type is None else action.type
            yield action.dest, opt_type, is_list

    def __init__(self, args: list[str], config: dict[str, Any],
                 cert_dir: str, temp_dir: str, work_dir: str, debug_mode: bool = False):
        self.__debug_mode = debug_mode
        self.__cert_dir, self.__temp_dir, self.__work_dir = cert_dir, temp_dir, work_dir
        self.__options = [x for x in self.__build_options()]  # list[(NAME, TYPE, IS_LIST)]
        self.__debug('Bootstrap options', self.__options)

        env_opts = self.__load_from_env()
        self.__debug('Environment options', env_opts)
        cfg_opts = self.__load_from_config(config)
        self.__debug('Configure file options', cfg_opts)
        cli_opts = self.__load_from_args(args)
        self.__debug('Command line options', cli_opts)

        options = env_opts | cfg_opts | cli_opts
        self.__opts = {x: y for x, y in options.items() if y != False}
        self.__debug('Bootstrap final options', self.__opts)

    def __load_from_args(self, raw_args: list[str]) -> dict[str, Any]:
        """ Loading options from command line arguments. """
        args = self.__parser.parse_args(raw_args)
        self.__debug('Command line arguments', args)
        arg_filter = lambda x: x is not None and x is not False
        return {x: y for x, y in vars(args).items() if arg_filter(y)}

    def __load_from_config(self, config: dict[str, Any]) -> dict[str, Any]:
        """ Loading options from configure file. """
        self.__debug('Configure file', config)
        options = {x[0].replace('_', '-'): x[0] for x in self.__options}
        return {options[x]: config[x] for x in options if x in config}

    def __load_from_env(self) -> dict[str, Any]:
        """ Loading options from environment variables. """
        def __convert(opt_raw: str, opt_field: str, opt_type: type) -> tuple[str, Any]:
            if opt_type is str:
                return opt_field, opt_raw
            elif opt_type is int:
                return opt_field, int(opt_raw)
            elif opt_type is bool:
                return opt_field, opt_raw.upper() in ['ON', 'TRUE']

        self.__debug('Environment variables', os.environ)
        options = {x.upper(): (x, t) for x, t, is_list in self.__options if not is_list}  # filter non-list options
        return dict([__convert(os.environ[x], *y) for x, y in options.items() if x in os.environ])

    def release(self) -> list[str]:
        """ Construct the startup arguments for syncplay server. """
        args = ['--port', str(self.__opts.get('port', 8999))]
        if 'password' in self.__opts:
            args += ['--password', self.__opts['password']]
        if 'motd' in self.__opts:
            args += ['--motd-file', self.__temp_file('motd.data', self.__opts['motd'])]

        salt = self.__opts.get('salt', None if 'random_salt' in self.__opts else '')
        if salt is not None:
            args += ['--salt', salt]  # using random salt without this option
        for opt in ['isolate_rooms', 'disable_chat', 'disable_ready']:
            if opt in self.__opts:
                args.append(f'--{opt}'.replace('_', '-'))

        if 'enable_stats' in self.__opts:
            args += ['--stats-db-file', os.path.join(self.__work_dir, 'stats.db')]
        if 'enable_tls' in self.__opts:
            args += ['--tls', self.__cert_dir]
        if 'persistent' in self.__opts:
            args += ['--rooms-db-file', os.path.join(self.__work_dir, 'rooms.db')]

        if 'max_username' in self.__opts:
            args += ['--max-username-length', str(self.__opts['max_username'])]
        if 'max_chat_message' in self.__opts:
            args += ['--max-chat-message-length', str(self.__opts['max_chat_message'])]
        if 'permanent_rooms' in self.__opts:
            rooms = '\n'.join(self.__opts['permanent_rooms'])
            args += ['--permanent-rooms-file', self.__temp_file('rooms.list', rooms)]

        if 'listen_ipv4' in self.__opts and 'listen_ipv6' in self.__opts:
            args += ['--interface-ipv4', self.__opts['listen_ipv4']]
            args += ['--interface-ipv6', self.__opts['listen_ipv6']]
        elif 'listen_ipv4' in self.__opts:
            args += ['--ipv4-only', '--interface-ipv4', self.__opts['listen_ipv4']]
        elif 'listen_ipv6' in self.__opts:
            args += ['--ipv6-only', '--interface-ipv6', self.__opts['listen_ipv6']]

        self.__debug('Syncplay startup arguments', args)
        return args


def syncplay_boot() -> None:
    """ Bootstrap the syncplay server. """
    temp_dir = os.environ.get('TEMP_DIR', '/tmp')
    work_dir = os.environ.get('WORK_DIR', '/data')
    cert_dir = os.environ.get('CERT_DIR', '/certs')
    config_file = os.environ.get('CONFIG', 'config.yml')
    debug_mode = os.environ.get('DEBUG', '').upper() in ['ON', 'TRUE']

    config = yaml.safe_load(open(config_file).read()) if os.path.exists(config_file) else {}
    bootstrap = SyncplayBoot(sys.argv[1:], config, cert_dir, temp_dir, work_dir, debug_mode)
    sys.argv = ['syncplay'] + bootstrap.release()


if __name__ == '__main__':
    syncplay_boot()
    sys.exit(ep_server.main())