add security auditing of python code using Bandit during CICD

.github/workflows/ci.yml

@@ -103,6 +103,7 @@ jobs:
         cd src/cli
         pip install -r requirements-lint.txt
         flake8 .
+        bandit -r ./onefuzz/
         black onefuzz examples tests --check
         isort --profile black ./onefuzz ./examples/ ./tests/ --check
         pytest -v tests
@@ -208,6 +209,7 @@ jobs:
         pip install -r requirements-dev.txt
         pytest
         flake8 .
+        bandit -r ./__app__/
         black ./__app__/ ./tests --check
         isort --profile black ./__app__/ ./tests --check
         mypy __app__ ./tests

src/api-service/__app__/onefuzzlib/azure/storage.py

@@ -86,7 +86,9 @@ def choose_account(storage_type: StorageType) -> str:
     # Use a random secondary storage account if any are available.  This
     # reduces IOP contention for the Storage Queues, which are only available
     # on primary accounts
-    return random.choice(accounts[1:])
+    #
+    # security note: this is not used as a security feature
+    return random.choice(accounts[1:])  # nosec
 
 
 @cached

src/api-service/requirements-dev.txt

@@ -4,3 +4,4 @@ mypy
 isort
 vulture
 black
+bandit

src/ci/onefuzztypes.sh

@@ -14,6 +14,7 @@ python setup.py sdist bdist_wheel
 pip install -r requirements-lint.txt
 black ./onefuzztypes --check
 flake8 ./onefuzztypes
+bandit -r ./onefuzztypes
 isort --profile black ./onefuzztypes --check
 mypy ./onefuzztypes --ignore-missing-imports
 pytest -v tests

src/cli/onefuzz/api.py

@@ -66,7 +66,10 @@ def is_uuid(value: str) -> bool:
 
 def wsl_path(path: str) -> str:
     if which("wslpath"):
-        return subprocess.check_output(["wslpath", "-w", path]).decode().strip()
+        # security note: path is a temporary path constructed by this library
+        return (
+            subprocess.check_output(["wslpath", "-w", path]).decode().strip()  # nosec
+        )
     return path
 
 
