added basic cli-logger, added logic to drop bogons-traffic on wan

Author: superstes

scripts/cli.sh

@@ -0,0 +1,23 @@
+#!/bin/bash
+
+set -eo pipefail
+
+if [ -z "$1" ]
+then
+  SRC='172.17.10.5'
+else
+  SRC="$1"
+fi
+
+if [ -z "$2" ]
+then
+  DST='1.1.1.1'
+else
+  DST="$2"
+fi
+
+cd "$(dirname "$0")/.."
+
+export DEBUG=1
+
+python3 src/firewall_test/cli.py --firewall-system 'linux_netfilter' --file-interfaces 'testdata/plugin_translate_linux_interfaces.json' --file-routes 'testdata/plugin_translate_linux_routes.json' --file-route-rules 'testdata/plugin_translate_linux_route-rules.json' --src-ip "$SRC" --dst-ip "$DST"

src/firewall_test/cli.py

@@ -1,3 +1,4 @@
+from os import environ
 from os import path as os_path
 from sys import path as sys_path
 from argparse import ArgumentParser
@@ -76,6 +77,7 @@ def main():
             l4_proto=args.proto_l4,
         )
 
+    print()
     loaded = load(
         system=args.firewall_system,
         file_interfaces=args.file_interfaces,
@@ -84,7 +86,11 @@ def main():
     )
     s = Simulator(**loaded)
     r = s.run(packet)
-    print('\n', r.to_json())
+
+    if 'DEBUG' in environ:
+        print('\n', r.to_json())
+
+    print()
 
 
 if __name__ == '__main__':

src/firewall_test/plugins/translate/config.py

@@ -0,0 +1,5 @@
+from ipaddress import ip_network
+
+DEFAULT_ROUTE_IP4 = ip_network('0.0.0.0/0')
+DEFAULT_ROUTE_IP6 = ip_network('::/0')
+DEFAULT_ROUTES = [DEFAULT_ROUTE_IP4, DEFAULT_ROUTE_IP6]

src/firewall_test/plugins/translate/linux.py

@@ -1,6 +1,7 @@
 from ipaddress import ip_address, ip_network, IPv4Address
 from json import loads as json_loads
 
+from plugins.translate.config import DEFAULT_ROUTE_IP4, DEFAULT_ROUTE_IP6
 from plugins.translate.abstract import TranslatePluginStaticRoutes, TranslatePluginStaticRouteRules, \
     StaticRoute, StaticRouteRule, TranslatePluginNetworkInterfaces, NetworkInterface
 
@@ -28,8 +29,8 @@ def _parse_rule(raw: dict) -> dict:
 
         if src == 'all':
             r['src'] = [
-                ip_network('0.0.0.0/0'),
-                ip_network('::/0'),
+                DEFAULT_ROUTE_IP4,
+                DEFAULT_ROUTE_IP6,
             ]
 
         else:
@@ -63,13 +64,13 @@ def _parse_route(raw: dict) -> dict:
 
         if raw.get('dst') == 'default':
             if r['gw'] is None:
-                r['net'] = ip_network('0.0.0.0/0')
+                r['net'] = DEFAULT_ROUTE_IP4
 
             elif r['gw'].find(':') == -1:
-                r['net'] = ip_network('0.0.0.0/0')
+                r['net'] = DEFAULT_ROUTE_IP4
 
             else:
-                r['net'] = ip_network('::/0')
+                r['net'] = DEFAULT_ROUTE_IP6
 
         else:
             r['net'] = ip_network(raw['dst'])

src/firewall_test/simulator/02_routes_test.py

@@ -22,8 +22,7 @@ def test_router_dst_route():
 
     packet = PacketIP(src='192.168.0.10', dst='1.1.1.1', l3_proto='ip4')
     r = router.get_route(packet)
-    assert len(r) == 1
-    r = r[0]
+    assert r is not None
     assert r.net == ip_network('0.0.0.0/0')
     assert r.table == 'default'
 
@@ -41,15 +40,13 @@ def test_router_src_route():
 
     packet = PacketIP(src='192.168.0.10', dst='1.1.1.1', l3_proto='ip4')
     r = router.get_src_route(packet)
-    assert len(r) == 1
-    r = r[0]
+    assert r is not None
     assert r.net == ip_network('0.0.0.0/0')
     assert r.table == 'default'
 
     packet = PacketIP(src='10.255.255.20', dst='1.1.1.1', l3_proto='ip4')
     r = router.get_src_route(packet)
-    assert len(r) == 2
-    r = r[0]
+    assert r is not None
     assert r.net == ip_network('10.255.255.0/24')
     assert r.table == 'default'
 

src/firewall_test/simulator/12_main_test.py

