diff --git a/data/templates/ipsec/ios_profile.j2 b/data/templates/ipsec/ios_profile.j2
index a9ae1c7a9..935acbf8e 100644
--- a/data/templates/ipsec/ios_profile.j2
+++ b/data/templates/ipsec/ios_profile.j2
@@ -1,109 +1,107 @@
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<!-- Set the name to whatever you like, it is used in the profile list on the device -->
<key>PayloadDisplayName</key>
<string>{{ profile_name }}</string>
<!-- This is a reverse-DNS style unique identifier used to detect duplicate profiles -->
<key>PayloadIdentifier</key>
<string>{{ rfqdn }}</string>
<!-- A globally unique identifier, use uuidgen on Linux/Mac OS X to generate it -->
<key>PayloadUUID</key>
<string>{{ '' | get_uuid }}</string>
<key>PayloadType</key>
<string>Configuration</string>
<key>PayloadVersion</key>
<integer>1</integer>
<key>PayloadContent</key>
<array>
<!-- It is possible to add multiple VPN payloads with different identifiers/UUIDs and names -->
<dict>
<!-- This is an extension of the identifier given above -->
<key>PayloadIdentifier</key>
<string>{{ rfqdn }}.conf1</string>
<!-- A globally unique identifier for this payload -->
<key>PayloadUUID</key>
<string>{{ '' | get_uuid }}</string>
<key>PayloadType</key>
<string>com.apple.vpn.managed</string>
<key>PayloadVersion</key>
<integer>1</integer>
<!-- This is the name of the VPN connection as seen in the VPN application later -->
<key>UserDefinedName</key>
<string>{{ vpn_name }}</string>
<key>VPNType</key>
<string>IKEv2</string>
<key>IKEv2</key>
<dict>
<!-- Hostname or IP address of the VPN server -->
<key>RemoteAddress</key>
<string>{{ remote }}</string>
<!-- Remote identity, can be a FQDN, a userFQDN, an IP or (theoretically) a certificate's subject DN. Can't be empty.
IMPORTANT: DNs are currently not handled correctly, they are always sent as identities of type FQDN -->
<key>RemoteIdentifier</key>
<string>{{ authentication.local_id if authentication.local_id is vyos_defined else 'VyOS' }}</string>
<!-- Local IKE identity, same restrictions as above. If it is empty the client's IP address will be used -->
<key>LocalIdentifier</key>
<string></string>
<!-- Optional, if it matches the CN of the root CA certificate (not the full subject DN) a certificate request will be sent
NOTE: If this is not configured make sure to configure leftsendcert=always on the server, otherwise it won't send its certificate -->
<key>ServerCertificateIssuerCommonName</key>
- <string>{{ ca_cn }}</string>
+ <string>{{ ca_common_name }}</string>
<!-- Optional, the CN or one of the subjectAltNames of the server certificate to verify it, if not set RemoteIdentifier will be used -->
<key>ServerCertificateCommonName</key>
- <string>{{ cert_cn }}</string>
+ <string>{{ cert_common_name }}</string>
<!-- The server is authenticated using a certificate -->
<key>AuthenticationMethod</key>
<string>Certificate</string>
<!-- The client uses EAP to authenticate -->
<key>ExtendedAuthEnabled</key>
<integer>1</integer>
<!-- The next two dictionaries are optional (as are the keys in them), but it is recommended to specify them as the default is to use 3DES.
IMPORTANT: Because only one proposal is sent (even if nothing is configured here) it must match the server configuration -->
<key>IKESecurityAssociationParameters</key>
<dict>
<!-- @see https://developer.apple.com/documentation/networkextension/nevpnikev2encryptionalgorithm -->
<key>EncryptionAlgorithm</key>
<string>{{ ike_encryption.encryption }}</string>
<!-- @see https://developer.apple.com/documentation/networkextension/nevpnikev2integrityalgorithm -->
<key>IntegrityAlgorithm</key>
<string>{{ ike_encryption.hash }}</string>
<!-- @see https://developer.apple.com/documentation/networkextension/nevpnikev2diffiehellmangroup -->
<key>DiffieHellmanGroup</key>
<integer>{{ ike_encryption.dh_group }}</integer>
</dict>
<key>ChildSecurityAssociationParameters</key>
<dict>
<key>EncryptionAlgorithm</key>
<string>{{ esp_encryption.encryption }}</string>
<key>IntegrityAlgorithm</key>
<string>{{ esp_encryption.hash }}</string>
<key>DiffieHellmanGroup</key>
<integer>{{ ike_encryption.dh_group }}</integer>
</dict>
</dict>
</dict>
-{% if certs is vyos_defined %}
+{% if ca_certificates is vyos_defined %}
<!-- This payload is optional but it provides an easy way to install the CA certificate together with the configuration -->
-{% for cert in certs %}
- <!-- Payload for: {{ cert.ca_cn }} -->
+{% for ca in ca_certificates %}
+ <!-- Payload for: {{ ca.ca_name }} -->
<dict>
<key>PayloadIdentifier</key>
- <string>org.{{ cert.ca_cn | lower | replace(' ', '.') | replace('_', '.') }}</string>
+ <string>org.{{ ca.ca_name | lower | replace(' ', '.') | replace('_', '.') }}</string>
<key>PayloadUUID</key>
- <string>{{ cert.ca_cn | generate_uuid4 }}</string>
+ <string>{{ ca.ca_name | get_uuid }}</string>
<key>PayloadType</key>
<string>com.apple.security.root</string>
<key>PayloadVersion</key>
<integer>1</integer>
<!-- This is the Base64 (PEM) encoded CA certificate -->
<key>PayloadContent</key>
- <data>
- {{ cert.ca_cert }}
- </data>
+ <data>{{ ca.ca_chain }}</data>
</dict>
{% endfor %}
{% endif %}
</array>
</dict>
</plist>
diff --git a/python/vyos/template.py b/python/vyos/template.py
index cdcbdc6c3..a4fff649c 100644
--- a/python/vyos/template.py
+++ b/python/vyos/template.py
@@ -1,848 +1,855 @@
# Copyright 2019-2023 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library. If not, see <http://www.gnu.org/licenses/>.
import functools
import os
from jinja2 import Environment
from jinja2 import FileSystemLoader
from jinja2 import ChainableUndefined
from vyos.defaults import directories
from vyos.utils.dict import dict_search_args
from vyos.utils.file import makedir
from vyos.utils.permission import chmod
from vyos.utils.permission import chown
# We use a mutable global variable for the default template directory
# to make it possible to call scripts from this repository
# outside of live VyOS systems.
# If something (like the image build scripts)
# want to call a script, they can modify the default location
# to the repository path.
DEFAULT_TEMPLATE_DIR = directories["templates"]
# Holds template filters registered via register_filter()
_FILTERS = {}
_TESTS = {}
# reuse Environments with identical settings to improve performance
@functools.lru_cache(maxsize=2)
def _get_environment(location=None):
if location is None:
loc_loader=FileSystemLoader(DEFAULT_TEMPLATE_DIR)
else:
loc_loader=FileSystemLoader(location)
env = Environment(
# Don't check if template files were modified upon re-rendering
auto_reload=False,
# Cache up to this number of templates for quick re-rendering
cache_size=100,
loader=loc_loader,
trim_blocks=True,
undefined=ChainableUndefined,
extensions=['jinja2.ext.loopcontrols']
)
env.filters.update(_FILTERS)
env.tests.update(_TESTS)
return env
def register_filter(name, func=None):
"""Register a function to be available as filter in templates under given name.
It can also be used as a decorator, see below in this module for examples.
:raise RuntimeError:
when trying to register a filter after a template has been rendered already
:raise ValueError: when trying to register a name which was taken already
"""
if func is None:
return functools.partial(register_filter, name)
if _get_environment.cache_info().currsize:
raise RuntimeError(
"Filters can only be registered before rendering the first template"
)
if name in _FILTERS:
raise ValueError(f"A filter with name {name!r} was registered already")
_FILTERS[name] = func
return func
def register_test(name, func=None):
"""Register a function to be available as test in templates under given name.
It can also be used as a decorator, see below in this module for examples.
:raise RuntimeError:
when trying to register a test after a template has been rendered already
:raise ValueError: when trying to register a name which was taken already
"""
if func is None:
return functools.partial(register_test, name)
if _get_environment.cache_info().currsize:
raise RuntimeError(
"Tests can only be registered before rendering the first template"
)
if name in _TESTS:
raise ValueError(f"A test with name {name!r} was registered already")
_TESTS[name] = func
return func
def render_to_string(template, content, formater=None, location=None):
"""Render a template from the template directory, raise on any errors.
:param template: the path to the template relative to the template folder
:param content: the dictionary of variables to put into rendering context
:param formater:
if given, it has to be a callable the rendered string is passed through
The parsed template files are cached, so rendering the same file multiple times
does not cause as too much overhead.
If used everywhere, it could be changed to load the template from Python
environment variables from an importable Python module generated when the Debian
package is build (recovering the load time and overhead caused by having the
file out of the code).
"""
template = _get_environment(location).get_template(template)
rendered = template.render(content)
if formater is not None:
rendered = formater(rendered)
return rendered
def render(
destination,
template,
content,
formater=None,
permission=None,
user=None,
group=None,
location=None,
):
"""Render a template from the template directory to a file, raise on any errors.
:param destination: path to the file to save the rendered template in
:param permission: permission bitmask to set for the output file
:param user: user to own the output file
:param group: group to own the output file
All other parameters are as for :func:`render_to_string`.
"""
# Create the directory if it does not exist
folder = os.path.dirname(destination)
makedir(folder, user, group)
# As we are opening the file with 'w', we are performing the rendering before
# calling open() to not accidentally erase the file if rendering fails
rendered = render_to_string(template, content, formater, location)
# Write to file
with open(destination, "w") as file:
chmod(file.fileno(), permission)
chown(file.fileno(), user, group)
file.write(rendered)
##################################
# Custom template filters follow #
##################################
@register_filter('force_to_list')
def force_to_list(value):
""" Convert scalars to single-item lists and leave lists untouched """
if isinstance(value, list):
return value
else:
return [value]
@register_filter('seconds_to_human')
def seconds_to_human(seconds, separator=""):
""" Convert seconds to human-readable values like 1d6h15m23s """
from vyos.utils.convert import seconds_to_human
return seconds_to_human(seconds, separator=separator)
@register_filter('bytes_to_human')
def bytes_to_human(bytes, initial_exponent=0, precision=2):
""" Convert bytes to human-readable values like 1.44M """
from vyos.utils.convert import bytes_to_human
return bytes_to_human(bytes, initial_exponent=initial_exponent, precision=precision)
@register_filter('human_to_bytes')
def human_to_bytes(value):
""" Convert a data amount with a unit suffix to bytes, like 2K to 2048 """
from vyos.utils.convert import human_to_bytes
return human_to_bytes(value)
@register_filter('ip_from_cidr')
def ip_from_cidr(prefix):
""" Take an IPv4/IPv6 CIDR host and strip cidr mask.
Example:
192.0.2.1/24 -> 192.0.2.1, 2001:db8::1/64 -> 2001:db8::1
"""
from ipaddress import ip_interface
return str(ip_interface(prefix).ip)
@register_filter('address_from_cidr')
def address_from_cidr(prefix):
""" Take an IPv4/IPv6 CIDR prefix and convert the network to an "address".
Example:
192.0.2.0/24 -> 192.0.2.0, 2001:db8::/48 -> 2001:db8::
"""
from ipaddress import ip_network
return str(ip_network(prefix).network_address)
@register_filter('bracketize_ipv6')
def bracketize_ipv6(address):
""" Place a passed IPv6 address into [] brackets, do nothing for IPv4 """
if is_ipv6(address):
return f'[{address}]'
return address
@register_filter('dot_colon_to_dash')
def dot_colon_to_dash(text):
""" Replace dot and colon to dash for string
Example:
192.0.2.1 => 192-0-2-1, 2001:db8::1 => 2001-db8--1
"""
text = text.replace(":", "-")
text = text.replace(".", "-")
return text
@register_filter('generate_uuid4')
def generate_uuid4(text):
""" Generate random unique ID
Example:
% uuid4()
UUID('958ddf6a-ef14-4e81-8cfb-afb12456d1c5')
"""
from uuid import uuid4
return uuid4()
@register_filter('netmask_from_cidr')
def netmask_from_cidr(prefix):
""" Take CIDR prefix and convert the prefix length to a "subnet mask".
Example:
- 192.0.2.0/24 -> 255.255.255.0
- 2001:db8::/48 -> ffff:ffff:ffff::
"""
from ipaddress import ip_network
return str(ip_network(prefix).netmask)
@register_filter('netmask_from_ipv4')
def netmask_from_ipv4(address):
""" Take IP address and search all attached interface IP addresses for the
given one. After address has been found, return the associated netmask.
Example:
- 172.18.201.10 -> 255.255.255.128
"""
from netifaces import interfaces
from netifaces import ifaddresses
from netifaces import AF_INET
for interface in interfaces():
tmp = ifaddresses(interface)
if AF_INET in tmp:
for af_addr in tmp[AF_INET]:
if 'addr' in af_addr:
if af_addr['addr'] == address:
return af_addr['netmask']
raise ValueError
@register_filter('is_ip_network')
def is_ip_network(addr):
""" Take IP(v4/v6) address and validate if the passed argument is a network
or a host address.
Example:
- 192.0.2.0 -> False
- 192.0.2.10/24 -> False
- 192.0.2.0/24 -> True
- 2001:db8:: -> False
- 2001:db8::100 -> False
- 2001:db8::/48 -> True
- 2001:db8:1000::/64 -> True
"""
try:
from ipaddress import ip_network
# input variables must contain a / to indicate its CIDR notation
if len(addr.split('/')) != 2:
raise ValueError()
ip_network(addr)
return True
except:
return False
@register_filter('network_from_ipv4')
def network_from_ipv4(address):
""" Take IP address and search all attached interface IP addresses for the
given one. After address has been found, return the associated network
address.
Example:
- 172.18.201.10 has mask 255.255.255.128 -> network is 172.18.201.0
"""
netmask = netmask_from_ipv4(address)
from ipaddress import ip_interface
cidr_prefix = ip_interface(f'{address}/{netmask}').network
return address_from_cidr(cidr_prefix)
@register_filter('is_interface')
def is_interface(interface):
""" Check if parameter is a valid local interface name """
from vyos.utils.network import interface_exists
return interface_exists(interface)
@register_filter('is_ip')
def is_ip(addr):
""" Check addr if it is an IPv4 or IPv6 address """
return is_ipv4(addr) or is_ipv6(addr)
@register_filter('is_ipv4')
def is_ipv4(text):
""" Filter IP address, return True on IPv4 address, False otherwise """
from ipaddress import ip_interface
try: return ip_interface(text).version == 4
except: return False
@register_filter('is_ipv6')
def is_ipv6(text):
""" Filter IP address, return True on IPv6 address, False otherwise """
from ipaddress import ip_interface
try: return ip_interface(text).version == 6
except: return False
@register_filter('first_host_address')
def first_host_address(prefix):
""" Return first usable (host) IP address from given prefix.
Example:
- 10.0.0.0/24 -> 10.0.0.1
- 2001:db8::/64 -> 2001:db8::
"""
from ipaddress import ip_interface
tmp = ip_interface(prefix).network
return str(tmp.network_address +1)
@register_filter('last_host_address')
def last_host_address(text):
""" Return first usable IP address from given prefix.
Example:
- 10.0.0.0/24 -> 10.0.0.254
- 2001:db8::/64 -> 2001:db8::ffff:ffff:ffff:ffff
"""
from ipaddress import ip_interface
from ipaddress import IPv4Network
from ipaddress import IPv6Network
addr = ip_interface(text)
if addr.version == 4:
return str(IPv4Network(addr).broadcast_address - 1)
return str(IPv6Network(addr).broadcast_address)
@register_filter('inc_ip')
def inc_ip(address, increment):
""" Increment given IP address by 'increment'
Example (inc by 2):
- 10.0.0.0/24 -> 10.0.0.2
- 2001:db8::/64 -> 2001:db8::2
"""
from ipaddress import ip_interface
return str(ip_interface(address).ip + int(increment))
@register_filter('dec_ip')
def dec_ip(address, decrement):
""" Decrement given IP address by 'decrement'
Example (inc by 2):
- 10.0.0.0/24 -> 10.0.0.2
- 2001:db8::/64 -> 2001:db8::2
"""
from ipaddress import ip_interface
return str(ip_interface(address).ip - int(decrement))
@register_filter('compare_netmask')
def compare_netmask(netmask1, netmask2):
"""
Compare two IP netmask if they have the exact same size.
compare_netmask('10.0.0.0/8', '20.0.0.0/8') -> True
compare_netmask('10.0.0.0/8', '20.0.0.0/16') -> False
"""
from ipaddress import ip_network
try:
return ip_network(netmask1).netmask == ip_network(netmask2).netmask
except:
return False
@register_filter('isc_static_route')
def isc_static_route(subnet, router):
# https://ercpe.de/blog/pushing-static-routes-with-isc-dhcp-server
# Option format is:
# <netmask>, <network-byte1>, <network-byte2>, <network-byte3>, <router-byte1>, <router-byte2>, <router-byte3>
# where bytes with the value 0 are omitted.
from ipaddress import ip_network
net = ip_network(subnet)
# add netmask
string = str(net.prefixlen) + ','
# add network bytes
if net.prefixlen:
width = net.prefixlen // 8
if net.prefixlen % 8:
width += 1
string += ','.join(map(str,tuple(net.network_address.packed)[:width])) + ','
# add router bytes
string += ','.join(router.split('.'))
return string
@register_filter('is_file')
def is_file(filename):
if os.path.exists(filename):
return os.path.isfile(filename)
return False
@register_filter('get_dhcp_router')
def get_dhcp_router(interface):
""" Static routes can point to a router received by a DHCP reply. This
helper is used to get the current default router from the DHCP reply.
Returns False of no router is found, returns the IP address as string if
a router is found.
"""
lease_file = directories['isc_dhclient_dir'] + f'/dhclient_{interface}.leases'
if not os.path.exists(lease_file):
return None
from vyos.utils.file import read_file
for line in read_file(lease_file).splitlines():
if 'option routers' in line:
(_, _, address) = line.split()
return address.rstrip(';')
@register_filter('natural_sort')
def natural_sort(iterable):
import re
from jinja2.runtime import Undefined
if isinstance(iterable, Undefined) or iterable is None:
return list()
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))]
return sorted(iterable, key=alphanum_key)
@register_filter('get_ipv4')
def get_ipv4(interface):
""" Get interface IPv4 addresses"""
from vyos.ifconfig import Interface
return Interface(interface).get_addr_v4()
@register_filter('get_ipv6')
def get_ipv6(interface):
""" Get interface IPv6 addresses"""
from vyos.ifconfig import Interface
return Interface(interface).get_addr_v6()
@register_filter('get_ip')
def get_ip(interface):
""" Get interface IP addresses"""
from vyos.ifconfig import Interface
return Interface(interface).get_addr()
def get_first_ike_dh_group(ike_group):
if ike_group and 'proposal' in ike_group:
for priority, proposal in ike_group['proposal'].items():
if 'dh_group' in proposal:
return 'dh-group' + proposal['dh_group']
return 'dh-group2' # Fallback on dh-group2
@register_filter('get_esp_ike_cipher')
def get_esp_ike_cipher(group_config, ike_group=None):
pfs_lut = {
'dh-group1' : 'modp768',
'dh-group2' : 'modp1024',
'dh-group5' : 'modp1536',
'dh-group14' : 'modp2048',
'dh-group15' : 'modp3072',
'dh-group16' : 'modp4096',
'dh-group17' : 'modp6144',
'dh-group18' : 'modp8192',
'dh-group19' : 'ecp256',
'dh-group20' : 'ecp384',
'dh-group21' : 'ecp521',
'dh-group22' : 'modp1024s160',
'dh-group23' : 'modp2048s224',
'dh-group24' : 'modp2048s256',
'dh-group25' : 'ecp192',
'dh-group26' : 'ecp224',
'dh-group27' : 'ecp224bp',
'dh-group28' : 'ecp256bp',
'dh-group29' : 'ecp384bp',
'dh-group30' : 'ecp512bp',
'dh-group31' : 'curve25519',
'dh-group32' : 'curve448'
}
ciphers = []
if 'proposal' in group_config:
for priority, proposal in group_config['proposal'].items():
# both encryption and hash need to be specified for a proposal
if not {'encryption', 'hash'} <= set(proposal):
continue
tmp = '{encryption}-{hash}'.format(**proposal)
if 'prf' in proposal:
tmp += '-' + proposal['prf']
if 'dh_group' in proposal:
tmp += '-' + pfs_lut[ 'dh-group' + proposal['dh_group'] ]
elif 'pfs' in group_config and group_config['pfs'] != 'disable':
group = group_config['pfs']
if group_config['pfs'] == 'enable':
group = get_first_ike_dh_group(ike_group)
tmp += '-' + pfs_lut[group]
ciphers.append(tmp)
return ciphers
@register_filter('get_uuid')
-def get_uuid(interface):
+def get_uuid(seed):
""" Get interface IP addresses"""
- from uuid import uuid1
- return uuid1()
+ if seed:
+ from hashlib import md5
+ from uuid import UUID
+ tmp = md5()
+ tmp.update(seed.encode('utf-8'))
+ return str(UUID(tmp.hexdigest()))
+ else:
+ from uuid import uuid1
+ return uuid1()
openvpn_translate = {
'des': 'des-cbc',
'3des': 'des-ede3-cbc',
'bf128': 'bf-cbc',
'bf256': 'bf-cbc',
'aes128gcm': 'aes-128-gcm',
'aes128': 'aes-128-cbc',
'aes192gcm': 'aes-192-gcm',
'aes192': 'aes-192-cbc',
'aes256gcm': 'aes-256-gcm',
'aes256': 'aes-256-cbc'
}
@register_filter('openvpn_cipher')
def get_openvpn_cipher(cipher):
if cipher in openvpn_translate:
return openvpn_translate[cipher].upper()
return cipher.upper()
@register_filter('openvpn_ncp_ciphers')
def get_openvpn_ncp_ciphers(ciphers):
out = []
for cipher in ciphers:
if cipher in openvpn_translate:
out.append(openvpn_translate[cipher])
else:
out.append(cipher)
return ':'.join(out).upper()
@register_filter('snmp_auth_oid')
def snmp_auth_oid(type):
if type not in ['md5', 'sha', 'aes', 'des', 'none']:
raise ValueError()
OIDs = {
'md5' : '.1.3.6.1.6.3.10.1.1.2',
'sha' : '.1.3.6.1.6.3.10.1.1.3',
'aes' : '.1.3.6.1.6.3.10.1.2.4',
'des' : '.1.3.6.1.6.3.10.1.2.2',
'none': '.1.3.6.1.6.3.10.1.2.1'
}
return OIDs[type]
@register_filter('nft_action')
def nft_action(vyos_action):
if vyos_action == 'accept':
return 'return'
return vyos_action
@register_filter('nft_rule')
def nft_rule(rule_conf, fw_hook, fw_name, rule_id, ip_name='ip'):
from vyos.firewall import parse_rule
return parse_rule(rule_conf, fw_hook, fw_name, rule_id, ip_name)
@register_filter('nft_default_rule')
def nft_default_rule(fw_conf, fw_name, family):
output = ['counter']
default_action = fw_conf['default_action']
#family = 'ipv6' if ipv6 else 'ipv4'
if 'default_log' in fw_conf:
action_suffix = default_action[:1].upper()
output.append(f'log prefix "[{family}-{fw_name[:19]}-default-{action_suffix}]"')
#output.append(nft_action(default_action))
output.append(f'{default_action}')
if 'default_jump_target' in fw_conf:
target = fw_conf['default_jump_target']
def_suffix = '6' if family == 'ipv6' else ''
output.append(f'NAME{def_suffix}_{target}')
output.append(f'comment "{fw_name} default-action {default_action}"')
return " ".join(output)
@register_filter('nft_state_policy')
def nft_state_policy(conf, state):
out = [f'ct state {state}']
if 'log' in conf:
log_state = state[:3].upper()
log_action = (conf['action'] if 'action' in conf else 'accept')[:1].upper()
out.append(f'log prefix "[STATE-POLICY-{log_state}-{log_action}]"')
if 'log_level' in conf:
log_level = conf['log_level']
out.append(f'level {log_level}')
out.append('counter')
if 'action' in conf:
out.append(conf['action'])
return " ".join(out)
@register_filter('nft_intra_zone_action')
def nft_intra_zone_action(zone_conf, ipv6=False):
if 'intra_zone_filtering' in zone_conf:
intra_zone = zone_conf['intra_zone_filtering']
fw_name = 'ipv6_name' if ipv6 else 'name'
name_prefix = 'NAME6_' if ipv6 else 'NAME_'
if 'action' in intra_zone:
if intra_zone['action'] == 'accept':
return 'return'
return intra_zone['action']
elif dict_search_args(intra_zone, 'firewall', fw_name):
name = dict_search_args(intra_zone, 'firewall', fw_name)
return f'jump {name_prefix}{name}'
return 'return'
@register_filter('nft_nested_group')
def nft_nested_group(out_list, includes, groups, key):
if not vyos_defined(out_list):
out_list = []
def add_includes(name):
if key in groups[name]:
for item in groups[name][key]:
if item in out_list:
continue
out_list.append(item)
if 'include' in groups[name]:
for name_inc in groups[name]['include']:
add_includes(name_inc)
for name in includes:
add_includes(name)
return out_list
@register_filter('nat_rule')
def nat_rule(rule_conf, rule_id, nat_type, ipv6=False):
from vyos.nat import parse_nat_rule
return parse_nat_rule(rule_conf, rule_id, nat_type, ipv6)
@register_filter('nat_static_rule')
def nat_static_rule(rule_conf, rule_id, nat_type):
from vyos.nat import parse_nat_static_rule
return parse_nat_static_rule(rule_conf, rule_id, nat_type)
@register_filter('conntrack_rule')
def conntrack_rule(rule_conf, rule_id, action, ipv6=False):
ip_prefix = 'ip6' if ipv6 else 'ip'
def_suffix = '6' if ipv6 else ''
output = []
if 'inbound_interface' in rule_conf:
ifname = rule_conf['inbound_interface']
if ifname != 'any':
output.append(f'iifname {ifname}')
if 'protocol' in rule_conf:
if action != 'timeout':
proto = rule_conf['protocol']
else:
for protocol, protocol_config in rule_conf['protocol'].items():
proto = protocol
output.append(f'meta l4proto {proto}')
tcp_flags = dict_search_args(rule_conf, 'tcp', 'flags')
if tcp_flags and action != 'timeout':
from vyos.firewall import parse_tcp_flags
output.append(parse_tcp_flags(tcp_flags))
for side in ['source', 'destination']:
if side in rule_conf:
side_conf = rule_conf[side]
prefix = side[0]
if 'address' in side_conf:
address = side_conf['address']
operator = ''
if address[0] == '!':
operator = '!='
address = address[1:]
output.append(f'{ip_prefix} {prefix}addr {operator} {address}')
if 'port' in side_conf:
port = side_conf['port']
operator = ''
if port[0] == '!':
operator = '!='
port = port[1:]
output.append(f'th {prefix}port {operator} {port}')
if 'group' in side_conf:
group = side_conf['group']
if 'address_group' in group:
group_name = group['address_group']
operator = ''
if group_name[0] == '!':
operator = '!='
group_name = group_name[1:]
output.append(f'{ip_prefix} {prefix}addr {operator} @A{def_suffix}_{group_name}')
# Generate firewall group domain-group
elif 'domain_group' in group:
group_name = group['domain_group']
operator = ''
if group_name[0] == '!':
operator = '!='
group_name = group_name[1:]
output.append(f'{ip_prefix} {prefix}addr {operator} @D_{group_name}')
elif 'network_group' in group:
group_name = group['network_group']
operator = ''
if group_name[0] == '!':
operator = '!='
group_name = group_name[1:]
output.append(f'{ip_prefix} {prefix}addr {operator} @N{def_suffix}_{group_name}')
if 'port_group' in group:
group_name = group['port_group']
if proto == 'tcp_udp':
proto = 'th'
operator = ''
if group_name[0] == '!':
operator = '!='
group_name = group_name[1:]
output.append(f'{proto} {prefix}port {operator} @P_{group_name}')
if action == 'ignore':
output.append('counter notrack')
output.append(f'comment "ignore-{rule_id}"')
else:
output.append(f'counter ct timeout set ct-timeout-{rule_id}')
output.append(f'comment "timeout-{rule_id}"')
return " ".join(output)
@register_filter('conntrack_ct_policy')
def conntrack_ct_policy(protocol_conf):
output = []
for item in protocol_conf:
item_value = protocol_conf[item]
output.append(f'{item}: {item_value}')
return ", ".join(output)
@register_filter('range_to_regex')
def range_to_regex(num_range):
"""Convert range of numbers or list of ranges
to regex
% range_to_regex('11-12')
'(1[1-2])'
% range_to_regex(['11-12', '14-15'])
'(1[1-2]|1[4-5])'
"""
from vyos.range_regex import range_to_regex
if isinstance(num_range, list):
data = []
for entry in num_range:
if '-' not in entry:
data.append(entry)
else:
data.append(range_to_regex(entry))
return f'({"|".join(data)})'
if '-' not in num_range:
return num_range
regex = range_to_regex(num_range)
return f'({regex})'
@register_test('vyos_defined')
def vyos_defined(value, test_value=None, var_type=None):
"""
Jinja2 plugin to test if a variable is defined and not none - vyos_defined
will test value if defined and is not none and return true or false.
If test_value is supplied, the value must also pass == test_value to return true.
If var_type is supplied, the value must also be of the specified class/type
Examples:
1. Test if var is defined and not none:
{% if foo is vyos_defined %}
...
{% endif %}
2. Test if variable is defined, not none and has value "something"
{% if bar is vyos_defined("something") %}
...
{% endif %}
Parameters
----------
value : any
Value to test from ansible
test_value : any, optional
Value to test in addition of defined and not none, by default None
var_type : ['float', 'int', 'str', 'list', 'dict', 'tuple', 'bool'], optional
Type or Class to test for
Returns
-------
boolean
True if variable matches criteria, False in other cases.
Implementation inspired and re-used from https://github.com/aristanetworks/ansible-avd/
"""
from jinja2 import Undefined
if isinstance(value, Undefined) or value is None:
# Invalid value - return false
return False
elif test_value is not None and value != test_value:
# Valid value but not matching the optional argument
return False
elif str(var_type).lower() in ['float', 'int', 'str', 'list', 'dict', 'tuple', 'bool'] and str(var_type).lower() != type(value).__name__:
# Invalid class - return false
return False
else:
# Valid value and is matching optional argument if provided - return true
return True
diff --git a/src/op_mode/ikev2_profile_generator.py b/src/op_mode/ikev2_profile_generator.py
index 4ac4fb14a..169a15840 100755
--- a/src/op_mode/ikev2_profile_generator.py
+++ b/src/op_mode/ikev2_profile_generator.py
@@ -1,236 +1,246 @@
#!/usr/bin/env python3
#
# Copyright (C) 2021-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 argparse
from sys import exit
from socket import getfqdn
from cryptography.x509.oid import NameOID
from vyos.configquery import ConfigTreeQuery
+from vyos.pki import CERT_BEGIN
+from vyos.pki import CERT_END
+from vyos.pki import find_chain
+from vyos.pki import encode_certificate
from vyos.pki import load_certificate
from vyos.template import render_to_string
from vyos.utils.io import ask_input
# Apple profiles only support one IKE/ESP encryption cipher and hash, whereas
# VyOS comes with a multitude of different proposals for a connection.
#
# We take all available proposals from the VyOS CLI and ask the user which one
# he would like to get enabled in his profile - thus there is limited possibility
# to select a proposal that is not supported on the connection profile.
#
# IOS supports IKE-SA encryption algorithms:
# - DES
# - 3DES
# - AES-128
# - AES-256
# - AES-128-GCM
# - AES-256-GCM
# - ChaCha20Poly1305
#
vyos2apple_cipher = {
'3des' : '3DES',
'aes128' : 'AES-128',
'aes256' : 'AES-256',
'aes128gcm128' : 'AES-128-GCM',
'aes256gcm128' : 'AES-256-GCM',
'chacha20poly1305' : 'ChaCha20Poly1305',
}
# Windows supports IKE-SA encryption algorithms:
# - DES3
# - AES128
# - AES192
# - AES256
# - GCMAES128
# - GCMAES192
# - GCMAES256
#
vyos2windows_cipher = {
'3des' : 'DES3',
'aes128' : 'AES128',
'aes192' : 'AES192',
'aes256' : 'AES256',
'aes128gcm128' : 'GCMAES128',
'aes192gcm128' : 'GCMAES192',
'aes256gcm128' : 'GCMAES256',
}
# IOS supports IKE-SA integrity algorithms:
# - SHA1-96
# - SHA1-160
# - SHA2-256
# - SHA2-384
# - SHA2-512
#
vyos2apple_integrity = {
'sha1' : 'SHA1-96',
'sha1_160' : 'SHA1-160',
'sha256' : 'SHA2-256',
'sha384' : 'SHA2-384',
'sha512' : 'SHA2-512',
}
# Windows supports IKE-SA integrity algorithms:
# - SHA1-96
# - SHA1-160
# - SHA2-256
# - SHA2-384
# - SHA2-512
#
vyos2windows_integrity = {
'sha1' : 'SHA196',
'sha256' : 'SHA256',
'aes128gmac' : 'GCMAES128',
'aes192gmac' : 'GCMAES192',
'aes256gmac' : 'GCMAES256',
}
# IOS 14.2 and later do no support dh-group 1,2 and 5. Supported DH groups would
# be: 14, 15, 16, 17, 18, 19, 20, 21, 31
ios_supported_dh_groups = ['14', '15', '16', '17', '18', '19', '20', '21', '31']
# Windows 10 only allows a limited set of DH groups
windows_supported_dh_groups = ['1', '2', '14', '24']
parser = argparse.ArgumentParser()
parser.add_argument('--os', const='all', nargs='?', choices=['ios', 'windows'], help='Operating system used for config generation', required=True)
parser.add_argument("--connection", action="store", help='IPsec IKEv2 remote-access connection name from CLI', required=True)
parser.add_argument("--remote", action="store", help='VPN connection remote-address where the client will connect to', required=True)
parser.add_argument("--profile", action="store", help='IKEv2 profile name used in the profile list on the device')
parser.add_argument("--name", action="store", help='VPN connection name as seen in the VPN application later')
args = parser.parse_args()
ipsec_base = ['vpn', 'ipsec']
config_base = ipsec_base + ['remote-access', 'connection']
pki_base = ['pki']
conf = ConfigTreeQuery()
if not conf.exists(config_base):
exit('IPsec remote-access is not configured!')
profile_name = 'VyOS IKEv2 Profile'
if args.profile:
profile_name = args.profile
vpn_name = 'VyOS IKEv2 VPN'
if args.name:
vpn_name = args.name
conn_base = config_base + [args.connection]
if not conf.exists(conn_base):
exit(f'IPsec remote-access connection "{args.connection}" does not exist!')
data = conf.get_config_dict(conn_base, key_mangling=('-', '_'),
get_first_key=True, no_tag_node_value_mangle=True)
data['profile_name'] = profile_name
data['vpn_name'] = vpn_name
data['remote'] = args.remote
# This is a reverse-DNS style unique identifier used to detect duplicate profiles
tmp = getfqdn().split('.')
tmp = reversed(tmp)
data['rfqdn'] = '.'.join(tmp)
pki = conf.get_config_dict(pki_base, get_first_key=True)
cert_name = data['authentication']['x509']['certificate']
-data['certs'] = []
+cert_data = load_certificate(pki['certificate'][cert_name]['certificate'])
+data['cert_common_name'] = cert_data.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value
+data['ca_common_name'] = cert_data.issuer.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value
+data['ca_certificates'] = []
-for ca_name in data['authentication']['x509']['ca_certificate']:
- tmp = {}
- ca_cert = load_certificate(pki['ca'][ca_name]['certificate'])
- cert = load_certificate(pki['certificate'][cert_name]['certificate'])
-
-
- tmp['ca_cn'] = ca_cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value
- tmp['cert_cn'] = cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value
- tmp['ca_cert'] = conf.value(pki_base + ['ca', ca_name, 'certificate'])
-
- data['certs'].append(tmp)
+loaded_ca_certs = {load_certificate(c['certificate'])
+ for c in pki['ca'].values()} if 'ca' in pki else {}
+for ca_name in data['authentication']['x509']['ca_certificate']:
+ loaded_ca_cert = load_certificate(pki['ca'][ca_name]['certificate'])
+ ca_full_chain = find_chain(loaded_ca_cert, loaded_ca_certs)
+ for ca in ca_full_chain:
+ tmp = {
+ 'ca_name' : ca.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value,
+ 'ca_chain' : encode_certificate(ca).replace(CERT_BEGIN, '').replace(CERT_END, '').replace('\n', ''),
+ }
+ data['ca_certificates'].append(tmp)
+
+# Remove duplicate list entries for CA certificates, as they are added by their common name
+# https://stackoverflow.com/a/9427216
+data['ca_certificates'] = [dict(t) for t in {tuple(d.items()) for d in data['ca_certificates']}]
esp_proposals = conf.get_config_dict(ipsec_base + ['esp-group', data['esp_group'], 'proposal'],
key_mangling=('-', '_'), get_first_key=True)
ike_proposal = conf.get_config_dict(ipsec_base + ['ike-group', data['ike_group'], 'proposal'],
key_mangling=('-', '_'), get_first_key=True)
-
# This script works only for Apple iOS/iPadOS and Windows. Both operating systems
# have different limitations thus we load the limitations based on the operating
# system used.
vyos2client_cipher = vyos2apple_cipher if args.os == 'ios' else vyos2windows_cipher;
vyos2client_integrity = vyos2apple_integrity if args.os == 'ios' else vyos2windows_integrity;
supported_dh_groups = ios_supported_dh_groups if args.os == 'ios' else windows_supported_dh_groups;
# Create a dictionary containing client conform IKE settings
ike = {}
count = 1
for _, proposal in ike_proposal.items():
if {'dh_group', 'encryption', 'hash'} <= set(proposal):
if (proposal['encryption'] in set(vyos2client_cipher) and
proposal['hash'] in set(vyos2client_integrity) and
proposal['dh_group'] in set(supported_dh_groups)):
# We 're-code' from the VyOS IPsec proposals to the Apple naming scheme
proposal['encryption'] = vyos2client_cipher[ proposal['encryption'] ]
proposal['hash'] = vyos2client_integrity[ proposal['hash'] ]
ike.update( { str(count) : proposal } )
count += 1
# Create a dictionary containing Apple conform ESP settings
esp = {}
count = 1
for _, proposal in esp_proposals.items():
if {'encryption', 'hash'} <= set(proposal):
if proposal['encryption'] in set(vyos2client_cipher) and proposal['hash'] in set(vyos2client_integrity):
# We 're-code' from the VyOS IPsec proposals to the Apple naming scheme
proposal['encryption'] = vyos2client_cipher[ proposal['encryption'] ]
proposal['hash'] = vyos2client_integrity[ proposal['hash'] ]
esp.update( { str(count) : proposal } )
count += 1
try:
if len(ike) > 1:
# Propare the input questions for the user
tmp = '\n'
for number, options in ike.items():
tmp += f'({number}) Encryption {options["encryption"]}, Integrity {options["hash"]}, DH group {options["dh_group"]}\n'
tmp += '\nSelect one of the above IKE groups: '
data['ike_encryption'] = ike[ ask_input(tmp, valid_responses=list(ike)) ]
else:
data['ike_encryption'] = ike['1']
if len(esp) > 1:
tmp = '\n'
for number, options in esp.items():
tmp += f'({number}) Encryption {options["encryption"]}, Integrity {options["hash"]}\n'
tmp += '\nSelect one of the above ESP groups: '
data['esp_encryption'] = esp[ ask_input(tmp, valid_responses=list(esp)) ]
else:
data['esp_encryption'] = esp['1']
except KeyboardInterrupt:
exit("Interrupted")
print('\n\n==== <snip> ====')
if args.os == 'ios':
print(render_to_string('ipsec/ios_profile.j2', data))
print('==== </snip> ====\n')
print('Save the XML from above to a new file named "vyos.mobileconfig" and E-Mail it to your phone.')
elif args.os == 'windows':
print(render_to_string('ipsec/windows_profile.j2', data))
print('==== </snip> ====\n')