Resolve aliases for black box and built-in function calls.

* Allow trigger words to be fully qualified to reduce false positives

Author: wchresta

examples/vulnerable_code/command_injection_with_aliases.py

@@ -0,0 +1,12 @@
+import os
+import os as myos
+from os import system
+from os import system as mysystem
+from subprocess import call as mycall, Popen as mypopen
+
+os.system("ls")
+myos.system("ls")
+system("ls")
+mysystem("ls")
+mycall("ls")
+mypopen("ls")

pyt/cfg/alias_helper.py

@@ -74,3 +74,22 @@ def retrieve_import_alias_mapping(names_list):
  if alias.asname:
  import_alias_names[alias.asname] = alias.name
  return import_alias_names
+
+
+def fully_qualify_alias_labels(label, aliases):
+ """Replace any aliases in label with the fully qualified name.
+
+ Args:
+ label -- A label : str representing a name (e.g. myos.system)
+ aliases -- A dict of {alias: real_name} (e.g. {'myos': 'os'})
+
+ >>> fully_qualify_alias_labels('myos.mycall', {'myos':'os'})
+ 'os.mycall'
+ """
+ for alias, full_name in aliases.items():
+ if label == alias:
+ return full_name
+ if label.startswith(alias+'.'):
+ return full_name + label[len(alias):]
+ return label
+

pyt/cfg/stmt_visitor.py

@@ -9,7 +9,8 @@
  handle_aliases_in_init_files,
  handle_fdid_aliases,
  not_as_alias_handler,
- retrieve_import_alias_mapping
+ retrieve_import_alias_mapping,
+ fully_qualify_alias_labels
 )
 from ..core.ast_helper import (
  generate_ast,
@@ -61,6 +62,7 @@
 class StmtVisitor(ast.NodeVisitor):
  def __init__(self, allow_local_directory_imports=True):
  self._allow_local_modules = allow_local_directory_imports
+ self.bb_or_bi_aliases = {}
  super().__init__()
 
  def visit_Module(self, node):
@@ -624,6 +626,10 @@ def add_blackbox_or_builtin_call(self, node, blackbox): # noqa: C901
 
  call_function_label = call_label_visitor.result[:call_label_visitor.result.find('(')]
 
+ # Check if function call matches a blackbox/built-in alias and if so, resolve it
+ # This resolves aliases like "from os import system as mysys" as: mysys -> os.system
+ call_function_label = fully_qualify_alias_labels(call_function_label, self.bb_or_bi_aliases)
+
  # Create e.g. ~call_1 = ret_func_foo
  LHS = CALL_IDENTIFIER + 'call_' + str(saved_function_call_index)
  RHS = 'ret_' + call_function_label + '('
@@ -810,7 +816,6 @@ def add_module( # noqa: C901
  module_path = module[1]
 
  parent_definitions = self.module_definitions_stack[-1]
- # The only place the import_alias_mapping is updated
  parent_definitions.import_alias_mapping.update(import_alias_mapping)
  parent_definitions.import_names = local_names
 
@@ -1052,7 +1057,13 @@ def visit_Import(self, node):
  retrieve_import_alias_mapping(node.names)
  )
  for alias in node.names:
- if alias.name not in uninspectable_modules:
+ if alias.name in uninspectable_modules:
+ # The module is uninspectable (so blackbox or built-in). If it has an alias, we remember
+ # the alias so we can do fully qualified name resolution for blackbox- and built-in trigger words
+ # e.g. we want a call to "os.system" be recognised, even if we do "import os as myos"
+ if alias.asname is not None and alias.asname != alias.name:
+ self.bb_or_bi_aliases[alias.asname] = alias.name
+ else:
  log.warn("Cannot inspect module %s", alias.name)
  uninspectable_modules.add(alias.name) # Don't repeatedly warn about this
  return IgnoredNode()
@@ -1094,7 +1105,14 @@ def visit_ImportFrom(self, node):
  retrieve_import_alias_mapping(node.names),
  from_from=True
  )
- if node.module not in uninspectable_modules:
+
+ if node.module in uninspectable_modules:
+ # Remember aliases for blackboxed and built-in imports such that we can label them fully qualified
+ # e.g. we want a call to "os.system" be recognised, even if we do "from os import system"
+ # from os import system as mysystem -> module=os, name=system, asname=mysystem
+ for name in node.names:
+ self.bb_or_bi_aliases[name.asname or name.name] = "{}.{}".format(node.module, name.name)
+ else:
  log.warn("Cannot inspect module %s", node.module)
  uninspectable_modules.add(node.module)
  return IgnoredNode()

pyt/vulnerability_definitions/all_trigger_words.pyt

@@ -31,17 +31,17 @@
  ]
  },
  "execute(": {},
- "system(": {},
+ "os.system(": {},
  "filter(": {},
  "subprocess.call(": {},
+ "subprocess.Popen(": {},
  "render_template(": {},
  "set_cookie(": {},
  "redirect(": {},
  "url_for(": {},
  "flash(": {},
  "jsonify(": {},
  "render(": {},
- "render_to_response(": {},
- "Popen(": {}
+ "render_to_response(": {}
  }
 }

tests/main_test.py

@@ -108,11 +108,11 @@ def test_targets_with_recursive(self):
  excluded_files = ""
 
  included_files = discover_files(targets, excluded_files, True)
- self.assertEqual(len(included_files), 33)
+ self.assertEqual(len(included_files), 34)
 
  def test_targets_with_recursive_and_excluded(self):
  targets = ["examples/vulnerable_code/"]
  excluded_files = "inter_command_injection.py"
 
  included_files = discover_files(targets, excluded_files, True)
- self.assertEqual(len(included_files), 32)
+ self.assertEqual(len(included_files), 33)

tests/vulnerabilities/vulnerabilities_test.py

@@ -482,6 +482,21 @@ def test_list_append_taints_list(self):
  )
  self.assert_length(vulnerabilities, expected_length=1)
 
+ def test_import_bb_or_bi_with_alias(self):
+ self.cfg_create_from_file('examples/vulnerable_code/command_injection_with_aliases.py')
+
+ EXPECTED = ['Entry module',
+ "~call_1 = ret_os.system('ls')",
+ "~call_2 = ret_os.system('ls')",
+ "~call_3 = ret_os.system('ls')",
+ "~call_4 = ret_os.system('ls')",
+ "~call_5 = ret_subprocess.call('ls')",
+ "~call_6 = ret_subprocess.Popen('ls')",
+ 'Exit module'
+ ]
+ for node, expected_label in zip(self.cfg.nodes, EXPECTED):
+ self.assertEqual(node.label, expected_label)
+
 
 class EngineDjangoTest(VulnerabilitiesBaseTestCase):
  def run_analysis(self, path):