基于github_monitor的源代码监控+自动化监控任务生成+webhook实现飞书机器人通知_github-monitor

基于github_monitor的源代码监控+自动化监控任务生成+webhook实现飞书机器人通知

1.实现背景

现在很多企业为了防止源代码泄漏，都需要对github进行监控，防止员工将内部代码私自上传到公开的github上。本文也是围绕这一点。

2.GITHUB监控部署

本文采用的项目是 github_monitor,地址 https://github.com/VKSRC/Github-Monitor,相关部署按照官方教程即可比较简单，这里不过多赘述。这个项目的优点如下；

• 支持一个任务添加多个关键词
• 匹配模式多
• 支持屏蔽仓库或作者
• 可以针对性的设置监控扫描时间

github监控需要添加token才能正常运行，创建token可以参考https://blog.csdn.net/qq_26849933/article/details/125075608

3.自动化监控任务生成

通常情况下，如果我们需要对accesskey之类的大量数据需要监控，手动导致非常麻烦，我们可以通过脚本的方式来实现批量导入，这里提供我写的一个简单的脚本。
只需要配置脚本参数 即可实现全自动的监控任务生成，提供的关键字文本需要按行存放，一个任务的关键词数目均可自由设置。
如base_name =“腾讯云key”，length =15，则会生成按照顺序生成腾讯云key1-15/腾讯云key16-30的任务名称。

#用于批量导入github监控的数据内容
import requests,os,json

#配置参数
base_url = ""  #部署的项目地址
user = ""  #管理员账号的用户名
password = ""  #管理员账号的密码
base_name = " "   #想要批量生成的任务名
start = 1  #任务起始序号
length = 15   #单个监控任务的查询数目
token = ""    #发起请求的token，通过账号密码自动获取
file_path = ''   #需要进行监控的关键字文本，一行一个关键词


#获取 token
def get_token():
    headers = {
                'Content-Type': 'application/json;charset=UTF-8'
            }    
    data = {
        "username":user,
        "password":password
        }
    re = requests.post(url=base_url+"/api/login/", headers=headers, json=data) 
    global token 
    token = json.loads(re.content.decode('utf8').replace("'", '"'))['token']



#添加监控任务
def add_task(name,keywords):
    headers = {
                'Content-Type': 'application/json;charset=UTF-8',
                'Authorization':'Token '+ token
            }
    data ={
        "name":name,
        "keywords":keywords,
        "match_method":1,
        "ignore_org":"",
        "ignore_repo":"",
        "mail":"",
        "pages":5,
        "interval":60
    }
    re = requests.post(url=base_url+"/api/monitor/task.json", headers=headers, json=data)
    if re.status_code ==201:
        print("任务",name,"添加成功！")
    else:
        print("任务",name,"添加失败！",re.status_code)



#处理需要上传数据
def deal_data(base_name,start,length):
    data = ""
    temp = 0
    for line in open(file_path):
        data = data + line
        temp = temp + 1
        start = start + 1
        if temp == length:
            print(data)
            temp = 0
            task_name = base_name + str(start-length) + '-'+ str(start-1)
            add_task(task_name, data)
            data = ''
    if data != '':
        task_name = base_name + str((int)(start/length)*length+1) + '-'+ str(start-1)
        add_task(task_name, data)


get_token()
deal_data(base_name,start,length)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

4.接入飞书hook

由于这个项目本身不支持飞书机器人，因此我自己写了一个简单的飞书机器人推送脚本。推送内容包含命中关键字/发现时间/文件名称/命中内容/访问地址这几个基础信息，如有需要可以自行修改。

4.1 系统配置修改

由于github_monitor本身的数据库通过docker启动，只配置了expose参数，不支持外网主机访问，因此我们需要修改项目目录下的docker-compose.yaml文件
将配置中的

expose:
3306
1
2

修改成

ports:
	3306:3306
1
2

然后重启服务即可

4.2机器人实现

添加机器人的方式很简单，可以参考官方文档 https://open.feishu.cn/document/ukTMukTMukTM/ucTM5YjL3ETO24yNxkjN#d3815c88
这个hook脚本设计的是每次查询最近一个小时内是否有新抓取的未处理监控项
可以挂在服务器上设置需要1小时运行一次的定时任务来处理，具体多久一次 可以按照需求来定，只需更改sql语句上的时间即可。

#github-monitor 飞书钩子脚本

import mysql.connector,requests

#推送消息至飞书
def push_message(data):
    message = {
	"msg_type": "post",
	"content": {
		"post": {
			"zh_cn": {
				"title": "GITHUB监控推送",
				"content": [
					[{
							"tag": "text",
							"text": "命中规则:  {}\n".format(data[1])
						},
                        {
							"tag": "text",
							"text": "发现时间:  {} \n".format(data[13])
						},
                    	{
							"tag": "text",
							"text": "文件名称:  {} \n".format(data[6])
						},
                    	{
							"tag": "text",
							"text": "命中内容:  {} \n".format(data[3])
						},                        
						{
							"tag": "a",
							"text": "详情查看",
							"href": "{}".format(data[4])
						}
					]
				]
			}
		}
	}
}  
    header = { 'Content-Type': 'application/json;charset=UTF-8'}
    re = requests.post(url='飞书hook地址',json=message,headers=header)
    print(re.status_code)
    print(re.content)



#检查是否有新的泄漏通知
def check_update():
    mydb=mysql.connector.connect(
        host="",
    user="",
    passwd="",
    database="github"
    )
    cursor=mydb.cursor()
    cursor.execute("select * from Leakage where add_time > now()-INTERVAL 1 HOUR and status =0")
    dataset = cursor.fetchall()
    if len(dataset)> 0:
        for data in dataset:
            push_message(data)

check_update()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63