extending tests for firewall chain handling

Author: superstes

src/firewall_test/simulator/05_firewall_chains_test.py

@@ -1,25 +1,98 @@
 import pytest
 
 from testdata_test import TESTDATA_FILE_NF_RULESET
-from config import ProtoL3IP4, ProtoL3IP4IP6, ProtoL3IP6, FlowInput, FlowOutput, FlowForward
+from config import ProtoL3IP4, ProtoL3IP6  # RuleActionAccept, RuleActionDrop, RuleActionReject, RuleActionSNAT
+from simulator.firewall import Firewall
 
 with open(TESTDATA_FILE_NF_RULESET, 'r', encoding='utf-8') as f:
     TESTDATA_RULESET = f.read()
 
 
+def _init_test() -> Firewall:
+    from plugins.system.system_linux_netfilter import SystemLinuxNetfilter
+    from plugins.translate.netfilter.ruleset import NetfilterRuleset
+    from simulator.firewall import Firewall
 
-def test_firewall_chains_basic():
+    ruleset = NetfilterRuleset(TESTDATA_RULESET).get()
+    return Firewall(
+        system=SystemLinuxNetfilter,
+        ruleset=ruleset,
+    )
+
+
+@pytest.mark.parametrize(
+    'src,dst,table_name,ipp,is_none',
+    [
+        ('192.168.0.10', '1.1.1.1', 'main', ProtoL3IP4, False),
+        ('192.168.0.10', '1.1.1.1', 'main', ProtoL3IP6, True),
+        ('192.168.0.10', '1.1.1.1', 'early', ProtoL3IP4, False),
+        ('::1', '2003::2', 'main', ProtoL3IP6, False),
+        ('::1', '2003::2', 'main', ProtoL3IP4, True),
+    ]
+)
+def test_firewall_chain_by_name_and_family(src, dst, table_name, ipp, is_none):
     from simulator.packet import PacketIP
     from simulator.firewall import Firewall
-    from plugins.translate.abstract import Table
+    from plugins.translate.abstract import Table, Ruleset
     from plugins.system.system_linux_netfilter import SystemLinuxNetfilter
-    from plugins.translate.netfilter.ruleset import NetfilterRuleset
+    from plugins.translate.netfilter.ruleset import NetfilterChainOutput as Chain
+
+    table_early = Table(
+        name='early',
+        priority=-1,
+        chains=[
+            Chain(name='input', hook='input', policy=Chain.POLICY_ACCEPT, family=ProtoL3IP4, rules=[]),
+            Chain(name='input', hook='input', policy=Chain.POLICY_ACCEPT, family=ProtoL3IP6, rules=[]),
+        ],
+    )
+    table_main = Table(
+        name='main',
+        chains=[
+            Chain(name='input', hook='input', policy=Chain.POLICY_DROP, family=ProtoL3IP4, rules=[]),
+            Chain(name='input', hook='input', policy=Chain.POLICY_DROP, family=ProtoL3IP6, rules=[]),
+            Chain(name='input', hook='input', policy=Chain.POLICY_DROP, rules=[]),
+        ],
+    )
 
-    ruleset = NetfilterRuleset(TESTDATA_RULESET).get()
     fw = Firewall(
         system=SystemLinuxNetfilter,
-        ruleset=ruleset,
+        ruleset=Ruleset(tables=[table_early, table_main]),
     )
-    # packet = PacketIP(src=src, dst=dst)
-    # table = Table(name='test', chains=[], family=ipp)
-    # assert fw._run_tables._is_matching_table(packet=packet, table=table) == matching
+    packet = PacketIP(src=src, dst=dst)
+    table = None
+    for t in fw.ruleset.tables:
+        if t.name == table_name:
+            table = t
+
+        for c in t.chains:
+            c.run_table = t
+
+    c = fw._run_tables._run_chains._get_chain_by_name_and_family(
+        packet=packet,
+        name='input',
+        family=ipp,
+        table=table,
+    )
+    if is_none:
+        assert c is None
+
+    else:
+        assert c is not None
+        assert c.name == 'input'
+        assert c.hook == 'input'
+        assert c.family == ipp
+        assert c.run_table.name == table_name
+
+
+# todo: test multiple testdata-rulesets for edge-case handling (RunFirewallChain.process)
+#   sub-chain (jump) terminal action should end hook
+#   sub-chain (goto) should end parent chain
+#   dnat
+#   snat
+#   masquerade
+#   action return
+#   action continue (?)
+#   lazy action
+#   lazy action mixed with 'quick' action
+#   chain default-policy
+

src/firewall_test/simulator/07_firewall_e2e_test.py renamed to src/firewall_test/simulator/06_firewall_e2e_test.py

@@ -16,9 +16,12 @@ def __init__(self, fw, run_tables):
         self._fw = fw
         self._run_tables = run_tables
 
-    def _get_chain_by_name_and_family(self, packet: PACKET_KINDS, name: str, family: str) -> (Chain, None):
-        for table in self._run_tables.get_tables(packet):
-            for chain in self._run_tables.get_chains(packet=packet, table=table):
+    def _get_chain_by_name_and_family(self, packet: PACKET_KINDS, table: Table, name: str, family: str) -> (Chain, None):
+        for t in self._run_tables.get_tables(packet):
+            if t.name != table.name or t.family != table.family or t.priority != table.priority:
+                continue
+
+            for chain in self._run_tables.get_chains(packet=packet, table=t):
                 if chain.name == name and chain.family == family:
                     return chain
 
@@ -101,6 +104,7 @@ def process(self, chain: Chain, packet: PACKET_KINDS) -> (bool, (Rule, None)):
                     packet=packet,
                     name=result.target_chain_name,
                     family=chain.family,
+                    table=chain.run_table,
                 )
                 if target_chain is None:
                     log_warn(
@@ -146,7 +150,7 @@ def process(self, chain: Chain, packet: PACKET_KINDS) -> (bool, (Rule, None)):
             log_debug('Firewall', f'> Chain {chain.name} | Applying lazy-action: {action_str}')
             return not issubclass(lazy_action, RuleActionKindTerminalKill), lazy_rule
 
-        if chain.policy in [RuleActionDrop, RuleActionReject]:
+        if chain.type == chain.TYPE_FILTER and chain.policy in [RuleActionDrop, RuleActionReject]:
             return False, None
 
         return True, None