bandit: a static security analyser

.bandit

@@ -0,0 +1,20 @@
+# NOTE: At present Bandit does *not* automatically source the .bandit
+# configuration file, invoke like so:
+#   $ bandit -r --ini .bandit
+# For development it may be convenient to use flake8-bandit.
+
+[bandit]
+targets: cylc/flow
+skips: B404,B607
+# B404: import_subprocess - consider security implications
+#       * Ignored as the use of subprocesses is a project wide decision.
+#       * The purpose of cylc is to run user defined code therefore the use
+#         of subprocesses is unavoidable.
+# B607: Starting a process with a partial executable path
+#       * Ignored as Cylc needs to be able to call out to executables which
+#         may have to be installed into environments so we cannot specify
+#         absolute paths. In some cases this may be required for portability.
+#       * Users can modify their $PATH to get Cylc to call malicious scripts,
+#         however, the purpose of Cylc Flow is to run user-defined code making
+#         this a moot point, note all code is run as the user. Cylc Flow does
+#         *not* provide multi-user functionality.

.github/workflows/test_fast.yml

@@ -70,6 +70,10 @@ jobs:
         run: |
           pytest tests/unit
 
+      - name: Bandit
+        run: |
+          bandit -r --ini .bandit cylc/flow
+
       - name: Integration Tests
         run: |
           pytest tests/integration

cylc/flow/dbstatecheck.py

@@ -75,9 +75,16 @@ def display_maps(res):
     def get_remote_point_format(self):
         """Query a remote workflow database for a 'cycle point format' entry"""
         for row in self.conn.execute(
-                r"SELECT value FROM " + CylcWorkflowDAO.TABLE_WORKFLOW_PARAMS +
-                r" WHERE key==?",
-                ['cycle_point_format']):
+            rf'''
+                SELECT
+                    value
+                FROM
+                    {CylcWorkflowDAO.TABLE_WORKFLOW_PARAMS}
+                WHERE
+                    key==?
+            ''',  # nosec (table name is code constant)
+            ['cycle_point_format']
+        ):
             return row[0]
 
     def state_lookup(self, state):
