Add direct-only OpenCRE map analysis

Author: Bornunique911

application/tests/web_main_test.py

@@ -736,28 +736,53 @@ def test_gap_analysis_supports_opencre_as_standard(
 
     @patch.object(web_main.gap_analysis, "schedule")
     @patch.object(db, "Node_collection")
-    def test_gap_analysis_preserves_multiple_opencre_overlaps(
+    def test_gap_analysis_returns_only_direct_opencre_mappings(
         self, db_mock, schedule_mock
     ) -> None:
         compare = defs.Standard(
             name="CWE",
             sectionID="1004",
             section="Sensitive Cookie Without 'HttpOnly' Flag",
         )
-        opencre_documents = []
-        internal_ids = []
-
-        for i in range(8):
+        direct_cre = defs.CRE(
+            id="804-220",
+            name="Set httponly attribute for cookie-based session tokens",
+            description="",
+        )
+        direct_cre.add_link(
+            defs.Link(ltype=defs.LinkTypes.LinkedTo, document=compare.shallow_copy())
+        )
+        auto_linked_cres = []
+        for i, cre_id in enumerate(
+            [
+                "117-371",
+                "166-151",
+                "284-521",
+                "368-633",
+                "612-252",
+                "664-080",
+                "801-310",
+            ],
+            start=1,
+        ):
             cre = defs.CRE(
-                id=f"170-77{i}",
-                name=f"Cryptography {i}",
+                id=cre_id,
+                name=f"Automatically mapped CRE {i}",
                 description="",
             )
-            compare.add_link(
-                defs.Link(ltype=defs.LinkTypes.LinkedTo, document=cre.shallow_copy())
+            cre.add_link(
+                defs.Link(
+                    ltype=defs.LinkTypes.AutomaticallyLinkedTo,
+                    document=compare.shallow_copy(),
+                )
             )
-            opencre_documents.append(cre)
-            internal_ids.append(SimpleNamespace(id=f"cre-internal-{i}"))
+            auto_linked_cres.append(cre)
+
+        opencre_documents = [direct_cre] + auto_linked_cres
+        internal_ids = [
+            SimpleNamespace(id=f"cre-internal-{i}")
+            for i in range(len(opencre_documents))
+        ]
 
         db_mock.return_value.get_gap_analysis_result.return_value = None
         db_mock.return_value.gap_analysis_exists.return_value = False
@@ -782,17 +807,79 @@ def test_gap_analysis_preserves_multiple_opencre_overlaps(
         payload = json.loads(response.data)
         self.assertEqual(200, response.status_code)
         self.assertIn("result", payload)
-        self.assertIn(compare.id, payload["result"])
-        self.assertEqual(8, len(payload["result"][compare.id]["paths"]))
-        self.assertEqual(
-            8,
-            len(
-                {
-                    path["end"]["id"]
-                    for path in payload["result"][compare.id]["paths"].values()
-                }
-            ),
+        self.assertEqual([compare.id], list(payload["result"].keys()))
+        self.assertEqual(1, len(payload["result"][compare.id]["paths"]))
+        path = next(iter(payload["result"][compare.id]["paths"].values()))
+        self.assertEqual(direct_cre.id, path["end"]["id"])
+        self.assertEqual(direct_cre.name, path["end"]["name"])
+        self.assertEqual(compare.id, path["path"][0]["start"]["id"])
+        self.assertEqual(direct_cre.id, path["path"][0]["end"]["id"])
+        schedule_mock.assert_not_called()
+
+    @patch.object(web_main.gap_analysis, "schedule")
+    @patch.object(db, "Node_collection")
+    def test_gap_analysis_returns_only_direct_opencre_mappings_when_opencre_is_left(
+        self, db_mock, schedule_mock
+    ) -> None:
+        compare = defs.Standard(
+            name="CWE",
+            sectionID="1004",
+            section="Sensitive Cookie Without 'HttpOnly' Flag",
+        )
+        direct_cre = defs.CRE(
+            id="804-220",
+            name="Set httponly attribute for cookie-based session tokens",
+            description="",
+        )
+        direct_cre.add_link(
+            defs.Link(ltype=defs.LinkTypes.LinkedTo, document=compare.shallow_copy())
+        )
+        indirect_cre = defs.CRE(
+            id="117-371",
+            name="Use a centralized access control mechanism",
+            description="",
+        )
+        indirect_cre.add_link(
+            defs.Link(
+                ltype=defs.LinkTypes.AutomaticallyLinkedTo,
+                document=compare.shallow_copy(),
+            )
         )
+
+        opencre_documents = [direct_cre, indirect_cre]
+        internal_ids = [
+            SimpleNamespace(id=f"cre-internal-{i}")
+            for i in range(len(opencre_documents))
+        ]
+
+        db_mock.return_value.get_gap_analysis_result.return_value = None
+        db_mock.return_value.gap_analysis_exists.return_value = False
+        db_mock.return_value.get_nodes.side_effect = lambda name=None, **kwargs: (
+            [compare] if name == "CWE" else []
+        )
+        db_mock.return_value.session.query.return_value.all.return_value = internal_ids
+        db_mock.return_value.get_CREs.side_effect = lambda internal_id=None, **kwargs: [
+            next(
+                cre
+                for index, cre in enumerate(opencre_documents)
+                if internal_id == f"cre-internal-{index}"
+            )
+        ]
+
+        with self.app.test_client() as client:
+            response = client.get(
+                "/rest/v1/map_analysis?standard=OpenCRE&standard=CWE",
+                headers={"Content-Type": "application/json"},
+            )
+
+        payload = json.loads(response.data)
+        self.assertEqual(200, response.status_code)
+        self.assertEqual([direct_cre.id], list(payload["result"].keys()))
+        self.assertEqual(1, len(payload["result"][direct_cre.id]["paths"]))
+        path = next(iter(payload["result"][direct_cre.id]["paths"].values()))
+        self.assertEqual(compare.id, path["end"]["id"])
+        self.assertEqual(direct_cre.id, path["path"][0]["start"]["id"])
+        self.assertEqual(compare.id, path["path"][0]["end"]["id"])
         schedule_mock.assert_not_called()
 
     def test_gap_analysis_weak_links_no_cache(self) -> None:

application/web/web_main.py

@@ -310,71 +310,44 @@ def _get_map_analysis_documents(
     return collection.get_nodes(name=standard)
 
 
-def _get_document_cre_ids(document: defs.Document) -> list[str]:
-    if document.doctype == defs.Credoctypes.CRE:
-        return [document.id]
-    return [
-        link.document.id
-        for link in document.links
-        if link.document.doctype == defs.Credoctypes.CRE
-    ]
-
-
-def _build_direct_overlap_path(
-    base_document: defs.Document, cre_id: str, compare_document: defs.Document
-) -> dict[str, Any] | None:
-    if base_document.doctype == defs.Credoctypes.CRE:
-        if compare_document.doctype == defs.Credoctypes.CRE:
-            return None
-        return {
-            "end": compare_document.shallow_copy(),
-            "path": [
-                {
-                    "start": base_document.shallow_copy(),
-                    "end": compare_document.shallow_copy(),
-                    "relationship": "LINKED_TO",
-                    "score": 0,
-                }
-            ],
-            "score": 0,
-        }
-
-    if compare_document.doctype == defs.Credoctypes.CRE:
-        return {
-            "end": compare_document.shallow_copy(),
-            "path": [
-                {
-                    "start": base_document.shallow_copy(),
-                    "end": compare_document.shallow_copy(),
-                    "relationship": "LINKED_TO",
-                    "score": 0,
-                }
-            ],
-            "score": 0,
-        }
-
+def _build_direct_link_path(
+    start_document: defs.Document, end_document: defs.Document
+) -> dict[str, Any]:
     return {
-        "end": compare_document.shallow_copy(),
+        "end": end_document.shallow_copy(),
         "path": [
             {
-                "start": base_document.shallow_copy(),
-                "end": defs.CRE(id=cre_id).shallow_copy(),
+                "start": start_document.shallow_copy(),
+                "end": end_document.shallow_copy(),
                 "relationship": "LINKED_TO",
                 "score": 0,
-            },
-            {
-                "start": defs.CRE(id=cre_id).shallow_copy(),
-                "end": compare_document.shallow_copy(),
-                "relationship": "LINKED_TO",
-                "score": 0,
-            },
+            }
         ],
         "score": 0,
     }
 
 
-def _make_direct_overlap_path_key(compare_document: defs.Document, cre_id: str) -> str:
-    return f"{compare_document.id}::{cre_id}"
+def _make_direct_link_path_key(end_document: defs.Document) -> str:
+    return end_document.id
+
+
+def _add_direct_link_result(
+    grouped_paths: dict[str, dict[str, Any]],
+    start_document: defs.Document,
+    end_document: defs.Document,
+) -> None:
+    shared_paths = grouped_paths.setdefault(
+        start_document.id,
+        {
+            "start": start_document.shallow_copy(),
+            "paths": {},
+            "extra": 0,
+        },
+    )["paths"]
+    shared_paths.setdefault(
+        _make_direct_link_path_key(end_document),
+        _build_direct_link_path(start_document, end_document),
+    )
 
 
 def _build_direct_cre_overlap_map_analysis(
@@ -385,34 +358,36 @@ def _build_direct_cre_overlap_map_analysis(
     if len(standards) < 2:
         return None
 
-    base_nodes = _get_map_analysis_documents(standards[0], collection)
-    compare_nodes = _get_map_analysis_documents(standards[1], collection)
+    base_standard = standards[0]
+    compare_standard = standards[1]
+    base_nodes = _get_map_analysis_documents(base_standard, collection)
+    compare_nodes = _get_map_analysis_documents(compare_standard, collection)
     if not base_nodes or not compare_nodes:
         return None
 
-    compare_nodes_by_cre: dict[str, list[defs.Document]] = {}
-    for compare_node in compare_nodes:
-        for cre_id in _get_document_cre_ids(compare_node):
-            compare_nodes_by_cre.setdefault(cre_id, []).append(compare_node)
+    base_is_opencre = base_standard == OPENCRE_STANDARD_NAME
+    opencre_nodes = base_nodes if base_is_opencre else compare_nodes
+    standard_nodes = compare_nodes if base_is_opencre else base_nodes
+
+    standard_nodes_by_id = {
+        standard_node.id: standard_node for standard_node in standard_nodes
+    }
+    direct_pairs: list[tuple[defs.CRE, defs.Document]] = []
+    for opencre_node in opencre_nodes:
+        for link in opencre_node.links:
+            if link.ltype != defs.LinkTypes.LinkedTo:
+                continue
+            standard_node = standard_nodes_by_id.get(link.document.id)
+            if not standard_node:
+                continue
+            direct_pairs.append((opencre_node, standard_node))
 
     grouped_paths: dict[str, dict[str, Any]] = {}
-    for base_node in base_nodes:
-        shared_paths: dict[str, Any] = {}
-        for cre_id in _get_document_cre_ids(base_node):
-            for compare_node in compare_nodes_by_cre.get(cre_id, []):
-                path = _build_direct_overlap_path(base_node, cre_id, compare_node)
-                if not path:
-                    continue
-                shared_paths.setdefault(
-                    _make_direct_overlap_path_key(compare_node, cre_id), path
-                )
-
-        if shared_paths:
-            grouped_paths[base_node.id] = {
-                "start": base_node.shallow_copy(),
-                "paths": shared_paths,
-                "extra": 0,
-            }
+    for opencre_node, standard_node in direct_pairs:
+        if base_is_opencre:
+            _add_direct_link_result(grouped_paths, opencre_node, standard_node)
+        else:
+            _add_direct_link_result(grouped_paths, standard_node, opencre_node)
 
     if not grouped_paths:
         return None
@@ -431,7 +406,6 @@ def map_analysis() -> Any:
         posthog.capture(f"map_analysis", f"standards:{standards}")
 
     database = db.Node_collection()
-    standards = request.args.getlist("standard")
     standards_hash = gap_analysis.make_resources_key(standards)
 
     if OPENCRE_STANDARD_NAME in standards: