import subprocess import time import os import sys import json import copy sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))) import importlib from contextlib import redirect_stdout from support.packet_generator.workpath import workdir from support.ui_utils.element_position.map_element_position_library import * import re import ast from pprint import pprint from pprint import PrettyPrinter def local_run(): """本地执行所有用例""" print("Get-Content result.txt -Encoding UTF8 -Wait") parameter = { "username": "zcw3", "password": "qa111111", "test_pc_ip": "192.168.64.93", "test_subcriber_id": "test6491", "api_server": "http://192.168.44.72", "debug_flag": "local", "script_type": "ui", "is_log": 1, "env": "tsgx", "vsys_id": 1, "root_path": workdir, "path": workdir + "/testcase/ui", "module_name": "manipulation", "test_case_name": os.path.basename(__file__)[:-3] } parameter = replace_paras(parameter) cur_path = os.getcwd() print(os.getcwd()) all_list = os.listdir() man_list = [] # 要执行的测试文件名称 tmp_no = ["pre_reqheader", "xly_reqheader", "suff_reqheader", "pre_resheader", "xly_resheader", "pre_resheader", "doh"] for i in all_list: if "man_" in i: i_1 = i[:-3] # 临时过滤掉不执行header的其它匹配 flag_1 = 0 for j in range(len(tmp_no)): if tmp_no[j] in i_1: flag_1 = 1 break if flag_1 == 0: man_list.append(i_1) n = 1 sum_all = len(man_list) # 打开文件准备写入 with open("C:\\zcw\\tsg_test\\tests\\manipulation\\result.txt", "w", encoding="utf-8") as f: # 使用redirect_stdout将print的输出重定向到文件 with redirect_stdout(f): for mani in man_list: print("执行第{}个用例，还剩{}个用例未执行".format(n, sum_all - n)) mo = importlib.import_module(mani) parameter["test_case_name"] = mani rl = mo.run(parameter) n += 1 # 手动刷新缓冲区，确保所有输出写入文件 f.flush() print("Done writing to result.txt") def diff_ms(ms_json_path='C:/Users/root/Downloads/Metersphere_Api_TSG (9).json', case_directory_path='C:/zcw/tsg_test/tests/ui/manipulation', file_startwith = "man_", file_endswith = ".py"): """ 比较MS和自动化脚本不同内容 :param ms_json_path: MS导出的json文件路径 :param case_directory_path: 对比用例case所在目录的路径 :param file_startwith: 用例文件开始字符，例如 man_xxx_.py 选择 man_ :param file_endswith: 用例文件结束字符，例如 man_xxx_.py 选择 .py """ # 存储文件名（去掉.py后缀） file_names = [] # 获取目录下所有文件的名称 for filename in os.listdir(case_directory_path): if filename.endswith(file_endswith) and filename.startswith(file_startwith): # 去掉.py后缀并添加到列表中 name_without_extension = filename[:-3] file_names.append(name_without_extension) # 读取MS json文件内容 with open(ms_json_path, 'r', encoding="utf-8") as f: tmp_d = json.load(f) #print(tmp_d) ms_names = [] # MS case名称和请求体用例名称字段的列表 [{'ms_name': 'man_mon', 'test_case_name': 'man_mon'}] ms_case_names = [] # 只有MS case名称 ["man_mon"] # 从MS中读取到name for i in range(len(tmp_d["cases"])): ms_one_dict = {} #print(tmp_d["cases"][i]["name"]) #print(tmp_d["cases"][i]["request"]) request_dict = json.loads(tmp_d["cases"][i]["request"]) body_dict = json.loads(request_dict["body"]["raw"]) #print(request_dict) #print(body_dict) #print(body_dict["test_case_name"]) ms_one_dict["ms_name"] = tmp_d["cases"][i]["name"] ms_one_dict["test_case_name"] = body_dict["test_case_name"] ms_names.append(ms_one_dict) ms_case_names.append(tmp_d["cases"][i]["name"]) #print(ms_one_dict) # print(file_names) # print(ms_names) print("自动化脚本个数：{}".format(len(file_names))) print("MS统计脚本个数：{}".format(len(ms_names))) # 排查MS中ms_name、test_case_name名称不同选项 diff_name = [] for i in range(len(ms_names)): if ms_names[i]["ms_name"] != ms_names[i]["test_case_name"]: diff_name.append(ms_names[i]) print("MS的用例名称和请求体中test_case_name不一致共有 {} 个：包括：{}".format(len(diff_name), diff_name)) # 排查自动化脚本比MS多的用例 print("检查自动化脚本和MS脚本不同用例：") file_names_set = set(file_names) ms_case_names_set = set(ms_case_names) only_in_file_names_set = file_names_set.difference(ms_case_names_set) only_in_ms_case_names_set = ms_case_names_set.difference(file_names_set) print("自动化脚本比MS多 {} 个用例：包括：{}".format(len(only_in_file_names_set), list(only_in_file_names_set))) print("MS比自动化脚本多 {} 个用例：包括：{}".format(len(only_in_ms_case_names_set), list(only_in_ms_case_names_set))) def get_excel_formate_from_case(case_directory_path='C:/zcw/tsg_test/tests/ui/manipulation', file_startwith = "man_", file_endswith = ".py"): # 使用excel统计测试用例， # 存储文件名（去掉.py后缀） file_names = [] # 获取目录下所有文件的名称 for filename in os.listdir(case_directory_path): if filename.endswith(file_endswith) and filename.startswith(file_startwith): # 去掉.py后缀并添加到列表中 name_without_extension = filename[:-3] file_names.append(name_without_extension) a = [["aa", 1, 1, 0, 1], ["aa", 1, 1, 0, 1]] excel_list = [] # 表头参数 first_table_header = ['action', 'srcip', 'user', 'srcport', 'dstip', 'fqdn', 'dstport', 'intip', 'intport', 'extip', 'extport', 'app', 'url', 'reqheader', 'reqbody', 'resheader', 'resbody', 'logall', 'case'] excel_list.append(first_table_header) for i in range(len(file_names)): raw = file_names[i] tmp_raw_list = [] raw_dict = { "action_col" : 0, "srcip_col" : 0, "user_col" : 0, "srcport_col" : 0, "dstip_col" : 0, "fqdn_col" : 0, "dstport_col" : 0, "intip_col" : 0, "intport_col" : 0, "extip_col" : 0, "extport_col" : 0, "app_col" : 0, "url_col" : 0, "reqheader_col" : 0, "reqbody_col" : 0, "resheader_col" : 0, "resbody_col" : 0, "logall_col" : 0, "case_col" : 0 } # 列统计 if "allow" in raw: raw_dict["action_col"] = "allow" elif "deny" in raw: raw_dict["action_col"] = "deny" elif "monitor" in raw: raw_dict["action_col"] = "monitor" elif "redirect" in raw: raw_dict["action_col"] = "redirect" elif "replace" in raw: raw_dict["action_col"] = "replace" elif "hijack" in raw: raw_dict["action_col"] = "hijack" elif "insert" in raw: raw_dict["action_col"] = "insert" elif "edit_element" in raw: raw_dict["action_col"] = "edit_element" else: raw_dict["action_col"] = "run_script" # 列统计 if "srcip" in raw: raw_dict["srcip_col"] = 1 if "user" in raw: raw_dict["user_col"] = 1 if "srcport" in raw: raw_dict["srcport_col"] = 1 if "dstip" in raw: raw_dict["dstip_col"] = 1 if "fqdn" in raw: raw_dict["fqdn_col"] = 1 if "dstport" in raw: raw_dict["dstport_col"] = 1 if "intip" in raw: raw_dict["intip_col"] = 1 if "intport" in raw: raw_dict["intport_col"] = 1 if "extip" in raw: raw_dict["extip_col"] = 1 if "extport" in raw: raw_dict["extport_col"] = 1 # application if "http" in raw: raw_dict["app_col"] = "http" else: raw_dict["app_col"] = "doh" if "url" in raw: raw_dict["url_col"] = 1 if "reqheader" in raw: raw_dict["reqheader_col"] = 1 if "reqbody" in raw: raw_dict["reqbody_col"] = 1 if "resheader" in raw: raw_dict["resheader_col"] = 1 if "resbody" in raw: raw_dict["resbody_col"] = 1 if "logall" in raw: raw_dict["logall_col"] = 1 raw_dict["case_col"] = raw # 组成一行数据 for j in range(len(first_table_header)): col_name = "{}_col".format(first_table_header[j]) tmp_raw_list.append(raw_dict[col_name]) # 组成excel数据 excel_list.append(copy.deepcopy(tmp_raw_list)) #print(file_names) print(excel_list) # 将excel list数据写入text，包含有制表符 # 写入文件 file_path = "test_excel_info.txt" # 确定每一列的宽度 # column_widths = [max(len(str(item)) for item in column) for column in zip(*excel_list)] # # 格式化字符串，每列保持对齐 # with open(file_path, "w") as file: # for row in excel_list: # aligned_row = [f"{item:<{width}}" for item, width in zip(row, column_widths)] # file.write("\t".join(aligned_row) + "\n") # 简单模式，没有对其显示 with open(file_path, "w") as f: for row in excel_list: f.write("\t".join(map(str, row)) + "\n") test_excel_info_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), file_path) print("完成。原数据保存路径：{}".format(test_excel_info_path)) def replace_test_data_from_py(case_directory_path='C:/zcw/tsg_test/tests/ui/目录名', file_startwith = "man_", file_endswith = ".py"): # 替换源文件中test_data内容 #原始数据格式内容，替换其中的 “ replace_text” resource_text = \ """# -*- coding: UTF-8 -*- import time import os import sys sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))) import traceback from datetime import datetime from support.instance_utils.test_data_instance import TestDataInstance from support.report_update import ReportUpdate def run(parameter): try: print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Begin to run test case: " + parameter["test_case_name"], flush=True) # 参数初始化 exception_result = "" result = {} # 脚本启动时间 script_start_time = time.time() # 测试数据 replace_text # 测试用例实例化 create = CreatePolicy(test_data, parameter) result = create.create_policy() # 脚本结束时间和耗时 script_end_time = time.time() duration = script_end_time - script_start_time print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Duration of running the test case: ", "{:.3f}".format(duration), flush=True) print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Finish test case: " + parameter["test_case_name"], flush=True) return result except Exception as e: exception_result = str(e) print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Error: ", e, flush=True) traceback.print_exc() return "Error: " + str(e) finally: # 清理环境 if isinstance(create, CreatePolicy): create.clean_up() # 生成csv报告 update = ReportUpdate() update.write_result(parameter, result, exception_result) if __name__ == '__main__': from support.ui_utils.element_position.map_element_position_library import replace_paras from support.ui_utils.workpath import workdir parameter = { "username": "zcw3", "password": "qa111111", "test_pc_ip": "192.168.64.93", "test_subcriber_id": "test6491", "api_server": "http://192.168.44.72", "debug_flag": "local", "script_type": "api", # api ui 空字符串 "is_log": 1, "env": "tsgx", "vsys_id": 1, "root_path": workdir, "path": workdir + "/testcase/ui", "module_name": "manipulation", "test_case_name": os.path.basename(__file__)[:-3] } parameter = replace_paras(parameter) run(parameter) """ #print(resource_text) file_names = [] # 获取目录下所有文件的名称 for filename in os.listdir(case_directory_path): if filename.endswith(file_endswith) and filename.startswith(file_startwith): # 去掉.py后缀并添加到列表中 file_names.append(filename) for i in range(len(file_names)): file_0 = file_names[i] print(file_0) tmp_resource_text = copy.deepcopy(resource_text) with open(file_0, "r", encoding="utf-8") as f: text = f.read() # 正则表达式匹配从# 测试数据到log_in = LogIn()之间的内容 pattern = re.compile(r'测试数据(.*?)log_in = LogIn\(\)', re.DOTALL) matches = re.findall(pattern, text) if len(matches) == 0: print("{}，替换失败".format(file_0)) continue match_text = matches[0] match_text = match_text.rstrip() match_text = match_text.lstrip("\n") # 输出匹配的内容 #print(match_text) result_text = tmp_resource_text.replace(" replace_text", match_text) # 覆盖源文件内容 谨慎执行 # with open(file_0, "w", encoding="utf-8") as f2: # f2.write(result_text) # 覆盖源文件内容 谨慎执行 print("{}，替换完成".format(file_0)) #break print("替换完成") def replace_test_data_from_py_2(case_directory_path='C:/zcw/tsg_test/tests/目录名', name = "name"): file_names = [] # 获取目录下所有文件的名称 faile_path = os.path.join(case_directory_path, name) print(faile_path) resource_text = "" parameter = [] parameter = {"test_pc_ip":'2.2.2.2', "test_case_name":parameter["test_case_name"]} tmp_resource_text = copy.deepcopy(resource_text) with open(faile_path, "r", encoding="utf-8") as f: text = f.read() #print(text) # 正则表达式匹配从# 测试数据到log_in = LogIn()之间的内容 pattern = re.compile(r'测试数据(.*?)# 测试用例实例化', re.DOTALL) matches = re.findall(pattern, text) match_text = matches[0] match_text = match_text.rstrip() match_text = match_text.lstrip() match_text = match_text.lstrip("test_data =") match_text = match_text.lstrip() match_text = match_text.replace('parameter["test_case_name"]', '"tmp_name"') match_text = match_text.replace('parameter["test_pc_ip"]', 'parameter["test_pc_ip"]') match_text = match_text.replace('"$" + parameter["test_subcriber_id"]', '"$test_tmp_subid"') # print(match_text) # print(type(match_text)) match_dict = ast.literal_eval(match_text) print(match_dict) object_ip = { "negate_option": False, "or_conditions": [ { "attribute_name": "ATTR_SOURCE_IP", "name": "manipulation_source_ip", "type": "ip", "sub_type": "ip", "statistics_option": "none", "member_type": "item", "items": [ { "op": "add", "ip": parameter["test_pc_ip"], "interval": "0-65535" } ] } ] } object_port = { "negate_option": False, "or_conditions": [ { "attribute_name": "ATTR_INTERNAL_PORT", "name": "manipulation_internal_port", "type": "port", "statistics_option": "none", "member_type": "item", "items": [ { "op": "add", "interval": "443" } ] } ] } object_fqdn = { "negate_option": False, "or_conditions": [ { "attribute_name": "ATTR_SERVER_FQDN", "name": "manipulation_fqdn", "type": "fqdn", "statistics_option": "none", "member_type": "item", "items": [ { "op": "add", "expr_type": "and", "expression": "baidu.com" } ] } ] } object_subid = { "negate_option": False, "or_conditions": [ { "attribute_name": "ATTR_SUBSCRIBER_ID", "name": "manipulation_subid", "type": "subscriberid", "statistics_option": "none", "member_type": "item", "items": [ { "op": "add", "expr_type": "and", "expression": "baidu.com" } ] } ] } object_url = { "negate_option": False, "or_conditions": [ { "attribute_name": "ATTR_HTTP_URL", "name": "manipulation_url", "type": "url", "statistics_option": "none", "member_type": "item", "items": [ { "op": "add", "expr_type": "and", "expression": "baidu" } ] } ] } object_keyword = { "negate_option": False, "or_conditions": [ { "attribute_name": "ATTR_HTTP_REQ_HDR", "name": "manipulation_http_reqheader", "type": "keyword", "statistics_option": "none", "member_type": "item", "items": [ { "op": "add", "expr_type": "and", "expression": "Chrome" } ] } ] } object_app = { "negate_option": False, "or_conditions": [ { "attribute_name": "ATTR_APP_ID", "type": "application", "items": ["http"] } ] } profile_response = { "sub_action": "block" } profile_element = { "sub_action": "edit_element", "rules": [ # { # "anchor_element": { # "contained_keyword": "tessa1", # "search_scope": "whole_file", # #"start_indicator": "testsss" # }, # "target_element": { # "element_treatment": "mark", # mark remove # "target_distance_from_matching": 2 # } # } ] } profile_replace_file = { "sub_action": "replace_file", "replacement_file": { "name": "test", "file_path": "hijack_png_1.png", "content_name": "hijack_png_1.png", "content_type": "application / vnd.android.package-archive" } } profile_insert_css = { "sub_action": "inject_css", "css_file": { "name": "automanipulation_insert_profile", "file_path": "insert_css_1.css" } } profile_insert_js = { "sub_action": "inject_javascript", "injection_section": "html_body", # html_head "js_file": { "name": "manipulation_insert_profile", "file_path": "insert_js_1.js" } } profile_redirect = { "code": 302, "to": "http://sss.ss.dd.sss" } profile_replace_text = { "sub_action": "replace_text", "rules": [ # { # "regex_enable": 0, # "search_in": "http_req_uri", # "find": "sdsd", # "replace_with": "ggss" # } ] } profile_lua = { "lua_script": { "name": "test", "file_path": "run_script_replace_reqbody.lua", "max_exec_time": 100 } } policy_configuration = { "name": parameter["test_case_name"], "type": "proxy_manipulation", "action": "allow", "and_conditions": [ ], "is_enabled": 1, "log_option": "metadata", } traffic_generation = { "tool": "ssl", # or trex/http "command": "curl --compressed -kv --connect-timeout 30 --max-time 60 --user-agent \"Wget (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.70 Safari/537.36\" https://www.baidu.com" } verification_result = { "excepted_traffic_result": "关于百度,CN=TSG", "expected_metric": {"hits": 1}, "expected_log": [ ] } print() for k,v in match_dict["condition"].items(): #print(k, "====", v) map_attr = { "source_ip": "ATTR_SOURCE_IP", "destination_ip": "ATTR_DESTINATION_IP", "internal_ip": "ATTR_INTERNAL_IP", "external_ip": "ATTR_EXTERNAL_IP", "source_port": "ATTR_SOURCE_PORT", "destination_port": "ATTR_DESTINATION_PORT", "external_port": "ATTR_EXTERNAL_PORT", "internal_port": "ATTR_INTERNAL_PORT", "sub_id": "ATTR_SUBSCRIBER_ID", "server_fqdn": "ATTR_SERVER_FQDN", "url": "ATTR_HTTP_URL", "request_header": "ATTR_HTTP_REQ_HDR", "response_header": "ATTR_HTTP_RES_HDR", "request_body": "ATTR_HTTP_REQ_BODY", "response_body": "ATTR_HTTP_RES_BODY", } map_name = { "source_ip": "manipulation_source_ip", "destination_ip": "manipulation_destination_ip", "internal_ip": "manipulation_internal_ip", "external_ip": "manipulation_external_ip", "source_port": "manipulation_src_port", "destination_port": "manipulation_destination_port", "external_port": "manipulation_external_port", "internal_port": "manipulation_internal_port", "sub_id": "manipulation_subid", "server_fqdn": "manipulation_fqdn", "url": "manipulation_url", "request_header": "manipulation_http_reqheader", "response_header": "manipulation_http_resheader", "request_body": "manipulation_keywords_request_body", "response_body": "manipulation_keywords_response_body", } if k in ["source_ip", "internal_ip", "destination_ip", "external_ip"]: manipulation_source_ip = copy.deepcopy(object_ip) manipulation_source_ip["or_conditions"][0]["attribute_name"] = map_attr[k] manipulation_source_ip["or_conditions"][0]["name"] = map_name[k] manipulation_source_ip["or_conditions"][0]["items"][0]["ip"] = v[0]["item"][0]["item_value"] if len(v[0]["item"]) > 1: print("......................含有两个item.......................手动添加") policy_configuration["and_conditions"].append(manipulation_source_ip) elif k in ["internal_port", "source_port", "destination_port", "external_port"]: manipulation_object_port = copy.deepcopy(object_port) manipulation_object_port["or_conditions"][0]["attribute_name"] = map_attr[k] manipulation_object_port["or_conditions"][0]["name"] = map_name[k] manipulation_object_port["or_conditions"][0]["items"][0]["interval"] = v[0]["item"][0]["item_value"] if len(v[0]["item"]) > 1: print("......................含有两个item.......................手动添加") policy_configuration["and_conditions"].append(manipulation_object_port) elif k in ["server_fqdn"]: manipulation_object_fqdn = copy.deepcopy(object_fqdn) manipulation_object_fqdn["or_conditions"][0]["attribute_name"] = map_attr[k] manipulation_object_fqdn["or_conditions"][0]["name"] = map_name[k] manipulation_object_fqdn["or_conditions"][0]["items"][0]["expression"] = v[0]["items"][0]["item_value"] if len(v[0]["items"]) > 1: print("......................含有两个item.......................手动添加") # 替换匹配符 tmp_exp = manipulation_object_fqdn["or_conditions"][0]["items"][0]["expression"] if tmp_exp.startswith("*"): # *ssdsds tmp_exp_1 = "{}$".format(tmp_exp[1:]) elif tmp_exp.endswith("*"): # ssdsds* tmp_exp_1 = "^{}".format(tmp_exp[:-1]) elif tmp_exp.startswith("$"): tmp_exp_1 = "^{}$".format(tmp_exp[1:]) else: tmp_exp_1 = tmp_exp manipulation_object_fqdn["or_conditions"][0]["items"][0]["expression"] = tmp_exp_1 policy_configuration["and_conditions"].append(manipulation_object_fqdn) elif k in ["sub_id"]: manipulation_object_subid = copy.deepcopy(object_subid) manipulation_object_subid["or_conditions"][0]["attribute_name"] = map_attr[k] manipulation_object_subid["or_conditions"][0]["name"] = map_name[k] manipulation_object_subid["or_conditions"][0]["items"][0]["expression"] = v[0]["items"][0]["item_value"] if len(v[0]["items"]) > 1: print("......................含有两个item.......................手动添加") # 替换匹配符 tmp_exp = manipulation_object_subid["or_conditions"][0]["items"][0]["expression"] if tmp_exp.startswith("*"): # *ssdsds tmp_exp_1 = "{}$".format(tmp_exp[1:]) elif tmp_exp.endswith("*"): # ssdsds* tmp_exp_1 = "^{}".format(tmp_exp[:-1]) elif tmp_exp.startswith("$"): tmp_exp_1 = "^{}$".format(tmp_exp[1:]) else: tmp_exp_1 = "^{}$".format(tmp_exp) manipulation_object_subid["or_conditions"][0]["items"][0]["expression"] = tmp_exp_1 policy_configuration["and_conditions"].append(manipulation_object_subid) elif k == "application": application = copy.deepcopy(object_app) policy_configuration["and_conditions"].append(application) elif k == "protocol_filed": for j in range(len(v)): if "item" in v[j]: item = "item" else: item = "items" for i in range(len(v[j][item])): item_type = v[j][item][i]["item_type"] if item_type in ["request_header", "response_header", "request_body", "response_body"]: if "item_key" in v[j][item][i]: item_key = v[j][item][i]["item_key"] else: item_key = v[j][item][i]["item_type"] item_value = v[j][item][i]["item_value"] http_header = copy.deepcopy(object_keyword) if item_key in ["User-Agent", "Content-Type", "Set-Cookie", "Cookie", "response_body", "request_body"]: http_header["or_conditions"][0]["attribute_name"] = map_attr[item_type] http_header["or_conditions"][0]["name"] = map_name[item_type] if item_value == "HEX": http_header["or_conditions"][0]["items"][0]["expression"] = "|{}|".format(v[j][item][i]["value"][0]) elif item_value == "REGEX": http_header["or_conditions"][0]["items"][0]["expr_type"] = "regex" http_header["or_conditions"][0]["items"][0]["expression"] = v[j][item][i]["value"][0] elif "range" in v[j]["item"][i] and len(v[0]["item"][i]["range"])>0: offset = v[j][item][i]["range"][0]["offset"] depth = v[j][item][i]["range"][0]["depth"] tmp_0 =v[j][item][i]["value"][0] tmp_item = "(offset={},depth={}){}".format(offset, depth, tmp_0) http_header["or_conditions"][0]["items"][0]["expression"] = tmp_item else: http_header["or_conditions"][0]["items"][0]["expression"] = v[j][item][i]["value"][0] # 替换匹配符 tmp_exp = http_header["or_conditions"][0]["items"][0]["expression"] if tmp_exp.startswith("*"): # *ssdsds tmp_exp_1 = "{}$".format(tmp_exp[1:]) elif tmp_exp.endswith("*"): # ssdsds* tmp_exp_1 = "^{}".format(tmp_exp[:-1]) elif tmp_exp.startswith("$"): tmp_exp_1 = "^{}$".format(tmp_exp[1:]) else: tmp_exp_1 = tmp_exp http_header["or_conditions"][0]["items"][0]["expression"] = tmp_exp_1 policy_configuration["and_conditions"].append(http_header) else: raise elif item_type in ["url"]: url = copy.deepcopy(object_url) url["or_conditions"][0]["attribute_name"] = map_attr[item_type] url["or_conditions"][0]["name"] = map_name[item_type] url["or_conditions"][0]["items"][0]["expression"] = v[j][item][i]["item_value"] # 替换匹配符 tmp_exp = url["or_conditions"][0]["items"][0]["expression"] if tmp_exp.startswith("*"): # *ssdsds tmp_exp_1 = "{}$".format(tmp_exp[1:]) elif tmp_exp.endswith("*"): # ssdsds* tmp_exp_1 = "^{}".format(tmp_exp[:-1]) elif tmp_exp.startswith("$"): tmp_exp_1 = "^{}$".format(tmp_exp[1:]) else: tmp_exp_1 = tmp_exp url["or_conditions"][0]["items"][0]["expression"] = tmp_exp_1 policy_configuration["and_conditions"].append(url) else: raise else: print("######没有对应的转换值#######：{}======={}".format(k,v)) raise if "action_parameter" in match_dict and len(match_dict["action_parameter"]) > 0: for k,v in match_dict["action_parameter"].items(): if k in ["deny_response_text", "response_page", "deny_response_block"]: response = copy.deepcopy(profile_response) if k == "deny_response_text": response_code = v[0]["response_code"] response_content = v[0]["response_content"] response["manipulation_block"] = "warning_page" response["code"] = response_code response["message"] = response_content elif k == "response_page": response_code = v[0]["response_code"] profile_file = { "name": "test", "format": "html", "file_path": "Response-Pages_1.html" } profile_file["name"] = v[0]["profile_file"]["name"] profile_file["format"] = "html" profile_file["file_path"] = v[0]["profile_file"]["file"] response["manipulation_block"] = "warning_page" response["code"] = response_code response["html_profile"] = profile_file elif k == "deny_response_block": pass else: raise policy_configuration["action_parameter"] = response elif k in ["edit_element_parameter"]: edit_element = copy.deepcopy(profile_element) for i in range(len(v)): search_scope = v[i]["search_scope"] contained_keyword = v[i]["contained_keyword"] element_treatment = v[i]["treatment"] target_distance_from_matching = int(v[i]["distance_from_anchor"]) start_indicator = v[i]["start_indicator"] tmp_dict = { "anchor_element": {}, "target_element": {} } tmp_dict["anchor_element"]["contained_keyword"] = contained_keyword tmp_dict["anchor_element"]["search_scope"] = search_scope if start_indicator != "test_keywork": tmp_dict["anchor_element"]["start_indicator"] = start_indicator tmp_dict["target_element"]["element_treatment"] = element_treatment tmp_dict["target_element"]["target_distance_from_matching"] = target_distance_from_matching edit_element["rules"].append(copy.deepcopy(tmp_dict)) policy_configuration["action_parameter"] = edit_element elif k in ["hijack_file"]: replace_file = copy.deepcopy(profile_replace_file) p_name = v[0]["profile_file"]["name"] file_path = v[0]["profile_file"]["file"] content_name = v[0]["profile_file"]["download_name"] map_type = { "gif" : "image/gif", "jpeg" : "image/jpeg", "png" : "image/png", "svg" : "image/svg+xml", "exe" : "application/x-msdos-program", "apk" : "application/vnd.android.package-archive", "html" : "text/html" } tmp_type = v[0]["profile_file"]["file_type"] content_type = map_type[tmp_type] replace_file["replacement_file"]["name"] = p_name replace_file["replacement_file"]["file_path"] = file_path replace_file["replacement_file"]["content_name"] = content_name replace_file["replacement_file"]["content_type"] = content_type policy_configuration["action_parameter"] = replace_file elif k in ["insert_script"]: p_name = v[0]["profile_file"]["name"] file_path = v[0]["profile_file"]["script"] script_type = v[0]["profile_file"]["script_type"] if script_type == "css": insert_profile = copy.deepcopy(profile_insert_css) insert_profile["css_file"]["name"] = p_name insert_profile["css_file"]["file_path"] = file_path elif script_type == "js": map_sec = { "Before Page Load": "html_head", "After Page Load": "html_body" } insert_position = v[0]["profile_file"]["insert_position"] insert_profile = copy.deepcopy(profile_insert_js) insert_profile["js_file"]["name"] = p_name insert_profile["js_file"]["file_path"] = file_path insert_profile["injection_section"]= map_sec[insert_position] policy_configuration["action_parameter"] = insert_profile elif k in ["redirect_url"]: code = v[0]["response_code"] to = v[0]["redirect_url"] redirect_action = copy.deepcopy(profile_redirect) redirect_action["code"] = code redirect_action["to"] = to policy_configuration["action_parameter"] = redirect_action elif k in ["replace_parameter"]: replace_text = copy.deepcopy(profile_replace_text) map_reg = { "off": 0, "on": 1 } map_search = { "HTTP Request-URI": "http_req_uri", "HTTP Request Header": "http_req_header", "HTTP Response Header": "http_resp_header", "HTTP Request Body": "http_req_body", "HTTP Response Body": "http_resp_body", } for i in range(len(v)): find = v[i]["find"] replace_with = v[i]["replace_with"] tmp_reg = v[i]["regex"] if tmp_reg == "on": print("+++++++++++++++++正则校验+++++++++++++") tmp_search = v[i]["search_in"] tmp_dict = {} tmp_dict["regex_enable"] = map_reg[tmp_reg] tmp_dict["search_in"] = map_search[tmp_search] tmp_dict["find"] = find tmp_dict["replace_with"] = replace_with replace_text["rules"].append(copy.deepcopy(tmp_dict)) policy_configuration["action_parameter"] = replace_text elif k in ["run_script"]: p_name = v[0]["profile_file"]["name"] file_path = v[0]["profile_file"]["script"] max_exec_time = v[0]["profile_file"]["max_exec_time"] p_lua = copy.deepcopy(profile_lua) p_lua["lua_script"]["name"] = p_name p_lua["lua_script"]["file_path"] = file_path p_lua["lua_script"]["max_exec_time"] = int(max_exec_time) policy_configuration["action_parameter"] = p_lua # profile_lua = { # "lua_script": { # "name": "test", # "file_path": "run_script_replace_reqbody.lua", # "max_exec_time": 10 # 提取traffic_generation traffic_generation["tool"] = match_dict["traffic"]["protocol"] traffic_generation["command"] = match_dict["traffic"]["command"] # 提取 verification_result verification_result["excepted_traffic_result"] = match_dict["expected_return"] verification_result["expected_metric"]["hits"] = match_dict["counters"]["hits"] verification_result["expected_log"] = match_dict["log_query_param"] if match_dict["rule_action"] in ["allow"]: policy_configuration["action"] = "allow" elif match_dict["rule_action"] in ["deny"]: policy_configuration["action"] = "deny" elif match_dict["rule_action"] in ["edit_element", "hijack", "insert", "replace"]: policy_configuration["action"] = "modify" elif match_dict["rule_action"] in ["monitor"]: policy_configuration["action"] = "monitor" elif match_dict["rule_action"] in ["redirect"]: policy_configuration["action"] = "redirect" elif match_dict["rule_action"] in ["run_script"]: policy_configuration["action"] = "execute" print() print(str(policy_configuration).replace("'", "\"")) # print() # repr(print(str(traffic_generation).replace("'", "\""))) # print() # print(str(verification_result).replace("'", "\"")) print() for i in range(len(verification_result["expected_log"])): if i + 1 < len(verification_result["expected_log"]): print(str(verification_result["expected_log"][i]).replace("'", "\"") + ",") else: print(str(verification_result["expected_log"][i]).replace("'", "\"")) # 输出匹配的内容 #print(match_text) result_text = tmp_resource_text.replace(" replace_text", match_text) def re_name(case_directory_path='C:/zcw/tsg_test/tests/目录名', file_startwith = "man_"): # 存储文件名（去掉.py后缀） file_names = [] # 获取目录下所有文件的名称 # for filename in os.listdir(case_directory_path): # if filename.endswith(file_endswith) and filename.startswith(file_startwith): # 去掉.py后缀并添加到列表中 # pass # if "run_script" in filename: # tmp = filename.replace("run_script", "execute") # old_path = os.path.join(case_directory_path, filename) # new_path = os.path.join(case_directory_path, tmp) # os.rename(old_path, new_path) # elif "man_replace_" in filename: # tmp = filename.replace("replace", "modify_replacetext") # old_path = os.path.join(case_directory_path, filename) # new_path = os.path.join(case_directory_path, tmp) # os.rename(old_path, new_path) # elif "man_hijack" in filename: # tmp = filename.replace("hijack", "modify_replacefile") # old_path = os.path.join(case_directory_path, filename) # new_path = os.path.join(case_directory_path, tmp) # os.rename(old_path, new_path) # elif "edit_element" in filename: # tmp = filename.replace("edit_element", "modify_editelement") # old_path = os.path.join(case_directory_path, filename) # new_path = os.path.join(case_directory_path, tmp) # os.rename(old_path, new_path) # if "injectcss" in filename: # tmp = filename.replace("injectcss", "injectjs") # old_path = os.path.join(case_directory_path, filename) # new_path = os.path.join(case_directory_path, tmp) # os.rename(old_path, new_path) n = 0 for filename in os.listdir(case_directory_path): if filename.endswith(file_endswith) and filename.startswith(file_startwith): # 去掉.py后缀并添加到列表中 if len(filename) - 3 > 100: print(filename, " ======= " ,(len(filename)-3)) n = n + 1 print(n) if __name__ == '__main__': # 定义目录和目标文件 需要调整的变量 ms_json_path = 'C:/Users/root/Downloads/Metersphere_Api_TSG (32).json' case_directory_path = 'C:/zcw/tsg_test/tests/manipulation' file_startwith = "man_" file_endswith = ".py" # 本地运行所有用例 # local_run() # 对比MS工具 diff_ms(ms_json_path=ms_json_path, case_directory_path=case_directory_path, file_startwith=file_startwith, file_endswith=file_endswith) # 生成excle使用数据工具 #get_excel_formate_from_case(case_directory_path=case_directory_path, file_startwith=file_startwith, file_endswith=file_endswith) # 替换test_data #replace_test_data_from_py(case_directory_path=case_directory_path, file_startwith=file_startwith, file_endswith=file_endswith) # 重命名 #re_name(case_directory_path=case_directory_path, file_startwith=file_startwith)