FreedomBox/actions/firewall

#!/usr/bin/python3
#
# This file is part of FreedomBox.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
"""
Configuration helper for FreedomBox firewall interface.
"""

import argparse
import subprocess

import augeas

from plinth import action_utils


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

    # Setup
    subparsers.add_parser('setup', help='Perform basic firewall setup')

    # Get status
    subparsers.add_parser('get-status',
                          help='Get whether firewalld is running')

    # Get service status
    get_enabled_services = subparsers.add_parser(
        'get-enabled-services', help='Get list of enabled services')
    get_enabled_services.add_argument(
        '--zone', help='Zone from which the list is to be retrieved',
        required=True)

    # Get interface status
    get_interfaces = subparsers.add_parser(
        'get-interfaces', help='Get list of interfaces in a zone')
    get_interfaces.add_argument(
        '--zone', help='Zone from which the list is to be retrieved',
        required=True)

    # Add a service
    add_service = subparsers.add_parser('add-service', help='Add a service')
    add_service.add_argument('service', help='Name of the service to add')
    add_service.add_argument(
        '--zone', help='Zone to which service is to be added', required=True)

    # Remove a service status
    remove_service = subparsers.add_parser('remove-service',
                                           help='Remove a service')
    remove_service.add_argument('service',
                                help='Name of the service to remove')
    remove_service.add_argument(
        '--zone', help='Zone from which service is to be removed',
        required=True)

    subparsers.required = True
    return parser.parse_args()


def _flush_iptables_rules():
    """Flush firewalld iptables rules before restarting it.

    This is workaround for
    https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=914694

    This workaround can be removed if the bug is fixed or if firewalld starts
    defaulting to nftables again.

    The bug leads to firewalld failing to flush rules when there are custom
    chains in the rules. This only happens on firewalld iptables backend when
    nftables is running with iptables compatibility.

    Flushing the tables before a restart will make the restart succeed and
    after the restart nftables backend is used avoiding the problem.

    """
    rule_template = '*{table}\n-F\n-X\n-Z\nCOMMIT\n'
    iptables_rules = ''
    ip6tables_rules = ''
    for table in ['security', 'raw', 'mangle', 'nat', 'filter']:
        iptables_rules += rule_template.format(table=table)
        ip6tables_rules += rule_template.format(table=table)

    subprocess.run(['iptables-restore'], input=iptables_rules.encode(),
                   check=True)
    subprocess.run(['ip6tables-restore'], input=iptables_rules.encode(),
                   check=True)


def set_firewall_backend(backend):
    """Set FirewallBackend attribute to the specified string."""
    conf_file = '/etc/firewalld/firewalld.conf'
    aug = augeas.Augeas(flags=augeas.Augeas.NO_LOAD +
                        augeas.Augeas.NO_MODL_AUTOLOAD)

    # lens for shell-script config file
    aug.set('/augeas/load/Shellvars/lens', 'Shellvars.lns')
    aug.set('/augeas/load/Shellvars/incl[last() + 1]', conf_file)
    aug.load()

    old_backend = aug.get('/files/{}/FirewallBackend'.format(conf_file))
    aug.set('/files/{}/FirewallBackend'.format(conf_file),
            '{}'.format(backend))
    aug.save()

    if old_backend == 'iptables':
        _flush_iptables_rules()

    if backend != old_backend:
        action_utils.service_restart('firewalld')


def subcommand_setup(_):
    """Perform basic firewalld setup."""
    action_utils.service_enable('firewalld')
    subprocess.call(['firewall-cmd', '--set-default-zone=external'])
    set_firewall_backend('nftables')

    add_service('external', 'http')
    add_service('internal', 'http')
    add_service('external', 'https')
    add_service('internal', 'https')
    add_service('internal', 'dns')
    add_service('internal', 'dhcp')


def subcommand_get_status(_):
    """Print status of the firewalld service"""
    subprocess.call(['firewall-cmd', '--state'])


def subcommand_get_enabled_services(arguments):
    """Print the status of variours services"""
    subprocess.call(
        ['firewall-cmd', '--zone', arguments.zone, '--list-services'])


def subcommand_get_interfaces(arguments):
    """Print the list of interfaces in a zone."""
    subprocess.call(
        ['firewall-cmd', '--zone', arguments.zone, '--list-interfaces'])


def subcommand_add_service(arguments):
    """Permit a service in the firewall."""
    add_service(arguments.zone, arguments.service)


def add_service(zone, service):
    """Permit a service in the firewall."""
    subprocess.call(['firewall-cmd', '--zone', zone, '--add-service', service])
    subprocess.call([
        'firewall-cmd', '--zone', zone, '--permanent', '--add-service', service
    ])


def subcommand_remove_service(arguments):
    """Block a service in the firewall"""
    subprocess.call([
        'firewall-cmd', '--zone', arguments.zone, '--remove-service',
        arguments.service
    ])
    subprocess.call([
        'firewall-cmd', '--zone', arguments.zone, '--permanent',
        '--remove-service', arguments.service
    ])


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