diff --git a/application/tests/cwe_parser_test.py b/application/tests/cwe_parser_test.py index 6f12b18c4..1af4fc625 100644 --- a/application/tests/cwe_parser_test.py +++ b/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 = """ Allowed XXE entry. - + This entry should not be imported. + + Prohibited + + + + Prohibited category entry. + + Prohibited + + + """ diff --git a/application/utils/external_project_parsers/parsers/cwe.py b/application/utils/external_project_parsers/parsers/cwe.py index 36fadbb36..3000aa01a 100644 --- a/application/utils/external_project_parsers/parsers/cwe.py +++ b/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: diff --git a/scripts/update-cwe.sh b/scripts/update-cwe.sh index 5ea22e08e..421e96716 100755 --- a/scripts/update-cwe.sh +++ b/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" \ No newline at end of file