FreedomBox/actions/sharing

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

import argparse
import json
import os
import pathlib
import re

import augeas

from plinth import action_utils

APACHE_CONFIGURATION = '/etc/apache2/conf-available/sharing-freedombox.conf'


def parse_arguments():
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest='subcommand', help='Sub command')

    subparsers.add_parser('list', help='List all existing shares')

    add_parser = subparsers.add_parser('add', help='Add a new share')
    add_parser.add_argument('--name', required=True, help='Name of the share')
    add_parser.add_argument('--path', required=True, help='Disk path to share')
    add_parser.add_argument('--groups', nargs='*',
                            help='List of groups that can access the share')
    add_parser.add_argument('--is-public', required=False, default=False,
                            action="store_true",
                            help='Allow public access to this share')

    remove_parser = subparsers.add_parser('remove',
                                          help='Remove an existing share')
    remove_parser.add_argument('--name', required=True,
                               help='Name of the share to remove')

    subparsers.required = True
    return parser.parse_args()


def load_augeas():
    """Initialize augeas for this app's configuration file."""
    aug = augeas.Augeas(flags=augeas.Augeas.NO_LOAD +
                        augeas.Augeas.NO_MODL_AUTOLOAD)
    aug.set('/augeas/load/Httpd/lens', 'Httpd.lns')
    aug.set('/augeas/load/Httpd/incl[last() + 1]', APACHE_CONFIGURATION)
    aug.load()

    aug.defvar('conf', '/files' + APACHE_CONFIGURATION)

    return aug


def subcommand_add(arguments):
    """Add a share to Apache configuration."""
    name = arguments.name
    path = '"' + arguments.path.replace('"', r'\"') + '"'
    groups = arguments.groups
    is_public = arguments.is_public
    url = '/share/' + name

    if not os.path.exists(APACHE_CONFIGURATION):
        pathlib.Path(APACHE_CONFIGURATION).touch()

    aug = load_augeas()
    shares = _list(aug)
    if any([share for share in shares if share['name'] == name]):
        raise Exception('Share already present')

    aug.set('$conf/directive[last() + 1]', 'Alias')
    aug.set('$conf/directive[last()]/arg[1]', url)
    aug.set('$conf/directive[last()]/arg[2]', path)

    aug.set('$conf/Location[last() + 1]/arg', url)

    aug.set('$conf/Location[last()]/directive[last() + 1]', 'Include')
    aug.set('$conf/Location[last()]/directive[last()]/arg',
            'includes/freedombox-sharing.conf')

    if not is_public:
        aug.set('$conf/Location[last()]/directive[last() + 1]', 'Include')
        aug.set('$conf/Location[last()]/directive[last()]/arg',
                'includes/freedombox-single-sign-on.conf')

        aug.set('$conf/Location[last()]/IfModule/arg', 'mod_auth_pubtkt.c')
        aug.set('$conf/Location[last()]/IfModule/directive[1]', 'TKTAuthToken')
        for group_name in groups:
            aug.set(
                '$conf/Location[last()]/IfModule/directive[1]/arg[last() + 1]',
                group_name)
    else:
        aug.set('$conf/Location[last()]/directive[last() + 1]', 'Require')
        aug.set('$conf/Location[last()]/directive[last()]/arg[1]', 'all')
        aug.set('$conf/Location[last()]/directive[last()]/arg[2]', 'granted')

    aug.save()

    with action_utils.WebserverChange() as webserver_change:
        webserver_change.enable('sharing-freedombox')


def subcommand_remove(arguments):
    """Remove a share from Apache configuration."""
    url_to_remove = '/share/' + arguments.name

    aug = load_augeas()

    for directive in aug.match('$conf/directive'):
        if aug.get(directive) != 'Alias':
            continue

        url = aug.get(directive + '/arg[1]')
        if url == url_to_remove:
            aug.remove(directive)

    for location in aug.match('$conf/Location'):
        url = aug.get(location + '/arg')
        if url == url_to_remove:
            aug.remove(location)

    aug.save()

    with action_utils.WebserverChange() as webserver_change:
        webserver_change.enable('sharing-freedombox')


def _get_name_from_url(url):
    """Return the name of the share given the URL for it."""
    matches = re.match(r'/share/([a-z0-9\-]*)', url)
    if not matches:
        raise ValueError

    return matches[1]


def _list(aug=None):
    """List all Apache configuration shares."""
    if not aug:
        aug = load_augeas()

    shares = []

    for match in aug.match('$conf/directive'):
        if aug.get(match) != 'Alias':
            continue

        url = aug.get(match + '/arg[1]')
        path = aug.get(match + '/arg[2]')

        path = path.removesuffix('"').removeprefix('"')
        path = path.replace(r'\"', '"')
        try:
            name = _get_name_from_url(url)
            shares.append({
                'name': name,
                'path': path,
                'url': '/share/' + name
            })
        except ValueError:
            continue

    for location in aug.match('$conf/Location'):
        url = aug.get(location + '/arg')

        try:
            name = _get_name_from_url(url)
        except ValueError:
            continue

        groups = []
        for group in aug.match(location + '//directive["TKTAuthToken"]/arg'):
            groups.append(aug.get(group))

        def _is_public():
            """Must contain the line 'Require all granted'."""
            require = location + '//directive["Require"]'
            return bool(aug.match(require)) and aug.get(
                require +
                '/arg[1]') == 'all' and aug.get(require +
                                                '/arg[2]') == 'granted'

        for share in shares:
            if share['name'] == name:
                share['groups'] = groups
                share['is_public'] = _is_public()

    return shares


def subcommand_list(_):
    """List all Apache configuration shares and print as JSON."""
    print(json.dumps({'shares': _list()}))


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()