Retry transient failures during upstream sync

Bornunique911 · Bornunique911 · commit 336be1aeb7c7 · 2026-04-27T02:31:34.000+05:30
diff --git a/application/cmd/cre_main.py b/application/cmd/cre_main.py
@@ -37,6 +37,51 @@
 app = None
 
 
+def fetch_upstream_json(
+    path: str,
+    timeout: Optional[float] = None,
+    max_attempts: Optional[int] = None,
+    backoff_seconds: Optional[float] = None,
+) -> Dict[str, Any]:
+    base_url = os.environ.get("CRE_UPSTREAM_API_URL", "https://opencre.org/rest/v1")
+    timeout = timeout or float(os.environ.get("CRE_UPSTREAM_TIMEOUT_SECONDS", "30"))
+    max_attempts = max_attempts or int(os.environ.get("CRE_UPSTREAM_MAX_ATTEMPTS", "4"))
+    backoff_seconds = backoff_seconds or float(
+        os.environ.get("CRE_UPSTREAM_RETRY_BACKOFF_SECONDS", "2")
+    )
+    url = f"{base_url}{path}"
+    last_error: Optional[Exception] = None
+
+    for attempt in range(1, max_attempts + 1):
+        try:
+            response = requests.get(url, timeout=timeout)
+            if response.status_code == 200:
+                return response.json()
+
+            status_error = RuntimeError(
+                f"cannot connect to upstream status code {response.status_code}"
+            )
+            # Retry only on transient upstream failures.
+            if response.status_code < 500 and response.status_code != 429:
+                raise status_error
+            last_error = status_error
+        except requests.exceptions.RequestException as exc:
+            last_error = exc
+
+        if attempt < max_attempts:
+            logger.warning(
+                "upstream fetch failed for %s on attempt %s/%s, retrying",
+                url,
+                attempt,
+                max_attempts,
+            )
+            time.sleep(backoff_seconds * attempt)
+
+    if last_error:
+        raise RuntimeError(f"upstream fetch failed for {url}") from last_error
+    raise RuntimeError(f"upstream fetch failed for {url}")
+
+
 def register_node(node: defs.Node, collection: db.Node_collection) -> db.Node:
     """
     for each link find if either the root node or the link have a CRE,
@@ -354,6 +399,8 @@ def register_standard(
 ):
     if os.environ.get("CRE_NO_GEN_EMBEDDINGS") == "1":
         generate_embeddings = False
+    if os.environ.get("CRE_NO_CALCULATE_GAP_ANALYSIS"):
+        calculate_gap_analysis = False
 
     if not standard_entries:
         logger.warning("register_standard() called with no standard_entries")
@@ -590,15 +637,7 @@ def download_graph_from_upstream(cache: str) -> None:
     collection = db_connect(path=cache).with_graph()
 
     def download_cre_from_upstream(creid: str):
-        cre_response = requests.get(
-            os.environ.get("CRE_UPSTREAM_API_URL", "https://opencre.org/rest/v1")
-            + f"/id/{creid}"
-        )
-        if cre_response.status_code != 200:
-            raise RuntimeError(
-                f"cannot connect to upstream status code {cre_response.status_code}"
-            )
-        data = cre_response.json()
+        data = fetch_upstream_json(f"/id/{creid}")
         credict = data["data"]
         cre = defs.Document.from_dict(credict)
         if cre.id in imported_cres:
@@ -610,15 +649,7 @@ def download_cre_from_upstream(creid: str):
             if link.document.doctype == defs.Credoctypes.CRE:
                 download_cre_from_upstream(link.document.id)
 
-    root_cres_response = requests.get(
-        os.environ.get("CRE_UPSTREAM_API_URL", "https://opencre.org/rest/v1")
-        + "/root_cres"
-    )
-    if root_cres_response.status_code != 200:
-        raise RuntimeError(
-            f"cannot connect to upstream status code {root_cres_response.status_code}"
-        )
-    data = root_cres_response.json()
+    data = fetch_upstream_json("/root_cres")
     for root_cre in data["data"]:
         cre = defs.Document.from_dict(root_cre)
         register_cre(cre, collection)
@@ -900,6 +931,54 @@ def run(args: argparse.Namespace) -> None:  # pragma: no cover
         BaseParser().register_resource(
             secure_headers.SecureHeaders, db_connection_str=args.cache_file
         )
+    if args.owasp_top10_2025_in:
+        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 args.owasp_api_top10_2023_in:
+        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 args.owasp_kubernetes_top10_2022_in:
+        from application.utils.external_project_parsers.parsers import (
+            owasp_kubernetes_top10_2022,
+        )
+
+        BaseParser().register_resource(
+            owasp_kubernetes_top10_2022.OwaspKubernetesTop10_2022,
+            db_connection_str=args.cache_file,
+        )
+    if args.owasp_kubernetes_top10_2025_in:
+        from application.utils.external_project_parsers.parsers import (
+            owasp_kubernetes_top10_2025,
+        )
+
+        BaseParser().register_resource(
+            owasp_kubernetes_top10_2025.OwaspKubernetesTop10_2025,
+            db_connection_str=args.cache_file,
+        )
+    if args.owasp_llm_top10_2025_in:
+        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 args.owasp_aisvs_in:
+        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
 
diff --git a/application/tests/cre_main_test.py b/application/tests/cre_main_test.py
@@ -7,6 +7,8 @@
 from unittest import mock
 from unittest.mock import Mock, patch
 from rq import Queue, job
+import requests
+from rq import Queue, job
 from application.utils import redis
 from application.prompt_client import prompt_client as prompt_client
 from application.tests.utils import data_gen
@@ -470,6 +472,111 @@ def test_register_cre(self) -> None:
             ],
         )
 
+    @patch("application.cmd.cre_main.time.sleep")
+    @patch("application.cmd.cre_main.requests.get")
+    def test_fetch_upstream_json_retries_transient_failures(
+        self, mock_get, mock_sleep
+    ) -> None:
+        transient_error = requests.exceptions.ConnectionError("reset by peer")
+        success_response = Mock()
+        success_response.status_code = 200
+        success_response.json.return_value = {"data": []}
+        mock_get.side_effect = [transient_error, success_response]
+
+        data = main.fetch_upstream_json("/root_cres")
+
+        self.assertEqual(data, {"data": []})
+        self.assertEqual(mock_get.call_count, 2)
+        mock_sleep.assert_called_once()
+
+    def test_parse_file(self) -> None:
+        file: List[Dict[str, Any]] = [
+            {
+                "description": "Verify that approved cryptographic algorithms are used in the generation, seeding, and verification.",
+                "doctype": defs.Credoctypes.CRE,
+                "id": "157-573",
+                "links": [
+                    {
+                        "type": defs.LinkTypes.LinkedTo,
+                        "document": {
+                            "doctype": defs.Credoctypes.Standard,
+                            "name": "TOP10",
+                            "section": "https://owasp.org/www-project-top-ten/2017/A5_2017-Broken_Access_Control",
+                        },
+                    },
+                    {
+                        "type": defs.LinkTypes.LinkedTo,
+                        "document": {
+                            "doctype": defs.Credoctypes.Standard,
+                            "name": "ISO 25010",
+                            "section": "Secure data storage",
+                        },
+                    },
+                ],
+                "name": "CREDENTIALS_MANAGEMENT_CRYPTOGRAPHIC_DIRECTIVES",
+            },
+            {
+                "description": "Desc",
+                "doctype": defs.Credoctypes.CRE,
+                "id": "141-141",
+                "name": "name",
+            },
+        ]
+        expected = [
+            defs.CRE(
+                doctype=defs.Credoctypes.CRE,
+                id="157-573",
+                description="Verify that approved cryptographic algorithms are used in the generation, seeding, and verification.",
+                name="CREDENTIALS_MANAGEMENT_CRYPTOGRAPHIC_DIRECTIVES",
+                links=[
+                    defs.Link(
+                        document=defs.Standard(
+                            doctype=defs.Credoctypes.Standard,
+                            name="TOP10",
+                            section="https://owasp.org/www-project-top-ten/2017/A5_2017-Broken_Access_Control",
+                        ),
+                        ltype=defs.LinkTypes.LinkedTo,
+                    ),
+                    defs.Link(
+                        document=defs.Standard(
+                            doctype=defs.Credoctypes.Standard,
+                            name="ISO 25010",
+                            section="Secure data storage",
+                        ),
+                        ltype=defs.LinkTypes.LinkedTo,
+                    ),
+                ],
+            ),
+            defs.CRE(id="141-141", description="Desc", name="name"),
+        ]
+        with self.assertLogs("application.cmd.cre_main", level=logging.FATAL) as logs:
+            # negative test first parse_file accepts a list of objects
+            result = main.parse_file(
+                filename="tests",
+                yamldocs=[
+                    "no",
+                    "valid",
+                    "objects",
+                    "here",
+                    {
+                        "1": 2,
+                    },
+                ],
+                scollection=self.collection,
+            )
+
+            self.assertEqual(result, None)
+            self.assertIn(
+                "CRITICAL:application.cmd.cre_main:Malformed file tests, skipping",
+                logs.output,
+            )
+
+        self.maxDiff = None
+
+        res = main.parse_file(
+            filename="tests", yamldocs=file, scollection=self.collection
+        )
+        self.assertCountEqual(res, expected)
     @patch.object(main, "db_connect")
     @patch.object(Queue, "enqueue_call")
     @patch.object(redis, "wait_for_jobs")
