FreedomBox/actions/openvpn

#!/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 OpenVPN server.
"""

import argparse
import os
import subprocess

import augeas

from plinth import action_utils, utils

KEYS_DIRECTORY = '/etc/openvpn/freedombox-keys'

DH_KEY = '/etc/openvpn/freedombox-keys/pki/dh.pem'

OLD_SERVER_CONFIGURATION_PATH = '/etc/openvpn/freedombox.conf'
SERVER_CONFIGURATION_PATH = '/etc/openvpn/server/freedombox.conf'

OLD_SERVICE_NAME = 'openvpn@freedombox'
SERVICE_NAME = 'openvpn-server@freedombox'

CA_CERTIFICATE_PATH = os.path.join(KEYS_DIRECTORY, 'pki', 'ca.crt')
USER_CERTIFICATE_PATH = os.path.join(KEYS_DIRECTORY, 'pki', 'issued',
                                     '{username}.crt')
USER_KEY_PATH = os.path.join(KEYS_DIRECTORY, 'pki', 'private',
                             '{username}.key')
ATTR_FILE = os.path.join(KEYS_DIRECTORY, 'pki', 'index.txt.attr')

SERVER_CONFIGURATION = '''
port 1194
proto udp
dev tun
ca /etc/openvpn/freedombox-keys/pki/ca.crt
cert /etc/openvpn/freedombox-keys/pki/issued/server.crt
key /etc/openvpn/freedombox-keys/pki/private/server.key
dh /etc/openvpn/freedombox-keys/pki/dh.pem
server 10.91.0.0 255.255.255.0
keepalive 10 120
cipher AES-256-CBC
comp-lzo
verb 3
'''

CLIENT_CONFIGURATION = '''
client
remote {remote} 1194
proto udp
dev tun
nobind
remote-cert-tls server
cipher AES-256-CBC
comp-lzo
redirect-gateway
verb 3
<ca>
{ca}</ca>
<cert>
{cert}</cert>
<key>
{key}</key>'''

CERTIFICATE_CONFIGURATION = {
    'EASYRSA_BATCH': '1',
    'EASYRSA_KEY_SIZE': '4096',
    'KEY_CONFIG': '/usr/share/easy-rsa/openssl-easyrsa.cnf',
    'KEY_DIR': KEYS_DIRECTORY,
    'EASYRSA_OPENSSL': 'openssl',
    'EASYRSA_CA_EXPIRE': '3650',
    'EASYRSA_REQ_EXPIRE': '3650',
    'EASYRSA_REQ_COUNTRY': 'US',
    'EASYRSA_REQ_PROVINCE': 'NY',
    'EASYRSA_REQ_CITY': 'New York',
    'EASYRSA_REQ_ORG': 'FreedomBox',
    'EASYRSA_REQ_EMAIL': 'me@freedombox',
    'EASYRSA_REQ_OU': 'Home',
    'EASYRSA_REQ_NAME': 'FreedomBox'
}

COMMON_ARGS = {'env': CERTIFICATE_CONFIGURATION, 'cwd': KEYS_DIRECTORY}


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('is-setup', help='Return whether setup is completed')
    subparsers.add_parser('setup', help='Setup OpenVPN server configuration')
    subparsers.add_parser(
        'upgrade',
        help='Upgrade OpenVPN server configuration from older configuration')

    get_profile = subparsers.add_parser(
        'get-profile', help='Return the OpenVPN profile of a user')
    get_profile.add_argument('username', help='User to get profile for')
    get_profile.add_argument('remote_server',
                             help='The server name for the user to connect')

    subparsers.required = True
    return parser.parse_args()


def subcommand_is_setup(_):
    """Return whether setup is complete."""
    print('true' if utils.is_non_empty_file(DH_KEY) else 'false')


def subcommand_setup(_):
    """Setup configuration, CA and certificates."""
    _create_server_config()
    _create_certificates()
    _setup_firewall()
    action_utils.service_enable(SERVICE_NAME)
    action_utils.service_restart(SERVICE_NAME)


def subcommand_upgrade(_):
    """Upgrade from an older version if configured.

    Otherwise do nothing.
    """
    if os.path.exists(OLD_SERVER_CONFIGURATION_PATH):
        os.rename(OLD_SERVER_CONFIGURATION_PATH, SERVER_CONFIGURATION_PATH)

    if action_utils.service_is_enabled(OLD_SERVICE_NAME):
        action_utils.service_disable(OLD_SERVICE_NAME)
        action_utils.service_enable(SERVICE_NAME)


def _create_server_config():
    """Write server configuration."""
    with open(SERVER_CONFIGURATION_PATH, 'w') as file_handle:
        file_handle.write(SERVER_CONFIGURATION)


def _setup_firewall():
    """Add TUN device to internal zone in firewalld."""
    subprocess.call(
        ['firewall-cmd', '--zone', 'internal', '--add-interface', 'tun+'])
    subprocess.call([
        'firewall-cmd', '--permanent', '--zone', 'internal', '--add-interface',
        'tun+'
    ])


def _create_certificates():
    """Generate CA and server certificates."""
    try:
        os.mkdir(KEYS_DIRECTORY, 0o700)
    except FileExistsError:
        pass

    subprocess.check_call(['/usr/share/easy-rsa/easyrsa', 'init-pki'],
                          **COMMON_ARGS)
    subprocess.check_call(
        ['/usr/share/easy-rsa/easyrsa', 'build-ca', 'nopass'], **COMMON_ARGS)
    subprocess.check_call([
        '/usr/share/easy-rsa/easyrsa', 'build-server-full', 'server', 'nopass'
    ], **COMMON_ARGS)
    subprocess.check_call(['/usr/share/easy-rsa/easyrsa', 'gen-dh'],
                          **COMMON_ARGS)


def subcommand_get_profile(arguments):
    """Return the profile for a user."""
    username = arguments.username
    remote_server = arguments.remote_server

    if username == 'ca' or username == 'server':
        raise Exception('Invalid username')

    user_certificate = USER_CERTIFICATE_PATH.format(username=username)
    user_key = USER_KEY_PATH.format(username=username)

    if not _is_non_empty_file(user_certificate) or \
       not _is_non_empty_file(user_key):
        set_unique_subject('no')  # Set unique subject in attribute file to no
        subprocess.check_call([
            '/usr/share/easy-rsa/easyrsa', 'build-client-full', username,
            'nopass'
        ], **COMMON_ARGS)

    user_certificate_string = _read_file(user_certificate)
    user_key_string = _read_file(user_key)
    ca_string = _read_file(CA_CERTIFICATE_PATH)

    profile = CLIENT_CONFIGURATION.format(
        ca=ca_string, cert=user_certificate_string, key=user_key_string,
        remote=remote_server)

    print(profile)


def set_unique_subject(value):
    """ Sets the unique_subject value to a particular value"""
    aug = load_augeas()
    aug.set('/files' + ATTR_FILE + '/unique_subject', value)
    aug.save()


def _read_file(filename):
    """Return the entire contents of a file as string."""
    with open(filename, 'r') as file_handle:
        return ''.join(file_handle.readlines())


def _is_non_empty_file(filepath):
    """Return wheather a file exists and is not zero size."""
    return os.path.isfile(filepath) and os.path.getsize(filepath) > 0


def load_augeas():
    """Initialize Augeas."""
    aug = augeas.Augeas(flags=augeas.Augeas.NO_LOAD +
                        augeas.Augeas.NO_MODL_AUTOLOAD)

    # shell-script config file lens
    aug.set('/augeas/load/Simplevars/lens', 'Simplevars.lns')
    aug.set('/augeas/load/Simplevars/incl[last() + 1]', ATTR_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()