diff options
Diffstat (limited to 'smoketest/scripts/cli/test_service_https.py')
| -rwxr-xr-x | smoketest/scripts/cli/test_service_https.py | 231 |
1 files changed, 214 insertions, 17 deletions
diff --git a/smoketest/scripts/cli/test_service_https.py b/smoketest/scripts/cli/test_service_https.py index 04c4a2e51..b29be9eda 100755 --- a/smoketest/scripts/cli/test_service_https.py +++ b/smoketest/scripts/cli/test_service_https.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2019-2024 VyOS maintainers and contributors +# 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 @@ -16,6 +16,8 @@ import unittest import json +import psutil +import time from requests import request from urllib3.exceptions import InsecureRequestWarning @@ -26,6 +28,7 @@ from base_vyostest_shim import ignore_warning from vyos.utils.file import read_file from vyos.utils.file import write_file from vyos.utils.process import call +from vyos.utils.process import cmd from vyos.utils.process import process_named_running from vyos.xml_ref import default_value @@ -34,6 +37,9 @@ from vyos.configsession import ConfigSessionError base_path = ['service', 'https'] pki_base = ['pki'] +address = '127.0.0.1' +key = 'VyOS-key' + cert_data = """ MIICFDCCAbugAwIBAgIUfMbIsB/ozMXijYgUYG80T1ry+mcwCgYIKoZIzj0EAwIw WTELMAkGA1UEBhMCR0IxEzARBgNVBAgMClNvbWUtU3RhdGUxEjAQBgNVBAcMCVNv @@ -112,6 +118,94 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): # Check for stopped process self.assertFalse(process_named_running(PROCESS_NAME)) + # always forward to base class + super().tearDown() + + def _api_get_background_operations(self): + url = f'https://{address}/retrieve/background-operations' + r = request('POST', url, verify=False, json={'key': key}) + self.assertEqual(r.status_code, 200) + body = r.json() + self.assertTrue(body.get('success')) + ops = body.get('data', {}).get('operations', []) + return ops + + def _wait_no_active_operations(self, timeout: int = 30): + # wait until no queued/running operations remain + deadline = time.time() + timeout + statuses = ('queued', 'running') + + while time.time() < deadline: + ops = self._api_get_background_operations() + ops = [op for op in ops if op.get('status') in statuses] + if not ops: + return + sleep(0.25) + self.fail('Timeout waiting for background operations to finish') + + def assertBackgroundOpResponseIsOk(self, response): + self.assertEqual(response.status_code, 200) + body = response.json() + self.assertTrue(body.get('success')) + + data = body.get('data', {}) + self.assertIsInstance(data, dict) + self.assertIn('operation', data) + op = data['operation'] + self.assertIn('op_id', op) + self.assertIn('status', op) + + def test_listen_address(self): + test_prefix = ['192.0.2.1/26', '2001:db8:1::ffff/64'] + test_addr = [ i.split('/')[0] for i in test_prefix ] + for i, addr in enumerate(test_prefix): + self.cli_set(['interfaces', 'dummy', f'dum{i}', 'address', addr]) + + key = 'MySuperSecretVyOS' + self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) + # commit base config first, for testing update of listen-address + self.cli_commit() + + for addr in test_addr: + self.cli_set(base_path + ['listen-address', addr]) + self.cli_commit() + + res = set() + t = psutil.net_connections(kind="tcp") + for c in t: + if c.laddr.port == 443: + res.add(c.laddr.ip) + + self.assertEqual(res, set(test_addr)) + + def test_listen_address_vrf(self): + # Verify that HTTPS service can be configured with a listen-address + # inside a VRF. Regression test: the port availability check used to + # fail because it ran in the default namespace where the VRF address + # is unreachable. + vrf = 'mgmt' + vrf_table = '1337' + test_addr = '192.0.2.1' + test_prefix = f'{test_addr}/26' + interface = 'dum0' + + self.cli_set(['interfaces', 'dummy', interface, 'address', test_prefix]) + self.cli_set(['interfaces', 'dummy', interface, 'vrf', vrf]) + self.cli_set(['vrf', 'name', vrf, 'table', vrf_table]) + + self.cli_set( + base_path + ['api', 'keys', 'id', 'key-01', 'key', 'MySuperSecretVyOS'] + ) + self.cli_set(base_path + ['listen-address', test_addr]) + self.cli_set(base_path + ['vrf', vrf]) + self.cli_commit() + + # Verify nginx is running inside the VRF + tmp = cmd(f'ip vrf pids {vrf}') + self.assertIn(PROCESS_NAME, tmp) + + self.cli_delete(['interfaces', 'dummy', interface]) + self.cli_delete(['vrf', 'name', vrf]) def test_certificate(self): cert_name = 'test_https' @@ -330,8 +424,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_add_delete(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/retrieve' payload = {'data': '{"op": "showConfig", "path": []}', 'key': f'{key}'} headers = {} @@ -361,8 +453,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_show(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/show' headers = {} @@ -379,8 +469,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_generate(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/generate' headers = {} @@ -397,8 +485,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_configure(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/configure' headers = {} conf_interface = 'dum0' @@ -423,8 +509,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_config_file(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/config-file' headers = {} @@ -441,8 +525,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_reset(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/reset' headers = {} @@ -459,8 +541,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_image(self): - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/image' headers = {} @@ -502,8 +582,6 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): @ignore_warning(InsecureRequestWarning) def test_api_config_file_load_http(self): # Test load config from HTTP URL - address = '127.0.0.1' - key = 'VyOS-key' url = f'https://{address}/config-file' url_config = f'https://{address}/configure' headers = {} @@ -546,6 +624,125 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase): call(f'sudo rm -f {nginx_tmp_site}') call('sudo systemctl reload nginx') + @ignore_warning(InsecureRequestWarning) + def test_api_configure_background(self): + url = f'https://{address}/configure' + conf_interface = 'dum8' + conf_address = '192.0.2.88/32' + + # Enable REST API + self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) + self.cli_set(base_path + ['api', 'rest']) + self.cli_commit() + + payload_path = [ + 'interfaces', + 'dummy', + conf_interface, + 'address', + ] + params = {'in_background': True} + payload = { + 'data': json.dumps( + {'op': 'set', 'path': payload_path, 'value': conf_address} + ), + 'key': key, + } + + r = request('POST', url, verify=False, params=params, data=payload) + self.assertBackgroundOpResponseIsOk(r) + body = r.json() + op = body.get('data', {}).get('operation', []) + + # Operation should appear as active shortly + ops = self._api_get_background_operations() + self.assertTrue( + any(o.get('op_id') == op.get('op_id') for o in ops), + 'Queued operation is not visible in `/retrieve/background-operations`', + ) + + # Wait until done + self._wait_no_active_operations() + + # Verify config applied (using CLI show) + self.assertIn(conf_address, self.op_mode(['show', 'configuration', 'commands'])) + + @ignore_warning(InsecureRequestWarning) + def test_api_configure_section_background(self): + url = f'https://{address}/configure-section' + conf_interface = 'dum9' + conf_address = '192.0.2.99/32' + + # Enable REST API + self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) + self.cli_set(base_path + ['api', 'rest']) + self.cli_commit() + + # Configure-section payload: set a full section + # example: set interfaces dummy dum8 address 192.0.2.99/32 + payload = { + 'data': json.dumps( + { + 'op': 'set', + 'path': ['interfaces', 'dummy', conf_interface], + 'section': { + 'address': [conf_address], + }, + } + ), + 'key': key, + } + params = {'in_background': True} + + r = request('POST', url, verify=False, params=params, data=payload) + self.assertBackgroundOpResponseIsOk(r) + + # Wait until done + self._wait_no_active_operations() + + # Verify section applied + self.assertIn(conf_address, self.op_mode(['show', 'configuration', 'commands'])) + + @ignore_warning(InsecureRequestWarning) + def test_api_configure_background_ops_over_max(self): + max_ops = 128 + + # Enable REST API + self.cli_set(base_path + ['api', 'keys', 'id', 'key-01', 'key', key]) + self.cli_set(base_path + ['api', 'rest']) + self.cli_commit() + + op_ids = [] + params = {'in_background': True} + url = f'https://{address}/configure' + + # Create many non-existent configurations to fill the queue. + for i in range(max_ops + 5): + config_name = f'invalid-test-option-{i}' + payload_path = ['system', config_name] + payload = { + 'data': json.dumps( + {'op': 'set', 'path': payload_path, 'value': config_name} + ), + 'key': key, + } + + with self.subTest(payload_path=payload_path): + r = request('POST', url, verify=False, params=params, data=payload) + self.assertBackgroundOpResponseIsOk(r) + + body = r.json() + op = body.get('data', {}).get('operation', []) + op_ids.append(op) + + # Wait for queue to drain + self._wait_no_active_operations(timeout=120) + + # Verify pruning: oldest `op_id` should be absent, and count should be <= `max_ops` + ops = self._api_get_background_operations() + self.assertLessEqual(len(ops), max_ops) + self.assertFalse(any(o.get('op_id') == op_ids[0] for o in ops)) + if __name__ == '__main__': unittest.main(verbosity=5) |