github-cve-monitor/github_cve_monitor.py

#!/usr/bin/python3
# -*- coding:utf-8 -*-
# @Author : yhy&ddm&w4ter

# 每3分钟检测一次github是否有新的cve漏洞提交记录，若有则通过server酱和钉钉机器人推送（二者配置一个即可）
# 建议使用screen命令运行在自己的linux vps后台上，就可以愉快的接收各种cve了

# https://my.oschina.net/u/4581868/blog/4380482
# https://github.com/kiang70/Github-Monitor
import json

import requests, time, re
import dingtalkchatbot.chatbot as cb
import datetime
import hashlib
from lxml import etree
import traceback

github_headers = {
    'Authorization': "token xxxxxx"  # 替换自己的github token     https://github.com/settings/tokens/new
}


# 抓取本年cve
def getNews():
    try:
        # 抓取本年的
        year = datetime.datetime.now().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()
        cve_total_count = json_str['total_count']
        cve_description = json_str['items'][0]['description']
        cve_url = json_str['items'][0]['html_url']
        cve_name = json_str['items'][0]['name']
        return cve_total_count, cve_description, cve_url,cve_name

    except Exception as e:
        print(e, "github链接不通")
        return '', '', ''


# 通过 pushed_at 检查工具是否更新
def get_pushed_at_time(tools_list):
    total_list = []
    for url in tools_list:
        pushed_at = requests.get(url, headers=github_headers, timeout=10).json()['pushed_at']
        total_list.append(pushed_at)

    return total_list


def get_update_log(url):
    # 考虑到有的工具没有 releases, 则通过 commits 记录获取更新描述

    # 判断是否有 releases 记录
    json_str = requests.get(url + '/releases', headers=github_headers, timeout=10).json()
    if len(json_str) != 0:
        try:
            update_log = json_str[0]['body']
        except Exception as e:
            update_log = "作者未写更新内容"
        download_url = json_str[0]['html_url']
        tools_version = json_str[0]['name']
        return update_log, download_url,len(json_str),tools_version
    else:
        json_str = requests.get(url + '/commits', headers=github_headers, timeout=10).json()
        update_log = json_str[0]['commit']['message']
        download_url = json_str[0]['html_url']
        return update_log, download_url


# 创建md5对象
def nmd5(str):
    m = hashlib.md5()
    b = str.encode(encoding='utf-8')
    m.update(b)
    str_md5 = m.hexdigest()
    return str_md5


# 有道翻译
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'
    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',
        'to': 'AUTO',
        'smartresult': 'dict',
        'client': 'fanyideskweb',
        'salt': salt,
        'sign': sign,
        'lts': lts,
        'bv': bv,
        'doctype': 'json',
        'version': '2.1',
        'keyfrom': 'fanyi.web',
        '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/',
        'Origin': 'http://fanyi.youdao.com',
        'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
        'X-Requested-With': 'XMLHttpRequest',
        'Accept': 'application/json, text/javascript, */*; q=0.01',
        'Accept-Encoding': 'gzip, deflate',
        'Accept-Language': 'zh-CN,zh;q=0.9',
        'Connection': 'keep-alive',
        '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):
    # 将此处换为钉钉机器人的api
    webhook = 'xxxxxxxx'
    secretKey = 'xxxxxxxx'  # 替换自己的加签， 钉钉中机器人管理 - 加签  双击，右键复制
    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):
    # 将 xxxx 换成自己的server SCKEY
    uri = 'https://sc.ftqq.com/xxxx.send?text={}&desp={}'.format(text, msg)
    requests.get(uri, headers=github_headers, timeout=10)


# 添加Telegram Bot推送支持
def tgbot(text, msg):
    import telegram
    # Your Telegram Bot Token
    bot = telegram.Bot(token='123456:aaa-sdasdsa')
    group_id = 'Your Group ID'
    bot.send_message(chat_id=group_id, text='{}\r\n{}'.format(text, msg))


# 通过检查name 和 description 中是否存在test字样，排除test
def regular(req):
    cve_name = req['items'][0]['name']
    cve_description = req['items'][0]['description']
    if cve_name.lower().find('test') == -1 and cve_description.lower().find('test') == -1:
        return True
    return False


# 根据cve 名字，获取描述，并翻译
def get_cve_des_zh(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)
    html = etree.HTML(response.text)
    des = html.xpath('//*[@id="GeneratedTable"]/table//tr[4]/td/text()')[0].strip()
    return translate(des)


def sendNews(tools_list):
    while True:
        try:
            print("cve 和 github 发布工具 监控中 ...")
            # 抓取本年的cve
            year = datetime.datetime.now().year
            api = "https://api.github.com/search/repositories?q=CVE-{}&sort=updated".format(year)
            # 请求API
            req = requests.get(api, headers=github_headers, timeout=10).json()
            total_count = req['total_count']

            # 通过 pushed_at 检查工具是否更新
            time_list1 = get_pushed_at_time(tools_list)

            # 监控时间间隔3分钟
            time.sleep(180)

            # 检查name 和 description 中是否存在test字样 和 是否更新
            if regular(req) and total_count != getNews()[0]:
                # 推送正文内容
                # 推送标题
                text = r'有新的CVE送达！'
                # 获取 cve 名字 ，根据cve 名字，获取描述，并翻译
                cve_name = re.findall('(CVE\-\d+\-\d+)', getNews()[3])[0].upper()
                cve_zh = get_cve_des_zh(cve_name)
                msg = "CVE编号：" + cve_name + "\r\n" + "Github地址：" + str(getNews()[2]) + "\r\n" + "CVE描述：" +"\r\n"+ cve_zh

                # 三选一即可，没配置的 注释或者删掉
                # server(text, msg)
                dingding(text, msg)
                # tgbot(text,msg)
                print(msg)

            time_list2 = get_pushed_at_time(tools_list)

            for i in range(len(tools_list)):
                # 两次时间不相等，则代表工具更新
                if time_list1[i] != time_list2[i]:
                    # get_update_log_info = get_update_log(tools_list[i])
                    if len(get_update_log(tools_list[i])) == 2:
                        update_log  = get_update_log(tools_list[i])[0]
                        download_url = get_update_log(tools_list[i])[1]
                        tools_name = tools_list[i].split('/')[-1]
                        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)
                        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:
            print("Program runing error:{}".format(traceback.print_exc()))


if __name__ == '__main__':
    tools_list = [
        "https://api.github.com/repos/BeichenDream/Godzilla",
        "https://api.github.com/repos/rebeyond/Behinder",
        "https://api.github.com/repos/AntSwordProject/antSword",
        "https://api.github.com/repos/j1anFen/shiro_attack",
        "https://api.github.com/repos/yhy0/ExpDemo-JavaFX",
        "https://api.github.com/repos/yhy0/github-cve-monitor",
        "https://api.github.com/repos/gentilkiwi/mimikatz",
        "https://api.github.com/repos/ehang-io/nps",
        "https://api.github.com/repos/chaitin/xray",
        "https://api.github.com/repos/FunnyWolf/pystinger",
        "https://api.github.com/repos/L-codes/Neo-reGeorg",
        "https://api.github.com/repos/shadow1ng/fscan",
        "https://api.github.com/repos/SafeGroceryStore/MDUT",
        "https://api.github.com/repos/EdgeSecurityTeam/Vulnerability",
    ]
    
    sendNews(tools_list)