Merge branch 'main' into feat/frontend-useuser-header-auth

Author: skypank-coder

application/database/db.py

@@ -23,7 +23,7 @@
 from flask_sqlalchemy.model import DefaultMeta
 from sqlalchemy import func, delete, cast as sql_cast, literal, or_
 from sqlalchemy.dialects.postgresql import JSONB
-from sqlalchemy.exc import OperationalError, IntegrityError
+from sqlalchemy.exc import OperationalError, IntegrityError, SQLAlchemyError
 
 from neomodel import (
     config,
@@ -2162,6 +2162,10 @@ def text_search(self, text: str) -> List[Optional[cre_defs.Document]]:
                 CRE ids before it performs a free text search
            Anything else will be a case insensitive LIKE query in the database
         """
+        search_terms = self._expand_text_search_aliases(text)
+        if not search_terms:
+            return []
+        text = text.strip()
         # structured text search first
         cre_id_search = r"CRE(:| )(?P<id>\d+-\d+)"
         cre_naked_id_search = r"\d\d\d-\d\d\d"
@@ -2213,34 +2217,81 @@ def text_search(self, text: str) -> List[Optional[cre_defs.Document]]:
             if results:
                 return list(set(results))
         # fuzzy matches second
-        args = [f"%{text}%", "", "", "", "", ""]
         results = {}
-        s = set([p for p in permutations(args, 6)])
-        for combo in s:
-            nodes = self.get_nodes(
-                name=combo[0],
-                section=combo[1],
-                subsection=combo[2],
-                link=combo[3],
-                description=combo[4],
-                partial=True,
-                ntype=None,  # type: ignore
-                sectionID=combo[5],
-            )
-            if nodes:
-                for node in nodes:
-                    node_key = f"{node.name}:{node.version}:{node.section}:{node.sectionID}:{node.subsection}:"
-                    results[node_key] = node
-        args = [f"%{text}%", None, None]
-        for combo in permutations(args, 3):
-            cres = self.get_CREs(
-                name=combo[0], external_id=combo[1], description=combo[2], partial=True
-            )
-            if cres:
-                for cre in cres:
-                    results[cre.id] = cre
+        for search_term in search_terms:
+            args = [f"%{search_term}%", "", "", "", "", ""]
+            s = set([p for p in permutations(args, 6)])
+            for combo in s:
+                nodes = self.get_nodes(
+                    name=combo[0],
+                    section=combo[1],
+                    subsection=combo[2],
+                    link=combo[3],
+                    description=combo[4],
+                    partial=True,
+                    ntype=None,  # type: ignore
+                    sectionID=combo[5],
+                )
+                if nodes:
+                    for node in nodes:
+                        node_key = f"{node.name}:{node.version}:{node.section}:{node.sectionID}:{node.subsection}:"
+                        results[node_key] = node
+            args = [f"%{search_term}%", None, None]
+            for combo in permutations(args, 3):
+                cres = self.get_CREs(
+                    name=combo[0],
+                    external_id=combo[1],
+                    description=combo[2],
+                    partial=True,
+                )
+                if cres:
+                    for cre in cres:
+                        results[cre.id] = cre
         return list(results.values())
 
+    def _expand_text_search_aliases(self, text: str) -> List[str]:
+        normalized = text.strip()
+        if not normalized:
+            return []
+        lowered = normalized.lower()
+        aliases = {
+            "xxe": [
+                "XXE",
+                "XML External Entity",
+                "XML External Entity Reference",
+            ],
+            "xss": [
+                "XSS",
+                "Cross Site Scripting",
+                "Cross-Site Scripting",
+            ],
+            "ssrf": [
+                "SSRF",
+                "Server Side Request Forgery",
+                "Server-Side Request Forgery",
+            ],
+            "csrf": [
+                "CSRF",
+                "Cross Site Request Forgery",
+                "Cross-Site Request Forgery",
+            ],
+            "rce": [
+                "RCE",
+                "Remote Code Execution",
+            ],
+        }
+        expanded = [normalized]
+        expanded.extend(aliases.get(lowered, []))
+        deduped = []
+        seen = set()
+        for entry in expanded:
+            key = entry.lower()
+            if key in seen:
+                continue
+            seen.add(key)
+            deduped.append(entry)
+        return deduped
+
     def get_root_cres(self):
         """Returns CRES that only have "Contains" links"""
         # select name  from cre where cre.id not in (select cre from cre_links where type="Contains") and cre.id not in (select "group" from cre_links where type="Is Part OF");
@@ -2289,7 +2340,7 @@ def health_check(self) -> Dict[str, Any]:
                 "db_reachable": False,
                 "reason": "database unreachable",
             }
-        except Exception:  # pragma: no cover - defensive, never fail open
+        except SQLAlchemyError:  # pragma: no cover - defensive, never fail open
             return {
                 "ok": False,
                 "db_reachable": False,

application/tests/cwe_parser_test.py

@@ -267,6 +267,32 @@ def iter_content(self, chunk_size=None):
             deserialization_cre.todict(),
         )
 
+    @patch.object(requests, "get")
+    def test_register_CWE_skips_prohibited_entries(self, mock_requests) -> None:
+        tmpdir = mkdtemp()
+        tmpFile = os.path.join(tmpdir, "cwe.xml")
+        tmpzip = os.path.join(tmpdir, "cwe.zip")
+        with open(tmpFile, "w") as cx:
+            cx.write(self.CWE_prohibited_xml)
+        with zipfile.ZipFile(tmpzip, "w", zipfile.ZIP_DEFLATED) as zipf:
+            zipf.write(tmpFile, arcname="cwe.xml")
+
+        class fakeRequest:
+            def iter_content(self, chunk_size=None):
+                with open(tmpzip, "rb") as zipf:
+                    return [zipf.read()]
+
+        mock_requests.return_value = fakeRequest()
+
+        entries = cwe.CWE().parse(
+            cache=self.collection,
+            ph=prompt_client.PromptHandler(database=self.collection),
+        )
+        imported_cwes = {node.sectionID: node for node in entries.results["CWE"]}
+
+        self.assertIn("611", imported_cwes)
+        self.assertNotIn("9999", imported_cwes)
+
     CWE_xml = """<?xml version="1.0" encoding="UTF-8"?>
 <Weakness_Catalog Name="CWE" Version="4.10" Date="2023-01-31" xmlns="http://cwe.mitre.org/cwe-6"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
@@ -739,4 +765,17 @@ def iter_content(self, chunk_size=None):
       </Weakness>
    </Weaknesses>
 </Weakness_Catalog>
+"""
+
+    CWE_prohibited_xml = """<?xml version="1.0" encoding="UTF-8"?>
+<Weakness_Catalog Name="CWE" Version="4.10" Date="2023-01-31" xmlns="http://cwe.mitre.org/cwe-6">
+   <Weaknesses>
+      <Weakness ID="611" Name="Improper Restriction of XML External Entity Reference" Abstraction="Base" Structure="Simple" Status="Draft">
+         <Description>Allowed XXE entry.</Description>
+      </Weakness>
+      <Weakness ID="9999" Name="Prohibited Example" Abstraction="Base" Structure="Simple" Status="PROHIBITED">
+         <Description>This entry should not be imported.</Description>
+      </Weakness>
+   </Weaknesses>
+</Weakness_Catalog>
 """

application/tests/db_test.py

@@ -1054,6 +1054,33 @@ def test_text_search(self) -> None:
             res = self.collection.text_search(k)
             self.assertCountEqual(res, val)
 
+    def test_text_search_expands_xxe_aliases(self) -> None:
+        xxe_cre = defs.CRE(
+            id="764-507",
+            name="Restrict XML parsing (against XXE)",
+        )
+        cwe_611 = defs.Standard(
+            name="CWE",
+            section="Improper Restriction of XML External Entity Reference",
+            sectionID="611",
+            hyperlink="https://cwe.mitre.org/data/definitions/611.html",
+        )
+
+        self.collection.add_cre(xxe_cre)
+        self.collection.add_node(cwe_611)
+
+        results = self.collection.text_search("XXE")
+
+        self.assertIn("764-507", [doc.id for doc in results if doc.doctype == "CRE"])
+        self.assertTrue(
+            any(
+                doc.doctype == "Standard"
+                and doc.name == "CWE"
+                and doc.sectionID == "611"
+                for doc in results
+            )
+        )
+
     def test_dbNodeFromNode(self) -> None:
         data = {
             "tool": defs.Tool(