DevSecOps | 2024-10-09

如何用 Python 手撸一个 GitLab 代码安全审计工具？

本文分享了极狐GitLab 的代码安全审计 & 审计事件流功能，而且演示如何用 Python 编写一个安全审计流接收器，通过接收安全审计日志并分析后发出通知。

本文内容比较丰富，主要分为以下几个部分：

关于代码安全审计
极狐GitLab 安全审计
极狐GitLab 安全审计流
用 Python 构建审计流目的地接受器
结束语
代码附录

代码安全审计

所谓代码安全审计，就是对代码仓库的所有操作进行相应的记录，目的是为了方便安全部门对代码仓库的操作进行安全审计，或者是代码仓库出问题以后，通过审计日志发现问题所在。大白话说就是看看谁对仓库做了什么操作，比如常规的仓库克隆、拉取、推送，当然最可怕的就是传说中的删除跑路或者修改仓库的可见性（从私有修改为公开，很多著名的信息泄露就是由仓库可见性修改引起的）。还有这些年很常见的，有员工在离职前疯狂下载代码，然后当作自己的知识产权，从而带离公司。

这每一件发生在公司内部都是一件大事，毕竟现在数字化时代，很多企业的核心资产就是“一坨坨”的代码。那能够避免这种事情发生或者在事件发生后能及时找到“肇事者”的方法其实就是代码安全审计，这玩意的英文名称叫做Audit event。当然，国内很多开发者可能也叫做“代码追踪”，“代码泄露之类的”。Whatever，不管叫什么，核心就是希望能够用一些手段来保护代码的安全，不要被偷、不要被删，所有的操作都要留下痕迹，而且这痕迹至少要包含三个要素：

Who：事件的操作主体。主要是指对代码进行操作的人，一般来讲当然就是公司内部的研发人员啦；
When：事件发生的时间。主要是指操作是什么时间段发生的；
What：操作主体做了什么具体操作。主要就是看看对仓库代码都做了啥，克隆还是推送，拉取还是删库等。

说半天，这玩意到底咋做呢？

说白了，只能依靠平台自身，平台要是自带了这个功能，那就方便很多，要是不带就没办法了。

极狐GitLab 安全审计 & 安全审计流

好巧不巧的是，GitLab 本身就自带了这个功能，而且随着版本的迭代更新，审计的事件也越来越多，到目前为止（最新为 17.4 版本）审计事件已经多到130+ 项，从实例到群组、到项目，都有。

极狐GitLab 安全审计事件全貌

需要注意的是：安全审计和安全审计流都属于极狐GitLab 专业版及以上功能，但是当前可以申请免费试用 60天。在官网申请后会立马收到一个 license，导入即可！

安全审计功能

极狐GitLab 审计事件可以在实例、群组、项目三个级别查看，路径分别为（以 17.4 为例）：

实例：管理中心 --> 监控 --> 审计事件
群组：群组 --> 安全 --> 审计事件
项目：项目 --> 安全 --> 审计事件

比如添加一个项目，会产生对应的审计事件：

极狐GitLab 审计事件

安全审计事件流

极狐GitLab 审计事件流功能可以将审计事件流发送到外部的流数据系统（可以接受并处理 JSON 格式的数据），然后再由流数据系统对数据进行分析、存储、可视化及告警等操作。

{
	"severity": "INFO",
	"time": "2024-09-26T08:54:16.339Z",
	"correlation_id": "01J8PRKGB20R989VA752DN9ES4",
	"meta.caller_id": "PostReceive",
	"meta.remote_ip": "127.0.0.1",
	"meta.feature_category": "source_code_management",
	"meta.user": "root",
	"meta.user_id": 1,
	"meta.project": "devsecops/ai",
	"meta.root_namespace": "devsecops",
	"meta.client_id": "user/1",
	"meta.root_caller_id": "POST /api/:version/internal/post_receive",
	"id": 274,
	"author_id": 1,
	"entity_id": 7,
	"entity_type": "Project",
	"details": {
		"push_access_levels": ["Maintainers"],
		"merge_access_levels": ["Maintainers"],
		"allow_force_push": false,
		"code_owner_approval_required": false,
		"event_name": "protected_branch_created",
		"author_name": "Administrator",
		"author_class": "User",
		"target_id": 7,
		"target_type": "ProtectedBranch",
		"target_details": "main",
		"custom_message": "Added protected branch with [allowed to push: [\"Maintainers\"], allowed to merge: [\"Maintainers\"], allow force push: false, code owner approval required: false]",
		"ip_address": "218.60.118.175",
		"entity_path": "devsecops/ai"
	},
	"ip_address": "218.60.118.175",
	"author_name": "Administrator",
	"entity_path": "devsecops/ai",
	"target_details": "main",
	"created_at": "2024-09-26T08:54:16.308Z",
	"target_type": "ProtectedBranch",
	"target_id": 7,
	"push_access_levels": ["Maintainers"],
	"merge_access_levels": ["Maintainers"],
	"allow_force_push": false,
	"code_owner_approval_required": false,
	"event_name": "protected_branch_created",
	"author_class": "User",
	"custom_message": "Added protected branch with [allowed to push: [\"Maintainers\"], allowed to merge: [\"Maintainers\"], allow force push: false, code owner approval required: false]"
}

极狐GitLab 可以将审计日志以 JSON 的方式往外发，只要有一个服务能够接受这些 JSON 格式的数据就可以。而且极狐GitLab 本身支持添加第三方的流接收器。

