MasonLiu/github-cve-monitor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

所有的web请求都加上了异常处理

Browse Source 
1.修复红队工具和cve重复推送问题
2.待增加用户黑名单功能，防止恶意批量创建CVE仓库造成的无意义推送
This commit is contained in:
wuyoukm 2021-09-02 14:29:09 +08:00 committed by GitHub
parent eaa826a3a2
commit 7fdde32532
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 32 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
github_cve_monitor.py
View File

@ -197,17 +197,20 @@ def load_tools_list():
def get_pushed_at_time(tools_list): def get_pushed_at_time(tools_list):
tools_info_list = [] tools_info_list = []
for url in tools_list: for url in tools_list:
tools_json = requests.get(url, headers=github_headers, timeout=10).json()
pushed_at_tmp = tools_json['pushed_at']
pushed_at = re.findall('\d{4}-\d{2}-\d{2}', pushed_at_tmp)[0] #获取的是API上的时间
tools_name = tools_json['name']
api_url = tools_json['url']
try: try:
releases_json = requests.get(url+"/releases", headers=github_headers, timeout=10).json() tools_json = requests.get(url, headers=github_headers, timeout=10).json()
tag_name = releases_json[0]['tag_name'] pushed_at_tmp = tools_json['pushed_at']
pushed_at = re.findall('\d{4}-\d{2}-\d{2}', pushed_at_tmp)[0] #获取的是API上的时间
tools_name = tools_json['name']
api_url = tools_json['url']
try:
releases_json = requests.get(url+"/releases", headers=github_headers, timeout=10).json()
tag_name = releases_json[0]['tag_name']
except Exception as e:
tag_name = "no releases"
tools_info_list.append({"tools_name":tools_name,"pushed_at":pushed_at,"api_url":api_url,"tag_name":tag_name})
except Exception as e: except Exception as e:
tag_name = "no releases" logging.error("get_pushed_at_time BUG -> {}".format(e))
tools_info_list.append({"tools_name":tools_name,"pushed_at":pushed_at,"api_url":api_url,"tag_name":tag_name})
return tools_info_list return tools_info_list
#根据红队名名称查询数据库红队工具的更新时间以及版本名称并返回 #根据红队名名称查询数据库红队工具的更新时间以及版本名称并返回
@ -245,7 +248,7 @@ def get_tools_update_list(data):
# todo BUG在数组 # todo BUG在数组
return tools_update_list return tools_update_list
#获取更新信息并发送到对应社交软件 #获取更新信息并发送到对应社交软件
def send_dingding(url,query_pushed_at,query_tag_name): def send_body(url,query_pushed_at,query_tag_name):
# 考虑到有的工具没有 releases, 则通过 commits 记录获取更新描述 # 考虑到有的工具没有 releases, 则通过 commits 记录获取更新描述
# 判断是否有 releases 记录 # 判断是否有 releases 记录
json_str = requests.get(url + '/releases', headers=github_headers, timeout=10).json() json_str = requests.get(url + '/releases', headers=github_headers, timeout=10).json()
@ -393,21 +396,30 @@ def dingding(text, msg,webhook,secretKey):
ding.send_text(msg='{}\r\n{}'.format(text, msg), is_at_all=False) ding.send_text(msg='{}\r\n{}'.format(text, msg), is_at_all=False)
# server酱 http://sc.ftqq.com/?c=code # server酱 http://sc.ftqq.com/?c=code
def server(text, msg,sckey): def server(text, msg,sckey):
uri = 'https://sc.ftqq.com/{}.send?text={}&desp={}'.format(sckey,text, msg)# 将 xxxx 换成自己的server SCKEY try:
requests.get(uri, headers=github_headers, timeout=10) uri = 'https://sc.ftqq.com/{}.send?text={}&desp={}'.format(sckey,text, msg)# 将 xxxx 换成自己的server SCKEY
requests.get(uri, headers=github_headers, timeout=10)
except Exception as e:
logging.error("server酱 BUG -> {}".format(e))
# 添加Telegram Bot推送支持 # 添加Telegram Bot推送支持
def tgbot(text, msg,token,group_id): def tgbot(text, msg,token,group_id):
import telegram import telegram
bot = telegram.Bot(token='xxx'.format(token))# Your Telegram Bot Token try:
bot.send_message(chat_id=group_id, text='{}\r\n{}'.format(text, msg)) bot = telegram.Bot(token='xxx'.format(token))# Your Telegram Bot Token
bot.send_message(chat_id=group_id, text='{}\r\n{}'.format(text, msg))
except Exception as e:
logging.error("tgbot BUG -> {}".format(e))
# 根据cve 名字，获取描述，并翻译 # 根据cve 名字，获取描述，并翻译
def get_cve_des_zh(cve): def get_cve_des_zh(cve):
time.sleep(3) time.sleep(3)
query_cve_url = "https://cve.mitre.org/cgi-bin/cvename.cgi?name=" + cve try:
response = requests.get(query_cve_url, headers=github_headers, timeout=10) query_cve_url = "https://cve.mitre.org/cgi-bin/cvename.cgi?name=" + cve
html = etree.HTML(response.text) response = requests.get(query_cve_url, headers=github_headers, timeout=10)
des = html.xpath('//*[@id="GeneratedTable"]/table//tr[4]/td/text()')[0].strip() html = etree.HTML(response.text)
return translate(des) des = html.xpath('//*[@id="GeneratedTable"]/table//tr[4]/td/text()')[0].strip()
return translate(des)
except Exception as e:
logging.error("CVE 官网连接 不通 具体BUG -> {}".format(e))
#发送CVE信息到钉钉 #发送CVE信息到钉钉
def sendNews(data): def sendNews(data):
try: try:
@ -460,7 +472,7 @@ if __name__ == '__main__':
for i in range(len(data3)): for i in range(len(data3)):
try: try:
logging.error("[+++] data3 数据 : api_url - > {} pushed_at - > {} tag_name - > {}".format(data3[i]['api_url'],data3[i]['pushed_at'],data3[i]['tag_name'])) logging.error("[+++] data3 数据 : api_url - > {} pushed_at - > {} tag_name - > {}".format(data3[i]['api_url'],data3[i]['pushed_at'],data3[i]['tag_name']))
send_dingding(data3[i]['api_url'],data3[i]['pushed_at'],data3[i]['tag_name']) send_body(data3[i]['api_url'],data3[i]['pushed_at'],data3[i]['tag_name'])
except Exception as e: except Exception as e:
print("main函数 try循环 遇到错误-->{}".format(e)) print("main函数 try循环 遇到错误-->{}".format(e))
logging.error("main函数 try循环 遇到错误-->{}".format(e)) logging.error("main函数 try循环 遇到错误-->{}".format(e))