Cannot seek stdin on pipe (#496)

* add namespaces for parent attributes

* pylint formatting changes

* added _Seeker for running seek on sys.stdin

* Update node_visitor.py

* Update general_hardcoded_password.py

* Update general_hardcoded_password.py

* pep8 fixes

* added list handling for hard fname swaps

* updated manager

* maintaining list order

* Update manager.py

* Update manager.py

* Update manager.py

* Update issue.py

* Update node_visitor.py

* Update manager.py

* Update issue.py

* Update context.py

* Update issue.py

* Update manager.py

* Update node_visitor.py

* Update tester.py

* Update issue.py

* Update manager.py

* Update context.py

* Update node_visitor.py

* Update manager.py

Co-authored-by: wxu <[email protected]>
Co-authored-by: Eric Brown <[email protected]>

Author: 3 people

bandit/core/context.py

@@ -315,3 +315,7 @@ def is_module_imported_like(self, module):
     @property
     def filename(self):
         return self._context.get("filename")
+
+    @property
+    def file_data(self):
+        return self._context.get("file_data")

bandit/core/issue.py

@@ -93,6 +93,7 @@ def __init__(
         self.text = text
         self.ident = ident
         self.fname = ""
+        self.fdata = None
         self.test = ""
         self.test_id = test_id
         self.lineno = lineno
@@ -171,9 +172,17 @@ def get_code(self, max_lines=3, tabbed=False):
         lmin = max(1, self.lineno - max_lines // 2)
         lmax = lmin + len(self.linerange) + max_lines - 1
 
+        if self.fname == "<stdin>":
+            self.fdata.seek(0)
+            for line_num in range(1, lmin):
+                self.fdata.readline()
+
         tmplt = "%i\t%s" if tabbed else "%i %s"
         for line in range(lmin, lmax):
-            text = linecache.getline(self.fname, line)
+            if self.fname == "<stdin>":
+                text = self.fdata.readline()
+            else:
+                text = linecache.getline(self.fname, line)
 
             if isinstance(text, bytes):
                 text = text.decode("utf-8")

bandit/core/manager.py

@@ -4,6 +4,7 @@
 # SPDX-License-Identifier: Apache-2.0
 import collections
 import fnmatch
+import io
 import json
 import logging
 import os
@@ -277,8 +278,12 @@ def run_tests(self):
                     self._show_progress("%s.. " % count, flush=True)
             try:
                 if fname == "-":
-                    sys.stdin = os.fdopen(sys.stdin.fileno(), "rb", 0)
-                    self._parse_file("<stdin>", sys.stdin, new_files_list)
+                    open_fd = os.fdopen(sys.stdin.fileno(), "rb", 0)
+                    fdata = io.BytesIO(open_fd.read())
+                    new_files_list = [
+                        "<stdin>" if x == "-" else x for x in new_files_list
+                    ]
+                    self._parse_file("<stdin>", fdata, new_files_list)
                 else:
                     with open(fname, "rb") as fdata:
                         self._parse_file(fname, fdata, new_files_list)
@@ -333,7 +338,7 @@ def _parse_file(self, fname, fdata, new_files_list):
 
             except tokenize.TokenError:
                 pass
-            score = self._execute_ast_visitor(fname, data, nosec_lines)
+            score = self._execute_ast_visitor(fname, fdata, data, nosec_lines)
             self.scores.append(score)
             self.metrics.count_issues(
                 [
@@ -360,7 +365,7 @@ def _parse_file(self, fname, fdata, new_files_list):
             LOG.debug("  Exception string: %s", e)
             LOG.debug("  Exception traceback: %s", traceback.format_exc())
 
-    def _execute_ast_visitor(self, fname, data, nosec_lines):
+    def _execute_ast_visitor(self, fname, fdata, data, nosec_lines):
         """Execute AST parse on each file
 
         :param fname: The name of the file being parsed
@@ -370,7 +375,13 @@ def _execute_ast_visitor(self, fname, data, nosec_lines):
         """
         score = []
         res = b_node_visitor.BanditNodeVisitor(
-            fname, self.b_ma, self.b_ts, self.debug, nosec_lines, self.metrics
+            fname,
+            fdata,
+            self.b_ma,
+            self.b_ts,
+            self.debug,
+            nosec_lines,
+            self.metrics,
         )
 
         score = res.process(data)

bandit/core/node_visitor.py

@@ -15,7 +15,9 @@
 
 
 class BanditNodeVisitor:
-    def __init__(self, fname, metaast, testset, debug, nosec_lines, metrics):
+    def __init__(
+        self, fname, fdata, metaast, testset, debug, nosec_lines, metrics
+    ):
         self.debug = debug
         self.nosec_lines = nosec_lines
         self.seen = 0
@@ -25,6 +27,7 @@ def __init__(self, fname, metaast, testset, debug, nosec_lines, metrics):
         }
         self.depth = 0
         self.fname = fname
+        self.fdata = fdata
         self.metaast = metaast
         self.testset = testset
         self.imports = set()
@@ -37,7 +40,7 @@ def __init__(self, fname, metaast, testset, debug, nosec_lines, metrics):
         try:
             self.namespace = b_utils.get_module_qualname_from_path(fname)
         except b_utils.InvalidModulePath:
-            LOG.info(
+            LOG.warning(
                 "Unable to find qualified name for module: %s", self.fname
             )
             self.namespace = ""
@@ -214,6 +217,7 @@ def pre_visit(self, node):
         self.context["node"] = node
         self.context["linerange"] = b_utils.linerange_fix(node)
         self.context["filename"] = self.fname
+        self.context["file_data"] = self.fdata
 
         self.seen += 1
         LOG.debug(

bandit/core/tester.py

@@ -61,6 +61,7 @@ def run_tests(self, raw_context, checktype):
                         result.fname = temp_context["filename"].decode("utf-8")
                     else:
                         result.fname = temp_context["filename"]
+                    result.fdata = temp_context["file_data"]
 
                     if result.lineno is None:
                         result.lineno = temp_context["lineno"]