From cbb6c944fea616547cec43f7f1ed6ea3cc4beb54 Mon Sep 17 00:00:00 2001 From: Christian Breunig Date: Tue, 22 Apr 2025 16:31:09 +0200 Subject: vyos.utils: T7122: fix IPv6 support in check_port_availability() Commit 4523e9c897b3 ("wireguard: T3763: Added check for listening port availability") added a function to check if a port is free to use or already occupied by a different running service. This has been done by trying to bind a socket to said given port. Unfortunately there is no support for IPv6 address-fdamily in both socketserver.TCPServer or socketserver.UDPServer. This must be done manually by deriving TCPServer and setting self.address_family for IPv6. The new implementation gets rid of both TCPServer and UDPServer and replaces it with a simple socket binding to a given IPv4/IPv6 address or any interface/ address if unspecified. In addition build time tests are added for the function to check for proper behavior during build time of vyos-1x. --- src/tests/test_utils_network.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) (limited to 'src/tests') diff --git a/src/tests/test_utils_network.py b/src/tests/test_utils_network.py index d68dec16f..92fde447d 100644 --- a/src/tests/test_utils_network.py +++ b/src/tests/test_utils_network.py @@ -1,4 +1,4 @@ -# Copyright (C) 2020-2024 VyOS maintainers and contributors +# Copyright (C) 2020-2025 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 @@ -43,3 +43,12 @@ class TestVyOSUtilsNetwork(TestCase): self.assertFalse(vyos.utils.network.is_loopback_addr('::2')) self.assertFalse(vyos.utils.network.is_loopback_addr('192.0.2.1')) + + def test_check_port_availability(self): + self.assertTrue(vyos.utils.network.check_port_availability('::1', 8080)) + self.assertTrue(vyos.utils.network.check_port_availability('127.0.0.1', 8080)) + self.assertTrue(vyos.utils.network.check_port_availability(None, 8080, protocol='udp')) + # We do not have 192.0.2.1 configured on this system + self.assertFalse(vyos.utils.network.check_port_availability('192.0.2.1', 443)) + # We do not have 2001:db8::1 configured on this system + self.assertFalse(vyos.utils.network.check_port_availability('2001:db8::1', 80, protocol='udp')) -- cgit v1.2.3 From aff2835d7b6226e4b89f51e3f6133da26f3a07bf Mon Sep 17 00:00:00 2001 From: Christian Breunig Date: Sun, 4 May 2025 11:23:54 +0200 Subject: vyos.template: T7122: add Jinja2 clever function helper to read vyos.defaults Add a new category if Jinja2 operands. We already have filters and tests, but sometimes we would like to call a Python function without and data "|" piped to it - that's what they call a clever-function. {{ get_default_port(NAME) }} can be used to retrieve the value from vyos.defaults.internal_ports[NAME] within Jinja2. We no longer need to extend the dictionary with arbitrary data retrieved from vyos.defaults, we can now simply register another clever-function to the Jinja2 backend. --- python/vyos/template.py | 44 ++++++++++++++++++++++++++++++++++++++++++-- src/tests/test_template.py | 9 +++++++++ 2 files changed, 51 insertions(+), 2 deletions(-) (limited to 'src/tests') diff --git a/python/vyos/template.py b/python/vyos/template.py index 513490963..11e1cc50f 100755 --- a/python/vyos/template.py +++ b/python/vyos/template.py @@ -36,6 +36,7 @@ DEFAULT_TEMPLATE_DIR = directories["templates"] # Holds template filters registered via register_filter() _FILTERS = {} _TESTS = {} +_CLEVER_FUNCTIONS = {} # reuse Environments with identical settings to improve performance @functools.lru_cache(maxsize=2) @@ -58,6 +59,7 @@ def _get_environment(location=None): ) env.filters.update(_FILTERS) env.tests.update(_TESTS) + env.globals.update(_CLEVER_FUNCTIONS) return env @@ -77,7 +79,7 @@ def register_filter(name, func=None): "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") + raise ValueError(f"A filter with name {name!r} was already registered") _FILTERS[name] = func return func @@ -97,10 +99,30 @@ def register_test(name, func=None): "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") + raise ValueError(f"A test with name {name!r} was already registered") _TESTS[name] = func return func +def register_clever_function(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_clever_function, name) + if _get_environment.cache_info().currsize: + raise RuntimeError( + "Clever functions can only be registered before rendering the" \ + "first template") + if name in _CLEVER_FUNCTIONS: + raise ValueError(f"A clever function with name {name!r} was already "\ + "registered") + _CLEVER_FUNCTIONS[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. @@ -1052,3 +1074,21 @@ def vyos_defined(value, test_value=None, var_type=None): else: # Valid value and is matching optional argument if provided - return true return True + +@register_clever_function('get_default_port') +def get_default_port(service): + """ + Jinja2 plugin to retrieve common service port number from vyos.defaults + class form a Jinja2 template. This removes the need to hardcode, or pass in + the data using the general dictionary. + + Added to remove code complexity and make it easier to read. + + Example: + {{ get_default_port('certbot_haproxy') }} + """ + from vyos.defaults import internal_ports + if service not in internal_ports: + raise RuntimeError(f'Service "{service}" not found in internal ' \ + 'vyos.defaults.internal_ports dict!') + return internal_ports[service] diff --git a/src/tests/test_template.py b/src/tests/test_template.py index 6377f6da5..7cae867a0 100644 --- a/src/tests/test_template.py +++ b/src/tests/test_template.py @@ -190,3 +190,12 @@ class TestVyOSTemplate(TestCase): for group_name, group_config in data['ike_group'].items(): ciphers = vyos.template.get_esp_ike_cipher(group_config) self.assertIn(IKEv2_DEFAULT, ','.join(ciphers)) + + def test_get_default_port(self): + from vyos.defaults import internal_ports + + with self.assertRaises(RuntimeError): + vyos.template.get_default_port('UNKNOWN') + + self.assertEqual(vyos.template.get_default_port('certbot_haproxy'), + internal_ports['certbot_haproxy']) -- cgit v1.2.3