# -*- coding: UTF-8 -*- import json import time import pytz import requests import os import configparser from support.organize_config import * from support.packet_generator.workpath import workdir from datetime import datetime from support.ui_utils.element_position.policy_element_position import * class QueryRuleLog: def __init__(self, parameter, policy_configuration, token, traffic_result): self.parameter = parameter self.policy_configuration = policy_configuration self.token = token self.traffic_result = traffic_result def query_rule_log(self, traffic_generation, verification_result, rule_uuids_tuple, start_time): try: self.rule_type = self.policy_configuration["type"] if self.rule_type == "security" or self.rule_type == "proxy_intercept" or self.rule_type == "proxy_manipulation" or self.rule_type == "monitor" or self.rule_type == "statistics" or self.rule_type == "service_chaining" or self.rule_type == "dos_protection": print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Start to verify the effect of the policy rule by api.", flush=True) log_result = None for _ in range(int(90)): time.sleep(10) utc_tz = pytz.timezone('UTC') current_utc_time = datetime.now(utc_tz) end_time = current_utc_time.strftime('%Y-%m-%dT%H:%M:%SZ') if log_result == None: log_dict, error = self.get_log(traffic_generation, rule_uuids_tuple, start_time, end_time) #print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "DEBGU:start_time:{},end_time:{};log_dict:{}, error:{}".format(start_time, end_time, log_dict, error), flush=True) if len(error) != 0: return error if self.rule_type == "dos_protection": log_result = self.verify_dos_log(log_dict, verification_result, start_time, end_time) else: log_result = self.verify_log(log_dict, verification_result) if log_result != None: break if log_result == True: print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The log result checked by calling api is passed.', flush=True) elif log_result == False: print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The log result checked by calling api is failed.', flush=True) elif log_result == None: if not verification_result["expected_log"]: print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The log result checked by calling api is passed.', flush=True) else: print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The log result checked by calling api is none.', flush=True) elif log_result == "no_set": print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The log result checked by calling api is no_set.', flush=True) return log_result except Exception as e: print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "When querying rule log, the exception error: ", str(e), flush=True) return "When querying rule log, the exception error: " + str(e) def get_log(self, traffic_generation, rule_uuids_tuple, start_time, end_time): try: self.traffic_generation = traffic_generation rule_uuids_list = list(rule_uuids_tuple) rule_uuid = rule_uuids_list[0]["uuid"] headers = {'Content-Type': 'application/json', 'Authorization': self.token} dsl_dict = {"dsl": "", "vsys": 1} log_condition = { "name": "log-query", "limit": 20, "data_source": "", "fields": "", "intervals": [] } log_condition_dict = json.loads(json.dumps(log_condition)) if "rule_number" in self.policy_configuration: rule_number = self.policy_configuration["rule_number"] else: rule_number = 1 for _ in range(rule_number): if self.rule_type == 'security': schema_type = log_source = 'security_event' log_query_rule_type = 'security_rule_uuid_list' elif self.rule_type == 'proxy_intercept': schema_type = log_source = 'session_record' log_query_rule_type = 'proxy_rule_uuid_list' elif self.rule_type == 'proxy_manipulation': schema_type = log_source = 'proxy_event' log_query_rule_type = 'proxy_rule_uuid_list' elif self.rule_type == 'monitor': schema_type = log_source = 'monitor_event' log_query_rule_type = 'monitor_rule_uuid_list' elif self.rule_type == 'statistics': schema_type = log_source = 'session_record' log_query_rule_type = 'statistics_rule_uuid_list' elif self.rule_type == "dos_protection": schema_type = log_source = "dos_event" elif self.rule_type == "service_chaining": schema_type = log_source = 'session_record' log_query_rule_type = 'sc_rule_uuid_list' fields = self.get_log_schema(self.token, schema_type, self.parameter["api_server"], self.parameter["vsys"]) log_condition_dict['fields'] = fields log_condition_dict['data_source'] = log_source log_condition_dict["intervals"].append(start_time + '/' + end_time) log_condition_dict['vsys'] = self.parameter["vsys"] if self.is_attribute_name_exsit("ATTR_SUBSCRIBER_ID") == True: log_filter = f"vsys_id in ({int(self.parameter['vsys'])}) AND subscriber_id='{self.parameter['test_subcriber_id']}' AND has({log_query_rule_type}, '{rule_uuid}')" log_condition_dict['filter'] = log_filter else: if self.traffic_generation["tool"] == "trex": if self.rule_type == "dos_protection": log_condition_dict['filter'] = f"destination_ip='{traffic_generation['servers_start_ip']}'" else: log_filter = f"vsys_id in ({int(self.parameter['vsys'])}) AND client_ip={traffic_generation['clients_start_ip']} AND has({log_query_rule_type}, '{rule_uuid}')" log_condition_dict['filter'] = log_filter log_condition_dict['filter'] = log_condition_dict['filter'].replace(f"client_ip={traffic_generation['clients_start_ip']}",f"client_ip='{traffic_generation['clients_start_ip']}'") else: # log_filter = f"client_ip={self.parameter['test_pc_ip']} AND has({log_query_rule_type}, '{rule_uuid}')" log_filter = f"vsys_id in ({int(self.parameter['vsys'])}) AND client_ip={self.parameter['test_pc_ip']} AND has({log_query_rule_type}, '{rule_uuid}')" log_condition_dict['filter'] = log_filter log_condition_dict['filter'] = log_condition_dict['filter'].replace(f"client_ip={self.parameter['test_pc_ip']}", f"client_ip='{self.parameter['test_pc_ip']}'") dsl_dict["dsl"] = log_condition_dict dsl_dict['vsys'] = int(self.parameter["vsys"]) url = self.parameter["api_server"] + "/v1/logs/query" response = requests.post(url, headers=headers, json=dsl_dict, verify=False) job_dict = json.loads(response.text) job_id = job_dict['data']['job']['job_id'] """ {"execution_mode":"oneshot","query_jobs":[{"id":"e9df1ca8-e270-427d-9baf-3c0e8fb13813","is_saved_query":0,"with_result":false}],"vsys":5} { "code": 200, "msg": "Success", "data": [{ "reason": null, "start_time": "2024-11-27T02:27:21Z", "is_done": true, "job_id": "e9df1ca8-e270-427d-9baf-3c0e8fb13813", "is_canceled": false, "done_progress": 1, "end_time": "2024-11-27T02:27:21Z", "links": { "status": "/v1/query/job/e9df1ca8-e270-427d-9baf-3c0e8fb13813", "count": "/v1/query/job/e9df1ca8-e270-427d-9baf-3c0e8fb13813/count", "list": "/v1/query/job/e9df1ca8-e270-427d-9baf-3c0e8fb13813/list", "timeline": "/v1/query/job/e9df1ca8-e270-427d-9baf-3c0e8fb13813/timeline" }, "is_failed": false }], "success": true } query_job_dict = { "query_jobs": [{"id": job_id, "is_saved_query":0, "with_result":False}], "vsys": int(self.parameter["vsys"]), "limit": 20, "offset": 0 } response = requests.post(url, headers=headers, json=query_job_dict, verify=False) if response.status_code == 200: query_job_dict = json.loads(response.text) is_done = query_job_dict['data'][0]['is_done'] is_failed = query_job_dict['data'][0]['is_failed'] reason = query_job_dict['data'][0]["reason"] if is_done and not is_failed: query_list_dict = { "query_jobs":[{"id": job_id,"query_option":"list"}], "vsys":int(self.parameter["vsys"]),"limit":20,"offset":0 } response = requests.post(url, headers=headers, json=query_list_dict, verify=False) assert response.status_code == 200 log_list = json.loads(response.text) log_list = log_list['data']['list'] return log_list, "" elif is_done and is_failed: return "", reason else: return [], "" """ query_list_dict = { "query_jobs":[{"id": job_id,"query_option":"list"}], "vsys":int(self.parameter["vsys"]),"limit":20,"offset":0 } time.sleep(2) response = requests.post(url, headers=headers, json=query_list_dict, verify=False) assert response.status_code == 200 log_list = json.loads(response.text) log_list = log_list['data']['list'] return log_list, "" except Exception as e: print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "When getting log, the exception error: ", e, flush=True) return "", "When getting log, the exception error: " + str(e) def is_attribute_name_exsit(self, attribute_name): if "and_conditions" in self.policy_configuration: and_conditions = self.policy_configuration["and_conditions"] for i in range(len(and_conditions)): or_conditions = and_conditions[i]["or_conditions"] for j in range(len(or_conditions)): if or_conditions[j]["attribute_name"] == attribute_name: return True return False def get_log_schema(self, token, schema_type, api_host, vsys): headers = {'Content-Type': 'application/json', 'Authorization': token} url = api_host + "/v1/logs/schema/" + schema_type vsys = {"vsys": vsys} response = requests.get(url, headers=headers, params=vsys, verify=False) assert response.status_code == 200 log_schema = json.loads(response.text) log_schema = log_schema['data']['fields'] log_schema = [field['name'] for field in log_schema] return log_schema def verify_log(self, log_dict, verification_result): log_query = verification_result['expected_log'] temp_log_result_list = [] if len(log_dict) > 0 and len(log_query) > 0: for log in log_dict: for log_query_param in log_query: query_field_key = log_query_param["query_field_key"] query_value = log_query_param["query_value"] exclude_fields = [ "packet_capture_file", "http_response_body", "http_request_body", "monitor_mirrored_pkts", "monitor_mirrored_bytes", "client_port", "ssl_ech_flag", "ssl_esni_flag", "sessions", "session_rate", "packets", "packet_rate", "bytes", "bit_rate", "rule_uuid", "conditions" ] dos_verification_fields = [ "sessions", "session_rate", "packets", "packet_rate", "bytes", "bit_rate", "rule_uuid", "source_ip", "destination_ip" ] if query_field_key in log and query_field_key not in (exclude_fields, dos_verification_fields) and log[query_field_key] == query_value: temp_log_result_list.append(True) elif query_field_key in {"packet_capture_file", "http_response_body", "http_request_body"}: # 为true时，表示支持下载；为false时，表示不支持下载 if query_value == True: if len(log[query_field_key]) > 0: packet_capture_url = log[query_field_key] download_parameter = {"url": packet_capture_url, "exportObj": True} url = self.parameter["api_server"] + "/v1/util/download-file" + "?vsys_id=" + str(self.parameter["vsys"]) headers = {"Content-Type": "application/json", "Authorization": self.token} response = requests.get(url, headers=headers, params=download_parameter, verify=False) if response.status_code == 200: temp_log_result_list.append(True) else: temp_log_result_list.append(False) else: temp_log_result_list.append(False) elif query_value == False: if len(log[query_field_key]) > 0 : temp_log_result_list.append(False) else: temp_log_result_list.append(True) elif query_field_key in {"monitor_mirrored_pkts", "monitor_mirrored_bytes"}: actual_bytes_value = self.traffic_result['total_bytes'] actual_pkts_value = self.traffic_result["total_packets"] mirror_bytes_value = 0 mirror_pkts_value = 0 application = self.get_application_from_configration() enable = self.policy_configuration["action_parameter"]["traffic_mirroring"]["enable"] if enable == 1: attribute_name_list = self.get_attribute_name_from_configration() # 根据不同的协议，对mirror相关参数进行赋值 if self.is_exist_in_list(application, ["ftp", "http", "https", "ssl"]): if len(attribute_name_list) > 0: if self.is_exist_in_list(attribute_name_list, ["ATTR_SSL_CN", "ATTR_SSL_SAN"]): mirror_bytes_value = actual_bytes_value - 4619 mirror_pkts_value = actual_pkts_value - 9 elif self.is_exist_in_list(attribute_name_list, ["ATTR_HTTP_RES_HDR"]): mirror_bytes_value = actual_bytes_value - 381 mirror_pkts_value = actual_pkts_value - 5 elif self.is_exist_in_list(attribute_name_list, ["ATTR_FTP_ACCOUNT"]): mirror_bytes_value = actual_bytes_value - 266 mirror_pkts_value = actual_pkts_value - 4 else: mirror_bytes_value = actual_bytes_value - 192 mirror_pkts_value = actual_pkts_value - 3 else: mirror_bytes_value = actual_bytes_value - 192 mirror_pkts_value = actual_pkts_value - 3 # if len(self.policy_configuration["action_parameter"]["traffic_mirroring"]["mirroring_profile"]) > 0: if "mirroring_profile" in self.policy_configuration["action_parameter"]["traffic_mirroring"]: vlan_count = len(self.policy_configuration["action_parameter"]["traffic_mirroring"]["mirroring_profile"]["vlan_array"]) num_vlans = vlan_count mirror_bytes_value *= num_vlans mirror_pkts_value *= num_vlans elif self.is_exist_in_list(application, ["mail"]): if len(attribute_name_list) > 0: if self.is_exist_in_list(attribute_name_list, ["ATTR_MAIL_FROM", "ATTR_MAIL_TO"]): mirror_bytes_value = actual_bytes_value - 19578 mirror_pkts_value = actual_pkts_value - 158 elif self.is_exist_in_list(attribute_name_list, ["ATTR_MAIL_ATT_CONTENT", "ATTR_MAIL_ATT_NAME"]): mirror_bytes_value = actual_bytes_value - 34554 mirror_pkts_value = actual_pkts_value - 170 elif self.is_exist_in_list(attribute_name_list, ["ATTR_MAIL_SUBJECT"]): mirror_bytes_value = actual_bytes_value - 19578 mirror_pkts_value = actual_pkts_value - 158 elif self.is_exist_in_list(attribute_name_list, ["ATTR_MAIL_CONTENT"]): mirror_bytes_value = actual_bytes_value - 21092 mirror_pkts_value = actual_pkts_value - 159 else: mirror_bytes_value = actual_bytes_value - 967 mirror_pkts_value = actual_pkts_value - 9 else: mirror_bytes_value = actual_bytes_value - 967 mirror_pkts_value = actual_pkts_value - 9 # if len(self.policy_configuration["action_parameter"]["traffic_mirroring"]["mirroring_profile"]) > 0: if "mirroring_profile" in self.policy_configuration["action_parameter"]["traffic_mirroring"]: vlan_count = len(self.policy_configuration["action_parameter"]["traffic_mirroring"]["mirroring_profile"]["vlan_array"]) num_vlans = vlan_count mirror_bytes_value *= num_vlans mirror_pkts_value *= num_vlans elif self.is_exist_in_list(application, ["", "quic", "dns"]): mirror_bytes_value = actual_bytes_value mirror_pkts_value = actual_pkts_value if "mirroring_profile" in self.policy_configuration["action_parameter"]["traffic_mirroring"]: # if len(self.policy_configuration["action_parameter"]["traffic_mirroring"]["mirroring_profile"]) > 0: vlan_count = len(self.policy_configuration["action_parameter"]["traffic_mirroring"]["mirroring_profile"]["vlan_array"]) num_vlans = vlan_count mirror_bytes_value *= num_vlans mirror_pkts_value *= num_vlans # 讲上述mirror的值赋值给对应的verification_result的expected_log if self.is_exist_in_list(application, ["ftp", "http", "ssl", "dns", "quic", "", "mail"]): if query_field_key == "monitor_mirrored_pkts" and log[query_field_key] == mirror_pkts_value: temp_log_result_list.append(True) for log_query_param in log_query: if log_query_param["query_field_key"] == "monitor_mirrored_pkts": log_query_param["query_value"] = mirror_pkts_value elif query_field_key == "monitor_mirrored_bytes" and log[query_field_key] == mirror_bytes_value: temp_log_result_list.append(True) for log_query_param in log_query: if log_query_param["query_field_key"] == "monitor_mirrored_bytes": log_query_param["query_value"] = mirror_bytes_value else: temp_log_result_list.append(False) elif query_field_key in {"client_port"}: if "_" in query_value: start, end = map(int, query_value.split('-')) if start <= log[query_field_key] <= end: temp_log_result_list.append(True) else: temp_log_result_list.append(False) else: if log[query_field_key] == query_value: temp_log_result_list.append(True) else: temp_log_result_list.append(False) elif query_field_key in {"ssl_ech_flag", "ssl_esni_flag"}: if query_value == "True": query_value = 1 elif query_value == "False": query_value = 0 if log[query_field_key] == query_value: temp_log_result_list.append(True) else: temp_log_result_list.append(False) elif query_field_key in {"proxy_pinning_status", "proxy_intercept_status", "proxy_passthrough_reason"}: actual_value = log[query_field_key] # 日志查询到的值 expected_value = query_value # 断言的预期值 if query_field_key == "proxy_passthrough_reason": if expected_value in {"EV Certificate", "Certificate Transparency", "Protocol Errors", "Mutual Authentication", "Certificate Not Installed", "Certificate Pinning"} and actual_value in {""}: # 证书动态放行会产生两条日志，其中一条大概率是拦截的日志，则跳过断言 continue else: if actual_value == expected_value: temp_log_result_list.append(True) else: temp_log_result_list.append(False) elif query_field_key == "proxy_intercept_status": # proxy_intercept_status的值需要转换 if expected_value == "passthrough": # proxy_intercept_status， 0=passthrough, 1=intercept expected_value = 0 else: expected_value = 1 if expected_value == 0 and actual_value == 1: continue else: if actual_value == expected_value: temp_log_result_list.append(True) else: temp_log_result_list.append(False) elif query_field_key in {"sc_rsp_raw_uuid_list", "sc_rsp_decrypted_uuid_list"}: query_value = self.policy_configuration["action_parameter"]["sff_profiles"][0]["service_func_profiles"] conf_path = os.path.join(workdir, "configuration_file.ini") conf = configparser.ConfigParser() conf.read(conf_path, encoding="utf-8") active_dst_ip_list = conf.get("sc_active_dst_ip", "ip_list") effective_device_tag_list = ["group-xxg-tsgx", "center-xxg-tsgx"] if log[query_field_key] == [] and self.policy_configuration["action_parameter"]["sf_configuration"][0]["connectivity"]["method"] == "vxlan_g" and self.policy_configuration["action_parameter"]["sf_configuration"][0]["connectivity"]["dest_ip"] not in active_dst_ip_list: temp_log_result_list.append(True) elif log[query_field_key] == [] and self.policy_configuration["action_parameter"]["sf_configuration"][0]["device_group"]["value"] not in effective_device_tag_list: temp_log_result_list.append(True) elif log[query_field_key] == [] and self.policy_configuration["action_parameter"]["sf_configuration"][0]["admin_status"] == 0: temp_log_result_list.append(True) elif query_value == log[query_field_key]: temp_log_result_list.append(True) else: temp_log_result_list.append(False) elif query_field_key in {"sent_pkts", "received_pkts", "sent_bytes", "received_bytes"}: if query_field_key == "sent_pkts": query_value = self.traffic_result["total_packets_sent"] elif query_field_key == "received_pkts": query_value = self.traffic_result["total_packets_received"] elif query_field_key == "sent_bytes": query_value = self.traffic_result["total_bytes_sent"] elif query_field_key == "received_bytes": query_value = self.traffic_result["total_bytes_received"] if query_value == log[query_field_key]: temp_log_result_list.append(True) else: temp_log_result_list.append(False) else: temp_log_result_list.append(False) if self.rule_type == 'proxy_intercept' and len(temp_log_result_list) < len(log_query): if True in temp_log_result_list: log_result = True else: log_result = False elif False not in temp_log_result_list: log_result = True else: log_result = False elif len(log_dict) == 0 and len(log_query) > 0: log_result = None elif len(log_query) == 0: log_result = True return log_result def get_application_from_configration(self): and_conditions = self.policy_configuration["and_conditions"] # action_parameter = self.policy_configuration["action_parameter"] for i in range(len(and_conditions)): or_conditions = and_conditions[i]["or_conditions"] for j in range(len(or_conditions)): if or_conditions[j]["attribute_name"] == "ATTR_APP_ID": application = or_conditions[j]["items"] return application def is_exist_in_list(self, application, actual_application_list): application_set = set(actual_application_list) for element in application: if element not in application_set: return False return True def get_attribute_name_from_configration(self): attribute_name_list = [] # 只收集Protocol Filed相关的Attribute Name，所以需要根据实际情况补充exclude_list？？ exclude_list = ["ATTR_SOURCE_IP", "ATTR_DESTINATION_IP", "ATTR_APP_ID"] and_conditions = self.policy_configuration["and_conditions"] for i in range(len(and_conditions)): or_conditions = and_conditions[i]["or_conditions"] for j in range(len(or_conditions)): if or_conditions[j]["attribute_name"] not in exclude_list: attribute_name_list.append(or_conditions[j]["attribute_name"]) return attribute_name_list def verify_dos_log(self, log_dict, verification_result, start_time, end_time): self.verification_result = verification_result log_query = verification_result['expected_log'] temp_log_result_list = [] if len(log_dict) > 0 and len(log_query) > 0: count = -1 for log in log_dict: count += 1 for log_query_param in log_query: query_field_key = log_query_param["query_field_key"] # query_value = log_query_param["query_value"] dos_verification_fields = [ "sessions", "session_rate", "packets", "packet_rate", "bytes", "bit_rate", "rule_uuid", "source_ip", "destination_ip" ] if query_field_key in dos_verification_fields: temp_log_result_list = self.verify_value_from_dos_event(log, query_field_key, count, start_time, end_time) if len(temp_log_result_list) < len(log_query): if True in temp_log_result_list: log_result = True else: log_result = False elif False not in temp_log_result_list: log_result = True else: log_result = False elif len(log_dict) == 0 and len(log_query) > 0: log_result = None elif len(log_query) == 0: log_result = True return log_result def verify_value_from_dos_event(self, log, query_field_key, count, start_time, end_time): session_number, total_packets, total_bytes = self.get_value_from_session_record(start_time, end_time) if self.policy_configuration["action_parameter"]["mitigation"]["behavior"] == "deny": expected_session_number = session_number elif self.policy_configuration["action_parameter"]["mitigation"]["behavior"] == "none" and log["sessions"] < session_number: expected_session_number = log["sessions"] else: # log_result_list.append(False) # expected_session_number = -1 return [False] log_result_list = [] time_diff = log["end_time"] - log["start_time"] expected_packets_count = expected_session_number * total_packets expected_bytes_count = expected_session_number * total_bytes if query_field_key == "sessions": if expected_session_number == log[query_field_key]: log_result_list.append(True) self.verification_result["expected_log"][count]["query_value"] = log[query_field_key] # count，第几次循环，则给log_query_param中的第几次-1的数赋值 else: log_result_list.append(False) elif query_field_key == "packets": if expected_packets_count == log[query_field_key]: log_result_list.append(True) self.verification_result["expected_log"][count]["query_value"] = log[query_field_key] else: log_result_list.append(False) elif query_field_key == "bytes": #大于0且是个整数 if log[query_field_key] > 0 and expected_bytes_count == log[query_field_key]: log_result_list.append(True) self.verification_result["expected_log"][count]["query_value"] = log[query_field_key] else: log_result_list.append(False) elif query_field_key == "bit_rate": bits_rate_value = (log["bytes"] * 8) / time_diff if bits_rate_value == log[query_field_key]: log_result_list.append(True) self.verification_result["expected_log"][count]["query_value"] = log[query_field_key] else: log_result_list.append(False) elif query_field_key == "session_rate": session_rate = expected_session_number / time_diff if session_rate == log[query_field_key]: log_result_list.append(True) self.verification_result["expected_log"][count]["query_value"] = log[query_field_key] else: log_result_list.append(False) elif query_field_key == "packet_rate": packet_rate = expected_packets_count / time_diff if packet_rate == log[query_field_key]: log_result_list.append(True) self.verification_result["expected_log"][count]["query_value"] = log[query_field_key] else: log_result_list.append(False) elif query_field_key in ["basic_attack_type", "basic_sessions","basic_session_rate", "basic_packets", "basic_packet_rate","basic_bytes", "basic_bit_rate"]: log_result_list.append(True) elif query_field_key == "rule_uuid": if len(log["rule_uuid"]) > 0: log_result_list.append(True) self.verification_result["expected_log"][count]["query_value"] = log["rule_uuid"] else: log_result_list.append(False) elif query_field_key == "source_ip" : if self.traffic_generation["clients_start_ip"] == log ["source_ip"]: log_result_list.append(True) self.verification_result["expected_log"][count]["query_value"] = log["source_ip"] else: log_result_list.append(False) elif query_field_key == "destination_ip": if self.traffic_generation["servers_start_ip"] == log[query_field_key]: log_result_list.append(True) self.verification_result["expected_log"][count]["query_value"] = log[query_field_key] else: log_result_list.append(False) else: log_result_list.append(False) return log_result_list def get_value_from_session_record(self, start_time, end_time): headers = {'Content-Type': 'application/json', 'Authorization': self.token} log_condition = { "page_no": 1, "page_size": 20, "source": "session_record", "columns": None, "start_time": "", "end_time": "", "filter": "", "vsys": 1 } schema_type = "session_record" log_condition_dict = json.loads(json.dumps(log_condition)) fields = self.get_log_schema(self.token, schema_type, self.parameter["api_server"], self.parameter["vsys"]) log_condition_dict['columns'] = fields log_condition_dict['start_time'] = start_time log_condition_dict['end_time'] = end_time log_condition_dict['vsys'] = self.parameter["vsys"] log_condition_dict['source'] = "session_record" log_condition_dict['identifier_name'] = "session-record-list" log_condition_dict['execution_mode'] = 'oneshot' if self.policy_configuration["action_parameter"]["mitigation"]['behavior']== "none": log_condition_dict['filter'] = f"server_ip='{self.traffic_generation['servers_start_ip']}'" elif self.policy_configuration["action_parameter"]["mitigation"]['behavior'] == "deny": json_str = json.dumps(self.policy_configuration) if self.is_attribute_name_exsit("ATTR_APP_ID") and "dns" in json_str: log_condition_dict['filter'] = f"server_ip='{self.traffic_generation['servers_start_ip']}'AND received_pkts in (0)" else: log_condition_dict['filter'] = f"server_ip='{self.traffic_generation['servers_start_ip']}'AND received_pkts in (0,1)" url = self.parameter["api_server"] + "/v1/logs/query".format(self.parameter["vsys"]) response = requests.post(url, headers=headers, json=log_condition_dict, verify=False) assert response.status_code == 200 log_list = json.loads(response.text) log_list = log_list['data']['list'] log_entry= log_list[0] received_pkts = log_entry['received_pkts'] sent_pkts = log_entry['sent_pkts'] total_packets = received_pkts + sent_pkts received_bytes = log_entry['received_bytes'] sent_bytes = log_entry['sent_bytes'] total_bytes = received_bytes + sent_bytes session_number = len(log_list) return session_number, total_packets, total_bytes