feat(firewall): add support for source MAC addresses in firewall rules

packages/ns-api/README.md

@@ -299,6 +299,7 @@ Response:
           "type": null
         }
       ],
+      "src_mac": [],
       "dest_ip": [
         {
           "value": "192.168.122.1",
@@ -328,6 +329,8 @@ Response:
 }
 ```
 
+`src_mac` is a plain list of source MAC addresses, empty if the rule does not match by MAC.
+
 ### list-output-rules
 
 List output rules in order:
@@ -606,7 +609,7 @@ Response:
 
 Add a rule:
 ```
-api-cli ns.firewall add-rule --data '{"name": "r1", "src": "lan", "src_ip": [], "dest": "wan", "dest_ip": ["1.2.3.4"], "proto": [], "dest_port": ", "target": "ACCEPT", "ns_service": "ssh", "enabled": true, "log": false, "ns_tag": [], "add_to_top": false}'
+api-cli ns.firewall add-rule --data '{"name": "r1", "src": "lan", "src_ip": [], "dest": "wan", "dest_ip": ["1.2.3.4"], "proto": [], "dest_port": ", "target": "ACCEPT", "ns_service": "ssh", "enabled": true, "log": false, "ns_tag": [], "add_to_top": false, "src_mac": []}'
 ```
 
 Response example:
@@ -616,8 +619,15 @@ Response example:
 
 The `proto` and `dest_port` field will be validated and saved only if `ns_service` is set to `custom`.
 
+`src_mac` is an optional list of lowercase, colon-separated source MAC addresses (default `[]`).
+It is mutually exclusive with `src_ip` and `ns_src`, and it is not allowed on output rules (empty `src`).
+When empty, the `src_mac` option is removed from the rule.
+
 Possible validation errors:
 - `invalid_format` for `dest_ip` and `src_ip`
+- `error.invalid_mac_address` for `src_mac`, if an address does not match `^([0-9a-f]{2}:){5}[0-9a-f]{2}$`
+- `error.src_mac_conflict` for `src_mac`, if `src_ip` or `ns_src` is also set
+- `error.src_mac_not_allowed` for `src_mac`, if the rule is an output rule
 - `invalid_target` for `target`
 - if `ns_service` is `custom`: `invalid_proto` for `proto` and `invalid_port` for `dest_port`
 - if `ns_service` is not `cutom` or `*`: `invalid_service` for `ns_service
@@ -626,7 +636,7 @@ Possible validation errors:
 
 Edit an existing rule:
 ```
-api-cli ns.firewall edit-rule --data '{"id": "ns_206325d3", "name": "r1", "src": "lan", "src_ip": [], "dest": "wan", "dest_ip": ["1.2.3.4"], "proto": ["tcp"], "dest_port": "22", "target": "ACCEPT", "ns_service": "ssh", "enabled": true, "log": false, "ns_tag": []}'
+api-cli ns.firewall edit-rule --data '{"id": "ns_206325d3", "name": "r1", "src": "lan", "src_ip": [], "dest": "wan", "dest_ip": ["1.2.3.4"], "proto": ["tcp"], "dest_port": "22", "target": "ACCEPT", "ns_service": "ssh", "enabled": true, "log": false, "ns_tag": [], "src_mac": []}'
 ```
 
 Response example:

packages/ns-api/files/ns.firewall

@@ -42,7 +42,8 @@ if cmd == 'list':
             "add_to_top": False,
             "ns_src": "database/id",
             "ns_dst": "database/id",
-            "ns_link": "database/id"
+            "ns_link": "database/id",
+            "src_mac": []
         },
         'edit-rule': {
             "id": "ns_1234",
@@ -61,7 +62,8 @@ if cmd == 'list':
             "add_to_top": False,
             "ns_src": "database/id",
             "ns_dst": "database/id",
-            "ns_link": "database/id"
+            "ns_link": "database/id",
+            "src_mac": []
         },
         'delete-rule': { "id": "ns_1234" },
         'enable-rule': { "id": "ns_1234" },
@@ -124,7 +126,8 @@ elif cmd == 'call':
                     data['add_to_top'],
                     data.get('ns_src'),
                     data.get('ns_dst'),
-                    data.get('ns_link')
+                    data.get('ns_link'),
+                    data.get('src_mac', [])
             )
             print(json.dumps({"id": ret}))
         elif action == "edit-rule":
@@ -146,7 +149,8 @@ elif cmd == 'call':
                     data['ns_tag'],
                     data.get('ns_src'),
                     data.get('ns_dst'),
-                    data.get('ns_link')
+                    data.get('ns_link'),
+                    data.get('src_mac', [])
             )
             print(json.dumps({"id": ret}))
         elif action == "delete-rule":

packages/ns-ui/Makefile

