From 0da3813007f4e3eec58c40ef7b6782305db81362 Mon Sep 17 00:00:00 2001 From: bornunique911 Date: Tue, 24 Mar 2026 23:47:44 +0530 Subject: [PATCH] Add specialized OWASP cheat sheet map analysis behavior --- application/tests/web_main_test.py | 256 +++++++++++++++++++++++++++++ application/web/web_main.py | 170 ++++++++++++++++++- 2 files changed, 425 insertions(+), 1 deletion(-) diff --git a/application/tests/web_main_test.py b/application/tests/web_main_test.py index 2efd71758..763e32ae9 100644 --- a/application/tests/web_main_test.py +++ b/application/tests/web_main_test.py @@ -34,6 +34,32 @@ def get_status(self): return rq.job.JobStatus.STARTED +class MockFinishedJobResult: + class Type: + SUCCESSFUL = "successful" + FAILED = "failed" + + type = Type.SUCCESSFUL + + @property + def return_value(self): + return [ + [ + "OWASP Top 10 for LLM and Gen AI Apps 2025", + "OWASP Cheat Sheets", + ], + {"status": "done"}, + ] + + +class MockFinishedJob: + def get_status(self): + return rq.job.JobStatus.FINISHED + + def latest_result(self): + return MockFinishedJobResult() + + class TestMain(unittest.TestCase): def tearDown(self) -> None: sqla.session.remove() @@ -878,6 +904,180 @@ def test_gap_analysis_returns_direct_cre_overlap_when_backends_fail( self.assertIn(compare.id, payload["result"][base.id]["paths"]) db_mock.return_value.add_gap_analysis_result.assert_called_once() + @patch.object(web_main.cre_main, "fetch_upstream_json") + @patch.object(web_main.gap_analysis, "schedule") + @patch.object(db, "Node_collection") + def test_gap_analysis_filters_generic_cheatsheets_for_llm_top10( + self, db_mock, schedule_mock, upstream_fetch_mock + ) -> None: + broad_cre = defs.CRE(id="064-808", name="Output encoding", description="") + llm_specific_cre = defs.CRE( + id="161-451", name="Prompt boundary protection", description="" + ) + + llm = defs.Standard( + name="OWASP Top 10 for LLM and Gen AI Apps 2025", + sectionID="LLM05", + section="Improper Output Handling", + ) + llm.add_link( + defs.Link(ltype=defs.LinkTypes.LinkedTo, document=broad_cre.shallow_copy()) + ) + llm.add_link( + defs.Link( + ltype=defs.LinkTypes.LinkedTo, document=llm_specific_cre.shallow_copy() + ) + ) + + generic_cheatsheet = defs.Standard( + name="OWASP Cheat Sheets", + section="Cross Site Scripting Prevention Cheat Sheet", + ) + generic_cheatsheet.add_link( + defs.Link(ltype=defs.LinkTypes.LinkedTo, document=broad_cre.shallow_copy()) + ) + + llm_cheatsheet = defs.Standard( + name="OWASP Cheat Sheets", + section="LLM Prompt Injection Prevention Cheat Sheet", + ) + llm_cheatsheet.add_link( + defs.Link( + ltype=defs.LinkTypes.LinkedTo, document=llm_specific_cre.shallow_copy() + ) + ) + + db_mock.return_value.get_gap_analysis_result.return_value = None + db_mock.return_value.gap_analysis_exists.return_value = False + db_mock.return_value.get_nodes.side_effect = lambda name=None, **kwargs: ( + [llm] + if name == "OWASP Top 10 for LLM and Gen AI Apps 2025" + else ( + [generic_cheatsheet, llm_cheatsheet] + if name == "OWASP Cheat Sheets" + else [] + ) + ) + schedule_mock.side_effect = RuntimeError("redis unavailable") + upstream_fetch_mock.side_effect = RuntimeError("upstream unavailable") + + with self.app.test_client() as client: + response = client.get( + "/rest/v1/map_analysis?standard=OWASP%20Top%2010%20for%20LLM%20and%20Gen%20AI%20Apps%202025&standard=OWASP%20Cheat%20Sheets", + headers={"Content-Type": "application/json"}, + ) + + payload = json.loads(response.data) + self.assertEqual(200, response.status_code) + self.assertIn("result", payload) + self.assertEqual( + "AI / LLM Cheat Sheets", + payload["specialized_cheatsheet_section"]["category"], + ) + self.assertIn(llm.id, payload["result"]) + self.assertNotIn(llm_cheatsheet.id, payload["result"][llm.id]["paths"]) + self.assertNotIn(generic_cheatsheet.id, payload["result"][llm.id]["paths"]) + + @patch.object(web_main.cre_main, "fetch_upstream_json") + @patch.object(web_main.gap_analysis, "schedule") + @patch.object(db, "Node_collection") + def test_gap_analysis_keeps_prompt_injection_cheatsheet_for_llm01( + self, db_mock, schedule_mock, upstream_fetch_mock + ) -> None: + prompt_cre = defs.CRE( + id="161-451", name="Prompt boundary protection", description="" + ) + + llm = defs.Standard( + name="OWASP Top 10 for LLM and Gen AI Apps 2025", + sectionID="LLM01", + section="Prompt Injection", + ) + llm.add_link( + defs.Link(ltype=defs.LinkTypes.LinkedTo, document=prompt_cre.shallow_copy()) + ) + + llm_cheatsheet = defs.Standard( + name="OWASP Cheat Sheets", + section="LLM Prompt Injection Prevention Cheat Sheet", + ) + llm_cheatsheet.add_link( + defs.Link(ltype=defs.LinkTypes.LinkedTo, document=prompt_cre.shallow_copy()) + ) + + db_mock.return_value.get_gap_analysis_result.return_value = None + db_mock.return_value.gap_analysis_exists.return_value = False + db_mock.return_value.get_nodes.side_effect = lambda name=None, **kwargs: ( + [llm] + if name == "OWASP Top 10 for LLM and Gen AI Apps 2025" + else [llm_cheatsheet] + if name == "OWASP Cheat Sheets" + else [] + ) + schedule_mock.side_effect = RuntimeError("redis unavailable") + upstream_fetch_mock.side_effect = RuntimeError("upstream unavailable") + + with self.app.test_client() as client: + response = client.get( + "/rest/v1/map_analysis?standard=OWASP%20Top%2010%20for%20LLM%20and%20Gen%20AI%20Apps%202025&standard=OWASP%20Cheat%20Sheets", + headers={"Content-Type": "application/json"}, + ) + + payload = json.loads(response.data) + self.assertEqual(200, response.status_code) + self.assertIn(llm.id, payload["result"]) + self.assertIn(llm_cheatsheet.id, payload["result"][llm.id]["paths"]) + + @patch.object(web_main.cre_main, "fetch_upstream_json") + @patch.object(web_main.gap_analysis, "schedule") + @patch.object(db, "Node_collection") + def test_gap_analysis_adds_specialized_section_for_api_cheatsheets( + self, db_mock, schedule_mock, upstream_fetch_mock + ) -> None: + authz_cre = defs.CRE( + id="117-371", name="Centralized access control", description="" + ) + base = defs.Standard( + name="OWASP API Security Top 10 2023", + sectionID="API1", + section="Broken Object Level Authorization", + ) + base.add_link( + defs.Link(ltype=defs.LinkTypes.LinkedTo, document=authz_cre.shallow_copy()) + ) + + compare = defs.Standard( + name="OWASP Cheat Sheets", + section="Authorization Cheat Sheet", + ) + compare.add_link( + defs.Link(ltype=defs.LinkTypes.LinkedTo, document=authz_cre.shallow_copy()) + ) + + db_mock.return_value.get_gap_analysis_result.return_value = None + db_mock.return_value.gap_analysis_exists.return_value = False + db_mock.return_value.get_nodes.side_effect = lambda name=None, **kwargs: ( + [base] + if name == "OWASP API Security Top 10 2023" + else [compare] if name == "OWASP Cheat Sheets" else [] + ) + schedule_mock.side_effect = RuntimeError("redis unavailable") + upstream_fetch_mock.side_effect = RuntimeError("upstream unavailable") + + with self.app.test_client() as client: + response = client.get( + "/rest/v1/map_analysis?standard=OWASP%20API%20Security%20Top%2010%202023&standard=OWASP%20Cheat%20Sheets", + headers={"Content-Type": "application/json"}, + ) + + payload = json.loads(response.data) + self.assertEqual(200, response.status_code) + self.assertEqual( + "API Cheat Sheets", + payload["specialized_cheatsheet_section"]["category"], + ) + self.assertIn(base.id, payload["specialized_cheatsheet_section"]["result"]) + @patch.object(web_main.gap_analysis, "schedule") @patch.object(db, "Node_collection") def test_gap_analysis_supports_opencre_as_standard( @@ -1222,6 +1422,62 @@ def test_gap_analysis_heroku_cache_miss_returns_404( self.assertEqual(404, response.status_code) redis_conn_mock.assert_not_called() + @patch.object(web_main.redis, "connect") + @patch.object(rq.job.Job, "fetch") + @patch.object(db, "Node_collection") + def test_ma_job_results_adds_specialized_cheatsheet_section( + self, db_mock, fetch_mock, redis_connect_mock + ) -> None: + shared_cre = defs.CRE(id="161-451", name="Prompt boundary protection", description="") + llm = defs.Standard( + name="OWASP Top 10 for LLM and Gen AI Apps 2025", + sectionID="LLM01", + section="Prompt Injection", + ) + llm.add_link( + defs.Link(ltype=defs.LinkTypes.LinkedTo, document=shared_cre.shallow_copy()) + ) + cheatsheet = defs.Standard( + name="OWASP Cheat Sheets", + section="LLM Prompt Injection Prevention Cheat Sheet", + ) + cheatsheet.add_link( + defs.Link(ltype=defs.LinkTypes.LinkedTo, document=shared_cre.shallow_copy()) + ) + + cached_gap = { + "result": { + llm.id: { + "start": llm.shallow_copy().todict(), + "paths": { + cheatsheet.id: { + "end": cheatsheet.shallow_copy().todict(), + "path": [], + "score": 0, + } + }, + "extra": 0, + } + } + } + + fetch_mock.return_value = MockFinishedJob() + redis_connect_mock.return_value.exists.return_value = True + db_mock.return_value.get_gap_analysis_result.return_value = json.dumps(cached_gap) + + with self.app.test_client() as client: + response = client.get( + "/rest/v1/ma_job_results?id=ABC", + headers={"Content-Type": "application/json"}, + ) + + payload = json.loads(response.data) + self.assertEqual(200, response.status_code) + self.assertEqual( + "AI / LLM Cheat Sheets", + payload["specialized_cheatsheet_section"]["category"], + ) + @patch.object(redis, "from_url") @patch.object(db, "Node_collection") def test_standards_from_db(self, node_mock, redis_conn_mock) -> None: diff --git a/application/web/web_main.py b/application/web/web_main.py index 8b6dd823c..b610d8b28 100644 --- a/application/web/web_main.py +++ b/application/web/web_main.py @@ -93,6 +93,49 @@ KUBERNETES_TOP10_2025_STANDARD_NAME = "OWASP Kubernetes Top Ten 2025 (Draft)" KUBERNETES_TOP10_2022_STANDARD_NAME = "OWASP Kubernetes Top Ten 2022" OWASP_CHEATSHEETS_STANDARD_NAME = "OWASP Cheat Sheets" +SPECIALIZED_CHEATSHEET_GROUPS = { + "AI / LLM Cheat Sheets": { + "standards": {LLM_TOP10_STANDARD_NAME, AISVS_STANDARD_NAME}, + "sections": { + "LLM Prompt Injection Prevention Cheat Sheet", + "AI Agent Security Cheat Sheet", + "Secure AI Model Ops Cheat Sheet", + }, + }, + "API Cheat Sheets": { + "standards": {API_TOP10_STANDARD_NAME}, + "sections": { + "REST Security Cheat Sheet", + "Authorization Cheat Sheet", + "Server Side Request Forgery Prevention Cheat Sheet", + "Web Service Security Cheat Sheet", + }, + }, + "Cloud Cheat Sheets": { + "standards": { + CCM_STANDARD_NAME, + KUBERNETES_TOP10_2025_STANDARD_NAME, + KUBERNETES_TOP10_2022_STANDARD_NAME, + }, + "sections": { + "Docker Security Cheat Sheet", + "Kubernetes Security Cheat Sheet", + "Secure Cloud Architecture Cheat Sheet", + }, + }, +} +AI_LLM_SECTION_CHEATSHEET_MATCHES = { + "LLM01": {"LLM Prompt Injection Prevention Cheat Sheet"}, + "LLM02": {"AI Agent Security Cheat Sheet"}, + "LLM03": {"Secure AI Model Ops Cheat Sheet"}, + "LLM04": {"Secure AI Model Ops Cheat Sheet"}, + "LLM05": set(), + "LLM06": {"AI Agent Security Cheat Sheet"}, + "LLM07": {"AI Agent Security Cheat Sheet"}, + "LLM08": {"AI Agent Security Cheat Sheet"}, + "LLM09": set(), + "LLM10": set(), +} def _ga_timeout_seconds() -> int: @@ -411,6 +454,18 @@ def _build_direct_cre_overlap_map_analysis( if not base_nodes or not compare_nodes: return None + specialized_group = _get_specialized_cheatsheet_group(standards) + if specialized_group: + allowed_sections = SPECIALIZED_CHEATSHEET_GROUPS[specialized_group]["sections"] + if standards[1] == OWASP_CHEATSHEETS_STANDARD_NAME: + compare_nodes = [ + node for node in compare_nodes if node.section in allowed_sections + ] + elif standards[0] == OWASP_CHEATSHEETS_STANDARD_NAME: + base_nodes = [ + node for node in base_nodes if node.section in allowed_sections + ] + if not base_nodes or not compare_nodes: return None @@ -428,6 +483,10 @@ def _build_direct_cre_overlap_map_analysis( if link.document.doctype != defs.Credoctypes.CRE: continue for compare_node in compare_nodes_by_cre.get(link.document.id, []): + if not _is_allowed_specialized_cheatsheet_match( + specialized_group, base_node, compare_node + ): + continue shared_paths.setdefault( compare_node.id, { @@ -469,6 +528,52 @@ def _build_direct_cre_overlap_map_analysis( ) return result +def _get_specialized_cheatsheet_group(standards: list[str]) -> str | None: + if OWASP_CHEATSHEETS_STANDARD_NAME not in standards: + return None + + for category, config in SPECIALIZED_CHEATSHEET_GROUPS.items(): + if any(standard in config["standards"] for standard in standards): + return category + return None + + +def _build_specialized_cheatsheet_section( + standards: list[str], gap_analysis_result: dict[str, Any] | None +) -> dict[str, Any] | None: + category = _get_specialized_cheatsheet_group(standards) + if not category or not gap_analysis_result: + return None + + result = gap_analysis_result.get("result") + if not isinstance(result, dict): + return None + + return { + "category": category, + "standards": standards, + "result": result, + } + + +def _is_allowed_specialized_cheatsheet_match( + category: str | None, base_node: defs.Standard, compare_node: defs.Standard +) -> bool: + if category != "AI / LLM Cheat Sheets": + return True + + if base_node.name == LLM_TOP10_STANDARD_NAME and compare_node.name == OWASP_CHEATSHEETS_STANDARD_NAME: + allowed = AI_LLM_SECTION_CHEATSHEET_MATCHES.get(base_node.sectionID or "", set()) + return compare_node.section in allowed + + if ( + base_node.name == OWASP_CHEATSHEETS_STANDARD_NAME + and compare_node.name == LLM_TOP10_STANDARD_NAME + ): + allowed = AI_LLM_SECTION_CHEATSHEET_MATCHES.get(compare_node.sectionID or "", set()) + return base_node.section in allowed + + return True def extend_cre_with_tag_links( cre: defs.CRE, collection: db.Node_collection ) -> defs.CRE: @@ -698,8 +803,15 @@ def map_analysis() -> Any: gap_analysis_result = database.get_gap_analysis_result(standards_hash) if gap_analysis_result: result = flask_json.loads(gap_analysis_result) + specialized_cheatsheet_section = _build_specialized_cheatsheet_section( + standards, result + ) if owasp_top10_comparison: result["owasp_top10_comparison"] = owasp_top10_comparison + if specialized_cheatsheet_section: + result["specialized_cheatsheet_section"] = ( + specialized_cheatsheet_section + ) return jsonify(result) # On Heroku (read-only), check if standards exist before attempting Redis/queue operations @@ -733,15 +845,29 @@ def map_analysis() -> Any: standards, standards_hash, database ) if upstream_gap_analysis: + specialized_cheatsheet_section = _build_specialized_cheatsheet_section( + standards, upstream_gap_analysis + ) if owasp_top10_comparison: upstream_gap_analysis["owasp_top10_comparison"] = owasp_top10_comparison + if specialized_cheatsheet_section: + upstream_gap_analysis["specialized_cheatsheet_section"] = ( + specialized_cheatsheet_section + ) return jsonify(upstream_gap_analysis) direct_gap_analysis = _build_direct_cre_overlap_map_analysis( standards, standards_hash, database ) if direct_gap_analysis: + specialized_cheatsheet_section = _build_specialized_cheatsheet_section( + standards, direct_gap_analysis + ) if owasp_top10_comparison: direct_gap_analysis["owasp_top10_comparison"] = owasp_top10_comparison + if specialized_cheatsheet_section: + direct_gap_analysis["specialized_cheatsheet_section"] = ( + specialized_cheatsheet_section + ) return jsonify(direct_gap_analysis) if owasp_top10_comparison: return jsonify({"owasp_top10_comparison": owasp_top10_comparison}) @@ -756,21 +882,42 @@ def map_analysis() -> Any: standards, standards_hash, database ) if upstream_gap_analysis: + specialized_cheatsheet_section = _build_specialized_cheatsheet_section( + standards, upstream_gap_analysis + ) if owasp_top10_comparison: upstream_gap_analysis["owasp_top10_comparison"] = owasp_top10_comparison + if specialized_cheatsheet_section: + upstream_gap_analysis["specialized_cheatsheet_section"] = ( + specialized_cheatsheet_section + ) return jsonify(upstream_gap_analysis) direct_gap_analysis = _build_direct_cre_overlap_map_analysis( standards, standards_hash, database ) if direct_gap_analysis: + specialized_cheatsheet_section = _build_specialized_cheatsheet_section( + standards, direct_gap_analysis + ) if owasp_top10_comparison: direct_gap_analysis["owasp_top10_comparison"] = owasp_top10_comparison + if specialized_cheatsheet_section: + direct_gap_analysis["specialized_cheatsheet_section"] = ( + specialized_cheatsheet_section + ) return jsonify(direct_gap_analysis) if owasp_top10_comparison: return jsonify({"owasp_top10_comparison": owasp_top10_comparison}) raise if owasp_top10_comparison: gap_analysis_dict["owasp_top10_comparison"] = owasp_top10_comparison + specialized_cheatsheet_section = _build_specialized_cheatsheet_section( + standards, gap_analysis_dict + ) + if specialized_cheatsheet_section: + gap_analysis_dict["specialized_cheatsheet_section"] = ( + specialized_cheatsheet_section + ) if "result" in gap_analysis_dict: return jsonify(gap_analysis_dict) if gap_analysis_dict.get("error"): @@ -778,15 +925,29 @@ def map_analysis() -> Any: standards, standards_hash, database ) if upstream_gap_analysis: + specialized_cheatsheet_section = _build_specialized_cheatsheet_section( + standards, upstream_gap_analysis + ) if owasp_top10_comparison: upstream_gap_analysis["owasp_top10_comparison"] = owasp_top10_comparison + if specialized_cheatsheet_section: + upstream_gap_analysis["specialized_cheatsheet_section"] = ( + specialized_cheatsheet_section + ) return jsonify(upstream_gap_analysis) direct_gap_analysis = _build_direct_cre_overlap_map_analysis( standards, standards_hash, database ) if direct_gap_analysis: + specialized_cheatsheet_section = _build_specialized_cheatsheet_section( + standards, direct_gap_analysis + ) if owasp_top10_comparison: direct_gap_analysis["owasp_top10_comparison"] = owasp_top10_comparison + if specialized_cheatsheet_section: + direct_gap_analysis["specialized_cheatsheet_section"] = ( + specialized_cheatsheet_section + ) return jsonify(direct_gap_analysis) if owasp_top10_comparison: return jsonify({"owasp_top10_comparison": owasp_top10_comparison}) @@ -865,7 +1026,14 @@ def fetch_job() -> Any: if ga: # logger.__delattr__("and results in cache") ga = flask_json.loads(ga) - if ga.get("result"): + specialized_cheatsheet_section = ( + _build_specialized_cheatsheet_section(standards, ga) + ) + if specialized_cheatsheet_section: + ga["specialized_cheatsheet_section"] = ( + specialized_cheatsheet_section + ) + if "result" in ga: return jsonify(ga) else: logger.error(