Merge pull request #71 from LoRexxar/develop

update to 1.9.4

cobra/__version__.py

@@ -7,7 +7,7 @@
 __issue_page__ = 'https://github.com/LoRexxar/Cobra-W/issues/new'
 __python_version__ = sys.version.split()[0]
 __platform__ = platform.platform()
-__version__ = '1.9.3'
+__version__ = '1.9.4'
 __author__ = 'LoRexxar'
 __author_email__ = 'LoRexxar@gmail.com'
 __license__ = 'MIT License'

cobra/core_engine/php/parser.py

@@ -630,13 +630,12 @@ def parameters_back(param, nodes, function_params=None, lineno=0,
         node = nodes[-1]
 
         # 加入扫描范围check, 如果当前行数大于目标行数，直接跳过(等于会不会有问题呢？)
-        if node.lineno >= lineno:
+        if node.lineno >= int(lineno):
             return parameters_back(param, nodes[:-1], function_params, lineno,
                                      function_flag=0, vul_function=vul_function,
                                      file_path=file_path,
                                      isback=isback, parent_node=0)
 
-
         if isinstance(node, php.Assignment) and param_name == get_node_name(node.node):  # 回溯的过程中，对出现赋值情况的节点进行跟踪
             param_node = get_node_name(node.node)  # param_node为被赋值的变量
             param_expr, expr_lineno, is_re = get_expr_name(node.expr)  # param_expr为赋值表达式,param_expr为变量或者列表
@@ -862,91 +861,58 @@ def parameters_back(param, nodes, function_params=None, lineno=0,
 
             if isinstance(node.node, php.Block):  # if里可能是代码块，也可能就一句语句
                 if_nodes = node.node.nodes
-                if_node_lineno = node.node.lineno
             elif node.node is not None:
                 if_nodes = [node.node]
-                if_node_lineno = node.node.lineno
             else:
                 if_nodes = []
-                if_node_lineno = 0
 
             # 进入分析if内的代码块，如果返回参数不同于进入参数，那么在不同的代码块中，变量值不同，不能统一处理，需要递归进入不同的部分
             is_co, cp, expr_lineno = parameters_back(param, if_nodes, function_params, lineno,
                                                      function_flag=function_flag, vul_function=vul_function,
                                                      file_path=file_path, isback=isback, parent_node=node)
 
-            if is_co == 3 and cp != param:  # 理由如上
-                is_co, cp, expr_lineno = parameters_back(cp, nodes[:-1], function_params, lineno,
-                                                         function_flag=function_flag, vul_function=vul_function,
-                                                         file_path=file_path, isback=isback,
-                                                         parent_node=parent_node)  # 找到可控的输入时，停止递归
-                # return is_co, cp, expr_lineno
-
-            if is_co is not 1 and node.elseifs != [] and node.elseifs.lineno < lineno:  # elseif可能有多个，所以需要列表
+            if is_co is not 1 and node.elseifs != []:  # elseif可能有多个，所以需要列表
 
                 logger.debug("[AST] param {} line {} in new branch for else if".format(param, node.lineno))
 
                 for node_elseifs_node in node.elseifs:
                     if isinstance(node_elseifs_node.node, php.Block):
                         elif_nodes = node_elseifs_node.node.nodes
-                        elif_node_lineno = node_elseifs_node.node.lineno
                     elif node_elseifs_node.node is not None:
                         elif_nodes = [node_elseifs_node.node]
-                        elif_node_lineno = node_elseifs_node.node.lineno
                     else:
                         elif_nodes = []
-                        elif_node_lineno = 0
 
                     is_co, cp, expr_lineno = parameters_back(param, elif_nodes, function_params, lineno,
                                                              function_flag=function_flag, vul_function=vul_function,
                                                              file_path=file_path,
                                                              isback=isback, parent_node=node)
 
-                    if is_co == 3 and cp != param:  # 理由如上
-                        is_co, cp, expr_lineno = parameters_back(cp, nodes[:-1], function_params, lineno,
-                                                                 function_flag=function_flag, vul_function=vul_function,
-                                                                 file_path=file_path,
-                                                                 isback=isback, parent_node=parent_node)  # 找到可控的输入时，停止递归
-                        # return is_co, cp, expr_lineno
-                    else:
-                        break
-
-            if is_co is not 1 and node.else_ != [] and node.else_ is not None and node.else_.lineno < lineno:
+            if is_co is not 1 and node.else_ != [] and node.else_ is not None:
 
                 logger.debug("[AST] param {} line {} in new branch for else".format(param, node.else_.lineno))
 
                 if isinstance(node.else_.node, php.Block):
                     else_nodes = node.else_.node.nodes
-                    else_node_lineno = node.else_.node.lineno
                 elif node.else_.node is not None:
                     else_nodes = [node.else_.node]
-                    else_node_lineno = node.else_.node.lineno
                 else:
                     else_nodes = []
-                    else_node_lineno = 0
 
                 is_co, cp, expr_lineno = parameters_back(param, else_nodes, function_params, lineno,
                                                          function_flag=function_flag, vul_function=vul_function,
                                                          file_path=file_path, isback=isback, parent_node=node)
 
-                if is_co == 3 and cp != param:  # 理由如上
-                    is_co, cp, expr_lineno = parameters_back(cp, nodes[:-1], function_params, lineno,
-                                                             function_flag=function_flag, vul_function=vul_function,
-                                                             file_path=file_path,
-                                                             isback=isback, parent_node=parent_node)  # 找到可控的输入时，停止递归
-                    # return is_co, cp, expr_lineno
+                if is_co in [-1, 1, 2]:  # 目标确定直接返回
+                    return is_co, cp, expr_lineno
 
             # 添加一种情况，如果只有if没有else，则会考虑不进入循环的特殊情况
             if is_co is not 1 and (node.else_ == [] or node.else_ is None):
 
                 logger.debug("[AST] param {} line {} in new branch skip if/else".format(param, node.lineno))
 
-                is_co, cp, expr_lineno = parameters_back(param, nodes[:-1], function_params, lineno,
-                                                         function_flag=function_flag, vul_function=vul_function,
-                                                         file_path=file_path,
-                                                         isback=isback, parent_node=parent_node)  # 找到可控的输入时，停止递归
-
-            # return is_co, cp, expr_lineno
+            if is_co in [-1, 1, 2]:  # 目标确定直接返回
+                return is_co, cp, expr_lineno
 
         elif isinstance(node, php.For):
             for_nodes = node.node.nodes
@@ -960,6 +926,9 @@ def parameters_back(param, nodes, function_params=None, lineno=0,
                                                      isback=isback, parent_node=node)
             function_flag = 0
 
+            if is_co in [-1, 1, 2]:  # 目标确定直接返回
+                return is_co, cp, expr_lineno
+
         if is_co == 3 or int(lineno) == node.lineno:  # 当is_co为True时找到可控，停止递归
             is_co, cp, expr_lineno = parameters_back(param, nodes[:-1], function_params, lineno,
                                                      function_flag=function_flag, vul_function=vul_function,

cobra/engine.py

@@ -147,9 +147,9 @@ def score2level(score):
         return '{l}-{s}: {ast}'.format(l=level[:1], s=score_full, ast=a)
 
 
-def scan_single(target_directory, single_rule, files=None, language=None, secret_name=None, is_unconfirm=False):
+def scan_single(target_directory, single_rule, files=None, language=None, secret_name=None, is_unconfirm=False, newcore_function_list=[]):
     try:
-        return SingleRule(target_directory, single_rule, files, language, secret_name, is_unconfirm).process()
+        return SingleRule(target_directory, single_rule, files, language, secret_name, is_unconfirm, newcore_function_list).process()
     except Exception:
         raise
 
@@ -160,6 +160,7 @@ def scan(target_directory, a_sid=None, s_sid=None, special_rules=None, language=
     vulnerabilities = r.vulnerabilities
     rules = r.rules(special_rules)
     find_vulnerabilities = []
+    newcore_function_list = {}
 
     def store(result):
         if result is not None and isinstance(result, list) is True:
@@ -171,7 +172,7 @@ def store(result):
 
     async def start_scan(target_directory, rule, files, language, secret_name):
 
-        result = scan_single(target_directory, rule, files, language, secret_name, is_unconfirm)
+        result = scan_single(target_directory, rule, files, language, secret_name, is_unconfirm, newcore_function_list)
         store(result)
 
     if len(rules) == 0:
@@ -264,6 +265,19 @@ async def start_scan(target_directory, rule, files, language, secret_name):
             logger.info(
                 '[SCAN] Not Trigger Rules ({l}): {r}'.format(l=len(diff_rules), r=','.join(diff_rules)))
 
+    # show detail about newcore function list
+    table2 = PrettyTable(
+        ['#', 'NewFunction', 'Related Rules id'])
+
+    table2.align = 'l'
+    idy = 0
+    for new_function_name in newcore_function_list:
+        table2.add_row([idy + 1, new_function_name, newcore_function_list[new_function_name]])
+        idy += 1
+
+    if len(newcore_function_list) > 0:
+        logger.info("[SCAN] New evil Function list by NewCore:\r\n{}".format(table2))
+
     # completed running data
     if s_sid is not None:
         Running(s_sid).data({
@@ -284,7 +298,7 @@ async def start_scan(target_directory, rule, files, language, secret_name):
 
 
 class SingleRule(object):
-    def __init__(self, target_directory, single_rule, files, language=None, secret_name=None, is_unconfirm=False):
+    def __init__(self, target_directory, single_rule, files, language=None, secret_name=None, is_unconfirm=False, newcore_function_list=[]):
         self.target_directory = target_directory
         self.find = Tool().find
         self.grep = Tool().grep
@@ -302,6 +316,9 @@ def __init__(self, target_directory, single_rule, files, language=None, secret_n
         """
         self.rule_vulnerabilities = []
 
+        # new core function list
+        self.newcore_function_list = newcore_function_list
+
         logger.info("[!] Start scan [CVI-{sr_id}]".format(sr_id=self.sr.svid))
 
     def origin_results(self):
@@ -492,10 +509,12 @@ def process(self):
                     self.rule_vulnerabilities.append(vulnerability)
                 else:
                     if reason == 'New Core':  # 新的规则
-                        logger.debug('[CVI-{cvi}] [NEW-VUL] New Rules init')
+
+                        logger.debug('[CVI-{cvi}] [NEW-VUL] New Rules init'.format(cvi=self.sr.svid))
                         new_rule_vulnerabilities = NewCore(self.sr, self.target_directory, data, self.files, 0,
                                                            languages=self.languages, secret_name=self.secret_name,
-                                                           is_unconfirm=self.is_unconfirm)
+                                                           is_unconfirm=self.is_unconfirm,
+                                                           newcore_function_list=self.newcore_function_list)
 
                         if len(new_rule_vulnerabilities) > 0:
                             self.rule_vulnerabilities.extend(new_rule_vulnerabilities)
@@ -1044,7 +1063,7 @@ def auto_parse_match(single_match, svid, language):
     return mr
 
 
-def NewCore(old_single_rule, target_directory, new_rules, files, count=0, languages=None, secret_name=None, is_unconfirm=False):
+def NewCore(old_single_rule, target_directory, new_rules, files, count=0, languages=None, secret_name=None, is_unconfirm=False, newcore_function_list=[]):
     """
     处理新的规则生成
     :param languages: 
@@ -1084,6 +1103,17 @@ def NewCore(old_single_rule, target_directory, new_rules, files, count=0, langua
     sr.svid = svid
     sr.language = language
 
+    # check vul rule exist
+    if vul_function in newcore_function_list:
+        logger.debug('[CVI-{cvi}] [NEW-VUL] New Rules {macth} exist.'.format(cvi=svid, macth=vul_function))
+
+        if svid not in newcore_function_list[vul_function]:
+            newcore_function_list[vul_function].append(svid)
+
+        return []
+    else:
+        newcore_function_list[vul_function] = [svid]
+
     # grep
 
     try:
@@ -1148,7 +1178,8 @@ def NewCore(old_single_rule, target_directory, new_rules, files, count=0, langua
                 if reason == 'New Core':  # 新的规则
                     logger.debug('[CVI-{cvi}] [NEW-VUL] New Rules init')
                     new_rule_vulnerabilities = NewCore(sr, target_directory, data, files, count,
-                                                       secret_name=secret_name, is_unconfirm=is_unconfirm)
+                                                       secret_name=secret_name, is_unconfirm=is_unconfirm,
+                                                       newcore_function_list=newcore_function_list)
 
                     if not new_rule_vulnerabilities:
                         return rule_vulnerabilities

cobra/pretreatment.py

@@ -168,23 +168,32 @@ async def pre_ast(self):
                         self.pre_result[filepath]['ast_nodes'] = all_nodes
 
                     except SyntaxError as e:
-                        logger.warning('[AST] [ERROR] parser {}: {}'.format(filepath, traceback.format_exc()))
+                        logger.warning('[AST] [ERROR] parser {} SyntaxError'.format(filepath))
+                        continue
 
                     except AssertionError as e:
                         logger.warning('[AST] [ERROR] parser {}: {}'.format(filepath, traceback.format_exc()))
+                        continue
 
                     except:
                         logger.warning('[AST] something error, {}'.format(traceback.format_exc()))
+                        continue
 
                     # 搜索所有的常量
                     for node in all_nodes:
                         if isinstance(node, php.FunctionCall) and node.name == "define":
                             define_params = node.params
-                            logger.debug(
-                                "[AST][Pretreatment] new define {}={}".format(define_params[0].node,
-                                                                              define_params[1].node))
 
-                            self.define_dict[define_params[0].node] = define_params[1].node
+                            if define_params:
+                                logger.debug(
+                                    "[AST][Pretreatment] new define {}={}".format(define_params[0].node,
+                                                                                  define_params[1].node))
+
+                                key = define_params[0].node
+                                if isinstance(key, php.Constant):
+                                    key = key.name
+
+                                self.define_dict[key] = define_params[1].node
 
             elif fileext[0] in ext_dict['chromeext'] and 'chromeext' in self.lan:
 

docs/changelog.md

@@ -162,4 +162,8 @@
     - 修复了因为每5行扫描而导致的无法定位参数问题 #68
     - 修复了在if/else递归逻辑中错误的逻辑关
     - 我必须要承认，由于对php结构的不熟悉，对于大多出后加的语法接口，基本都是出现问题之后才写的，相互割裂太严重，导致引入了过多的递归跳出逻辑，严重影响到了对作用域的控制
-
+- 2020-04-26
+    - Cobra-W 1.9.4
+    - 修复了全局变量扫描的时候遇到的语法错误 #69
+    - 修复了if/else中不正确的进出逻辑
+    - 添加了新恶意函数列表的展示，并优化其逻辑