@@ -23,7 +23,7 @@ def test_basic():
     assert r.packet.ni_out == 'wan'
     assert r.flow_type == FLOW_FORWARD
 
-    assert r.route_src[0].net == ip_network('172.17.0.0/16')
-    assert r.route_src[0].ni == 'docker0'
-    assert r.route_dst[0].net == ip_network('0.0.0.0/0')
-    assert r.route_dst[0].ni == 'wan'
+    assert r.route_src.net == ip_network('172.17.0.0/16')
+    assert r.route_src.ni == 'docker0'
+    assert r.route_dst.net == ip_network('0.0.0.0/0')
+    assert r.route_dst.ni == 'wan'

src/firewall_test/simulator/logger.py

@@ -0,0 +1,36 @@
+from sys import stdout
+from os import environ
+
+COLOR_OK = '\x1b[1;32m'
+COLOR_WARN = '\x1b[1;33m'
+COLOR_INFO = '\x1b[1;34m'
+COLOR_ERROR = '\x1b[1;31m'
+COLOR_DEBUG = '\x1b[35m'
+RESET_STYLE = '\x1b[0m'
+
+
+def _log(label: str, msg: str, color: str, symbol: str):
+    stdout.write(
+        color + symbol + ' ' + label.upper() + ': ' + msg + RESET_STYLE + '\n',
+    )
+
+
+def log_debug(label: str, msg: str):
+    if 'DEBUG' in environ:
+        _log(label='DEBUG ' + label, msg=msg, color=COLOR_DEBUG, symbol='🛈')
+
+
+def log_ok(label: str, msg: str):
+    _log(label=label, msg=msg, color=COLOR_OK, symbol=' ✓')
+
+
+def log_info(label: str, msg: str):
+    _log(label=label, msg=msg, color=COLOR_INFO, symbol='🛈')
+
+
+def log_warn(label: str, msg: str):
+    _log(label=label, msg=msg, color=COLOR_WARN, symbol='⚠')
+
+
+def log_error(label: str, msg: str):
+    _log(label=label, msg=msg, color=COLOR_ERROR, symbol=' ✖')

src/firewall_test/simulator/main.py

@@ -3,9 +3,11 @@
 
 from simulator.packet import PacketIP
 from simulator.routes import Router
+from simulator.logger import log_info, log_error, log_ok
 
-from plugins.translate.abstract import NetworkInterface, StaticRoute, StaticRouteRule
 from plugins.system.abstract import FirewallSystem
+from plugins.translate.config import DEFAULT_ROUTES
+from plugins.translate.abstract import NetworkInterface, StaticRoute, StaticRouteRule
 
 FLOW_INPUT = 'input'
 FLOW_OUTPUT = 'output'
@@ -29,37 +31,61 @@ def __init__(self, packet: PacketIP, simulator):
         self.local_src, packet.ni_in = self._is_ip_local(packet.src)
         self.local_dst, packet.ni_out = self._is_ip_local(packet.dst)
         self.flow_type = self._get_flow_type()
+        # log_info('Firewall', f'Flow-type: {self.flow_type}')
 
         self.route_src = self._s.router.get_src_route(self.packet)
         self._update_packet_ni_in()
+        if packet.ni_in is not None:
+            log_info('Firewall', f'Packet inbound-interface: {packet.ni_in}')
 
         if self.route_src is None:
-            raise ConnectionError('No Source-Route found')
+            log_error('Router', 'No Source-Route found')
+            return
 
+        # todo: prerouting firewall-filters
         # todo: DNAT
         self._dnat_done = True
         self.dnat = None
+        if self.dnat is not None:
+            log_info('Firewall', f'Performed DNAT: {self.dnat}')
 
         self.local_dst, packet.ni_out = self._is_ip_local(packet.dst)
         self.flow_type = self._get_flow_type()
+
         self.route_dst = self._s.router.get_route(self.packet)
         self._update_packet_ni_out()
+        if packet.ni_out is not None:
+            log_info('Firewall', f'Packet outbound-interface: {packet.ni_out}')
 
         if self.route_dst is None:
-            raise ConnectionError('No Destination-Route found')
+            log_error('Router', 'No Destination-Route found')
+            return
+
+        log_info('Firewall', f'Flow-type: {self.flow_type}')
+
+        if self._is_bogon_to_wan() and self._s.system.FIREWALL_WAN_DROP_BOGONS:
+            log_error('Firewall', 'Dropping traffic to WAN targeting bogons')
+            return
+
+        # todo: main firewall-filters
 
         # todo: SNAT
         self.snat = None
