From f9fe9e0e399e2c84b2deccdb379855af8e758540 Mon Sep 17 00:00:00 2001 From: bornunique911 Date: Thu, 23 Jul 2026 22:52:22 +0530 Subject: [PATCH 1/4] cwe: remove prohibited vulnerability-mapping entries on refresh --- application/tests/cwe_parser_test.py | 49 ++++- .../external_project_parsers/parsers/cwe.py | 189 ++++++++++++------ scripts/update-cwe.sh | 2 +- 3 files changed, 179 insertions(+), 61 deletions(-) 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..7cfcdf89b 100644 --- a/application/utils/external_project_parsers/parsers/cwe.py +++ b/application/utils/external_project_parsers/parsers/cwe.py @@ -5,6 +5,7 @@ from pathlib import Path import requests from typing import Dict, List +from sqlalchemy import func from application.database import db from application.defs import cre_defs as defs import shutil @@ -70,6 +71,72 @@ 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) + + embeddings = cache.get_embeddings_for_doc(db.nodeFromDB(entry)) + if embeddings: + cache.session.delete(embeddings) + cache.session.delete(entry) + + cache.session.commit() + logger.info( + "Deleted %s prohibited CWE entries from the local database", + len(entries), + ) + def link_cwe_to_capec_cre( self, cwe: defs.Standard, cache: db.Node_collection, capec_id: str ) -> defs.Standard: @@ -177,68 +244,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..41577b893 100755 --- a/scripts/update-cwe.sh +++ b/scripts/update-cwe.sh @@ -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" +echo "CWE parser will skip entries whose vulnerability mapping is marked PROHIBITED" exec python "$ROOT_DIR/cre.py" --cwe_in --cache_file "$CACHE_FILE" From f0b3d333e5207fcff8b070056f0ea1087049acd4 Mon Sep 17 00:00:00 2001 From: bornunique911 Date: Fri, 24 Jul 2026 00:03:14 +0530 Subject: [PATCH 2/4] CWE parser: handle prohibited entries fully, delete all embeddings, and improve update script and fixed linting issue. --- .../external_project_parsers/parsers/cwe.py | 28 +++++++++++++------ scripts/update-cwe.sh | 10 +++---- 2 files changed, 25 insertions(+), 13 deletions(-) diff --git a/application/utils/external_project_parsers/parsers/cwe.py b/application/utils/external_project_parsers/parsers/cwe.py index 7cfcdf89b..3000aa01a 100644 --- a/application/utils/external_project_parsers/parsers/cwe.py +++ b/application/utils/external_project_parsers/parsers/cwe.py @@ -6,6 +6,7 @@ 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 @@ -126,16 +127,27 @@ def delete_cwe_entries(self, cache: db.Node_collection, cwe_ids: List[str]) -> N for link in entry_links: cache.session.delete(link) - embeddings = cache.get_embeddings_for_doc(db.nodeFromDB(entry)) - if embeddings: - cache.session.delete(embeddings) + # 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) - cache.session.commit() - logger.info( - "Deleted %s prohibited CWE entries from the local database", - len(entries), - ) + 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 diff --git a/scripts/update-cwe.sh b/scripts/update-cwe.sh index 41577b893..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 skip entries whose vulnerability mapping is marked PROHIBITED" -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 From ee2cfcf42d7d1627d16f2d01350304f2d228862f Mon Sep 17 00:00:00 2001 From: bornunique911 Date: Sat, 1 Aug 2026 00:05:52 +0530 Subject: [PATCH 3/4] fix(cwe): broaden DB exception handling, remove redundant test writes, and rename shadowed variable --- application/tests/cwe_parser_test.py | 4 +--- application/utils/external_project_parsers/parsers/cwe.py | 8 ++++---- 2 files changed, 5 insertions(+), 7 deletions(-) diff --git a/application/tests/cwe_parser_test.py b/application/tests/cwe_parser_test.py index 1af4fc625..1b07b01a6 100644 --- a/application/tests/cwe_parser_test.py +++ b/application/tests/cwe_parser_test.py @@ -296,7 +296,7 @@ def iter_content(self, chunk_size=None): @patch.object(requests, "get") def test_register_CWE_removes_stale_prohibited_entries(self, mock_requests) -> None: - stale_category = self.collection.add_node( + self.collection.add_node( defs.Standard( name="CWE", sectionID="16", @@ -304,8 +304,6 @@ def test_register_CWE_removes_stale_prohibited_entries(self, mock_requests) -> N 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") diff --git a/application/utils/external_project_parsers/parsers/cwe.py b/application/utils/external_project_parsers/parsers/cwe.py index 3000aa01a..509668433 100644 --- a/application/utils/external_project_parsers/parsers/cwe.py +++ b/application/utils/external_project_parsers/parsers/cwe.py @@ -6,7 +6,7 @@ import requests from typing import Dict, List from sqlalchemy import func -from sqlalchemy.exc import IntegrityError +from sqlalchemy.exc import IntegrityError, SQLAlchemyError from application.database import db from application.defs import cre_defs as defs import shutil @@ -145,7 +145,7 @@ def delete_cwe_entries(self, cache: db.Node_collection, cwe_ids: List[str]) -> N "Deleted %s prohibited CWE entries from the local database", len(entries), ) - except IntegrityError as e: + except SQLAlchemyError as e: cache.session.rollback() logger.error("Failed to delete prohibited CWE entries: %s", e) @@ -309,9 +309,9 @@ def register_cwe(self, cache: db.Node_collection, xml_file: str): cwe=cwe, cache=cache, capec_id=capec_id ) else: - id = lst["@CAPEC_ID"] + capec_id_value = lst["@CAPEC_ID"] cwe = self.link_cwe_to_capec_cre( - cwe=cwe, cache=cache, capec_id=id + cwe=cwe, cache=cache, capec_id=capec_id_value ) else: logger.info( From 60547512fd84bfbf4abda86aa02784ab9550491c Mon Sep 17 00:00:00 2001 From: bornunique911 Date: Sun, 2 Aug 2026 12:00:31 +0530 Subject: [PATCH 4/4] fix(cwe): skip prohibited entries by Status or Mapping_Notes and remove stale ones --- application/utils/external_project_parsers/parsers/cwe.py | 1 + scripts/update-cwe.sh | 8 +++----- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/application/utils/external_project_parsers/parsers/cwe.py b/application/utils/external_project_parsers/parsers/cwe.py index 509668433..6120f7c18 100644 --- a/application/utils/external_project_parsers/parsers/cwe.py +++ b/application/utils/external_project_parsers/parsers/cwe.py @@ -148,6 +148,7 @@ def delete_cwe_entries(self, cache: db.Node_collection, cwe_ids: List[str]) -> N except SQLAlchemyError as e: cache.session.rollback() logger.error("Failed to delete prohibited CWE entries: %s", e) + raise RuntimeError("Failed to delete prohibited CWE entries") from e def link_cwe_to_capec_cre( self, cwe: defs.Standard, cache: db.Node_collection, capec_id: str diff --git a/scripts/update-cwe.sh b/scripts/update-cwe.sh index 421e96716..24ab81215 100755 --- a/scripts/update-cwe.sh +++ b/scripts/update-cwe.sh @@ -15,10 +15,8 @@ fi source "$VENV_DIR/bin/activate" -if ! python -c "import requests" >/dev/null 2>&1; then - echo "Installing Python runtime dependencies" - pip install -r "$ROOT_DIR/requirements.txt" -fi +echo "Installing Python runtime dependencies" +pip install -r "$ROOT_DIR/requirements.txt" if [[ -f "$CACHE_FILE" ]]; then cp "$CACHE_FILE" "$BACKUP_FILE" @@ -30,4 +28,4 @@ export CRE_NO_GEN_EMBEDDINGS="${CRE_NO_GEN_EMBEDDINGS:-1}" echo "Importing latest MITRE CWE data into $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 +exec python "$ROOT_DIR/cre.py" --cwe_in --cache_file "$CACHE_FILE"