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

Merge pull request #8 from wuyoukm/master

Browse Source 
细化更新
This commit is contained in:
yhy 2021-07-30 16:54:58 +08:00 committed by GitHub
commit dbe5790593
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 101 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

151
github_cve_monitor.py
View File

@ -7,6 +7,7 @@
# https://my.oschina.net/u/4581868/blog/4380482 # https://my.oschina.net/u/4581868/blog/4380482
# https://github.com/kiang70/Github-Monitor # https://github.com/kiang70/Github-Monitor
import json
import requests, time, re import requests, time, re
import dingtalkchatbot.chatbot as cb import dingtalkchatbot.chatbot as cb
@ -15,20 +16,20 @@ import hashlib
from lxml import etree from lxml import etree
import traceback import traceback
github_headers = { github_headers = {
'Authorization': "token xxxxxx" #替换自己的github token https://github.com/settings/tokens/new 'Authorization': "token xxxxxx" # 替换自己的github token https://github.com/settings/tokens/new
} }
# 抓取本年cve # 抓取本年cve
def getNews(): def getNews():
try: try:
# 抓取本年的 # 抓取本年的
year = datetime.datetime.now().year year = datetime.datetime.now().year
api = "https://api.github.com/search/repositories?q=CVE-{}&sort=updated".format(year) api = "https://api.github.com/search/repositories?q=CVE-{}&sort=updated".format(year)
json_str = requests.get(api, headers = github_headers, timeout= 10).json() json_str = requests.get(api, headers=github_headers, timeout=10).json()
cve_total_count = json_str['total_count'] cve_total_count = json_str['total_count']
cve_description = json_str['items'][0]['description'] cve_description = json_str['items'][0]['description']
cve_url = json_str['items'][0]['html_url'] cve_url = json_str['items'][0]['html_url']
return cve_total_count, cve_description, cve_url return cve_total_count, cve_description, cve_url
@ -36,26 +37,32 @@ def getNews():
print(e, "github链接不通") print(e, "github链接不通")
return '', '', '' return '', '', ''
# 通过 pushed_at 检查工具是否更新 # 通过 pushed_at 检查工具是否更新
def get_pushed_at_time(tools_list): def get_pushed_at_time(tools_list):
total_list = [] total_list = []
for url in tools_list: for url in tools_list:
pushed_at = requests.get(url, headers = github_headers, timeout= 10).json()['pushed_at'] pushed_at = requests.get(url, headers=github_headers, timeout=10).json()['pushed_at']
total_list.append(pushed_at) total_list.append(pushed_at)
return total_list return total_list
def get_update_log(url): def get_update_log(url):
# 考虑到有的工具没有 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()
if len(json_str) != 0: if len(json_str) != 0:
update_log = json_str[0]['body'] try:
update_log = json_str[0]['body']
except Exception as e:
update_log = "作者未写更新内容"
download_url = json_str[0]['html_url'] download_url = json_str[0]['html_url']
return update_log, download_url tools_version = json_str[0]['name']
return update_log, download_url,len(json_str),tools_version
else: else:
json_str = requests.get(url + '/commits', headers = github_headers, timeout= 10).json() json_str = requests.get(url + '/commits', headers=github_headers, timeout=10).json()
update_log = json_str[0]['commit']['message'] update_log = json_str[0]['commit']['message']
download_url = json_str[0]['html_url'] download_url = json_str[0]['html_url']
return update_log, download_url return update_log, download_url
@ -67,18 +74,19 @@ def nmd5(str):
b = str.encode(encoding='utf-8') b = str.encode(encoding='utf-8')
m.update(b) m.update(b)
str_md5 = m.hexdigest() str_md5 = m.hexdigest()
return str_md5 return str_md5
# 有道翻译 # 有道翻译
def translate(word): def translate(word):
headerstr = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36' headerstr = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36'
bv = nmd5(headerstr) bv = nmd5(headerstr)
lts = str(round(time.time()*1000)) lts = str(round(time.time() * 1000))
salt = lts + '90' salt = lts + '90'
# 如果翻译失败，{'errorCode': 50} 请查看 fanyi.min.js: https://shared.ydstatic.com/fanyi/newweb/v1.1.7/scripts/newweb/fanyi.min.js # 如果翻译失败，{'errorCode': 50} 请查看 fanyi.min.js: https://shared.ydstatic.com/fanyi/newweb/v1.1.7/scripts/newweb/fanyi.min.js
# 搜索 fanyideskweb sign: n.md5("fanyideskweb" + e + i + "Y2FYu%TNSbMCxc3t2u^XT") ，Y2FYu%TNSbMCxc3t2u^XT是否改变，替换即可 # 搜索 fanyideskweb sign: n.md5("fanyideskweb" + e + i + "Y2FYu%TNSbMCxc3t2u^XT") ，Y2FYu%TNSbMCxc3t2u^XT是否改变，替换即可
strexample = 'fanyideskweb' + word + salt + 'Y2FYu%TNSbMCxc3t2u^XT' strexample = 'fanyideskweb' + word + salt + 'Y2FYu%TNSbMCxc3t2u^XT'
sign = nmd5(strexample) sign = nmd5(strexample)
data = { data = {
@ -87,8 +95,8 @@ def translate(word):
'to': 'AUTO', 'to': 'AUTO',
'smartresult': 'dict', 'smartresult': 'dict',
'client': 'fanyideskweb', 'client': 'fanyideskweb',
'salt':salt, 'salt': salt,
'sign':sign, 'sign': sign,
'lts': lts, 'lts': lts,
'bv': bv, 'bv': bv,
'doctype': 'json', 'doctype': 'json',
@ -96,20 +104,20 @@ def translate(word):
'keyfrom': 'fanyi.web', 'keyfrom': 'fanyi.web',
'action': 'FY_BY_CLICKBUTTION', 'action': 'FY_BY_CLICKBUTTION',
} }
url='http://fanyi.youdao.com/translate_o?smartresult=dict&smartresult=rule' url = 'http://fanyi.youdao.com/translate_o?smartresult=dict&smartresult=rule'
header={ header = {
'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36',
'Referer':'http://fanyi.youdao.com/', 'Referer': 'http://fanyi.youdao.com/',
'Origin': 'http://fanyi.youdao.com', 'Origin': 'http://fanyi.youdao.com',
'Content-Type':'application/x-www-form-urlencoded; charset=UTF-8', 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'X-Requested-With':'XMLHttpRequest', 'X-Requested-With': 'XMLHttpRequest',
'Accept':'application/json, text/javascript, */*; q=0.01', 'Accept': 'application/json, text/javascript, */*; q=0.01',
'Accept-Encoding':'gzip, deflate', 'Accept-Encoding': 'gzip, deflate',
'Accept-Language':'zh-CN,zh;q=0.9', 'Accept-Language': 'zh-CN,zh;q=0.9',
'Connection': 'keep-alive', 'Connection': 'keep-alive',
'Host': 'fanyi.youdao.com', 'Host': 'fanyi.youdao.com',
'cookie':'_ntes_nnid=937f1c788f1e087cf91d616319dc536a,1564395185984; OUTFOX_SEARCH_USER_ID_NCOO=; OUTFOX_SEARCH_USER_ID=-10218418@11.136.67.24; JSESSIONID=; ___rl__test__cookies=1' 'cookie': '_ntes_nnid=937f1c788f1e087cf91d616319dc536a,1564395185984; OUTFOX_SEARCH_USER_ID_NCOO=; OUTFOX_SEARCH_USER_ID=-10218418@11.136.67.24; JSESSIONID=; ___rl__test__cookies=1'
} }
res = requests.post(url=url, data=data, headers=header) res = requests.post(url=url, data=data, headers=header)
@ -126,29 +134,31 @@ def translate(word):
def dingding(text, msg): def dingding(text, msg):
# 将此处换为钉钉机器人的api # 将此处换为钉钉机器人的api
webhook = 'xxxxxxxx' webhook = 'xxxxxxxx'
secretKey = 'xxxxxxxx' # 替换自己的加签， 钉钉中机器人管理 - 加签 双击，右键复制 secretKey = 'xxxxxxxx' # 替换自己的加签， 钉钉中机器人管理 - 加签 双击，右键复制
ding = cb.DingtalkChatbot(webhook, secret = secretKey) ding = cb.DingtalkChatbot(webhook, secret=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): def server(text, msg):
# 将 xxxx 换成自己的server SCKEY # 将 xxxx 换成自己的server SCKEY
uri = 'https://sc.ftqq.com/xxxx.send?text={}&desp={}'.format(text, msg) uri = 'https://sc.ftqq.com/xxxx.send?text={}&desp={}'.format(text, msg)
requests.get(uri, headers = github_headers, timeout= 10) requests.get(uri, headers=github_headers, timeout=10)
# 添加Telegram Bot推送支持 # 添加Telegram Bot推送支持
def tgbot(text,msg): def tgbot(text, msg):
import telegram import telegram
# Your Telegram Bot Token # Your Telegram Bot Token
bot = telegram.Bot(token='123456:aaa-sdasdsa') bot = telegram.Bot(token='123456:aaa-sdasdsa')
group_id='Your Group ID' group_id = 'Your Group ID'
bot.send_message(chat_id=group_id,text='{}\r\n{}'.format(text,msg)) bot.send_message(chat_id=group_id, text='{}\r\n{}'.format(text, msg))
# 通过检查name 和 description 中是否存在test字样，排除test # 通过检查name 和 description 中是否存在test字样，排除test
def regular(req): def regular(req):
cve_name = req['items'][0]['name'] cve_name = req['items'][0]['name']
cve_description = req['items'][0]['description'] cve_description = req['items'][0]['description']
if cve_name.lower().find('test') == -1 and cve_description.lower().find('test') == -1: if cve_name.lower().find('test') == -1 and cve_description.lower().find('test') == -1:
return True return True
return False return False
@ -157,20 +167,21 @@ def regular(req):
# 根据cve 名字，获取描述，并翻译 # 根据cve 名字，获取描述，并翻译
def get_cve_des_zh(cve): def get_cve_des_zh(cve):
query_cve_url = "https://cve.mitre.org/cgi-bin/cvename.cgi?name=" + cve query_cve_url = "https://cve.mitre.org/cgi-bin/cvename.cgi?name=" + cve
response = requests.get(query_cve_url, headers = github_headers, timeout= 10) response = requests.get(query_cve_url, headers=github_headers, timeout=10)
html = etree.HTML(response.text) html = etree.HTML(response.text)
des = html.xpath('//*[@id="GeneratedTable"]/table//tr[4]/td/text()')[0].strip() des = html.xpath('//*[@id="GeneratedTable"]/table//tr[4]/td/text()')[0].strip()
return translate(des) return translate(des)
def sendNews(tools_list): def sendNews(tools_list):
while True: while True:
try: try:
print("cve 和 github 发布工具 监控中 ...") print("cve 和 github 发布工具 监控中 ...")
# 抓取本年的cve # 抓取本年的cve
year = datetime.datetime.now().year year = datetime.datetime.now().year
api = "https://api.github.com/search/repositories?q=CVE-{}&sort=updated".format(year) api = "https://api.github.com/search/repositories?q=CVE-{}&sort=updated".format(year)
# 请求API # 请求API
req = requests.get(api, headers = github_headers, timeout= 10).json() req = requests.get(api, headers=github_headers, timeout=10).json()
total_count = req['total_count'] total_count = req['total_count']
# 通过 pushed_at 检查工具是否更新 # 通过 pushed_at 检查工具是否更新
@ -185,10 +196,10 @@ def sendNews(tools_list):
# 推送标题 # 推送标题
text = r'有新的CVE送达！' text = r'有新的CVE送达！'
# 获取 cve 名字 ，根据cve 名字，获取描述，并翻译 # 获取 cve 名字 ，根据cve 名字，获取描述，并翻译
cve_name = re.findall("(?i)CVE\-\d+\-\d+", req['items'][0]['name'])[0].upper() cve_name = re.findall('(CVE\-\d+\-\d+)', req['items'][0]['name'])[0].upper()
cve_zh = get_cve_des_zh(cve_name) cve_zh = get_cve_des_zh(cve_name)
msg = "CVE编号："+ cve_name + "\r\n"+"Github地址："+ str(getNews()[2]) + "\r\n" + "CVE描述："+ cve_zh msg = "CVE编号：" + cve_name + "\r\n" + "Github地址：" + str(getNews()[2]) + "\r\n" + "CVE描述：" +"\r\n"+ cve_zh
# 三选一即可，没配置的 注释或者删掉 # 三选一即可，没配置的 注释或者删掉
# server(text, msg) # server(text, msg)
dingding(text, msg) dingding(text, msg)
@ -200,15 +211,55 @@ def sendNews(tools_list):
for i in range(len(tools_list)): for i in range(len(tools_list)):
# 两次时间不相等，则代表工具更新 # 两次时间不相等，则代表工具更新
if time_list1[i] != time_list2[i]: if time_list1[i] != time_list2[i]:
update_log, download_url = get_update_log(tools_list[i]) # get_update_log_info = get_update_log(tools_list[i])
tools_name = tools_list[i].split('/')[-1] if len(get_update_log(tools_list[i])) == 2:
text = r'** ' + tools_name + r' ** 工具更新啦!' update_log = get_update_log(tools_list[i])[0]
body = "工具名称：" + tools_name + "\r\n" + "工具地址：" + download_url + "\r\n" + "工具更新日志：" + update_log download_url = get_update_log(tools_list[i])[1]
# 三选一即可，没配置的 注释或者删掉 tools_name = tools_list[i].split('/')[-1]
# server(text, body) text = r'** ' + tools_name + r' ** 工具更新啦!'
dingding(text, body) body = "工具名称：" + tools_name + "\r\n" + "工具地址：" + download_url + "\r\n" + "工具更新日志：" + "\r\n" + update_log
# tgbot(text,body) # 三选一即可，没配置的 注释或者删掉
print(body) # server(text, body)
dingding(text, body)
# tgbot(text,body)
print(body)
elif len(get_update_log(tools_list[i])) == 4:
one_all_info = get_update_log(tools_list[i])
release_len_one = one_all_info[2]
time.sleep(120)
two_all_info = get_update_log(tools_list[i])
release_len_two = two_all_info[2]
if release_len_one != release_len_two:
update_log = two_all_info[0]
download_url = two_all_info[1]
tools_version = two_all_info[3]
tools_name = tools_list[i].split('/')[-1]
text = r'** ' + tools_name + r' ** 工具版本更新啦!'
body = "工具名称：" + tools_name + "\r\n"+"当前最新版本："+tools_version+"\r\n"+ "工具下载地址：" + download_url + "\r\n" + "工具更新日志：" + "\r\n" + update_log
# 三选一即可，没配置的 注释或者删掉
# server(text, body)
dingding(text, body)
# tgbot(text,body)
else:
commits_url = tools_list[i]+"/commits"
commits_url_response_json = requests.get(commits_url).text
commits_json = json.loads(commits_url_response_json)
tools_name = tools_list[i].split('/')[-1]
download_url = commits_json[0]['html_url']
try:
update_log = commits_json[0]['commit']['message']
except Exception as e:
update_log = "作者未写更新内容，具体点击更新详情地址的URL进行查看"
text = r'** ' + tools_name + r' ** 工具小更新了一波!'
body = "工具名称：" + tools_name + "\r\n" + "更新详情地址：" + download_url + "\r\n" + "工具更新日志：" + "\r\n" + update_log
# 三选一即可，没配置的 注释或者删掉
# server(text, body)
dingding(text, body)
# tgbot(text,body)
except Exception as e: except Exception as e:
print("Program runing error:{}".format(traceback.print_exc())) print("Program runing error:{}".format(traceback.print_exc()))
@ -227,6 +278,6 @@ if __name__ == '__main__':
"https://api.github.com/repos/chaitin/xray", "https://api.github.com/repos/chaitin/xray",
"https://api.github.com/repos/FunnyWolf/pystinger", "https://api.github.com/repos/FunnyWolf/pystinger",
"https://api.github.com/repos/L-codes/Neo-reGeorg", "https://api.github.com/repos/L-codes/Neo-reGeorg",
"https://api.github.com/repos/shadow1ng/fscan", "https://api.github.com/repos/shadow1ng/fscan",
] ]
sendNews(tools_list) sendNews(tools_list)