Make cleaner to remove schain firewall rules config

core/schains/firewall/nftables.py

@@ -52,6 +52,7 @@ def __init__(self, table: str = TABLE, chain: str = CHAIN) -> None:
         self._nftables = importlib.import_module('nftables')
         self.nft = self._nftables.Nftables()
         self.nft.set_json_output(True)
+        self.nft.set_stateless_output(True)
 
     @classmethod
     def rule_to_expr(cls, rule: SChainRule, counter: bool = True) -> list:
@@ -179,6 +180,7 @@ def create_chain(self, first_port: int, last_port: int) -> None:
                 )
             )
             self.add_schain_drop_rule(first_port, last_port)
+            self.save_rules()
 
     def delete_chain(self) -> None:
         if self.has_chain(self.chain):
@@ -330,7 +332,9 @@ def get_plain_chain_rules(self) -> str:
         self.nft.set_json_output(False)
         output = ''
         try:
-            rc, output, error = self.run_cmd(f'list chain {self.FAMILY} {self.table} {self.chain}')
+            rc, output, error = self.run_cmd(
+                f'list chain {self.FAMILY} {self.table} {self.chain}'
+            )
             if rc != 0:
                 raise NFTablesCmdFailedError(f"Failed to get table content: {error}")
         finally:
@@ -346,7 +350,7 @@ def save_rules(self) -> None:
 
     def remove_saved_rules(self) -> None:
         nft_chain_path = os.path.join(NFT_CHAIN_BASE_PATH, f'{self.chain}.conf')
-        shutil.rmtree(nft_chain_path)
+        os.remove(nft_chain_path)
 
     def cleanup(self) -> None:
         self.delete_chain()

tests/firewall/nftables_test.py

@@ -1,11 +1,14 @@
 import concurrent.futures
 import importlib
+import os
+import shutil
 import time
 
 import pytest
 
-from core.schains.firewall.nftables import NFTablesController
+from core.schains.firewall.nftables import NFTablesController, NFT_CHAIN_BASE_PATH
 from core.schains.firewall.types import SChainRule
+from tools.helper import run_cmd
 
 
 @pytest.fixture
@@ -15,6 +18,16 @@ def nf_test_tables():
     return nft
 
 
+@pytest.fixture()
+def nft_chain_folder():
+    path = '/etc/nft.conf.d/chains'
+    try:
+        os.makedirs(path)
+        yield path
+    finally:
+        shutil.rmtree(path)
+
+
 @pytest.fixture
 def filter_table(nf_test_tables):
     print(nf_test_tables.cmd('add table inet firewall'))
@@ -67,6 +80,27 @@ def test_nftables_controller_duplicates(custom_chain):
     ]
 
 
+def test_create_delete_chain(filter_table, nft_chain_folder):
+    chain_name = 'test-chain'
+
+    output = run_cmd(['nft', 'list', 'chains']).stdout.decode('utf-8')
+    output == 'table inet firewall {\n}\n'
+    nft_chain_path = os.path.join(NFT_CHAIN_BASE_PATH, f'skale-{chain_name}.conf')
+    assert not os.path.isfile(nft_chain_path)
+
+    manager = NFTablesController(chain=chain_name)
+    manager.create_chain(first_port=10000, last_port=10063)
+
+    chains = run_cmd(['nft', 'list', 'chains']).stdout.decode('utf-8')
+    assert chains == 'table inet firewall {\n\tchain skale-test-chain {\n\t\ttype filter hook input priority filter; policy accept;\n\t}\n}\n'  # noqa
+    assert os.path.isfile(nft_chain_path)
+
+    manager.cleanup()
+    chains = run_cmd(['nft', 'list', 'chains']).stdout.decode('utf-8')
+    assert chains == 'table inet firewall {\n}\n'
+    assert not os.path.isfile(nft_chain_path)
+
+
 def add_remove_rule(srule, refresh):
     manager = NFTablesController()
     manager.add_rule(srule)