[Fix] Overlapping stored flows

CHANGELOG.rst

@@ -21,6 +21,13 @@ Removed
 Security
 ========
 
+[5.1.0] - 2021-11.08
+********************
+
+Added
+=====
+- Augmented ``_add_flow_store`` to overwrite overlapping flows
+
 [5.0.0] - 2021-11.05
 ********************
 

kytos.json

@@ -3,7 +3,7 @@
   "username": "kytos",
   "name": "flow_manager",
   "description": "Manage switches' flows through a REST API.",
-  "version": "5.0.0",
+  "version": "5.1.0",
   "napp_dependencies": ["kytos/of_core", "kytos/storehouse"],
   "license": "MIT",
   "url": "https://github.com/kytos/flow_manager.git",

main.py

@@ -7,7 +7,7 @@
 from threading import Lock
 
 from flask import jsonify, request
-from napps.kytos.flow_manager.match import match_flow
+from napps.kytos.flow_manager.match import match_flow, match_strict_flow
 from napps.kytos.flow_manager.storehouse import StoreHouse
 from napps.kytos.of_core.flow import FlowFactory
 from napps.kytos.of_core.settings import STATS_INTERVAL
@@ -314,9 +314,8 @@ def _del_matched_flows_store(self, flow_dict, switch):
             del stored_flows_box["id"]
             self.stored_flows = deepcopy(stored_flows_box)
 
-    # pylint: disable=fixme
     def _add_flow_store(self, flow_dict, switch):
-        """Try to add a flow dict in the store."""
+        """Try to add a flow dict in the store idempotently."""
         installed_flow = {}
         installed_flow["flow"] = flow_dict
         installed_flow["created_at"] = now().strftime("%Y-%m-%dT%H:%M:%S")
@@ -326,11 +325,23 @@ def _add_flow_store(self, flow_dict, switch):
         if switch.id not in stored_flows_box:
             stored_flows_box[switch.id] = OrderedDict()
 
-        # TODO handle issue 23 (overlapping FlowMod add)
         if not stored_flows_box[switch.id].get(cookie):
             stored_flows_box[switch.id][cookie] = [installed_flow]
         else:
-            stored_flows_box[switch.id][cookie].append(installed_flow)
+            version = switch.connection.protocol.version
+            stored_flows = stored_flows_box[switch.id].get(cookie, [])
+            for i, stored_flow in enumerate(stored_flows):
+                if all(
+                    (
+                        stored_flow["flow"].get("priority", 0)
+                        == flow_dict.get("priority", 0),
+                        match_strict_flow(flow_dict, version, stored_flow["flow"]),
+                    )
+                ):
+                    stored_flows_box[switch.id][cookie][i] = installed_flow
+                    break
+            else:
+                stored_flows_box[switch.id][cookie].append(installed_flow)
 
         stored_flows_box["id"] = "flow_persistence"
         self.storehouse.save_flow(stored_flows_box)

match.py

@@ -21,6 +21,17 @@ def match_flow(flow_to_install, version, stored_flow_dict):
     raise NotImplementedError(f"Unsupported OpenFlow version {version}")
 
 
+def match_strict_flow(flow_to_install, version, stored_flow_dict) -> None:
+    """Match the flow strictly.
+
+    It has support for only for (OF 1.3) flows.
+    If all fields match, return the flow, otherwise return False.
+    """
+    if version != 0x04:
+        raise NotImplementedError(f"Unsupported OpenFlow version {version}")
+    return match13_strict(flow_to_install, stored_flow_dict)
+
+
 def _get_match_fields(flow_dict):
     """Generate match fields."""
     match_fields = {}
@@ -105,17 +116,34 @@ def match10_no_strict(flow_dict, args):
     return flow_dict
 
 
-def match13_no_strict(flow_to_install, stored_flow_dict):
-    """Match a flow that is either exact or more specific (non-strict) (OF1.3).
-
-    Return the flow if any fields match, otherwise, return False.
-    """
+def _match_cookie(flow_to_install, stored_flow_dict):
+    """Check if a the cookie and its mask matches between the flows."""
     cookie = flow_to_install.get("cookie", 0) & flow_to_install.get("cookie_mask", 0)
     cookie_stored = stored_flow_dict.get("cookie", 0) & flow_to_install.get(
         "cookie_mask", 0
     )
     if cookie and cookie != cookie_stored:
         return False
