summaryrefslogtreecommitdiff
path: root/src/conf_mode/vpn_openconnect.py
blob: 68da70d7dd8fe6207137f7e86c03987af4c95b56 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
#!/usr/bin/env python3
#
# Copyright (C) 2018-2023 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import os
from sys import exit

from vyos.config import Config
from vyos.configdict import dict_merge
from vyos.pki import wrap_certificate
from vyos.pki import wrap_private_key
from vyos.template import render
from vyos.util import call
from vyos.util import check_port_availability
from vyos.util import is_systemd_service_running
from vyos.util import is_listen_port_bind_service
from vyos.util import dict_search
from vyos.xml import defaults
from vyos import ConfigError
from crypt import crypt, mksalt, METHOD_SHA512
from time import sleep

from vyos import airbag
airbag.enable()

cfg_dir        = '/run/ocserv'
ocserv_conf    = cfg_dir + '/ocserv.conf'
ocserv_passwd  = cfg_dir + '/ocpasswd'
ocserv_otp_usr = cfg_dir + '/users.oath'
radius_cfg     = cfg_dir + '/radiusclient.conf'
radius_servers = cfg_dir + '/radius_servers'

# Generate hash from user cleartext password
def get_hash(password):
    return crypt(password, mksalt(METHOD_SHA512))


def _default_dict_cleanup(origin: dict, default_values: dict) -> dict:
    """
    https://vyos.dev/T2665
    Clear unnecessary key values in merged config by dict_merge function
    :param origin: config
    :type origin: dict
    :param default_values: default values
    :type default_values: dict
    :return: merged dict
    :rtype: dict
    """
    if 'mode' in origin["authentication"] and "local" in \
            origin["authentication"]["mode"]:
        del origin['authentication']['local_users']['username']['otp']
        if not origin["authentication"]["local_users"]["username"]:
            raise ConfigError(
                'Openconnect authentication mode local requires at least one user')
        default_ocserv_usr_values = \
        default_values['authentication']['local_users']['username']['otp']
        for user, params in origin['authentication']['local_users'][
            'username'].items():
            # Not every configuration requires OTP settings
            if origin['authentication']['local_users']['username'][user].get(
                    'otp'):
                origin['authentication']['local_users']['username'][user][
                    'otp'] = dict_merge(default_ocserv_usr_values,
                                        origin['authentication'][
                                            'local_users']['username'][user][
                                            'otp'])

    if 'mode' in origin["authentication"] and "radius" in \
            origin["authentication"]["mode"]:
        del origin['authentication']['radius']['server']['port']
        if not origin["authentication"]['radius']['server']:
            raise ConfigError(
                'Openconnect authentication mode radius requires at least one RADIUS server')
        default_values_radius_port = \
        default_values['authentication']['radius']['server']['port']
        for server, params in origin['authentication']['radius'][
            'server'].items():
            if 'port' not in params:
                params['port'] = default_values_radius_port

    if 'mode' in origin["accounting"] and "radius" in \
            origin["accounting"]["mode"]:
        del origin['accounting']['radius']['server']['port']
        if not origin["accounting"]['radius']['server']:
            raise ConfigError(
                'Openconnect accounting mode radius requires at least one RADIUS server')
        default_values_radius_port = \
            default_values['accounting']['radius']['server']['port']
        for server, params in origin['accounting']['radius'][
            'server'].items():
            if 'port' not in params:
                params['port'] = default_values_radius_port
    return origin


def get_config(config=None):
    if config:
        conf = config
    else:
        conf = Config()
    base = ['vpn', 'openconnect']
    if not conf.exists(base):
        return None

    ocserv = conf.get_config_dict(base, key_mangling=('-', '_'), get_first_key=True)
    # We have gathered the dict representation of the CLI, but there are default
    # options which we need to update into the dictionary retrived.
    default_values = defaults(base)
    ocserv = dict_merge(default_values, ocserv)
    # workaround a "know limitation" - https://vyos.dev/T2665
    ocserv = _default_dict_cleanup(ocserv, default_values)
    if ocserv:
        ocserv['pki'] = conf.get_config_dict(['pki'], key_mangling=('-', '_'),
                                get_first_key=True, no_tag_node_value_mangle=True)

    return ocserv

def verify(ocserv):
    if ocserv is None:
        return None
    # Check if listen-ports not binded other services
    # It can be only listen by 'ocserv-main'
    for proto, port in ocserv.get('listen_ports').items():
        if check_port_availability(ocserv['listen_address'], int(port), proto) is not True and \
                not is_listen_port_bind_service(int(port), 'ocserv-main'):
            raise ConfigError(f'"{proto}" port "{port}" is used by another service')

    # Check accounting
    if "accounting" in ocserv:
        if "mode" in ocserv["accounting"] and "radius" in ocserv["accounting"]["mode"]:
            if "authentication" not in ocserv or "mode" not in ocserv["authentication"]:
                raise ConfigError('Accounting depends on OpenConnect authentication configuration')
            elif "radius" not in ocserv["authentication"]["mode"]:
                raise ConfigError('RADIUS accounting must be used with RADIUS authentication')

    # Check authentication
    if "authentication" in ocserv:
        if "mode" in ocserv["authentication"]:
            if "local" in ocserv["authentication"]["mode"]:
                if "radius" in ocserv["authentication"]["mode"]:
                    raise ConfigError('OpenConnect authentication modes are mutually-exclusive, remove either local or radius from your configuration')
                if not ocserv["authentication"]["local_users"]:
                    raise ConfigError('openconnect mode local required at least one user')
                if not ocserv["authentication"]["local_users"]["username"]:
                    raise ConfigError('openconnect mode local required at least one user')
                else:
                    # For OTP mode: verify that each local user has an OTP key
                    if "otp" in ocserv["authentication"]["mode"]["local"]:
                        users_wo_key = []
                        for user, user_config in ocserv["authentication"]["local_users"]["username"].items():
                            # User has no OTP key defined
                            if dict_search('otp.key', user_config) == None:
                                users_wo_key.append(user)
                        if users_wo_key:
                            raise ConfigError(f'OTP enabled, but no OTP key is configured for these users:\n{users_wo_key}')
                    # For password (and default) mode: verify that each local user has password
                    if "password" in ocserv["authentication"]["mode"]["local"] or "otp" not in ocserv["authentication"]["mode"]["local"]:
                        users_wo_pswd = []
                        for user in ocserv["authentication"]["local_users"]["username"]:
                            if not "password" in ocserv["authentication"]["local_users"]["username"][user]:
                                users_wo_pswd.append(user)
                        if users_wo_pswd:
                            raise ConfigError(f'password required for users:\n{users_wo_pswd}')
        else:
            raise ConfigError('openconnect authentication mode required')
    else:
        raise ConfigError('openconnect authentication credentials required')

    # Check ssl
    if 'ssl' not in ocserv:
        raise ConfigError('openconnect ssl required')

    if not ocserv['pki'] or 'certificate' not in ocserv['pki']:
        raise ConfigError('PKI not configured')

    ssl = ocserv['ssl']
    if 'certificate' not in ssl:
        raise ConfigError('openconnect ssl certificate required')

    cert_name = ssl['certificate']

    if cert_name not in ocserv['pki']['certificate']:
        raise ConfigError('Invalid openconnect ssl certificate')

    cert = ocserv['pki']['certificate'][cert_name]

    if 'certificate' not in cert:
        raise ConfigError('Missing certificate in PKI')

    if 'private' not in cert or 'key' not in cert['private']:
        raise ConfigError('Missing private key in PKI')

    if 'ca_certificate' in ssl:
        if 'ca' not in ocserv['pki']:
            raise ConfigError('PKI not configured')

        if ssl['ca_certificate'] not in ocserv['pki']['ca']:
            raise ConfigError('Invalid openconnect ssl CA certificate')

    # Check network settings
    if "network_settings" in ocserv:
        if "push_route" in ocserv["network_settings"]:
            # Replace default route
            if "0.0.0.0/0" in ocserv["network_settings"]["push_route"]:
                ocserv["network_settings"]["push_route"].remove("0.0.0.0/0")
                ocserv["network_settings"]["push_route"].append("default")
        else:
            ocserv["network_settings"]["push_route"] = ["default"]
    else:
        raise ConfigError('openconnect network settings required')

def generate(ocserv):
    if not ocserv:
        return None

    if "radius" in ocserv["authentication"]["mode"]:
        if dict_search(ocserv, 'accounting.mode.radius'):
            # Render radius client configuration
            render(radius_cfg, 'ocserv/radius_conf.j2', ocserv)
            merged_servers = ocserv["accounting"]["radius"]["server"] | ocserv["authentication"]["radius"]["server"]
            # Render radius servers
            # Merge the accounting and authentication servers into a single dictionary
            render(radius_servers, 'ocserv/radius_servers.j2', {'server': merged_servers})
        else:
            # Render radius client configuration
            render(radius_cfg, 'ocserv/radius_conf.j2', ocserv)
            # Render radius servers
            render(radius_servers, 'ocserv/radius_servers.j2', ocserv["authentication"]["radius"])
    elif "local" in ocserv["authentication"]["mode"]:
        # if mode "OTP", generate OTP users file parameters
        if "otp" in ocserv["authentication"]["mode"]["local"]:
            if "local_users" in ocserv["authentication"]:
                for user in ocserv["authentication"]["local_users"]["username"]:
                    # OTP token type from CLI parameters:
                    otp_interval = str(ocserv["authentication"]["local_users"]["username"][user]["otp"].get("interval"))
                    token_type = ocserv["authentication"]["local_users"]["username"][user]["otp"].get("token_type")
                    otp_length = str(ocserv["authentication"]["local_users"]["username"][user]["otp"].get("otp_length"))
                    if token_type == "hotp-time":
                        otp_type = "HOTP/T" + otp_interval
                    elif token_type == "hotp-event":
                        otp_type = "HOTP/E"
                    else:
                        otp_type = "HOTP/T" + otp_interval
                    ocserv["authentication"]["local_users"]["username"][user]["otp"]["token_tmpl"] = otp_type + "/" + otp_length
        # if there is a password, generate hash
        if "password" in ocserv["authentication"]["mode"]["local"] or not "otp" in ocserv["authentication"]["mode"]["local"]:
            if "local_users" in ocserv["authentication"]:
                for user in ocserv["authentication"]["local_users"]["username"]:
                    ocserv["authentication"]["local_users"]["username"][user]["hash"] = get_hash(ocserv["authentication"]["local_users"]["username"][user]["password"])

        if "password-otp" in ocserv["authentication"]["mode"]["local"]:
            # Render local users ocpasswd
            render(ocserv_passwd, 'ocserv/ocserv_passwd.j2', ocserv["authentication"]["local_users"])
            # Render local users OTP keys
            render(ocserv_otp_usr, 'ocserv/ocserv_otp_usr.j2', ocserv["authentication"]["local_users"])
        elif "password" in ocserv["authentication"]["mode"]["local"]:
            # Render local users ocpasswd
            render(ocserv_passwd, 'ocserv/ocserv_passwd.j2', ocserv["authentication"]["local_users"])
        elif "otp" in ocserv["authentication"]["mode"]["local"]:
            # Render local users OTP keys
            render(ocserv_otp_usr, 'ocserv/ocserv_otp_usr.j2', ocserv["authentication"]["local_users"])
        else:
            # Render local users ocpasswd
            render(ocserv_passwd, 'ocserv/ocserv_passwd.j2', ocserv["authentication"]["local_users"])
    else:
        if "local_users" in ocserv["authentication"]:
            for user in ocserv["authentication"]["local_users"]["username"]:
                ocserv["authentication"]["local_users"]["username"][user]["hash"] = get_hash(ocserv["authentication"]["local_users"]["username"][user]["password"])
            # Render local users
            render(ocserv_passwd, 'ocserv/ocserv_passwd.j2', ocserv["authentication"]["local_users"])

    if "ssl" in ocserv:
        cert_file_path = os.path.join(cfg_dir, 'cert.pem')
        cert_key_path = os.path.join(cfg_dir, 'cert.key')
        ca_cert_file_path = os.path.join(cfg_dir, 'ca.pem')

        if 'certificate' in ocserv['ssl']:
            cert_name = ocserv['ssl']['certificate']
            pki_cert = ocserv['pki']['certificate'][cert_name]

            with open(cert_file_path, 'w') as f:
                f.write(wrap_certificate(pki_cert['certificate']))

            if 'private' in pki_cert and 'key' in pki_cert['private']:
                with open(cert_key_path, 'w') as f:
                    f.write(wrap_private_key(pki_cert['private']['key']))

        if 'ca_certificate' in ocserv['ssl']:
            ca_name = ocserv['ssl']['ca_certificate']
            pki_ca_cert = ocserv['pki']['ca'][ca_name]

            with open(ca_cert_file_path, 'w') as f:
                f.write(wrap_certificate(pki_ca_cert['certificate']))

    # Render config
    render(ocserv_conf, 'ocserv/ocserv_config.j2', ocserv)


def apply(ocserv):
    if not ocserv:
        call('systemctl stop ocserv.service')
        for file in [ocserv_conf, ocserv_passwd, ocserv_otp_usr]:
            if os.path.exists(file):
                os.unlink(file)
    else:
        call('systemctl reload-or-restart ocserv.service')
        counter = 0
        while True:
            # exit early when service runs
            if is_systemd_service_running("ocserv.service"):
                break
            sleep(0.250)
            if counter > 5:
                raise ConfigError('openconnect failed to start, check the logs for details')
                break
            counter += 1


if __name__ == '__main__':
    try:
        c = get_config()
        verify(c)
        generate(c)
        apply(c)
    except ConfigError as e:
        print(e)
        exit(1)