@@ -102,7 +109,14 @@ def workflow_state_query(
         else:
             target_table = CylcWorkflowDAO.TABLE_TASK_STATES
 
-        stmt = "select {0} from {1}".format(mask, target_table)
+        stmt = rf'''
+            SELECT
+                {mask}
+            FROM
+                {target_table}
+        '''  # nosec
+        # * mask is hardcoded
+        # * target_table is a code constant
         if task is not None:
             stmt_wheres.append("name==?")
             stmt_args.append(task)

cylc/flow/install_plugins/log_vc_info.py

@@ -192,8 +192,15 @@ def _run_cmd(vcs: str, args: Iterable[str], cwd: Union[Path, str]) -> str:
     """
     cmd = [vcs, *args]
     try:
-        proc = Popen(
-            cmd, cwd=cwd, stdin=DEVNULL, stdout=PIPE, stderr=PIPE, text=True)
+        proc = Popen(  # nosec
+            cmd,
+            cwd=cwd,
+            stdin=DEVNULL,
+            stdout=PIPE,
+            stderr=PIPE,
+            text=True,
+        )
+        # commands are defined in constants at top of module
     except FileNotFoundError as exc:
         # This will only be raised if the VCS command is not installed,
         # otherwise Popen() will succeed with a non-zero return code

cylc/flow/job_file.py

@@ -81,9 +81,13 @@ def write(self, local_job_file_path, job_conf, check_syntax=True):
         # check syntax
         if check_syntax:
             try:
-                with Popen(
-                        ['/usr/bin/env', 'bash', '-n', tmp_name],
-                        stderr=PIPE, stdin=DEVNULL) as proc:
+                with Popen(  # nosec
+                    ['/usr/bin/env', 'bash', '-n', tmp_name],
+                    stderr=PIPE,
+                    stdin=DEVNULL,
+                    # * the purpose of this is to evaluate user devined code
+                    #   prior to it being executed
+                ) as proc:
                     if proc.wait():
                         # This will leave behind the temporary file,
                         # which is useful for debugging syntax errors, etc.

cylc/flow/job_runner_handlers/background.py

@@ -66,7 +66,7 @@ def submit(cls, job_file_path, submit_opts):
             # process can detach as a process group leader and not subjected to
             # SIGHUP from the current process.
             # TODO: close stdout? Maybe atexit?
-            proc = Popen(
+            proc = Popen(  # nosec
                 [
                     "nohup",
                     "bash",
@@ -81,6 +81,7 @@ def submit(cls, job_file_path, submit_opts):
                 stdout=DEVNULL,
                 stderr=STDOUT
             )
+            # * the purpose of this call is to run user defined code
         except OSError as exc:
             # subprocess.Popen has a bad habit of not setting the
             # filename of the executable when it raises an OSError.

cylc/flow/parsec/validate.py

@@ -71,7 +71,9 @@ class ParsecValidator:
     # Parameterized names containing at least one comma.
     _REC_MULTI_PARAM = re.compile(r'<[\w]+,.*?>')
 
-    SELF_REFERENCE_PATTERNS = ['localhost', '127.0.0.1', '0.0.0.0']
+    SELF_REFERENCE_PATTERNS = ['localhost', '127.0.0.1', '0.0.0.0']  # nosec
+    # * these strings are used for validation purposes
+    # * they are not used for binding
 
     # Value type constants
     V_BOOLEAN = 'V_BOOLEAN'

cylc/flow/platforms.py

@@ -156,7 +156,7 @@ def platform_from_name(
     for platform_name_re in reversed(list(platform_groups)):
         if re.fullmatch(platform_name_re, platform_name):
             platform_group = deepcopy(platform_name)
-            platform_name = random.choice(
+            platform_name = random.choice(  # nosec (not crypto related)
                 platform_groups[platform_name_re]['platforms']
             )
 
@@ -562,7 +562,7 @@ def get_random_platform_for_install_target(
     """Return a randomly selected platform (dict) for given install target."""
     platforms = get_all_platforms_for_install_target(install_target)
     try:
-        return random.choice(platforms)
+        return random.choice(platforms)  # nosec (not crypto related)
     except IndexError:
         # No platforms to choose from
         raise PlatformLookupError(

cylc/flow/prerequisite.py

@@ -154,7 +154,9 @@ def _conditional_is_satisfied(self):
 
         """
         try:
-            res = eval(self.conditional_expression)
+            res = eval(self.conditional_expression)  # nosec
+            # * the expression is constructed internally
+            # * https://github.com/cylc/cylc-flow/issues/4403
         except (SyntaxError, ValueError) as exc:
             err_msg = str(exc)
             if str(exc).find("unexpected EOF") != -1:

cylc/flow/remote.py

@@ -41,7 +41,12 @@ def get_proc_ancestors():
     pid = os.getpid()
     ancestors = []
     while True:
-        p = Popen(["ps", "-p", str(pid), "-oppid="], stdout=PIPE, stderr=PIPE)
+        p = Popen(  # nosec
+            ["ps", "-p", str(pid), "-oppid="],
+            stdout=PIPE,
+            stderr=PIPE,
+        )
+        # * there is no untrusted output
         ppid = p.communicate()[0].decode().strip()
         if not ppid:
             return ancestors
@@ -116,7 +121,13 @@ def run_cmd(
         stdin = read
 
     try:
-        proc = Popen(command, stdin=stdin, stdout=stdout, stderr=stderr)
+        proc = Popen(  # nosec
+            command,
+            stdin=stdin,
+            stdout=stdout,
+            stderr=stderr,
+        )
+        # * this see CODACY ISSUE comment above
     except OSError as exc:
         sys.exit(r'ERROR: %s: %s' % (
             exc, ' '.join(quote(item) for item in command)))

cylc/flow/rundb.py

@@ -494,16 +494,25 @@ def select_workflow_params(self, callback):
 
         E.g. a row might be ['UTC mode', '1']
         """
-        stmt = f"SELECT key, value FROM {self.TABLE_WORKFLOW_PARAMS}"
+        stmt = rf'''
+            SELECT
+                key, value
+            FROM
+                {self.TABLE_WORKFLOW_PARAMS}
+        '''  # nosec (table name is code constant)
         for row_idx, row in enumerate(self.connect().execute(stmt)):
             callback(row_idx, list(row))
 
     def select_workflow_params_restart_count(self):
         """Return number of restarts in workflow_params table."""
-        stmt = f"""
-            SELECT value FROM {self.TABLE_WORKFLOW_PARAMS}
-            WHERE key == 'n_restart';
-        """
+        stmt = rf"""
+            SELECT
+                value
+            FROM
+                {self.TABLE_WORKFLOW_PARAMS}
+            WHERE
+                key == 'n_restart'
+        """  # nosec (table name is code constant)
         result = self.connect().execute(stmt).fetchone()
         return int(result[0]) if result else 0
 
@@ -514,8 +523,13 @@ def select_workflow_template_vars(self, callback):
             [key,value]
         """
         for row_idx, row in enumerate(self.connect().execute(
-                r"SELECT key,value FROM %s" % (
-                    self.TABLE_WORKFLOW_TEMPLATE_VARS))):
+                rf'''
+                    SELECT
+                        key, value
+                    FROM
+                        {self.TABLE_WORKFLOW_TEMPLATE_VARS}
+                '''  # nosec (table name is code constant)
+        )):
             callback(row_idx, list(row))
 
     def select_task_action_timers(self, callback):
