From a39f8a73746ac4f478abec69270ae2f9e7464152 Mon Sep 17 00:00:00 2001 From: Shaurya Upadhyay Date: Tue, 21 Jul 2026 17:52:50 +0530 Subject: [PATCH 1/6] =?UTF-8?q?fix:=20code=20quality=20round=202=20?= =?UTF-8?q?=E2=80=94=20typo,=20dead=20import,=20PEP=208=20identity=20check?= =?UTF-8?q?s,=20unnecessary=20f-strings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix misspelled variable pentalty -> penalty in gap_analysis.py (3 occurrences) - Remove unused import json as _json in cre_main.py - Use is None instead of == None per PEP 8 (6 occurrences across 5 files) - Strip unnecessary f-string prefixes from strings with no placeholders (production code only) --- application/cmd/cre_main.py | 9 +++---- application/database/db.py | 10 +++---- application/database/inmemory_graph.py | 2 +- application/prompt_client/prompt_client.py | 4 +-- .../parsers/export_format_parser.py | 2 +- application/utils/gap_analysis.py | 6 ++--- application/utils/oscal_utils.py | 2 +- application/utils/spreadsheet_parsers.py | 4 +-- application/web/web_main.py | 26 +++++++++---------- 9 files changed, 32 insertions(+), 33 deletions(-) diff --git a/application/cmd/cre_main.py b/application/cmd/cre_main.py index 82b892c7a..5f137d96d 100644 --- a/application/cmd/cre_main.py +++ b/application/cmd/cre_main.py @@ -11,7 +11,6 @@ from collections import deque from typing import Any, Callable, Dict, List, Optional, Tuple, TYPE_CHECKING import hashlib -import json as _json from rq import Queue, job, exceptions from sqlalchemy import not_ @@ -1256,13 +1255,13 @@ def populate_neo4j_db(cache: str): ): logger.info("Skipping Neo4j population as per environment variables") return - logger.info(f"Populating neo4j DB: Connecting to SQL DB") + logger.info("Populating neo4j DB: Connecting to SQL DB") database = db_connect(path=cache) if database.neo_db: - logger.info(f"Populating neo4j DB: Populating") + logger.info("Populating neo4j DB: Populating") database.neo_db.populate_DB(database.session) - logger.info(f"Populating neo4j DB: Complete") + logger.info("Populating neo4j DB: Complete") else: logger.warning( - f"Populating neo4j DB: database.neo_db is None, skipping population" + "Populating neo4j DB: database.neo_db is None, skipping population" ) diff --git a/application/database/db.py b/application/database/db.py index a7c50b443..da33d099f 100644 --- a/application/database/db.py +++ b/application/database/db.py @@ -722,7 +722,7 @@ class NeoDocument(StructuredNode): @classmethod def to_cre_def(self, node, parse_links=True): - raise Exception(f"Shouldn't be parsing a NeoDocument") + raise Exception("Shouldn't be parsing a NeoDocument") @classmethod def get_links(self, links_dict): @@ -744,7 +744,7 @@ class NeoNode(NeoDocument): @classmethod def to_cre_def(self, node, parse_links=True): - raise Exception(f"Shouldn't be parsing a NeoNode") + raise Exception("Shouldn't be parsing a NeoNode") class NeoStandard(NeoNode): @@ -2352,7 +2352,7 @@ def add_internal_link( ltype (cre_defs.LinkTypes, optional): the linktype Returns: the cre_defs.Link or None in case of error (cycle) """ - if ltype == None: + if ltype is None: raise ValueError("Every link should have a link type") if ltype == cre_defs.LinkTypes.PartOf: @@ -3280,7 +3280,7 @@ def gap_analysis( key = node.id if not key: logger.error( - f"key is empty, this is a bug and this gap analysis will not progress" + "key is empty, this is a bug and this gap analysis will not progress" ) continue if key not in grouped_paths: @@ -3306,7 +3306,7 @@ def gap_analysis( end_key = path["end"].id if not end_key: logger.error( - f"end_key is empty, this is a bug and this gap analysis will not progress" + "end_key is empty, this is a bug and this gap analysis will not progress" ) continue path["score"] = get_path_score(path) diff --git a/application/database/inmemory_graph.py b/application/database/inmemory_graph.py index be747326e..e4407899f 100644 --- a/application/database/inmemory_graph.py +++ b/application/database/inmemory_graph.py @@ -81,7 +81,7 @@ def get_hierarchy(self, rootIDs: List[str], creID: str): if creID in rootIDs: return 0 - if self.__parent_child_subgraph == None: + if self.__parent_child_subgraph is None: if len(self.__graph.edges) == 0: raise ValueError("Graph has no edges") include_cres = [] diff --git a/application/prompt_client/prompt_client.py b/application/prompt_client/prompt_client.py index 9527df765..add7663f2 100644 --- a/application/prompt_client/prompt_client.py +++ b/application/prompt_client/prompt_client.py @@ -396,7 +396,7 @@ def find_missing_embeddings(self, database: db.Node_collection) -> List[str]: Returns: List[str]: a list of db ids which do not have embeddings """ - logger.info(f"syncing nodes with embeddings") + logger.info("syncing nodes with embeddings") missing_embeddings = [] for doc_type in cre_defs.Credoctypes: db_ids = [] @@ -1211,7 +1211,7 @@ def get_id_of_most_similar_cre_paginated( if max_similarity < similarity_threshold: logger.info( - f"there is no good cre candidate for this standard section, returning nothing" + "there is no good cre candidate for this standard section, returning nothing" ) return None, None return most_similar_id, max_similarity diff --git a/application/utils/external_project_parsers/parsers/export_format_parser.py b/application/utils/external_project_parsers/parsers/export_format_parser.py index 699accfe0..f07cab84c 100644 --- a/application/utils/external_project_parsers/parsers/export_format_parser.py +++ b/application/utils/external_project_parsers/parsers/export_format_parser.py @@ -85,7 +85,7 @@ def parse_export_format(lfile: List[Dict[str, Any]]) -> Dict[str, List[defs.Docu logger.warning( f"Link between {highest_cre.name} and {working_cre.name} already exists" ) - elif highest_cre == None: + elif highest_cre is None: highest_cre = working_cre highest_index = i diff --git a/application/utils/gap_analysis.py b/application/utils/gap_analysis.py index 2c9099c9a..3160b0ef1 100644 --- a/application/utils/gap_analysis.py +++ b/application/utils/gap_analysis.py @@ -97,9 +97,9 @@ def get_path_score(path): if step["relationship"] == "CONTAINS": penalty_type = f"CONTAINS_{get_relation_direction(step, previous_id)}" - pentalty = PENALTIES[penalty_type] - score += pentalty - step["score"] = pentalty + penalty = PENALTIES[penalty_type] + score += penalty + step["score"] = penalty previous_id = get_next_id(step, previous_id) return score diff --git a/application/utils/oscal_utils.py b/application/utils/oscal_utils.py index 9f4228557..0b26a6758 100644 --- a/application/utils/oscal_utils.py +++ b/application/utils/oscal_utils.py @@ -47,7 +47,7 @@ def document_to_oscal( version=version, links=[common.Link(href=hyperlink)], ) - if uuid == None or uuid == "": + if uuid is None or uuid == "": uuid = str(uuid4()) c = catalog.Catalog(metadata=m, uuid=uuid) controls: List[catalog.Control] = [] diff --git a/application/utils/spreadsheet_parsers.py b/application/utils/spreadsheet_parsers.py index 638fe3b0f..1d069aae2 100644 --- a/application/utils/spreadsheet_parsers.py +++ b/application/utils/spreadsheet_parsers.py @@ -290,7 +290,7 @@ def parse_export_format(lfile: List[Dict[str, Any]]) -> Dict[str, List[defs.Docu logger.warning( f"Link between {highest_cre.name} and {working_cre.name} already exists" ) - elif highest_cre == None: + elif highest_cre is None: highest_cre = working_cre highest_index = i @@ -499,7 +499,7 @@ def parse_hierarchical_export_format( current_hierarchy, name = get_highest_cre_name( mapping=mapping, highest_hierarchy=max_hierarchy ) - if name == None: # skip empty lines + if name is None: # skip empty lines continue if current_hierarchy > 0: # find the previous higher CRE so we can link diff --git a/application/web/web_main.py b/application/web/web_main.py index 4075eda61..49c17576b 100644 --- a/application/web/web_main.py +++ b/application/web/web_main.py @@ -211,7 +211,7 @@ def find_node_by_name( sectionID: str = "", ) -> Any: if posthog: - posthog.capture(f"find_node_by_name", f"name:{name};nodeType{ntype}") + posthog.capture("find_node_by_name", f"name:{name};nodeType{ntype}") database = db.Node_collection() opt_section = section or request.args.get("section") @@ -303,7 +303,7 @@ def find_node_by_name( def find_document_by_tag() -> Any: tags = request.args.getlist("tag") if posthog: - posthog.capture(f"find_document_by_tag", f"tags:{tags}") + posthog.capture("find_document_by_tag", f"tags:{tags}") database = db.Node_collection() # opt_osib = request.args.get("osib") @@ -342,7 +342,7 @@ def find_document_by_tag() -> Any: def map_analysis() -> Any: standards = request.args.getlist("standard") if posthog: - posthog.capture(f"map_analysis", f"standards:{standards}") + posthog.capture("map_analysis", f"standards:{standards}") database = db.Node_collection() if len(standards) < 2: @@ -448,7 +448,7 @@ def map_analysis() -> Any: def map_analysis_weak_links() -> Any: standards = request.args.getlist("standard") if posthog: - posthog.capture(f"map_analysis_weak_links", f"standards:{standards}") + posthog.capture("map_analysis_weak_links", f"standards:{standards}") key = request.args.get("key") cache_key = gap_analysis.make_subresources_key(standards=standards, key=key) @@ -536,7 +536,7 @@ def fetch_job() -> Any: @app.route("/rest/v1/standards", methods=["GET"]) def standards() -> Any: if posthog: - posthog.capture(f"standards", "") + posthog.capture("standards", "") database = db.Node_collection() standards = list(database.standards()) @@ -605,7 +605,7 @@ def text_search() -> Any: if not text: return jsonify({"error": "text parameter is required"}), 400 if posthog: - posthog.capture(f"text_search", f"text:{text}") + posthog.capture("text_search", f"text:{text}") opt_format = request.args.get("format") documents = database.text_search(text) @@ -665,7 +665,7 @@ def find_root_cres() -> Any: """ if posthog: - posthog.capture(f"find_root_cres", "") + posthog.capture("find_root_cres", "") database = db.Node_collection() # opt_osib = request.args.get("osib") @@ -746,7 +746,7 @@ def smartlink( # ATTENTION: DO NOT MESS WITH THIS FUNCTIONALITY WITHOUT A TICKET AND CORE CONTRIBUTORS APPROVAL! # CRITICAL FUNCTIONALITY DEPENDS ON THIS! if posthog: - posthog.capture(f"smartlink", f"name:{name}") + posthog.capture("smartlink", f"name:{name}") database = db.Node_collection() opt_version = request.args.get("version") @@ -802,7 +802,7 @@ def smartlink( ) return redirect(redirectors.redirect(name, section)) else: - logger.warning(f"not sure what happened, 404") + logger.warning("not sure what happened, 404") return abort(404, "Document does not exist") @@ -826,7 +826,7 @@ def deeplink( opt_version = request.args.get("version") opt_subsection = request.args.get("subsection") if posthog: - posthog.capture(f"deeplink", f"name:{name}") + posthog.capture("deeplink", f"name:{name}") if opt_section: opt_section = urllib.parse.unquote(opt_section) @@ -1153,7 +1153,7 @@ def admin_import_run_apply(run_id: str) -> Any: def chat_cre() -> Any: message = request.get_json(force=True) if posthog: - posthog.capture(f"chat_cre", "") + posthog.capture("chat_cre", "") database = db.Node_collection() # Lazy import to avoid loading heavy prompt/ML dependencies at web boot. @@ -1389,7 +1389,7 @@ def put_user_resources() -> Any: def all_cres() -> Any: database = db.Node_collection() if posthog: - posthog.capture(f"all_cres", "") + posthog.capture("all_cres", "") page = 1 per_page = ITEMS_PER_PAGE @@ -1417,7 +1417,7 @@ def all_cres() -> Any: @app.route("/rest/v1/cre_csv", methods=["GET"]) def get_cre_csv() -> Any: if posthog: - posthog.capture(f"get_cre_csv", "") + posthog.capture("get_cre_csv", "") database = db.Node_collection() root_cres = database.get_root_cres() From fb627a04915618413fcaed9b9d8fa009b941c571 Mon Sep 17 00:00:00 2001 From: Shaurya Upadhyay Date: Fri, 24 Jul 2026 02:56:33 +0530 Subject: [PATCH 2/6] fix: change remaining _json references to json --- application/cmd/cre_main.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/application/cmd/cre_main.py b/application/cmd/cre_main.py index 5f137d96d..e9872e752 100644 --- a/application/cmd/cre_main.py +++ b/application/cmd/cre_main.py @@ -461,7 +461,7 @@ def _standard_structure_fingerprint(resource_name: str) -> str: tuple(sorted(links)), ) ) - payload = _json.dumps( + payload = json.dumps( sorted(rows), sort_keys=True, separators=(",", ":"), ensure_ascii=False ) return hashlib.sha256(payload.encode("utf-8")).hexdigest() @@ -697,7 +697,7 @@ def download_gap_analysis_from_upstream(cache: str) -> None: tojson = res.json() if "result" not in tojson: continue - payload = _json.dumps({"result": tojson.get("result")}) + payload = json.dumps({"result": tojson.get("result")}) if not gap_analysis.primary_gap_analysis_payload_is_material( payload ): @@ -714,7 +714,7 @@ def download_gap_analysis_from_upstream(cache: str) -> None: tojson = res.json() if "result" not in tojson: continue - payload = _json.dumps({"result": tojson.get("result")}) + payload = json.dumps({"result": tojson.get("result")}) if not gap_analysis.primary_gap_analysis_payload_is_material( payload ): From 24ffdb2ceea744907ef3e896bc2d354394f06b70 Mon Sep 17 00:00:00 2001 From: Shaurya Upadhyay Date: Sat, 8 Aug 2026 16:32:11 +0530 Subject: [PATCH 3/6] fix: Secure redirect URL to resolve CodeQL alert --- application/web/web_main.py | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/application/web/web_main.py b/application/web/web_main.py index 49c17576b..9aeb8ceb6 100644 --- a/application/web/web_main.py +++ b/application/web/web_main.py @@ -794,16 +794,16 @@ def smartlink( if found_section_id: return redirect(f"/node/{ntype}/{name}/sectionid/{section}") return redirect(f"/node/{ntype}/{name}/section/{section}") - elif doctype == defs.Credoctypes.Standard.value and redirectors.redirect( - name, section - ): - logger.info( - f"did not find node of type {ntype}, name {name} and section {section}, redirecting to external resource" - ) - return redirect(redirectors.redirect(name, section)) - else: - logger.warning("not sure what happened, 404") - return abort(404, "Document does not exist") + elif doctype == defs.Credoctypes.Standard.value: + url = redirectors.redirect(name, section) + if url and isinstance(url, str) and url.startswith("https://"): + logger.info( + f"did not find node of type {ntype}, name {name} and section {section}, redirecting to external resource" + ) + return redirect(url) + + logger.warning("not sure what happened, 404") + return abort(404, "Document does not exist") @openapi_documented("deeplink") From 6771031bf876766fd2c3c4bc438fda2ca1d54fe6 Mon Sep 17 00:00:00 2001 From: Shaurya Upadhyay Date: Sat, 8 Aug 2026 16:43:12 +0530 Subject: [PATCH 4/6] style: run black formatter --- application/web/web_main.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/application/web/web_main.py b/application/web/web_main.py index 9aeb8ceb6..8738320f7 100644 --- a/application/web/web_main.py +++ b/application/web/web_main.py @@ -49,7 +49,6 @@ import oauthlib import google.auth.transport.requests - ITEMS_PER_PAGE = 20 MAX_ITEMS_PER_PAGE = 100 OPENCRE_STANDARD_NAME = gap_analysis.OPENCRE_STANDARD_NAME @@ -801,7 +800,7 @@ def smartlink( f"did not find node of type {ntype}, name {name} and section {section}, redirecting to external resource" ) return redirect(url) - + logger.warning("not sure what happened, 404") return abort(404, "Document does not exist") From fab0b9a41b5953e864b77e827147947d8f2be05e Mon Sep 17 00:00:00 2001 From: Shaurya Upadhyay Date: Sat, 8 Aug 2026 16:46:22 +0530 Subject: [PATCH 5/6] test: Add regression tests for rejected redirect values --- application/tests/web_main_test.py | 28 +++++++++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/application/tests/web_main_test.py b/application/tests/web_main_test.py index 589d37007..1bff0f94e 100644 --- a/application/tests/web_main_test.py +++ b/application/tests/web_main_test.py @@ -113,7 +113,7 @@ def test_extend_cre_with_tag_links(self) -> None: } for c, v in cres.items(): - res = web_main.extend_cre_with_tag_links( # type:ignore # mypy bug + res = web_main.extend_cre_with_tag_links( # type: ignore # mypy bug v, collection=collection ) self.assertCountEqual(res.links, v.links) @@ -715,6 +715,32 @@ def test_smartlink_critical_edge_cases(self) -> None: self.assertEqual(status, 302) self.assertEqual(location, "/node/standard/VERSTD/sectionid/200") + @patch("application.utils.redirectors.redirect") + def test_smartlink_rejected_redirects(self, mock_redirect) -> None: + """Test that invalid URLs returned by redirectors.redirect result in a 404.""" + with self.app.test_client() as client: + # 1. http:// URL + mock_redirect.return_value = ( + "http://cwe.mitre.org/data/definitions/999.html" + ) + response = client.get("/smartlink/standard/CWE/999") + self.assertEqual(404, response.status_code) + mock_redirect.assert_called_once_with("CWE", "999") + + mock_redirect.reset_mock() + # 2. javascript: URL + mock_redirect.return_value = "javascript:alert(1)" + response = client.get("/smartlink/standard/CWE/999") + self.assertEqual(404, response.status_code) + mock_redirect.assert_called_once_with("CWE", "999") + + mock_redirect.reset_mock() + # 3. non-string result + mock_redirect.return_value = {"url": "https://cwe.mitre.org"} + response = client.get("/smartlink/standard/CWE/999") + self.assertEqual(404, response.status_code) + mock_redirect.assert_called_once_with("CWE", "999") + @patch.object(redis, "from_url") @patch.object(db, "Node_collection") def test_gap_analysis_from_cache_full_response( From b593ba0117567787e268f260438cfb537f829535 Mon Sep 17 00:00:00 2001 From: Shaurya Upadhyay Date: Sat, 8 Aug 2026 17:27:56 +0530 Subject: [PATCH 6/6] fix: Secure redirect URL to resolve CodeQL alert properly --- application/utils/redirectors.py | 9 +++++++-- application/web/web_main.py | 9 ++++++++- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/application/utils/redirectors.py b/application/utils/redirectors.py index 99eabcd72..2568b94d4 100644 --- a/application/utils/redirectors.py +++ b/application/utils/redirectors.py @@ -1,9 +1,14 @@ +import urllib.parse + + def cwe_redirector(cwe_id: int): - return f"https://cwe.mitre.org/data/definitions/{cwe_id}.html" + return ( + f"https://cwe.mitre.org/data/definitions/{urllib.parse.quote(str(cwe_id))}.html" + ) def capec_redirector(capec_id: int) -> str: - return f"https://capec.mitre.org/data/definitions/{capec_id}.html" + return f"https://capec.mitre.org/data/definitions/{urllib.parse.quote(str(capec_id))}.html" def redirect(node_type, node_id): diff --git a/application/web/web_main.py b/application/web/web_main.py index 8738320f7..5f049f693 100644 --- a/application/web/web_main.py +++ b/application/web/web_main.py @@ -795,7 +795,14 @@ def smartlink( return redirect(f"/node/{ntype}/{name}/section/{section}") elif doctype == defs.Credoctypes.Standard.value: url = redirectors.redirect(name, section) - if url and isinstance(url, str) and url.startswith("https://"): + if ( + url + and isinstance(url, str) + and ( + url.startswith("https://cwe.mitre.org/") + or url.startswith("https://capec.mitre.org/") + ) + ): logger.info( f"did not find node of type {ntype}, name {name} and section {section}, redirecting to external resource" )