cwe: remove prohibited vulnerability-mapping entries on refresh

application/tests/cwe_parser_test.py

@@ -292,6 +292,42 @@ def iter_content(self, chunk_size=None):
 
         self.assertIn("611", imported_cwes)
         self.assertNotIn("9999", imported_cwes)
+        self.assertNotIn("16", imported_cwes)
+
+    @patch.object(requests, "get")
+    def test_register_CWE_removes_stale_prohibited_entries(self, mock_requests) -> None:
+        stale_category = self.collection.add_node(
+            defs.Standard(
+                name="CWE",
+                sectionID="16",
+                section="Configuration",
+                hyperlink="https://cwe.mitre.org/data/definitions/16.html",
+            )
+        )
+        self.collection.session.add(stale_category)
+        self.collection.session.commit()
+
+        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()
+
+        cwe.CWE().parse(
+            cache=self.collection,
+            ph=prompt_client.PromptHandler(database=self.collection),
+        )
+
+        self.assertEqual(self.collection.get_nodes(name="CWE", sectionID="16"), [])
 
     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"
@@ -773,9 +809,20 @@ def iter_content(self, chunk_size=None):
       <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">
+      <Weakness ID="9999" Name="Prohibited Example" Abstraction="Base" Structure="Simple" Status="Draft">
          <Description>This entry should not be imported.</Description>
+         <Mapping_Notes>
+            <Usage>Prohibited</Usage>
+         </Mapping_Notes>
       </Weakness>
    </Weaknesses>
