Update python-app.yml

.github/workflows/python-app.yml

@@ -1,28 +1,25 @@
-# This workflow will install Python dependencies, run tests and lint with a single version of Python
+# This workflow will install Python dependencies, run tests and lint with multiple versions of Python
 # For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
-
 name: Python application
-
 on:
   push:
     branches: [ "master", "develop" ]
   pull_request:
     branches: [ "master", "develop" ]
-
 permissions:
   contents: read
-
 jobs:
   build:
-
     runs-on: ubuntu-latest
-
+    strategy:
+      matrix:
+        python-version: ["3.9", "3.10", "3.11", "3.12"]
     steps:
     - uses: actions/checkout@v3
-    - name: Set up Python 3.11
+    - name: Set up Python ${{ matrix.python-version }}
       uses: actions/setup-python@v3
       with:
-        python-version: "3.11"
+        python-version: ${{ matrix.python-version }}
     - name: Setup timezone
       uses: zcong1993/setup-timezone@master
       with:

README.md

@@ -36,7 +36,7 @@ See how is ExaFS integrated into the network in the picture below.
 
 ![ExaFS schema](./docs/app_schema_en.png)
 
-The central part of the ExaFS is a web application, written in Python3.6 with Flask framework. It provides a user interface for ExaBGP rule CRUD operations. The application also provides the REST API with CRUD operations for the configuration rules. The web app uses Shibboleth authorization; the REST API is using token-based authorization. 
+The central part of the ExaFS is a web application, written in Python with Flask framework. It provides a user interface for ExaBGP rule CRUD operations. The application also provides the REST API with CRUD operations for the configuration rules. The web app uses Shibboleth authorization; the REST API is using token-based authorization. 
 
 The app creates the ExaBGP commands and forwards them to ExaBGP process. All rules are carefully validated, and only valid rules are stored in the database and sent to the ExaBGP connector.
 

flowapp/services/rule_service.py

