Merge pull request #2 from MikPisula/latest

chore: Update Python typing

Author: superstes

.gitignore

@@ -5,6 +5,7 @@ src/*.egg-info
 dist/
 .idea/
 .vscode/
+__pycache__
 
 testdata/custom_*
 scripts/cli_custom.sh

src/firewall_test/plugins/system/abstract_rule_match.py

@@ -11,10 +11,10 @@ class RuleMatchResult:
     def __init__(
             self,
             matched: bool,
-            action: (type[RuleAction], None),
-            target_chain_name: (str, None),
-            target_nat_ip: ((IPv4Address, IPv6Address), None),
-            target_nat_port: (int, None),
+            action: type[RuleAction]|None,
+            target_chain_name: str|None,
+            target_nat_ip: IPv4Address|IPv6Address|None,
+            target_nat_port: int|None,
     ):
         self.matched = matched
         self.action = action

src/firewall_test/plugins/system/firewall_netfilter.py

@@ -2,15 +2,15 @@
 from plugins.system.abstract_rule_match import RuleMatcher, RuleMatchResult
 from plugins.translate.abstract import Rule
 from plugins.translate.netfilter.parse import NftRule
-from simulator.packet import PacketIP, PacketTCPUDP, PacketICMP
+from simulator.packet import PacketIP, PacketTCPUDP
 from utils.logger import log_debug, log_warn
 
 # todo: add explicit match-tests
 
 
 # pylint: disable=R0912
 class RuleMatcherNetfilter(RuleMatcher):
-    def matches(self, packet: (PacketIP, PacketTCPUDP, PacketICMP), rule: Rule) -> RuleMatchResult:
+    def matches(self, packet: PacketIP, rule: Rule) -> RuleMatchResult:
         """
         :param packet: Packet to match
         :param rule: Rule to check

src/firewall_test/plugins/system/firewall_opnsense.py

@@ -1,5 +1,5 @@
 from config import ProtoL3IP4IP6, RuleActionGoTo
-from simulator.packet import PACKET_KINDS, PacketTCPUDP
+from simulator.packet import PacketIP, PacketTCPUDP
 from plugins.translate.abstract import Rule
 from plugins.system.abstract_rule_match import RuleMatcher, RuleMatchResult
 from plugins.translate.opnsense.rule import OPNsenseRule, RULE_SEQUENCE_NEXT_CHAIN
@@ -10,7 +10,7 @@
 
 # pylint: disable=R0912
 class RuleMatcherOPNsense(RuleMatcher):
-    def matches(self, packet: PACKET_KINDS, rule: Rule) -> RuleMatchResult:
+    def matches(self, packet: PacketIP, rule: Rule) -> RuleMatchResult:
         """
         :param packet: Packet to match
         :param rule: Rule to check

src/firewall_test/plugins/translate/abstract.py

@@ -14,13 +14,13 @@ def __init__(self, raw: any):
         self.raw = raw
 
     @abstractmethod
-    def get(self) -> (dict, list[dict]):
+    def get(self) -> dict|list[dict]:
         pass
 
 
 class TranslateOutput(ABC):
     @abstractmethod
-    def dump(self) -> (dict, list[dict]):
+    def dump(self) -> dict|list[dict]:
         pass
 
     @abstractmethod
@@ -304,12 +304,12 @@ class Chain(TranslateOutput):
 
     # pylint: disable=W0622
     def __init__(
-        self, name: str, hook: (str, None), policy: (None, RuleActionAccept, RuleActionDrop, RuleActionReject),
+        self, name: str, hook: str|None, policy: RuleActionAccept|RuleActionDrop|RuleActionReject|None,
             rules: list[Rule], priority: int = 0, type: str = 'filter', family: type[ProtoL3] = ProtoL3IP4IP6,
     ):
         self.name = name
         self.type = type
-        self.family: type[ProtoL3] = family
+        self.family = family
         self.hook = hook
         self.priority = priority
         self.policy = policy
@@ -388,7 +388,7 @@ def __init__(
         self.type = type
         self.priority = priority
         self.chains = chains
-        self.family: type[ProtoL3] = family
+        self.family = family
 
     def dump(self) -> dict:
         return {

src/firewall_test/plugins/translate/netfilter/elements.py

@@ -94,7 +94,7 @@ class NftMatch:
     OP_GT = '>'
     OP_LT = '<'
 
-    def __init__(self, operator: str, left: (str, dict), right: (str, dict, list)):
+    def __init__(self, operator: str, left: str|dict, right: str|dict|list):
         self.operator = operator
         self._left = left
 
@@ -270,7 +270,7 @@ def update_value_type(self):
 
         self.value = values
 
-    def get_matches(self) -> (dict, None):
+    def get_matches(self) -> dict|None:
         matches = {}
         if self.match_proto_l3:
             if self.value_proto_l3:

src/firewall_test/plugins/translate/opnsense/rule.py

@@ -1,6 +1,6 @@
 from ipaddress import IPv4Network, IPv6Network
 
-from config import ProtoL3IP4IP6, PROTOS_L3, PROTOS_L4
+from config import ProtoL3IP4IP6, ProtoL3, ProtoL4
 from utils.logger import rule_repr
 
 # pylint: disable=R0801
@@ -24,8 +24,8 @@ def __init__(
             nis: list[str] = None,
             ni_direction: str = None,
             desc: str = None,
-            ipprotocol: PROTOS_L3 = ProtoL3IP4IP6,
-            protocol: PROTOS_L4 = None,
+            ipprotocol: ProtoL3 = ProtoL3IP4IP6,
+            protocol: ProtoL4 = None,
             source: list[(IPv4Network, IPv6Network)] = None,
             destination: list[(IPv4Network, IPv6Network)] = None,
             source_port: list[int] = None,
@@ -85,7 +85,7 @@ def match_ip_daddr(self) -> bool:
 
         return False
 
-    def get_matches(self) -> (dict, None):
+    def get_matches(self) -> dict|None:
         matches = {}
         if self.ipp is not None:
             matches['proto_l3'] = self.ipp.N

src/firewall_test/plugins/translate/opnsense/ruleset.py

@@ -2,6 +2,7 @@
 from ipaddress import ip_network, summarize_address_range, ip_address
 import xml.etree.ElementTree as ET
 from xml.etree.ElementTree import Element
+from typing import Any
 
 from requests import get as http_get
 from requests import Response as HttpResponse
@@ -167,7 +168,7 @@ def get(self) -> Ruleset:
 
     ### RULES ###
 
-    def _parse_rule_address(self, value: str) -> (list, None):
+    def _parse_rule_address(self, value: str) -> list|None:
         values = split_csv(value)
         out = []
 
@@ -197,7 +198,7 @@ def _parse_rule_address(self, value: str) -> (list, None):
 
         return out
 
-    def _parse_rule_network(self, value: str) -> (list, None):
+    def _parse_rule_network(self, value: str) -> list|None:
         values = split_csv(value)
         out = []
 
@@ -227,7 +228,7 @@ def _parse_rule_network(self, value: str) -> (list, None):
 
         return out
 
-    def _parse_rule_port(self, value: str) -> (list, None):
+    def _parse_rule_port(self, value: str) -> list|None:
         values = split_csv(value)
         out = []
 
@@ -252,7 +253,7 @@ def _parse_rule_port(self, value: str) -> (list, None):
         return out
 
     @staticmethod
-    def log_unsupported_rule(chain: Chain, rule_raw: dict, rule: dict, result: (any, None), invalid: bool) -> bool:
+    def log_unsupported_rule(chain: Chain, rule_raw: dict, rule: dict, result: Any, invalid: bool) -> bool:
         if invalid:
             return invalid
 
@@ -592,7 +593,7 @@ def _parse_alias_iplist_plain(self, url: str) -> (list[str], None):
         return content
 
     @staticmethod
-    def _download_alias_iplist(url: str) -> (HttpResponse, None):
+    def _download_alias_iplist(url: str) -> HttpResponse|None:
         url = url.strip()
         if not url.startswith('http'):
             log_warn('Firewall Plugin', f'Unsupported alias-type "urltable" URL: {url}')

src/firewall_test/simulator/firewall.py

@@ -7,7 +7,7 @@
 from plugins.system.abstract import FirewallSystem
 from plugins.system.abstract_rule_match import RuleMatchResult
 from plugins.translate.abstract import Ruleset, Table, Chain, Rule
-from simulator.packet import PACKET_KINDS, PacketTCPUDP
+from simulator.packet import PacketIP, PacketTCPUDP
 from utils.logger import log_debug, log_info, log_warn
 
 
@@ -16,7 +16,7 @@ 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, table: Table, name: str, family: str) -> (Chain, None):
+    def _get_chain_by_name_and_family(self, packet: PacketIP, 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
@@ -41,7 +41,7 @@ def _log_match(self, chain: Chain, rule: Rule, debug: bool = False):
             log_info(label='Firewall', v1=msg, v2=v2)
 
     # pylint: disable=R0911,R0912
-    def process(self, chain: Chain, packet: PACKET_KINDS) -> (bool, (Rule, None)):
+    def process(self, chain: Chain, packet: PacketIP) -> tuple[bool, Rule|None]:
         """
         :param chain: Firewall chain to process; if any rule has an action that targets another chain - it will also be processed
         :param packet: Packet to match
@@ -52,8 +52,8 @@ def process(self, chain: Chain, packet: PACKET_KINDS) -> (bool, (Rule, None)):
 
         rule_matcher = self._fw.system.get_rule_matcher()(chain.run_table)
         chain.rules.sort(key=lambda r: r.seq, reverse=False)
-        lazy_action: (None, RuleAction) = None
-        lazy_rule: (None, Rule) = None
+        lazy_action: RuleAction|None = None
+        lazy_rule: Rule|None = None
 
         for rule in chain.rules:
             result: RuleMatchResult = rule_matcher.matches(packet=packet, rule=rule)
@@ -162,7 +162,7 @@ def __init__(self, fw):
         self._run_chains = RunFirewallChain(fw=fw, run_tables=self)
 
     @staticmethod
-    def _is_matching_table(packet: PACKET_KINDS, table: Table, ignore_type: list[str] = None) -> bool:
+    def _is_matching_table(packet: PacketIP, table: Table, ignore_type: list[str] = None) -> bool:
         if ignore_type is not None and table.type in ignore_type:
             return False
 
@@ -177,14 +177,14 @@ def _is_matching_table(packet: PACKET_KINDS, table: Table, ignore_type: list[str
 
         return False
 
-    def get_tables(self, packet: PACKET_KINDS, ignore_type: list[str] = None) -> list[Table]:
+    def get_tables(self, packet: PacketIP, ignore_type: list[str] = None) -> list[Table]:
         return [
             t for t in self._fw.ruleset.tables
             if self._is_matching_table(packet=packet, table=t, ignore_type=ignore_type)
         ]
 
     @staticmethod
-    def _is_matching_chain(packet: PACKET_KINDS, chain: Chain, ignore_type: list[str] = None) -> bool:
+    def _is_matching_chain(packet: PacketIP, chain: Chain, ignore_type: list[str] = None) -> bool:
         if ignore_type is not None and chain.type in ignore_type:
             return False
 
@@ -200,7 +200,7 @@ def _is_matching_chain(packet: PACKET_KINDS, chain: Chain, ignore_type: list[str
 
         return False
 
-    def get_chains(self, packet: PACKET_KINDS, table: Table, ignore_type: list[str] = None):
+    def get_chains(self, packet: PacketIP, table: Table, ignore_type: list[str] = None):
         return [
             c for c in table.chains
             if self._is_matching_chain(packet=packet, chain=c, ignore_type=ignore_type)
@@ -283,8 +283,8 @@ def _inherit_table_priority_to_chain(table: Table, chain: Chain):
             chain.priority = table.priority
 
     def _process_by_table_prio(
-            self, tables: list[Table], callback_chain_filter: Callable[[Chain], bool], packet: PACKET_KINDS,
-    ) -> (bool, (Rule, None)):
+            self, tables: list[Table], callback_chain_filter: Callable[[Chain], bool], packet: PacketIP,
+    ) -> tuple[bool, Rule|None]:
         for table in self._sort_tables_by_priority(tables):
             chains = [
                 c for c in self.get_chains(packet=packet, table=table)
@@ -300,8 +300,8 @@ def _process_by_table_prio(
         return True, None
 
     def _process_by_chain_prio(
-            self, tables: list[Table], callback_chain_filter: Callable[[Chain], bool], packet: PACKET_KINDS,
-    ) -> (bool, (Rule, None)):
+            self, tables: list[Table], callback_chain_filter: Callable[[Chain], bool], packet: PacketIP,
+    ) -> tuple[bool, Rule|None]:
         chains: list[Chain] = []
         for table in tables:
             for chain in self.get_chains(packet=packet, table=table):
@@ -323,7 +323,7 @@ def _process_by_chain_prio(
 
         return True, None
 
-    def _process_chain(self, chain: Chain, packet: PACKET_KINDS) -> (bool, (Rule, None)):
+    def _process_chain(self, chain: Chain, packet: PacketIP) -> tuple[bool, Rule|None]:
         """
         see: RunFirewallChain.process
         """
@@ -343,7 +343,7 @@ def _chain_filter_pre_routing(self, chain: Chain, flow: type[Flow]) -> bool:
         before_dnat = self._is_chain_before_eq(chain=chain, **self._fw.system.FIREWALL_NAT[flow]['dnat'])
         return chain.type != chain.TYPE_NAT and before_dnat
 
-    def process_pre_routing(self, packet: PACKET_KINDS, flow: type[Flow]) -> (bool, (Rule, None)):
+    def process_pre_routing(self, packet: PacketIP, flow: type[Flow]) -> tuple[bool, Rule|None]:
         """
         :param packet: Packet to process
         :param flow: traffic flow-type
@@ -368,7 +368,7 @@ def _chain_filter_dnat(self, chain: Chain, flow: type[Flow]) -> bool:
             chain.hook == chain_dnat['hook'] and \
             chain.priority == chain_dnat['priority']
 
-    def process_dnat(self, packet: PACKET_KINDS, flow: type[Flow]) -> (bool, (Rule, None)):
+    def process_dnat(self, packet: PacketIP, flow: type[Flow]) -> tuple[bool, Rule|None]:
         """
         :param packet: Packet to process
         :param flow: traffic flow-type
@@ -400,7 +400,7 @@ def _chain_filter_main(self, chain: Chain, flow: type[Flow]) -> bool:
 
         return chain.type != chain.TYPE_NAT and after_dnat and before_snat
 
-    def process_main(self, packet: PACKET_KINDS, flow: type[Flow]) -> (bool, (Rule, None)):
+    def process_main(self, packet: PacketIP, flow: type[Flow]) -> tuple[bool, Rule|None]:
         """
         :param packet: Packet to process
         :param flow: traffic flow-type
@@ -425,7 +425,7 @@ def _chain_filter_snat(self, chain: Chain, flow: type[Flow]) -> bool:
             chain.hook == chain_snat['hook'] and \
             chain.priority == chain_snat['priority']
 
-    def process_snat(self, packet: PACKET_KINDS, flow: type[Flow]) -> (bool, (Rule, None)):
+    def process_snat(self, packet: PacketIP, flow: type[Flow]) -> tuple[bool, Rule|None]:
         """
         :param packet: Packet to process
         :param flow: traffic flow-type
@@ -451,7 +451,7 @@ def _chain_filter_egress(self, chain: Chain, flow: type[Flow]) -> bool:
         after_snat = self._is_chain_after(chain=chain, **self._fw.system.FIREWALL_NAT[flow]['snat'])
         return chain.type != chain.TYPE_NAT and after_snat
 
-    def process_egress(self, packet: PACKET_KINDS, flow: type[Flow]) -> (bool, (Rule, None)):
+    def process_egress(self, packet: PacketIP, flow: type[Flow]) -> tuple[bool, Rule|None]:
         """
         :param packet: Packet to process
         :param flow: traffic flow-type
@@ -475,15 +475,15 @@ def __init__(self, system: type[FirewallSystem], ruleset: Ruleset):
 
         self._run_tables = RunFirewallTables(self)
 
-    def process_pre_routing(self, packet: PACKET_KINDS, flow: type[Flow]) -> (bool, (Rule, None)):
+    def process_pre_routing(self, packet: PacketIP, flow: type[Flow]) -> tuple[bool, Rule|None]:
         log_info('Firewall', v3='Processing Pre-Routing Filter-Hooks')
         if flow == FlowInputForward:
             # before DNAT we cannot know for sure
             flow = FlowInput
 
         return self._run_tables.process_pre_routing(packet=packet, flow=flow)
 
-    def process_dnat(self, packet: PACKET_KINDS, flow: type[Flow]) -> (bool, (Rule, None)):
+    def process_dnat(self, packet: PacketIP, flow: type[Flow]) -> tuple[bool, Rule|None]:
         if flow == FlowInputForward:
             # before DNAT we cannot know for sure
             flow = FlowInput
@@ -496,12 +496,12 @@ def process_dnat(self, packet: PACKET_KINDS, flow: type[Flow]) -> (bool, (Rule,
 
         return self._run_tables.process_dnat(packet=packet, flow=flow)
 
-    def process_main(self, packet: PACKET_KINDS, flow: type[Flow]) -> (bool, (Rule, None)):
+    def process_main(self, packet: PacketIP, flow: type[Flow]) -> tuple[bool, Rule|None]:
         log_info('Firewall', v3='Processing Main Filter-Hooks')
 
         return self._run_tables.process_main(packet=packet, flow=flow)
 
-    def process_snat(self, packet: PACKET_KINDS, flow: type[Flow]) -> (bool, (Rule, None)):
+    def process_snat(self, packet: PacketIP, flow: type[Flow]) -> tuple[bool, Rule|None]:
         if not self.system.FIREWALL_SNAT or 'snat' not in self.system.FIREWALL_NAT[flow]:
             # system or flow has no SNAT capability
             return False, None
@@ -510,7 +510,7 @@ def process_snat(self, packet: PACKET_KINDS, flow: type[Flow]) -> (bool, (Rule,
 
         return self._run_tables.process_snat(packet=packet, flow=flow)
 
-    def process_egress(self, packet: PACKET_KINDS, flow: type[Flow]) -> (bool, (Rule, None)):
+    def process_egress(self, packet: PacketIP, flow: type[Flow]) -> tuple[bool, Rule|None]:
         if 'snat' not in self.system.FIREWALL_NAT[flow]:
             # already processed all chains
             return True, None