fix: address codeql alerts on raises, tar links, and duplicate tests

Author: KAJdev

runpod/apps/api.py

@@ -41,7 +41,8 @@ async def _execute(
                 # transport-level failures (ssl hiccups, resets, dns)
                 # are transient; graphql/auth errors propagate untouched
                 last_exc = exc
-        raise last_exc  # type: ignore[misc]
+        assert last_exc is not None  # loop always runs at least once
+        raise last_exc
 
     # -- endpoints (dev session provisioning) --
 

runpod/apps/targets.py

@@ -184,7 +184,8 @@ async def _request_json(
             # decided retryability
             last_exc = exc
             continue
-    raise last_exc  # type: ignore[misc]
+    assert last_exc is not None  # loop always runs at least once
+    raise last_exc
 
 
 async def _post_json(

runpod/apps/tasks.py

@@ -281,6 +281,7 @@ async def poll_result(self) -> Optional[Dict[str, Any]]:
         """
         url = f"{_proxy_url(self.pod_id)}/result"
         attempts = 6
+        data: Dict[str, Any] = {}
         async with aiohttp.ClientSession() as session:
             for attempt in range(attempts):
                 async with session.get(

runpod/runtimes/bootstrap.py

@@ -117,15 +117,28 @@ def _locate():
                 # and absolute paths
                 tar.extractall(path=APP_DIR, filter="data")
             else:
-                for member in tar.getmembers():
+                members = tar.getmembers()
+                for member in members:
                     dest = os.path.realpath(
                         os.path.join(APP_DIR, member.name)
                     )
                     if not dest.startswith(target + os.sep) and dest != target:
                         raise PhaseError(
                             "locate", f"unsafe tar member path: {member.name}"
                         )
-                tar.extractall(path=APP_DIR)
+                    # links can point outside and later members write
+                    # through them; artifacts never contain links
+                    if member.issym() or member.islnk():
+                        raise PhaseError(
+                            "locate",
+                            f"unsafe tar link member: {member.name}",
+                        )
+                    if not (member.isfile() or member.isdir()):
+                        raise PhaseError(
+                            "locate",
+                            f"unsupported tar member type: {member.name}",
+                        )
+                tar.extractall(path=APP_DIR, members=members)  # noqa: S202 - members validated above
     except (OSError, tarfile.TarError) as exc:
         raise PhaseError("locate", f"failed to extract artifact: {exc}")
 

tests/test_apps/test_model.py

@@ -72,9 +72,12 @@ def test_task_decorator_rejects_model(self):
 
         app = App("modeltest")
 
+        # model= is not a parameter of @app.task; passed via a kwargs
+        # dict so the intentional misuse is constructed at runtime
+        kwargs = {"name": "train", "gpu": "4090", "model": Model("org/name")}
         with pytest.raises(TypeError, match="model"):
 
-            @app.task(name="train", gpu="4090", model=Model("org/name"))
+            @app.task(**kwargs)
             def train():
                 pass
 

tests/test_apps/test_tasks.py

@@ -472,27 +472,6 @@ async def test_execute_timeout(self):
                 await execution.execute({"fn": "t"}, timeout=10)
 
 
-class TestUnwrapTaskResponse:
-    def test_failure_raises(self):
-        from runpod.apps.errors import RemoteExecutionError
-        from runpod.apps.tasks import unwrap_task_response
-
-        with pytest.raises(RemoteExecutionError, match="worker oom"):
-            unwrap_task_response({"success": False, "error": "worker oom"})
-
-    def test_json_result(self):
-        from runpod.apps.tasks import unwrap_task_response
-
-        assert unwrap_task_response(
-            {"success": True, "json_result": {"x": 1}}
-        ) == {"x": 1}
-
-    def test_pickled_result(self):
-        from runpod.apps.tasks import unwrap_task_response
-
-        assert unwrap_task_response({"success": True, "result": _b64(7)}) == 7
-
-
 class TestTaskJob:
     def _job(self):
         from runpod.apps.tasks import TaskExecution, TaskJob