Skip to content

Commit 8d57002

Browse files
committed
Add OWASP comparison and fallback map analysis support
1 parent b3fedd8 commit 8d57002

2 files changed

Lines changed: 621 additions & 4 deletions

File tree

application/tests/web_main_test.py

Lines changed: 304 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -518,6 +518,63 @@ def test_find_root_cres(self) -> None:
518518
self.assertEqual(json.loads(response.data.decode()), expected)
519519
self.assertEqual(200, response.status_code)
520520

521+
def test_find_root_cres_featured_standards(self) -> None:
522+
collection = db.Node_collection().with_graph()
523+
524+
root_cre = defs.CRE(id="111-115", description="CA", name="CA", tags=["ta"])
525+
collection.add_cre(root_cre)
526+
527+
featured_nodes = [
528+
defs.Standard(
529+
name="OWASP Top 10 for LLM and Gen AI Apps 2025",
530+
sectionID="LLM01",
531+
section="Prompt Injection",
532+
),
533+
defs.Standard(
534+
name="OWASP AI Security Verification Standard (AISVS)",
535+
sectionID="AISVS1",
536+
section="Training Data Governance & Bias Management",
537+
),
538+
defs.Standard(
539+
name="OWASP API Security Top 10 2023",
540+
sectionID="API1",
541+
section="Broken Object Level Authorization",
542+
),
543+
defs.Standard(
544+
name="Cloud Controls Matrix",
545+
sectionID="AIS-01",
546+
section="Application and Interface Security",
547+
),
548+
defs.Standard(
549+
name="OWASP Kubernetes Top Ten 2025 (Draft)",
550+
sectionID="K01",
551+
section="Insecure Workload Configurations",
552+
),
553+
]
554+
for node in featured_nodes:
555+
collection.add_node(node)
556+
557+
with self.app.test_client() as client:
558+
response = client.get(
559+
"/rest/v1/root_cres",
560+
headers={"Content-Type": "application/json"},
561+
)
562+
563+
self.assertEqual(200, response.status_code)
564+
payload = response.get_json()
565+
self.assertIn("data", payload)
566+
self.assertIn("featured_standards", payload)
567+
self.assertCountEqual(
568+
["AI", "API", "Cloud"], payload["featured_standards"].keys()
569+
)
570+
self.assertEqual(2, len(payload["featured_standards"]["AI"]))
571+
self.assertEqual(1, len(payload["featured_standards"]["API"]))
572+
self.assertEqual(2, len(payload["featured_standards"]["Cloud"]))
573+
self.assertEqual(
574+
"OWASP Top 10 for LLM and Gen AI Apps 2025",
575+
payload["featured_standards"]["AI"][0]["name"],
576+
)
577+
521578
def test_smartlink(self) -> None:
522579
self.maxDiff = None
523580
collection = db.Node_collection().with_graph()
@@ -623,6 +680,253 @@ def test_gap_analysis_from_cache_full_response(
623680
self.assertEqual(200, response.status_code)
624681
self.assertEqual(expected, json.loads(response.data))
625682

683+
@patch.object(redis, "from_url")
684+
@patch.object(db, "Node_collection")
685+
def test_gap_analysis_adds_owasp_top10_comparison_section(
686+
self, db_mock, redis_conn_mock
687+
) -> None:
688+
expected = {"result": "hello"}
689+
top10_2021 = [
690+
defs.Standard(
691+
name="OWASP Top 10 2021",
692+
sectionID="A01",
693+
section="Broken Access Controls",
694+
hyperlink="https://owasp.org/Top10/A01_2021-Broken_Access_Control/",
695+
),
696+
defs.Standard(
697+
name="OWASP Top 10 2021",
698+
sectionID="A02",
699+
section="Cryptographic Failures",
700+
hyperlink="https://owasp.org/Top10/A02_2021-Cryptographic_Failures/",
701+
),
702+
]
703+
top10_2025 = [
704+
defs.Standard(
705+
name="OWASP Top 10 2025",
706+
sectionID="A01",
707+
section="Broken Access Control",
708+
hyperlink="https://owasp.org/Top10/2025/A01_2025-Broken_Access_Control/",
709+
),
710+
defs.Standard(
711+
name="OWASP Top 10 2025",
712+
sectionID="A02",
713+
section="Security Misconfiguration",
714+
hyperlink="https://owasp.org/Top10/2025/A02_2025-Security_Misconfiguration/",
715+
),
716+
]
717+
redis_conn_mock.return_value.exists.return_value = True
718+
redis_conn_mock.return_value.get.return_value = json.dumps(expected)
719+
db_mock.return_value.get_gap_analysis_result.return_value = json.dumps(expected)
720+
db_mock.return_value.gap_analysis_exists.return_value = True
721+
722+
def get_nodes_side_effect(name=None, **kwargs):
723+
if name == "OWASP Top 10 2021":
724+
return top10_2021
725+
if name == "OWASP Top 10 2025":
726+
return top10_2025
727+
return []
728+
729+
db_mock.return_value.get_nodes.side_effect = get_nodes_side_effect
730+
731+
with self.app.test_client() as client:
732+
response = client.get(
733+
"/rest/v1/map_analysis?standard=OWASP%20Top%2010%202021&standard=OWASP%20Top%2010%202025",
734+
headers={"Content-Type": "application/json"},
735+
)
736+
737+
payload = json.loads(response.data)
738+
self.assertEqual(200, response.status_code)
739+
self.assertEqual("hello", payload["result"])
740+
self.assertIn("owasp_top10_comparison", payload)
741+
self.assertEqual(
742+
["OWASP Top 10 2021", "OWASP Top 10 2025"],
743+
payload["owasp_top10_comparison"]["standards"],
744+
)
745+
self.assertEqual(
746+
"Broken Access Controls",
747+
payload["owasp_top10_comparison"]["items"][0]["top10_2021"]["section"],
748+
)
749+
self.assertEqual(
750+
"Broken Access Control",
751+
payload["owasp_top10_comparison"]["items"][0]["top10_2025"]["section"],
752+
)
753+
754+
@patch.object(web_main.gap_analysis, "schedule")
755+
@patch.object(db, "Node_collection")
756+
def test_gap_analysis_returns_owasp_comparison_when_schedule_fails(
757+
self, db_mock, schedule_mock
758+
) -> None:
759+
top10_2021 = [
760+
defs.Standard(
761+
name="OWASP Top 10 2021",
762+
sectionID="A01",
763+
section="Broken Access Controls",
764+
hyperlink="https://owasp.org/Top10/A01_2021-Broken_Access_Control/",
765+
)
766+
]
767+
top10_2025 = [
768+
defs.Standard(
769+
name="OWASP Top 10 2025",
770+
sectionID="A01",
771+
section="Broken Access Control",
772+
hyperlink="https://owasp.org/Top10/2025/A01_2025-Broken_Access_Control/",
773+
)
774+
]
775+
db_mock.return_value.get_gap_analysis_result.return_value = None
776+
db_mock.return_value.gap_analysis_exists.return_value = False
777+
778+
def get_nodes_side_effect(name=None, **kwargs):
779+
if name == "OWASP Top 10 2021":
780+
return top10_2021
781+
if name == "OWASP Top 10 2025":
782+
return top10_2025
783+
return []
784+
785+
db_mock.return_value.get_nodes.side_effect = get_nodes_side_effect
786+
schedule_mock.side_effect = RuntimeError("redis unavailable")
787+
788+
with self.app.test_client() as client:
789+
response = client.get(
790+
"/rest/v1/map_analysis?standard=OWASP%20Top%2010%202021&standard=OWASP%20Top%2010%202025",
791+
headers={"Content-Type": "application/json"},
792+
)
793+
794+
payload = json.loads(response.data)
795+
self.assertEqual(200, response.status_code)
796+
self.assertIn("owasp_top10_comparison", payload)
797+
self.assertEqual(
798+
"Broken Access Controls",
799+
payload["owasp_top10_comparison"]["items"][0]["top10_2021"]["section"],
800+
)
801+
802+
@patch.object(web_main.cre_main, "fetch_upstream_json")
803+
@patch.object(web_main.gap_analysis, "schedule")
804+
@patch.object(db, "Node_collection")
805+
def test_gap_analysis_returns_upstream_result_when_schedule_fails(
806+
self, db_mock, schedule_mock, upstream_fetch_mock
807+
) -> None:
808+
expected = {"result": {"A01": {"start": "x"}}}
809+
db_mock.return_value.get_gap_analysis_result.return_value = None
810+
db_mock.return_value.gap_analysis_exists.return_value = False
811+
schedule_mock.side_effect = RuntimeError("redis unavailable")
812+
upstream_fetch_mock.return_value = expected
813+
814+
with self.app.test_client() as client:
815+
response = client.get(
816+
"/rest/v1/map_analysis?standard=OWASP%20Top%2010%202021&standard=OWASP%20Cheat%20Sheets",
817+
headers={"Content-Type": "application/json"},
818+
)
819+
820+
self.assertEqual(200, response.status_code)
821+
self.assertEqual(expected, json.loads(response.data))
822+
upstream_fetch_mock.assert_called_once_with(
823+
"/map_analysis?standard=OWASP%20Top%2010%202021&standard=OWASP%20Cheat%20Sheets",
824+
timeout=5.0,
825+
max_attempts=1,
826+
backoff_seconds=0.5,
827+
)
828+
db_mock.return_value.add_gap_analysis_result.assert_called_once_with(
829+
cache_key="OWASP Top 10 2021 >> OWASP Cheat Sheets",
830+
ga_object=json.dumps(expected),
831+
)
832+
833+
@patch.object(web_main.cre_main, "fetch_upstream_json")
834+
@patch.object(web_main.gap_analysis, "schedule")
835+
@patch.object(db, "Node_collection")
836+
def test_gap_analysis_returns_direct_cre_overlap_when_backends_fail(
837+
self, db_mock, schedule_mock, upstream_fetch_mock
838+
) -> None:
839+
shared_cre = defs.CRE(id="170-772", name="Cryptography", description="")
840+
base = defs.Standard(
841+
name="OWASP Top 10 2025",
842+
sectionID="A04",
843+
section="Cryptographic Failures",
844+
)
845+
base.add_link(
846+
defs.Link(ltype=defs.LinkTypes.LinkedTo, document=shared_cre.shallow_copy())
847+
)
848+
compare = defs.Standard(
849+
name="OWASP Web Security Testing Guide (WSTG)",
850+
section="WSTG-CRYP-04",
851+
)
852+
compare.add_link(
853+
defs.Link(ltype=defs.LinkTypes.LinkedTo, document=shared_cre.shallow_copy())
854+
)
855+
856+
db_mock.return_value.get_gap_analysis_result.return_value = None
857+
db_mock.return_value.gap_analysis_exists.return_value = False
858+
db_mock.return_value.get_nodes.side_effect = lambda name=None, **kwargs: (
859+
[base]
860+
if name == "OWASP Top 10 2025"
861+
else [compare] if name == "OWASP Web Security Testing Guide (WSTG)" else []
862+
)
863+
schedule_mock.side_effect = RuntimeError("redis unavailable")
864+
upstream_fetch_mock.side_effect = RuntimeError("upstream unavailable")
865+
866+
with self.app.test_client() as client:
867+
response = client.get(
868+
"/rest/v1/map_analysis?standard=OWASP%20Top%2010%202025&standard=OWASP%20Web%20Security%20Testing%20Guide%20(WSTG)",
869+
headers={"Content-Type": "application/json"},
870+
)
871+
872+
payload = json.loads(response.data)
873+
self.assertEqual(200, response.status_code)
874+
self.assertIn("result", payload)
875+
self.assertIn(base.id, payload["result"])
876+
self.assertIn(compare.id, payload["result"][base.id]["paths"])
877+
db_mock.return_value.add_gap_analysis_result.assert_called_once()
878+
879+
@patch.object(web_main.gap_analysis, "schedule")
880+
@patch.object(db, "Node_collection")
881+
def test_gap_analysis_normalizes_owasp_top_2025_alias(
882+
self, db_mock, schedule_mock
883+
) -> None:
884+
top10_2021 = [
885+
defs.Standard(
886+
name="OWASP Top 10 2021",
887+
sectionID="A01",
888+
section="Broken Access Controls",
889+
hyperlink="https://owasp.org/Top10/A01_2021-Broken_Access_Control/",
890+
)
891+
]
892+
top10_2025 = [
893+
defs.Standard(
894+
name="OWASP Top 10 2025",
895+
sectionID="A01",
896+
section="Broken Access Control",
897+
hyperlink="https://owasp.org/Top10/2025/A01_2025-Broken_Access_Control/",
898+
)
899+
]
900+
db_mock.return_value.get_gap_analysis_result.return_value = None
901+
db_mock.return_value.gap_analysis_exists.return_value = False
902+
903+
def get_nodes_side_effect(name=None, **kwargs):
904+
if name == "OWASP Top 10 2021":
905+
return top10_2021
906+
if name == "OWASP Top 10 2025":
907+
return top10_2025
908+
return []
909+
910+
db_mock.return_value.get_nodes.side_effect = get_nodes_side_effect
911+
schedule_mock.side_effect = RuntimeError("redis unavailable")
912+
913+
with self.app.test_client() as client:
914+
response = client.get(
915+
"/rest/v1/map_analysis?standard=OWASP%20Top%2010%202021&standard=OWASP%20Top%202025",
916+
headers={"Content-Type": "application/json"},
917+
)
918+
919+
payload = json.loads(response.data)
920+
self.assertEqual(200, response.status_code)
921+
self.assertIn("owasp_top10_comparison", payload)
922+
self.assertEqual(
923+
"Broken Access Control",
924+
payload["owasp_top10_comparison"]["items"][0]["top10_2025"]["section"],
925+
)
926+
schedule_mock.assert_called_once_with(
927+
["OWASP Top 10 2021", "OWASP Top 10 2025"], db_mock.return_value
928+
)
929+
626930
@patch.object(db, "Node_collection")
627931
@patch.object(rq.job.Job, "fetch")
628932
@patch.object(rq.Queue, "enqueue_call")

0 commit comments

Comments
 (0)