feat: added functions

Added index for policies
Added store for policies and rules
Added tests
Added ValidationError Exception

src/nethsec/mwan/__init__.py

@@ -1,14 +1,18 @@
 #!/usr/bin/python3
+import json
+import subprocess
+
+from euci import EUci
+
+from nethsec import utils
+from nethsec.utils import ValidationError
+
 
 #
 # Copyright (C) 2023 Nethesis S.r.l.
 # SPDX-License-Identifier: GPL-2.0-only
 #
 
-from euci import EUci
-
-from nethsec import utils
-
 
 def __generate_metric(e_uci: EUci, interface_metrics: list[int] = None, metric: int = 1) -> int:
     """
@@ -45,12 +49,12 @@ def __store_interface(e_uci: EUci, name: str) -> tuple[bool, bool]:
         to network interface
 
     Raises:
-        ValueError: if interface name is not defined in /etc/config/network
+        ValidationError: if interface name is not defined in /etc/config/network
     """
     # checking if interface is configured
     available_interfaces = utils.get_all_by_type(e_uci, 'network', 'interface')
     if name not in available_interfaces.keys():
-        raise ValueError(name, 'invalid')
+        raise ValidationError('name', 'invalid', name)
 
     created_interface = False
     # if no interface with name exists, create one with defaults
@@ -110,34 +114,81 @@ def __store_member(e_uci: EUci, interface_name: str, metric: int, weight: int) -
     return member_config_name, changed
 
 
+def store_rule(e_uci: EUci, name: str, policy: str, protocol: str = None,
+               source_addresses: str = None, source_ports: str = None,
+               destination_addresses: str = None, destination_ports: str = None) -> str:
+    """
+    Stores a rule for mwan3
+    Args:
+        e_uci: EUci instance
+        name: name of the rule, must be unique
+        policy: policy to use for the rule, must be already set
+        protocol: must be one of tcp, udp, icmp or all, defaults to 'all'
+        source_addresses: source addresses to match
+        source_ports: source ports to match or range
+        destination_addresses: destination addresses to match
+        destination_ports: destination ports to match or range
+
+    Returns:
+        name of the rule created
+
+    Raises:
+        ValidationError if name is not unique or policy is not valid
+    """
+    rule_config_name = utils.get_id(name, 15)
+    if rule_config_name in e_uci.get('mwan3').keys():
+        raise ValidationError('name', 'unique', name)
+    if policy not in utils.get_all_by_type(e_uci, 'mwan3', 'policy').keys():
+        raise ValidationError('policy', 'invalid', policy)
+    e_uci.set('mwan3', rule_config_name, 'rule')
+    e_uci.set('mwan3', rule_config_name, 'label', name)
+    e_uci.set('mwan3', rule_config_name, 'use_policy', policy)
+    if protocol is not None:
+        e_uci.set('mwan3', rule_config_name, 'proto', protocol)
+    if source_addresses is not None:
+        e_uci.set('mwan3', rule_config_name, 'src_ip', source_addresses)
+    if source_ports is not None:
+        e_uci.set('mwan3', rule_config_name, 'src_port', source_ports)
+    if destination_addresses is not None:
+        e_uci.set('mwan3', rule_config_name, 'dest_ip', destination_addresses)
+    if destination_ports is not None:
+        e_uci.set('mwan3', rule_config_name, 'dest_port', destination_ports)
+
+    e_uci.save('mwan3')
+    return f'mwan3.{rule_config_name}'
+
+
 def store_policy(e_uci: EUci, name: str, interfaces: list[dict]) -> list[str]:
     """
     Stores a policy for mwan3, takes care of creating interfaces and members.
     Args:
         e_uci: EUci instance
         name: name of policy
-        interfaces: list of interfaces to add to policy, must have a name, metric and weight fields
+        interfaces: list of interfaces to add to policy, must have name, metric and weight fields
 
     Returns:
         list of changed configuration
 
     Raises:
-        ValueError: if name is not unique
+        ValidationError: if name is not unique
     """
     changed_config = []
     # generate policy name
     policy_config_name = utils.get_id(name)
     # make sure name is not something that already exists
     if policy_config_name in e_uci.get('mwan3').keys():
-        raise ValueError(name, 'unique')
+        raise ValidationError('name', 'unique', name)
     # generate policy config with corresponding name
     e_uci.set('mwan3', policy_config_name, 'policy')
-    e_uci.set('mwan3', policy_config_name, 'name', name)
+    e_uci.set('mwan3', policy_config_name, 'label', name)
     changed_config.append(f'mwan3.{policy_config_name}')
 
     member_names: list[str] = []
     for interface in interfaces:
-        added_mwan_interface, updated_interface = __store_interface(e_uci, interface['name'])
+        try:
+            added_mwan_interface, updated_interface = __store_interface(e_uci, interface['name'])
+        except ValidationError:
+            raise ValidationError('interfaces', 'invalid', interface['name'])
         if added_mwan_interface:
             changed_config.append(f'mwan3.{interface["name"]}')
         if updated_interface:
