diff --git a/interface-definitions/nat.xml.in b/interface-definitions/nat.xml.in
index a06ceefb6..0a639bd80 100644
--- a/interface-definitions/nat.xml.in
+++ b/interface-definitions/nat.xml.in
@@ -1,158 +1,158 @@
<?xml version="1.0"?>
<interfaceDefinition>
<node name="nat" owner="${vyos_conf_scripts_dir}/nat.py">
<properties>
<help>Network Address Translation (NAT) parameters</help>
<priority>220</priority>
</properties>
<children>
<node name="destination">
<properties>
<help>Destination NAT settings</help>
</properties>
<children>
#include <include/nat-rule.xml.i>
<tagNode name="rule">
<children>
- #include <include/inbound-interface.xml.i>
+ #include <include/firewall/inbound-interface.xml.i>
<node name="translation">
<properties>
<help>Inside NAT IP (destination NAT only)</help>
</properties>
<children>
<leafNode name="address">
<properties>
<help>IP address, subnet, or range</help>
<valueHelp>
<format>ipv4</format>
<description>IPv4 address to match</description>
</valueHelp>
<valueHelp>
<format>ipv4net</format>
<description>IPv4 prefix to match</description>
</valueHelp>
<valueHelp>
<format>ipv4range</format>
<description>IPv4 address range to match</description>
</valueHelp>
<constraint>
<validator name="ipv4-prefix"/>
<validator name="ipv4-address"/>
<validator name="ipv4-range"/>
</constraint>
</properties>
</leafNode>
#include <include/nat-translation-port.xml.i>
#include <include/nat-translation-options.xml.i>
<node name="redirect">
<properties>
<help>Redirect to local host</help>
</properties>
<children>
#include <include/nat-translation-port.xml.i>
</children>
</node>
</children>
</node>
</children>
</tagNode>
</children>
</node>
<node name="source">
<properties>
<help>Source NAT settings</help>
</properties>
<children>
#include <include/nat-rule.xml.i>
<tagNode name="rule">
<properties>
<help>Rule number for NAT</help>
<valueHelp>
<format>u32:1-999999</format>
<description>Number of NAT rule</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 1-999999"/>
</constraint>
<constraintErrorMessage>NAT rule number must be between 1 and 999999</constraintErrorMessage>
</properties>
<children>
- #include <include/nat-interface.xml.i>
+ #include <include/firewall/outbound-interface.xml.i>
<node name="translation">
<properties>
<help>Outside NAT IP (source NAT only)</help>
</properties>
<children>
<leafNode name="address">
<properties>
<help>IP address, subnet, or range</help>
<completionHelp>
<list>masquerade</list>
</completionHelp>
<valueHelp>
<format>ipv4</format>
<description>IPv4 address to match</description>
</valueHelp>
<valueHelp>
<format>ipv4net</format>
<description>IPv4 prefix to match</description>
</valueHelp>
<valueHelp>
<format>ipv4range</format>
<description>IPv4 address range to match</description>
</valueHelp>
<valueHelp>
<format>masquerade</format>
<description>NAT to the primary address of outbound-interface</description>
</valueHelp>
<constraint>
<validator name="ipv4-prefix"/>
<validator name="ipv4-address"/>
<validator name="ipv4-range"/>
<regex>(masquerade)</regex>
</constraint>
</properties>
</leafNode>
#include <include/nat-translation-port.xml.i>
#include <include/nat-translation-options.xml.i>
</children>
</node>
</children>
</tagNode>
</children>
</node>
<node name="static">
<properties>
<help>Static NAT (one-to-one)</help>
</properties>
<children>
<tagNode name="rule">
<properties>
<help>Rule number for NAT</help>
</properties>
<children>
#include <include/generic-description.xml.i>
<node name="destination">
<properties>
<help>NAT destination parameters</help>
</properties>
<children>
#include <include/ipv4-address-prefix.xml.i>
</children>
</node>
#include <include/inbound-interface.xml.i>
<node name="translation">
<properties>
<help>Translation address or prefix</help>
</properties>
<children>
#include <include/ipv4-address-prefix.xml.i>
</children>
</node>
</children>
</tagNode>
</children>
</node>
</children>
</node>
</interfaceDefinition>
diff --git a/python/vyos/nat.py b/python/vyos/nat.py
index cc3c8103d..e32b5ae74 100644
--- a/python/vyos/nat.py
+++ b/python/vyos/nat.py
@@ -1,284 +1,304 @@
#!/usr/bin/env python3
#
# Copyright (C) 2022 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/>.
from vyos.template import is_ip_network
from vyos.utils.dict import dict_search_args
from vyos.template import bracketize_ipv6
def parse_nat_rule(rule_conf, rule_id, nat_type, ipv6=False):
output = []
ip_prefix = 'ip6' if ipv6 else 'ip'
log_prefix = ('DST' if nat_type == 'destination' else 'SRC') + f'-NAT-{rule_id}'
log_suffix = ''
if ipv6:
log_prefix = log_prefix.replace("NAT-", "NAT66-")
ignore_type_addr = False
translation_str = ''
if 'inbound_interface' in rule_conf:
- ifname = rule_conf['inbound_interface']
- if ifname != 'any':
- output.append(f'iifname "{ifname}"')
+ operator = ''
+ if 'interface_name' in rule_conf['inbound_interface']:
+ iiface = rule_conf['inbound_interface']['interface_name']
+ if iiface[0] == '!':
+ operator = '!='
+ iiface = iiface[1:]
+ output.append(f'iifname {operator} {{{iiface}}}')
+ else:
+ iiface = rule_conf['inbound_interface']['interface_group']
+ if iiface[0] == '!':
+ operator = '!='
+ iiface = iiface[1:]
+ output.append(f'iifname {operator} @I_{iiface}')
if 'outbound_interface' in rule_conf:
- ifname = rule_conf['outbound_interface']
- if ifname != 'any':
- output.append(f'oifname "{ifname}"')
+ operator = ''
+ if 'interface_name' in rule_conf['outbound_interface']:
+ oiface = rule_conf['outbound_interface']['interface_name']
+ if oiface[0] == '!':
+ operator = '!='
+ oiface = oiface[1:]
+ output.append(f'oifname {operator} {{{oiface}}}')
+ else:
+ oiface = rule_conf['outbound_interface']['interface_group']
+ if oiface[0] == '!':
+ operator = '!='
+ oiface = oiface[1:]
+ output.append(f'oifname {operator} @I_{oiface}')
if 'protocol' in rule_conf and rule_conf['protocol'] != 'all':
protocol = rule_conf['protocol']
if protocol == 'tcp_udp':
protocol = '{ tcp, udp }'
output.append(f'meta l4proto {protocol}')
if 'packet_type' in rule_conf:
output.append(f'pkttype ' + rule_conf['packet_type'])
if 'exclude' in rule_conf:
translation_str = 'return'
log_suffix = '-EXCL'
elif 'translation' in rule_conf:
addr = dict_search_args(rule_conf, 'translation', 'address')
port = dict_search_args(rule_conf, 'translation', 'port')
if 'redirect' in rule_conf['translation']:
translation_output = [f'redirect']
redirect_port = dict_search_args(rule_conf, 'translation', 'redirect', 'port')
if redirect_port:
translation_output.append(f'to {redirect_port}')
else:
translation_prefix = nat_type[:1]
translation_output = [f'{translation_prefix}nat']
if addr and is_ip_network(addr):
if not ipv6:
map_addr = dict_search_args(rule_conf, nat_type, 'address')
translation_output.append(f'{ip_prefix} prefix to {ip_prefix} {translation_prefix}addr map {{ {map_addr} : {addr} }}')
ignore_type_addr = True
else:
translation_output.append(f'prefix to {addr}')
elif addr == 'masquerade':
if port:
addr = f'{addr} to '
translation_output = [addr]
log_suffix = '-MASQ'
else:
translation_output.append('to')
if addr:
addr = bracketize_ipv6(addr)
translation_output.append(addr)
options = []
addr_mapping = dict_search_args(rule_conf, 'translation', 'options', 'address_mapping')
port_mapping = dict_search_args(rule_conf, 'translation', 'options', 'port_mapping')
if addr_mapping == 'persistent':
options.append('persistent')
if port_mapping and port_mapping != 'none':
options.append(port_mapping)
translation_str = " ".join(translation_output) + (f':{port}' if port else '')
if options:
translation_str += f' {",".join(options)}'
if not ipv6 and 'backend' in rule_conf['load_balance']:
hash_input_items = []
current_prob = 0
nat_map = []
for trans_addr, addr in rule_conf['load_balance']['backend'].items():
item_prob = int(addr['weight'])
upper_limit = current_prob + item_prob - 1
hash_val = str(current_prob) + '-' + str(upper_limit)
element = hash_val + " : " + trans_addr
nat_map.append(element)
current_prob = current_prob + item_prob
elements = ' , '.join(nat_map)
if 'hash' in rule_conf['load_balance'] and 'random' in rule_conf['load_balance']['hash']:
translation_str += ' numgen random mod 100 map ' + '{ ' + f'{elements}' + ' }'
else:
for input_param in rule_conf['load_balance']['hash']:
if input_param == 'source-address':
param = 'ip saddr'
elif input_param == 'destination-address':
param = 'ip daddr'
elif input_param == 'source-port':
prot = rule_conf['protocol']
param = f'{prot} sport'
elif input_param == 'destination-port':
prot = rule_conf['protocol']
param = f'{prot} dport'
hash_input_items.append(param)
hash_input = ' . '.join(hash_input_items)
translation_str += f' jhash ' + f'{hash_input}' + ' mod 100 map ' + '{ ' + f'{elements}' + ' }'
for target in ['source', 'destination']:
if target not in rule_conf:
continue
side_conf = rule_conf[target]
prefix = target[:1]
addr = dict_search_args(side_conf, 'address')
if addr and not (ignore_type_addr and target == nat_type):
operator = ''
if addr[:1] == '!':
operator = '!='
addr = addr[1:]
output.append(f'{ip_prefix} {prefix}addr {operator} {addr}')
addr_prefix = dict_search_args(side_conf, 'prefix')
if addr_prefix and ipv6:
operator = ''
if addr_prefix[:1] == '!':
operator = '!='
addr_prefix = addr_prefix[1:]
output.append(f'ip6 {prefix}addr {operator} {addr_prefix}')
port = dict_search_args(side_conf, 'port')
if port:
protocol = rule_conf['protocol']
if protocol == 'tcp_udp':
protocol = 'th'
operator = ''
if port[:1] == '!':
operator = '!='
port = port[1:]
output.append(f'{protocol} {prefix}port {operator} {{ {port} }}')
if 'group' in side_conf:
group = side_conf['group']
if 'address_group' in group and not (ignore_type_addr and target == nat_type):
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_{group_name}')
# Generate firewall group domain-group
elif 'domain_group' in group and not (ignore_type_addr and target == nat_type):
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 and not (ignore_type_addr and target == nat_type):
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_{group_name}')
if 'mac_group' in group:
group_name = group['mac_group']
operator = ''
if group_name[0] == '!':
operator = '!='
group_name = group_name[1:]
output.append(f'ether {prefix}addr {operator} @M_{group_name}')
if 'port_group' in group:
proto = rule_conf['protocol']
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}')
output.append('counter')
if 'log' in rule_conf:
output.append(f'log prefix "[{log_prefix}{log_suffix}]"')
if translation_str:
output.append(translation_str)
output.append(f'comment "{log_prefix}"')
return " ".join(output)
def parse_nat_static_rule(rule_conf, rule_id, nat_type):
output = []
log_prefix = ('STATIC-DST' if nat_type == 'destination' else 'STATIC-SRC') + f'-NAT-{rule_id}'
log_suffix = ''
ignore_type_addr = False
translation_str = ''
if 'inbound_interface' in rule_conf:
ifname = rule_conf['inbound_interface']
ifprefix = 'i' if nat_type == 'destination' else 'o'
if ifname != 'any':
output.append(f'{ifprefix}ifname "{ifname}"')
if 'exclude' in rule_conf:
translation_str = 'return'
log_suffix = '-EXCL'
elif 'translation' in rule_conf:
translation_prefix = nat_type[:1]
translation_output = [f'{translation_prefix}nat']
addr = dict_search_args(rule_conf, 'translation', 'address')
map_addr = dict_search_args(rule_conf, 'destination', 'address')
if nat_type == 'source':
addr, map_addr = map_addr, addr # Swap
if addr and is_ip_network(addr):
translation_output.append(f'ip prefix to ip {translation_prefix}addr map {{ {map_addr} : {addr} }}')
ignore_type_addr = True
elif addr:
translation_output.append(f'to {addr}')
options = []
addr_mapping = dict_search_args(rule_conf, 'translation', 'options', 'address_mapping')
port_mapping = dict_search_args(rule_conf, 'translation', 'options', 'port_mapping')
if addr_mapping == 'persistent':
options.append('persistent')
if port_mapping and port_mapping != 'none':
options.append(port_mapping)
if options:
translation_output.append(",".join(options))
translation_str = " ".join(translation_output)
prefix = nat_type[:1]
addr = dict_search_args(rule_conf, 'translation' if nat_type == 'source' else nat_type, 'address')
if addr and not ignore_type_addr:
output.append(f'ip {prefix}addr {addr}')
output.append('counter')
if translation_str:
output.append(translation_str)
if 'log' in rule_conf:
output.append(f'log prefix "[{log_prefix}{log_suffix}]"')
output.append(f'comment "{log_prefix}"')
return " ".join(output)
diff --git a/smoketest/scripts/cli/test_nat.py b/smoketest/scripts/cli/test_nat.py
index 703e5ab28..2f744a2f7 100755
--- a/smoketest/scripts/cli/test_nat.py
+++ b/smoketest/scripts/cli/test_nat.py
@@ -1,294 +1,296 @@
#!/usr/bin/env python3
#
# Copyright (C) 2020-2022 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 jmespath
import json
import os
import unittest
from base_vyostest_shim import VyOSUnitTestSHIM
from vyos.configsession import ConfigSessionError
from vyos.utils.process import cmd
from vyos.utils.dict import dict_search
base_path = ['nat']
src_path = base_path + ['source']
dst_path = base_path + ['destination']
static_path = base_path + ['static']
nftables_nat_config = '/run/nftables_nat.conf'
nftables_static_nat_conf = '/run/nftables_static-nat-rules.nft'
class TestNAT(VyOSUnitTestSHIM.TestCase):
@classmethod
def setUpClass(cls):
super(TestNAT, cls).setUpClass()
# ensure we can also run this test on a live system - so lets clean
# out the current configuration :)
cls.cli_delete(cls, base_path)
def tearDown(self):
self.cli_delete(base_path)
self.cli_commit()
self.assertFalse(os.path.exists(nftables_nat_config))
self.assertFalse(os.path.exists(nftables_static_nat_conf))
def verify_nftables(self, nftables_search, table, inverse=False, args=''):
nftables_output = cmd(f'sudo nft {args} list table {table}')
for search in nftables_search:
matched = False
for line in nftables_output.split("\n"):
if all(item in line for item in search):
matched = True
break
self.assertTrue(not matched if inverse else matched, msg=search)
def wait_for_domain_resolver(self, table, set_name, element, max_wait=10):
# Resolver no longer blocks commit, need to wait for daemon to populate set
count = 0
while count < max_wait:
code = run(f'sudo nft get element {table} {set_name} {{ {element} }}')
if code == 0:
return True
count += 1
sleep(1)
return False
def test_snat(self):
rules = ['100', '110', '120', '130', '200', '210', '220', '230']
outbound_iface_100 = 'eth0'
outbound_iface_200 = 'eth1'
nftables_search = ['jump VYOS_PRE_SNAT_HOOK']
for rule in rules:
network = f'192.168.{rule}.0/24'
# depending of rule order we check either for source address for NAT
# or configured destination address for NAT
if int(rule) < 200:
self.cli_set(src_path + ['rule', rule, 'source', 'address', network])
- self.cli_set(src_path + ['rule', rule, 'outbound-interface', outbound_iface_100])
+ self.cli_set(src_path + ['rule', rule, 'outbound-interface', 'interface-name', outbound_iface_100])
self.cli_set(src_path + ['rule', rule, 'translation', 'address', 'masquerade'])
nftables_search.append([f'saddr {network}', f'oifname "{outbound_iface_100}"', 'masquerade'])
else:
self.cli_set(src_path + ['rule', rule, 'destination', 'address', network])
- self.cli_set(src_path + ['rule', rule, 'outbound-interface', outbound_iface_200])
+ self.cli_set(src_path + ['rule', rule, 'outbound-interface', 'interface-name', outbound_iface_200])
self.cli_set(src_path + ['rule', rule, 'exclude'])
nftables_search.append([f'daddr {network}', f'oifname "{outbound_iface_200}"', 'return'])
self.cli_commit()
self.verify_nftables(nftables_search, 'ip vyos_nat')
def test_snat_groups(self):
address_group = 'smoketest_addr'
address_group_member = '192.0.2.1'
+ interface_group = 'smoketest_ifaces'
+ interface_group_member = 'bond.99'
rule = '100'
- outbound_iface = 'eth0'
self.cli_set(['firewall', 'group', 'address-group', address_group, 'address', address_group_member])
+ self.cli_set(['firewall', 'group', 'interface-group', interface_group, 'interface', interface_group_member])
self.cli_set(src_path + ['rule', rule, 'source', 'group', 'address-group', address_group])
- self.cli_set(src_path + ['rule', rule, 'outbound-interface', outbound_iface])
+ self.cli_set(src_path + ['rule', rule, 'outbound-interface', 'interface-group', interface_group])
self.cli_set(src_path + ['rule', rule, 'translation', 'address', 'masquerade'])
self.cli_commit()
nftables_search = [
[f'set A_{address_group}'],
[f'elements = {{ {address_group_member} }}'],
- [f'ip saddr @A_{address_group}', f'oifname "{outbound_iface}"', 'masquerade']
+ [f'ip saddr @A_{address_group}', f'oifname @I_{interface_group}', 'masquerade']
]
self.verify_nftables(nftables_search, 'ip vyos_nat')
self.cli_delete(['firewall'])
def test_dnat(self):
rules = ['100', '110', '120', '130', '200', '210', '220', '230']
inbound_iface_100 = 'eth0'
inbound_iface_200 = 'eth1'
inbound_proto_100 = 'udp'
inbound_proto_200 = 'tcp'
nftables_search = ['jump VYOS_PRE_DNAT_HOOK']
for rule in rules:
port = f'10{rule}'
self.cli_set(dst_path + ['rule', rule, 'source', 'port', port])
self.cli_set(dst_path + ['rule', rule, 'translation', 'address', '192.0.2.1'])
self.cli_set(dst_path + ['rule', rule, 'translation', 'port', port])
rule_search = [f'dnat to 192.0.2.1:{port}']
if int(rule) < 200:
self.cli_set(dst_path + ['rule', rule, 'protocol', inbound_proto_100])
- self.cli_set(dst_path + ['rule', rule, 'inbound-interface', inbound_iface_100])
+ self.cli_set(dst_path + ['rule', rule, 'inbound-interface', 'interface-name', inbound_iface_100])
rule_search.append(f'{inbound_proto_100} sport {port}')
rule_search.append(f'iifname "{inbound_iface_100}"')
else:
self.cli_set(dst_path + ['rule', rule, 'protocol', inbound_proto_200])
- self.cli_set(dst_path + ['rule', rule, 'inbound-interface', inbound_iface_200])
+ self.cli_set(dst_path + ['rule', rule, 'inbound-interface', 'interface-name', inbound_iface_200])
rule_search.append(f'iifname "{inbound_iface_200}"')
nftables_search.append(rule_search)
self.cli_commit()
self.verify_nftables(nftables_search, 'ip vyos_nat')
def test_snat_required_translation_address(self):
# T2813: Ensure translation address is specified
rule = '5'
self.cli_set(src_path + ['rule', rule, 'source', 'address', '192.0.2.0/24'])
# check validate() - translation address not specified
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(src_path + ['rule', rule, 'translation', 'address', 'masquerade'])
self.cli_commit()
def test_dnat_negated_addresses(self):
# T3186: negated addresses are not accepted by nftables
rule = '1000'
self.cli_set(dst_path + ['rule', rule, 'destination', 'address', '!192.0.2.1'])
self.cli_set(dst_path + ['rule', rule, 'destination', 'port', '53'])
- self.cli_set(dst_path + ['rule', rule, 'inbound-interface', 'eth0'])
+ self.cli_set(dst_path + ['rule', rule, 'inbound-interface', 'interface-name', 'eth0'])
self.cli_set(dst_path + ['rule', rule, 'protocol', 'tcp_udp'])
self.cli_set(dst_path + ['rule', rule, 'source', 'address', '!192.0.2.1'])
self.cli_set(dst_path + ['rule', rule, 'translation', 'address', '192.0.2.1'])
self.cli_set(dst_path + ['rule', rule, 'translation', 'port', '53'])
self.cli_commit()
def test_nat_no_rules(self):
# T3206: deleting all rules but keep the direction 'destination' or
# 'source' resulteds in KeyError: 'rule'.
#
# Test that both 'nat destination' and 'nat source' nodes can exist
# without any rule
self.cli_set(src_path)
self.cli_set(dst_path)
self.cli_set(static_path)
self.cli_commit()
def test_dnat_without_translation_address(self):
- self.cli_set(dst_path + ['rule', '1', 'inbound-interface', 'eth1'])
+ self.cli_set(dst_path + ['rule', '1', 'inbound-interface', 'interface-name', 'eth1'])
self.cli_set(dst_path + ['rule', '1', 'destination', 'port', '443'])
self.cli_set(dst_path + ['rule', '1', 'protocol', 'tcp'])
self.cli_set(dst_path + ['rule', '1', 'packet-type', 'host'])
self.cli_set(dst_path + ['rule', '1', 'translation', 'port', '443'])
self.cli_commit()
nftables_search = [
['iifname "eth1"', 'tcp dport 443', 'pkttype host', 'dnat to :443']
]
self.verify_nftables(nftables_search, 'ip vyos_nat')
def test_static_nat(self):
dst_addr_1 = '10.0.1.1'
translate_addr_1 = '192.168.1.1'
dst_addr_2 = '203.0.113.0/24'
translate_addr_2 = '192.0.2.0/24'
ifname = 'eth0'
self.cli_set(static_path + ['rule', '10', 'destination', 'address', dst_addr_1])
self.cli_set(static_path + ['rule', '10', 'inbound-interface', ifname])
self.cli_set(static_path + ['rule', '10', 'translation', 'address', translate_addr_1])
self.cli_set(static_path + ['rule', '20', 'destination', 'address', dst_addr_2])
self.cli_set(static_path + ['rule', '20', 'inbound-interface', ifname])
self.cli_set(static_path + ['rule', '20', 'translation', 'address', translate_addr_2])
self.cli_commit()
nftables_search = [
[f'iifname "{ifname}"', f'ip daddr {dst_addr_1}', f'dnat to {translate_addr_1}'],
[f'oifname "{ifname}"', f'ip saddr {translate_addr_1}', f'snat to {dst_addr_1}'],
[f'iifname "{ifname}"', f'dnat ip prefix to ip daddr map {{ {dst_addr_2} : {translate_addr_2} }}'],
[f'oifname "{ifname}"', f'snat ip prefix to ip saddr map {{ {translate_addr_2} : {dst_addr_2} }}']
]
self.verify_nftables(nftables_search, 'ip vyos_static_nat')
def test_dnat_redirect(self):
dst_addr_1 = '10.0.1.1'
dest_port = '5122'
protocol = 'tcp'
redirected_port = '22'
ifname = 'eth0'
self.cli_set(dst_path + ['rule', '10', 'destination', 'address', dst_addr_1])
self.cli_set(dst_path + ['rule', '10', 'destination', 'port', dest_port])
self.cli_set(dst_path + ['rule', '10', 'protocol', protocol])
- self.cli_set(dst_path + ['rule', '10', 'inbound-interface', ifname])
+ self.cli_set(dst_path + ['rule', '10', 'inbound-interface', 'interface-name', ifname])
self.cli_set(dst_path + ['rule', '10', 'translation', 'redirect', 'port', redirected_port])
self.cli_set(dst_path + ['rule', '20', 'destination', 'address', dst_addr_1])
self.cli_set(dst_path + ['rule', '20', 'destination', 'port', dest_port])
self.cli_set(dst_path + ['rule', '20', 'protocol', protocol])
- self.cli_set(dst_path + ['rule', '20', 'inbound-interface', ifname])
+ self.cli_set(dst_path + ['rule', '20', 'inbound-interface', 'interface-name', ifname])
self.cli_set(dst_path + ['rule', '20', 'translation', 'redirect'])
self.cli_commit()
nftables_search = [
[f'iifname "{ifname}"', f'ip daddr {dst_addr_1}', f'{protocol} dport {dest_port}', f'redirect to :{redirected_port}'],
[f'iifname "{ifname}"', f'ip daddr {dst_addr_1}', f'{protocol} dport {dest_port}', f'redirect']
]
self.verify_nftables(nftables_search, 'ip vyos_nat')
def test_nat_balance(self):
ifname = 'eth0'
member_1 = '198.51.100.1'
weight_1 = '10'
member_2 = '198.51.100.2'
weight_2 = '90'
member_3 = '192.0.2.1'
weight_3 = '35'
member_4 = '192.0.2.2'
weight_4 = '65'
dst_port = '443'
- self.cli_set(dst_path + ['rule', '1', 'inbound-interface', ifname])
+ self.cli_set(dst_path + ['rule', '1', 'inbound-interface', 'interface-name', ifname])
self.cli_set(dst_path + ['rule', '1', 'protocol', 'tcp'])
self.cli_set(dst_path + ['rule', '1', 'destination', 'port', dst_port])
self.cli_set(dst_path + ['rule', '1', 'load-balance', 'hash', 'source-address'])
self.cli_set(dst_path + ['rule', '1', 'load-balance', 'hash', 'source-port'])
self.cli_set(dst_path + ['rule', '1', 'load-balance', 'hash', 'destination-address'])
self.cli_set(dst_path + ['rule', '1', 'load-balance', 'hash', 'destination-port'])
self.cli_set(dst_path + ['rule', '1', 'load-balance', 'backend', member_1, 'weight', weight_1])
self.cli_set(dst_path + ['rule', '1', 'load-balance', 'backend', member_2, 'weight', weight_2])
- self.cli_set(src_path + ['rule', '1', 'outbound-interface', ifname])
+ self.cli_set(src_path + ['rule', '1', 'outbound-interface', 'interface-name', ifname])
self.cli_set(src_path + ['rule', '1', 'load-balance', 'hash', 'random'])
self.cli_set(src_path + ['rule', '1', 'load-balance', 'backend', member_3, 'weight', weight_3])
self.cli_set(src_path + ['rule', '1', 'load-balance', 'backend', member_4, 'weight', weight_4])
self.cli_commit()
nftables_search = [
[f'iifname "{ifname}"', f'tcp dport {dst_port}', f'dnat to jhash ip saddr . tcp sport . ip daddr . tcp dport mod 100 map', f'0-9 : {member_1}, 10-99 : {member_2}'],
[f'oifname "{ifname}"', f'snat to numgen random mod 100 map', f'0-34 : {member_3}, 35-99 : {member_4}']
]
self.verify_nftables(nftables_search, 'ip vyos_nat')
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/src/conf_mode/nat.py b/src/conf_mode/nat.py
index 52a7a71fd..cb97a8662 100755
--- a/src/conf_mode/nat.py
+++ b/src/conf_mode/nat.py
@@ -1,237 +1,243 @@
#!/usr/bin/env python3
#
# Copyright (C) 2020-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 jmespath
import json
import os
from sys import exit
from netifaces import interfaces
from vyos.base import Warning
from vyos.config import Config
from vyos.configdep import set_dependents, call_dependents
from vyos.template import render
from vyos.template import is_ip_network
from vyos.utils.kernel import check_kmod
from vyos.utils.dict import dict_search
from vyos.utils.dict import dict_search_args
from vyos.utils.process import cmd
from vyos.utils.process import run
from vyos.utils.network import is_addr_assigned
from vyos import ConfigError
from vyos import airbag
airbag.enable()
k_mod = ['nft_nat', 'nft_chain_nat']
nftables_nat_config = '/run/nftables_nat.conf'
nftables_static_nat_conf = '/run/nftables_static-nat-rules.nft'
valid_groups = [
'address_group',
'domain_group',
'network_group',
'port_group'
]
def get_config(config=None):
if config:
conf = config
else:
conf = Config()
base = ['nat']
nat = conf.get_config_dict(base, key_mangling=('-', '_'),
get_first_key=True,
with_recursive_defaults=True)
set_dependents('conntrack', conf)
if not conf.exists(base):
nat['deleted'] = ''
return nat
nat['firewall_group'] = conf.get_config_dict(['firewall', 'group'], key_mangling=('-', '_'), get_first_key=True,
no_tag_node_value_mangle=True)
return nat
def verify_rule(config, err_msg, groups_dict):
""" Common verify steps used for both source and destination NAT """
if (dict_search('translation.port', config) != None or
dict_search('translation.redirect.port', config) != None or
dict_search('destination.port', config) != None or
dict_search('source.port', config)):
if config['protocol'] not in ['tcp', 'udp', 'tcp_udp']:
raise ConfigError(f'{err_msg}\n' \
'ports can only be specified when protocol is '\
'either tcp, udp or tcp_udp!')
if is_ip_network(dict_search('translation.address', config)):
raise ConfigError(f'{err_msg}\n' \
'Cannot use ports with an IPv4 network as translation address as it\n' \
'statically maps a whole network of addresses onto another\n' \
'network of addresses')
for side in ['destination', 'source']:
if side in config:
side_conf = config[side]
if len({'address', 'fqdn'} & set(side_conf)) > 1:
raise ConfigError('Only one of address, fqdn or geoip can be specified')
if 'group' in side_conf:
if len({'address_group', 'network_group', 'domain_group'} & set(side_conf['group'])) > 1:
raise ConfigError('Only one address-group, network-group or domain-group can be specified')
for group in valid_groups:
if group in side_conf['group']:
group_name = side_conf['group'][group]
error_group = group.replace("_", "-")
if group in ['address_group', 'network_group', 'domain_group']:
types = [t for t in ['address', 'fqdn'] if t in side_conf]
if types:
raise ConfigError(f'{error_group} and {types[0]} cannot both be defined')
if group_name and group_name[0] == '!':
group_name = group_name[1:]
group_obj = dict_search_args(groups_dict, group, group_name)
if group_obj is None:
raise ConfigError(f'Invalid {error_group} "{group_name}" on nat rule')
if not group_obj:
Warning(f'{error_group} "{group_name}" has no members!')
if dict_search_args(side_conf, 'group', 'port_group'):
if 'protocol' not in config:
raise ConfigError('Protocol must be defined if specifying a port-group')
if config['protocol'] not in ['tcp', 'udp', 'tcp_udp']:
raise ConfigError('Protocol must be tcp, udp, or tcp_udp when specifying a port-group')
if 'load_balance' in config:
for item in ['source-port', 'destination-port']:
if item in config['load_balance']['hash'] and config['protocol'] not in ['tcp', 'udp']:
raise ConfigError('Protocol must be tcp or udp when specifying hash ports')
count = 0
if 'backend' in config['load_balance']:
for member in config['load_balance']['backend']:
weight = config['load_balance']['backend'][member]['weight']
count = count + int(weight)
if count != 100:
Warning(f'Sum of weight for nat load balance rule is not 100. You may get unexpected behaviour')
def verify(nat):
if not nat or 'deleted' in nat:
# no need to verify the CLI as NAT is going to be deactivated
return None
if dict_search('source.rule', nat):
for rule, config in dict_search('source.rule', nat).items():
err_msg = f'Source NAT configuration error in rule {rule}:'
if 'outbound_interface' in config:
- if config['outbound_interface'] not in 'any' and config['outbound_interface'] not in interfaces():
- Warning(f'rule "{rule}" interface "{config["outbound_interface"]}" does not exist on this system')
+ if 'interface_name' in config['outbound_interface'] and 'interface_group' in config['outbound_interface']:
+ raise ConfigError(f'Cannot specify both interface-group and interface-name for nat source rule "{rule}"')
+ elif 'interface_name' in config['outbound_interface']:
+ if config['outbound_interface']['interface_name'] not in 'any' and config['outbound_interface']['interface_name'] not in interfaces():
+ Warning(f'rule "{rule}" interface "{config["outbound_interface"]["interface_name"]}" does not exist on this system')
if not dict_search('translation.address', config) and not dict_search('translation.port', config):
if 'exclude' not in config and 'backend' not in config['load_balance']:
raise ConfigError(f'{err_msg} translation requires address and/or port')
addr = dict_search('translation.address', config)
if addr != None and addr != 'masquerade' and not is_ip_network(addr):
for ip in addr.split('-'):
if not is_addr_assigned(ip):
Warning(f'IP address {ip} does not exist on the system!')
# common rule verification
verify_rule(config, err_msg, nat['firewall_group'])
if dict_search('destination.rule', nat):
for rule, config in dict_search('destination.rule', nat).items():
err_msg = f'Destination NAT configuration error in rule {rule}:'
if 'inbound_interface' in config:
- if config['inbound_interface'] not in 'any' and config['inbound_interface'] not in interfaces():
- Warning(f'rule "{rule}" interface "{config["inbound_interface"]}" does not exist on this system')
+ if 'interface_name' in config['inbound_interface'] and 'interface_group' in config['inbound_interface']:
+ raise ConfigError(f'Cannot specify both interface-group and interface-name for destination nat rule "{rule}"')
+ elif 'interface_name' in config['inbound_interface']:
+ if config['inbound_interface']['interface_name'] not in 'any' and config['inbound_interface']['interface_name'] not in interfaces():
+ Warning(f'rule "{rule}" interface "{config["inbound_interface"]["interface_name"]}" does not exist on this system')
if not dict_search('translation.address', config) and not dict_search('translation.port', config) and 'redirect' not in config['translation']:
if 'exclude' not in config and 'backend' not in config['load_balance']:
raise ConfigError(f'{err_msg} translation requires address and/or port')
# common rule verification
verify_rule(config, err_msg, nat['firewall_group'])
if dict_search('static.rule', nat):
for rule, config in dict_search('static.rule', nat).items():
err_msg = f'Static NAT configuration error in rule {rule}:'
if 'inbound_interface' not in config:
raise ConfigError(f'{err_msg}\n' \
'inbound-interface not specified')
# common rule verification
verify_rule(config, err_msg, nat['firewall_group'])
return None
def generate(nat):
if not os.path.exists(nftables_nat_config):
nat['first_install'] = True
render(nftables_nat_config, 'firewall/nftables-nat.j2', nat)
render(nftables_static_nat_conf, 'firewall/nftables-static-nat.j2', nat)
# dry-run newly generated configuration
tmp = run(f'nft -c -f {nftables_nat_config}')
if tmp > 0:
raise ConfigError('Configuration file errors encountered!')
tmp = run(f'nft -c -f {nftables_static_nat_conf}')
if tmp > 0:
raise ConfigError('Configuration file errors encountered!')
return None
def apply(nat):
cmd(f'nft -f {nftables_nat_config}')
cmd(f'nft -f {nftables_static_nat_conf}')
if not nat or 'deleted' in nat:
os.unlink(nftables_nat_config)
os.unlink(nftables_static_nat_conf)
call_dependents()
return None
if __name__ == '__main__':
try:
check_kmod(k_mod)
c = get_config()
verify(c)
generate(c)
apply(c)
except ConfigError as e:
print(e)
exit(1)
diff --git a/src/migration-scripts/nat/5-to-6 b/src/migration-scripts/nat/5-to-6
new file mode 100755
index 000000000..de3830582
--- /dev/null
+++ b/src/migration-scripts/nat/5-to-6
@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 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/>.
+
+# T5643: move from 'set nat [source|destination] rule X [inbound-interface|outbound interface] <iface>'
+# to
+# 'set nat [source|destination] rule X [inbound-interface|outbound interface] interface-name <iface>'
+
+from sys import argv,exit
+from vyos.configtree import ConfigTree
+
+if len(argv) < 2:
+ print("Must specify file name!")
+ exit(1)
+
+file_name = argv[1]
+
+with open(file_name, 'r') as f:
+ config_file = f.read()
+
+config = ConfigTree(config_file)
+
+if not config.exists(['nat']):
+ # Nothing to do
+ exit(0)
+
+for direction in ['source', 'destination']:
+ # If a node doesn't exist, we obviously have nothing to do.
+ if not config.exists(['nat', direction]):
+ continue
+
+ # However, we also need to handle the case when a 'source' or 'destination' sub-node does exist,
+ # but there are no rules under it.
+ if not config.list_nodes(['nat', direction]):
+ continue
+
+ for rule in config.list_nodes(['nat', direction, 'rule']):
+ base = ['nat', direction, 'rule', rule]
+ for iface in ['inbound-interface','outbound-interface']:
+ if config.exists(base + [iface]):
+ tmp = config.return_value(base + [iface])
+ config.delete(base + [iface])
+ config.set(base + [iface, 'interface-name'], value=tmp)
+
+try:
+ with open(file_name, 'w') as f:
+ f.write(config.to_string())
+except OSError as e:
+ print("Failed to save the modified config: {}".format(e))
+ exit(1)