@@ -12,7 +12,7 @@ PKG_RELEASE:=1
 
 PKG_SOURCE_PROTO:=git
 PKG_SOURCE_URL:=https://github.com/NethServer/nethsecurity-ui.git
-PKG_SOURCE_VERSION:=$(PKG_VERSION)
+PKG_SOURCE_VERSION:=34e8f1baa15e7d55eedfbcb74175d327c7254e85
 PKG_SOURCE_SUBDIR:=nethsecurity-ui-$(PKG_SOURCE_VERSION)
 PKG_BUILD_DIR:=$(BUILD_DIR)/$(PKG_SOURCE_SUBDIR)
 PKG_MIRROR_HASH:=skip

packages/python3-nethsec/src/nethsec/firewall/__init__.py

@@ -1128,6 +1128,11 @@ def enrich_rule(uci, rule: dict) -> dict:
             rule['dest_ip'] = list(rule['dest_ip'])
         elif type(rule['dest_ip'] ) == str:
             rule['dest_ip'] = [rule['dest_ip']]
+    src_mac = rule.get('src_mac', [])
+    if type(src_mac) == str:
+        rule['src_mac'] = [src_mac]
+    else:
+        rule['src_mac'] = list(src_mac)
     # try to gather info on src_ip and dest_ip
     for key in ['src_ip', 'dest_ip']:
         tmp = []
@@ -1453,7 +1458,7 @@ def validate_port_format(port: str) -> bool:
         return False
     return True
 
-def validate_rule(uci, src: str, src_ip: list[str], dest: str, dest_ip: list[str], proto: list, dest_port: list[str], target: str, service: str, ns_src: str, ns_dst: str):
+def validate_rule(uci, src: str, src_ip: list[str], dest: str, dest_ip: list[str], proto: list, dest_port: list[str], target: str, service: str, ns_src: str, ns_dst: str, src_mac: list[str] = None):
     """
     Validate rule.
 
@@ -1468,10 +1473,19 @@ def validate_rule(uci, src: str, src_ip: list[str], dest: str, dest_ip: list[str
         service: service name
         ns_src: an object in the form `<database>/<id>`
         ns_dst: an object in the form `<database>/<id>`
+        src_mac: a list of source MAC addresses, lowercase and colon-separated
 
     Raises:
         ValidationError: if rule is invalid
     """
+    if src_mac:
+        if not src:
+            # output rules: the source is always the firewall itself
+            raise utils.ValidationError('src_mac', 'error.src_mac_not_allowed', ','.join(src_mac))
+        if src_ip or ns_src:
+            raise utils.ValidationError('src_mac', 'error.src_mac_conflict', ','.join(src_mac))
+        for mac in src_mac:
+            utils.validate_mac_address(mac, 'src_mac')
     if ns_src:
         if not objects.object_exists(uci, ns_src):
             raise utils.ValidationError('ns_src', 'object_not_found', ns_src)
@@ -1519,7 +1533,7 @@ def get_service_by_name(name: str) -> dict:
     return None
 
 def setup_rule(uci, id: str, name: str, src: str, src_ip: list[str], dest: str, dest_ip: list[str], proto: list, dest_port: list[str], target: str, service: str,
-                enabled: bool = True, log: bool = False, tag = [], ns_src: str = None, ns_dst: str = None, ns_link: str = None) -> None:
+                enabled: bool = True, log: bool = False, tag = [], ns_src: str = None, ns_dst: str = None, ns_link: str = None, src_mac: list[str] = None) -> None:
     """
     Set up a rule in the firewall config.
 
@@ -1541,10 +1555,18 @@ def setup_rule(uci, id: str, name: str, src: str, src_ip: list[str], dest: str,
             ns_src: an object in the form `<database>/<id>`
             ns_dst: an object in the form `<database>/<id>`
             ns_link: a string in the form `<database>/<id>` that is used to link this rule to another entity
+            src_mac: a list of source MAC addresses
     """
     uci.set('firewall', id, 'name', name)
     uci.set('firewall', id, 'src', src)
     uci.set('firewall', id, 'src_ip', src_ip)
+    if src_mac:
+        uci.set('firewall', id, 'src_mac', src_mac)
+    else:
+        try:
+            uci.delete('firewall', id, 'src_mac')
+        except:
+            pass
     uci.set('firewall', id, 'dest', dest)
     uci.set('firewall', id, 'dest_ip', dest_ip)
 
