Tetragon作为基于eBPF的安全工具，能够直接在内核中执行安全策略，无需多次上下文切换即可完成行为监控与拦截，将其与Python结合可以快速实现定制化的安全防护逻辑。下面先介绍基础的环境准备与核心概念。
Tetragon基础环境准备
首先需要在系统中安装Tetragon，同时安装Python的调用依赖。以Ubuntu系统为例，安装Tetragon的命令如下：
# 添加Tetragon仓库 curl -s https://packages.cilium.io/gpg | sudo apt-key add - echo "deb [signed-by=/usr/share/keyrings/cilium-archive-keyring.gpg] https://packages.cilium.io/deb ./" | sudo tee /etc/apt/sources.list.d/cilium.list # 安装Tetragon sudo apt update sudo apt install tetragon # 启动Tetragon服务 sudo systemctl start tetragon
Python端需要安装requests库用于调用Tetragon的API接口，执行以下命令安装：
pip install requests
eBPF安全策略核心概念
Tetragon的eBPF安全策略主要通过TracingPolicy资源定义，核心包含以下几个部分：
- spec：策略的具体规则定义，包含监控的对象、触发条件、执行动作
- selector：选择策略作用的对象，比如指定进程、用户、命名空间等
- match：匹配规则，定义需要监控的系统调用或行为类型
- actions：匹配到规则后执行的动作，比如记录日志、拦截操作等
Python配置基础eBPF安全策略
下面的示例通过Python调用Tetragon的API，创建一个监控文件写入操作的eBPF安全策略，当指定目录下的文件被写入时记录日志。
import requests
import json
# Tetragon API地址，默认本地端口
TETRAGON_API = "http://127.0.0.1:1234/api/v1.0"
def create_file_write_policy():
# 定义TracingPolicy策略内容
policy = {
"apiVersion": "cilium.io/v1alpha1",
"kind": "TracingPolicy",
"metadata": {
"name": "file-write-monitor"
},
"spec": {
"selectors": [
{
"matchSelectors": [
{
"matchEvents": [
{
"event": "write"
}
],
"matchArgs": [
{
"index": 0,
"operator": "Prefix",
"values": ["/tmp/sensitive/"]
}
]
}
]
}
],
"actions": [
{
"type": "log"
}
]
}
}
# 调用API创建策略
response = requests.post(
f"{TETRAGON_API}/tracingpolicies",
headers={"Content-Type": "application/json"},
data=json.dumps(policy)
)
if response.status_code == 201:
print("文件写入监控策略创建成功")
else:
print(f"策略创建失败，状态码：{response.status_code}，返回信息：{response.text}")
if __name__ == "__main__":
create_file_write_policy()
策略验证与事件查看
策略创建完成后，可以在/tmp/sensitive/目录下执行文件写入操作，然后通过Python调用API查看捕获的安全事件：
import requests
TETRAGON_API = "http://127.0.0.1:1234/api/v1.0"
def get_security_events():
# 获取最近的安全事件
response = requests.get(f"{TETRAGON_API}/events")
if response.status_code == 200:
events = response.json()
for event in events:
# 只打印文件写入相关事件
if event.get("event_type") == "write":
print(f"捕获到文件写入事件：{event}")
else:
print(f"获取事件失败，状态码：{response.status_code}")
if __name__ == "__main__":
get_security_events()
进阶：配置拦截型安全策略
除了记录日志，还可以配置拦截动作，阻止不符合预期的操作。下面的示例创建一个拦截非法进程启动的eBPF安全策略：
import requests
import json
TETRAGON_API = "http://127.0.0.1:1234/api/v1.0"
def create_process_block_policy():
policy = {
"apiVersion": "cilium.io/v1alpha1",
"kind": "TracingPolicy",
"metadata": {
"name": "illegal-process-block"
},
"spec": {
"selectors": [
{
"matchSelectors": [
{
"matchEvents": [
{
"event": "exec"
}
],
"matchArgs": [
{
"index": 0,
"operator": "Equal",
"values": ["/usr/bin/illegal_app"]
}
]
}
]
}
],
"actions": [
{
"type": "block"
}
]
}
}
response = requests.post(
f"{TETRAGON_API}/tracingpolicies",
headers={"Content-Type": "application/json"},
data=json.dumps(policy)
)
if response.status_code == 201:
print("进程拦截策略创建成功")
else:
print(f"策略创建失败，状态码：{response.status_code}，返回信息：{response.text}")
if __name__ == "__main__":
create_process_block_policy()
策略管理常用操作
实际使用中还需要对策略进行查询、更新、删除操作，以下是常用的Python实现：
import requests
TETRAGON_API = "http://127.0.0.1:1234/api/v1.0"
def list_policies():
# 列出所有策略
response = requests.get(f"{TETRAGON_API}/tracingpolicies")
if response.status_code == 200:
policies = response.json()
print("当前所有安全策略：")
for policy in policies:
print(f"策略名称：{policy['metadata']['name']}")
else:
print(f"获取策略列表失败，状态码：{response.status_code}")
def delete_policy(policy_name):
# 删除指定策略
response = requests.delete(f"{TETRAGON_API}/tracingpolicies/{policy_name}")
if response.status_code == 200:
print(f"策略{policy_name}删除成功")
else:
print(f"策略删除失败，状态码：{response.status_code}")
if __name__ == "__main__":
list_policies()
# delete_policy("file-write-monitor")
注意事项
- Tetragon需要内核版本不低于4.14，且开启eBPF相关配置才能正常工作
- 策略的匹配规则需要准确，错误的规则可能导致正常业务被拦截
- 生产环境中建议对策略进行版本管理，避免误修改导致安全问题
- Python调用API时需要注意权限问题，确保运行Python进程的用户有访问Tetragon API的权限