diff --git a/op-mode-definitions/vpn-ipsec.xml.in b/op-mode-definitions/vpn-ipsec.xml.in
index c7ba780a3..b551af2be 100644
--- a/op-mode-definitions/vpn-ipsec.xml.in
+++ b/op-mode-definitions/vpn-ipsec.xml.in
@@ -1,296 +1,296 @@
<?xml version="1.0" encoding="UTF-8"?>
<interfaceDefinition>
<node name="reset">
<children>
<node name="vpn">
<properties>
<help>Reset Virtual Private Network (VPN) information</help>
</properties>
<children>
<node name="ipsec">
<properties>
<help>Reset IPSec VPN sessions</help>
</properties>
<children>
<tagNode name="profile">
<properties>
<help>Reset a specific tunnel for given DMVPN profile</help>
<completionHelp>
<path>vpn ipsec profile</path>
</completionHelp>
</properties>
<children>
<tagNode name="tunnel">
<properties>
<help>Reset a specific tunnel for given DMVPN profile</help>
<completionHelp>
<script>sudo ${vyos_completion_dir}/list_ipsec_profile_tunnels.py --profile ${COMP_WORDS[4]}</script>
</completionHelp>
</properties>
<children>
<tagNode name="remote-host">
<properties>
<help>Reset a specific tunnel for given DMVPN NBMA</help>
<completionHelp>
<list><x.x.x.x> <h:h:h:h:h:h:h:h></list>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/ipsec.py reset_profile_dst --profile="$5" --tunnel="$7" --nbma-dst="$9"</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/ipsec.py reset_profile_all --profile="$5" --tunnel="$7"</command>
</tagNode>
</children>
</tagNode>
<node name="remote-access">
<properties>
<help>Reset remote access IPSec VPN connections</help>
</properties>
<children>
<node name="all">
<properties>
<help>Reset all users current remote access IPSec VPN sessions</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/ipsec.py reset_ra</command>
</node>
<tagNode name="user">
<properties>
<help>Reset specified user current remote access IPsec VPN session(s)</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/ipsec.py reset_ra --user="$6"</command>
</tagNode>
</children>
</node>
<node name="site-to-site">
<properties>
<help>Reset site-to-site IPSec VPN connections</help>
</properties>
<children>
<node name="all">
<properties>
<help>Reset all site-to-site IPSec VPN sessions</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/ipsec.py reset_all_peers</command>
</node>
<tagNode name="peer">
<properties>
<help>Reset all tunnels for given peer</help>
<completionHelp>
<path>vpn ipsec site-to-site peer</path>
</completionHelp>
</properties>
<children>
<tagNode name="tunnel">
<properties>
<help>Reset a specific tunnel for given peer</help>
<completionHelp>
<path>vpn ipsec site-to-site peer ${COMP_WORDS[5]} tunnel</path>
</completionHelp>
</properties>
<command>sudo ${vyos_op_scripts_dir}/ipsec.py reset_peer --peer="$6" --tunnel="$8"</command>
</tagNode>
<node name="vti">
<properties>
<help>Reset the VTI tunnel for given peer</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/ipsec.py reset_peer --peer="$6" --tunnel="vti"</command>
</node>
</children>
<command>sudo ${vyos_op_scripts_dir}/ipsec.py reset_peer --peer="$6"</command>
</tagNode>
</children>
</node>
</children>
</node>
</children>
</node>
</children>
</node>
<node name="restart">
<children>
<node name="ipsec">
<properties>
<help>Restart the IPsec VPN process</help>
</properties>
<command>if systemctl is-active --quiet strongswan; then sudo systemctl restart strongswan ; echo "IPsec process restarted"; else echo "IPsec process not running" ; fi</command>
</node>
</children>
</node>
<node name="show">
<children>
<node name="vpn">
<properties>
<help>Show Virtual Private Network (VPN) information</help>
</properties>
<children>
<node name="debug">
<properties>
<help>Show VPN debugging information</help>
</properties>
<children>
<tagNode name="peer">
<properties>
<help>Show debugging information for a peer</help>
<completionHelp>
<path>vpn ipsec site-to-site peer</path>
</completionHelp>
</properties>
<children>
<tagNode name="tunnel">
<properties>
<help>Show debug information for peer tunnel</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/vpn_ipsec.py --action="vpn-debug" --name="$5" --tunnel="$7"</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/vpn_ipsec.py --action="vpn-debug" --name="$5" --tunnel="all"</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/vpn_ipsec.py --action="vpn-debug" --name="all"</command>
</node>
<node name="ike">
<properties>
<help>Show Internet Key Exchange (IKE) information</help>
</properties>
<children>
<node name="sa">
<properties>
<help>Show all currently active IKE Security Associations (SA)</help>
</properties>
<children>
<node name="nat-traversal">
<properties>
<help>Show all currently active IKE Security Associations (SA) that are using NAT Traversal</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/vpn_ike_sa.py --nat="yes"</command>
</node>
<tagNode name="peer">
<properties>
<help>Show all currently active IKE Security Associations (SA) for a peer</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/vpn_ike_sa.py --peer="$6"</command>
</tagNode>
</children>
<command>sudo ${vyos_op_scripts_dir}/vpn_ike_sa.py</command>
</node>
<node name="secrets">
<properties>
<help>Show all the pre-shared key secrets</help>
</properties>
- <command>sudo cat /etc/ipsec.secrets | sed 's/#.*//'</command>
+ <command>${vyos_op_scripts_dir}/ipsec.py show_psk</command>
</node>
<node name="status">
<properties>
<help>Show summary of IKE process information</help>
</properties>
<command>if systemctl is-active --quiet strongswan ; then systemctl status strongswan ; else echo "Process is not running" ; fi</command>
</node>
</children>
</node>
<node name="ipsec">
<properties>
<help>Show Internet Protocol Security (IPsec) information</help>
</properties>
<children>
<node name="connections">
<properties>
<help>Show VPN connections</help>
</properties>
<command>sudo ${vyos_op_scripts_dir}/ipsec.py show_connections</command>
</node>
<node name="policy">
<properties>
<help>Show the in-kernel crypto policies</help>
</properties>
<command>sudo ip xfrm policy list</command>
</node>
<node name="remote-access">
<properties>
<help>Show active VPN server sessions</help>
</properties>
<children>
<node name="detail">
<properties>
<help>Show detail active IKEv2 RA sessions</help>
</properties>
<command>if systemctl is-active --quiet strongswan ; then sudo ${vyos_op_scripts_dir}/ipsec.py show_ra_detail; else echo "IPsec process not running" ; fi</command>
</node>
<tagNode name="connection-id">
<properties>
<help>Show detail active IKEv2 RA sessions by connection-id</help>
</properties>
<command>if systemctl is-active --quiet strongswan ; then sudo ${vyos_op_scripts_dir}/ipsec.py show_ra_detail --conn-id="$6"; else echo "IPsec process not running" ; fi</command>
</tagNode>
<node name="summary">
<properties>
<help>Show active IKEv2 RA sessions summary</help>
</properties>
<command>if systemctl is-active --quiet strongswan ; then sudo ${vyos_op_scripts_dir}/ipsec.py show_ra_summary; else echo "IPsec process not running" ; fi</command>
</node>
<tagNode name="username">
<properties>
<help>Show detail active IKEv2 RA sessions by username</help>
</properties>
<command>if systemctl is-active --quiet strongswan ; then sudo ${vyos_op_scripts_dir}/ipsec.py show_ra_detail --username="$6"; else echo "IPsec process not running" ; fi</command>
</tagNode>
</children>
</node>
<node name="sa">
<properties>
<help>Show all active IPsec Security Associations (SA)</help>
</properties>
<children>
<!--
<node name="detail">
<properties>
<help>Show Detail on all active IPsec Security Associations (SA)</help>
</properties>
<command></command>
</node>
<tagNode name="stats">
<properties>
<help>Show statistics for all currently active IPsec Security Associations (SA)</help>
<valueHelp>
<format>txt</format>
<description>Show Statistics for SAs associated with a specific peer</description>
</valueHelp>
</properties>
<children>
<tagNode name="tunnel">
<properties>
<help>Show Statistics for SAs associated with a specific peer</help>
</properties>
<command></command>
</tagNode>
</children>
<command></command>
</tagNode>
-->
<node name="detail">
<properties>
<help>Show Verbose Detail on all active IPsec Security Associations (SA)</help>
</properties>
<command>if systemctl is-active --quiet strongswan ; then sudo ${vyos_op_scripts_dir}/ipsec.py show_sa_detail ; else echo "IPsec process not running" ; fi</command>
</node>
</children>
<command>if systemctl is-active --quiet strongswan ; then sudo ${vyos_op_scripts_dir}/ipsec.py show_sa ; else echo "IPsec process not running" ; fi</command>
</node>
<node name="state">
<properties>
<help>Show the in-kernel crypto state</help>
</properties>
<command>sudo ip xfrm state list</command>
</node>
<node name="status">
<properties>
<help>Show status of IPsec process</help>
</properties>
<command>if systemctl is-active --quiet strongswan >/dev/null ; then echo -e "IPsec Process Running: $(pgrep charon)\n$(sudo /usr/sbin/ipsec status)" ; else echo "IPsec process not running" ; fi</command>
</node>
</children>
</node>
</children>
</node>
</children>
</node>
</interfaceDefinition>
diff --git a/src/op_mode/ipsec.py b/src/op_mode/ipsec.py
index 57d3cfed9..44d41219e 100755
--- a/src/op_mode/ipsec.py
+++ b/src/op_mode/ipsec.py
@@ -1,789 +1,828 @@
#!/usr/bin/env python3
#
# Copyright (C) 2022-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 re
import sys
import typing
from hurry import filesize
from re import split as re_split
from tabulate import tabulate
from vyos.utils.convert import convert_data
from vyos.utils.convert import seconds_to_human
from vyos.utils.process import cmd
from vyos.configquery import ConfigTreeQuery
import vyos.opmode
import vyos.ipsec
def _convert(text):
return int(text) if text.isdigit() else text.lower()
def _alphanum_key(key):
return [_convert(c) for c in re_split('([0-9]+)', str(key))]
def _get_raw_data_sas():
try:
get_sas = vyos.ipsec.get_vici_sas()
sas = convert_data(get_sas)
return sas
except (vyos.ipsec.ViciInitiateError) as err:
raise vyos.opmode.UnconfiguredSubsystem(err)
def _get_output_swanctl_sas_from_list(ra_output_list: list) -> str:
"""
Template for output for VICI
Inserts \n after each IKE SA
:param ra_output_list: IKE SAs list
:type ra_output_list: list
:return: formatted string
:rtype: str
"""
output = '';
for sa_val in ra_output_list:
for sa in sa_val.values():
swanctl_output: str = cmd(
f'sudo swanctl -l --ike-id {sa["uniqueid"]}')
output = f'{output}{swanctl_output}\n\n'
return output
def _get_formatted_output_sas(sas):
sa_data = []
for sa in sas:
for parent_sa in sa.values():
# create an item for each child-sa
for child_sa in parent_sa.get('child-sas', {}).values():
# prepare a list for output data
sa_out_name = sa_out_state = sa_out_uptime = sa_out_bytes = sa_out_packets = sa_out_remote_addr = sa_out_remote_id = sa_out_proposal = 'N/A'
# collect raw data
sa_name = child_sa.get('name')
sa_state = child_sa.get('state')
sa_uptime = child_sa.get('install-time')
sa_bytes_in = child_sa.get('bytes-in')
sa_bytes_out = child_sa.get('bytes-out')
sa_packets_in = child_sa.get('packets-in')
sa_packets_out = child_sa.get('packets-out')
sa_remote_addr = parent_sa.get('remote-host')
sa_remote_id = parent_sa.get('remote-id')
sa_proposal_encr_alg = child_sa.get('encr-alg')
sa_proposal_integ_alg = child_sa.get('integ-alg')
sa_proposal_encr_keysize = child_sa.get('encr-keysize')
sa_proposal_dh_group = child_sa.get('dh-group')
# format data to display
if sa_name:
sa_out_name = sa_name
if sa_state:
if sa_state == 'INSTALLED':
sa_out_state = 'up'
else:
sa_out_state = 'down'
if sa_uptime:
sa_out_uptime = seconds_to_human(sa_uptime)
if sa_bytes_in and sa_bytes_out:
bytes_in = filesize.size(int(sa_bytes_in))
bytes_out = filesize.size(int(sa_bytes_out))
sa_out_bytes = f'{bytes_in}/{bytes_out}'
if sa_packets_in and sa_packets_out:
packets_in = filesize.size(int(sa_packets_in),
system=filesize.si)
packets_out = filesize.size(int(sa_packets_out),
system=filesize.si)
packets_str = f'{packets_in}/{packets_out}'
sa_out_packets = re.sub(r'B', r'', packets_str)
if sa_remote_addr:
sa_out_remote_addr = sa_remote_addr
if sa_remote_id:
sa_out_remote_id = sa_remote_id
# format proposal
if sa_proposal_encr_alg:
sa_out_proposal = sa_proposal_encr_alg
if sa_proposal_encr_keysize:
sa_proposal_encr_keysize_str = sa_proposal_encr_keysize
sa_out_proposal = f'{sa_out_proposal}_{sa_proposal_encr_keysize_str}'
if sa_proposal_integ_alg:
sa_proposal_integ_alg_str = sa_proposal_integ_alg
sa_out_proposal = f'{sa_out_proposal}/{sa_proposal_integ_alg_str}'
if sa_proposal_dh_group:
sa_proposal_dh_group_str = sa_proposal_dh_group
sa_out_proposal = f'{sa_out_proposal}/{sa_proposal_dh_group_str}'
# add a new item to output data
sa_data.append([
sa_out_name, sa_out_state, sa_out_uptime, sa_out_bytes,
sa_out_packets, sa_out_remote_addr, sa_out_remote_id,
sa_out_proposal
])
headers = [
"Connection", "State", "Uptime", "Bytes In/Out", "Packets In/Out",
"Remote address", "Remote ID", "Proposal"
]
sa_data = sorted(sa_data, key=_alphanum_key)
output = tabulate(sa_data, headers)
return output
# Connections block
def _get_convert_data_connections():
try:
get_connections = vyos.ipsec.get_vici_connections()
connections = convert_data(get_connections)
return connections
except (vyos.ipsec.ViciInitiateError) as err:
raise vyos.opmode.UnconfiguredSubsystem(err)
def _get_parent_sa_proposal(connection_name: str, data: list) -> dict:
"""Get parent SA proposals by connection name
if connections not in the 'down' state
Args:
connection_name (str): Connection name
data (list): List of current SAs from vici
Returns:
str: Parent SA connection proposal
AES_CBC/256/HMAC_SHA2_256_128/MODP_1024
"""
if not data:
return {}
for sa in data:
# check if parent SA exist
if connection_name not in sa.keys():
continue
if 'encr-alg' in sa[connection_name]:
encr_alg = sa.get(connection_name, '').get('encr-alg')
cipher = encr_alg.split('_')[0]
mode = encr_alg.split('_')[1]
encr_keysize = sa.get(connection_name, '').get('encr-keysize')
integ_alg = sa.get(connection_name, '').get('integ-alg')
# prf_alg = sa.get(connection_name, '').get('prf-alg')
dh_group = sa.get(connection_name, '').get('dh-group')
proposal = {
'cipher': cipher,
'mode': mode,
'key_size': encr_keysize,
'hash': integ_alg,
'dh': dh_group
}
return proposal
return {}
def _get_parent_sa_state(connection_name: str, data: list) -> str:
"""Get parent SA state by connection name
Args:
connection_name (str): Connection name
data (list): List of current SAs from vici
Returns:
Parent SA connection state
"""
ike_state = 'down'
if not data:
return ike_state
for sa in data:
# check if parent SA exist
for connection, connection_conf in sa.items():
if connection_name != connection:
continue
if connection_conf['state'].lower() == 'established':
ike_state = 'up'
return ike_state
def _get_child_sa_state(connection_name: str, tunnel_name: str,
data: list) -> str:
"""Get child SA state by connection and tunnel name
Args:
connection_name (str): Connection name
tunnel_name (str): Tunnel name
data (list): List of current SAs from vici
Returns:
str: `up` if child SA state is 'installed' otherwise `down`
"""
child_sa = 'down'
if not data:
return child_sa
for sa in data:
# check if parent SA exist
if connection_name not in sa.keys():
continue
child_sas = sa[connection_name]['child-sas']
# Get all child SA states
# there can be multiple SAs per tunnel
child_sa_states = [
v['state'] for k, v in child_sas.items() if
v['name'] == tunnel_name
]
return 'up' if 'INSTALLED' in child_sa_states else child_sa
def _get_child_sa_info(connection_name: str, tunnel_name: str,
data: list) -> dict:
"""Get child SA installed info by connection and tunnel name
Args:
connection_name (str): Connection name
tunnel_name (str): Tunnel name
data (list): List of current SAs from vici
Returns:
dict: Info of the child SA in the dictionary format
"""
for sa in data:
# check if parent SA exist
if connection_name not in sa.keys():
continue
child_sas = sa[connection_name]['child-sas']
# Get all child SA data
# Skip temp SA name (first key), get only SA values as dict
# {'OFFICE-B-tunnel-0-46': {'name': 'OFFICE-B-tunnel-0'}...}
# i.e get all data after 'OFFICE-B-tunnel-0-46'
child_sa_info = [
v for k, v in child_sas.items() if 'name' in v and
v['name'] == tunnel_name and v['state'] == 'INSTALLED'
]
return child_sa_info[-1] if child_sa_info else {}
def _get_child_sa_proposal(child_sa_data: dict) -> dict:
if child_sa_data and 'encr-alg' in child_sa_data:
encr_alg = child_sa_data.get('encr-alg')
cipher = encr_alg.split('_')[0]
mode = encr_alg.split('_')[1]
key_size = child_sa_data.get('encr-keysize')
integ_alg = child_sa_data.get('integ-alg')
dh_group = child_sa_data.get('dh-group')
proposal = {
'cipher': cipher,
'mode': mode,
'key_size': key_size,
'hash': integ_alg,
'dh': dh_group
}
return proposal
return {}
def _get_raw_data_connections(list_connections: list, list_sas: list) -> list:
"""Get configured VPN IKE connections and IPsec states
Args:
list_connections (list): List of configured connections from vici
list_sas (list): List of current SAs from vici
Returns:
list: List and status of IKE/IPsec connections/tunnels
"""
base_dict = []
for connections in list_connections:
base_list = {}
for connection, conn_conf in connections.items():
base_list['ike_connection_name'] = connection
base_list['ike_connection_state'] = _get_parent_sa_state(
connection, list_sas)
base_list['ike_remote_address'] = conn_conf['remote_addrs']
base_list['ike_proposal'] = _get_parent_sa_proposal(
connection, list_sas)
base_list['local_id'] = conn_conf.get('local-1', '').get('id')
base_list['remote_id'] = conn_conf.get('remote-1', '').get('id')
base_list['version'] = conn_conf.get('version', 'IKE')
base_list['children'] = []
children = conn_conf['children']
for tunnel, tun_options in children.items():
state = _get_child_sa_state(connection, tunnel, list_sas)
local_ts = tun_options.get('local-ts')
remote_ts = tun_options.get('remote-ts')
dpd_action = tun_options.get('dpd_action')
close_action = tun_options.get('close_action')
sa_info = _get_child_sa_info(connection, tunnel, list_sas)
esp_proposal = _get_child_sa_proposal(sa_info)
base_list['children'].append({
'name': tunnel,
'state': state,
'local_ts': local_ts,
'remote_ts': remote_ts,
'dpd_action': dpd_action,
'close_action': close_action,
'sa': sa_info,
'esp_proposal': esp_proposal
})
base_dict.append(base_list)
return base_dict
def _get_raw_connections_summary(list_conn, list_sas):
import jmespath
data = _get_raw_data_connections(list_conn, list_sas)
match = '[*].children[]'
child = jmespath.search(match, data)
tunnels_down = len([k for k in child if k['state'] == 'down'])
tunnels_up = len([k for k in child if k['state'] == 'up'])
tun_dict = {
'tunnels': child,
'total': len(child),
'down': tunnels_down,
'up': tunnels_up
}
return tun_dict
def _get_formatted_output_conections(data):
from tabulate import tabulate
data_entries = ''
connections = []
for entry in data:
tunnels = []
ike_name = entry['ike_connection_name']
ike_state = entry['ike_connection_state']
conn_type = entry.get('version', 'IKE')
remote_addrs = ','.join(entry['ike_remote_address'])
local_ts, remote_ts = '-', '-'
local_id = entry['local_id']
remote_id = entry['remote_id']
proposal = '-'
if entry.get('ike_proposal'):
proposal = (f'{entry["ike_proposal"]["cipher"]}_'
f'{entry["ike_proposal"]["mode"]}/'
f'{entry["ike_proposal"]["key_size"]}/'
f'{entry["ike_proposal"]["hash"]}/'
f'{entry["ike_proposal"]["dh"]}')
connections.append([
ike_name, ike_state, conn_type, remote_addrs, local_ts, remote_ts,
local_id, remote_id, proposal
])
for tun in entry['children']:
tun_name = tun.get('name')
tun_state = tun.get('state')
conn_type = 'IPsec'
local_ts = '\n'.join(tun.get('local_ts'))
remote_ts = '\n'.join(tun.get('remote_ts'))
proposal = '-'
if tun.get('esp_proposal'):
proposal = (f'{tun["esp_proposal"]["cipher"]}_'
f'{tun["esp_proposal"]["mode"]}/'
f'{tun["esp_proposal"]["key_size"]}/'
f'{tun["esp_proposal"]["hash"]}/'
f'{tun["esp_proposal"]["dh"]}')
connections.append([
tun_name, tun_state, conn_type, remote_addrs, local_ts,
remote_ts, local_id, remote_id, proposal
])
connection_headers = [
'Connection', 'State', 'Type', 'Remote address', 'Local TS',
'Remote TS', 'Local id', 'Remote id', 'Proposal'
]
output = tabulate(connections, connection_headers, numalign='left')
return output
# Connections block end
def _get_childsa_id_list(ike_sas: list) -> list:
"""
Generate list of CHILD SA ids based on list of OrderingDict
wich is returned by vici
:param ike_sas: list of IKE SAs generated by vici
:type ike_sas: list
:return: list of IKE SAs ids
:rtype: list
"""
list_childsa_id: list = []
for ike in ike_sas:
for ike_sa in ike.values():
for child_sa in ike_sa['child-sas'].values():
list_childsa_id.append(child_sa['uniqueid'].decode('ascii'))
return list_childsa_id
def _get_all_sitetosite_peers_name_list() -> list:
"""
Return site-to-site peers configuration
:return: site-to-site peers configuration
:rtype: list
"""
conf: ConfigTreeQuery = ConfigTreeQuery()
config_path = ['vpn', 'ipsec', 'site-to-site', 'peer']
peers_config = conf.get_config_dict(config_path, key_mangling=('-', '_'),
get_first_key=True,
no_tag_node_value_mangle=True)
peers_list: list = []
for name in peers_config:
peers_list.append(name)
return peers_list
def reset_peer(peer: str, tunnel: typing.Optional[str] = None):
# Convert tunnel to Strongwan format of CHILD_SA
tunnel_sw = None
if tunnel:
if tunnel.isnumeric():
tunnel_sw = f'{peer}-tunnel-{tunnel}'
elif tunnel == 'vti':
tunnel_sw = f'{peer}-vti'
try:
sa_list: list = vyos.ipsec.get_vici_sas_by_name(peer, tunnel_sw)
if not sa_list:
raise vyos.opmode.IncorrectValue(
f'Peer\'s {peer} SA(s) not found, aborting')
if tunnel and sa_list:
childsa_id_list: list = _get_childsa_id_list(sa_list)
if not childsa_id_list:
raise vyos.opmode.IncorrectValue(
f'Peer {peer} tunnel {tunnel} SA(s) not found, aborting')
vyos.ipsec.terminate_vici_by_name(peer, tunnel_sw)
print(f'Peer {peer} reset result: success')
except (vyos.ipsec.ViciInitiateError) as err:
raise vyos.opmode.UnconfiguredSubsystem(err)
except (vyos.ipsec.ViciCommandError) as err:
raise vyos.opmode.IncorrectValue(err)
def reset_all_peers():
sitetosite_list = _get_all_sitetosite_peers_name_list()
if sitetosite_list:
for peer_name in sitetosite_list:
try:
reset_peer(peer_name)
except (vyos.opmode.IncorrectValue) as err:
print(err)
print('Peers reset result: success')
else:
raise vyos.opmode.UnconfiguredSubsystem(
'VPN IPSec site-to-site is not configured, aborting')
def _get_ra_session_list_by_username(username: typing.Optional[str] = None):
"""
Return list of remote-access IKE_SAs uniqueids
:param username:
:type username:
:return:
:rtype:
"""
list_sa_id = []
sa_list = _get_raw_data_sas()
for sa_val in sa_list:
for sa in sa_val.values():
if 'remote-eap-id' in sa:
if username:
if username == sa['remote-eap-id']:
list_sa_id.append(sa['uniqueid'])
else:
list_sa_id.append(sa['uniqueid'])
return list_sa_id
def reset_ra(username: typing.Optional[str] = None):
#Reset remote-access ipsec sessions
if username:
list_sa_id = _get_ra_session_list_by_username(username)
else:
list_sa_id = _get_ra_session_list_by_username()
if list_sa_id:
vyos.ipsec.terminate_vici_ikeid_list(list_sa_id)
def reset_profile_dst(profile: str, tunnel: str, nbma_dst: str):
if profile and tunnel and nbma_dst:
ike_sa_name = f'dmvpn-{profile}-{tunnel}'
try:
# Get IKE SAs
sa_list = convert_data(
vyos.ipsec.get_vici_sas_by_name(ike_sa_name, None))
if not sa_list:
raise vyos.opmode.IncorrectValue(
f'SA(s) for profile {profile} tunnel {tunnel} not found, aborting')
sa_nbma_list = list([x for x in sa_list if
ike_sa_name in x and x[ike_sa_name][
'remote-host'] == nbma_dst])
if not sa_nbma_list:
raise vyos.opmode.IncorrectValue(
f'SA(s) for profile {profile} tunnel {tunnel} remote-host {nbma_dst} not found, aborting')
# terminate IKE SAs
vyos.ipsec.terminate_vici_ikeid_list(list(
[x[ike_sa_name]['uniqueid'] for x in sa_nbma_list if
ike_sa_name in x]))
# initiate IKE SAs
for ike in sa_nbma_list:
if ike_sa_name in ike:
vyos.ipsec.vici_initiate(ike_sa_name, 'dmvpn',
ike[ike_sa_name]['local-host'],
ike[ike_sa_name]['remote-host'])
print(
f'Profile {profile} tunnel {tunnel} remote-host {nbma_dst} reset result: success')
except (vyos.ipsec.ViciInitiateError) as err:
raise vyos.opmode.UnconfiguredSubsystem(err)
except (vyos.ipsec.ViciCommandError) as err:
raise vyos.opmode.IncorrectValue(err)
def reset_profile_all(profile: str, tunnel: str):
if profile and tunnel:
ike_sa_name = f'dmvpn-{profile}-{tunnel}'
try:
# Get IKE SAs
sa_list: list = convert_data(
vyos.ipsec.get_vici_sas_by_name(ike_sa_name, None))
if not sa_list:
raise vyos.opmode.IncorrectValue(
f'SA(s) for profile {profile} tunnel {tunnel} not found, aborting')
# terminate IKE SAs
vyos.ipsec.terminate_vici_by_name(ike_sa_name, None)
# initiate IKE SAs
for ike in sa_list:
if ike_sa_name in ike:
vyos.ipsec.vici_initiate(ike_sa_name, 'dmvpn',
ike[ike_sa_name]['local-host'],
ike[ike_sa_name]['remote-host'])
print(
f'Profile {profile} tunnel {tunnel} remote-host {ike[ike_sa_name]["remote-host"]} reset result: success')
print(f'Profile {profile} tunnel {tunnel} reset result: success')
except (vyos.ipsec.ViciInitiateError) as err:
raise vyos.opmode.UnconfiguredSubsystem(err)
except (vyos.ipsec.ViciCommandError) as err:
raise vyos.opmode.IncorrectValue(err)
def show_sa(raw: bool):
sa_data = _get_raw_data_sas()
if raw:
return sa_data
return _get_formatted_output_sas(sa_data)
def _get_output_sas_detail(ra_output_list: list) -> str:
"""
Formate all IKE SAs detail output
:param ra_output_list: IKE SAs list
:type ra_output_list: list
:return: formatted RA IKE SAs detail output
:rtype: str
"""
return _get_output_swanctl_sas_from_list(ra_output_list)
def show_sa_detail(raw: bool):
sa_data = _get_raw_data_sas()
if raw:
return sa_data
return _get_output_sas_detail(sa_data)
def show_connections(raw: bool):
list_conns = _get_convert_data_connections()
list_sas = _get_raw_data_sas()
if raw:
return _get_raw_data_connections(list_conns, list_sas)
connections = _get_raw_data_connections(list_conns, list_sas)
return _get_formatted_output_conections(connections)
def show_connections_summary(raw: bool):
list_conns = _get_convert_data_connections()
list_sas = _get_raw_data_sas()
if raw:
return _get_raw_connections_summary(list_conns, list_sas)
def _get_ra_sessions(username: typing.Optional[str] = None) -> list:
"""
Return list of remote-access IKE_SAs from VICI by username.
If username unspecified, return all remote-access IKE_SAs
:param username: Username of RA connection
:type username: str
:return: list of ra remote-access IKE_SAs
:rtype: list
"""
list_sa = []
sa_list = _get_raw_data_sas()
for conn in sa_list:
for sa in conn.values():
if 'remote-eap-id' in sa:
if username:
if username == sa['remote-eap-id']:
list_sa.append(conn)
else:
list_sa.append(conn)
return list_sa
def _filter_ikesas(list_sa: list, filter_key: str, filter_value: str) -> list:
"""
Filter IKE SAs by specifice key
:param list_sa: list of IKE SAs
:type list_sa: list
:param filter_key: Filter Key
:type filter_key: str
:param filter_value: Filter Value
:type filter_value: str
:return: Filtered list of IKE SAs
:rtype: list
"""
filtered_sa_list = []
for conn in list_sa:
for sa in conn.values():
if sa[filter_key] and sa[filter_key] == filter_value:
filtered_sa_list.append(conn)
return filtered_sa_list
def _get_last_installed_childsa(sa: dict) -> str:
"""
Return name of last installed active Child SA
:param sa: Dictionary with Child SAs
:type sa: dict
:return: Name of the Last installed active Child SA
:rtype: str
"""
child_sa_name = None
child_sa_id = 0
for sa_name, child_sa in sa['child-sas'].items():
if child_sa['state'] == 'INSTALLED':
if child_sa_id == 0 or int(child_sa['uniqueid']) > child_sa_id:
child_sa_id = int(child_sa['uniqueid'])
child_sa_name = sa_name
return child_sa_name
def _get_formatted_ike_proposal(sa: dict) -> str:
"""
Return IKE proposal string in format
EncrALG-EncrKeySize/PFR/HASH/DH-GROUP
:param sa: IKE SA
:type sa: dict
:return: IKE proposal string
:rtype: str
"""
proposal = ''
proposal = f'{proposal}{sa["encr-alg"]}' if 'encr-alg' in sa else proposal
proposal = f'{proposal}-{sa["encr-keysize"]}' if 'encr-keysize' in sa else proposal
proposal = f'{proposal}/{sa["prf-alg"]}' if 'prf-alg' in sa else proposal
proposal = f'{proposal}/{sa["integ-alg"]}' if 'integ-alg' in sa else proposal
proposal = f'{proposal}/{sa["dh-group"]}' if 'dh-group' in sa else proposal
return proposal
def _get_formatted_ipsec_proposal(sa: dict) -> str:
"""
Return IPSec proposal string in format
Protocol: EncrALG-EncrKeySize/HASH/PFS
:param sa: Child SA
:type sa: dict
:return: IPSec proposal string
:rtype: str
"""
proposal = ''
proposal = f'{proposal}{sa["protocol"]}' if 'protocol' in sa else proposal
proposal = f'{proposal}:{sa["encr-alg"]}' if 'encr-alg' in sa else proposal
proposal = f'{proposal}-{sa["encr-keysize"]}' if 'encr-keysize' in sa else proposal
proposal = f'{proposal}/{sa["integ-alg"]}' if 'integ-alg' in sa else proposal
proposal = f'{proposal}/{sa["dh-group"]}' if 'dh-group' in sa else proposal
return proposal
def _get_output_ra_sas_detail(ra_output_list: list) -> str:
"""
Formate RA IKE SAs detail output
:param ra_output_list: IKE SAs list
:type ra_output_list: list
:return: formatted RA IKE SAs detail output
:rtype: str
"""
return _get_output_swanctl_sas_from_list(ra_output_list)
def _get_formatted_output_ra_summary(ra_output_list: list):
sa_data = []
for conn in ra_output_list:
for sa in conn.values():
sa_id = sa['uniqueid'] if 'uniqueid' in sa else ''
sa_username = sa['remote-eap-id'] if 'remote-eap-id' in sa else ''
sa_protocol = f'IKEv{sa["version"]}' if 'version' in sa else ''
sa_remotehost = sa['remote-host'] if 'remote-host' in sa else ''
sa_remoteid = sa['remote-id'] if 'remote-id' in sa else ''
sa_ike_proposal = _get_formatted_ike_proposal(sa)
sa_tunnel_ip = sa['remote-vips'][0]
child_sa_key = _get_last_installed_childsa(sa)
if child_sa_key:
child_sa = sa['child-sas'][child_sa_key]
sa_ipsec_proposal = _get_formatted_ipsec_proposal(child_sa)
sa_state = "UP"
sa_uptime = seconds_to_human(sa['established'])
else:
sa_ipsec_proposal = ''
sa_state = "DOWN"
sa_uptime = ''
sa_data.append(
[sa_id, sa_username, sa_protocol, sa_state, sa_uptime,
sa_tunnel_ip,
sa_remotehost, sa_remoteid, sa_ike_proposal,
sa_ipsec_proposal])
headers = ["Connection ID", "Username", "Protocol", "State", "Uptime",
"Tunnel IP", "Remote Host", "Remote ID", "IKE Proposal",
"IPSec Proposal"]
sa_data = sorted(sa_data, key=_alphanum_key)
output = tabulate(sa_data, headers)
return output
def show_ra_detail(raw: bool, username: typing.Optional[str] = None,
conn_id: typing.Optional[str] = None):
list_sa: list = _get_ra_sessions()
if username:
list_sa = _filter_ikesas(list_sa, 'remote-eap-id', username)
elif conn_id:
list_sa = _filter_ikesas(list_sa, 'uniqueid', conn_id)
if not list_sa:
raise vyos.opmode.IncorrectValue(
f'No active connections found, aborting')
if raw:
return list_sa
return _get_output_ra_sas_detail(list_sa)
def show_ra_summary(raw: bool):
list_sa: list = _get_ra_sessions()
if not list_sa:
raise vyos.opmode.IncorrectValue(
f'No active connections found, aborting')
if raw:
return list_sa
return _get_formatted_output_ra_summary(list_sa)
+# PSK block
+def _get_raw_psk():
+ conf: ConfigTreeQuery = ConfigTreeQuery()
+ config_path = ['vpn', 'ipsec', 'authentication', 'psk']
+ psk_config = conf.get_config_dict(config_path, key_mangling=('-', '_'),
+ get_first_key=True,
+ no_tag_node_value_mangle=True)
+
+ psk_list = []
+ for psk, psk_data in psk_config.items():
+ psk_data['psk'] = psk
+ psk_list.append(psk_data)
+
+ return psk_list
+
+
+def _get_formatted_psk(psk_list):
+ headers = ["PSK", "Id", "Secret"]
+ formatted_data = []
+
+ for psk_data in psk_list:
+ formatted_data.append([psk_data["psk"], "\n".join(psk_data["id"]), psk_data["secret"]])
+
+ return tabulate(formatted_data, headers=headers)
+
+
+def show_psk(raw: bool):
+ config = ConfigTreeQuery()
+ if not config.exists('vpn ipsec authentication psk'):
+ raise vyos.opmode.UnconfiguredSubsystem('VPN ipsec psk authentication is not configured')
+
+ psk = _get_raw_psk()
+ if raw:
+ return psk
+ return _get_formatted_psk(psk)
+
+# PSK block end
+
+
if __name__ == '__main__':
try:
res = vyos.opmode.run(sys.modules[__name__])
if res:
print(res)
except (ValueError, vyos.opmode.Error) as e:
print(e)
sys.exit(1)