@@ -1666,7 +1688,7 @@ def reorder_firewall_config(uci):
     uci.save('firewall')
 
 def add_rule(uci, name: str, src: str, src_ip: list[str], dest: str, dest_ip: list[str], proto: list, dest_port: list[str], target: str, service: str,
-            enabled: bool = True, log: bool = False, tag = [], add_to_top: bool = False, ns_src: str = None, ns_dst: str = None, ns_link: str = None) -> str:
+            enabled: bool = True, log: bool = False, tag = [], add_to_top: bool = False, ns_src: str = None, ns_dst: str = None, ns_link: str = None, src_mac: list[str] = None) -> str:
     """
     Add rule to firewall config.
 
@@ -1689,14 +1711,15 @@ def add_rule(uci, name: str, src: str, src_ip: list[str], dest: str, dest_ip: li
         ns_src: an object in the form `<database>/<id>`
         ns_dst: an object in the form `<database>/<id>`
         ns_link: a string in the form `<database>/<id>` that is used to link this rule to another entity
+        src_mac: a list of source MAC addresses, lowercase and colon-separated
 
     Returns:
         name of rule config that was added
     """
-    validate_rule(uci, src, src_ip, dest, dest_ip, proto, dest_port, target, service, ns_src, ns_dst)
+    validate_rule(uci, src, src_ip, dest, dest_ip, proto, dest_port, target, service, ns_src, ns_dst, src_mac)
     rule = utils.get_random_id()
     uci.set('firewall', rule, 'rule')
-    setup_rule(uci, rule, name, src, src_ip, dest, dest_ip, proto, dest_port, target, service, enabled, log, tag, ns_src, ns_dst, ns_link)
+    setup_rule(uci, rule, name, src, src_ip, dest, dest_ip, proto, dest_port, target, service, enabled, log, tag, ns_src, ns_dst, ns_link, src_mac)
     reorder_firewall_config(uci)
     update_firewall_rules(uci) # expand objects and save
 
@@ -1714,7 +1737,7 @@ def add_rule(uci, name: str, src: str, src_ip: list[str], dest: str, dest_ip: li
     return rule
 
 def edit_rule(uci, id: str, name: str, src: str, src_ip: list[str], dest: str, dest_ip: list[str], proto: list, dest_port: list[str], target: str, service: str, 
-            enabled: bool = True, log: bool = False, tag = [], ns_src: str = None, ns_dst: str = None, ns_link: str = None) -> str:
+            enabled: bool = True, log: bool = False, tag = [], ns_src: str = None, ns_dst: str = None, ns_link: str = None, src_mac: list[str] = None) -> str:
     """
     Edit rule in firewall config.
 
@@ -1736,14 +1759,15 @@ def edit_rule(uci, id: str, name: str, src: str, src_ip: list[str], dest: str, d
         ns_src: an object in the form `<database>/<id>`
         ns_dst: an object in the form `<database>/<id>`
         ns_link: a string in the form `<database>/<id>` that is used to link this rule to another entity
+        src_mac: a list of source MAC addresses, lowercase and colon-separated
 
     Returns:
         name of rule config that was edited
     """
     if not uci.get('firewall', id, default=None):
         raise utils.ValidationError("id", "rule_does_not_exists", id)  
-    validate_rule(uci, src, src_ip, dest, dest_ip, proto, dest_port, target, service, ns_src, ns_dst)
-    setup_rule(uci, id, name, src, src_ip, dest, dest_ip, proto, dest_port, target, service, enabled, log, tag, ns_src, ns_dst, ns_link)
+    validate_rule(uci, src, src_ip, dest, dest_ip, proto, dest_port, target, service, ns_src, ns_dst, src_mac)
+    setup_rule(uci, id, name, src, src_ip, dest, dest_ip, proto, dest_port, target, service, enabled, log, tag, ns_src, ns_dst, ns_link, src_mac)
     update_firewall_rules(uci) # expand objects and save
     return id
 

packages/python3-nethsec/src/nethsec/utils/__init__.py

@@ -106,6 +106,22 @@ def validate_uci_name(name, parameter='name'):
     if not name or sanitize(name) != name:
         raise ValidationError(parameter, 'invalid_name', name)
 
