From f4bad53e73a90675a151d3797c698583e2936ab5 Mon Sep 17 00:00:00 2001 From: wuyoukm <28343933+wuyoukm@users.noreply.github.com> Date: Wed, 25 Aug 2021 17:22:48 +0800 Subject: [PATCH] Update github_cve_monitor.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 增加配置文件，更加方便 --- github_cve_monitor.py | 101 ++++++++++++++++++++++++++++-------------- 1 file changed, 67 insertions(+), 34 deletions(-) diff --git a/github_cve_monitor.py b/github_cve_monitor.py index bed72a8..28f8255 100644 --- a/github_cve_monitor.py +++ b/github_cve_monitor.py @@ -13,15 +13,39 @@ import requests, time, re import dingtalkchatbot.chatbot as cb import datetime import hashlib +import yaml from lxml import etree import sqlite3 -file = "toollist.txt" -github_headers = { - 'Authorization': "token ****" # 替换自己的github token https://github.com/settings/tokens/new -} + today_cve_info_tmp = [] tools_update_list = [] +#读取配置文件 +def load_config(): + with open('config.yaml', 'r') as f: + config = yaml.load(f,Loader=yaml.FullLoader) + github_token = config['all_config']['github_token'] + if int(config['all_config']['dingding'][0].split(":")[1]) == 1: + dingding_webhook = "https:"+config['all_config']['dingding'][1].split(":")[2] + dingding_secretKey = config['all_config']['dingding'][2].split(":")[1] + app_name = config['all_config']['dingding'][3].split(":")[1] + return app_name,github_token,dingding_webhook,dingding_secretKey + elif int(config['all_config']['server'][0].split(":")[1]) == 1: + server_sckey = config['all_config']['server'][1].split(":")[1] + app_name = config['all_config']['server'][2].split(":")[1] + return app_name,github_token,server_sckey + elif int(config['all_config']['tgbot'][0].split(":")[1]) ==1 : + tgbot_token = config['all_config']['tgbot'][1].split(":")[1] + tgbot_group_id = config['all_config']['tgbot'][2].split(":")[1] + app_name = config['all_config']['tgbot'][3].split(":")[1] + return app_name,github_token,tgbot_token,tgbot_group_id + elif int(config['all_config']['tgbot'][0].split(":")[1]) == 0 and int(config['all_config']['server'][0].split(":")[1]) == 0 and int(config['all_config']['dingding'][0].split(":")[1]) == 0: + print("[-] 配置文件有误，三个社交软件的enable不能为0") + +github_headers = { + 'Authorization': "token {}".format(load_config()[1]) # 替换自己的github token https://github.com/settings/tokens/new +} + #初始化创建数据库 def create_database(): conn = sqlite3.connect('data.db') @@ -118,17 +142,15 @@ def tools_insert_into_sqlite3(data): print("tools_insert_into_sqlite3 函数 插入数据成功！") conn.close() #读取本地红队工具链接文件转换成list -def timing_update_tools_list(file): - result = [] - with open(file,'r') as f: - for line in f: - result.append(list(line.strip('\n').split(','))) - return result +def load_tools_list(): + with open('tools_list.yaml', 'r') as f: + list = yaml.load(f,Loader=yaml.FullLoader) + return list['tools_list'] #获取红队工具的名称，更新时间，版本名称信息 def get_pushed_at_time(tools_list): tools_info_list = [] for url in tools_list: - tools_json = requests.get(url[0], headers=github_headers, timeout=10).json() + 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] tools_name = tools_json['name'] @@ -160,7 +182,7 @@ def get_tools_update_list(data): #返回数据库里面的时间和版本 tools_update_list.append({"api_url":dist['api_url'],"pushed_at":today_tools_pushed_at,"tag_name":query_result[1]}) return tools_update_list -#获取更新信息并发送到钉钉 +#获取更新信息并发送到对应社交软件 def send_dingding(url,query_pushed_at,query_tag_name): # 考虑到有的工具没有 releases, 则通过 commits 记录获取更新描述 # 判断是否有 releases 记录 @@ -181,7 +203,12 @@ def send_dingding(url,query_pushed_at,query_tag_name): tools_name = url.split('/')[-1] text = r'** ' + tools_name + r' ** 工具,版本更新啦!' body = "工具名称：" + tools_name + "\r\n" + "工具地址：" + download_url + "\r\n" + "工具更新日志：" + "\r\n" + update_log - dingding(text, body) + if load_config()[0] == "dingding": + dingding(text, body,load_config()[2],load_config()[3]) + elif load_config()[0] == "server": + server(text, body,load_config()[2]) + elif load_config()[0] == "tgbot": + tgbot(text,body,load_config()[2],load_config()[3]) conn = sqlite3.connect('data.db') cur = conn.cursor() sql_grammar = "UPDATE redteam_tools_monitor SET tag_name = '{}' WHERE tools_name='{}'".format(tag_name,tools_name) @@ -203,7 +230,12 @@ def send_dingding(url,query_pushed_at,query_tag_name): update_log = "作者未写更新内容，具体点击更新详情地址的URL进行查看" text = r'** ' + tools_name + r' ** 工具小更新了一波!' body = "工具名称：" + tools_name + "\r\n" + "更新详情地址：" + download_url + "\r\n" + "commit更新日志：" + "\r\n" + update_log - dingding(text, body) + if load_config()[0] == "dingding": + dingding(text, body,load_config()[2],load_config()[3]) + elif load_config()[0] == "server": + server(text, body,load_config()[2]) + elif load_config()[0] == "tgbot": + tgbot(text,body,load_config()[2],load_config()[3]) conn = sqlite3.connect('data.db') cur = conn.cursor() sql_grammar = "UPDATE redteam_tools_monitor SET pushed_at = '{}' WHERE tools_name='{}'".format(new_pushed_at,tools_name) @@ -220,7 +252,12 @@ def send_dingding(url,query_pushed_at,query_tag_name): tools_name = url.split('/')[-1] text = r'** ' + tools_name + r' ** 工具更新啦!' body = "工具名称：" + tools_name + "\r\n" + "工具地址：" + download_url + "\r\n" + "commit更新日志：" + "\r\n" + update_log - dingding(text, body) + if load_config()[0] == "dingding": + dingding(text, body, load_config()[2], load_config()[3]) + elif load_config()[0] == "server": + server(text, body, load_config()[2]) + elif load_config()[0] == "tgbot": + tgbot(text, body, load_config()[2], load_config()[3]) conn = sqlite3.connect('data.db') cur = conn.cursor() sql_grammar = "UPDATE redteam_tools_monitor SET pushed_at = '{}' WHERE tools_name='{}'".format(new_pushed_at,tools_name) @@ -243,12 +280,10 @@ def translate(word): bv = nmd5(headerstr) lts = str(round(time.time() * 1000)) salt = lts + '90' - # 如果翻译失败，{'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是否改变，替换即可 strexample = 'fanyideskweb' + word + salt + 'Y2FYu%TNSbMCxc3t2u^XT' sign = nmd5(strexample) - data = { 'i': word, 'from': 'AUTO', @@ -265,7 +300,6 @@ def translate(word): 'action': 'FY_BY_CLICKBUTTION', } url = 'http://fanyi.youdao.com/translate_o?smartresult=dict&smartresult=rule' - 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', 'Referer': 'http://fanyi.youdao.com/', @@ -279,30 +313,26 @@ def translate(word): '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' } - res = requests.post(url=url, data=data, headers=header) result_dict = res.json() - result = "" for json_str in result_dict['translateResult'][0]: tgt = json_str['tgt'] result += tgt return result + # 钉钉 -def dingding(text, msg): - webhook = '*****' # 将此处换为钉钉机器人的api - secretKey = '****' # 替换自己的加签， 钉钉中机器人管理 - 加签 双击，右键复制 +def dingding(text, msg,webhook,secretKey): ding = cb.DingtalkChatbot(webhook, secret=secretKey) ding.send_text(msg='{}\r\n{}'.format(text, msg), is_at_all=False) # server酱 http://sc.ftqq.com/?c=code -def server(text, msg): - uri = 'https://sc.ftqq.com/xxxx.send?text={}&desp={}'.format(text, msg)# 将 xxxx 换成自己的server SCKEY +def server(text, msg,sckey): + uri = 'https://sc.ftqq.com/{}.send?text={}&desp={}'.format(sckey,text, msg)# 将 xxxx 换成自己的server SCKEY requests.get(uri, headers=github_headers, timeout=10) # 添加Telegram Bot推送支持 -def tgbot(text, msg): +def tgbot(text, msg,token,group_id): import telegram - bot = telegram.Bot(token='123456:aaa-sdasdsa')# Your Telegram Bot Token - group_id = 'Your Group ID' + bot = telegram.Bot(token='xxx'.format(token))# Your Telegram Bot Token bot.send_message(chat_id=group_id, text='{}\r\n{}'.format(text, msg)) # 根据cve 名字，获取描述，并翻译 def get_cve_des_zh(cve): @@ -321,11 +351,14 @@ def sendNews(data): try: cve_name = re.findall('(cve\-\d+\-\d+)', data[i]['cve_name'])[0].upper() cve_zh = get_cve_des_zh(cve_name) - msg = "CVE编号：" + cve_name + "\r\n" + "Github地址：" + str(data[i]['cve_url']) + "\r\n" + "CVE描述：" + "\r\n" + cve_zh - dingding(text, msg) + body = "CVE编号：" + cve_name + "\r\n" + "Github地址：" + str(data[i]['cve_url']) + "\r\n" + "CVE描述：" + "\r\n" + cve_zh + if load_config()[0] == "dingding": + dingding(text, body, load_config()[2], load_config()[3]) + elif load_config()[0] == "server": + server(text, body, load_config()[2]) + elif load_config()[0] == "tgbot": + tgbot(text, body, load_config()[2], load_config()[3]) print("钉钉 发送 CVE 成功") - # server(text, msg) - # tgbot(text,msg) except IndexError: pass except Exception as e: @@ -336,7 +369,7 @@ if __name__ == '__main__': try: #初始化部分 create_database() - tools_list = timing_update_tools_list(file) + tools_list = load_tools_list() tools_data = get_pushed_at_time(tools_list) tools_insert_into_sqlite3(tools_data) @@ -348,7 +381,7 @@ if __name__ == '__main__': cve_insert_into_sqlite3(today_cve_data) #红队工具部分 time.sleep(3) - tools_list = timing_update_tools_list(file) + tools_list = load_tools_list() data2 = get_pushed_at_time(tools_list) data3 = get_tools_update_list(data2) for i in range(len(data3)):