+        if self.snat is not None:
+            log_info('Firewall', f'Performed SNAT: {self.snat}')
+
+        # todo: egress firewall-filters
+
+        log_ok('Firewall', 'Packet passed')
 
     def dump(self) -> dict:
         return {
             'packet': self.packet.dump(),
-            'ipp': self._ipp,
             'src_is_local': self.local_src,
             'dst_is_local': self.local_dst,
             'flow_type': self.flow_type,
-            'route_src': [route.dump() for route in self.route_src],
-            'route_dst': [route.dump() for route in self.route_dst],
+            'route_src': self.route_src.dump() if self.route_src is not None else None,
+            'route_dst': self.route_dst.dump() if self.route_dst is not None else None,
             'dnat': self.dnat,
             'snat': self.snat,
         }
@@ -91,19 +117,26 @@ def _update_packet_ni_in(self):
         if self.packet.ni_in is not None:
             return
 
-        if len(self.route_src) == 0:
+        if self.route_src is None:
             return
 
-        self.packet.ni_in = self.route_src[0].ni
+        self.packet.ni_in = self.route_src.ni
 
     def _update_packet_ni_out(self) -> (str, None):
         if self.packet.ni_out is not None:
             return
 
-        if len(self.route_dst) == 0:
+        if self.route_dst is None:
             return
 
-        self.packet.ni_out = self.route_dst[0].ni
+        self.packet.ni_out = self.route_dst.ni
+
+    def _is_bogon_to_wan(self) -> bool:
+        if self.route_dst.net in DEFAULT_ROUTES and \
+                not self.packet.dst.is_global:
+            return True
+
+        return False
 
 
 class Simulator:

src/firewall_test/simulator/routes.py

@@ -1,6 +1,10 @@
+from os import environ
+
 from plugins.system.abstract import FirewallSystem
 from plugins.translate.abstract import StaticRouteRule, StaticRoute
+
 from simulator.packet import PacketIP
+from simulator.logger import log_debug
 
 
 class Router:
@@ -70,18 +74,18 @@ def _get_matching_rules(self, packet: PacketIP) -> list[StaticRouteRule]:
     def _get_matching_routes(self, packet: PacketIP, matching_rules: list[StaticRouteRule], src_route: bool = False) -> list[StaticRoute]:
         # todo: ignore routes of interfaces that are down
 
+        route_for = packet.dst
+        if src_route:
+            route_for = packet.src
+
         routes = []
         if len(matching_rules) > 0:
             for rule in matching_rules:
                 for route in self.rule_route_mapping[rule]:
-                    if packet.dst in route.net:
+                    if route_for in route.net:
                         routes.append(route)
 
         else:
-            route_for = packet.dst
-            if src_route:
-                route_for = packet.src
-
             for route in self.routes:
                 if route_for in route.net:
                     routes.append(route)
@@ -123,23 +127,25 @@ def _sort_routes(self, matching_routes: list[StaticRoute]) -> list[StaticRoute]:
 
         return sorted_routes
 
-    def get_route(self, packet: PacketIP) -> list[StaticRoute]:
-        matching_rules = self._get_matching_rules(packet)
-        matching_routes = self._get_matching_routes(packet, matching_rules)
+    def get_route(self, packet: PacketIP) -> (StaticRoute, None):
+        rules = self._get_matching_rules(packet)
+        routes = self._get_matching_routes(packet, rules)
+        routes = self._sort_routes(routes)
+        if 'DEBUG' in environ:
+            log_debug('Router', f'Packet {packet.dump()} | Destination routes: {[r.dump() for r in routes]}')
 
-        print(f"ROUTE RULES MATCHED: {matching_rules}")
-        print(f"ROUTES MATCHED: {matching_routes}")
+        if len(routes) == 0:
+            return None
 
-        sorted_routes = self._sort_routes(matching_routes)
+        return routes[0]
 
-        print(f"ROUTES SORTED: {sorted_routes}")
-        return sorted_routes
-
-    def get_src_route(self, packet: PacketIP) -> list[StaticRoute]:
-        matching_routes = self._get_matching_routes(packet, [], src_route=True)
-        print(f"ROUTES MATCHED: {matching_routes}")
+    def get_src_route(self, packet: PacketIP) -> (StaticRoute, None):
+        routes = self._get_matching_routes(packet, [], src_route=True)
+        routes = self._sort_routes(routes)
+        if 'DEBUG' in environ:
+            log_debug('Router', f'Packet {packet.dump()} | Source routes: {[r.dump() for r in routes]}')
 
-        sorted_routes = self._sort_routes(matching_routes)
+        if len(routes) == 0:
+            return None
 
-        print(f"ROUTES SORTED: {sorted_routes}")
-        return sorted_routes
+        return routes[0]