@@ -153,6 +204,90 @@ def store_policy(e_uci: EUci, name: str, interfaces: list[dict]) -> list[str]:
 
     e_uci.set('mwan3', policy_config_name, 'use_member', member_names)
 
+    if len(utils.get_all_by_type(e_uci, 'mwan3', 'rule')) == 0:
+        changed_config.append(store_rule(e_uci, 'Default Rule', policy_config_name))
+
     e_uci.save('mwan3')
     e_uci.save('network')
     return changed_config
+
+
+def __fetch_interface_status(interface_name: str) -> str:
+    try:
+        output = subprocess.check_output([
+            'ubus',
+            'call',
+            'mwan3',
+            'status',
+            '{"section": "interfaces"}'
+        ]).decode('utf-8')
+        decoded_output = json.JSONDecoder().decode(output)
+        return decoded_output['interfaces'][interface_name]['status']
+    except:
+        return 'unknown'
+
+
+def __parse_member(e_uci: EUci, member_name: str) -> dict:
+    """
+    Parses a member configuration and returns formatted data.
+    Args:
+        e_uci: EUci instance
+        member_name: member name
+
+    Returns:
+        dict with member data
+    """
+    member_data = e_uci.get_all('mwan3', member_name)
+    return {
+        'name': member_name,
+        'interface': member_data['interface'],
+        'metric': member_data['metric'],
+        'weight': member_data['weight'],
+        'status': __fetch_interface_status(member_data['interface'])
+    }
+
+
+def index_policies(e_uci: EUci) -> list[dict]:
+    """
+    Returns a list of policies with their members, interfaces and metrics/weights.
+    Args:
+        e_uci: EUci instance
+
+    Returns:
+        list of dicts with policy data
+    """
+    data = []
+    policies = utils.get_all_by_type(e_uci, 'mwan3', 'policy')
+    # iterate over policies
+    for policy_name in policies.keys():
+        policy = policies[policy_name]
+        policy_data = {
+            'name': policy_name,
+        }
+        # add label only if present
+        if 'label' in policy:
+            policy_data['label'] = policy['label']
+
+        # add members
+        members = []
+        if 'use_member' in policy:
+            policy_data['members'] = {}
+            for member in policy['use_member']:
+                members.append(__parse_member(e_uci, member))
+
+        # infer policy type by metrics
+        metrics = [int(member['metric']) for member in members]
+        if all(metric == metrics[0] for metric in metrics):
+            policy_data['type'] = 'balance'
+        elif all(metrics.index(metric) == key for key, metric in enumerate(metrics)):
+            policy_data['type'] = 'backup'
+        else:
+            policy_data['type'] = 'custom'
+
+        unique_metrics = list(set(metrics))
+        for unique_metric in unique_metrics:
+            policy_data['members'][unique_metric] = list(filter(lambda x: x['metric'] == str(unique_metric), members))
+
+        # append policy to data
+        data.append(policy_data)
+    return data

src/nethsec/utils/__init__.py

