Add OWASP AI resource importers

application/cmd/cre_main.py

@@ -730,7 +730,11 @@ def backfill_gap_analysis_only(
     if os.environ.get("CRE_NO_NEO4J") != "1":
         populate_neo4j_db(db_connection_str)
 
-    gap_analysis.backfill_opencre_direct_pairs(collection, refresh=True)
+    if os.environ.get("OPENCRE_ENV") not in {"heroku", "opencreorg"}:
+        gap_analysis.backfill_opencre_direct_pairs(collection, refresh=True)
+    else:
+        logger.warning("Skipping GA recomputation on production environment")
+        return
 
     missing = _missing_ga_pairs(collection)
     if max_pairs > 0:
@@ -911,6 +915,44 @@ def run(args: argparse.Namespace) -> None:  # pragma: no cover
         BaseParser().register_resource(
             secure_headers.SecureHeaders, db_connection_str=args.cache_file
         )
+    if getattr(args, "owasp_top10_2025_in", False):
+        from application.utils.external_project_parsers.parsers import owasp_top10_2025
+
+        BaseParser().register_resource(
+            owasp_top10_2025.OwaspTop10_2025, db_connection_str=args.cache_file
+        )
+    if getattr(args, "owasp_api_top10_2023_in", False):
+        from application.utils.external_project_parsers.parsers import (
+            owasp_api_top10_2023,
+        )
+
+        BaseParser().register_resource(
+            owasp_api_top10_2023.OwaspApiTop10_2023,
+            db_connection_str=args.cache_file,
+        )
+    if getattr(args, "owasp_kubernetes_top10_2022_in", False):
+        logger.warning(
+            "--owasp_kubernetes_top10_2022_in requested, but no Kubernetes 2022 parser module is present in this branch; skipping"
+        )
+    if getattr(args, "owasp_kubernetes_top10_2025_in", False):
+        logger.warning(
+            "--owasp_kubernetes_top10_2025_in requested, but no Kubernetes 2025 parser module is present in this branch; skipping"
+        )
+    if getattr(args, "owasp_llm_top10_2025_in", False):
+        from application.utils.external_project_parsers.parsers import (
+            owasp_llm_top10_2025,
+        )
+
+        BaseParser().register_resource(
+            owasp_llm_top10_2025.OwaspLlmTop10_2025,
+            db_connection_str=args.cache_file,
+        )
+    if getattr(args, "owasp_aisvs_in", False):
+        from application.utils.external_project_parsers.parsers import owasp_aisvs
+
+        BaseParser().register_resource(
+            owasp_aisvs.OwaspAisvs, db_connection_str=args.cache_file
+        )
     if args.pci_dss_4_in:
         from application.utils.external_project_parsers.parsers import pci_dss
 

application/tests/cheatsheet_extractor_test.py

@@ -0,0 +1,169 @@
+import unittest
+from application.utils.external_project_parsers.parsers.cheatsheet_extractor import (
+    extract_cheatsheet_record,
+)
+from application.defs.cheatsheet_defs import SUMMARY_MAX_LENGTH
+
+SOURCE_PATH = "cheatsheets/Secrets_Management_Cheat_Sheet.md"
+EXPECTED_SOURCE_ID = "Secrets_Management_Cheat_Sheet"
+EXPECTED_HYPERLINK = (
+    "https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html"
+)
+
+NORMAL_MD = """\
+# Secrets Management Cheat Sheet
+
+## Introduction
+Storage guidance.
+
+## Architectural Patterns
+Use vaults and environment isolation.
+"""
+
+MISSING_H1_MD = """\
+## Introduction
+No H1 present.
+
+## Details
+More content.
+"""
+
+EMPTY_MD = ""
+
+# No ## Introduction heading exists, so summary extraction
+# falls back to the first available body content.
+BODY_UNDER_H1_MD = """\
+# Single Heading Cheat Sheet
+
+Body text directly under H1, no subheadings at all.
+"""
+
+# Leading whitespace before H1 and malformed ## headings
+# should still be normalized and extracted correctly.
+MALFORMED_MD = """\
+   # Malformed Title
+
+##malformed
+
+## Introduction
+Some intro text.
+
+## Valid Heading
+"""
+
+
+class TestNormal(unittest.TestCase):
+    def setUp(self):
+        self.record = extract_cheatsheet_record(NORMAL_MD, SOURCE_PATH)
+
+    # source-derived fields should remain deterministic and
+    # independent of markdown content across all extraction paths.
+    def test_source(self):
+        self.assertEqual(self.record.source, "owasp_cheatsheets")
+
+    def test_source_id(self):
+        self.assertEqual(self.record.source_id, EXPECTED_SOURCE_ID)
+
+    def test_hyperlink(self):
+        self.assertEqual(self.record.hyperlink, EXPECTED_HYPERLINK)
+
+    def test_raw_markdown_path(self):
+        self.assertEqual(self.record.raw_markdown_path, SOURCE_PATH)
+
+    def test_title(self):
+        self.assertEqual(self.record.title, "Secrets Management Cheat Sheet")
+
+    def test_summary(self):
+        self.assertEqual(self.record.summary, "Storage guidance.")
+
+    def test_summary_bounded(self):
+        # Summary truncation is enforced centrally via
+        # CheatsheetRecord.__post_init__.
+        self.assertLessEqual(len(self.record.summary), SUMMARY_MAX_LENGTH)
+
+    def test_headings(self):
+        self.assertIn("Introduction", self.record.headings)
+        self.assertIn("Architectural Patterns", self.record.headings)
+
+    def test_fallback_not_used(self):
+        self.assertEqual(self.record.metadata["fallback_used"], "false")
+
+
+class TestMissingH1(unittest.TestCase):
+    def setUp(self):
+        self.record = extract_cheatsheet_record(MISSING_H1_MD, SOURCE_PATH)
+
+    def test_title_is_fallback(self):
+        self.assertEqual(self.record.title, "No title found.")
+
+    def test_summary_from_introduction(self):
+        self.assertIn("no h1", self.record.summary.lower())
+
+    def test_headings_extracted(self):
+        self.assertIn("Introduction", self.record.headings)
+        self.assertIn("Details", self.record.headings)
+
+    def test_fallback_used(self):
+        self.assertEqual(self.record.metadata["fallback_used"], "true")
+
+
+class TestEmptyMarkdown(unittest.TestCase):
+    def setUp(self):
+        self.record = extract_cheatsheet_record(EMPTY_MD, SOURCE_PATH)
+
+    def test_title_is_fallback(self):
+        self.assertEqual(self.record.title, "No title found.")
+
+    def test_summary_no_summary_found(self):
+        # Empty markdown should trigger terminal summary fallback.
+        self.assertEqual(self.record.summary, "No summary found.")
+
+    def test_headings_empty(self):
+        self.assertEqual(self.record.headings, [])
+
+    def test_fallback_used(self):
+        self.assertEqual(self.record.metadata["fallback_used"], "true")
+
+
+class TestBodyUnderH1(unittest.TestCase):
+    def setUp(self):
+        self.record = extract_cheatsheet_record(BODY_UNDER_H1_MD, SOURCE_PATH)
+
+    def test_title(self):
+        self.assertEqual(self.record.title, "Single Heading Cheat Sheet")
+
+    def test_summary_from_fallback_via_h1(self):
+        # Summary fallback should extract body content beneath the H1 section.
+        self.assertIn("body text", self.record.summary.lower())
+
+    def test_headings_empty(self):
+        # No valid ## headings should produce an empty headings list.
+        self.assertEqual(self.record.headings, [])
+
+    def test_fallback_used(self):
+        self.assertEqual(self.record.metadata["fallback_used"], "true")
+
+
+class TestMalformedHeadings(unittest.TestCase):
+    def setUp(self):
+        self.record = extract_cheatsheet_record(MALFORMED_MD, SOURCE_PATH)
+
+    def test_malformed_h1_extracted(self):
+        self.assertEqual(self.record.title, "Malformed Title")
+
+    def test_malformed_h2_in_headings(self):
+        self.assertIn("malformed", self.record.headings)
+
+    def test_valid_headings_also_extracted(self):
+        self.assertIn("Introduction", self.record.headings)
+        self.assertIn("Valid Heading", self.record.headings)
+
+    def test_summary_from_introduction(self):
+        self.assertIn("intro", self.record.summary.lower())
+
+    def test_fallback_not_used(self):
+        self.assertEqual(self.record.metadata["fallback_used"], "false")
+
+
+if __name__ == "__main__":
+    unittest.main()

application/tests/cheatsheets_parser_test.py

@@ -34,7 +34,13 @@ class Repo:
         repo.working_dir = loc
         cre = defs.CRE(name="blah", id="223-780")
         self.collection.add_cre(cre)
-        with open(os.path.join(os.path.join(loc, "cheatsheets"), "cs.md"), "w") as mdf:
+        with open(
+            os.path.join(
+                os.path.join(loc, "cheatsheets"),
+                "Secrets_Management_Cheat_Sheet.md",
+            ),
+            "w",
+        ) as mdf:
             mdf.write(cs)
         mock_clone.return_value = repo
         entries = cheatsheets_parser.Cheatsheets().parse(
@@ -45,22 +51,22 @@ class Repo:
         # verify the external tagging convention, not just enum wiring.
         expected = defs.Standard(
             name="OWASP Cheat Sheets",
-            hyperlink="https://github.com/foo/bar/tree/master/cs.md",
+            hyperlink="https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html",
             section="Secrets Management Cheat Sheet",
-            links=[defs.Link(document=cre, ltype=defs.LinkTypes.LinkedTo)],
+            links=[defs.Link(document=cre, ltype=defs.LinkTypes.AutomaticallyLinkedTo)],
             tags=[
                 "family:guidance",
                 "subtype:cheatsheet",
-                "source:owasp_cheatsheets",
                 "audience:developer",
                 "maturity:stable",
+                "source:owasp_cheatsheets",
             ],
         )
         self.maxDiff = None
         for name, nodes in entries.results.items():
             self.assertEqual(name, parser.name)
             self.assertEqual(len(nodes), 1)
-            self.assertCountEqual(expected.todict(), nodes[0].todict())
+            self.assertEqual(expected.todict(), nodes[0].todict())
 
     cheatsheets_md = """ # Secrets Management Cheat Sheet
 

application/tests/owasp_aisvs_parser_test.py

@@ -0,0 +1,66 @@
+import unittest
+
+from application import create_app, sqla  # type: ignore
+from application.database import db
+from application.defs import cre_defs as defs
+from application.prompt_client import prompt_client
+from application.utils.external_project_parsers.parsers import owasp_aisvs
+
+
+class TestOwaspAisvsParser(unittest.TestCase):
+    def tearDown(self) -> None:
+        sqla.session.remove()
+        sqla.drop_all()
+        self.app_context.pop()
+
+    def setUp(self) -> None:
+        self.app = create_app(mode="test")
+        self.app_context = self.app.app_context()
+        self.app_context.push()
+        sqla.create_all()
+        self.collection = db.Node_collection()
+
+    def test_parse(self) -> None:
+        for cre_id, name in [
+            ("227-045", "Identify sensitive data and subject it to a policy"),
+            (
+                "307-507",
+                "Allow only trusted sources both build time and runtime; therefore perform integrity checks on all resources and code",
+            ),
+            (
+                "162-655",
+                "Documentation of all components' business or security function",
+            ),
+        ]:
+            self.collection.add_cre(defs.CRE(id=cre_id, name=name, description=""))
+
+        result = owasp_aisvs.OwaspAisvs().parse(
+            self.collection, prompt_client.PromptHandler(database=self.collection)
+        )
+
+        entries = result.results["OWASP AI Security Verification Standard (AISVS)"]
+        self.assertEqual(14, len(entries))
+        self.assertEqual("AISVS1", entries[0].sectionID)
+        self.assertEqual(
+            "Training Data Governance & Bias Management", entries[0].section
+        )
+        self.assertEqual(
+            "https://github.com/OWASP/AISVS/blob/main/1.0/en/0x10-C01-Training-Data-Integrity-and-Traceability.md",
+            entries[0].hyperlink,
+        )
+        self.assertEqual(
+            ["227-045", "307-507"],
+            [link.document.id for link in entries[0].links],
+        )
+        self.assertEqual("AISVS14", entries[-1].sectionID)
+        self.assertEqual(
+            "Human Oversight, Accountability & Governance", entries[-1].section
+        )
+        self.assertEqual(
+            "https://github.com/OWASP/AISVS/blob/main/1.0/en/0x10-C14-Human-Oversight.md",
+            entries[-1].hyperlink,
+        )
+        self.assertEqual(
+            ["162-655"],
+            [link.document.id for link in entries[-1].links],
+        )

application/tests/owasp_api_top10_2023_parser_test.py

@@ -0,0 +1,47 @@
+import unittest
+
+from application import create_app, sqla  # type: ignore
+from application.database import db
+from application.defs import cre_defs as defs
+from application.prompt_client import prompt_client
+from application.utils.external_project_parsers.parsers import owasp_api_top10_2023
+
+
+class TestOwaspApiTop10_2023Parser(unittest.TestCase):
+    def tearDown(self) -> None:
+        sqla.session.remove()
+        sqla.drop_all()
+        self.app_context.pop()
+
+    def setUp(self) -> None:
+        self.app = create_app(mode="test")
+        self.app_context = self.app.app_context()
+        self.app_context.push()
+        sqla.create_all()
+        self.collection = db.Node_collection()
+
+    def test_parse(self) -> None:
+        for cre_id, name in [
+            ("304-667", "Protect API against unauthorized access/modification (IDOR)"),
+            ("724-770", "Technical application access control"),
+            ("715-223", "Ensure trusted origin of third party resources"),
+        ]:
+            self.collection.add_cre(defs.CRE(id=cre_id, name=name, description=""))
+
+        result = owasp_api_top10_2023.OwaspApiTop10_2023().parse(
+            self.collection, prompt_client.PromptHandler(database=self.collection)
+        )
+
+        entries = result.results["OWASP API Security Top 10 2023"]
+        self.assertEqual(10, len(entries))
+        self.assertEqual("API1", entries[0].sectionID)
+        self.assertEqual("Broken Object Level Authorization", entries[0].section)
+        self.assertEqual(
+            ["304-667", "724-770"],
+            [link.document.id for link in entries[0].links],
+        )
+        self.assertEqual("API10", entries[-1].sectionID)
+        self.assertEqual(
+            ["715-223"],
+            [link.document.id for link in entries[-1].links],
+        )