@@ -530,7 +533,9 @@ def _dbg_linux(
                     dbg += ["--batch"]
 
                     try:
-                        return subprocess.run(
+                        # security note: dbg is built from content coming from
+                        # the server, which is trusted in this context.
+                        return subprocess.run(  # nosec
                             dbg, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
                         ).stdout.decode(errors="ignore")
                     except subprocess.CalledProcessError as err:
@@ -539,7 +544,9 @@ def _dbg_linux(
                         )
                         raise err
                 else:
-                    subprocess.call(dbg)
+                    # security note: dbg is built from content coming from the
+                    # server, which is trusted in this context.
+                    subprocess.call(dbg)  # nosec
         return None
 
     def _dbg_windows(
@@ -565,7 +572,9 @@ def _dbg_windows(
 
                     logging.debug("launching: %s", dbg)
                     try:
-                        return subprocess.run(
+                        # security note: dbg is built from content coming from the server,
+                        # which is trusted in this context.
+                        return subprocess.run(  # nosec
                             dbg, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
                         ).stdout.decode(errors="ignore")
                     except subprocess.CalledProcessError as err:
@@ -575,7 +584,9 @@ def _dbg_windows(
                         raise err
             else:
                 logging.debug("launching: %s", dbg)
-                subprocess.call(dbg)
+                # security note:  dbg is built from content coming from the
+                # server, which is trusted in this context.
+                subprocess.call(dbg)  # nosec
 
         return None
 

src/cli/onefuzz/debug.py

@@ -360,7 +360,12 @@ def download_files(self, job_id: UUID_EXPANSION, output: Directory) -> None:
             if not os.path.exists(outdir):
                 os.makedirs(outdir)
             self.logger.info("downloading: %s", name)
-            subprocess.check_output([azcopy, "sync", to_download[name], outdir])
+            # security note: the src for azcopy comes from the server which is
+            # trusted in this context, while the destination is provided by the
+            # user
+            subprocess.check_output(  # nosec
+                [azcopy, "sync", to_download[name], outdir]
+            )
 
 
 class DebugLog(Command):

src/cli/onefuzz/job_templates/handlers.py

@@ -68,9 +68,12 @@ def _define_missing_containers(
                 if container_type == container.type:
                     seen = True
             if not seen:
-                assert isinstance(request.user_fields["project"], str)
-                assert isinstance(request.user_fields["name"], str)
-                assert isinstance(request.user_fields["build"], str)
+                if not isinstance(request.user_fields["project"], str):
+                    raise TypeError
+                if not isinstance(request.user_fields["name"], str):
+                    raise TypeError
+                if not isinstance(request.user_fields["build"], str):
+                    raise TypeError
                 container_name = _build_container_name(
                     self.onefuzz,
                     container_type,

src/cli/onefuzz/rdp.py

@@ -25,7 +25,9 @@ def rdp_connect(ip: str, password: str, *, port: int) -> Generator:
         )
 
         encrypted = (
-            subprocess.check_output(
+            # security note: script includes content coming from the server,
+            # which is trusted in this context.
+            subprocess.check_output(  # nosec
                 ["powershell.exe", "-ExecutionPolicy", "Unrestricted", script]
             )
             .decode()
@@ -46,5 +48,7 @@ def rdp_connect(ip: str, password: str, *, port: int) -> Generator:
         os.chdir(tmpdir)
         cmd = ["mstsc.exe", FILENAME]
         logging.info("launching rdp: %s", " ".join(cmd))
-        yield subprocess.call(cmd)
+        # security note: mstsc.exe is called using a config with data coming
+        # from the server, which is trusted in this context.
+        yield subprocess.call(cmd)  # nosec
         os.chdir(previous)

src/cli/onefuzz/ssh.py

@@ -43,7 +43,8 @@ def temp_file(
             handle.write(content)
 
         if permissions is not None and platform.system() != "Windows":
-            subprocess.check_call(["chmod", permissions, full_path])
+            # security note: arguments all checked
+            subprocess.check_call(["chmod", permissions, full_path])  # nosec
 
         yield full_path
 
@@ -102,10 +103,14 @@ def ssh_connect(
         logging.info("launching ssh: %s", " ".join(cmd))
 
         if call:
-            yield subprocess.call(cmd)
+            # security note: command includes user provided arguments
+            # intentionally
+            yield subprocess.call(cmd)  # nosec
             return
 
-        with subprocess.Popen(
+        # security note: command includes user provided arguments
+        # intentionally
+        with subprocess.Popen(  # nosec
             cmd, stdin=PIPE, stdout=PIPE, stderr=PIPE, bufsize=0
         ) as ssh:
             yield ssh

src/cli/onefuzz/templates/ossfuzz.py

@@ -80,7 +80,9 @@ def _copy_all(self, src_sas: str, dst_sas: str) -> None:
             dst_sas,
         ]
         self.logger.info("copying base setup")
-        subprocess.check_output(cmd)
+        # security note: the source and destination container sas URLS are
+        # considerd trusted from the service
+        subprocess.check_output(cmd)  # nosec
 
     def _copy_exe(self, src_sas: str, dst_sas: str, target_exe: File) -> None:
         files: List[File] = [target_exe]
@@ -100,7 +102,9 @@ def _copy_exe(self, src_sas: str, dst_sas: str, target_exe: File) -> None:
                 dst_url,
             ]
             self.logger.info("uploading %s", path)
-            subprocess.check_output(cmd)
+            # security note: the source and destination container sas URLS
+            # are considerd trusted from the service
+            subprocess.check_output(cmd)  # nosec
 
     def libfuzzer(
         self,
@@ -141,7 +145,10 @@ def libfuzzer(
             container_sas[name] = sas_url
 
         self.logger.info("uploading build artifacts")
-        subprocess.check_output(
+
+        # security note: the container sas is considerd trusted from the
+        # service
+        subprocess.check_output(  # nosec
             [
                 "azcopy",
                 "sync",
@@ -151,7 +158,7 @@ def libfuzzer(
                 "*fuzzer_seed_corpus.zip",
             ]
         )
-        subprocess.check_output(
+        subprocess.check_output(  # nosec
             [
                 "azcopy",
                 "sync",

src/cli/requirements-lint.txt

@@ -4,3 +4,4 @@ pytest
 isort
 vulture
 black
+bandit

src/pytypes/requirements-lint.txt

@@ -5,3 +5,4 @@ isort
 pydantic
 vulture
 black
+bandit