Page MenuHomeVyOS Platform
Files F118309542

No OneTemporary
Actions

Size
11 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/smoketest/scripts/cli/test_system_flow-accounting.py b/smoketest/scripts/cli/test_system_flow-accounting.py
index 7de865b72..a5ab9b1fd 100755
--- a/smoketest/scripts/cli/test_system_flow-accounting.py
+++ b/smoketest/scripts/cli/test_system_flow-accounting.py
@@ -1,284 +1,298 @@
#!/usr/bin/env python3
#
# Copyright VyOS maintainers and contributors <maintainers@vyos.io>
#
# 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 unittest
from base_vyostest_shim import VyOSUnitTestSHIM
from vyos.configsession import ConfigSessionError
from vyos.ifconfig import Section
from vyos.utils.kernel import is_module_loaded
from vyos.utils.kernel import get_module_data
from vyos.utils.process import cmd
module_name = 'ipt_NETFLOW'
base_path = ['system', 'flow-accounting']
class TestSystemFlowAccounting(VyOSUnitTestSHIM.TestCase):
def _get_iptables_watched_interfaces(self, command, table, chain, column_name):
iptables_command = f'{command} -vn -t {table} -L {chain}'
data = cmd(iptables_command, message='Failed to get flows list')
data = data.splitlines()
self.assertGreaterEqual(
len(data), 2, "Unexpected output of {command}, should be at least two lines"
)
column_index = data[1].split().index(column_name)
interfaces = [
line.split()[column_index] for line in data[2:] if 'NETFLOW' in line
]
return interfaces
def _get_iptables_watched_ingress_interfaces(self, command):
return self._get_iptables_watched_interfaces(command, 'raw', 'PREROUTING', 'in')
def _get_iptables_watched_egress_interfaces(self, command):
return self._get_iptables_watched_interfaces(
command, 'mangle', 'POSTROUTING', 'out'
)
def _assert_ingress_interfaces(self, interfaces):
for command in 'iptables', 'ip6tables':
self.assertEqual(
set(self._get_iptables_watched_ingress_interfaces(command)),
set(interfaces),
command,
)
def _assert_egress_interfaces(self, interfaces):
for command in 'iptables', 'ip6tables':
self.assertEqual(
set(self._get_iptables_watched_egress_interfaces(command)),
set(interfaces),
command,
)
@classmethod
def setUpClass(cls):
super(TestSystemFlowAccounting, 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):
# after service removal process must no longer run
self.assertTrue(is_module_loaded(module_name))
self.cli_delete(base_path)
self.cli_commit()
# after service removal process must no longer run
self.assertFalse(is_module_loaded(module_name))
self._assert_ingress_interfaces([])
self._assert_egress_interfaces([])
# always forward to base class
super().tearDown()
def test_basic(self):
engine_id = '33'
self.cli_set(base_path + ['netflow', 'engine-id', engine_id])
# You need to configure at least one interface for flow-accounting
with self.assertRaises(ConfigSessionError):
self.cli_commit()
for interface in Section.interfaces('ethernet'):
self.cli_set(base_path + ['netflow', 'interface', interface])
# You need to configure at least one NetFlow server
with self.assertRaises(ConfigSessionError):
self.cli_commit()
netflow_server = '11.22.33.44'
self.cli_set(base_path + ['netflow', 'server', netflow_server])
# commit changes, this time should work
self.cli_commit()
# verify configuration
self._assert_ingress_interfaces(Section.interfaces('ethernet'))
self._assert_egress_interfaces([])
module_data = get_module_data(module_name)
self.assertEqual(engine_id, module_data['parameters']['engine_id'])
def test_netflow(self):
engine_id = '33'
max_flows = '667'
dummy_if = 'dum3842'
agent_address = '192.0.2.10'
version = '10'
active_timeout = '900'
inactive_timeout = '30'
source_ipv4_address = '192.0.2.1'
source_ipv6_address = '2001:db8::ab'
netflow_server = {
'11.22.33.44': {},
'55.66.77.88': {'port': '6000'},
'100.12.14.1': {'source-interface': dummy_if},
'203.0.113.21': {'port': '3000', 'source-address': source_ipv4_address},
'2001:db8::1': {'source-address': source_ipv6_address},
}
# ipt_NETFLOW sorts destinations by IP
expected_destination = '11.22.33.44:2055,55.66.77.88:6000,100.12.14.1:2055%dum3842,203.0.113.21:3000@192.0.2.1,[2001:db8::1]:2055@2001:db8::ab'
self.cli_set(
['interfaces', 'dummy', dummy_if, 'address', agent_address + '/32']
)
self.cli_set(
['interfaces', 'dummy', dummy_if, 'address', source_ipv4_address + '/32']
)
self.cli_set(
['interfaces', 'dummy', dummy_if, 'address', source_ipv6_address + '/128']
)
for interface in Section.interfaces('ethernet'):
self.cli_set(base_path + ['netflow', 'interface', interface])
self.cli_set(base_path + ['netflow', 'engine-id', engine_id])
self.cli_set(base_path + ['netflow', 'max-flows', max_flows])
self.cli_set(base_path + ['netflow', 'version', version])
self.cli_set(base_path + ['netflow', 'active-timeout', active_timeout])
self.cli_set(base_path + ['netflow', 'inactive-timeout', inactive_timeout])
# You need to configure at least one netflow server
with self.assertRaises(ConfigSessionError):
self.cli_commit()
for server, server_config in netflow_server.items():
self.cli_set(base_path + ['netflow', 'server', server])
if 'port' in server_config:
self.cli_set(base_path + ['netflow', 'server', server, 'port', server_config['port']])
if 'source-address' in server_config:
self.cli_set(
base_path
+ [
'netflow',
'server',
server,
'source-address',
server_config['source-address'],
]
)
if 'source-interface' in server_config:
self.cli_set(
base_path
+ [
'netflow',
'server',
server,
'source-interface',
server_config['source-interface'],
]
)
# commit changes
self.cli_commit()
module_data = get_module_data(module_name)
self.assertEqual(engine_id, module_data['parameters']['engine_id'])
self.assertEqual(max_flows, module_data['parameters']['maxflows'])
self.assertEqual(expected_destination, module_data['parameters']['destination'])
self.assertEqual(version, module_data['parameters']['protocol'])
self.assertEqual(active_timeout, module_data['parameters']['active_timeout'])
self.assertEqual(
inactive_timeout, module_data['parameters']['inactive_timeout']
)
# Test module reload with new parameters
engine_id = '73'
self.cli_set(base_path + ['netflow', 'engine-id', engine_id])
self.cli_commit()
module_data = get_module_data(module_name)
self.assertEqual(engine_id, module_data['parameters']['engine_id'])
self.cli_delete(['interfaces', 'dummy', dummy_if])
def test_iptables(self):
netflow_server = '11.22.33.44'
self.cli_set(base_path + ['netflow', 'server', netflow_server])
dummy_ifs = [
'dum4000',
'dum4001',
'dum4002',
'dum4003',
]
self.cli_set(['interfaces', 'dummy', dummy_ifs[0], 'address', '192.0.2.100/32'])
self.cli_set(['interfaces', 'dummy', dummy_ifs[1], 'address', '192.0.2.101/32'])
self.cli_set(['interfaces', 'dummy', dummy_ifs[2], 'address', '192.0.2.102/32'])
self.cli_set(['interfaces', 'dummy', dummy_ifs[3], 'address', '192.0.2.103/32'])
# * three interfaces
for i in range(3):
self.cli_set(base_path + ['netflow', 'interface', dummy_ifs[i]])
self.cli_commit()
self._assert_ingress_interfaces(dummy_ifs[0:3])
self._assert_egress_interfaces([])
# * Then delete one
self.cli_delete(base_path + ['netflow', 'interface', dummy_ifs[1]])
self.cli_commit()
self._assert_ingress_interfaces([dummy_ifs[0], dummy_ifs[2]])
self._assert_egress_interfaces([])
# * Then add one
self.cli_set(base_path + ['netflow', 'interface', dummy_ifs[3]])
self.cli_commit()
self._assert_ingress_interfaces([dummy_ifs[0], dummy_ifs[2], dummy_ifs[3]])
self._assert_egress_interfaces([])
# * enable egress
self.cli_set(base_path + ['enable-egress'])
self.cli_commit()
self._assert_ingress_interfaces([dummy_ifs[0], dummy_ifs[2], dummy_ifs[3]])
self._assert_egress_interfaces([dummy_ifs[0], dummy_ifs[2], dummy_ifs[3]])
def test_sampler(self):
# Separate test because if --enable-sampler is not given to configure of
# ipt_NETFLOW this parameter is not available
sampling_rate = '100'
self.cli_set(base_path + ['netflow', 'sampling-rate', sampling_rate])
for interface in Section.interfaces('ethernet'):
self.cli_set(base_path + ['netflow', 'interface', interface])
netflow_server = '11.22.33.44'
self.cli_set(base_path + ['netflow', 'server', netflow_server])
# commit changes, this time should work
self.cli_commit()
module_data = get_module_data(module_name)
if 'sampler' not in module_data['parameters']:
self.skipTest("ipt_NETFLOW has no sampler parameter")
self.assertEqual(
f'random:{sampling_rate}', module_data['parameters']['sampler']
)
+ def test_netflow_v5(self):
+ version = '5'
+ netflow_server = '203.0.113.27'
+
+ for interface in Section.interfaces('ethernet'):
+ self.cli_set(base_path + ['netflow', 'interface', interface])
+
+ self.cli_set(base_path + ['netflow', 'version', version])
+ self.cli_set(base_path + ['netflow', 'server', netflow_server])
+
+ # commit changes
+ self.cli_commit()
+
+
if __name__ == '__main__':
unittest.main(verbosity=2, failfast=VyOSUnitTestSHIM.TestCase.debug_on())

File Metadata

Mime Type
text/x-diff
Expires
Sat, Oct 17, 12:49 AM (19 h, 27 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
4416719
Default Alt Text
(11 KB)

Event Timeline