@@ -526,8 +540,14 @@ def select_task_action_timers(self, callback):
         attrs = []
         for item in self.TABLES_ATTRS[self.TABLE_TASK_ACTION_TIMERS]:
             attrs.append(item[0])
-        stmt = r"SELECT %s FROM %s" % (
-            ",".join(attrs), self.TABLE_TASK_ACTION_TIMERS)
+        stmt = rf'''
+            SELECT
+                {",".join(attrs)}
+            FROM
+                {self.TABLE_TASK_ACTION_TIMERS}
+        '''  # nosec
+        # * table name is code constant
+        # * attrs are code constants
         for row_idx, row in enumerate(self.connect().execute(stmt)):
             callback(row_idx, list(row))
 
@@ -541,17 +561,33 @@ def select_task_job(self, cycle, name, submit_num=None):
         for column in self.tables[self.TABLE_TASK_JOBS].columns[3:]:
             keys.append(column.name)
         if submit_num in [None, "NN"]:
-            stmt = (r"SELECT %(keys_str)s FROM %(table)s"
-                    r" WHERE cycle==? AND name==?"
-                    r" ORDER BY submit_num DESC LIMIT 1") % {
-                "keys_str": ",".join(keys),
-                "table": self.TABLE_TASK_JOBS}
+            stmt = f'''
+                SELECT
+                    {",".join(keys)}
+                FROM
+                    {self.TABLE_TASK_JOBS}
+                WHERE
+                    cycle==?
+                    AND name==?"
+                ORDER BY
+                    submit_num DESC LIMIT 1
+            '''  # nosec
+            # * table name is code constant
+            # * keys are code constants
             stmt_args = [cycle, name]
         else:
-            stmt = (r"SELECT %(keys_str)s FROM %(table)s"
-                    r" WHERE cycle==? AND name==? AND submit_num==?") % {
-                "keys_str": ",".join(keys),
-                "table": self.TABLE_TASK_JOBS}
+            stmt = f'''
+                SELECT
+                    {",".join(keys)}
+                FROM
+                    {self.TABLE_TASK_JOBS}
+                WHERE
+                    cycle==?
+                    AND name==?
+                    AND submit_num==?
+            '''  # nosec
+            # * table name is code constant
+            # * keys are code constants
             stmt_args = [cycle, name, submit_num]
         try:
             for row in self.connect().execute(stmt, stmt_args):
