Quickly deploy Syncplay server
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.
 
 

186 lines
10 KiB

#!/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())