@@ -352,35 +352,30 @@ def evaluate_rtbh_against_whitelists_check_results(
     Process all results for cases where rule is whitelisted by several whitelists.
     """
     for rule, whitelist_key, relation in results:
-        match relation:
-            case Relation.EQUAL:
-                model = whitelist_rtbh_rule(model, wl_cache[whitelist_key])
-                msg = f"RTBH Rule {model.id} {model} is equal to active whitelist {whitelist_key}. Rule is whitelisted."
+        if relation == Relation.EQUAL:
+            model = whitelist_rtbh_rule(model, wl_cache[whitelist_key])
+            msg = f"RTBH Rule {model.id} {model} is equal to active whitelist {whitelist_key}. Rule is whitelisted."
+            flashes.append(msg)
+            current_app.logger.info(msg)
+        elif relation == Relation.SUBNET:
+            parts = subtract_network(target=str(model), whitelist=whitelist_key)
+            wl_id = wl_cache[whitelist_key].id
+            msg = f"RTBH Rule {model.id} {model} is supernet of active whitelist {whitelist_key}. Rule is whitelisted, {len(parts)} subnet rules created."
+            flashes.append(msg)
+            current_app.logger.info(msg)
+            for network in parts:
+                new_rule = create_rtbh_from_whitelist_parts(model, wl_id, whitelist_key, network, author, user_id)
+                msg = f"Created RTBH rule {new_rule.id} {new_rule} for {network} parted by whitelist {whitelist_key}"
                 flashes.append(msg)
                 current_app.logger.info(msg)
-            case Relation.SUBNET:
-                parts = subtract_network(target=str(model), whitelist=whitelist_key)
-                wl_id = wl_cache[whitelist_key].id
-                msg = f"RTBH Rule {model.id} {model} is supernet of active whitelist {whitelist_key}. Rule is whitelisted, {len(parts)} subnet rules created."
-                flashes.append(msg)
-                current_app.logger.info(msg)
-                for network in parts:
-                    new_rule = create_rtbh_from_whitelist_parts(model, wl_id, whitelist_key, network, author, user_id)
-                    msg = (
-                        f"Created RTBH rule {new_rule.id} {new_rule} for {network} parted by whitelist {whitelist_key}"
-                    )
-                    flashes.append(msg)
-                    current_app.logger.info(msg)
-                model.rstate_id = 4
-                add_rtbh_rule_to_cache(model, wl_id, RuleOrigin.USER)
-                db.session.commit()
-            case Relation.SUPERNET:
-                model = whitelist_rtbh_rule(model, wl_cache[whitelist_key])
-                msg = (
-                    f"RTBH Rule {model.id} {model} is subnet of active whitelist {whitelist_key}. Rule is whitelisted."
-                )
-                current_app.logger.info(msg)
-                flashes.append(msg)
+            model.rstate_id = 4
+            add_rtbh_rule_to_cache(model, wl_id, RuleOrigin.USER)
+            db.session.commit()
+        elif relation == Relation.SUPERNET:
+            model = whitelist_rtbh_rule(model, wl_cache[whitelist_key])
+            msg = f"RTBH Rule {model.id} {model} is subnet of active whitelist {whitelist_key}. Rule is whitelisted."
+            current_app.logger.info(msg)
+            flashes.append(msg)
     return model
 
 

flowapp/services/whitelist_common.py

@@ -1,7 +1,7 @@
 from enum import Enum, auto
 from functools import lru_cache
 import ipaddress
-from typing import List, Tuple
+from typing import List, Tuple, Union
 from flowapp import db
 from flowapp.constants import RuleOrigin, RuleTypes
 from flowapp.models import RTBH, RuleWhitelistCache, Whitelist
@@ -59,7 +59,7 @@ def _is_same_ip_version(addr1: str, addr2: str) -> bool:
 
 
 @lru_cache(maxsize=1024)
-def get_network(address: str) -> ipaddress.IPv4Network | ipaddress.IPv6Network:
+def get_network(address: str) -> Union[ipaddress.IPv4Network, ipaddress.IPv6Network]:
     """
     Create and cache an IP network object.
 

flowapp/services/whitelist_service.py

@@ -84,37 +84,34 @@ def evaluate_whitelist_against_rtbh_check_results(
 
     for rule_key, whitelist_key, relation in results:
         current_app.logger.info(f"whitelist {whitelist_key} is {relation} to Rule {rule_key}")
-        match relation:
-            case Relation.EQUAL:
-                whitelist_rtbh_rule(rtbh_rule_cache[rule_key], whitelist_model)
-                withdraw_rtbh_route(rtbh_rule_cache[rule_key])
-                msg = "Existing active rule {rule_key} is equal to whitelist {whitelist_key}. Rule is now whitelisted."
+        if relation == Relation.EQUAL:
+            whitelist_rtbh_rule(rtbh_rule_cache[rule_key], whitelist_model)
+            withdraw_rtbh_route(rtbh_rule_cache[rule_key])
+            msg = "Existing active rule {rule_key} is equal to whitelist {whitelist_key}. Rule is now whitelisted."
+            flashes.append(msg)
+            current_app.logger.info(msg)
+        elif relation == Relation.SUBNET:
+            parts = subtract_network(target=rule_key, whitelist=whitelist_key)
+            wl_id = whitelist_model.id
+            msg = f"Rule {rule_key} is supernet of whitelist {whitelist_key}. Rule is whitelisted, {len(parts)} subnet rules will be created."
+            flashes.append(msg)
+            current_app.logger.info(msg)
+            for network in parts:
+                rule_model = rtbh_rule_cache[rule_key]
+                create_rtbh_from_whitelist_parts(rule_model, wl_id, whitelist_key, network)
+                msg = f"Created RTBH rule from {rule_model.id} {network} parted by whitelist {whitelist_key}."
                 flashes.append(msg)
                 current_app.logger.info(msg)
-            case Relation.SUBNET:
-                parts = subtract_network(target=rule_key, whitelist=whitelist_key)
-                wl_id = whitelist_model.id
-                msg = f"Rule {rule_key} is supernet of whitelist {whitelist_key}. Rule is whitelisted, {len(parts)} subnet rules will be created."
-                flashes.append(msg)
-                current_app.logger.info(msg)
-                for network in parts:
-                    rule_model = rtbh_rule_cache[rule_key]
-                    create_rtbh_from_whitelist_parts(rule_model, wl_id, whitelist_key, network)
-                    msg = f"Created RTBH rule from {rule_model.id} {network} parted by whitelist {whitelist_key}."
-                    flashes.append(msg)
-                    current_app.logger.info(msg)
-                rule_model.rstate_id = 4
-                add_rtbh_rule_to_cache(rule_model, wl_id, RuleOrigin.USER)
-                db.session.commit()
-            case Relation.SUPERNET:
-
-                whitelist_rtbh_rule(rtbh_rule_cache[rule_key], whitelist_model)
-                withdraw_rtbh_route(rtbh_rule_cache[rule_key])
-                msg = (
-                    f"Existing active rule {rule_key} is subnet of whitelist {whitelist_key}. Rule is now whitelisted."
-                )
-                current_app.logger.info(msg)
-                flashes.append(msg)
+            rule_model.rstate_id = 4
+            add_rtbh_rule_to_cache(rule_model, wl_id, RuleOrigin.USER)
+            db.session.commit()
+        elif relation == Relation.SUPERNET:
+
+            whitelist_rtbh_rule(rtbh_rule_cache[rule_key], whitelist_model)
+            withdraw_rtbh_route(rtbh_rule_cache[rule_key])
+            msg = f"Existing active rule {rule_key} is subnet of whitelist {whitelist_key}. Rule is now whitelisted."
+            current_app.logger.info(msg)
+            flashes.append(msg)
 
     return whitelist_model
 
@@ -154,13 +151,12 @@ def delete_whitelist(whitelist_id: int) -> List[str]:
         for cached_rule in cached_rules:
             rule_model_type = RuleTypes(cached_rule.rtype)
             cache_entries_count = RuleWhitelistCache.count_by_rule(cached_rule.rid, rule_model_type)
-            match rule_model_type:
-                case RuleTypes.IPv4:
-                    rule_model = db.session.get(Flowspec4, cached_rule.rid)
-                case RuleTypes.IPv6:
-                    rule_model = db.session.get(Flowspec6, cached_rule.rid)
-                case RuleTypes.RTBH:
-                    rule_model = db.session.get(RTBH, cached_rule.rid)
+            if rule_model_type == RuleTypes.IPv4:
+                rule_model = db.session.get(Flowspec4, cached_rule.rid)
+            elif rule_model_type == RuleTypes.IPv6:
+                rule_model = db.session.get(Flowspec6, cached_rule.rid)
+            elif rule_model_type == RuleTypes.RTBH:
+                rule_model = db.session.get(RTBH, cached_rule.rid)
             rorigin_type = RuleOrigin(cached_rule.rorigin)
             current_app.logger.debug(f"Rule {rule_model} has origin {rorigin_type}")
             if rorigin_type == RuleOrigin.WHITELIST: