appease bandit

Author: mr-c

Makefile

@@ -152,7 +152,7 @@ test: $(PYSOURCES) FORCE
 
 ## testcov                : run the wes-service test suite and collect coverage
 testcov: $(PYSOURCES)
-	pytest --cov ${PYTEST_EXTRA}
+	python -m pytest -rsx --cov ${PYTEST_EXTRA}
 
 sloccount.sc: $(PYSOURCES) Makefile
 	sloccount --duplicates --wide --details $^ > $@

test-requirements.txt

@@ -1 +1,2 @@
 pytest
+pytest-cov

wes_client/util.py

@@ -4,7 +4,8 @@
 import json
 import logging
 import os
-from subprocess import DEVNULL, CalledProcessError, check_call
+import sys
+from subprocess import DEVNULL, CalledProcessError, check_call  # nosec B404
 from typing import Any, Dict, List, Optional, Set, Tuple, Union, cast
 from urllib.request import pathname2url, urlopen
 
@@ -18,7 +19,10 @@
 def py3_compatible(filePath: str) -> bool:
     """Determines if a python file is 3.x compatible by seeing if it compiles in a subprocess"""
     try:
-        check_call(["python3", "-m", "py_compile", filePath], stderr=DEVNULL)
+        check_call(
+            [sys.executable, "-m", "py_compile", os.path.normpath(filePath)],
+            stderr=DEVNULL,
+        )  # nosec B603
     except CalledProcessError as e:
         raise RuntimeError("Python files must be 3.x compatible") from e
     return True
@@ -29,9 +33,7 @@ def get_version(extension: str, workflow_file: str) -> str:
     if extension == "py" and py3_compatible(workflow_file):
         return "3"
     elif extension == "cwl":
-        return cast(
-            str, yaml.load(open(workflow_file), Loader=yaml.FullLoader)["cwlVersion"]
-        )
+        return cast(str, yaml.safe_load(open(workflow_file))["cwlVersion"])
     else:  # Must be a wdl file.
         # Borrowed from https://github.com/Sage-Bionetworks/synapse-orchestrator/
         #               blob/develop/synorchestrator/util.py#L142
@@ -66,7 +68,7 @@ def wf_info(workflow_path: str) -> Tuple[str, str]:
             "http://"
         ):
             # If file not local go fetch it.
-            html = urlopen(workflow_path).read()
+            html = urlopen(workflow_path).read()  # nosec B310
             local_loc = os.path.join(os.getcwd(), "fetchedFromRemote." + file_type)
             with open(local_loc, "w") as f:
                 f.write(html.decode())
@@ -174,7 +176,7 @@ def build_wes_request(
                 attach_f: Any = open(attachment, "rb")
                 relpath = os.path.relpath(attachment, wfbase)
             elif attachment.startswith("http"):
-                attach_f = urlopen(attachment)
+                attach_f = urlopen(attachment)  # nosec B310
                 relpath = os.path.basename(attach_f)
 
             parts.append(("workflow_attachment", (relpath, attach_f)))
@@ -226,7 +228,7 @@ def get_service_info(self) -> Dict[str, Any]:
         :param host: Port where the post request will be sent and the wes server listens at (default 8080)
         :return: The body of the get result as a dictionary.
         """
-        postresult = requests.get(
+        postresult = requests.get(  # nosec B113
             f"{self.proto}://{self.host}/ga4gh/wes/v1/service-info",
             headers=self.auth,
         )
@@ -244,7 +246,7 @@ def list_runs(self) -> Dict[str, Any]:
         :param host: Port where the post request will be sent and the wes server listens at (default 8080)
         :return: The body of the get result as a dictionary.
         """
-        postresult = requests.get(
+        postresult = requests.get(  # nosec B113
             f"{self.proto}://{self.host}/ga4gh/wes/v1/runs", headers=self.auth
         )
         return wes_reponse(postresult)
@@ -266,7 +268,7 @@ def run(
         """
         attachments = list(expand_globs(attachments))
         parts = build_wes_request(wf, jsonyaml, attachments)
-        postresult = requests.post(
+        postresult = requests.post(  # nosec B113
             f"{self.proto}://{self.host}/ga4gh/wes/v1/runs",
             files=parts,
             headers=self.auth,
@@ -283,7 +285,7 @@ def cancel(self, run_id: str) -> Dict[str, Any]:
         :param host: Port where the post request will be sent and the wes server listens at (default 8080)
         :return: The body of the delete result as a dictionary.
         """
-        postresult = requests.post(
+        postresult = requests.post(  # nosec B113
             f"{self.proto}://{self.host}/ga4gh/wes/v1/runs/{run_id}/cancel",
             headers=self.auth,
         )
@@ -299,7 +301,7 @@ def get_run_log(self, run_id: str) -> Dict[str, Any]:
         :param host: Port where the post request will be sent and the wes server listens at (default 8080)
         :return: The body of the get result as a dictionary.
         """
-        postresult = requests.get(
+        postresult = requests.get(  # nosec B113
             f"{self.proto}://{self.host}/ga4gh/wes/v1/runs/{run_id}",
             headers=self.auth,
         )
@@ -315,7 +317,7 @@ def get_run_status(self, run_id: str) -> Dict[str, Any]:
         :param host: Port where the post request will be sent and the wes server listens at (default 8080)
         :return: The body of the get result as a dictionary.
         """
-        postresult = requests.get(
+        postresult = requests.get(  # nosec B113
             f"{self.proto}://{self.host}/ga4gh/wes/v1/runs/{run_id}/status",
             headers=self.auth,
         )

wes_client/wes_client_main.py

@@ -97,7 +97,9 @@ def main(argv: List[str] = sys.argv[1:]) -> int:
 
     if args.log:
         response = client.get_run_log(run_id=args.log)
-        sys.stdout.write(requests.get(response["run_log"]["stderr"], headers=auth).text)
+        sys.stdout.write(
+            requests.get(response["run_log"]["stderr"], headers=auth).text  # nosec B113
+        )
         return 0
 
     if args.get:
@@ -146,7 +148,7 @@ def main(argv: List[str] = sys.argv[1:]) -> int:
     try:
         # TODO: Only works with Arvados atm
         logging.info(str(s["run_log"]["stderr"]))
-        logs = requests.get(s["run_log"]["stderr"], headers=auth).text
+        logs = requests.get(s["run_log"]["stderr"], headers=auth).text  # nosec B113
         logging.info("Run log:\n" + logs)
     except InvalidSchema:
         logging.info("Run log:\n" + str(s["run_log"]["stderr"]))

wes_service/arvados_wes.py

@@ -3,7 +3,7 @@
 import logging
 import os
 import shutil
-import subprocess
+import subprocess  # nosec B404
 import tempfile
 import threading
 from typing import Any, Callable, Dict, List, Optional, Tuple, Union, cast
@@ -83,8 +83,9 @@ def catch_exceptions_wrapper(self: Any, *args: str, **kwargs: str) -> Any:
 
 class ArvadosBackend(WESBackend):
     def GetServiceInfo(self) -> Dict[str, Any]:
-        stdout, stderr = subprocess.Popen(
-            ["arvados-cwl-runner", "--version"], stderr=subprocess.PIPE
+        stdout, stderr = subprocess.Popen(  # nosec B603
+            [shutil.which("arvados-cwl-runner") or "arvados-cwl-runner", "--version"],
+            stderr=subprocess.PIPE,
         ).communicate()
         return {
             "workflow_type_versions": {
@@ -218,7 +219,7 @@ def invoke_cwl_runner(
                     cr_uuid, "Executing %s" % cmd, env["ARVADOS_API_TOKEN"]
                 )
 
-                proc = subprocess.Popen(
+                proc = subprocess.Popen(  # nosec B603
                     cmd,
                     env=env,
                     cwd=tempdir,

wes_service/cwl_runner.py

@@ -1,6 +1,6 @@
 import json
 import os
-import subprocess
+import subprocess  # nosec B404
 import uuid
 from typing import Any, Dict, List, Tuple, cast
 
@@ -72,7 +72,7 @@ def run(
 
         # build args and run
         command_args: List[str] = [runner] + extra2 + [workflow_url, jsonpath]
-        proc = subprocess.Popen(
+        proc = subprocess.Popen(  # nosec B603
             command_args, stdout=output, stderr=stderr, close_fds=True, cwd=tempdir
         )
         output.close()
@@ -162,7 +162,7 @@ def cancel(self) -> None:
 class CWLRunnerBackend(WESBackend):
     def GetServiceInfo(self) -> Dict[str, Any]:
         runner = cast(str, self.getopt("runner", default="cwl-runner"))
-        stdout, stderr = subprocess.Popen(
+        stdout, stderr = subprocess.Popen(  # nosec B603
             [runner, "--version"], stderr=subprocess.PIPE
         ).communicate()
         r = {

wes_service/toil_wes.py

@@ -2,7 +2,7 @@
 import logging
 import os
 import shutil
-import subprocess
+import subprocess  # nosec B404
 import time
 import uuid
 from multiprocessing import Process
@@ -131,7 +131,7 @@ def call_cmd(self, cmd: Union[List[str], str], cwd: str) -> int:
             self.outfile,
             self.errfile,
         )
-        process = subprocess.Popen(
+        process = subprocess.Popen(  # nosec B603
             cmd, stdout=stdout, stderr=stderr, close_fds=True, cwd=cwd
         )
         stdout.close()
@@ -287,8 +287,13 @@ def getstate(self) -> Tuple[str, int]:
                     open(self.staterrorfile, "a").close()
                     return "EXECUTOR_ERROR", 255
         if (
-            subprocess.run(
-                ["toil", "status", "--failIfNotComplete", self.jobstorefile]
+            subprocess.run(  # nosec B603
+                [
+                    shutil.which("toil") or "toil",
+                    "status",
+                    "--failIfNotComplete",
+                    self.jobstorefile,
+                ]
             ).returncode
             == 0
         ):