221 lines
8.3 KiB
Python
221 lines
8.3 KiB
Python
import os
|
||
import subprocess
|
||
from time import sleep
|
||
import platform
|
||
|
||
|
||
|
||
class Order:
|
||
def CMD(self,data):
|
||
result = os.popen(data)
|
||
# res = result.read().encoding('GBK')
|
||
res = result.read()
|
||
result.close()
|
||
# res = res.decode("unicode-escape")
|
||
return res
|
||
def Linux(self):
|
||
pass
|
||
# 根据证书颁发者名字判断证书是否替换
|
||
def Cert_Verification(self,data):
|
||
c = []
|
||
print(1)
|
||
#with open(r'C:\Users\iiesoft\AppData\Local\Programs\Python\Python36\Lib\site-packages\custometest\certificate.yaml', 'r') as foo:
|
||
with open(r'certificate.yaml', 'r') as foo:
|
||
print(2)
|
||
for line in foo.readlines():
|
||
if data in line:
|
||
print(line)
|
||
c.append('证书已替换')
|
||
else:
|
||
pass
|
||
if '证书已替换' in c:
|
||
# print('证书已替换')
|
||
foo.close()
|
||
return '证书已替换'
|
||
else:
|
||
# print('证书未替换')
|
||
foo.close()
|
||
return '证书未替换'
|
||
|
||
def Content_Type(self,data):
|
||
d = []
|
||
with open('certificate.yaml', 'r') as foo:
|
||
for line in foo.readlines():
|
||
if data in line:
|
||
# print(line)
|
||
d.append('Content_Type已替换')
|
||
else:
|
||
pass
|
||
if 'Content_Type已替换' in d:
|
||
# print('证书已替换')
|
||
foo.close()
|
||
return 'Content_Type已替换'
|
||
else:
|
||
# print('证书未替换')
|
||
foo.close()
|
||
return 'Content_Type未替换'
|
||
# curl路由内容设置
|
||
def curl_name(self,data):
|
||
#curl_name = 'curl -kv -m 10 -1 --trace C:/Users/iiesoft/AppData/Local/Programs/Python/Python36/Lib/site-packages/custometest/certificate.yaml '+data+'| iconv -f utf-8 -t gbk'
|
||
curl_name = 'curl -kv -m 10 -1 --trace certificate.yaml '+data+'| iconv -f utf-8 -t gbk'
|
||
return curl_name
|
||
# 控制器
|
||
def manu(self,url,Certificate):
|
||
# print(data['url'])
|
||
n = 0
|
||
while n != len(url):
|
||
b = self.curl_name(url[n])
|
||
d = self.CMD(b)
|
||
# print(d)
|
||
sleep(1)
|
||
if Certificate != "":
|
||
c =self.Cert_Verification(Certificate)
|
||
# f = self.Content_Type(data["Content_Type"])
|
||
sleep(1)
|
||
assert_cer = url[n]+c
|
||
# assert_Content_Type = data['Content_Type']+f
|
||
n+=1
|
||
return d,assert_cer
|
||
|
||
def FTP(self, ftp_type):
|
||
windows_path = os.getcwd()
|
||
linux_path = os.getcwd().replace('\\', '/')
|
||
# 判断FTP执行类型：（下载/登录）
|
||
if ftp_type == "下载":
|
||
# 调用cmd执行FTP下载文件
|
||
data = 'curl -m 20 ftp://202.38.97.230/pub/iso/linux/knoppix/KNOPPIX_V7.7.1DVD-2016-10-22-EN/dpkg-l-dvd-771.txt -u"anonymous:chrome@example.com" -o zmmtext123.txt'
|
||
d = self.CMD(data)
|
||
sleep(5)
|
||
fsize = os.path.getsize(linux_path + "/zmmtext123.txt") # 435814
|
||
if fsize == 435814:
|
||
return "ftp_success"
|
||
else:
|
||
return "ftp_fail"
|
||
elif ftp_type == "登录":
|
||
data = 'curl -m 10 ftp://202.38.97.230/pub/iso/linux/knoppix/KNOPPIX_V7.7.1DVD-2016-10-22-EN/dpkg-l-dvd-771.txt -u"anonymous:chrome@example.com" | iconv -f utf-8 -t gbk'
|
||
d = self.CMD(data)
|
||
# print(d)
|
||
if "Graphical (Xorg) program starter for ADRIANE" in d:
|
||
return "ftp_success"
|
||
else:
|
||
return "ftp_fail"
|
||
# FTP 下载
|
||
def FTP_down(self, ftp_url,file_size,file_name):
|
||
windows_path = os.getcwd()
|
||
linux_path = os.getcwd().replace('\\', '/')
|
||
# 判断FTP执行类型：（下载/登录）
|
||
# 调用cmd执行FTP下载文件
|
||
data = 'curl -m 20 '+ftp_url+ '-o '+ file_name + " ' "
|
||
print(data)
|
||
d = self.CMD(data)
|
||
sleep(5)
|
||
fsize = os.path.getsize(linux_path + "/"+file_name) # 435814
|
||
print(fsize)
|
||
if fsize == file_size:
|
||
return "ftp_success"
|
||
else:
|
||
return "ftp_fail"
|
||
|
||
# FTP 登录
|
||
def FTP_login(self, ftp_url,file_content):
|
||
SYS = self.Operating_System()
|
||
if SYS == "Windows":
|
||
data = 'curl -m 10 '+ftp_url+' | iconv -f utf-8 -t gbk'
|
||
d = self.CMD(data)
|
||
else:
|
||
data = 'curl -m 10 '+ftp_url
|
||
d = self.CMD(data)
|
||
|
||
if file_content in d:
|
||
return "ftp_success"
|
||
else:
|
||
return "ftp_fail"
|
||
|
||
# 判断当前操作系统
|
||
def Operating_System(self):
|
||
os_name = platform.system()
|
||
return os_name
|
||
|
||
#需要替换的内容进行循环替换 jsons为初始默认json datas为需要替换的内容
|
||
# 全局变量 null，将java中的空值（null）,装换位python中空值（""）
|
||
global null
|
||
null = ''
|
||
|
||
# 对需要替换的内容进行循环替换 jsons为初始默认json datas为需要替换的内容，header 启用自定义json
|
||
def Jsoneditmanu(self, jsons, datas=None,header=None):
|
||
#判断是否启用自定义json
|
||
if header != None:
|
||
header = eval(header)
|
||
# 返回 header
|
||
return header
|
||
# 判断是否需要更改json内容
|
||
elif datas != None:
|
||
# datas = eval(datas)
|
||
jsons = eval(jsons)
|
||
# 循环遍历替换json内容
|
||
for k, v in datas.items():
|
||
Order.UpdateAllvalues(self,jsons, k, v)
|
||
return jsons
|
||
else:
|
||
# 返回初始json
|
||
return jsons
|
||
|
||
# 循环嵌套替换
|
||
def UpdateAllvalues(self,mydict, key, value):
|
||
if isinstance(mydict, dict): # 使用isinstance检测数据类型，如果是字典
|
||
if key in mydict.keys(): # 替换字典第一层中所有key与传参一致的key
|
||
mydict[key] = value
|
||
for k in mydict.keys(): # 遍历字典的所有子层级，将子层级赋值为变量chdict，分别替换子层级第一层中所有key对应的value，最后在把替换后的子层级赋值给当前处理的key
|
||
chdict = mydict[k]
|
||
Order.UpdateAllvalues(self,chdict, key, value)
|
||
mydict[k] = chdict
|
||
elif isinstance(mydict, list): # 如是list
|
||
for element in mydict: # 遍历list元素，以下重复上面的操作
|
||
if isinstance(element, dict):
|
||
if key in element.keys():
|
||
element[key] = value
|
||
for k in element.keys():
|
||
chdict = element[k]
|
||
Order.UpdateAllvalues(self,chdict, key, value)
|
||
element[k] = chdict
|
||
|
||
|
||
|
||
if __name__ == '__main__':
|
||
datas = {"url":['https://www.baidu.com'],
|
||
"Certificate":"Tango Secure Gateway CA",
|
||
# "Content_Type":"text/html",
|
||
'log':'Security Event Logs',
|
||
"sni":['baidu'],
|
||
"intercept_code":"200",
|
||
"log_code":"200",
|
||
"certifucate":"1",
|
||
"log_content":"true"
|
||
}
|
||
# data= {"url":['https://www.baidu.com'],
|
||
# "Certificate":"Tango Secure Gateway CA"
|
||
# }
|
||
# url = ['https://www.baidu.com']
|
||
# url = ['https://www.baidu.com']
|
||
# url = ['https://www.baidu.com']
|
||
# # Certificate1 = "Tango Secure Gateway CA"
|
||
# Certificate = "baidu"
|
||
# a='Tango Secure Gateway CA'
|
||
# s = Order()
|
||
# b = s.manu(url,Certificate)
|
||
# print(b[1])
|
||
# FTP下载 传入ftp的路径和文件大小
|
||
ftp_url = 'ftp://202.38.97.230/pub/iso/linux/knoppix/KNOPPIX_V7.7.1DVD-2016-10-22-EN/dpkg-l-dvd-771.txt -u"anonymous:chrome@example.com" '
|
||
ftp_size = 435814
|
||
ftp_issue = s.FTP_down(ftp_url,ftp_size)
|
||
# FTP登录 传入ftp的路径和文件内容
|
||
ftp_url ='ftp://202.38.97.230/pub/iso/linux/knoppix/KNOPPIX_V7.7.1DVD-2016-10-22-EN/dpkg-l-dvd-771.txt -u"anonymous:chrome@example.com" '
|
||
file_content = "Graphical (Xorg) program starter for ADRIANE"
|
||
ftp_issue = s.FTP_login(ftp_url,file_content)
|
||
# for i in b:
|
||
# print(i)
|
||
# dd = s.CMD('curl -I https://www.baidu.com')
|
||
# print(dd)
|
||
# if "private, no-cache, no-store, proxy-revalidate, no-transform"in dd:
|
||
# print("ok")
|
||
# a ='curl -kv -1 --trace certificate.yaml https://www.baidu.com | iconv -f utf-8 -t gbk' |