+    return True
+
+
+def _match_keys(flow_to_install, stored_flow_dict, flow_to_install_keys):
+    """Check if certain keys on flow_to_install match on stored_flow_dict."""
+    for key in flow_to_install_keys:
+        if key not in stored_flow_dict["match"]:
+            return False
+        if flow_to_install["match"][key] != stored_flow_dict["match"].get(key):
+            return False
+    return True
+
+
+def match13_no_strict(flow_to_install, stored_flow_dict):
+    """Match a flow that is either exact or more specific (non-strict) (OF1.3).
+
+    Return the flow if any fields match, otherwise, return False.
+    """
+    if not _match_cookie(flow_to_install, stored_flow_dict):
+        return False
 
     if "match" not in flow_to_install or "match" not in stored_flow_dict:
         return stored_flow_dict
@@ -124,10 +152,35 @@ def match13_no_strict(flow_to_install, stored_flow_dict):
     if len(flow_to_install["match"]) > len(stored_flow_dict["match"]):
         return False
 
-    for key, value in flow_to_install.get("match").items():
-        if key not in stored_flow_dict["match"]:
-            return False
-        if value != stored_flow_dict["match"].get(key):
-            return False
+    if not _match_keys(
+        flow_to_install, stored_flow_dict, flow_to_install["match"].keys()
+    ):
+        return False
+    return stored_flow_dict
+
 
+def match13_strict(flow_to_install, stored_flow_dict):
+    """Match a flow strictly (OF1.3).
+
+    Return the flow if all fields match, otherwise, return False.
+    """
+    if not _match_cookie(flow_to_install, stored_flow_dict):
+        return False
+    if flow_to_install.get("priority", 0) != stored_flow_dict.get("priority", 0):
+        return False
+
+    if "match" not in flow_to_install and "match" not in stored_flow_dict:
+        return stored_flow_dict
+    if "match" not in flow_to_install and "match" in stored_flow_dict:
+        return False
+    if "match" in flow_to_install and "match" not in stored_flow_dict:
+        return False
+
+    if len(flow_to_install["match"]) != len(stored_flow_dict["match"]):
+        return False
+
+    if not _match_keys(
+        flow_to_install, stored_flow_dict, flow_to_install["match"].keys()
+    ):
+        return False
     return stored_flow_dict

setup.py

@@ -24,7 +24,7 @@
 BASE_ENV = Path(os.environ.get("VIRTUAL_ENV", "/"))
 
 NAPP_NAME = "flow_manager"
-NAPP_VERSION = "5.0.0"
+NAPP_VERSION = "5.1.0"
 
 # Kytos var folder
 VAR_PATH = BASE_ENV / "var" / "lib" / "kytos"

tests/unit/test_main.py

@@ -2,6 +2,9 @@
 from unittest import TestCase
 from unittest.mock import MagicMock, patch
 