可以在实例、群组级别添加事件流外部接收器：

实例：管理中心 --> 监控 --> 审计事件 --> 事件流
群组：群组 --> 安全 --> 审计事件 --> 事件流

比如在实例级别添加了一个事件流外部接收器：

事件流外部接受地址

主要参数：

目的地名称：写明事件流目的地名称，因为可以添加多个，因此需要用不同的名称来区分
目的地 URL：事件流目的地的地址，也就是接受 JSON 数据的服务地址。这也是本文的核心，这个服务可以自己构建一个。

用 Python 构建审计流目的地接受器

用 Python 主流的 web 框架都可以构建此类接收器，本文使用常用的 fastapi 来构建，代码如下：

from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.post("/jh-gitlab")
async def gitlab_payload(data: dict):
    audit_event_info = {
        "Action": data['details']['custom_message'],
        "Author": data['details']['author_name'],
        "IP Address": data['details']['ip_address'],
        "Entity Path": data['details']['entity_path'],
        "Target Details": data['target_details']
    }
    print(audit_event_info)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

前面看到实际的审计事件日志有很多信息，但是一般想要的就是开头提到的Who、When、What，对应日志里面的字段基本就是action、author、ipaddress、entity_path、target_details。所以，接收到数据以后，先把这些数据取出来，然后做下一步。

将上面的代码存到一个 python 文件里面，然后在服务器上运行起来即可：

python3 main.py
INFO:     Started server process [2140728]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

这时候对代码库做一次变更，比如来个暴力的，直接删除仓库，看看能接收到什么数据：

审计事件流

可以看到，将仓库删除的话，有两个动作：

1.修改仓库的名称

{
	"Action": "Changed name from DevSecOps / ai to DevSecOps / ai-deleted-7",
	"Author": "Administrator",
	"IP Address": "36.133.246.166",
	"Entity Path": "devsecops/ai-deleted-7",
	"Target Details": "devsecops/ai-deleted-7"
}

从上面的信息就能看出，是 adminstor（对，也就是管理员）把 devsecops群组下面的 ai项目删除了。

2. 将仓库标记为等待删除

{
	"Action": "Project marked for deletion",
	"Author": "Administrator",
	"IP Address": "36.133.246.166",
	"Entity Path": "devsecops/ai-deleted-7",
	"Target Details": "ai-deleted-7"
}

从上面的信息就能看出，项目 ai被标记为等待删除，这个可以在项目界面上看到：

被标记为等待删除的项目

接下来就要对不同的操作做一些区分了。因为不同的操作 action 的内容也不尽相同。当然，重要的是这些事件发生以后，如果想特别关注，那就搞一个通知发送机制。下面是一个发送到钉钉群的参考代码：

def notification(payload: dict):
    webhook_url = "https://oapi.dingtalk.com/robot/send?access_token=你的钉钉token"

    # 发送消息的内容
    message = {
        "msgtype": "text",
        "text": {
            "content" : "GitLab: {}".format(json.dumps(payload))
        }
    }

    # 发送 POST 请求
    headers = {'Content-Type': 'application/json'}
    response = requests.post(webhook_url, data=json.dumps(message), headers=headers)

    # 对结果进行判断
    if json.loads(response.text)['errcode'] == 0:
        print("Send Message Success!")
    else:
        print("Send Message Failed!")

然后对仓库做一些操作，比如新建项目、删除项目、克隆项目、推送代码等，就可以看到对应的消息发送到了钉钉群：

钉钉群消息通知

当然，如果觉得上面的这种方式不太容易理解的话，就做一个转换，把 Action 的内容转化成任何人都能看懂的消息，毕竟 git-upload-pack对很多人来说都不是很常见。就把这个任务交给对此感兴趣的小伙伴吧。

结束语

代码安全审计是安全合规非常重要的一环，但是同时也是很多企业容易忽略的一环，究其原因是能够具备如此完整功能的产品不是很多，因为这需要产品不断地持续迭代更新，而且得从早期就做好产品规划。而在这一点上，GitLab 是值得称赞的。当然，说再多也不去亲自去体验。欢迎感兴趣的小伙伴申请专业版免费使用 license 来体验完整的功能。

附录

把这个测试用的代码完整附录如下：


from fastapi import FastAPI
import uvicorn
import requests
import json

app = FastAPI()

@app.post("/jh-gitlab")
async def gitlab_payload(data: dict):
    # 抓取审计事件中的主要信息
    audit_event_info = {
        "Action": data['details']['custom_message'],
        "Author": data['details']['author_name'],
        "IP Address": data['details']['ip_address'],
        "Entity Path": data['details']['entity_path'],
        "Target Details": data['target_details']
    }
    print(audit_event_info)

    # 发送消息通知
    notification(audit_event_info)

def notification(payload: dict):
    webhook_url = "https://oapi.dingtalk.com/robot/send?access_token=你的钉钉 webhook token"

    # 发送消息的内容
    message = {
        "msgtype": "text",
        "text": {
            "content" : "GitLab: {}".format(json.dumps(payload))
        }
    }

    # 发送 POST 请求
    headers = {'Content-Type': 'application/json'}
    response = requests.post(webhook_url, data=json.dumps(message), headers=headers)
    print(response.text)
    if json.loads(response.text)['errcode'] == 0:
        print("Send Message Success!")
        return True
    else:
        print("Send Message Failed!")
        return json.loads(response.text)['errmsg']


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)