1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
|
# -*- coding: UTF-8 -*-
import configparser
import os
import time
from datetime import datetime
from selenium.webdriver.common.by import By
from support.ui_utils.element_position.policy_element_position import *
from support.ui_utils.policies.edit_rules import EditRules
from support.packet_generator.workpath import workdir
class QueryRuleMetric:
def __init__(self, policy_configuration, verification_result, driver, traffic_result):
self.policy_configuration = policy_configuration
self.verification_result = verification_result
self.driver = driver
self.traffic_result = traffic_result
def query_rule_metric(self):
try:
rule_type = self.policy_configuration["type"]
rule_action = self.policy_configuration["action"]
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Start to verify the policy rule metric by ui.", flush=True)
rules = EditRules(self.driver)
edit_rule_code, _ = rules.edit_rules(rule_type, rule_action)
if edit_rule_code != 200:
return "Fail to edit rule when ui verification."
if rule_type == "security" or rule_type == "proxy_intercept" or rule_type == "proxy_manipulation" or rule_type == "monitor" or rule_type == "statistics":
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Get hit count from ui detail page.", flush=True)
hit_count = -1
if rule_type == "statistics":
# 如果为Statistics Rule无对应的Log类型，则从Session Records查找
hit_count = self.driver.find_element(By.XPATH, statisticsRulePage_rightInfo_72hoursHitCount_posXpath).text
if rule_action == "shunt":
hit_count = self.driver.find_element(By.XPATH, "//td[normalize-space(text())='Hit Count (72 hours)']/parent::tr//span").text
else:
hit_count = self.driver.find_element(By.XPATH, "//td[normalize-space(text())='Hit Count (72 hours)']/parent::tr//span[@class='hit-count']").text
if int(hit_count) == self.verification_result["expected_metric"]["hits"] or int(hit_count) > 0 and self.verification_result["expected_metric"]["hits"] == "many":
metric_result = True
else:
metric_result = False
if metric_result == True:
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The metric result at ui page is {}.'.format(metric_result), flush=True)
elif metric_result == False:
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The metric result at ui page is {}.'.format(metric_result), flush=True)
return metric_result
elif rule_type == "service_chaining":
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Get value of sent pkts, received pkts, sent bytes, received bytes from UI detail page.", flush=True)
time.sleep(4)
sent_pkts_str = self.driver.find_element(By.XPATH, "//td[normalize-space(text())='Sent Packets']/parent::tr//td[2]").text
if sent_pkts_str == "-":
sent_pkts = 0
else:
sent_pkts, _ = self.extract_value("packets", sent_pkts_str)
sent_bytes_str = self.driver.find_element(By.XPATH, "//td[normalize-space(text())='Sent Bytes']/parent::tr//td[2]").text
if sent_bytes_str == "-":
sent_bytes = 0
else:
sent_bytes, _ = self.extract_value("bytes", sent_bytes_str)
received_pkts_str = self.driver.find_element(By.XPATH, "//td[normalize-space(text())='Received Packets']/parent::tr//td[2]").text
if received_pkts_str == "-":
received_pkts = 0
else:
received_pkts, _ = self.extract_value("packets", received_pkts_str)
received_bytes_str = self.driver.find_element(By.XPATH, "//td[normalize-space(text())='Received Bytes']/parent::tr//td[2]").text
if received_bytes_str == "-":
received_bytes = 0
else:
received_bytes, _ = self.extract_value("bytes", received_bytes_str)
parse = configparser.ConfigParser()
parse_dir = os.path.join(workdir, "configuration_file.ini")
parse.read(parse_dir, encoding="utf-8")
ip_str = parse.get("sc_active_dst_ip", "ip_list")
sc_active_ip_list = ip_str.split(",")
effective_device_tag = ["group-xxg-tsgx", "center-xxg-tsgx"]
sff_configuration = self.policy_configuration["action_parameter"]["sff_profiles"]
sf_configuration = self.policy_configuration["action_parameter"]["sff_profiles"][0]["service_func_profiles"]
if len(self.traffic_result) > 0:
if sent_pkts == 0 and received_pkts == 0:
negate_option = self.get_negate_option_from_configration()
if sf_configuration[0]["connectivity"]["method"] == "vxlan_g":
if sf_configuration[0]["connectivity"]["sf_dest_ip"] not in sc_active_ip_list:
metric_result = True
elif sf_configuration[0]["connectivity"]["sf_dest_ip"] in sc_active_ip_list and sf_configuration[0]["admin_status"] == 0:
metric_result = True
elif sf_configuration[0]["device_group"]["tag"] not in effective_device_tag and sff_configuration[0]["load_balance_localization"] == "nearby":
metric_result = True
elif negate_option == 1:
metric_result = True
else:
metric_result = False
elif sf_configuration[0]["connectivity"]["method"] == "layer2_switch":
if sf_configuration[0]["admin_status"] == 0:
metric_result = True
elif sf_configuration[0]["device_group"]["tag"] not in effective_device_tag and sff_configuration[0]["load_balance_localization"] == "nearby":
metric_result = True
elif negate_option == 1:
metric_result = True
else:
metric_result = False
else:
if self.policy_configuration["action_parameter"]["targeted_traffic"] == "raw":
byte_value = self.traffic_result["total_bytes"]
pkt_value = self.traffic_result["total_packets"]
if sf_configuration[0]["connectivity"]["method"] == "vxlan_g" and sf_configuration[0]["connectivity"]["sf_dest_ip"] in sc_active_ip_list:
if 1000 <= pkt_value < 10000:
pkt_value_temp = round(pkt_value/1000, 2)
elif pkt_value < 1000:
pkt_value_temp = pkt_value
if 1024 <= byte_value < 1048576:
total_bytes_temp = round(byte_value/1024.0, 2)
elif byte_value < 1024:
total_bytes_temp = byte_value
elif 1048576 <= byte_value < 1073741824:
total_bytes_temp = round(byte_value/1048576.0, 2)
if sff_configuration["type"] == 1:
if sent_bytes == received_bytes == total_bytes_temp and sent_pkts == received_pkts == pkt_value_temp:
metric_result = True
else:
metric_result = False
elif sff_configuration["type"] == 2:
if sent_bytes == total_bytes_temp and sent_pkts == pkt_value_temp and received_bytes == received_pkts == 0:
metric_result = True
else:
metric_result = False
else:
metric_result = False
elif sf_configuration[0]["connectivity"]["method"] == "layer2_switch":
if 1000 <= pkt_value < 10000:
pkt_value_temp = round(pkt_value/1000, 2)
elif pkt_value < 1000:
pkt_value_temp = pkt_value
if 1024 <= byte_value < 1048576:
total_bytes_temp = round(byte_value/1024.0, 2)
elif byte_value < 1024:
total_bytes_temp = byte_value
elif 1048576 <= byte_value < 1073741824:
total_bytes_temp = round(byte_value/1048576.0, 2)
if sff_configuration["type"] == 2:
if sent_bytes == total_bytes_temp and sent_pkts == pkt_value_temp and received_bytes == received_pkts == 0:
metric_result = True
else:
metric_result = False
else:
metric_result = False
elif self.policy_configuration["action_parameter"]["targeted_traffic"] == "decrypted":
if sf_configuration[0]["connectivity"]["method"] == "vxlan_g" and sf_configuration[0]["connectivity"]["sf_dest_ip"] in sc_active_ip_list:
if sff_configuration["type"] == 1:
if sent_bytes == received_bytes and sent_pkts == received_pkts:
metric_result = True
else:
metric_result = False
elif sff_configuration["type"] == 2:
if sent_bytes != 0 and sent_pkts != 0 and received_bytes == received_pkts == 0:
metric_result = True
else:
metric_result = False
else:
metric_result = False
elif sf_configuration[0]["connectivity"]["method"] == "layer2_switch":
if sff_configuration["type"] == 2:
if sent_bytes != 0 and sent_pkts != 0 and received_bytes == received_pkts == 0:
metric_result = True
else:
metric_result = False
else:
metric_result = False
if metric_result == True:
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The metric result at ui page is True', flush=True)
else:
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The metric result at ui page is False', flush=True)
return metric_result
elif rule_type == "dos_protection":
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "Get hit count from ui detail page.", flush=True)
hit_count = self.driver.find_element(By.XPATH, "//td[normalize-space(text())='Hit Count (72 hours)']/following-sibling::td").text
self.driver.find_element(By.ID, "Log").click()
self.driver.find_element(By.ID, "Log_DosEventLogs").click()
time.sleep(1)
self.driver.find_element(By.ID, "addFilter").click()
self.driver.find_element(By.XPATH, "//div[@class='el-input el-input--small el-input--suffix is-focus']//input").click()
self.driver.find_element(By.XPATH, "//div[@class='el-input el-input--small el-input--suffix is-focus']//input").send_keys("destination")
self.driver.find_element(By.ID, "Destination IP8-_Label_UnrecognizedExp_Expression_groupItem_TagsSearch_FacedtedSearch_Search_VPanel_DosLog_Home_App_anonymousComponent").click()
self.driver.find_element(By.ID, "operator=").click()
for t in range(len(self.verification_result)):
if self.verification_result[t]["query_field_key"] == "destination_ip":
destination_ip = self.verification_result[t]["query_value"]
self.driver.find_element(By.XPATH, "//div[@class='el-tooltip el-input el-input--mini']//input").send_keys(destination_ip)
btn = self.driver.find_element(By.ID, "facedQuery")
self.driver.execute_script("arguments[0].click()", btn)
log_number_str = self.driver.find_element(By.XPATH, "//div[@class='search-addition-info search-row']//div[1]").text
log_number = log_number_str.split(": ")[-1]
if (int(hit_count) == self.verification_result["expected_metric"]["hits"]) or (self.verification_result["expected_metric"]["hits"] == "many" and int(log_number) != 0):
metric_result = True
else:
metric_result = False
if metric_result == True:
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The metric result at ui page is {}.'.format(metric_result), flush=True)
elif metric_result == False:
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], 'The metric result at ui page is {}.'.format(metric_result), flush=True)
return metric_result
except Exception as e:
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datetime.now().microsecond)[:3], "When querying rule metric, the exception error: ", str(e), flush=True)
return "When querying rule metric, the exception error: " + str(e)
def extract_value(self, type, value):
if type == "packets":
if " " in value:
value_number = float(value.split(" ")[0])
value_unit = value.split(" ")[-1]
else:
value_number = int(value)
value_unit = ""
elif type == "bytes":
if " " in value:
value_unit = value.split(" ")[-1]
if value_unit == "B":
value_number = int(value.split(" ")[0])
else:
value_number = float(value.split(" ")[0])
else:
value_number = 0
value_unit = ""
return value_number, value_unit
def get_negate_option_from_configration(self):
negate_option_list = []
action_parameter = self.policy_configuration["action_parameter"]
for i in range(len(action_parameter)):
negate_option_list.append(action_parameter[i]["negate_option"])
for value in negate_option_list:
if value == 1:
return 1
return 0
|