+# pylint: disable=too-many-lines,fixme
+# TODO split this test suite in smaller ones
+
 from kytos.core.helpers import now
 from kytos.lib.helpers import (
     get_connection_mock,
@@ -405,6 +408,156 @@ def test_check_switch_consistency_add(self, *args):
         self.napp.check_switch_consistency(switch)
         mock_install_flows.assert_called()
 
+    @patch("napps.kytos.flow_manager.storehouse.StoreHouse.save_flow")
+    def test_add_overlapping_flow(self, *args):
+        """Test add an overlapping flow."""
+        (_,) = args
+        dpid = "00:00:00:00:00:00:00:01"
+        switch = get_switch_mock(dpid, 0x04)
+        switch.id = dpid
+        cookie = 0x20
+
+        self.napp.stored_flows = {
+            dpid: {
+                cookie: [
+                    {
+                        "flow": {
+                            "priority": 10,
+                            "cookie": 84114904,
+                            "match": {
+                                "ipv4_src": "192.168.1.1",
+                                "ipv4_dst": "192.168.0.2",
+                            },
+                            "actions": [{"action_type": "output", "port": 2}],
+                        }
+                    }
+                ]
+            }
+        }
+
+        new_actions = [{"action_type": "output", "port": 3}]
+        flow_dict = {
+            "priority": 10,
+            "cookie": cookie,
+            "match": {
+                "ipv4_src": "192.168.1.1",
+                "ipv4_dst": "192.168.0.2",
+            },
+            "actions": new_actions,
+        }
+
+        self.napp._add_flow_store(flow_dict, switch)
+        assert len(self.napp.stored_flows[dpid]) == 1
+        assert self.napp.stored_flows[dpid][0x20][0]["flow"]["actions"] == new_actions
+
+    @patch("napps.kytos.flow_manager.storehouse.StoreHouse.save_flow")
+    def test_add_overlapping_flow_multiple_stored(self, *args):
+        """Test add an overlapping flow with multiple flows stored."""
+        (_,) = args
+        dpid = "00:00:00:00:00:00:00:01"
+        switch = get_switch_mock(dpid, 0x04)
+        switch.id = dpid
+        cookie = 0x20
+
+        stored_flows_list = [
+            {
+                "flow": {
+                    "actions": [{"action_type": "output", "port": 2}],
+                    "match": {"dl_vlan": 100, "in_port": 1},
+                    "priority": 10,
+                },
+            },
+            {
+                "flow": {
+                    "actions": [{"action_type": "output", "port": 3}],
+                    "match": {"dl_vlan": 200, "in_port": 1},
+                    "priority": 10,
+                },
+            },
+            {
+                "flow": {
+                    "actions": [{"action_type": "output", "port": 4}],
+                    "match": {"dl_vlan": 300, "in_port": 1},
+                    "priority": 10,
+                },
+            },
+            {
+                "flow": {
+                    "actions": [{"action_type": "output", "port": 4}],
+                    "match": {"in_port": 1},
+                    "priority": 10,
+                },
+            },
+        ]
+
+        self.napp.stored_flows = {dpid: {cookie: list(stored_flows_list)}}
+
+        new_actions = [{"action_type": "output", "port": 3}]
+        overlapping_flow = {
+            "priority": 10,
+            "cookie": cookie,
+            "match": {
+                "in_port": 1,
+            },
+            "actions": new_actions,
+        }
+
+        self.napp._add_flow_store(overlapping_flow, switch)
+        assert len(self.napp.stored_flows[dpid][cookie]) == len(stored_flows_list)
+
+        # only the last flow is expected to be strictly matched
+        self.assertDictEqual(
+            self.napp.stored_flows[dpid][cookie][len(stored_flows_list) - 1]["flow"],
+            overlapping_flow,
+        )
+
+        # all flows except the last one should still be the same
+        for i in range(0, len(stored_flows_list) - 1):
+            self.assertDictEqual(
+                self.napp.stored_flows[dpid][cookie][i], stored_flows_list[i]
+            )
+
+    @patch("napps.kytos.flow_manager.storehouse.StoreHouse.save_flow")
+    def test_add_overlapping_flow_diff_priority(self, *args):
+        """Test that a different priority wouldn't overlap."""
+        (_,) = args
+        dpid = "00:00:00:00:00:00:00:01"
+        switch = get_switch_mock(dpid, 0x04)
+        switch.id = dpid
+
+        cookie = 0x20
+        self.napp.stored_flows = {
+            dpid: {
+                cookie: [
+                    {
+                        "flow": {
+                            "priority": 10,
+                            "cookie": 84114904,
+                            "match": {
+                                "ipv4_src": "192.168.1.1",
+                                "ipv4_dst": "192.168.0.2",
+                            },
+                            "actions": [{"action_type": "output", "port": 2}],
+                        }
+                    }
+                ]
+            }
+        }
+
+        new_actions = [{"action_type": "output", "port": 3}]
+        flow_dict = {
+            "priority": 11,
+            "cookie": cookie,
+            "match": {
+                "ipv4_src": "192.168.1.1",
+                "ipv4_dst": "192.168.0.2",
+            },
+            "actions": new_actions,
+        }
+
+        self.napp._add_flow_store(flow_dict, switch)
+        assert len(self.napp.stored_flows[dpid][cookie]) == 2
+
     @patch("napps.kytos.flow_manager.main.Main._install_flows")
     @patch("napps.kytos.flow_manager.main.FlowFactory.get_class")
     def test_check_switch_flow_not_missing(self, *args):