@@ -370,3 +370,10 @@ def generic_error(error):
       - A validation error object
     '''
     return {"error": error.strip().replace(" ", "_").lower()}
+
+
+class ValidationError(ValueError):
+    def __init__(self, parameter, message="", value=""):
+        self.parameter = parameter
+        self.message = message
+        self.value = value

tests/test_mwan.py

@@ -119,8 +119,9 @@ def test_create_interface(e_uci):
 
 
 def test_fail_create_invalid_interface(e_uci):
-    with pytest.raises(ValueError):
+    with pytest.raises(ValueError) as err:
         mwan.__store_interface(e_uci, 'RED_4')
+        assert err.value.args[0] == ('RED_4', 'invalid')
 
 
 def test_interface_avoid_edit_of_metric(e_uci):
@@ -152,10 +153,11 @@ def test_create_default_mwan(e_uci):
            'mwan3.ns_RED_1_M10_W200',
            'mwan3.RED_2',
            'network.RED_2',
-           'mwan3.ns_RED_2_M20_W100']
+           'mwan3.ns_RED_2_M20_W100',
+           'mwan3.ns_Default_Rule']
 
     assert e_uci.get('mwan3', 'ns_default') == 'policy'
-    assert e_uci.get('mwan3', 'ns_default', 'name') == 'default'
+    assert e_uci.get('mwan3', 'ns_default', 'label') == 'default'
     assert e_uci.get('mwan3', 'ns_default', 'use_member', list=True) == (
         'ns_RED_1_M10_W200', 'ns_RED_2_M20_W100')
 
@@ -171,3 +173,117 @@ def test_metric_generation(e_uci):
     assert mwan.__generate_metric(e_uci, [1, 4]) == 2
     assert mwan.__generate_metric(e_uci, [1, 2, 4]) == 3
     assert mwan.__generate_metric(e_uci, [4, 3, 1]) == 2
+
+
+def test_list_policies(e_uci):
+    mwan.store_policy(e_uci, 'backup', [
+        {
+            'name': 'RED_1',
+            'metric': '10',
+            'weight': '200',
+        },
+        {
+            'name': 'RED_2',
+            'metric': '20',
+            'weight': '100',
+        }
+    ])
+    mwan.store_policy(e_uci, 'balance', [
+        {
+            'name': 'RED_3',
+            'metric': '10',
+            'weight': '200',
+        },
+        {
+            'name': 'RED_2',
+            'metric': '10',
+            'weight': '100',
+        }
+    ])
+    mwan.store_policy(e_uci, 'custom', [
+        {
+            'name': 'RED_3',
+            'metric': '10',
+            'weight': '200',
+        },
+        {
+            'name': 'RED_2',
+            'metric': '10',
+            'weight': '100',
+        },
+        {
+            'name': 'RED_1',
+            'metric': '20',
+            'weight': '100',
+        }
+    ])
+    index = mwan.index_policies(e_uci)
+    # check backup policy
+    assert index[0]['name'] == 'ns_backup'
+    assert index[0]['label'] == 'backup'
+    assert index[0]['type'] == 'backup'
+    assert index[0]['members'][10][0]['name'] == 'ns_RED_1_M10_W200'
+    assert index[0]['members'][10][0]['interface'] == 'RED_1'
+    assert index[0]['members'][10][0]['metric'] == '10'
+    assert index[0]['members'][10][0]['weight'] == '200'
+    assert index[0]['members'][20][0]['name'] == 'ns_RED_2_M20_W100'
+    assert index[0]['members'][20][0]['interface'] == 'RED_2'
+    assert index[0]['members'][20][0]['metric'] == '20'
+    assert index[0]['members'][20][0]['weight'] == '100'
+    # check balance policy
+    assert index[1]['name'] == 'ns_balance'
+    assert index[1]['label'] == 'balance'
+    assert index[1]['type'] == 'balance'
+    assert index[1]['members'][10][0]['name'] == 'ns_RED_3_M10_W200'
+    assert index[1]['members'][10][0]['interface'] == 'RED_3'
+    assert index[1]['members'][10][0]['metric'] == '10'
+    assert index[1]['members'][10][0]['weight'] == '200'
+    assert index[1]['members'][10][1]['name'] == 'ns_RED_2_M10_W100'
+    assert index[1]['members'][10][1]['interface'] == 'RED_2'
+    assert index[1]['members'][10][1]['metric'] == '10'
+    assert index[1]['members'][10][1]['weight'] == '100'
+    # check custom policy
+    assert index[2]['name'] == 'ns_custom'
+    assert index[2]['label'] == 'custom'
+    assert index[2]['type'] == 'custom'
+    assert index[2]['members'][10][0]['name'] == 'ns_RED_3_M10_W200'
+    assert index[2]['members'][10][0]['interface'] == 'RED_3'
+    assert index[2]['members'][10][0]['metric'] == '10'
+    assert index[2]['members'][10][0]['weight'] == '200'
+    assert index[2]['members'][10][1]['name'] == 'ns_RED_2_M10_W100'
+    assert index[2]['members'][10][1]['interface'] == 'RED_2'
+    assert index[2]['members'][10][1]['metric'] == '10'
+    assert index[2]['members'][10][1]['weight'] == '100'
+    assert index[2]['members'][20][0]['name'] == 'ns_RED_1_M20_W100'
+    assert index[2]['members'][20][0]['interface'] == 'RED_1'
+    assert index[2]['members'][20][0]['metric'] == '20'
+    assert index[2]['members'][20][0]['weight'] == '100'
+
+
+def test_store_rule(e_uci):
+    mwan.store_policy(e_uci, 'default', [
+        {
+            'name': 'RED_1',
+            'metric': '20',
+            'weight': '100',
+        }
+    ])
+    assert mwan.store_rule(e_uci, 'additional rule', 'ns_default') == 'mwan3.ns_additional_r'
+    assert e_uci.get('mwan3', 'ns_additional_r') == 'rule'
+    assert e_uci.get('mwan3', 'ns_additional_r', 'label') == 'additional rule'
+    assert e_uci.get('mwan3', 'ns_additional_r', 'use_policy') == 'ns_default'
+
+
+def test_unique_rule(e_uci):
+    mwan.store_policy(e_uci, 'default', [
+        {
+            'name': 'RED_1',
+            'metric': '20',
+            'weight': '100',
+        }
+    ])
+    with pytest.raises(ValueError) as e:
+        mwan.store_rule(e_uci, 'additional rule', 'ns_default')
+        mwan.store_rule(e_uci, 'additional rule', 'ns_default')
+        assert e.value.args[0] == 'name'
+        assert e.value.args[1] == 'invalid'