FreedomBox/actions/coturn

#!/usr/bin/python3
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
Configuration helper for Coturn daemon.
"""

import argparse
import json
import pathlib
import random
import shutil
import string

import augeas

from plinth import action_utils

CONFIG_FILE = pathlib.Path('/etc/coturn/freedombox.conf')


def parse_arguments():
    """Return parsed command line arguments as dictionary."""
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest='subcommand', help='Sub command')

    subparsers.add_parser('setup', help='Setup Coturn server')
    subparsers.add_parser('get-config',
                          help='Return the current configuration')
    subparser = subparsers.add_parser('set-domain', help='Set the TLS domain')
    subparser.add_argument('domain_name', help='TLS domain name to set')

    subparsers.required = True
    return parser.parse_args()


def _key_path(key):
    """Return the augeas path for a key."""
    return '/files' + str(CONFIG_FILE) + '/' + key


def subcommand_setup(_):
    """Setup Coturn server."""
    CONFIG_FILE.parent.mkdir(exist_ok=True)
    if not CONFIG_FILE.exists():
        CONFIG_FILE.touch(0o640)
        shutil.chown(CONFIG_FILE, group='turnserver')

    action_utils.service_daemon_reload()
    action_utils.service_try_restart('coturn')

    aug = augeas_load()

    # XXX: Should we set listen, relay IP address to :: or dynamically
    # XXX: Should we set external-ip
    aug.set(_key_path('min-port'), '49152')
    aug.set(_key_path('max-port'), '50175')
    aug.set(_key_path('use-auth-secret'), 'true')
    if not aug.get(_key_path('static-auth-secret')):
        secret = ''.join(
            random.choice(string.ascii_letters + string.digits)
            for _ in range(64))
        aug.set(_key_path('static-auth-secret'), secret)

    aug.set(_key_path('cert'), '/etc/coturn/certs/cert.pem')
    aug.set(_key_path('pkey'), '/etc/coturn/certs/pkey.pem')
    aug.set(_key_path('no-tlsv1'), 'true')
    aug.set(_key_path('no-tlsv1_1'), 'true')
    aug.set(_key_path('no-cli'), 'true')

    aug.save()


def subcommand_get_config(_):
    """Return the current configuration in JSON format."""
    aug = augeas_load()
    config = {
        'static_auth_secret': aug.get(_key_path('static-auth-secret')),
        'realm': aug.get(_key_path('realm')),
    }
    print(json.dumps(config))


def subcommand_set_domain(arguments):
    """Set the TLS domain.

    This value is usually not stored. So, set realm value even though it is not
    needed to set realm for REST API based authentication.

    """
    aug = augeas_load()
    aug.set(_key_path('realm'), arguments.domain_name)
    aug.save()


def augeas_load():
    """Initialize Augeas."""
    aug = augeas.Augeas(flags=augeas.Augeas.NO_LOAD +
                        augeas.Augeas.NO_MODL_AUTOLOAD)
    aug.set('/augeas/load/Simplevars/lens', 'Simplevars.lns')
    aug.set('/augeas/load/Simplevars/incl[last() + 1]', str(CONFIG_FILE))
    aug.load()

    return aug


def main():
    """Parse arguments and perform all duties."""
    arguments = parse_arguments()

    subcommand = arguments.subcommand.replace('-', '_')
    subcommand_method = globals()['subcommand_' + subcommand]
    subcommand_method(arguments)


if __name__ == '__main__':
    main()