@@ -625,7 +661,12 @@ def select_task_job_run_times(self, callback):
 
     def select_task_job_platforms(self):
         """Return the set of platform names from task_jobs table."""
-        stmt = f"SELECT DISTINCT platform_name FROM {self.TABLE_TASK_JOBS}"
+        stmt = rf'''
+            SELECT DISTINCT
+                platform_name
+            FROM
+                {self.TABLE_TASK_JOBS}
+        '''  # nosec (table name is code constant)
         return {i[0] for i in self.connect().execute(stmt)}
 
     def select_submit_nums(self, name, point):
@@ -656,13 +697,23 @@ def select_submit_nums(self, name, point):
         return ret
 
     def select_xtriggers_for_restart(self, callback):
-        stm = r"SELECT signature,results FROM %s" % self.TABLE_XTRIGGERS
-        for row_idx, row in enumerate(self.connect().execute(stm, [])):
+        stmt = rf'''
+            SELECT
+                signature,results
+            FROM
+                {self.TABLE_XTRIGGERS}
+        '''  # nosec (table name is code constant)
+        for row_idx, row in enumerate(self.connect().execute(stmt, [])):
             callback(row_idx, list(row))
 
     def select_abs_outputs_for_restart(self, callback):
-        stm = r"SELECT cycle,name,output FROM %s" % self.TABLE_ABS_OUTPUTS
-        for row_idx, row in enumerate(self.connect().execute(stm, [])):
+        stmt = rf'''
+            SELECT
+                cycle, name, output
+            FROM
+                {self.TABLE_ABS_OUTPUTS}
+        '''  # nosec (table name is code constant)
+        for row_idx, row in enumerate(self.connect().execute(stmt, [])):
             callback(row_idx, list(row))
 
     def select_task_pool(self, callback):
@@ -739,7 +790,7 @@ def select_task_prerequisites(
         self, cycle: str, name: str
     ) -> List[Tuple[str, str, str, str]]:
         """Return prerequisites of a task of the given name & cycle point."""
-        stmt = f"""
+        stmt = rf"""
             SELECT
                 prereq_name,
                 prereq_cycle,
@@ -750,13 +801,18 @@ def select_task_prerequisites(
             WHERE
                 cycle == ? AND
                 name == ?
-        """
+        """  # nosec (table name is code constant)
         stmt_args = [cycle, name]
         return list(self.connect().execute(stmt, stmt_args))
 
     def select_tasks_to_hold(self) -> List[Tuple[str, str]]:
         """Return all tasks to hold stored in the DB."""
-        stmt = f"SELECT name, cycle FROM {self.TABLE_TASKS_TO_HOLD}"
+        stmt = rf'''
+            SELECT
+                name, cycle
+            FROM
+                {self.TABLE_TASKS_TO_HOLD}
+        '''  # nosec (table name is code constant)
         return list(self.connect().execute(stmt))
 
     def select_task_times(self):
@@ -765,7 +821,7 @@ def select_task_times(self):
         To make data interpretation easier, choose the most recent succeeded
         task to sample timings from.
         """
-        q = """
+        stmt = rf"""
             SELECT
                 name,
                 cycle,
@@ -775,18 +831,15 @@ def select_task_times(self):
                 time_run,
                 time_run_exit
             FROM
-                %(task_jobs)s
+                {self.TABLE_TASK_JOBS}
             WHERE
-                run_status = %(succeeded)d
-        """ % {
-            'task_jobs': self.TABLE_TASK_JOBS,
-            'succeeded': 0,
-        }
+                run_status = 0
+        """  # nosec (table name is code constant)
         columns = (
             'name', 'cycle', 'host', 'job_runner',
             'submit_time', 'start_time', 'succeed_time'
         )
-        return columns, list(self.connect().execute(q))
+        return columns, list(self.connect().execute(stmt))
 
     def vacuum(self):
         """Vacuum to the database."""