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
|
#!/usr/bin/env python3
#
# Copyright (C) 2018-2024 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.base import Warning
from vyos.config import Config
from vyos.configverify import verify_pki_certificate
from vyos.configverify import verify_pki_ca_certificate
from vyos.pki import wrap_certificate
from vyos.pki import wrap_private_key
from vyos.template import render
from vyos.utils.process import call
from vyos.utils.network import check_port_availability
from vyos.utils.process import is_systemd_service_running
from vyos.utils.network import is_listen_port_bind_service
from vyos.utils.dict import dict_search
from vyos import ConfigError
from passlib.hash import sha512_crypt
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 sha512_crypt.hash(password)
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,
with_recursive_defaults=True,
with_pki=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 not origin["accounting"]['radius']['server']:
raise ConfigError('OpenConnect accounting mode radius requires at least one RADIUS server')
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"] and
"radius" in ocserv["authentication"]["mode"]):
raise ConfigError('OpenConnect authentication modes are mutually-exclusive, remove either local or radius from your configuration')
if "radius" in ocserv["authentication"]["mode"]:
if not ocserv["authentication"]['radius']['server']:
raise ConfigError('OpenConnect authentication mode radius requires at least one RADIUS server')
if "local" in ocserv["authentication"]["mode"]:
if not ocserv.get("authentication", {}).get("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}')
# Validate that if identity-based-config is configured all child config nodes are set
if 'identity_based_config' in ocserv["authentication"]:
if 'disabled' not in ocserv["authentication"]["identity_based_config"]:
Warning("Identity based configuration files is a 3rd party addition. Use at your own risk, this might break the ocserv daemon!")
if 'mode' not in ocserv["authentication"]["identity_based_config"]:
raise ConfigError('OpenConnect radius identity-based-config enabled but mode not selected')
elif 'group' in ocserv["authentication"]["identity_based_config"]["mode"] and "radius" not in ocserv["authentication"]["mode"]:
raise ConfigError('OpenConnect config-per-group must be used with radius authentication')
if 'directory' not in ocserv["authentication"]["identity_based_config"]:
raise ConfigError('OpenConnect identity-based-config enabled but directory not set')
if 'default_config' not in ocserv["authentication"]["identity_based_config"]:
raise ConfigError('OpenConnect identity-based-config enabled but default-config not set')
else:
raise ConfigError('OpenConnect authentication mode required')
else:
raise ConfigError('OpenConnect authentication credentials required')
# Check ssl
if 'ssl' not in ocserv:
raise ConfigError('SSL missing on OpenConnect config!')
if 'certificate' not in ocserv['ssl']:
raise ConfigError('SSL certificate missing on OpenConnect config!')
verify_pki_certificate(ocserv, ocserv['ssl']['certificate'])
if 'ca_certificate' in ocserv['ssl']:
verify_pki_ca_certificate(ocserv, ocserv['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)
|