+_VALID_MAC_ADDRESS = re.compile(r'([0-9a-f]{2}:){5}[0-9a-f]{2}')
+
+def validate_mac_address(mac, parameter='mac'):
+    '''
+    Ensure the given MAC address is lowercase and colon-separated (e.g. '00:1a:2b:3c:4d:5e').
+
+    Arguments:
+      - mac -- the MAC address to validate
+      - parameter -- the parameter name to report in the ValidationError (default: 'mac')
+
+    Raises:
+      - ValidationError if the MAC address is not valid
+    '''
+    if not mac or not _VALID_MAC_ADDRESS.fullmatch(mac):
+        raise ValidationError(parameter, 'error.invalid_mac_address', mac)
+
 def get_all_by_type(uci, config, utype):
     '''
     Return all sections of the given utype from the given config

packages/python3-nethsec/tests/test_firewall.py

@@ -1013,6 +1013,48 @@ def test_edit_rule_to_notrack(u, mocker):
     firewall.edit_rule(u, rid, 'rule_to_change', 'lan', [], 'wan', [], [], '', 'NOTRACK', "*", True, False, [])
     assert u.get("firewall", rid, "target") == "NOTRACK"
 
+def test_add_rule_with_src_mac(u, mocker):
+    mocker.patch('builtins.open', mocker.mock_open(read_data=services_file))
+    mock_isfile = mocker.patch('os.path.isfile')
+    mock_isfile.return_value = True
+    macs = ['00:1a:2b:3c:4d:5e', 'aa:bb:cc:dd:ee:ff']
+    rid = firewall.add_rule(u, 'macrule', 'lan', [], 'wan', [], [], '', 'DROP', "*", True, False, [], False, src_mac=macs)
+    assert u.get_all("firewall", rid, "src_mac") == tuple(macs)
+    rule = next(r for r in firewall.list_forward_rules(u) if r['id'] == rid)
+    assert rule['src_mac'] == macs
+    assert rule['src_ip'] == []
+    firewall.delete_rule(u, rid)
+
+def test_edit_rule_src_mac_to_src_ip(u, mocker):
+    mocker.patch('builtins.open', mocker.mock_open(read_data=services_file))
+    mock_isfile = mocker.patch('os.path.isfile')
+    mock_isfile.return_value = True
+    rid = firewall.add_rule(u, 'macrule2', 'lan', [], 'wan', [], [], '', 'DROP', "*", True, False, [], False, src_mac=['00:1a:2b:3c:4d:5e'])
+    firewall.edit_rule(u, rid, 'macrule2', 'lan', ['192.168.1.22'], 'wan', [], [], '', 'DROP', "*", True, False, [], src_mac=[])
+    with pytest.raises(UciExceptionNotFound):
+        u.get_all("firewall", rid, "src_mac")
+    rule = next(r for r in firewall.list_forward_rules(u) if r['id'] == rid)
+    assert rule['src_mac'] == []
+    firewall.edit_rule(u, rid, 'macrule2', 'lan', [], 'wan', [], [], '', 'DROP', "*", True, False, [], src_mac=['aa:bb:cc:dd:ee:ff'])
+    assert u.get_all("firewall", rid, "src_mac") == ('aa:bb:cc:dd:ee:ff',)
+    with pytest.raises(UciExceptionNotFound):
+        u.get_all("firewall", rid, "src_ip")
+    firewall.delete_rule(u, rid)
+
+def test_add_rule_src_mac_validation(u, mocker):
+    mocker.patch('builtins.open', mocker.mock_open(read_data=services_file))
+    mock_isfile = mocker.patch('os.path.isfile')
+    mock_isfile.return_value = True
+    with pytest.raises(ValidationError) as e:
+        firewall.add_rule(u, 'badmac', 'lan', [], 'wan', [], [], '', 'DROP', "*", src_mac=['zz:zz:zz:zz:zz:zz'])
+    assert (e.value.parameter, e.value.message, e.value.value) == ('src_mac', 'error.invalid_mac_address', 'zz:zz:zz:zz:zz:zz')
+    with pytest.raises(ValidationError) as e:
+        firewall.add_rule(u, 'conflict', 'lan', ['192.168.1.22'], 'wan', [], [], '', 'DROP', "*", src_mac=['00:1a:2b:3c:4d:5e'])
+    assert e.value.message == 'error.src_mac_conflict'
+    with pytest.raises(ValidationError) as e:
+        firewall.add_rule(u, 'output', '', [], 'wan', [], [], '', 'DROP', "*", src_mac=['00:1a:2b:3c:4d:5e'])
+    assert e.value.message == 'error.src_mac_not_allowed'
+
 def test_delete_rule(u):
     ids =  firewall.list_rule_ids(u)
     id_to_delete = ids.pop()

packages/python3-nethsec/tests/test_utils.py

@@ -254,3 +254,12 @@ def test_validate_uci_name_invalid():
         with pytest.raises(utils.ValidationError) as e:
             utils.validate_uci_name(name)
         assert e.value.message == 'invalid_name'
+
+def test_validate_mac_address_valid():
+    utils.validate_mac_address('00:1a:2b:3c:4d:5e')
+
+@pytest.mark.parametrize('mac', ['', 'zz:zz:zz:zz:zz:zz', '00:1A:2B:3C:4D:5E', '00-1a-2b-3c-4d-5e', '00:1a:2b:3c:4d', '00:1a:2b:3c:4d:5e\n'])
+def test_validate_mac_address_invalid(mac):
+    with pytest.raises(utils.ValidationError) as e:
+        utils.validate_mac_address(mac, 'src_mac')
+    assert (e.value.parameter, e.value.message) == ('src_mac', 'error.invalid_mac_address')