+   <Categories>
+      <Category ID="16" Name="Configuration" Status="Obsolete">
+         <Summary>Prohibited category entry.</Summary>
+         <Mapping_Notes>
+            <Usage>Prohibited</Usage>
+         </Mapping_Notes>
+      </Category>
+   </Categories>
 </Weakness_Catalog>
 """

application/utils/external_project_parsers/parsers/cwe.py

@@ -5,6 +5,8 @@
 from pathlib import Path
 import requests
 from typing import Dict, List
+from sqlalchemy import func
+from sqlalchemy.exc import IntegrityError
 from application.database import db
 from application.defs import cre_defs as defs
 import shutil
@@ -70,6 +72,83 @@ def parse(self, cache: db.Node_collection, ph: prompt_client.PromptHandler):
     def make_hyperlink(self, cwe_id: int):
         return f"https://cwe.mitre.org/data/definitions/{cwe_id}.html"
 
+    def iter_catalog_entries(self, section: Dict) -> List[Dict]:
+        if not section:
+            return []
+
+        entries = []
+        for collection in section.values():
+            if isinstance(collection, list):
+                entries.extend(entry for entry in collection if isinstance(entry, Dict))
+            elif isinstance(collection, Dict):
+                entries.append(collection)
+        return entries
+
+    def get_mapping_usage(self, entry: Dict) -> str:
+        mapping_notes = entry.get("Mapping_Notes")
+        if not isinstance(mapping_notes, Dict):
+            return ""
+        usage = mapping_notes.get("Usage", "")
+        return str(usage).strip().lower()
+
+    def is_prohibited_entry(self, entry: Dict) -> bool:
+        return (
+            str(entry.get("@Status", "")).strip().lower() == "prohibited"
+            or self.get_mapping_usage(entry) == "prohibited"
+        )
+
+    def collect_prohibited_cwe_ids(self, weakness_catalog: Dict) -> List[str]:
+        prohibited_ids = []
+        for section_name in ("Weaknesses", "Categories"):
+            for entry in self.iter_catalog_entries(weakness_catalog.get(section_name)):
+                if entry.get("@ID") and self.is_prohibited_entry(entry):
+                    prohibited_ids.append(str(entry["@ID"]))
+        return prohibited_ids
+
+    def delete_cwe_entries(self, cache: db.Node_collection, cwe_ids: List[str]) -> None:
+        if not cwe_ids:
+            return
+
+        entries = (
+            cache.session.query(db.Node)
+            .filter(
+                func.lower(db.Node.name) == self.name.lower(),
+                db.Node.section_id.in_(cwe_ids),
+            )
+            .all()
+        )
+        if not entries:
+            return
+
+        for entry in entries:
+            entry_links = (
+                cache.session.query(db.Links).filter(db.Links.node == entry.id).all()
+            )
+            for link in entry_links:
+                cache.session.delete(link)
+
+            # Delete all embeddings associated with this node
+            # Use node_id (the foreign key column) instead of node (which doesn't exist)
+            embeddings = (
+                cache.session.query(db.Embeddings)
+                .filter(db.Embeddings.node_id == entry.id)
+                .all()
+            )
+            for emb in embeddings:
+                cache.session.delete(emb)
+
+            cache.session.delete(entry)
+
+        try:
+            cache.session.commit()
+            logger.info(
+                "Deleted %s prohibited CWE entries from the local database",
+                len(entries),
+            )
+        except IntegrityError as e:
+            cache.session.rollback()
+            logger.error("Failed to delete prohibited CWE entries: %s", e)
+
     def link_cwe_to_capec_cre(
         self, cwe: defs.Standard, cache: db.Node_collection, capec_id: str
     ) -> defs.Standard:
@@ -177,68 +256,72 @@ def register_cwe(self, cache: db.Node_collection, xml_file: str):
         related_ids_by_cwe = {}
         with open(xml_file, "r") as xml:
             weakness_catalog = xmltodict.parse(xml.read()).get("Weakness_Catalog")
-        for _, weaknesses in weakness_catalog.get("Weaknesses").items():
-            for weakness in weaknesses:
-                statuses[weakness["@Status"]] = 1
-                cwe = None
-                if weakness["@Status"] in self.allowed_statuses:
-                    cwes = cache.get_nodes(self.name, sectionID=weakness["@ID"])
-                    if cwes:  # update the CWE in the database
-                        cwe = cwes[0]
-                        cwe.section = weakness["@Name"]
-                        cwe.hyperlink = self.make_hyperlink(weakness["@ID"])
-                        cache.add_node(
-                            cwe,
-                            comparison_skip_attributes=[
-                                "link",
-                                "section",
-                                "version",
-                                "subsection",
-                                "tags",
-                                "description",
-                            ],
-                        )
-                    else:  # we found something new
-                        cwe = defs.Standard(
-                            name="CWE",
-                            sectionID=weakness["@ID"],
-                            section=weakness["@Name"],
-                            hyperlink=self.make_hyperlink(weakness["@ID"]),
-                            tags=base_parser_defs.build_tags(
-                                family=base_parser_defs.Family.TAXONOMY,
-                                subtype=base_parser_defs.Subtype.RISK_LIST,
-                                audience=base_parser_defs.Audience.DEVELOPER,
-                                maturity=base_parser_defs.Maturity.STABLE,
-                                source="cwe",
-                                extra=[],
-                            ),
-                        )
-                    logger.debug(f"Registered CWE with id {cwe.sectionID}")
-
-                    if weakness.get("Related_Attack_Patterns") and os.environ.get(
-                        "CRE_LINK_CWE_THROUGH_CAPEC"
-                    ):
-                        for lst in weakness["Related_Attack_Patterns"].values():
-                            for capec_entry in lst:
-                                if isinstance(capec_entry, Dict):
-                                    for _, capec_id in capec_entry.items():
-                                        cwe = self.link_cwe_to_capec_cre(
-                                            cwe=cwe, cache=cache, capec_id=capec_id
-                                        )
-                                else:
-                                    id = lst["@CAPEC_ID"]
+        prohibited_ids = self.collect_prohibited_cwe_ids(weakness_catalog)
+        self.delete_cwe_entries(cache, prohibited_ids)
+
+        for weakness in self.iter_catalog_entries(weakness_catalog.get("Weaknesses")):
+            statuses[weakness["@Status"]] = 1
+            cwe = None
+            if self.is_prohibited_entry(weakness):
+                continue
+            if weakness["@Status"] in self.allowed_statuses:
+                cwes = cache.get_nodes(self.name, sectionID=weakness["@ID"])
+                if cwes:  # update the CWE in the database
+                    cwe = cwes[0]
+                    cwe.section = weakness["@Name"]
+                    cwe.hyperlink = self.make_hyperlink(weakness["@ID"])
+                    cache.add_node(
+                        cwe,
+                        comparison_skip_attributes=[
+                            "link",
+                            "section",
+                            "version",
+                            "subsection",
+                            "tags",
+                            "description",
+                        ],
+                    )
+                else:  # we found something new
+                    cwe = defs.Standard(
+                        name="CWE",
+                        sectionID=weakness["@ID"],
+                        section=weakness["@Name"],
+                        hyperlink=self.make_hyperlink(weakness["@ID"]),
+                        tags=base_parser_defs.build_tags(
+                            family=base_parser_defs.Family.TAXONOMY,
+                            subtype=base_parser_defs.Subtype.RISK_LIST,
+                            audience=base_parser_defs.Audience.DEVELOPER,
+                            maturity=base_parser_defs.Maturity.STABLE,
+                            source="cwe",
+                            extra=[],
+                        ),
+                    )
+                logger.debug(f"Registered CWE with id {cwe.sectionID}")
+
+                if weakness.get("Related_Attack_Patterns") and os.environ.get(
+                    "CRE_LINK_CWE_THROUGH_CAPEC"
+                ):
+                    for lst in weakness["Related_Attack_Patterns"].values():
+                        for capec_entry in lst:
+                            if isinstance(capec_entry, Dict):
+                                for _, capec_id in capec_entry.items():
                                     cwe = self.link_cwe_to_capec_cre(
-                                        cwe=cwe, cache=cache, capec_id=id
+                                        cwe=cwe, cache=cache, capec_id=capec_id
                                     )
-                    else:
-                        logger.info(
-                            f"CWE '{cwe.sectionID}-{cwe.section}' does not have any related CAPEC attack patterns, skipping automated linking"
-                        )
-                    entries.append(cwe)
-                    entries_by_id[cwe.sectionID] = cwe
-                    related_ids_by_cwe[cwe.sectionID] = (
-                        self.collect_related_weakness_ids(weakness)
+                            else:
+                                id = lst["@CAPEC_ID"]
+                                cwe = self.link_cwe_to_capec_cre(
+                                    cwe=cwe, cache=cache, capec_id=id
+                                )
+                else:
+                    logger.info(
+                        f"CWE '{cwe.sectionID}-{cwe.section}' does not have any related CAPEC attack patterns, skipping automated linking"
                     )
+                entries.append(cwe)
+                entries_by_id[cwe.sectionID] = cwe
+                related_ids_by_cwe[cwe.sectionID] = self.collect_related_weakness_ids(
+                    weakness
+                )
 
         changed = True
         while changed:

scripts/update-cwe.sh

@@ -15,9 +15,9 @@ fi
 
 source "$VENV_DIR/bin/activate"
 
-if ! python -c "import pytest" >/dev/null 2>&1; then
-  echo "Installing Python development dependencies"
-  pip install -r "$ROOT_DIR/requirements-dev.txt"
+if ! python -c "import requests" >/dev/null 2>&1; then
+  echo "Installing Python runtime dependencies"
+  pip install -r "$ROOT_DIR/requirements.txt"
 fi
 
 if [[ -f "$CACHE_FILE" ]]; then
@@ -29,5 +29,5 @@ export CRE_NO_NEO4J="${CRE_NO_NEO4J:-1}"
 export CRE_NO_GEN_EMBEDDINGS="${CRE_NO_GEN_EMBEDDINGS:-1}"
 
 echo "Importing latest MITRE CWE data into $CACHE_FILE"
-echo "CWE parser will import only allowed statuses and skip PROHIBITED entries"
-exec python "$ROOT_DIR/cre.py" --cwe_in --cache_file "$CACHE_FILE"
+echo "CWE parser will skip entries marked PROHIBITED by either @Status attribute or Mapping_Notes -> Usage field, covering both weaknesses and categories."
+exec python "$ROOT_DIR/cre.py" --cwe_in --cache_file "$CACHE_FILE"