WELCOME

Elkeid®: Elkeid 是一款可以满足主机，容器与容器集群，Serverless 等多种工作负载安全需求的开源解决方案，源于字节跳动内部最佳实践。

源码仓库: github.com/bytedance/Elkeid

Elkeid®-HUB: 通用的规则/事件处理引擎，支持流式/离线数据处理，初衷是通过标准化的抽象语法/规则来解决复杂的数据/事件处理与外部系统联动需求。

社区版下载地址：github.com/bytedance/Elkeid-HUB

Elkeid Architecture

Introduction

Elkeid 具备以下主要能力：

• Elkeid 不仅具备传统的 HIDS(Host Intrusion Detection System) 的对于主机层入侵检测和恶意文件识别的能力，且对容器内的恶意行为也可以很好的识别，部署在宿主机即可以满足宿主机与其上容器内的反入侵安全需求，并且 Elkeid 底层强大的内核态数据采集能力可以满足大部分安全运营人员对于主机层数据的渴望。

• 对于运行的业务 Elkeid 具备 RASP 能力可以注入到业务进程内进行反入侵保护，不仅运维人员不需要再安装一个 Agent，业务也无需重启。

• 对于 K8s 本身 Elkeid 支持接入K8s Audit Log 对 K8s 系统进行入侵检测和风险识别。

• Elkeid 的规则引擎 Elkeid HUB 也可以很好的和外部多系统进行联动对接。

Ekeid 将这些能力都很好的融合在一个平台内，满足不同工作负载的复杂安全需求的同时，还能实现多组件能力关联，更难得的是每个组件均经过字节跳动海量数据和多年的实战检验。

Elkeid Community Edition Description

需要注意的是 Elkeid 开源版本 和完整版本存在差异，目前已开源的能力主要包括：

• 全部端上能力，即端上数据/资产/部分采集能力，内核态数据采集能力，RASP 探针部分等，并与字节跳动内部版本一致；
• 全部接入层能力，即 Agent Center，服务发现等，并与字节跳动内部版本一致；
• 提供社区版规则引擎即 Elkeid HUB，并配套少量策略作为示例使用；
• 提供社区版 Elkeid Console 与部分配套能力。

因此需要具备完整的反入侵与风险感知能力，还需要自行基于 Elkeid HUB 进行策略构建和对 Elkeid 采集的数据进行二次加工等工作。

从这里开始

• 快速部署

联系我们

• Lark

• 邮箱: elkeid#bytedance.com

协议

• Elkeid Driver: GPLv2
• Elkeid RASP: Apache-2.0
• Elkeid Agent: Apache-2.0
• Elkeid Server: Apache-2.0
• Elkeid Console: Elkeid LICENSE

快速部署

自动下载缺失预编译ko服务开启提示

服务背景： Elkeid Driver是在内核态工作的，由于内核要求加载的内核模块与内核版本强绑定，我们又无法占用客户机的资源在安装agent时在客户机上编译ko。因此，我们在release包中提供了预编译的ko，避免每次都需要手动编译ko，目前共计包含3435个预编译ko。但依旧存在两个问题无法解决，一是无法实时更新，上游发行版更新内核后，我们无法也没有人力同步更新预编译的ko到release中，二是覆盖范围有限，可能会遇见我们未曾使用过的发行版所使用的内核。为此，我们提供了自动下载缺失预编译ko的功能，此功能主要是通知到我们相关同学，该ko有客户在试用，尽快从上游更新或覆盖该发行版。 若您选择同意开启此服务，我们需要同时收集一些基础运行信息，以便我们根据不同需求的用户定制优先级排期，并给出合理的资源占用评估。填写的email信息仅用于区分来源身份，真实email或昵称均可。具体信息如下：

1. 缺失预编译ko的内核版本，服务器架构(仅为arm64或amd64二选一，不涉及任何其他cpu机器信息)。
2. agent center上agent的连接数，每30min收集一次。
3. agent center上agent的qps，包含send和receive，每30min收集一次，取30min的平均值。
4. hub input qps，每30min收集一次，取30min的平均值。
5. redis qps，每30min收集一次，取30min的平均值。
6. redis 内存占用，每30min收集一次，实时数值。
7. kafka 生产和消费的qps，每30min收集一次，取30min的平均值。
8. mongodb qps，每30min收集一次，取30min的平均值。

若您不同意开启此服务，您依旧可以使用release包中提供的预编译ko，其他功能不受影响。具体操作为在release界面下载ko_1.7.0.9.tar.xz，然后替换package/to_upload/agent/component/driver/ko.tar.xz，deploy期间会将ko解压到/elkeid/nginx/ElkeidAgent/agent/component/driver/ko目录中。相关收集信息和下载ko的代码均在已开源的manager代码中，是否开启相关功能取决于manager运行时conf目录下的elkeidup_config.yaml文件。若您在部署期间开启了此服务，但是需要在之后的流程中关闭，您可以将elkeidup_config.yaml文件中的report.enable_report设置为false，之后重启manager即可。

附：相关功能位于manager代码中的以下位置:

• 开关位于internal/monitor/report.go的InitReport()函数中，清空此函数内容，即可关闭功能入口。
• 收集信息项位于internal/monitor/report.go的heartbeatDefaultQuery结构中。
• 自动下载ko功能位于biz/handler/v6/ko.go的SendAgentDriverKoMissedMsg函数中。

Elkeid 完整部署

1、单机docker快速部署 (单机测试环境推荐)

注：宿主机请优先使用centos 7.x 或 debian 9/10，容器内的服务依赖了systemd，而systemd使用了cgroup，容器内外systemd版本相差过大会导致容器内的systemd运行异常，无法启动相应服务。

1.1、导入镜像

# 从release下载的是分卷的镜像，需要先合并镜像
wget https://github.com/bytedance/Elkeid/releases/download/v1.9.1/elkeidup_image_v1.9.1.tar.gz.00
wget https://github.com/bytedance/Elkeid/releases/download/v1.9.1/elkeidup_image_v1.9.1.tar.gz.01
wget https://github.com/bytedance/Elkeid/releases/download/v1.9.1/elkeidup_image_v1.9.1.tar.gz.02
wget https://github.com/bytedance/Elkeid/releases/download/v1.9.1/elkeidup_image_v1.9.1.tar.gz.03
cat elkeidup_image_v1.9.1.tar.gz.* > elkeidup_image_v1.9.1.tar.gz

#导入镜像
docker load -i elkeidup_image_v1.9.1.tar.gz

1.2、运行容器

docker run -d --name elkeid_community \
  --restart=unless-stopped \
  -v /sys/fs/cgroup:/sys/fs/cgroup:ro \
  -p 8071:8071 -p 8072:8072 -p 8080:8080 \
  -p 8081:8081 -p 8082:8082 -p 8089:8080  -p 8090:8090\
  --privileged \
  elkeid/all-in-one:v1.9.1

1.3、设置对外IP

使用本机IP，不能使用127.0.0.1。

docker exec -it elkeid_community bash

cd /root/.elkeidup/

# 命令为交互式
./elkeidup public {ip}


./elkeidup agent init
./elkeidup agent build
./elkeidup agent policy create

cat ~/.elkeidup/elkeid_passwd

部署过程中遇到Elkeid社区版信息收集声明，请参考自动下载缺失预编译ko服务开启提示和Agent Install Remark

1.4、访问前端console并安装Agent

顺利安装完成后，/root/.elkeidup/elkeid_passwd文件记录了各组件的密码和相关的url。

初始密码在构建镜像的时候已经固定了的，为了安全性请不要用于生产环境

访问elkeid_console，按照Console使用手册-安装配置界面的命令进行Agent安装部署。

2、使用elkeidup进行完整部署

2.1、配置目标机器root用户ssh免密登录

如果部署机器为本机，依旧需要配置本机免密登录，登录耗时需要小于1s。 可用以下命令进行验证，两次date命令的输出结果需要相同。

date && ssh root@{ip} date
# 输出时间差小于1s

2.2、解压release产物并配置目录

• 下载release产物（分卷压缩包），并合并压缩包

wget https://github.com/bytedance/Elkeid/releases/download/v1.9.1/elkeidup_package_v1.9.1.tar.gz.00
wget https://github.com/bytedance/Elkeid/releases/download/v1.9.1/elkeidup_package_v1.9.1.tar.gz.01
wget https://github.com/bytedance/Elkeid/releases/download/v1.9.1/elkeidup_package_v1.9.1.tar.gz.02
cat elkeidup_package_v1.9.1.tar.gz.* > elkeidup_package_v1.9.1.tar.gz

也可以参考从源码构建 Elkeid，自行编译和构建package包。

如果之前安装过，请删除/root/.elkeidup和/elkeid文件夹，避免造成干扰

• 解压release产物并配置目录

mkdir -p /root/.elkeidup && cd /root/.elkeidup
mv {DownloadDir}/elkeidup_package_v1.9.1.tar.gz elkeidup_package_v1.9.1.tar.gz
tar -xf elkeidup_package_v1.9.1.tar.gz
chmod a+x /root/.elkeidup/elkeidup

2.3、生成并修改config.yaml

ip为本机非127.0.0.1 ip，若不为单机部署，请参考资源手册修改config.yaml

cd /root/.elkeidup
./elkeidup init --host {ip}
mv config_example.yaml config.yaml

2.4、部署

cd /root/.elkeidup

# 命令为交互式
./elkeidup deploy

部署过程中遇到Elkeid社区版信息收集声明，请参考自动下载缺失预编译ko服务开启提示和Agent Install Remark

2.5、构建Agent

cd /root/.elkeidup

./elkeidup agent init
./elkeidup agent build
./elkeidup agent policy create

2.6、访问前端console并安装Agent

顺利安装完成后，执行cat /root/.elkeidup/elkeid_passwd将看到各组件的随机生成的密码和相关的url。

访问elkeid_console，按照Console使用手册-安装配置 界面的命令进行Agent安装部署。

3、Agent Install Remark

• Driver模块依赖预编译ko，具体支持列表参考：ko_list
• Driver 是否存在的方式：lsmod | grep hids_driver
  • 如果测试机器kernel版本不在支持列表中，请自行编译ko文件和生成sign文件(sha256)，并将其导入Nginx中。
  • 如果在执行elkeidup deploy中未同意声明，也需要自行编译ko或下载Release中对应的预编译ko(支持列表)和sign文件，并将其导入Nginx中。

3.1、ko导入Nginx方法

ko/sign文件的格式应该遵循：hids_driver_1.7.0.4_{uname -r}_{arch}.ko/sign格式， 文件需要放置在nginx对应服务器的：/elkeid/nginx/ElkeidAgent/agent/component/driver/ko下，并修改权限chown -R nginx:nginx /elkeid/nginx。放置完成后，重启Agent即可。

4、HTTPS配置

请参考Elkeid https配置文档

5、 重装指定组件

如果部分组件有更新，或者重新编译了部分组件，可以使用elkeidup reinstall命令重新安装指定组件。

例如 release:v1.9.1.1 中更新了Hub社区版，可以使用以下命令进行重新安装。

# {v1.9.1.1} 为 v1.9.1.1 解压后的package目录
# reinstall hub
cp {v1.9.1.1}/package/hub/hub.tar.gz /root/.elkeidup/package/hub/hub.tar.gz
cp {v1.9.1.1}/package/hub_leader/hub_leader.tar.gz /root/.elkeidup/package/hub_leader/hub_leader.tar.gz

/root/.elkeidup/elkeidup reinstall --component Hub
/root/.elkeidup/elkeidup reinstall --component HubLeader

6、资源占用

该章节仅用于说明如何计算服务端所需资源，因为不同服务器在不同业务和不同负载下产生的事件数量差别很大，不能仅通过agent数进行资源占用评估，推荐配置仅供参考。

6.1 配置计算

以下为需要重点评估的部分，其他部分资源占用相对较少并且非线性增长。

6.1.1 接入带宽

一个8C服务器在中等负载下，单个agent预计为30qps，占用带宽约7KB/S，在kafka上每小时占用磁盘约100MB。

6.1.2 AC

入带宽：所有agent带宽的和，同时需要考虑峰值提供冗余。

出带宽：数据写入到kafka，此部分是未压缩的数据，带宽需求为入带宽的4~6倍，如果超过千兆，需要使用万兆网卡。

CPU&内存：AC单个CPU约可以支撑400~500个agent，30Kqps，cpu内存比例为1:2。

6.1.3 Kafka

入带宽：同AC出带宽

出带宽：若仅用hub消费，则与入带宽相同，否则，每增加一个消费者，出带宽增加一倍。

集群：如果kafka为单节点，可以直接按上述数据计算带宽和磁盘要求，如果为3节点无副本，单台机器带宽需求和磁盘需要变为3分之1。

磁盘：默认仅保持一到两个小时的数据，数据以未压缩的形式保存在kafka的磁盘上，根据数据保存时间计算磁盘容量和冗余。

副本：若配置副本，磁盘需求对应翻倍。

6.1.4 Mongodb

Mongodb集群提供的仅是高可用，单节点与三节点，实际上消耗的均是master节点的资源。

1000以下：独占或等效独占的 4C8G 100G

1000-5000: 独占或等效独占的 8C16G 200G

5000以上：独占或等效独占的 16C32G 500G

6.1.5 Manager

每5000台Agent，提供一个4C8G的manager实例即可。

6.1.6 Hub

在满足通常安全需求时，单个Hub CPU预期支持 100核。但社区版尚未支持hub集群。

6.2 推荐配置

6.1.1 <1000台agent 无高可用

6.2.2 3000-5000 高可用

Elkeid HUB 单独部署

如果需要单独部署HUB，可以使用elkeidup中的--hub_only参数，具体步骤如下：

1、配置目标机器root用户ssh免密登录

如果部署机器为本机，依旧需要配置本机免密登录，登录耗时需要小于1s。 可用以下命令进行验证，两次date命令的输出结果需要相同。

date && ssh root@{ip} date
# 输出时间差小于1s

2、下载release产物并配置目录

mkdir -p /root/.elkeidup && cd /root/.elkeidup
wget https://github.com/bytedance/Elkeid/releases/download/v1.9.1/elkeidup_hub_v1.9.1.tar.gz -O elkeidup.tar.gz && tar -xf elkeidup.tar.gz
chmod a+x /root/.elkeidup/elkeidup

3、生成并修改config.yaml

ip为本机非 127.0.0.1 ip，若不为单机部署，请参考部署资源手册修改config.yaml

cd /root/.elkeidup
## 生成hub only 配置
./elkeidup init --host {ip} --hub_only
mv config_example.yaml config.yaml

4、部署

cd /root/.elkeidup

# 命令为交互式
./elkeidup deploy --hub_only

## status
./elkeidup status --hub_only

## undeploy
./elkeidup undeploy --hub_only

5、访问HUB前端

顺利安装完成后，执行cat /root/.elkeidup/elkeid_passwd将看到各组件的随机生成的密码和相关的url。

访问 elkeid_hub_frontend，参照 Elkeid HUB 社区版快速上手教程 使用。

6、HTTPS配置

请参考Elkeid https配置文档

Elkeid CWPP v1.9.1 社区版 资源配置手册

版本

社区版v1.9.1

架构介绍

注：目前受社区版限制，Hub部分仅支持单机部署

组件详情

配置文件说明

1. ssh_host 为通用配置，表示该组件在哪些机器上进行部署，若为数组类型，说明该组件支持集群部署，否则只支持单机部署，具体限制见配置文件注释。
2. quota为通用配置，最终会转变为cgroup限制。
3. 单机测试环境下，所有机器都填同一地址即可。

# redis 单台或3台，3台时为哨兵模式
redis:
  install: true
  quota: 1C2G
  ssh_host:
    - redis-1
    - redis-2
    - redis-3

# MongoDB 单台或3台，3台时为副本集模式
mongodb:
  install: true
  quota: 2C4G
  ssh_host:
    - monogo-1
    - monogo-2
    - monogo-3

# MongoDB 单台或3台，3台时为进群模式
kafka:
  install: true
  topic: hids_svr
  partition_num: 12 # 默认单topic分区数
  quota: 2C4G
  ssh_host:
    - kafka-1
    - kafka-2
    - kafka-3

# leader 社区版目前仅支持单机模式
leader:
  install: true
  quota: 1C2G
  ssh_host: leader-1

# nginx 单台多台即可，但其他组件默认只会使用第一台
nginx:
  install: true
  quota: 1C2G
  ssh_host:
    - nginx-1
    - nginx-2
  domain: # 指向nginx机器的域名，仅支持单个
  public_addr: # nginx机器的公网IP，仅支持单个

# sd 单台多台即可
service_discovery:
  install: true
  quota: 1C2G
  ssh_host:
    - sd-1
    - sd-2

# hub 社区版目前仅支持单机模式
hub:
  install: true
  quota: 2C4G
  ssh_host: hub-1

# manager 单台多台即可
manager:
  install: true
  quota: 2C4G
  ssh_host:
    - manager-1

# ac 单台多台即可
agent_center:
  install: true
  grpc_conn_limit: 1500 # 单个AC的最大连接数限制
  quota: 1C2G
  ssh_host:
    - ac-1

# prometheus 单台多台即可，默认只会请求第一台，第二台处于双写状态，不会被查询
prometheus:
  quota: 1C2G
  ssh_host:
    - prometheus-1

# grafana 仅支持一台
grafana:
  quota: 1C2G
  ssh_host: grafana-1

从源码构建 Elkeid CWPP

当前社区版本中，部分组件尚未开源，主要是Elkeidup和Hub的相关组件目前仅能提供社区版二进制，所以无法提供从零到一提供完全从源码构建的构建手册。您可以通过在安装前替换package中的指定文件，或安装后替换可执行程序的方式，达到运行从源码构建的可执行程序的效果。具体文件位置和对应关系说明如下。

安装前替换

Agent

Agent部分在elkeidup deploy过程中会从源码build，所以替换package中的以下文件即可，建议解压该文件，确认该文件与目录结构与替换前的文件相同。

package/agent/v1.9.1/agent/elkeid-agent-src_1.7.0.24.tar.gz

Driver Plugin

Driver plugin同样会在elkeidup deploy过程中会从源码build，所以同样替换package中的以下文件即可，建议解压该文件，确认该文件与目录结构与替换前的文件相同。

package/agent/v1.9.1/driver/driver-src_1.0.0.15.tar.gz

其他agent plugin

其他agent plugin都是预编译好的，按照各plugin的文档，编译好后替换对应的文件即可。注意plugin存在plg格式和tar.gz格式，plg格式为可执行文件，tar.gz为压缩包。版本号目前写死在elkeidup中，需要保持一致，请勿更改。

package/agent/v1.9.1/driver/driver-src_1.0.0.15.tar.gz
package/agent/v1.9.1/baseline/baseline-default-aarch64-1.0.1.23.tar.gz
package/agent/v1.9.1/baseline/baseline-default-x86_64-1.0.1.23.tar.gz
package/agent/v1.9.1/collector/collector-default-aarch64-1.0.0.140.plg
package/agent/v1.9.1/collector/collector-default-x86_64-1.0.0.140.plg
package/agent/v1.9.1/etrace/etrace-default-x86_64-1.0.0.92.tar.gz
package/agent/v1.9.1/journal_watcher/journal_watcher-default-aarch64-1.0.0.23.plg
package/agent/v1.9.1/journal_watcher/journal_watcher-default-x86_64-1.0.0.23.plg
package/agent/v1.9.1/rasp/rasp-default-x86_64-1.9.1.44.tar.gz
package/agent/v1.9.1/scanner/scanner-default-aarch64-3.1.9.6.tar.gz
package/agent/v1.9.1/scanner/scanner-default-x86_64-3.1.9.6.tar.gz

ko

默认deploy时不会降预编译的ko拷贝到nginx中，在release界面同时会提供预编译的ko，下载预编译的ko或自行编译ko后，替换以下文件即可，文件为tar.xz格式，解压后有一个ko文件夹，格式必须相同。

package/to_upload/agent/component/driver/ko.tar.xz

Manager & ServiceDiscovery & AgentCenter

编译好对应的二进制，解压以下路径的tar.gz，然后替换好二进制后打包回tar.gz即可。

# manager
package/manager/bin.tar.gz
# service discovery
package/service_discovery/bin.tar.gz
# agent center
package/agent_center/bin.tar.gz

安装后替换

Agent 相关

Agent部分均可以通过前端上传，具体见agent发布文档

ko

拷贝对应ko和sing文件到以下目录即可，之后执行命令修改目录权限

# ko 目录
/elkeid/nginx/ElkeidAgent/agent/component/driver/ko

# 修改权限
chown -R nginx:nginx /elkeid/nginx

Manager & ServiceDiscovery & AgentCenter

暂停服务，替换对应的二进制文件，然后重启服务

# manager
systemctl stop elkeid_manager
mv new_manager_bin /elkeid/manager/manager
systemctl start elkeid_manager

# service discovery
systemctl stop elkeid_sd
mv new_sd_bin /elkeid/service_discovery/sd
systemctl start elkeid_sd

# agent center
systemctl stop elkeid_ac
mv new_ac_bin /elkeid/agent_center/agent_center
systemctl start elkeid_ac

Elkeidup 社区版升级指南 1.7.1 --> 1.9.1

前言

首先需要配置elkeidup 1.7.1 与 1.9.1 版本共存，然后按情况进行切换。

详细操作请同时参照1.7.1 与 1.9.1 的文档。

# rename .elkeidup dir
cd /root
mv .elkeidup .elkeidup_v1.7.1
ln -s .elkeidup_v1.7.1 .elkeidup

# copy cert to v1.9.1
mkdir -p /root/.elkeidup_v1.9.1
cp -r /root/.elkeidup_v1.7.1/elkeid_passwd /root/.elkeidup_v1.9.1
cp -r /root/.elkeidup_v1.7.1/cert /root/.elkeidup_v1.9.1
# download v1.9.1 package to /root/.elkeidup_v1.9.1

切换到 1.7.1

rm /root/.elkeidup && ln -s /root/.elkeidup_v1.7.1 /root/.elkeidup

切换到 1.9.1

rm /root/.elkeidup && ln -s /root/.elkeidup_v1.9.1 /root/.elkeidup

后端

v1.9.1后端目前无法与v1.7.1兼容，需要卸载v1.7.1后端后重新安装v1.9.1。

备份数据

根据需要选择备份数据：

1. 备份 MongoDB：目录位于 /elkeid/mongodb 仅是备份DB，备份的数据无法直接使用，如果有恢复需求，目前尚无自动化脚本，需要手动转换。
2. 备份Hub策略：目录位于 /elkeid/hub 策略可以在Hub web界面中进行导入。

卸载v1.7.1

在卸载v1.7.1后端后，Agent将在1min后自动关闭所有插件，并进入守护状态，直到新的后端被安装

# 按照前言操作切换到 v1.7.1

cd /root/.elkeidup 
./elkeidup undeploy

安装v1.9.1

在安装v1.9.1后端后，Agent将在1min内重连，但此时还尚未加载任何插件，您可以在Console上看到这个状态

# 按照前言操作切换到 v1.9.1
# 安装文档详见v1.9.1 安装文档
cd /root/.elkeidup 
./elkeidup deploy

Agent

确认配置及状态

• /root/.elkeidup_v1.7.1/cert /root/.elkeidup_v1.9.1/cert 三个目录内的所有文件内容均保持一致

• /root/.elkeidup_v1.7.1/elkeid_server.yaml /root/.elkeidup_v1.9.1/elkeidup_config.yaml 三个文件中，下述相关配置均保持一致。

  •   注：具体字段filed值以v1.9.1为准，请勿直接覆盖。

  • nginx

    • domain
    • ssh_host
    • public_addr

  • mg

    • ssh_host

• 确认后端更新完成后，所有v1.7.1的Agent均已成功上线

Build v1.9.1组件

./elkeidup agent init
./elkeidup agent build
./elkeidup agent policy create

下发任务

可根据需要进行灰度升级，此时新上线/重连的客户端会自动拉取最新配置升级，其他客户端需要手动同步配置升级

1. 在 Elkeid Console - 任务管理 界面，点击“新建任务”，选择单台主机，点击下一步，选择“同步配置”任务类型，点击确认。随后，在此页面找到刚刚创建的任务，点击运行，观察升级后的主机是否符合预期。
2. 在 Elkeid Console - 任务管理 界面，点击“新建任务”，选择全部主机，点击下一步，选择“同步配置”任务类型，点击确认。随后，在此页面找到刚刚创建的任务，点击运行，即可对存量旧版本Agent进行升级。

Elkeid 社区版 扩容指南

ServiceDiscovery

自身扩容(依赖elkeidup)

1. 修改config.yaml 在sd中添加其他的host，登录条件与安装时相同。
2. 执行以下命令 elkeidup reinstall --component ServiceDiscovery --re-init

自身扩容(手动操作)

1. 拷贝已安装好SD机器的 /elkeid/service_discovery 到待扩容机器上。
2. 更新全部SD的配置文件 /elkeid/service_discovery/conf/conf.yaml 的 Cluster.Members项，该项为所有sd实例的数组，每台sd都要填写全部实例的地址。
3. 执行新SD实例的 /elkeid/service_discovery/install.sh ，会自动启动sd。
4. 重启所有旧的sd实例 systemctl restart elkeid_sd 。

同步修改上下游配置

sd目前同时被AgentCenter，Manager和Nginx所依赖，扩容SD后，需要同步重启。

• AgentCenter: 配置文件位于/elkeid/agent_center/conf/svr.yml 的 sd.addrs，重启命令 systemctl restart elkeid_ac。
• Manager: 配置文件位于/elkeid/manager/conf/svr.yml 的 sd.addrs，重启命令 systemctl restart elkeid_manager。
• Nginx: 配置文件位于/elkeid/nginx/nginx/nginx.conf 的 upstream sd_list，重启命令 systemctl restart elkeid_nginx。

AgentCenter

自身扩容(依赖elkeidup)

1. 修改config.yaml 在ac中添加其他的host，登录条件与安装时相同。
2. 执行以下命令 elkeidup reinstall --component AgentCenter --re-init

自身扩容(手动操作)

1. 拷贝已安装好AC机器的 /elkeid/agent_center 到待扩容机器上。
2. 执行新AC实例的 /elkeid/agent_center/install.sh ，会自动安装和启动AC。

同步修改上下游配置

若agent通过服务发现的方式连接到AC，则不需要手动同步上下游配置。

若agent通过编码的AC地址连接AC，需要重新编译agent，将新的AC地址加入到agent连接配置中。

Elkeid https配置文档

1 概述

• 默认情况下，Elkeid Console监听在8082和8072端口，Elkeid HUB监听在8081和8071端口。
• 如果需要使用HTTPS，可以使用8072和8071端口来访问。

2、使用企业内部证书

安装时生成的自签名证书位于nginx所在机器的 /elkeid/nginx/nginx 目录，包括以下两个文件：

server.key
server.crt

替换上述两个文件后，执行以下操作即可：

chown -R nginx:nginx /elkeid/nginx
systemctl restart elkeid_nginx

3、使用部署时生成的自签名证书

Elkeid 在部署时，仅能使用自签名证书，由于chrome的安全设置，无法直接访问，所有需要手动信任该证书才可使用https访问，具体操作参考如下： 以下示例假设nginx所在服务器为 console.elkeid.org，并配置了/etc/hosts或dns解析。

3.1、Macos

1. 访问 https://console.elkeid.org:8072/ 导出证书
2. 导入导出的证书并信任
3. 点击钥匙串访问，信任证书
4. 再次访问 https://console.elkeid.org:8072/

快速上手

Elkeid Console 主机安全保护平台使用手册

版本

社区版v1.9.1

安全概览

通过安全概览能快速了解覆盖范围，告警势态，运营情况(安全告警与安全风险处置情况)等的整体安全态势。

资产概览

查看当前在覆盖的主机和容器集群信息。

入侵告警

展示未处理告警的数量以及最近7天内的变化趋势。

漏洞风险

展示未处理漏洞的数量以及最近7天内的变化趋势。

基线风险

展示TOP3的基线风险数量。

主机风险分布

展示当前存在待处理告警、高可利用漏洞和基线风险的主机占比。

Agent 概览

展示Agent的在线情况和资源占用情况。

后端服务状态

展示后端服务的负载。

资产中心

运营人员可以通过资产中心对资产数，Agent运行情况，资产上详细信息进行查看。

主机列表

展示主机资产的列表以及各主机的风险情况。

点击"主机名称"可以进入查看此主机的资产信息和风险信息。

点击页面中各标签可以查看主机各类相关数据，目前支持下列数据。

• 安全告警
• 漏洞风险
• 基线风险
• 运行时安全告警
• 病毒查杀
• 性能监控
• 资产指纹

资产指纹

通过该功能查看各主机的开放端口，运行进程，系统用户，定时任务，系统服务，系统软件等详情。

点击页面中各标签可以查看相应资产数据，目前支持下列资产数据:

• 容器
• 开放端口
• 运行进程
• 系统用户
• 定时任务
• 系统服务
• 系统软件
• 系统完整性校验
• 应用
• 内核模块

点击各行的"主机名称"可以跳转到相应主机的详情页，查看此主机资产指纹数据

容器集群

展示已经接入容器集群防护能力的容器集群信息。

主机和容器防护

入侵检测

告警列表

告警列表可以查看当前环境内存在的安全告警列表。

点击"摘要"可以查看告警关键摘要信息。

点击”处理”可以进行告警的处理，目前处理操作支持"加白"，"已人工处理"和“误报”。

点击各行的“影响资产”可以跳转到关联的主机详情页面查看相关数据。

白名单

主机和容器防护告警的白名单。

风险防范

漏洞列表

漏洞列表页面可以查看当前环境内存在的安全漏洞风险，默认只显示高风险漏洞的信息，用户通过立即检查来检测环境中最新存在的安全漏洞。

在漏洞列表数据右侧点击详情可以查看漏洞信息以及影响的资产列表。

漏洞信息也可以通过处置与忽略进行标记。

基线检查

基线检查页面可以查看当前环境内存在的安全基线风险，可以通过立即检查来查看环境中最新存在的安全基线问题。

在基线列表右侧点击详情可以看到相应的基线详情。

点击加白名单可以进行加白操作。

应用运行时安全防护

运行状态

展示运行时安全组件的覆盖情况。

配置管理

进行运行时安全组件的配置管理。

入侵检测

告警列表

运行时安全检测发现的安全告警展示。

白名单

运行时安全检测告警过滤白名单规则展示。

容器集群防护

入侵检测

告警列表

容器集群防护发现的安全告警展示。

白名单

容器集群防护告警过滤白名单规则展示。

病毒扫描

病毒扫描

显示检测到的病毒文件信息。

白名单

对病毒检测结果进行过滤的白名单列表。

系统管理

任务管理

该功能主要用于重启客户端、同步配置等管理。

点击”新建任务“，可以进行任务的配置。

除"重启客户端"类型任务外，其他类型任务需要先在"组件管理"和"组件策略"界面进行Agent和插件的配置，才能进行相应操作。

组件管理

用于对Agent和相应插件进行升级配置的管理。

点击”新建组件“可以选择组件的类型是Agent还是插件，然后进行相应配置。

点击发布版本按钮，可以通过上传对应架构与发行版的组件文件生成版本实例。

组件策略

通过管理组件策略，调整Agent实际生效的组件版本，进而实现更新、卸载等操作。

点击新建策略，通过选择组件名称及相匹配的版本，将对应策略添加至策略组中。

通过屏蔽规则使得部分主机不加载(生效)对应组件。

安装配置

可以通过部署帮助来实现Agent的部署和卸载。

用户管理

可以在用户管理中对Elkeid Console进行管理，如修改密码，新增与删除用户。

点击"新增用户"可以设置增加新的用户，设置密码的时候请注意密码强度的要求。

容器集群配置

进行容器集群防护能力接入的配置。

通知管理

告警通知以及过期通知的配置管理，支持飞书，钉钉等。

系统监控

后端监控

查看后端服务所在主机的CPU，内存，磁盘和网络流量等使用情况。

后端服务

查看后端各服务模块的QPS，CPU和内存的使用情况。

服务告警

显示最近1小时/24小时发现的服务异常告警。

Elkeid HUB 社区版快速上手教程

本教程的前置条件

在开始本教程之前，请检查：

• 已经按照部署文档,使用elkeidup正确部署了 HUB。
• 至少有一个可以使用的数据输入源（Input）和数据存储的输出源（Output）。在HIDS场景中，数据输入源为 AgentCenter 配置文件中指定的 kafka ，输出源可以使用 ES 或 Kafka ，本教程中以 ES 为例。

e.g. 社区版默认已配置输入源(hids,k8s)可在前端看到,方便测试使用

Step 1. 访问并登录前端

使用部署前端机器的 IP 即可访问前端，登录信息为elkeidup部署创建的用户名和密码。

Step 2. 编写策略

基本概念介绍

RuleSet是HUB实现检测/响应的核心部分，可以根据业务需要对输入的数据进行检测响应，配合插件还能实现报警的直接推送或对消息的进一步处理。因此如果有额外的业务需要可能需要自行编写部分规则。

RuleSet是通过XML格式来描述的规则集，RuleSet分为两种类型rule 和 whitelist。rule为如果检测到会继续向后传递，whitelist则为检测到不向后传递，向后传递相当于检出，因此whitelist一般用于白名单。

Ruleset里可以包含一条或多条rule，多个rule之间的关系是'或'的关系，即如果一条数据可以同时命中多条rule。

使用前端编写规则

进入规则页->规则集页面，可以看到当前收藏的RuleSet和全部RuleSet。

当Type为Rule时会出现未检测时丢弃字段，意味未检测到是否丢弃，默认是True即为未检测到即丢弃，不向下传递，在这里我们设为True。创建完成后，在创建好的条目上点击规则按钮，进入该Ruleset详情。在RuleSet详情中点击新建会弹出表单编辑器。

HUB已经默认开放了数十条规则,可以查看已经编写的规则,进行相关策略编写

也可以根据Elkeid HUB 社区版使用手册 进行编写.编写完成后,可以在项目页新建project,将编写好的规则与输入输出进行组合.

下图即为hids告警处理的过程:数据按照dsl的顺序,依次经过RULESET.hids_detect、RULESET.hids_filter等规则进行处理,最后再通过RULESET.push_hids_alert推送到CWPP console.

完成以上步骤后，进入规则发布页面，会显示出刚才修改的全部内容，每一个条目对应着一个组件修改，点击 Diff 可以查看修改详情。检查无误后，点击提交，将变更下发到HUB集群。

任务提交后，会自动跳转到任务详情页面，显示当前任务执行进度。

配置下发完成后，需要启动刚才新建的两个项目，进入规则发布->项目操作页面，分别启动全部已有的 项目。

进阶操作

配置ES Index查看报警

此步骤适用于OutputType使用ES的用户，Kafka用户可以自行配置。

1. 在输出页配置es类型的输出,可开启AddTimestamp,方便在kibana页面配置相关索引

2. 编辑hids项目,加入刚编些好的es输出

3. 提交变更

4. 首先进入ES 的 stack management，选择kibana 的index patterns，点击 create index patten

5. 输入之前填入的ES output index name，以星号 * 作为后缀，这里以默认推荐优先，分别为alert 或者 raw_alert
   • 如果这时index中存在数据（即已经触发过告警流），那么应能正确匹配到alert或者 raw_alert index

6. 配置时间字段
   • 如果这时index中存在数据（即已经触发过告警流），那么应能正确匹配到timestamp字段，否则此处无下拉选择框

7. 去浏览数据
   • 进入discover 看板，选择刚才创建的 alert* 看板，调整右侧时间，即可看到告警

示例

sqlmap检测规则编写

在本教程中，会尝试在前端中编写一条规则，为检查执行sqlmap命令的规则.

在该检测场景中，我们只需要关注Execve相关的信息，因此我添加了data_type为59的过滤字段，因此该规则只会对data_type为59的数据进行处理。之后我添加了一个CheckNode，检查数据中argv字段中是否包含'sqlmap'，编写完的效果如下：

可以看到分别设置了三个CheckNode来进行检测，一是直接检测argv中是否包含sqlmap，二是检测exe字段是否包含python，三是使用正则来进行匹配是否为单独的word，当这三个同时满足，就会触发报警，编写好后，点击保存。

我们单独为该测试规则建立一个Output和Project，如下图所示：

进入测试环境执行sqlmap相关指令，在kibana中添加对应的index pattern，稍微等待一会就可以找到对应的报警结果。

可以看到出现了报警。

推送飞书插件编写

规则写完了，如果我想在发生该事件的时候飞书提醒我该如何实现呢？RuleSet并不支持这项功能，此时可以通过编写插件来实现。

创建并使用Python 插件的步骤如下：

1. 点击创建按钮

2. 按照需求，填写信息

3. 点击确认，完成创建

4. 编辑插件

默认为只读状态，需要点击编辑才能进行编辑

编辑后点击保存

5. 在Rule中添加action

6. 同策略发布相同，在策略发布界面发布策略

这样每当这个rule的条件被触发，就会调用这个插件进行报警。

Elkeid CWPP 应用运行时防护 （RASP） 使用指南

本指南涵盖以下功能：

• 通过 CWPP 对 应用运行时 组件进行运维。
• 控制 RASP 植入探针到目标进程中，完成运行时行为采集。
• 植入配置
• 阻断/过滤 配置
• 查看 CWPP 的告警事件。

安装/更新 RASP 组件

1. 确保组件列表中包含 rasp 组件。

如果没有 rasp 组件，需要新建组件，组件名为 rasp。

注意！由于 Agent 机制，插件名称与插件二进制名称应该一致。

发布版本，上传 tar.gz 格式的压缩包。 请使用 1.9.1.* 版本的插件。 压缩包地址：bytedance/Elkeid: releases

2. 确保组件策略中包含 rasp 组件

3. 同步策略到机器。

运行状态

部署 RASP 组件后，RASP 将会自动分析机器进程，对符合植入探针条件的进程信息将会上报到 运行状态。 右侧详情链接支持查看进程其他信息

配置

配置哪些进程将会开启 RASP 保护

点击新建配置 每条配置的各表单项目间为与的关系 每条配置间为或的关系

每条配置还可以配置阻断与过滤

• 阻断：对某个Hook函数的某个参数进行正则表达式匹配
  • 正则表达式匹配到时，函数抛出异常阻断该函数运行。
  • 正则表达式没有匹配到时，函数正常运行。
• 过滤：对某个Hook函数的参数进行正则表达式匹配
  • 包含：只上报匹配到的参数数据
  • 不包含：只上报匹配到以外的参数数据

入侵检测

RASP 探针植入目标进程后，将会持续上报应用行为，发现异常行为后将会产生事件与告警。

• 右侧告警数据可以检查参数详情与调用栈

Elkeid 数据接入

ElkeidData.xlsx

采集字段列表

数据类型分类

字段对应中文描述

Elkeid 原始数据接入指南

适用版本

Elkeid 社区版v1.9.1

综述

Elkeid为了优化数据流压力和追求数据流灵活性，选用PB格式来传递数据。PB在kafka中存储为 byte[] 形式的消息，因此在消费数据后需要按照protobuf的形式进行解压，否则看到将是乱码字段。因此对于一般情况下使用来说，转移成JSON 数据流会更符合大数据套件的数据接入情况。这里提供接受Elkeid数据流并转换成JSON数据流的两种样例，供使用者参考。

PB Schema

如下即为 HIDS的 PB的protobuf schema，需要注意的是Item类为了灵活的支持多种版本的数据更新，Map中通过K-V的方式包含了基本全部类型的数据格式。因此单独PB后需要针对Map遍历提取K-V对放入全体数据后才能较好的支持进入大数据套件。

syntax = "proto3";
option go_package = "hids_pb";
package hids_pb;

//server -> bmq
message MQData{
  int32 data_type = 1;
  int64 timestamp = 2;
  Item body = 3;

  string agent_id = 4;
  string in_ipv4_list = 5;
  string ex_ipv4_list = 6;
  string in_ipv6_list = 7;
  string ex_ipv6_list = 8;
  string hostname = 9;
  string version = 10;
  string product = 11;

  int64  time_pkg = 12;
  string psm_name = 13;
  string psm_path = 14;
  string tags = 15;
}

message Item{
  map<string,string> fields = 1;
}

MQData.proto

Map内的数据字段请参考 Elkeid数据说明

简便方式（Elkeid-HUB）

Elkeid原生自带基于golang的PB数据解析，因此可以通过利用Elkeid-HUB进行数据的解压和ETL操作。

配置输出端

首先找到对应的输出端，Elkeid原生支持将PB的数据流转换成JSON并以ES/Kafka等方式进行输出，此处以kafka为例，填入所需的Kafka Bootstrap Server 地址（含端口）和topic名即可。此处建议打开AddTimestamp追加输出HUB的时间（将Unix系统时间转换成时间戳格式）

配置Project

在project中新增project，并按如下新增到Kafka的输出。注意OUTPUT后的值要和上面配置输出端的OutputID 保持一致。配之后点击Confirm 保存

INPUT.hids --> OUTPUT.your_kafka

启动

点击 Rule Publish 提交更改。等待任务完成后点击Project Operation 将任务启动即可。之后即可在所配制的Kafka topic中看到JSON格式的数据字段

注意如果用户希望直接将HIDS数据写入ES，请参考上述 Elkeid HUB 配置

如果用户利用Logstash进行日志采集，可以参考Logstash Protobuf Plugin进行配置。

Consumer 样例

这里以Python Kafka 套件消费数据为主进行展示

需要 kafka-python 和 protobuf 两个 python3 lib

安装protoc

首先需要安装protobuf的 compiler，主要用于生成 MQData_pb2.py 这个文件。用户可以使用我们直接提供的文件，或者自行编译。这里以 linux 为例，执行下面语句即可安装3.14的 protoc

PROTOC_ZIP=protoc-3.14.0-linux-x86_64.zip
curl -OL https://github.com/protocolbuffers/protobuf/releases/download/v3.14.0/$PROTOC_ZIP
sudo unzip -o $PROTOC_ZIP -d /usr/local bin/protoc
sudo unzip -o $PROTOC_ZIP -d /usr/local 'include/*'
rm -f $PROTOC_ZIP

将上面的PB Schema进行编译

protoc -I=. --python_out=. ./MQData.proto

会生成 MQData_pb2.py 文件，将这个文件放到您的项目中。

MQData_pb2.py

编写consumer

因为Elkeid PB 将主要数据放入 body.field 这个map中，因此需要对数据进行拍平，生成原生的一级数据结构。这里提供一个 decoder 供Kafka Serializer使用

#!/usr/bin/python3

# decoder of Elkeid PB, input string and will dump json for you.
def pbDecoder(value):
    ret = {}
    aMQData = MQData.MQData();
    aMQData.ParseFromString(value)

    # common part of message
    ret["data_type"] = str(aMQData.data_type)
    ret["timestamp"] = str(aMQData.timestamp)
    ret["agent_id"] = aMQData.agent_id
    ret["in_ipv4_list"] = aMQData.in_ipv4_list
    ret["ex_ipv4_list"] = aMQData.ex_ipv4_list
    ret["in_ipv6_list"] = aMQData.in_ipv6_list
    ret["ex_ipv6_list"] = aMQData.ex_ipv6_list
    ret["hostname"] = aMQData.hostname
    ret["version"] = aMQData.version
    ret["product"] = aMQData.product
    ret["time_pkg"] = str(aMQData.time_pkg)
    ret["psm_name"] = aMQData.psm_name
    ret["psm_path"] = aMQData.psm_path
    ret["tags"] = aMQData.tags

    # major data part of message
    for key in aMQData.body.fields:
        ret[key] =  aMQData.body.fields[key]
    
    return json.dumps(ret)

然后创建 Kafka consumer，将上述的decoder作为 kafka 的 value_deserializer 传递进去。Elkeid的默认topic为 hids_svr

#!/usr/bin/python3
from kafka import KafkaConsumer
import MQData_pb2 as MQData # 刚刚编译的那个class文件
import json

# decoder of Elkeid PB, input string and will dump json for you.
def pbDecoder(value):
    ret = {}
    aMQData = MQData.MQData();
    aMQData.ParseFromString(value)

    # common part of message
    ret["data_type"] = str(aMQData.data_type)
    ret["timestamp"] = str(aMQData.timestamp)
    ret["agent_id"] = aMQData.agent_id
    ret["in_ipv4_list"] = aMQData.in_ipv4_list
    ret["ex_ipv4_list"] = aMQData.ex_ipv4_list
    ret["in_ipv6_list"] = aMQData.in_ipv6_list
    ret["ex_ipv6_list"] = aMQData.ex_ipv6_list
    ret["hostname"] = aMQData.hostname
    ret["version"] = aMQData.version
    ret["product"] = aMQData.product
    ret["time_pkg"] = str(aMQData.time_pkg)
    ret["psm_name"] = aMQData.psm_name
    ret["psm_path"] = aMQData.psm_path
    ret["tags"] = aMQData.tags

    # major data part of message
    for key in aMQData.body.fields:
        ret[key] =  aMQData.body.fields[key]
    
    return json.dumps(ret)


# To consume latest messages and auto-commit offsets
consumer = KafkaConsumer('hids_svr',
                         group_id='test',
                         auto_offset_reset='latest',
                         bootstrap_servers=['10.2.0.67:9092', '10.2.0.233:9092', '10.2.0.92:9092'],
                         value_deserializer = lambda m: pbDecoder(m))

# Print all message in JSON format
# this is the part that you need to code to your job
for message in consumer:
    # do something with message.value
    print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
                                          message.offset, message.key,
                                          message.value))

数据持久化建议

整体建议

数据在获取到Json格式后即可根据Elkeid 数据说明 的字段进行相应的存储工作。一般情况对于HIDS数据持久化存储需求应该符合使用方公司自身的数据中台规范需求。这里仅仅提供建议，而没有标准化方法可以直接使用。

大宽表

对于机器数量少于1w台的情况，我们建议用户将Elkeid数据以大宽表的形式存储。也就是说将所有数据用共同的表schema进行存储。我们建议用户将Elkeid数据按照时间进行分区，方便后续进行过期（TTL）设置。

建表DDL请参考如下：

CREATE TABLE `YOUR_DBNAME`.`hids_rawdata`(
    `agent_id` STRING COMMENT 'HIDS 的 agent_id',
    `argv` STRING COMMENT '进程命令行',
    `comm` STRING COMMENT '进程命令行首项',
    `data_type` STRING COMMENT '数据类型',
    `dip` STRING COMMENT '链接目的IP',
    `dport` STRING COMMENT '链接目的端口',
    `exe` STRING COMMENT '命令首项的文件路径',
    `exe_hash` STRING COMMENT '命令首项的文件哈希',
    `hostname` STRING COMMENT '主机名',
    `ld_preload` STRING COMMENT '执行时加载的额外so文件',
    `nodename` STRING COMMENT '节点名',
    `pgid` STRING COMMENT '进程组ID',
    `pid` STRING COMMENT '进程ID',
    `pid_tree` STRING COMMENT '进程链条',
    `ppid` STRING COMMENT '父进程ID',
    `run_path` STRING COMMENT '命令执行时所在的位置',
    `res` STRING COMMENT '返回值',
    `sa_family` STRING COMMENT '链接的协议组',
    `sessionid` STRING COMMENT 'linux进程会话ID',
    `sip` STRING COMMENT '链接源IP',
    `socket_pid` STRING COMMENT '进程链上存在socket的进程ID',
    `sport` STRING COMMENT '链接源端口',
    `ssh` STRING COMMENT '进程创建时的SSH环境变量信息',
    `stdin` STRING COMMENT '标准输入',
    `stdout` STRING COMMENT '标准输出',
    `tgid` STRING COMMENT '线程组ID',
    `time` STRING COMMENT '进程发生时间',
    `tty` STRING COMMENT 'tty字符终端',
    `uid` STRING COMMENT 'linux上用户ID',
    `username` STRING COMMENT 'linux上用户名',
    `query` STRING COMMENT 'DNS 查询语句',
    `qr` STRING COMMENT 'DNS 查询qr',
    `opcode` STRING COMMENT 'DNS 查询opcode',
    `rcode` STRING COMMENT 'DNS 查询rcode',
    `file_path` STRING COMMENT '创建文件路径',
    `ptrace_request` STRING COMMENT 'ptrace时的请求ID',
    `target_pid` STRING COMMENT 'ptrace时被调试进程ID',
    `lkm_file` STRING COMMENT '内核模块名（老）',
    `old_uid` STRING COMMENT '提权前linux用户ID',
    `module_name` STRING COMMENT '内核模块名',
    `hidden` STRING COMMENT '是否为隐藏内核模块',
    `syscall_number` STRING COMMENT '被Hook的syscall ID',
    `interrupt_number` STRING COMMENT '被Hook的中断 ID',
    `data` STRING COMMENT '数据（某些场景使用）',
    `socket_argv` STRING COMMENT '进程链上存在socket的进程命令行',
    `ppid_argv` STRING COMMENT '父进程命令行',
    `pgid_argv` STRING COMMENT '进程组长命令行',
    `connect_type` STRING COMMENT '链接的类型',
    `oldname` STRING COMMENT '原用户名',
    `newname` STRING COMMENT '新用户名',
    `mprotect_prot` STRING COMMENT 'mprotect_prot',
    `owner_pid` STRING COMMENT 'owner_pid',
    `owner_file` STRING COMMENT 'owner_file',
    `vm_file` STRING COMMENT 'vm_file',
    `status` STRING COMMENT 'ssh登陆状态',
    `types` STRING COMMENT 'ssh登陆类型',
    `invalid` STRING COMMENT 'ssh用户是否存在',
    `rawlog` STRING COMMENT 'ssh登陆日志',
    `authorized` STRING COMMENT 'kerberos登陆的linux用户名',
    `principal` STRING COMMENT 'kerberos登陆的kerb用户名',
    `user` STRING COMMENT 'ssh登陆的linux用户名',
    `ex_ipv4_list` STRING COMMENT '外网IPv4列表',
    `ex_ipv6_list` STRING COMMENT '外网IPv6列表',
    `in_ipv4_list` STRING COMMENT '内网IPv4列表',
    `in_ipv6_list` STRING COMMENT '内网IPv6列表',
    `version` STRING COMMENT 'HIDS版本号',
    `sid` STRING COMMENT '进程会话sid',
    `addr` STRING COMMENT 'addr',
    `target_argv` STRING COMMENT 'ptrace被调试进程的argv',
    `old_username` STRING COMMENT '提权前用户名',
    `lkm_hash` STRING COMMENT '内核模块哈希 ',
    `timestamp` STRING COMMENT '时间戳',
    `time_pkg` STRING COMMENT '上传时间',
    `tags` STRING COMMENT '机器标签',
    `ko_file` STRING COMMENT '新增内核模块名',
    `flags` STRING COMMENT 'memfd - flags',
    `fdname` STRING COMMENT 'memfd - flags',
    `new_name` STRING COMMENT 'mv/link/prctl 新文件名',
    `old_name` STRING COMMENT 'mv/link 旧文件名',
    `option` STRING COMMENT 'prctl',
    `pns` STRING COMMENT 'pid_namespace',
    `root_pns` STRING COMMENT 'root fs pid_namespace',
    `psm` STRING COMMENT 'pod psm',
    `pod_name` STRING COMMENT 'pod name',
    `sandbox_task_id` STRING COMMENT '沙箱任务id（含蜜罐）',
    `sandbox_task_type` STRING COMMENT '沙箱类型（含蜜罐）',
    `cpu` STRING COMMENT 'CPU Heartbeat',
    `io` STRING COMMENT 'IO Heartbeat',
    `kernel_version` STRING COMMENT 'kernel_version -- heartbeat',
    `memory` STRING COMMENT 'memory -- heartbeat',
    `net_type` STRING COMMENT 'net_type -- heartbeat',
    `platform` STRING COMMENT 'platform  -- heartbeat',
    `platform_version` STRING COMMENT 'platform_version -- heartbeat',
    `plugins` STRING COMMENT 'plugins  -- heartbeat',
    `slab` STRING COMMENT 'slab  -- heartbeat',
    `exe_size` STRING COMMENT 'exe 大小，静态检测',
    `psm_name` STRING COMMENT 'psm_name',
    `psm_path` STRING COMMENT 'psm_path',
    `product` STRING COMMENT '产品类型',
    `mode` STRING COMMENT 'mode ',
    `file` STRING COMMENT 'file ',
    `sec` STRING COMMENT 'sec ',
    `nsec` STRING COMMENT 'nsec ',
    `sig` STRING COMMENT 'sig',
    `dev` STRING COMMENT 'dev',
    `fstype` STRING COMMENT 'fstype',
    `wait` STRING COMMENT 'wait',
    `sb_id` STRING COMMENT 'sb_id',
    `manufacturer` STRING COMMENT 'manufacturer',
    `serial` STRING COMMENT 'serial',
    `action` STRING COMMENT 'action',
    `dpid` STRING COMMENT 'dpid',
    `dpid_argv` STRING COMMENT 'dpid_argv',
    `cred` STRING COMMENT 'cred',
    `dcred` STRING COMMENT 'dcred',
    `name` STRING COMMENT 'name',
    `idc` STRING COMMENT 'idc',
    `platform_family` STRING COMMENT 'platform_family',
    `net_mode` STRING COMMENT 'net mode ',
    `boot_at` STRING COMMENT 'boot_at',
    `started_at` STRING COMMENT 'started_at'
) COMMENT '全量数据' PARTITIONED BY (
    `date` STRING COMMENT 'date', 
    `hour` STRING COMMENT 'hour'
) ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' WITH SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat' TBLPROPERTIES (
    'transient_lastDdlTime' = '1649647413',
    'primary_key' = ''
)

这里需要注意的是，Elkeid数据中并不存在直接的date / hour字段 因此可以先对Elkeid数据所在HDFS分区进行外表建立 然后从外表将数据导入内表分区时手动设置partition

1. ALTER TABLE hids_rawdata ADD PARTITION (date='2008-08-08', hour='08')
2. LOAD DATA INPATH '/user/hids_data' INTO TABLE hids_rawdata PARTITION(dt='2008-08- 08', hour='08');

持久化存储预估资源

1 台健康Elkeid Agent在普遍业务场景下每小时产出数据 20MB左右

对于需要日志长期持久化的场景，这里给出存储需求列表供参考

开源策略说明

HIDS开源策略列表

RASP开源策略列表

K8S开源策略列表

k8s 开源策略列表

K8S开源策略编写说明

数据源

K8S策略基于K8S Audit Logs数据，具体的Audit Policy可以在平台中下载。在Console中配置好后，数据会经Agent Center上开启的Webhook写入到Kafka的k8stopic。也可以直接使用HUB中自带的k8s INPUT作为数据源。

开源策略说明

Project

在v1.9.1社区版中，我们编写了一部分示例策略用于开源，对应的HUB Project为kube_example和kube_workload。kube_example存放的策略为基础策略，kube_workload存放的为需要对数据进行处理后再进行检测的策略。

• kube_example

  INPUT.k8s --> RULESET.kube_detect
  RULESET.kube_detect --> RULESET.kube_alert_info
  RULESET.kube_alert_info --> RULESET.kube_add_cluster_info
  RULESET.kube_add_cluster_info --> RULESET.kube_push_alert

• kube_workload

  INPUT.k8s --> RULESET.kube_workloads
  RULESET.kube_workloads --> RULESET.kube_filter_workloads
  RULESET.kube_filter_workloads --> RULESET.kube_handle_workloads
  RULESET.kube_handle_workloads --> RULESET.kube_detect_workloads
  RULESET.kube_detect_workloads  --> RULESET.kube_alert_info
  RULESET.kube_alert_info --> RULESET.kube_add_cluster_info
  RULESET.kube_add_cluster_info --> RULESET.kube_push_alert

Ruleset

下面为一些调用HUB内置插件的规则集进行补充说明，其余规则可以直接在HUB前端进行查看。

kube_alert_info规则集对检出的告警添加告警数据字段，同时调用Kube_add_info插件添加告警的基础信息。该插件为HUB内置Modify插件，因此可以按需调用。

kube_add_cluster_info规则集调用Manager接口通过集群id获取集群信息，该流程通过调用KubeAddClusterInfo插件实现。该插件为HUB内置Modify插件。

kube_push_alert规则集调用Manager接口推送告警，该流程通过调用KubePushMsgToLeader插件实现。该插件为HUB内置Action插件。

下图为告警内容说明： workload相关检测策略通过Python Plugin进行实现，该插件于kube_handle_workloads中调用。

编写建议

在编写其余告警策略时，需要分别调用kube_alert_info、kube_add_cluster_info、kube_push_alert进行告警的信息填充，集群信息添加，告警的推送。如果新增告警类型，需要在kube_alert_info中进行添加，补充相关字段。

社区版与企业版能力差异

Elkeid Console社区版v1.9.1和企业版能力对比

Elkeid HUB 社区版和企业版能力对比

Elkeid

Elkeid®: Elkeid 是一款可以满足主机，容器与容器集群，Serverless 等多种工作负载安全需求的开源解决方案，源于字节跳动内部最佳实践。

源码仓库: github.com/bytedance/Elkeid

Elkeid 由两部分组成：

• 端上：在Linux主机中完成信息采集，与后端进行通信。
• 后端：处理端上通信，管理各个 Agent 与插件，提供管理接口和操作界面，对采集数据进行分析。

端上能力

插件

后端能力

Agent

关于 Elkeid Agent

Agent提供端上组件的基本能力支撑，包括数据通信、资源监控、组件版本控制、文件传输、机器基础信息采集等。

Agent本身不提供安全能力，作为一个插件底座以系统服务的方式运行。各类功能插件的策略存放于服务器端的配置，Agent接收到相应的控制指令及配置后对自身及插件进行开启、关闭、升级等操作。

Agent与Server之间采用bi-stream gRPC进行通信，基于自签名证书开启双向TLS验证，保障信道安全。其中，Agent -> Server 方向信息流动称为数据流，Server -> Agent 方向信息流动一般为控制流，使用protobuf的不同message类型。Agent本身支持客户端侧服务发现，也支持跨Region级别的通信配置，实现一个Agent包在多个网络隔离的环境下进行安装，基于底层一个TCP连接，在上层实现了Transfer与FileOp两种数据传输服务，支撑了插件本身的数据上报与Host中的文件交互。

Plugins作为安全能力插件，与Agent的进程关系一般为“父——子”进程。以两个pipe作为跨进程通信方式，plugins lib提供了Go与Rust的两个插件库，负责插件端信息的编码与发送。值得一提的是，插件发送数据后，会被编码为Protobuf二进制数据，Agent接收到后无需二次解编码，再其外层拼接好Header特征数据，直接传输给Server，一般情况下Server也无需解码，直接传输至后续数据流中，使用时进行解码，一定程度上降低了数据传输中多次编解码造成的额外性能开销。

Agent采用Go实现，在Linux下，通过systemd作为守护方式，受cgroup限制控制资源使用，支持aarch64与x86-64架构，最终编译、打包为deb与rpm包分发，格式均符合systemd及Debian、RHEL规范，可以直接提供至对应的软件仓库中进行后续版本维护。在后续版本中，将会发布用于Windows平台下的Agent。

运行时要求

Agent及Plugin提供的大部分功能需要以root权限运行在宿主机(Host)层面，在权限受限的容器中，部分功能可能会存在异常。

快速开始

通过 elkeidup 的完整部署，可以直接得到用于Debian/RHEL系列发行版的安装包，并按照 Elkeid Console - 安装配置 界面的命令进行安装部署。

手动编译

环境要求

• Go >= 1.18
• nFPM
• 成功部署的 Server (包含所有组件)

确认相关配置

• 需要确保 transport/connection 目录下的 ca.crt、client.key、client.crt 三个文件与Agent Center conf 目录下的同名文件保持一致。
• 需要确保 transport/connection/product.go 文件中的参数都配置妥当：
  • 如果是手动部署的Server：
    • serviceDiscoveryHost["default"] 需被赋值为 ServiceDiscovery 服务或代理服务的内网监听地址与端口，例如：serviceDiscoveryHost["default"] = "192.168.0.1:8088"
    • privateHost["default"] 需被赋值为 AgentCenter 服务或代理服务的内网监听地址与端口，例如：privateHost["default"] = "192.168.0.1:6751"
    • 如有Server的公网接入点，publicHost["default"] 需被赋值为 AgentCenter 服务或代理服务的外网监听地址与端口，例如：publicHost["default"]="203.0.113.1:6751"
  • 如果是通过 elkeidup 部署的Server，可以根据部署Server机器的 ~/.elkeidup/elkeidup_config.yaml 文件获得对应配置：
    • 在配置文件中找到 Nginx 服务的IP，具体的配置项为 nginx.sshhost[0].host
    • 在配置文件中找到 ServiceDiscovery 服务的IP，具体的配置项为 sd.sshhost[0].host
    • serviceDiscoveryHost["default"] 需被赋值为 ServiceDiscovery 服务的IP，并将端口号设置为8088，例如：serviceDiscoveryHost["default"] = "192.168.0.1:8088"
    • privateHost["default"] 需被赋值为 Nginx 服务的IP，并将端口号设置为8090，例如：privateHost["default"] = "192.168.0.1:8090"

编译

在Agent根目录，执行：

BUILD_VERSION=1.7.0.24 bash build.sh

在编译过程中，脚本会读取 BUILD_VERSION 环境变量设置版本信息，可根据实际需要进行修改。

编译成功后，在根目录的 output 目录下，应该可以看到2个deb与2个rpm文件，它们分别对应不同的系统架构。

版本升级

1. 如果没有创建过客户端类型的组件，请在 Elkeid Console-组件管理 界面新建对应组件。
2. 在 Elkeid Console - 组件管理 界面，找到“elkeid-agent”条目，点击右侧“发布版本”，填写版本信息并上传对应平台与架构的文件，点击确认。
3. 在 Elkeid Console - 组件策略 界面，(如有)删除旧的“elkeid-agent”版本策略，点击“新建策略”，选中刚刚发布的版本，点击确认。后续新安装的Agent均会自升级到最新版本。
4. 在 Elkeid Console - 任务管理 界面，点击“新建任务”，选择全部主机，点击下一步，选择“同步配置”任务类型，点击确认。随后，在此页面找到刚刚创建的任务，点击运行，即可对存量旧版本Agent进行升级。

License

Elkeid Agent is distributed under the Apache-2.0 license.

Server

源码：https://github.com/bytedance/Elkeid/blob/main/server

后台架构图

概述

Elkeid 后台大体包含5个模块：

1. AgentCenter(AC)，负责与Agent进行通信，采集Agent数据并简单处理后汇总到消息队列集群，同时也负责对Agent进行管理包括Agent的升级，配置修改，任务下发等。同时AC也对外提供HTTP接口，Manager通过这些HTTP接口实现对AC和Agent的管理和监控。
2. ServiceDiscovery(SD)，后台中的各个服务模块都需要向SD中心定时注册、同步服务信息，从而保证各个服务模块中的实例相互可见，便于直接通信。由于SD维护了各个注册服务的状态信息，所以当服务使用方在请求服务发现时，SD会进行负载均衡。比如Agent请求获取AC实例列表，SD直接返回负载压力最小的AC实例。
3. Manager，负责对整个后台进行管理并提供相关的查询、管理接口。包括管理AC集群，监控AC状态，控制AC服务相关参数，并通过AC管理所有的Agent，收集Agent运行状态，往Agent下发任务，同时manager也管理实时和离线计算集群。
4. Elkeid Console: Elkeid 前端部分。
5. Elkeid HUB :Elkeid HIDS RuleEngine。

简单来说就是AgentCenter收集Agent数据，Elkeid HUB对这些数据进行分析和检测，Manager管理着AgentCenter和这些计算模块，ServiceDiscovery把这些所有的服务、节点都串联了起来，通过Elkeid Console可查看告警和资产数据等。

功能特点

• 百万级Agent的后台架构解决方案
• 分布式，去中心化，集群高可用
• 部署简单，依赖少，便于维护

完整部署文档

• 通过ElkeidUp部署

编译

1. AgentCenter(AC): 在Elkeid/server/agent_center目录执行 ./build.sh，将会在在output目录下生成产物bin.tar.gz。
2. ServiceDiscovery(SD)：在Elkeid/server/service_discovery目录执行 ./build.sh将会在在output目录下生成产物bin.tar.gz。
3. Manager：在Elkeid/server/manager目录执行 ./build.sh，将会在在output目录下生成产物bin.tar.gz。

版本升级

参照从源码构建 Elkeid CWPP的后端部分，来部署或升级即可。

Console User Guide

• ELkeid Console User Guide

License

Elkeid Server are distributed under the Apache-2.0 license.

About Elkeid(AgentSmith-HIDS) Driver

Elkeid Driver 主要是为信息安全需求而设计的。

Elkeid Driver 主要通过 Kprobe Hook Kernel Funcion 来提供丰富而准确的数据收集功能，包括内核级进程执行探测，特权升级监控，网络审计等等。 并且支持 Linux namespace，因此对容器监控有着很好的实现。与传统的UserSpace HIDS相比，Elkeid由于驱动的存在提供了更全面的信息，并提高了性能。

凭借其出色的数据收集能力，Elkeid Driver还可以支持沙盒，蜜罐和审计等需求。

如果发现 Bug 欢迎提 Issue 或 加入飞书公开群参与讨论。

快速尝试

首先需要安装Linux Headers，Linux Headers 的版本必须等于 uname -r

# clone and build
git clone https://github.com/bytedance/Elkeid.git
cd Elkeid/driver/LKM/
make clean && make
< CentOS only: run build script instead >
sh ./centos_build_ko.sh

# load and test (should run as root)
insmod hids_driver.ko
dmesg | tail -n 20
test/rst -q
< "CTRL + C" to quit >

# unload
rmmod hids_driver

我们提供部分预编译好的 Ko 文件

我们提供了一些预编译好的 Elkeid 内核模块，这些 Ko 包括了 debian,centos,ubuntu 等发行版的不同内核版本。

预编译好的 Ko 文件

Ko 文件列表 若不再列表内，或下载失败，请自行编译 ko

获取方式

如果所有链接都获取失败，则说明 预编译的 Ko中，不包含当前系统的内核版本所需的 Ko，需要自行编译

wget "http://lf26-elkeid.bytetos.com/obj/elkeid-download/ko/hids_driver_1.7.0.4_$(uname -r).ko"
# or
curl -O "http://lf26-elkeid.bytetos.com/obj/elkeid-download/ko/hids_driver_1.7.0.4_$(uname -r).ko"
# other cdn
## "http://lf3-elkeid.bytetos.com/obj/elkeid-download/ko/hids_driver_1.7.0.4_$(uname -r).ko"
## "http://lf6-elkeid.bytetos.com/obj/elkeid-download/ko/hids_driver_1.7.0.4_$(uname -r).ko"
## "http://lf9-elkeid.bytetos.com/obj/elkeid-download/ko/hids_driver_1.7.0.4_$(uname -r).ko"

内核模块的测试方法

可以通过 LTP 或者 Kasan这两个方法对内核模块进行测试.

这里提供 LTP测试用例 文件

关于Linux发行版的兼容性

关于ARM64 (AArch64)支持

• 支持

关于Linux Kernel Version兼容性

• Linux Kernel Version >= 2.6.32 && <= 5.14.X

关于容器信息采集的兼容性

Hook List

Anti Rootkit List

关于驱动数据传输

驱动数据协议

上述Hook点每命中一次均会生成一条日志记录，每条日志包含多个数据项，数据项之间使用'\x17'作为间隔符。数据部分通常由公共数据和私有数据组合而成，值得注意的是Anti-rootkit没有公共数据。

公共数据

-------------------------------------------------------------------------------
|1        |2  |3  |4  |5   |6   |7   |8  |9   |10      |11       |12 |13      |
-------------------------------------------------------------------------------
|data_type|uid|exe|pid|ppid|pgid|tgid|sid|comm|nodename|sessionid|pns|root_pns|
-------------------------------------------------------------------------------

Write Data (1)

-----------
|14   |15 | 
-----------
|file||buf|
-----------

Open Data (2)

---------------------
|14   |15  |16      | 
---------------------
|flags|mode|filename|
---------------------

Mprotect Data (10)

-----------------------------------------------------
|14           |15       |16        |17     |18      |
-----------------------------------------------------
|mprotect_prot|owner_pid|owner_file|vm_file|pid_tree|
-----------------------------------------------------

Nanosleep Data (35)

----------
|14 |15  |
----------
|sec|nsec|
----------

Connect Data (42)

-----------------------------------
|14       |15 |16   |17 |18   |19 |
-----------------------------------
|sa_family|dip|dport|sip|sport|res|
-----------------------------------

Accept Data (43)

-----------------------------------
|14       |15 |16   |17 |18   |19 |
-----------------------------------
|sa_family|dip|dport|sip|sport|res|
-----------------------------------

Bind Data (49)

-------------------------
|14       |15 |16   |17 |
-------------------------
|sa_family|sip|sport|res|
-------------------------

Execve Data (59)

-----------------------------------------------------------------------------------------------------
|14  |15      |16   |17    |18 |19   |20 |21   |22       |23      |24 |25        |26 |27        |28 |
-----------------------------------------------------------------------------------------------------
|argv|run_path|stdin|stdout|dip|dport|sip|sport|sa_family|pid_tree|tty|socket_pid|ssh|ld_preload|res|
-----------------------------------------------------------------------------------------------------

Note:

• socket_exe/dip/dport/sip/sport/sa_family 来自于进程所持fd信息

• ssh/ld_preload 来自于进程的环境变量信息

Process Exit Data (60)

该数据没有私有数据，仅有公共数据

Kill Data (62)

----------------
|14        |15 |
----------------
|target_pid|sig|
----------------

Rename Data (82)

--------------------------
|14      |15      |16    | 
--------------------------
|old_name|new_name|sb_id|
-------------------------

Link Data (86)

--------------------------
|14      |15      |16    | 
--------------------------
|old_name|new_name|sb_id|
-------------------------

Ptrace Data (101)

----------------------------------------------
|14            |15        |16  |17  |18      |
----------------------------------------------
|ptrace_request|target_pid|addr|data|pid_tree|
----------------------------------------------

Setsid Data (112)

该数据没有私有数据，仅有公共数据

Prctl Data (157)

_________________
|14    |15      | 
-----------------
|option|new_name|
-----------------

Mount Data (165)

_____________________________________
|14      |15 |16       |17    |18   | 
-------------------------------------
|pid_tree|dev|file_path|fstype|flags|
-------------------------------------

Tkill Data (200)

----------------
|14        |15 |
----------------
|target_pid|sig|
----------------

Exit Group Data (231)

该数据没有私有数据，仅有公共数据

memfd_create Data (356)

______________
|14    |15   | 
--------------
|fdname|flags|
--------------

Dns Query Data (601)

--------------------------------------------------
|14   |15       |16 |17   |18 |19   |20    |21   |
--------------------------------------------------
|query|sa_family|dip|dport|sip|sport|opcode|rcode|
--------------------------------------------------

Create File data (602)

----------------------------------------------------------
|14 	  |15 |16   |17 |18   |19       |20        |21   |
----------------------------------------------------------
|file_path|dip|dport|sip|sport|sa_family|socket_pid|sb_id|
---------------------------------------------------------

Load Module Data (603)

----------------------------
|14      |15      |16      | 
----------------------------
|ko_file|pid_tree|run_path|
----------------------------

Update Cred Data (604)

----------------------
|14      |15     |16 | 
----------------------
|pid_tree|old_uid|res|
----------------------

Unlink Data (605)

------
|14  |
------
|file|
------

Rmdir Data (606)

------
|14  |
------
|file|
------

call_usermodehelper_exec Data (607)

-------------------------
|1        |2  |3   |4   |
-------------------------
|data_type|exe|argv|wait|
-------------------------

File Write Data (608)

------------
|14  |15   |
------------
|file|sb_id|
------------
需要通过 Diver Filter 加入待观察列表，详情见 "关于 Driver Filter" 部分

File Read Data (609)

------------
|14  |15   |
------------
|file|sb_id|
------------
需要通过 Diver Filter 加入待观察列表，详情见 "关于 Driver Filter" 部分

USB Device Event Data (610)

-----------------------------------------
|14          |15          |16    |17    |
-----------------------------------------
|product_info|manufacturer|serial|action|
-----------------------------------------
action = 1 is USB_DEVICE_ADD
action = 2 is USB_DEVICE_REMOVE

Privilege Escalation (611)

------------------------------
|14   |15      |16    |17    |
------------------------------
|p_pid|pid_tree|p_cred|c_cred|
------------------------------
p_cred = uid|euid|suid|fsuid|gid|egid|sgid|fsgid
c_cred = uid|euid|suid|fsuid|gid|egid|sgid|fsgid

Proc File Hook (700)

-----------------------
|1        |2          |
-----------------------
|data_type|module_name|
-----------------------

Syscall Table Hook Data (701)

--------------------------------------
|1        |2          |3             |
--------------------------------------
|data_type|module_name|syscall_number|
--------------------------------------

Hidden Kernel Module Data (702)

-----------------------
|1        |2          |
-----------------------
|data_type|module_name|
-----------------------

Interrupt Table Hook Data (703)

----------------------------------------
|1        |2          |3               |
----------------------------------------
|data_type|module_name|interrupt_number|
----------------------------------------

关于 Driver Filter

Elkeid驱动程序支持白名单以过滤出不需要的数据。 我们提供两种类型的白名单，'exe'白名单和'argv'白名单。 'exe'白名单作用于execve /create filte/ dns query/connect hook，而'argv'白名单仅作用于execve hook 。 出于性能和稳定性方面的考虑，‘exe’和‘argv’白名单容量为64。

白名单的字符串驱动位于: /dev/hids_driver_allowlist

Filter define is:

#define ADD_EXECVE_EXE_SHITELIST 89         /* Y */
#define DEL_EXECVE_EXE_SHITELIST 70         /* F */
#define DEL_ALL_EXECVE_EXE_SHITELIST 119    /* w */
#define EXECVE_EXE_CHECK 121                /* y */
#define PRINT_ALL_ALLOWLIST 46              /* . */
#define ADD_EXECVE_ARGV_SHITELIST 109       /* m */
#define DEL_EXECVE_ARGV_SHITELIST 74        /* J */
#define DEL_ALL_EXECVE_ARGV_SHITELIST 117   /* u */
#define EXECVE_ARGV_CHECK 122               /* z */

#define ADD_WRITE_NOTIFI 87                 /* W */
#define DEL_WRITE_NOTIFI 120                /* v */
#define ADD_READ_NOTIFI 82                  /* R */
#define DEL_READ_NOTIFI 115                 /* s */
#define DEL_ALL_NOTIFI 65                   /* A */

关于Elkeid Driver 性能

Testing Environment(VM):

Testing Load:

Key kprobe Handler Testing Result(90s)

udp_recvmsg_handler 仅工作在端口为 53 或 5353的情况

测试原始数据:Benchmark Data

关于部署

可以使用DKMS或者提前编译好ko文件然后进行下发

• install driver: insmod hids_driver.ko
• remove driver: first you need kill userspace agent and rmmod hids_driver.ko

已知问题

• 内核模块 hook 点初始化失败 : do_init_module
  在一些老版本的 ubuntu/centos 内核中出现，dmesg 会有如下输出:
  do_init_module register_kprobe failed, returned -2.
  内核模块仍然可以使用，但没有 do_init_module 数据

License

Elkeid kernel module are distributed under the GNU GPLv2 license.

RASP

Runtime application self-protection (RASP) 是一种应用安全防御技术，通过对应用运行时植入探针来采集运行时关键信息，通过分析运行时行为产生告警，安全工程师可以高效的根据告警对入侵和攻击进行排查和修复。

Elkeid RASP

• 源码: bytedance/Elkeid: rasp

简介

• 支持进程运行时类型分析。
• 支持以下类型运行时的动态探针植入。
  • CPython
  • Golang
  • JVM
  • NodeJS
  • PHP
• 探针将会上报运行时 Hook 信息。
• 兼容 Elkeid 技术栈。

Install

• build manually: GUIDE
  1. CMake 3.17+
  2. GCC 8+
  3. MUSL toolcahin 1.2.2 (download via CDN: link)
  4. RUST toolchain 1.40+
  5. JDK 11+(for Java probe)
  6. Python2 + Python3 + pip + wheel + header files (for python probe)
  7. PHP header files
  8. make build and install

git submodule update --recursive --init

make -j$(nproc) build \
    STATIC=TRUE \
    PY_PREBUILT=TRUE \
    CC=/opt/x86_64-linux-musl-1.2.2/bin/x86_64-linux-musl-gcc \
    CXX=/opt/x86_64-linux-musl-1.2.2/bin/x86_64-linux-musl-g++ \
    LD=/opt/x86_64-linux-musl-1.2.2/bin/x86_64-linux-musl-ld \
    CARGO_TARGET_X86_64_UNKNOWN_LINUX_MUSL_LINKER=/opt/x86_64-linux-musl-1.2.2/bin/x86_64-linux-musl-ld \
    GNU_CC=/opt/gcc-10.4.0/bin/gcc \
    GNU_CXX=/opt/gcc-10.4.0/bin/g++ \
    PHP_HEADERS=/path/to/php-headers \
    PYTHON2_INCLUDE=/path/to/include/python2.7 \
    PYTHON3_INCLUDE=/path/to/include/python3 \
    VERSION=0.0.0.1

# for Console/Agent plugin：
sha256sum "rasp_${VERSION}.tar.gz" && sha256sum output/rasp

# for install on buding machine:
sudo make install

• build with docker:

curl -fsSL https://lf3-static.bytednsdoc.com/obj/eden-cn/laahweh7uhwbps/php-headers.tar.gz | tar -xz -C rasp/php

docker run --rm -v $(pwd):/Elkeid \
    -v /tmp/cache/gradle:/root/.gradle \
    -v /tmp/cache/librasp:/Elkeid/rasp/librasp/target \
    -v /tmp/cache/rasp_server:/Elkeid/rasp/rasp_server/target \
    -v /tmp/cache/plugin:/Elkeid/rasp/plugin/target \
    -e MAKEFLAGS="-j$(nproc)" hackerl/rasp-toolchain \
    make -C /Elkeid/rasp build \
    STATIC=TRUE \
    PY_PREBUILT=TRUE \
    CC=/opt/x86_64-linux-musl-1.2.2/bin/x86_64-linux-musl-gcc \
    CXX=/opt/x86_64-linux-musl-1.2.2/bin/x86_64-linux-musl-g++ \
    LD=/opt/x86_64-linux-musl-1.2.2/bin/x86_64-linux-musl-ld \
    CARGO_TARGET_X86_64_UNKNOWN_LINUX_MUSL_LINKER=/opt/x86_64-linux-musl-1.2.2/bin/x86_64-linux-musl-ld \
    GNU_CC=/opt/gcc-10.4.0/bin/gcc GNU_CXX=/opt/gcc-10.4.0/bin/g++ \
    PHP_HEADERS=/Elkeid/rasp/php/php-headers \
    PYTHON2_INCLUDE=/usr/local/include/python2.7 \
    PYTHON3_INCLUDE=/usr/local/include/python3 \
    VERSION=0.0.0.1

• for single process inject

sudo env RUST_LOG=<loglevel> /etc/elkeid/plugin/RASP/elkeid_rasp -p <pid>

License

Elkeid RASP are distributed under the Apache-2.0 license.

JVM

原理

• RASP 将会尝试启动 JVM 的 attach 功能，并将探针 jar 植入到 JVM 中。
• 植入的探针将会利用 Java 提供的 instrument 能力（文档：https://docs.oracle.com/javase/8/docs/api/java/lang/instrument/Instrumentation.html），对部分函数的字节码进行修改，在函数入口插入拦截函数参数的字节码。
• 此时如果目标函数被执行，拦截的函数参数将会拷贝到跨线程的队列中，并由另一线程完成数据上报。

测试用例

每个线程无限循环进程创建，文件操作以及网络请求，覆盖大多数 hook 点。由于大多数是 IO 行为，整体 CPU 不会太高，但是产生的消息量巨大。

调用耗时

使用 StopWatch 在每次字节码插入时开始计时，并在执行原始函数时暂停，能够较为精确地记录 RASP 额外指令带来的性能消耗。

测试结果

单线程低并发：

20个线程高并发：

内存占用

内存占用：50M 注入后内存立刻增长 30M，包括 JVM 开启的 AttachListener 线程，加载 SmithAgent 以及初始化 RASP 守护线程等带来的消耗。netty 开始通过 unix socket 与 server 通信后，缓存队列以及 socket 通信再次占用 20M。

CPU 消耗

单线程低并发情况下，注入时 CPU 飙升，随后 Agent 初始再次导致一波尖刺，然后逐步稳定，对 CPU 负载影响微乎其微。

20个线程高并发情况下，状况类似。

Golang

简介

在以往的 RASP 解决方案中，部署方式通常需要业务参与，修改相关配置或是启动参数，这也就造成了 RASP 部署困难的窘境。更有甚者，由于 Golang 编译型语言的特性，多数 RASP 只能被迫选择在编译期集成进去，以降低技术实现成本，但这无疑又间接加大了部署推广的难度。 我始终认为限制 RASP 发展与推广的是部署，而并非是技术难度，许多厂商在各语言的技术实现上都有大同小异的成熟方案。例如使用 JVM 的 Instrumentation 功能动态修改 bytecode，可以在虚拟机功能的基础上实现稳定的运行时防护。 所以在 Elkeid RASP 的项目初期，团队就敲定了动态注入的部署方式，一切都朝着降低部署难度的目标靠拢。

原理

目前 Elkeid RASP 开源了 JVM、Python、Node 以及 Golang 四种语言的运行时保护功能，四种语言均支持对已存在的进程动态注入防护代码。其中 JVM、Node 依赖于虚拟机提供的机制，运行稳定，而 Python 与 Golang 则需要利用 ptrace 在进程层面上做注入。

进程注入

为了实现对 Python 以及 Golang 进程的动态防护，先不考虑运行时层面的代码篡改，我们至少需要一个能够在 Linux 任意进程空间内执行任意代码的工具。但是很可惜，Linux 没有类似于 CreateRemoteThread 的接口。 大多数的代码注入，都是使用 ptrace 篡改进程执行流程，调用 dlopen 加载动态库。而且大多数项目都会指出，该方式的不稳定性可能会导致进程永久卡住。因为 dlopen 底层会调用 malloc，而在 glibc 的官方文档中指明了 malloc 是不可重入函数。 在研究过程中，我发现了 mandibule 这个项目，它另辟蹊径地编写了一个 ELF Loader，再使用 ptrace 让该 ELF Loader 在目标进程内执行，加载一个全新的程序，执行完成后恢复主线程的寄存器。 由于作者已经放弃维护，而且我自己在使用过程中，发现了项目一些设计上的缺陷以及代码 bug。于是我借鉴了该思路，开发了 pangolin 这个工具，它可以在任意进程内临时运行另一个程序，细节可以看相关的 blog。

Inline Hook

借助 pangolin，我们可以在一个 Golang 的进程中执行任意代码，我们甚至可以篡改可执行段的机器码，很轻松地便可以对某个函数进行 Inline Hook。例如我们想对 Golang 的命令执行函数 exec.Command 进行 Inline Hook，那么可以在进程注入期间修改函数 os/exec.Command 的开头指令，使其执行时先跳转到我们编写的函数中。在我们自定义的函数中，便可以获取该函数调用时的入参以及调用栈，再通过某种通信方式传输出去，一个简单的 RASP 模型便完成了。 那么要完成该流程，我们需要一些先决条件：

• 在去除掉 ELF 符号信息的情况下，如何获取 Golang 的符号信息，以确定函数的地址，完成对函数的 Inline Hook 操作。
• Golang 如何进行函数调用，通过寄存器亦或是栈，我们又该如何读取函数的入参。
• 如何获取 Golang 当前函数的 Stack Frame 长度，用于定位上一层函数的返回地址，完成调用栈回溯。

Golang Runtime Symbol Information

在去除了 ELF 符号信息后，一个编译好的 Golang 程序还是可以正确地执行 debug.PrintStack 函数，那么便可以证明 Golang 内部必然还存在一个符号表。根据官方文档 Go 1.2 Runtime Symbol Information 的介绍，Golang 从 1.2 之后的版本内置了符号信息，对于 ELF 格式来说，通常放置在 .gopclntab 这个 section。 这些符号信息不仅包含了函数名称、函数地址范围以及函数栈帧长度信息，甚至还有相关的代码文件名，以及行号等源码信息。根据文档记录的信息格式，我们可以很轻松地解析出一个 Golang 二进制程序的符号表。

Golang build info

对于 Golang 1.13 以上编译出的二进制，可以使用 go version 命令查看编译时的 Golang 版本，由此说明二进制中内嵌了相关的编译信息。对于 ELF 格式来说，编译信息存放在 .go.buildinfo section，其中包含 Golang 版本号以及三方依赖库列表。值得注意的是，buildinfo 的格式在 Golang 1.18 版本发生了改变，弃用了数据指针，相关解析代码可查看官方仓库。

Golang internal ABI

接下来我们需要了解 Golang 编译出的机器码，是如何进行函数调用的，以及 Golang 的结构体在内存中是如何存放的，了解这些之后我们才能正确地取出函数入参。

memory layout

Golang 包含的内置类型描述，可以在官方文档 The Go Programming Language Specification 中找到。对于数值类型，Golang 规范了类型的内存占用大小，而字节对齐则随 CPU 架构不同而变化。对于复合类型，例如 string、slice 以及 map 等，内存占用大小由组成的基础类型及其字节对齐决定，而该复合类型的字节对齐由组成类型中最大的字节对齐决定。 文档中并未描述 string 等内置类型的底层内存排布，但是我们可以从一些文章，亦或是 CGO 生成的头文件中一窥究竟。 以下是我使用 cpp 对 x64 架构下 Golang 类型的描述，代码可以在 Elkeid 官方仓库中找到：

namespace go {
    typedef signed char Int8;
    typedef unsigned char Uint8;
    typedef short Int16;
    typedef unsigned short Uint16;
    typedef int Int32;
    typedef unsigned int Uint32;
    typedef long long Int64;
    typedef unsigned long long Uint64;
    typedef Int64 Int;
    typedef Uint64 Uint;
    typedef __SIZE_TYPE__ Uintptr;
    typedef float Float32;
    typedef double Float64;
    typedef float _Complex Complex64;
    typedef double _Complex Complex128;

    struct interface {
        void *t;
        void *v;
    };

    struct string {
        const char *data;
        ptrdiff_t length;
    };

    template<typename T>
    struct slice {
        T *values;
        Int count;
        Int capacity;
    };
}

可以看到 string 类型由两个字段组成，数据指针加上字符串长度，在内存中总共占用 16 字节。string 的内存对齐则由这两个字段决定，即 align(string) = max(align(const char *), align(ptrdiff_t))。 对于 int32 这些 Golang 的基础数值类型来说，其字节对齐与 cpp 默认的对齐一致。

但是 Golang 并不确保这些类型的字节对齐不变，官方似乎正在考虑改变 x86 上 int64 的字节对齐。现在我们已经了解了 Golang 类型的内存排布，那么对于任意入参的函数调用，我们都能准确的从内存中取出数据。现在剩下的问题便是函数调用发生时，参数将会存放在何处？

stack-based calling conventions

Golang 在 1.17 版本之前的函数调用中，参数与结果均存放在栈上。但由于栈上频繁的内存操作影响了运行性能，所以社区草拟了基于寄存器的调用约定方案，并在 1.17 版本后切换到该调用约定。 我们先了解 Golang 最原始的 ABI0，也就是基于栈的调用约定，细节描述可以从文档 A Quick Guide to Go's Assembler 找到。在函数调用发生时，调用者需要将参数以及返回值，从低地址向高地址依次排列在栈顶。 例如在调用函数 func A(a int32, b string) (int32, error) 时，我们需要按下列排布存放参数与返回值：

+------------------------------+
| 2nd result error.v           |
| 2nd result error.t           |
| 1st result int32             |
| <pointer-sized alignment>    |
| b string.length              |
| b string.data                |
| a int32                      |
+------------------------------+ ↓ stack pointer

先放入 4 字节的参数 a，接着放入 string 类型的参数 b。由于 string 类型的字节对齐是 8，而此时的地址为 sp + 4，所以需要填充 4 字节的空白区域，从 sp + 8 开始放置 string 的数据。参数存放完成后，如果此时的地址没有按指针大小对齐，则需要填充空白字节。例如在 amd64 架构上，最后一个 int32 的参数放置于地址 0x40000，占用 4 字节大小，那么我们需要再填充 4 字节空白数据，使得返回值存放地址为 0x40008，按当前架构的指针大小 8 对齐。 我们接着放入第一个 int32 的返回值，而第二个返回值类型为 error，实际上就是 interface 类型。由上一小结可知，interface 类型占用 16 字节，按 8 字节对齐，所以我们填充 4 字节后，放入 error 结构体。当然，对于返回值而言，我们并不会真正地写入数据，而是预留内存空间以供被调用者写入。

register-based calling conventions

对于基于寄存器的调用约定，调用者需要先尝试将参数放置于寄存器中。如果结构体太大，或是结构体中包含 Non-trivial arrays 类型成员导致无法存放，则会转而放置于栈上。对于 amd64 架构，Golang 使用X0 – X14 寄存器存放浮点数数据，而对于整数数值，则使用以下 9 个整数寄存器存放：

RAX, RBX, RCX, RDI, RSI, R8, R9, R10, R11

对于数值类型参数，我们可以直接将参数一一对应到寄存器中。而对于结构体类型，我们需要将结构体拆解成多个基础数值类型，然后进行对应放置。如果一个结构体拆解后，需要占用的寄存器数超过了剩余的寄存器数，则该整个结构体都只能放置于栈上。该部分细节繁杂，本文不作赘述，细节请看文档 Go internal ABI specification。

实现

Runtime conflict

有了上述理论支持后，我们现在可以着手编写钩子函数了。在钩子函数执行过程中，不能随意篡改堆栈上的数据，执行完成后需要恢复所有寄存器，并跳转到原函数继续执行。需要注意的是，我们必须时刻记住，执行钩子函数的是 Golang 的线程，那么就存在以下两个问题：

• Golang 为线程分配的栈空间很小，钩子函数如果使用过度会导致 Segmentation fault。
• 在 Golang 的线程中执行时，我们无法正常调用 glibc 函数，例如 malloc 依赖于 fs 寄存器指向的 TLS 结构，以保证线程安全，但 fs 在 1.17 版本以下的 Golang 线程中指向全局 G。

为了解决第一个问题，我们需要在钩子函数的入口处，申请一块足够大的内存替换当前栈。而对于第二点，我们只能使用freestanding 代码及 syscall 来完成参数读取与栈回溯操作。为了更好地解耦与复用，于是我开发了一个不依赖于 glibc 的小型 c-runtime，包含内联汇编编写的 syscall 以及必要的标准库函数。我们可以安全地在 Golang 线程中调用 c-runtime 中的任何函数，例如使用底层是无锁环形缓冲区和 mmap syscall 的 z_malloc 来分配堆空间。

钩子函数

下面是使用内联汇编编写的钩子函数 wrapper，可以通用地进行栈替换、寄存器备份以及函数跳转：

asm volatile(
    "mov $1, %%r12;"
    "mov %%rsp, %%r13;"
    "add $8, %%r13;"
    "and $15, %%r13;"
    "sub $16, %%rsp;"
    "movdqu %%xmm14, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm13, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm12, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm11, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm10, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm9, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm8, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm7, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm6, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm5, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm4, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm3, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm2, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm1, (%%rsp);"
    "sub $16, %%rsp;"
    "movdqu %%xmm0, (%%rsp);"
    "push %%r11;"
    "push %%r10;"
    "push %%r9;"
    "push %%r8;"
    "push %%rsi;"
    "push %%rdi;"
    "push %%rcx;"
    "push %%rbx;"
    "push %%rax;"
    "sub %%r13, %%rsp;"
    "mov %0, %%rdi;"
    "call z_malloc;"
    "cmp $0, %%rax;"
    "je end_%=;"
    "mov %%rsp, %%rdi;"
    "mov %%rax, %%rsp;"
    "add %0, %%rsp;"
    "push %%rax;"
    "push %%rdi;"
    "add $312, %%rdi;"
    "add %%r13, %%rdi;"
    "call %P1;"
    "mov %%rax, %%r12;"
    "pop %%rsi;"
    "pop %%rdi;"
    "mov %%rsi, %%rsp;"
    "call z_free;"
    "end_%=:"
    "add %%r13, %%rsp;"
    "pop %%rax;"
    "pop %%rbx;"
    "pop %%rcx;"
    "pop %%rdi;"
    "pop %%rsi;"
    "pop %%r8;"
    "pop %%r9;"
    "pop %%r10;"
    "pop %%r11;"
    "movdqu (%%rsp), %%xmm0;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm1;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm2;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm3;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm4;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm5;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm6;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm7;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm8;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm9;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm10;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm11;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm12;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm13;"
    "add $16, %%rsp;"
    "movdqu (%%rsp), %%xmm14;"
    "add $16, %%rsp;"
    "cmp $0, %%r12;"
    "je block_%=;"
    "jmp *%2;"
    "block_%=:"
    "ret;"
    ::
    "i"(STACK_SIZE),
    "i"(handler),
    "m"(origin)
);

r12和r13 寄存器是 Golang 中可以随意使用的临时寄存器，我们用 r12 来标识是否要阻断当前调用。而r13 用来参与计算，以确保调用 handler 时栈指针按 16 字节对齐，这是 amd64 下 gcc 的默认约定。 在代码的开头，我们先将 X0 - X14 浮点数寄存器推入栈中，接着推入 Golang 1.17 以上需要使用的整数寄存器。然后调用 z_malloc 申请 40K 的内存替换当前栈，再以原始栈指针为参数调用 handler。在 handler 函数中，根据 Golang 的版本不同，我们可以从栈上存储的寄存器中，或上一函数的 Stack frame 中读出入参。当然也可以根据原始栈指针读取返回地址，从 Golang 符号表中查找函数信息，然后根据 Stack frame 读出上一层的返回地址，循环往复完成栈回溯。 我们甚至可以在 handler 中判断参数是否合法，当参数匹配到我们设置的正则时，可以手动写入 error 返回值到栈上，并返回 false 以将 r12 寄存器置零完成阻断。在 handler 函数执行完成后，从栈上恢复寄存器，并根据 r12 决定返回还是跳转至原函数。 为了更好地进行参数读取和阻断，我使用Templates 编写了一套 Golang 类型反射库，可以在运行时获取 Golang 类型元数据。元数据包含类型的基础类型成员数，每个成员的相对偏移以及占用大小，还有该类型需要占用的浮点/整数寄存器数。在 handler 函数中，我们可以轻松地利用这些元数据分析 Golang 的参数内存布局，正确地取出数据。由于该部分代码细节繁多，限于本文篇幅所以不进行详细讲解，取参与回溯部分请直接阅读仓库代码。

回溯停止

对于调用栈的回溯，上面已经解析过了，我们可以取出当前栈顶的返回地址，在符号表中查找地址相关的函数的名称、文件、行号以及栈帧大小。获取栈帧大小后，取出 sp + framesize 的上一层返回地址，循环上述步骤即可。但有一个问题是，我们在哪里结束循环？调用链的层数一定有限，那么第一个函数是哪个？ 在 1.2 版本中，Golang 通过判断函数名是否为 runtime.goexit、runtime.rt0_go 等入口函数，由此决定是否终止回溯。而对于较新版本的 Golang ，符号信息中增加了一个 funcID 字段，通过 funcID 判断函数类型是否为入口函数。但 funcID 的本质与函数名比较无二，而且 funcID 在版本之间会发生变动，所以最后决定简单地使用函数名判断：

constexpr auto STACK_TOP_FUNCTION = {
        "runtime.mstart",
        "runtime.rt0_go",
        "runtime.mcall",
        "runtime.morestack",
        "runtime.lessstack",
        "runtime.asmcgocall",
        "runtime.externalthreadhandler",
        "runtime.goexit"
};

消息通信

成功获取入参和调用栈后，要如何把消息传输出去？如果需要进行 socket 通信，并且不阻塞 Golang 线程，那就需要驻留一个线程在 Golang 进程内，实现一个简单的生产者消费者模型。那么在无法使用 std::queue 等标准库的情况下，要怎么实现消费丢列，又该如何保证线程安全？ 为了尽可能地减少性能影响，我利用 gcc 内置的原子操作实现了一个定长的无锁环形缓冲区，并使用 c-runtime 中实现的 condition variable 做线程同步，实现了一个高效的消息队列。在每个钩子函数触发时，都会将入参和调用栈打包放入队列，如果队列已满则丢弃该消息。 在 pangolin 注入过程中，我们启动一个消费者线程，从消息队列中消费函数调用信息，序列化为 json 后通过 unix socket 传输到 server。

信号屏蔽

Golang 启动时会设置信号处理函数，而在进程收到信号时，内核会随机选择一个线程进行信号处理。我们在 Golang 进程中驻留的几个 cpp 线程有可能被选中用于执行处信号处理函数，但是处理函数默认当前处于 Golang 线程中，读取 fs 寄存器以访问 Golang 的全局 G，但此时 fs 所指向的其实是 glibc 的 TLS，于是导致异常退出。 为了避免这种情况发生，我们需要手动设置驻留的 cpp 线程，令其屏蔽所有信号：

sigset_t mask = {}; 
sigset_t origin_mask = {};

sigfillset(&mask);

if (pthread_sigmask(SIG_SETMASK, &mask, &origin_mask) != 0) {
    LOG_ERROR("set signal mask failed");
    quit(-1);
}

流程

在学习了原理和实现细节后，我们来解析一下 go-probe 的执行流程，入口函数如下：

#include "go/symbol/build_info.h"
#include "go/symbol/line_table.h"
#include "go/symbol/interface_table.h"
#include "go/api/api.h"
#include <zero/log.h>
#include <csignal>
#include <asm/api_hook.h>
#include <z_syscall.h>

void quit(int status) {
    uintptr_t address = 0;
    char *env = getenv("QUIT");

    if (!env) {
        LOG_WARNING("can't found quit env variable");
        z_exit_group(-1);
    }

    if (!zero::strings::toNumber(env, address, 16) || !address) {
        LOG_ERROR("invalid quit function address");
        z_exit_group(-1);
    }

    ((decltype(quit) *)address)(status);
}

int main() {
    INIT_FILE_LOG(zero::INFO, "go-probe");

    sigset_t mask = {};
    sigset_t origin_mask = {};

    sigfillset(&mask);

    if (pthread_sigmask(SIG_SETMASK, &mask, &origin_mask) != 0) {
        LOG_ERROR("set signal mask failed");
        quit(-1);
    }

    if (!gLineTable->load()) {
        LOG_ERROR("line table load failed");
        quit(-1);
    }

    if (gBuildInfo->load()) {
        LOG_INFO("go version: %s", gBuildInfo->mVersion.c_str());

        CInterfaceTable table = {};

        if (!table.load()) {
            LOG_ERROR("interface table load failed");
            quit(-1);
        }

        table.findByFuncName("errors.(*errorString).Error", (go::interface_item **)CAPIBase::errorInterface());
    }

    gSmithProbe->start();

    for (const auto &api : GOLANG_API) {
        for (unsigned int i = 0; i < gLineTable->mFuncNum; i++) {
            CFunc func = {};

            if (!gLineTable->getFunc(i, func))
                break;

            const char *name = func.getName();
            void *entry = (void *)func.getEntry();

            if ((api.ignoreCase ? strcasecmp(api.name, name) : strcmp(api.name, name)) == 0) {
                LOG_INFO("hook %s: %p", name, entry);

                if (hookAPI(entry, (void *)api.metadata.entry, api.metadata.origin) < 0) {
                    LOG_WARNING("hook %s failed", name);
                    break;
                }

                break;
            }
        }
    }

    pthread_sigmask(SIG_SETMASK, &origin_mask, nullptr);
    quit(0);

    return 0;
}

需要明确的是，go-probe 由 pangolin 注入到 Golang 进程的主线程中临时运行。同时 pangolin 使用 ptrace 持续监听该线程的 syscall 调用，拦截到 main 函数发出的 exit 或 exit_group 调用后，恢复线程状态并结束注入流程。然而可以看到，上面的代码中会优先调用环境变量 QUIT 指向的函数，这又是为何？ 在实际的部署过程中，由于资源限制等诸多特殊原因，pangolin 进程可能会在注入期间被 kill。那么此时 main 函数执行的 syscall 就无人拦截，exit_group 会真正地导致业务进程退出。为了让执行 go-probe 的线程能够自我恢复，pangolin 会提前将线程状态快照写入到 Golang 内存中。同时遗留在 Golang 进程中的 shellcode 包含一个 quit 函数，能够根据该快照主动恢复线程，类似于 glibc 的 setcontext，而 QUIT 环境变量正是 quit 的地址。 在初始化文件日志后，先令当前线程屏蔽所有信号，之后启动的所有子线程都会继承该设置。然后从 ELF 的 section 中加载符号表、编译信息以及 interface 表，并且为了支持阻断功能，查找 errors.(*errorString) 的地址并保存。执行 gSmithProbe->start() 启动通信客户端后，从符号表中查找 GOLANG_API 所有子项，并进行 Inline Hook。完成以上流程后，恢复信号掩码并调用 quit 函数以通知 pangolin 结束注入。

CPython

RASP CPython 耗时计算

OUT OF DATE 数据已过期，仅供参考。 新测算正在路上

单位ns

import time
import timeit


class Builtins:
    # avoid timeit using exec
    @staticmethod
    def bench_eval():
        st = time.perf_counter_ns()
        for i in range(100000):
            eval("1024*1024*{}".format(i), globals(), {})
        et = time.perf_counter_ns()
        return et - st

    @staticmethod
    def bench_open():
        st = time.perf_counter_ns()
        for i in range(100000):
            open("/etc/passwd").close()
        et = time.perf_counter_ns()
        return et - st

    @staticmethod
    def bench_exec():
        st = time.perf_counter_ns()
        for i in range(100000):
            exec("1024*1024*{}".format(i), globals(), {})
        et = time.perf_counter_ns()
        return et - st


class Requests():
    @staticmethod
    def bench_get():
        return timeit.timeit(timer=time.perf_counter_ns,
                             setup="import requests",
                             stmt="requests.get('http://localhost/')",
                             number=10000,
                             )

    @staticmethod
    def bench_post():
        return timeit.timeit(timer=time.perf_counter_ns,
                             setup="import requests",
                             stmt="requests.post('http://localhost/')",
                             number=10000,
                             )

    @staticmethod
    def bench_put():
        return timeit.timeit(timer=time.perf_counter_ns,
                             setup="import requests",
                             stmt="requests.put('http://localhost/')",
                             number=10000,
                             )

    @staticmethod
    def bench_head():
        return timeit.timeit(timer=time.perf_counter_ns,
                             setup="import requests",
                             stmt="requests.head('http://localhost/')",
                             number=10000,
                             )

    @staticmethod
    def bench_delete():
        return timeit.timeit(timer=time.perf_counter_ns,
                             setup="import requests",
                             stmt="requests.delete('http://localhost/')",
                             number=10000,
                             )

    @staticmethod
    def bench_options():
        return timeit.timeit(timer=time.perf_counter_ns,
                             setup="import requests",
                             stmt="requests.options('http://localhost/')",
                             number=10000,
                             )

    @staticmethod
    def bench_patch():
        return timeit.timeit(timer=time.perf_counter_ns,
                             setup="import requests",
                             stmt="requests.patch('http://localhost/')",
                             number=10000,
                             )

    @staticmethod
    def bench_reqeust():
        return timeit.timeit(timer=time.perf_counter_ns,
                             setup="import requests",
                             stmt="requests.request('GET', 'http://localhost/')",
                             number=10000,
                             )


class Os:
    @staticmethod
    def bench_system():
        return timeit.timeit(timer=time.perf_counter_ns,
                             setup='import os', stmt='os.system("w > /dev/null")', number=1000
                             )

    @staticmethod
    def bench_popen():
        return timeit.timeit(timer=time.perf_counter_ns,
                             setup="import os",
                             stmt="os.popen('w > /dev/null')", number=1000)

    @staticmethod
    def bench_spawnv():
        return timeit.timeit(timer=time.perf_counter_ns,
                             setup="import os",
                             stmt='os.spawnv(os.P_WAIT, "/bin/w", [">", "/dev/null"])',
                             number=1000)


def setup_rasp(iast=False):
    import rasp
    rasp.setup_var(iast, False)
    rasp.setup_client()
    rasp.hook()


if __name__ == "__main__":
    # print(Builtins.bench_eval())
    # os
    import sys
    point = sys.argv[1]
    if point == "os":
        raw_os_system = Os.bench_system()
        raw_os_popen = Os.bench_popen()
        raw_os_spawnv = Os.bench_spawnv()
        setup_rasp()
        rasp_os_system = Os.bench_system()
        rasp_os_popen = Os.bench_popen()
        rasp_os_spawnv = Os.bench_spawnv()
        print("os.system: ", raw_os_system, rasp_os_system,
              (rasp_os_system-raw_os_system)/1000)
        print("os.popen: ", raw_os_popen, rasp_os_popen,
              (rasp_os_popen-raw_os_popen)/1000)
        print("os.spawnv: ", raw_os_spawnv, rasp_os_spawnv,
              (rasp_os_spawnv-raw_os_spawnv)/1000)
    if point == "builtin":
        raw_eval = Builtins.bench_eval()
        raw_exec = Builtins.bench_exec()
        raw_open = Builtins.bench_open()
        setup_rasp()
        rasp_eval = Builtins.bench_eval()
        rasp_exec = Builtins.bench_exec()
        rasp_open = Builtins.bench_open()
        print("eval:", raw_eval, rasp_eval, (rasp_eval-raw_eval)/100000)
        print("exec:", raw_exec, rasp_exec, (rasp_exec-raw_exec)/100000)
        print("open:", raw_open, rasp_open, (rasp_open-raw_open)/100000)
    if point == "requests":
        requests_bench = [func for func in dir(
            Requests) if func.startswith("bench")]
        print(requests_bench)
        requests_res = {}
        for func in requests_bench:
            res = getattr(Requests, func)()
            print(res)
            requests_res["raw_{}".format(func)] = res
        setup_rasp(iast=True)
        for func in requests_bench:
            res = getattr(Requests, func)()
            print(res)
            requests_res["rasp_{}".format(func)] = res
        for func in requests_bench:
            raw = requests_res.get("raw_{}".format(func))
            rasp = requests_res.get("rasp_{}".format(func))
            print("requests.{}:".format(
                func[6:]), raw, rasp, (rasp-raw)/10000)

原理

PHP 解释器提供了挂钩函数与 opcode 的接口，我们需要基于 PHP 提供的头文件，使用 C 系语言构建一个 RASP 拓展。PHP 在加载拓展后，会依次调用模块导出的初始化函数，我们在函数中便可以完成函数与 opcode 的挂钩替换，获取 HTTP 请求参数、函数调用详情以及调用栈，传输到远端进行分析。 对于函数而言，直接搜索函数哈希表然后替换 internal_function.handler 指向自定义的 wrapper，便可以在函数被调用时获得控制权，而 opcode 则只需要简单的调用 zend_set_user_opcode_handler 指向自定义处理函数。 需要注意的是，PHP 的拓展本质是一个动态库，使用导出函数供解释器调用以完成加载。而且针对不同的 PHP 版本，都需要使用对应的头文件分别进行编译，否则无法保证正确加载，Elkeid RASP 每次发布都会附带常见的预编译版本。

多进程架构

PHP 通常不会以 CLI 模式单独运行，大多数会依赖于 PHP-FPM 或 Apache。在默认情况下 PHP 以单线程运行，明显这样是无法满足高并发处理需求的，所以 PHP-FPM 这些框架在主进程加载完模块后，会进行多次 fork 使用多进程模型分担流量压力。 在主进程加载 RASP 拓展初始化时，我们挂钩了函数与 opcode，之后主进程 fork 出多个进程，这些进程完全一致且函数与 opcode 都处于被挂钩了的状态，现在的问题是我们在 wrapper 函数中应该将调用数据发往哪里？ 为了减少性能影响，Elkeid RASP 系列产品都不会使用脚本引擎在本地进行检测，而是将函数调用数据序列化为 json 通过 unix socket 传输到服务器进行分析。那么为了进行异步的数据收发，通常需要一个单独的 eventloop 线程，但是多线程在进行 fork 时存在安全隐患可能导致死锁，而且单次 fork 只会复制当前调用线程。 为了在多个进程之间共享调用数据和配置，Elkeid RASP 模块在主进程初始化阶段会分配一段共享内存，其中包含一个用来传递调用信息的无锁环形缓冲区，以及一片储存配置的共享区域。分配完成后，调用 fork 复制一个 RASP 独占的新进程通过 unix socket 进行消息通信，从缓冲区消费数据发送到远端，以及从远端收取配置消息写入共享空间。

拓展生命周期

PHP 每个拓展加载后都会经历四个阶段：

• MINIT，这是模块的启动步骤，对于 RASP 而言通常在此处完成挂钩操作。
• RINIT，每个请求的到来都会触发，可以获取此次请求的详细信息进行分析。
• RSHUTDOWN，每个请求处理结束后触发，通常可以忽略该步骤。
• MSHUTDOWN，模块的卸载步骤，在此处进行资源释放和清理操作。

MINIT

RASP 在此处进行模块初始化和挂钩操作：

PHP_MINIT_FUNCTION (php_probe) {
    ZEND_INIT_MODULE_GLOBALS(php_probe, PHP_GINIT(php_probe), PHP_GSHUTDOWN(php_probe))

    if (!gAPIConfig || !gAPITrace)
        return FAILURE;

    if (fork() == 0) {
        INIT_FILE_LOG(zero::INFO, "php-probe");

        char name[16] = {};
        snprintf(name, sizeof(name), "probe(%d)", getppid());

        if (prctl(PR_SET_PDEATHSIG, SIGKILL) < 0) {
            LOG_ERROR("set death signal failed");
            exit(-1);
        }

        if (pthread_setname_np(pthread_self(), name) != 0) {
            LOG_ERROR("set process name failed");
            exit(-1);
        }

        gSmithProbe->start();

        exit(0);
    }

    for (const auto &api: PHP_API) {
        HashTable *hashTable = CG(function_table);

        if (api.cls) {
#if PHP_MAJOR_VERSION > 5
            auto cls = (zend_class_entry *) zend_hash_str_find_ptr(CG(class_table), api.cls, strlen(api.cls));

            if (!cls) {
                LOG_WARNING("can't found class: %s", api.cls);
                continue;
            }

            hashTable = &cls->function_table;
#else
            zend_class_entry **cls;

            if (zend_hash_find(CG(class_table), api.cls, strlen(api.cls) + 1, (void **)&cls) != SUCCESS) {
                LOG_WARNING("can't found class: %s", api.cls);
                continue;
            }

            hashTable = &(*cls)->function_table;
#endif
        }

#if PHP_MAJOR_VERSION > 5
        auto func = (zend_function *) zend_hash_str_find_ptr(hashTable, api.name, strlen(api.name));

        if (!func) {
            LOG_WARNING("can't found function: %s", api.name);
            continue;
        }
#else
        zend_function *func;

        if (zend_hash_find(hashTable, api.name, strlen(api.name) + 1, (void **)&func) != SUCCESS) {
            LOG_WARNING("can't found function: %s", api.name);
            continue;
        }
#endif

#if PHP_MAJOR_VERSION < 8
        if (func->internal_function.handler == ZEND_FN(display_disabled_function)) {
            LOG_WARNING("disabled function: %s", api.name);
            continue;
        }
#endif

        *api.metadata.origin = func->internal_function.handler;
        func->internal_function.handler = api.metadata.entry;
    }

    for (const auto &opcode: PHP_OPCODE)
        zend_set_user_opcode_handler(opcode.op, opcode.handler);

    return SUCCESS;
}

首先使用 ZEND_INIT_MODULE_GLOBALS 初始化全局变量，用来持续追踪单次请求的详细参数，可以将多次函数调用关联到某个具体的请求。另外需要注意，这里的全局变量并不是语言层面上的，因为 PHP 有多线程版本，在多线程中会为每个线程拷贝一份全新的变量，当然这些都会在头文件提供的宏中进行判断。

ZEND_BEGIN_MODULE_GLOBALS(php_probe)
    Request request{};
ZEND_END_MODULE_GLOBALS(php_probe)

gAPIConfig 与 gAPITrace 便是共享内存中的缓冲区与配置区域，然后 fork 一个进程进行消息通信，最后依次查找函数、挂钩函数以及挂钩 opcode。

RINIT

在这个阶段我们可以获取一个请求的详细信息，例如 url、header 或者 body，储存在全局变量的 request 中：

PHP_RINIT_FUNCTION (php_probe) {
    zval *server = HTTPGlobals(
            TRACK_VARS_SERVER
#if PHP_MAJOR_VERSION <= 5
            TSRMLS_CC
#endif
    );

    if (!server || Z_TYPE_P(server) != IS_ARRAY)
        return SUCCESS;

    auto fetch = [=](const HashTable *hashTable, const char *key) -> std::string {
        zval *val = hashFind(hashTable, key);

        if (!val)
            return "";

        return {Z_STRVAL_P(val), (std::size_t) Z_STRLEN_P(val)};
    };

    strncpy(PHP_PROBE_G(request).scheme, fetch(Z_ARRVAL_P(server), "REQUEST_SCHEME").c_str(), SMITH_FIELD_LENGTH - 1);
    strncpy(PHP_PROBE_G(request).host, fetch(Z_ARRVAL_P(server), "HTTP_HOST").c_str(), SMITH_FIELD_LENGTH - 1);
    strncpy(PHP_PROBE_G(request).serverName, fetch(Z_ARRVAL_P(server), "SERVER_NAME").c_str(), SMITH_FIELD_LENGTH - 1);
    strncpy(PHP_PROBE_G(request).serverAddress, fetch(Z_ARRVAL_P(server), "SERVER_ADDR").c_str(), SMITH_FIELD_LENGTH - 1);
    strncpy(PHP_PROBE_G(request).uri, fetch(Z_ARRVAL_P(server), "REQUEST_URI").c_str(), SMITH_FIELD_LENGTH - 1);
    strncpy(PHP_PROBE_G(request).query, fetch(Z_ARRVAL_P(server), "QUERY_STRING").c_str(), SMITH_FIELD_LENGTH - 1);
    strncpy(PHP_PROBE_G(request).method, fetch(Z_ARRVAL_P(server), "REQUEST_METHOD").c_str(), SMITH_FIELD_LENGTH - 1);
    strncpy(PHP_PROBE_G(request).remoteAddress, fetch(Z_ARRVAL_P(server), "REMOTE_ADDR").c_str(), SMITH_FIELD_LENGTH - 1);
    strncpy(PHP_PROBE_G(request).documentRoot, fetch(Z_ARRVAL_P(server), "DOCUMENT_ROOT").c_str(), SMITH_FIELD_LENGTH - 1);

    std::optional<short> port = zero::strings::toNumber<short>(fetch(Z_ARRVAL_P(server), "SERVER_PORT"));

    if (port)
        PHP_PROBE_G(request).port = *port;

    for (const auto &e: Z_ARRVAL_P(server)) {
        if (e.type != HASH_KEY_IS_STRING || Z_TYPE_P(e.value) != IS_STRING)
            continue;

        std::string override;
        std::string key = std::get<std::string>(e.key);

        if (key == "HTTP_CONTENT_TYPE" || key == "CONTENT_TYPE") {
            override = "content-type";
        } else if (key == "HTTP_CONTENT_LENGTH" || key == "CONTENT_LENGTH") {
            override = "content-length";
        } else if (zero::strings::startsWith(key, "HTTP_")) {
            override = zero::strings::tolower(key.substr(5));
            std::replace(override.begin(), override.end(), '_', '-');
        } else {
            continue;
        }

        strncpy(PHP_PROBE_G(request).headers[PHP_PROBE_G(request).header_count][0], override.c_str(), SMITH_FIELD_LENGTH - 1);
        strncpy(PHP_PROBE_G(request).headers[PHP_PROBE_G(request).header_count][1], Z_STRVAL_P(e.value), SMITH_FIELD_LENGTH - 1);

        if (++PHP_PROBE_G(request).header_count >= SMITH_HEADER_COUNT)
            break;
    }

    if (strcasecmp(PHP_PROBE_G(request).method, "post") != 0 && strcasecmp(PHP_PROBE_G(request).method, "put") != 0)
        return SUCCESS;

    auto begin = PHP_PROBE_G(request).headers;
    auto end = begin + PHP_PROBE_G(request).header_count;

    if (std::find_if(begin, end, [](const auto &header) {
        if (strcmp(header[0], "content-type") != 0)
            return false;

        return strncmp(header[1], "multipart/form-data", 19) == 0;
    }) != end) {
        strncpy(
                PHP_PROBE_G(request).body,
                toString(
                        HTTPGlobals(
                                TRACK_VARS_POST
#if PHP_MAJOR_VERSION <= 5
                                TSRMLS_CC
#endif
                        )
#if PHP_MAJOR_VERSION <= 5
                        TSRMLS_CC
#endif
                ).c_str(),
                SMITH_FIELD_LENGTH - 1
        );

        zval *files = HTTPGlobals(
                TRACK_VARS_FILES
#if PHP_MAJOR_VERSION <= 5
                TSRMLS_CC
#endif
        );

        if (!files || Z_TYPE_P(files) != IS_ARRAY)
            return SUCCESS;

        for (const auto &e: Z_ARRVAL_P(files)) {
            if (Z_TYPE_P(e.value) != IS_ARRAY)
                continue;

            strncpy(PHP_PROBE_G(request).files[PHP_PROBE_G(request).file_count].name, fetch(Z_ARRVAL_P(e.value), "name").c_str(), SMITH_FIELD_LENGTH - 1);
            strncpy(PHP_PROBE_G(request).files[PHP_PROBE_G(request).file_count].type, fetch(Z_ARRVAL_P(e.value), "type").c_str(), SMITH_FIELD_LENGTH - 1);
            strncpy(PHP_PROBE_G(request).files[PHP_PROBE_G(request).file_count].tmp_name, fetch(Z_ARRVAL_P(e.value), "tmp_name").c_str(), SMITH_FIELD_LENGTH - 1);

            if (++PHP_PROBE_G(request).file_count >= SMITH_FILE_COUNT)
                break;
        }

        return SUCCESS;
    }

    php_stream *stream = php_stream_open_wrapper("php://input", "rb", REPORT_ERRORS, nullptr);

    if (!stream)
        return SUCCESS;

    char buffer[1024] = {};

    int n = php_stream_read(stream, buffer, sizeof(buffer));

    if (n < 0) {
        php_stream_close(stream);
        return SUCCESS;
    }

    for (int i = 0, j = 0; i < n && j < SMITH_FIELD_LENGTH - 1; i++) {
        if (!isprint(buffer[i]))
            continue;

        PHP_PROBE_G(request).body[j++] = buffer[i];
    }

    php_stream_close(stream);

    return SUCCESS;
}

RSHUTDOWN

这个阶段时请求已经处理完成，我们需要将 request 清空：

PHP_RSHUTDOWN_FUNCTION (php_probe) {
    PHP_PROBE_G(request) = {};
    return SUCCESS;
}

MSHUTDOWN

在模块卸载的阶段，我们还原挂钩的函数以及 opcode，释放已分配的资源：

PHP_MSHUTDOWN_FUNCTION (php_probe) {
    for (const auto &api: PHP_API) {
        HashTable *hashTable = CG(function_table);

        if (api.cls) {
#if PHP_MAJOR_VERSION > 5
            auto cls = (zend_class_entry *) zend_hash_str_find_ptr(CG(class_table), api.cls, strlen(api.cls));

            if (!cls) {
                LOG_WARNING("can't found class: %s", api.cls);
                continue;
            }

            hashTable = &cls->function_table;
#else
            zend_class_entry **cls;

            if (zend_hash_find(CG(class_table), api.cls, strlen(api.cls) + 1, (void **)&cls) != SUCCESS) {
                LOG_WARNING("can't found class: %s", api.cls);
                continue;
            }

            hashTable = &(*cls)->function_table;
#endif
        }

#if PHP_MAJOR_VERSION > 5
        auto func = (zend_function *) zend_hash_str_find_ptr(hashTable, api.name, strlen(api.name));

        if (!func) {
            LOG_WARNING("can't found function: %s", api.name);
            continue;
        }
#else
        zend_function *func;

        if (zend_hash_find(hashTable, api.name, strlen(api.name) + 1, (void **)&func) != SUCCESS) {
            LOG_WARNING("can't found function: %s", api.name);
            continue;
        }
#endif

#if PHP_MAJOR_VERSION < 8
        if (func->internal_function.handler == ZEND_FN(display_disabled_function)) {
            LOG_WARNING("disabled function: %s", api.name);
            continue;
        }
#endif

        if (!*api.metadata.origin) {
            LOG_WARNING("null origin handler");
            continue;
        }

        func->internal_function.handler = *api.metadata.origin;
    }

    for (const auto &opcode: PHP_OPCODE)
        zend_set_user_opcode_handler(opcode.op, nullptr);

#ifdef ZTS
    ts_free_id(php_probe_globals_id);
#else
    PHP_GSHUTDOWN(php_probe)(&php_probe_globals);
#endif

    return SUCCESS;
}

变量

PHP 中所有的变量都储存在一个 zval 结构体中，实际上就是一个巨大的 union，根据类型获取相对应的值。为了更好的传递函数调用信息，我们需要将每个参数转为可读字符串类型，所以我们手写一个转换函数：

std::string toString(
        zval *val
#if PHP_MAJOR_VERSION <= 5
        TSRMLS_DC
#endif
) {
    if (!val)
        return "";

    switch (Z_TYPE_P(val)) {
        case IS_NULL:
            return "null";

#if PHP_MAJOR_VERSION > 5
        case IS_FALSE:
            return "false";

        case IS_TRUE:
            return "true";
#else
        case IS_BOOL:
            return Z_BVAL_P(val) ? "true" : "false";
#endif

        case IS_LONG:
            return std::to_string(Z_LVAL_P(val));

        case IS_DOUBLE:
            return std::to_string(Z_DVAL_P(val));

        case IS_STRING:
            return {Z_STRVAL_P(val), (std::size_t) Z_STRLEN_P(val)};

        case IS_ARRAY: {
            auto quoted = [](const std::string &str) -> std::string {
                std::stringstream ss;
                ss << std::quoted(str);

                return ss.str();
            };

            bool uneven = false;
            std::map<std::string, std::string> kv;

            for (const auto &e: Z_ARRVAL_P(val)) {
                switch (e.type) {
                    case HASH_KEY_IS_LONG:
                        kv.insert({
                            std::to_string(std::get<unsigned long>(e.key)),
                            toString(
                                    e.value
#if PHP_MAJOR_VERSION <= 5
                                    TSRMLS_CC
#endif
                            )
                        });

                        break;

                    case HASH_KEY_IS_STRING:
                        uneven = true;

                        kv.insert({
                            quoted(std::get<std::string>(e.key)),
                            toString(
                                    e.value
#if PHP_MAJOR_VERSION <= 5
                                    TSRMLS_CC
#endif
                            )
                        });

                        break;

                    default:
                        break;
                }
            }

            std::list<std::string> items;

            if (!uneven) {
                std::transform(
                        kv.begin(),
                        kv.end(),
                        std::back_inserter(items),
                        [](const auto &it) {
                            return it.second;
                        }
                );

                return zero::strings::join(items, " ");
            }

            std::transform(
                    kv.begin(),
                    kv.end(),
                    std::back_inserter(items),
                    [&](const auto &it) {
                        return it.first + ": " + quoted(it.second);
                    }
            );

            return zero::strings::format("{%s}", zero::strings::join(items, ", ").c_str());
        }

        case IS_RESOURCE: {
            const char *type = zend_rsrc_list_get_rsrc_type(
#if PHP_MAJOR_VERSION > 5
                    Z_RES_P(val)
#else
                    Z_LVAL_P(val) TSRMLS_CC
#endif
            );

            if (!type)
                break;

            if (strcmp(type, "curl") == 0)
                return curlInfo(
                        val
#if PHP_MAJOR_VERSION <= 5
                        TSRMLS_CC
#endif
                );

            return "resource";
        }

        default:
            break;
    }

    return "unknown";
}

值得注意的是，语言中常见的数组和字典在 PHP 中都是 array 类型，它实际上就是 kv 类型的哈希表，针对广义上的数组储存会将 key 设为连续的 index 数字，所以在转为字符串时可以将这些 key 忽略。

调用参数

函数挂钩后被调用时，我们临时获取到执行权，这是我们需要提取出调用参数以及调用栈，传递给远端分析，我们先来看函数原型：

void entry(zend_execute_data *execute_data, zval *return_value);

很显然我们需要从 execute_data 提取出参数，PHP 提供的一种兼容性较好的方式是调用 zend_parse_parameters：

long l;
char *s;
int s_len;
zval *param;

zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "lsz", &l, &s, &s_len, &param);

这是一个类似于 scanf 的公开接口，我们可以指定输入参数的具体类型，将其分解到对应的 C 语言类型中，或者使用 'z' 直接获取原始 zval。但是一个明显的缺点是，如果有两个入参不同的函数，我们便需要编写两个 wrapper 函数，并使用不同的 zend_parse_parameters 参数进行爬取。 简化一下，我们可以单纯地使用 'z' 提出出所有参数的原始 zval，以使用上一步的转换函数转为字符串，但是还存在的问题是函数的参数个数会不一样，而且 PHP 允许忽略可选参数，例如使用 'z|z' 爬取一个必要参数和一个可选参数。 为了更好地定义 wrapper 以及挂钩函数，我编写了一套模板用来提取参数：

template<int ClassID, int MethodID, bool CanBlock, bool Ret, int Required, int Optional = 0>
class APIEntry {
public:
    static constexpr auto getTypeSpec() {
        constexpr size_t length = Required + (Optional > 0 ? Optional + 1 : 0);
        std::array<char, length + 1> buffer = {};

        for (size_t i = 0; i < length; i++) {
            if (i == Required) {
                buffer[i] = '|';
                continue;
            }

            buffer[i] = 'z';
        }

        return buffer;
    }

    static void entry(INTERNAL_FUNCTION_PARAMETERS) {
        entry(std::make_index_sequence<Required + Optional>{}, INTERNAL_FUNCTION_PARAM_PASSTHRU);
    }

    template<size_t ...Index>
    static void entry(std::index_sequence<Index...>, INTERNAL_FUNCTION_PARAMETERS) {
        zval *args[sizeof...(Index)] = {};
        int argc = std::min(Required + Optional, (int)ZEND_NUM_ARGS());

#if PHP_MAJOR_VERSION > 5
        constexpr
#endif
        auto spec = getTypeSpec();

        if (zend_parse_parameters(
#if PHP_MAJOR_VERSION > 5
                argc,
#else
                argc TSRMLS_CC,
#endif
                spec.data(),
                &args[Index]...) != SUCCESS) {
            origin(INTERNAL_FUNCTION_PARAM_PASSTHRU);
            return;
        }

        Trace trace = {
                ClassID,
                MethodID
        };

        while (trace.count < std::min(argc, SMITH_ARG_COUNT)) {
            zval *arg = args[trace.count];

            if (!arg)
                continue;

            strncpy(
                    trace.args[trace.count++],
                    toString(
                            arg
#if PHP_MAJOR_VERSION <= 5
                            TSRMLS_CC
#endif
                    ).c_str(),
                    SMITH_ARG_LENGTH - 1
            );
        }

        std::vector<std::string> stackTrace = traceback(
#if PHP_MAJOR_VERSION <= 5
                TSRMLS_C
#endif
        );

        for (int i = 0; i < stackTrace.size() && i < SMITH_TRACE_COUNT; i++) {
            strncpy(trace.stackTrace[i], stackTrace[i].c_str(), SMITH_TRACE_LENGTH - 1);
        }

        trace.request = PHP_PROBE_G(request);

        if constexpr (CanBlock) {
            if (gAPIConfig->block(trace)) {
                trace.blocked = true;

                gAPITrace->enqueue(trace);

                zend_throw_exception(
                        nullptr,
                        "API blocked by RASP",
                        0
#if PHP_MAJOR_VERSION <= 5
                        TSRMLS_CC
#endif
                );

                return;
            }
        }

        if constexpr (!Ret) {
            if (!gAPIConfig->surplus(ClassID, MethodID))
                return;

            gAPITrace->enqueue(trace);
            origin(INTERNAL_FUNCTION_PARAM_PASSTHRU);

            return;
        }

        origin(INTERNAL_FUNCTION_PARAM_PASSTHRU);

        strncpy(
                trace.ret,
                toString(
                        return_value
#if PHP_MAJOR_VERSION <= 5
                        TSRMLS_CC
#endif
                ).c_str(),
                SMITH_ARG_LENGTH - 1
        );

        if (!gAPIConfig->surplus(ClassID, MethodID))
            return;

        gAPITrace->enqueue(trace);
    }

public:
    static constexpr APIMetadata metadata() {
        return {entry, &origin};
    }

public:
    static handler origin;
};

例如对于有可选参数的 readfile 函数，我们可以简单的定义 wrapper：

APIEntry<1, 1, false, false, 1, 1>::metadata()

Required 表明必需的参数个数，Optional 表示可选的参数个数，当然对于 readfile 而言，可选参数实际为 2 个，但是最后一个我们用不上所以忽略。

Opcode 参数

在 opcode 处理函数中获取参数有所不同，并且 opcode 限制了参数个数上限为 2。在 PHP7 之后，我们可以简单的使用 zend_get_zval_ptr 获取参数的原始 zval：

int entry(
        zend_execute_data *execute_data
#if PHP_MAJOR_VERSION <= 5
        TSRMLS_DC
#endif
) {
    zval *op1 = zend_get_zval_ptr(execute_data->opline, execute_data->opline->op1_type, &execute_data->opline->op1, execute_data);
    zval *op2 = zend_get_zval_ptr(execute_data->opline, execute_data->opline->op2_type, &execute_data->opline->op2, execute_data);

    return ZEND_USER_OPCODE_DISPATCH;
}

zend_get_zval_ptr 只会简单地拷贝一份 zval，并不会修改引用计数，但是在 PHP5 中这是危险的。由于设计问题，在 PHP5 中调用 zend_get_zval_ptr 会认为你当前持有该变量，获取参数后会进行释放，所以接口内部会修改引用计数，导致返回 ZEND_USER_OPCODE_DISPATCH 调用原始处理函数时，无法获取到正确的入参。针对这个问题，我们需要使用大量的宏判断以适配各版本。同样为了更好地批量定义 wrapper，我编写了一套可配置模板：

template<int ClassID, int MethodID, bool Extended = false>
class OpcodeEntry {
public:
    static int entry(
            zend_execute_data *execute_data
#if PHP_MAJOR_VERSION <= 5
            TSRMLS_DC
#endif
    ) {
#if PHP_VERSION_ID >= 80000
        zval *op1 = zend_get_zval_ptr(execute_data->opline, execute_data->opline->op1_type, &execute_data->opline->op1, execute_data);
        zval *op2 = zend_get_zval_ptr(execute_data->opline, execute_data->opline->op2_type, &execute_data->opline->op2, execute_data);
#elif PHP_VERSION_ID >= 70300
        zend_free_op should_free;
        zval *op1 = zend_get_zval_ptr(execute_data->opline, execute_data->opline->op1_type, &execute_data->opline->op1, execute_data, &should_free, BP_VAR_IS);
        zval *op2 = zend_get_zval_ptr(execute_data->opline, execute_data->opline->op2_type, &execute_data->opline->op2, execute_data, &should_free, BP_VAR_IS);
#elif PHP_VERSION_ID >= 70000
        zend_free_op should_free;
        zval *op1 = zend_get_zval_ptr(execute_data->opline->op1_type, &execute_data->opline->op1, execute_data, &should_free, BP_VAR_IS);
        zval *op2 = zend_get_zval_ptr(execute_data->opline->op2_type, &execute_data->opline->op2, execute_data, &should_free, BP_VAR_IS);
#elif PHP_VERSION_ID >= 50500
        auto extract = [&](zend_uchar type, znode_op *node) {
            switch (type) {
                case IS_TMP_VAR:
                    return &EX_TMP_VAR(execute_data, node->var)->tmp_var;

                case IS_VAR:
                    return EX_TMP_VAR(execute_data, node->var)->var.ptr;

                default:
                    break;
            }

            zend_free_op should_free;

            return zend_get_zval_ptr(type, node, execute_data, &should_free, BP_VAR_IS TSRMLS_CC);
        };

        zval *op1 = extract(execute_data->opline->op1_type, &execute_data->opline->op1);
        zval *op2 = extract(execute_data->opline->op2_type, &execute_data->opline->op2);
#elif PHP_VERSION_ID >= 50400
        auto extract = [&](zend_uchar type, znode_op *node) {
            switch (type) {
                case IS_TMP_VAR:
                    return &((temp_variable *)((char *)execute_data->Ts + node->var))->tmp_var;

                case IS_VAR:
                    return ((temp_variable *)((char *)execute_data->Ts + node->var))->var.ptr;

                default:
                    break;
            }

            zend_free_op should_free;

            return zend_get_zval_ptr(type, node, execute_data->Ts, &should_free, BP_VAR_IS TSRMLS_CC);
        };

        zval *op1 = extract(execute_data->opline->op1_type, &execute_data->opline->op1);
        zval *op2 = extract(execute_data->opline->op2_type, &execute_data->opline->op2);
#else
        auto extract = [&](int type, znode *node) {
            switch (type) {
                case IS_TMP_VAR:
                    return &((temp_variable *)((char *)execute_data->Ts + node->u.var))->tmp_var;

                case IS_VAR:
                    return ((temp_variable *)((char *)execute_data->Ts + node->u.var))->var.ptr;

                default:
                    break;
            }

            zend_free_op should_free;

            return zend_get_zval_ptr(node, execute_data->Ts, &should_free, BP_VAR_IS TSRMLS_CC);
        };

        zval *op1 = extract(execute_data->opline->op1.op_type, &execute_data->opline->op1);
        zval *op2 = extract(execute_data->opline->op2.op_type, &execute_data->opline->op2);
#endif

        Trace trace = {
                ClassID,
                MethodID
        };

        strncpy(
                trace.args[trace.count++],
                toString(
                        op1
#if PHP_MAJOR_VERSION <= 5
                        TSRMLS_CC
#endif
                ).c_str(),
                SMITH_ARG_LENGTH - 1
        );

        strncpy(
                trace.args[trace.count++],
                toString(
                        op2
#if PHP_MAJOR_VERSION <= 5
                        TSRMLS_CC
#endif
                ).c_str(),
                SMITH_ARG_LENGTH - 1
        );

        if constexpr (Extended) {
#if PHP_VERSION_ID >= 50400
            strncpy(trace.args[trace.count++], std::to_string(execute_data->opline->extended_value).c_str(), SMITH_ARG_LENGTH - 1);
#else
            strncpy(trace.args[trace.count++], std::to_string(Z_LVAL(execute_data->opline->op2.u.constant)).c_str(), SMITH_ARG_LENGTH - 1);
#endif
        }

        std::vector<std::string> stackTrace = traceback(
#if PHP_MAJOR_VERSION <= 5
                TSRMLS_C
#endif
        );

        for (int i = 0; i < stackTrace.size() && i < SMITH_TRACE_COUNT; i++) {
            strncpy(trace.stackTrace[i], stackTrace[i].c_str(), SMITH_TRACE_LENGTH - 1);
        }

        trace.request = PHP_PROBE_G(request);

        if (!gAPIConfig->surplus(ClassID, MethodID))
            return ZEND_USER_OPCODE_DISPATCH;

        gAPITrace->enqueue(trace);

        return ZEND_USER_OPCODE_DISPATCH;
    }
};

其中比较特殊的是 Extended，这是因为 PHP 中存在一个特殊的 opcode，也就是 ZEND_INCLUDE_OR_EVAL，它可以看做多个子 opcode 的集合。为了区分具体的操作是 eval 或是 include，我们需要额外的获取一个标志位，供服务端进行区分判断。

阻断

Elkeid RASP 系列产品都支持阻断功能，下发的规则为正则格式，当某次函数调用的入参匹配到规则时，便会抛出异常：

......
        if constexpr (CanBlock) {
            if (gAPIConfig->block(trace)) {
                trace.blocked = true;

                gAPITrace->enqueue(trace);

                zend_throw_exception(
                        nullptr,
                        "API blocked by RASP",
                        0
#if PHP_MAJOR_VERSION <= 5
                        TSRMLS_CC
#endif
                );

                return;
            }
        }
......

在获取到入参后，查询共享内存中的规则，如果匹配则调用 zend_throw_exception 抛出异常。

性能

资源占用：

• 单个 PHP 进程的内存占用增长约 7M。
• 单个 PHP 进程的 CPU 增长小于 %2。
• 对于一个 PHP 进程组，额外创建一个通信进程。

耗时统计： | api | average(ns) | tp90(ns) | tp95(ns) | tp99(ns) | | ---- | ---- | ---- | ---- | ---- | | passthru | 15421.034781554868 | 13197.200000000004 | 15493.39999999999 | 237840.09999999954 | | system | 9150.411468751894 | 16550.6 | 19003.0 | 37398.719999999885 | | exec | 8766.203311700127 | 16398.6 | 18939.799999999996 | 34817.0999999997 | | shell_exec | 9421.62037992353 | 17205.0 | 19705.0 | 43732.91999999992 | | proc_open | 8735.06981968308 | 17033 | 19621.0 | 34133.599999999955 | | popen | 6197.004372381126 | 6662.4 | 8780.499999999995 | 17723.260000000006 | | file | 6078.228969122295 | 14572.9 | 16554.799999999996 | 24646.309999999954 | | readfile | 3194.912538746733 | 3672.600000000002 | 4886.5999999999985 | 9566.199999999993 | | file_get_contents | 5957.900887753861 | 7671.0 | 8890.25 | 16371.149999999947 | | file_put_contents | 3224.3691446648013 | 2868.0 | 3262.1499999999996 | 8150.669999999993 | | copy | 2873.517952775073 | 2972.0 | 3451.0 | 9257.470000000032 | | rename | 3356.0933333333332 | 4983.6 | 6922.5999999999985 | 12561.160000000003 | | unlink | 2542.6100176710743 | 2908 | 3829.5 | 9480.699999999988 | | dir | 2603.6219809709687 | 2741.5 | 3217.5 | 8634.199999999983 | | opendir | 3853.008359265361 | 5571.0 | 6505.5999999999985 | 10775.199999999964 | | scandir | 3196.0514355528408 | 4123.200000000001 | 4949.549999999999 | 8905.97999999997 | | fopen | 3161.4416473176097 | 3168.5 | 3542.75 | 8217.750000000016 | | move_uploaded_file | 2862.6720479422734 | 2884.0 | 3171.7999999999993 | 8084.879999999997 | | splfileobject::construct | 4243.37449516583 | 5447.799999999999 | 6497.949999999999 | 11698.980000000003 | | socket_connect | 4211.529264111669 | 4307.0 | 5049.44999999999 | 11996.7 | | gethostbyname | 5103.737816465396 | 6755.800000000001 | 8366.199999999999 | 13759.359999999962 | | dns_get_record | 6716.215250491159 | 9247.0 | 10155.65 | 21651.48999999994 | | putenv | 11979.022368659695 | 11790 | 13426 | 67583.00000000041 | | curl_exec | 10650.08878674247 | 12138.7 | 14603.049999999996 | 47374.10000000011 | | ZEND_INCLUDE_OR_EVAL | 6083.132776567417 | 5832.0 | 6501.399999999994 | 15980.07000000004 |

assert 在 PHP7 中已经不再属于函数，pcntl_exec 相当于 exec 系统调用，在 PHP-FPM 中被禁止使用。

NodeJS

简介

• 兼容性：Node 8.6.0+
• 代码：bytedance/Elkeid: rasp/node

Inspector

根据Node.js的官方文档得知，在Node.js 8之后加入了新的调试机制(Inspector API)，在启动时加入”–inspect”参数或者向进程发送SIGUSR1信号，就会激活检查器。

node
kill -USR1 $(pidof node)

激活检查器后，终端会打印信息：

Debugger listening on ws://127.0.0.1:9229/f1f12ee0-4f35-4e61-836c-71d175a607e3

检查器监听本地端口9929，使用websocket协议，协议约定可见文档，连接检查器可以使用node自带的inspect client：

node inspect -p $(pidof node)

连接后可以通过”exec(‘console.log(123)’)”在目标进程中执行node代码，或者执行”repl”开启交互模式。但是自带的inspect client功能简单，如果需要更好的体验，可以使用chrome的调试模块，在chrome中访问”chrome://inspect”即可。

协议

node-inspect 就是检查器客户端的官方代码仓库，通过阅读 inspect_client.js发现，交互协议就是简单的json字符串，每个请求和响应通过id字段一一对应。 例如请求开启调试时，通过websocket协议发送字符串：

{
    "id": 1,
    "method": "Debugger.enable"
}

检查器返回响应：

{
    "id": 1,
    "result": {
        "debuggerId": "(D750AD89818A150E3155AC1D6ECB7BB)"
    }
}

如果你想在目标进程中执行代码，类似于”exec(‘console.log(123)’)”命令，你只需要使用Runtime.evaluate，发送请求：

{
    "id": 2,
    "method": "Runtime.evaluate",
    "params": {
        "expression": "console.log(123)",
        "includeCommandLineAPI": true
    }
}

参数需要设置”includeCommandLineAPI”，否则注入的代码无法使用require函数。

API

通过使用inspector在目标进程中执行"require('smith.js')"，进行API的hook：

const path = require('path');
const inspector = require('inspector');
const { SmithClient, OperateEnum } = require('./client');
const { baseProcess, spawnProcess, spawnSyncProcess, connectProcess } = require('./process');

const fs = require('fs');
const net = require('net');
const dns = require('dns');
const child_process = require('child_process');

const smith_client = new SmithClient();

smith_client.on('message', (message) => {
    switch (message.message_type) {
        case OperateEnum.detect:
            if (typeof require.main === 'undefined')
                break;

            const root = path.dirname(require.main.filename);
            const package = path.join(root, 'package.json');

            if (fs.existsSync(package)) {
                const data = fs.readFileSync(package);
                smith_client.postMessage(OperateEnum.detect, {'node': JSON.parse(data)});
            }

            break;

        default:
            break;
    }
});

smith_client.connect();

function smithHook(fn, classID, methodID, process=baseProcess) {
    return function(...args) {
        const smithTrace = {
            'class_id': classID,
            'method_id': methodID,
            'args': args.map(process),
            'stack_trace': new Error().stack.split('\n').slice(1).map(s => s.trim())
        }

        smith_client.postMessage(OperateEnum.trace, smithTrace);

        return fn.call(this, ...args);
    }
}

child_process.ChildProcess.prototype.spawn = smithHook(child_process.ChildProcess.prototype.spawn, 0, 0, spawnProcess);
child_process.spawnSync = smithHook(child_process.spawnSync, 0, 1, spawnSyncProcess);
child_process.execSync = smithHook(child_process.execSync, 0, 2);
child_process.execFileSync = smithHook(child_process.execFileSync, 0, 3);

fs.open = smithHook(fs.open, 1, 0);
fs.openSync = smithHook(fs.openSync, 1, 1);
fs.readFile = smithHook(fs.readFile, 1, 2);
fs.readFileSync = smithHook(fs.readFileSync, 1, 3);
fs.readdir = smithHook(fs.readdir, 1, 4);
fs.readdirSync = smithHook(fs.readdirSync, 1, 5);
fs.unlink = smithHook(fs.unlink, 1, 6);
fs.unlinkSync = smithHook(fs.unlinkSync, 1, 7);
fs.rmdir = smithHook(fs.rmdir, 1, 8);
fs.rmdirSync = smithHook(fs.rmdirSync, 1, 9);
fs.rename = smithHook(fs.rename, 1, 10);
fs.renameSync = smithHook(fs.renameSync, 1, 11);

net.Socket.prototype.connect = smithHook(net.Socket.prototype.connect, 2, 0, connectProcess);

dns.lookup = smithHook(dns.lookup, 3, 0);
dns.resolve = smithHook(dns.resolve, 3, 1);
dns.resolve4 = smithHook(dns.resolve4, 3, 2);
dns.resolve6 = smithHook(dns.resolve6, 3, 3);

eval = smithHook(eval, 4, 0);

if (inspector.url()) {
    setTimeout(() => {
        inspector.close();
    }, 500);
}

API列表：

• process
  • child_process.ChildProcess.prototype.spawn
  • child_process.spawnSync
  • child_process.execSync
  • child_process.execFileSync
• fs
  • fs.open
  • fs.openSync
  • fs.readFile
  • fs.readFileSync
  • fs.readdir
  • fs.readdirSync
  • fs.unlink
  • fs.unlinkSync
  • fs.rmdir
  • fs.rmdirSync
  • fs.rename
  • fs.renameSync
• net
  • net.Socket.prototype.connect
• dns
  • dns.lookup
  • dns.resolve
  • dns.resolve4
  • dns.resolve6

数据格式

使用测试代码：

const fs = require('fs');
const net = require('net');
const dns = require('dns');
const child_process = require('child_process');

function testProcess() {
    child_process.exec('ls');
}

function testFile() {
    fs.open('/tmp/test', 'w', () => {});
    fs.readFile('/tmp/test', () => {});
    fs.readdir('/tmp', () => {});
    fs.rename('/tmp/test', '/tmp/test1', () => {});
    fs.unlink('/tmp/test1', () => {});
}

function testNetwork() {
    net.connect(80, 'qq.com', () => {});
}

function testDns() {
    dns.lookup('qq.com', () => {});
}

setInterval(() => {
    testProcess();
    testFile();
    testNetwork();
    testDns();
}, 3000);

运行代码：

node test.json

node <path>/injector.js <pid of node> 'require("<rasp node path>")'

数据输出：

[{
    "pid": 3371235,
    "runtime": "node.js",
    "runtime_version": "v9.0.0",
    "time": 1614065334018,
    "message_type": 2,
    "probe_version": "1.0.0",
    "data": {
        "class_id": 0,
        "method_id": 0,
        "args": ["/bin/sh -c ls"],
        "stack_trace": ["at ChildProcess.spawn (/data00/home/liupan.patte/node-probe/smith.js:31:28)", "at exports.spawn (child_process.js:499:9)", "at Object.exports.execFile (child_process.js:209:15)", "at Object.exports.exec (child_process.js:139:18)", "at testProcess (/data00/home/liupan.patte/node-probe/test.js:7:19)", "at Timeout.setInterval [as _onTimeout] (/data00/home/liupan.patte/node-probe/test.js:27:5)", "at ontimeout (timers.js:478:11)", "at tryOnTimeout (timers.js:302:5)", "at Timer.listOnTimeout (timers.js:262:5)"]
    }
}, {
    "pid": 3371235,
    "runtime": "node.js",
    "runtime_version": "v9.0.0",
    "time": 1614065334021,
    "message_type": 2,
    "probe_version": "1.0.0",
    "data": {
        "class_id": 1,
        "method_id": 0,
        "args": ["/tmp/test", "w", null],
        "stack_trace": ["at Object.open (/data00/home/liupan.patte/node-probe/smith.js:31:28)", "at testFile (/data00/home/liupan.patte/node-probe/test.js:11:8)", "at Timeout.setInterval [as _onTimeout] (/data00/home/liupan.patte/node-probe/test.js:28:5)", "at ontimeout (timers.js:478:11)", "at tryOnTimeout (timers.js:302:5)", "at Timer.listOnTimeout (timers.js:262:5)"]
    }
}, {
    "pid": 3371235,
    "runtime": "node.js",
    "runtime_version": "v9.0.0",
    "time": 1614065334022,
    "message_type": 2,
    "probe_version": "1.0.0",
    "data": {
        "class_id": 1,
        "method_id": 3,
        "args": ["/tmp/test", null],
        "stack_trace": ["at Object.readFile (/data00/home/liupan.patte/node-probe/smith.js:31:28)", "at testFile (/data00/home/liupan.patte/node-probe/test.js:12:8)", "at Timeout.setInterval [as _onTimeout] (/data00/home/liupan.patte/node-probe/test.js:28:5)", "at ontimeout (timers.js:478:11)", "at tryOnTimeout (timers.js:302:5)", "at Timer.listOnTimeout (timers.js:262:5)"]
    }
}, {
    "pid": 3371235,
    "runtime": "node.js",
    "runtime_version": "v9.0.0",
    "time": 1614065334022,
    "message_type": 2,
    "probe_version": "1.0.0",
    "data": {
        "class_id": 1,
        "method_id": 5,
        "args": ["/tmp", null],
        "stack_trace": ["at Object.readdir (/data00/home/liupan.patte/node-probe/smith.js:31:28)", "at testFile (/data00/home/liupan.patte/node-probe/test.js:13:8)", "at Timeout.setInterval [as _onTimeout] (/data00/home/liupan.patte/node-probe/test.js:28:5)", "at ontimeout (timers.js:478:11)", "at tryOnTimeout (timers.js:302:5)", "at Timer.listOnTimeout (timers.js:262:5)"]
    }
}, {
    "pid": 3371235,
    "runtime": "node.js",
    "runtime_version": "v9.0.0",
    "time": 1614065334022,
    "message_type": 2,
    "probe_version": "1.0.0",
    "data": {
        "class_id": 1,
        "method_id": 11,
        "args": ["/tmp/test", "/tmp/test1", null],
        "stack_trace": ["at Object.rename (/data00/home/liupan.patte/node-probe/smith.js:31:28)", "at testFile (/data00/home/liupan.patte/node-probe/test.js:14:8)", "at Timeout.setInterval [as _onTimeout] (/data00/home/liupan.patte/node-probe/test.js:28:5)", "at ontimeout (timers.js:478:11)", "at tryOnTimeout (timers.js:302:5)", "at Timer.listOnTimeout (timers.js:262:5)"]
    }
}, {
    "pid": 3371235,
    "runtime": "node.js",
    "runtime_version": "v9.0.0",
    "time": 1614065334022,
    "message_type": 2,
    "probe_version": "1.0.0",
    "data": {
        "class_id": 1,
        "method_id": 7,
        "args": ["/tmp/test1", null],
        "stack_trace": ["at Object.unlink (/data00/home/liupan.patte/node-probe/smith.js:31:28)", "at testFile (/data00/home/liupan.patte/node-probe/test.js:15:8)", "at Timeout.setInterval [as _onTimeout] (/data00/home/liupan.patte/node-probe/test.js:28:5)", "at ontimeout (timers.js:478:11)", "at tryOnTimeout (timers.js:302:5)", "at Timer.listOnTimeout (timers.js:262:5)"]
    }
}, {
    "pid": 3371235,
    "runtime": "node.js",
    "runtime_version": "v9.0.0",
    "time": 1614065334023,
    "message_type": 2,
    "probe_version": "1.0.0",
    "data": {
        "class_id": 2,
        "method_id": 0,
        "args": [
            [{
                "port": 80,
                "host": "baidu.com"
            }, null]
        ],
        "stack_trace": ["at Socket.connect (/data00/home/liupan.patte/node-probe/smith.js:31:28)", "at Object.connect (net.js:107:35)", "at testNetwork (/data00/home/liupan.patte/node-probe/test.js:19:9)", "at Timeout.setInterval [as _onTimeout] (/data00/home/liupan.patte/node-probe/test.js:29:5)", "at ontimeout (timers.js:478:11)", "at tryOnTimeout (timers.js:302:5)", "at Timer.listOnTimeout (timers.js:262:5)"]
    }
}, {
    "pid": 3371235,
    "runtime": "node.js",
    "runtime_version": "v9.0.0",
    "time": 1614065334023,
    "message_type": 2,
    "probe_version": "1.0.0",
    "data": {
        "class_id": 3,
        "method_id": 0,
        "args": ["baidu.com", {
            "hints": 32
        }, null],
        "stack_trace": ["at /data00/home/liupan.patte/node-probe/smith.js:31:28", "at lookupAndConnect (net.js:1100:3)", "at Socket.connect (net.js:1034:5)", "at Socket.connect (/data00/home/liupan.patte/node-probe/smith.js:46:19)", "at Object.connect (net.js:107:35)", "at testNetwork (/data00/home/liupan.patte/node-probe/test.js:19:9)", "at Timeout.setInterval [as _onTimeout] (/data00/home/liupan.patte/node-probe/test.js:29:5)", "at ontimeout (timers.js:478:11)", "at tryOnTimeout (timers.js:302:5)", "at Timer.listOnTimeout (timers.js:262:5)"]
    }
}, {
    "pid": 3371235,
    "runtime": "node.js",
    "runtime_version": "v9.0.0",
    "time": 1614065334023,
    "message_type": 2,
    "probe_version": "1.0.0",
    "data": {
        "class_id": 3,
        "method_id": 0,
        "args": ["baidu.com", null],
        "stack_trace": ["at Object.lookup (/data00/home/liupan.patte/node-probe/smith.js:31:28)", "at testDns (/data00/home/liupan.patte/node-probe/test.js:23:9)", "at Timeout.setInterval [as _onTimeout] (/data00/home/liupan.patte/node-probe/test.js:30:5)", "at ontimeout (timers.js:478:11)", "at tryOnTimeout (timers.js:302:5)", "at Timer.listOnTimeout (timers.js:262:5)"]
    }
}]

问题
修改module.exports进行hook，无法影响同模块中的函数调用。
// a.js
function A() {
    console.log('A called');
}

function B() {
    A();
}

module.exports = {
    A,
    B
};

// b.js
const m = require('./a')

m.B();
m.A = () => {console.log('A hooked')};
m.B();

替换了m.A之后，外部对模块a.A的调用会被hook，但是m.B中对A的调用依旧使用原函数。

Agent

关于 Elkeid Agent

Agent提供端上组件的基本能力支撑，包括数据通信、资源监控、组件版本控制、文件传输、机器基础信息采集等。

Agent本身不提供安全能力，作为一个插件底座以系统服务的方式运行。各类功能插件的策略存放于服务器端的配置，Agent接收到相应的控制指令及配置后对自身及插件进行开启、关闭、升级等操作。

Agent与Server之间采用bi-stream gRPC进行通信，基于自签名证书开启双向TLS验证，保障信道安全。其中，Agent -> Server 方向信息流动称为数据流，Server -> Agent 方向信息流动一般为控制流，使用protobuf的不同message类型。Agent本身支持客户端侧服务发现，也支持跨Region级别的通信配置，实现一个Agent包在多个网络隔离的环境下进行安装，基于底层一个TCP连接，在上层实现了Transfer与FileOp两种数据传输服务，支撑了插件本身的数据上报与Host中的文件交互。

Plugins作为安全能力插件，与Agent的进程关系一般为“父——子”进程。以两个pipe作为跨进程通信方式，plugins lib提供了Go与Rust的两个插件库，负责插件端信息的编码与发送。值得一提的是，插件发送数据后，会被编码为Protobuf二进制数据，Agent接收到后无需二次解编码，再其外层拼接好Header特征数据，直接传输给Server，一般情况下Server也无需解码，直接传输至后续数据流中，使用时进行解码，一定程度上降低了数据传输中多次编解码造成的额外性能开销。

Agent采用Go实现，在Linux下，通过systemd作为守护方式，受cgroup限制控制资源使用，支持aarch64与x86-64架构，最终编译、打包为deb与rpm包分发，格式均符合systemd及Debian、RHEL规范，可以直接提供至对应的软件仓库中进行后续版本维护。在后续版本中，将会发布用于Windows平台下的Agent。

运行时要求

Agent及Plugin提供的大部分功能需要以root权限运行在宿主机(Host)层面，在权限受限的容器中，部分功能可能会存在异常。

快速开始

通过 elkeidup 的完整部署，可以直接得到用于Debian/RHEL系列发行版的安装包，并按照 Elkeid Console - 安装配置 界面的命令进行安装部署。

手动编译

环境要求

• Go >= 1.18
• nFPM
• 成功部署的 Server (包含所有组件)

确认相关配置

• 需要确保 transport/connection 目录下的 ca.crt、client.key、client.crt 三个文件与Agent Center conf 目录下的同名文件保持一致。
• 需要确保 transport/connection/product.go 文件中的参数都配置妥当：
  • 如果是手动部署的Server：
    • serviceDiscoveryHost["default"] 需被赋值为 ServiceDiscovery 服务或代理服务的内网监听地址与端口，例如：serviceDiscoveryHost["default"] = "192.168.0.1:8088"
    • privateHost["default"] 需被赋值为 AgentCenter 服务或代理服务的内网监听地址与端口，例如：privateHost["default"] = "192.168.0.1:6751"
    • 如有Server的公网接入点，publicHost["default"] 需被赋值为 AgentCenter 服务或代理服务的外网监听地址与端口，例如：publicHost["default"]="203.0.113.1:6751"
  • 如果是通过 elkeidup 部署的Server，可以根据部署Server机器的 ~/.elkeidup/elkeidup_config.yaml 文件获得对应配置：
    • 在配置文件中找到 Nginx 服务的IP，具体的配置项为 nginx.sshhost[0].host
    • 在配置文件中找到 ServiceDiscovery 服务的IP，具体的配置项为 sd.sshhost[0].host
    • serviceDiscoveryHost["default"] 需被赋值为 ServiceDiscovery 服务的IP，并将端口号设置为8088，例如：serviceDiscoveryHost["default"] = "192.168.0.1:8088"
    • privateHost["default"] 需被赋值为 Nginx 服务的IP，并将端口号设置为8090，例如：privateHost["default"] = "192.168.0.1:8090"

编译

在Agent根目录，执行：

BUILD_VERSION=1.7.0.24 bash build.sh

在编译过程中，脚本会读取 BUILD_VERSION 环境变量设置版本信息，可根据实际需要进行修改。

编译成功后，在根目录的 output 目录下，应该可以看到2个deb与2个rpm文件，它们分别对应不同的系统架构。

版本升级

1. 如果没有创建过客户端类型的组件，请在 Elkeid Console-组件管理 界面新建对应组件。
2. 在 Elkeid Console - 组件管理 界面，找到“elkeid-agent”条目，点击右侧“发布版本”，填写版本信息并上传对应平台与架构的文件，点击确认。
3. 在 Elkeid Console - 组件策略 界面，(如有)删除旧的“elkeid-agent”版本策略，点击“新建策略”，选中刚刚发布的版本，点击确认。后续新安装的Agent均会自升级到最新版本。
4. 在 Elkeid Console - 任务管理 界面，点击“新建任务”，选择全部主机，点击下一步，选择“同步配置”任务类型，点击确认。随后，在此页面找到刚刚创建的任务，点击运行，即可对存量旧版本Agent进行升级。

License

Elkeid Agent is distributed under the Apache-2.0 license.

RASP plugin

制品形式

• 包含有 rasp 插件以及各 runtime 探针的 tar.gz 压缩包。
• 可在 bytedance/Elkeid: release 页面下载最新的制品。

组成

RASP 插件包含以下部件：

├── rasp 插件主入口，由 agent 启动，与 agent 双向通信，与探针双向通信。
├── settings.toml 插件配置文件。
├── elkeid_rasp 仅用于对指定进程进行探针植入与通信使用，数据输出到 stdout。
├── lib-1.9.1.68 包含有各探针的目录，与版本号一致。
│     ├── golang
│     ├── java
│     ├── node
│     ├── pangolin
│     ├── php
│     ├── python
│     └── rasp_server
├── nsenter
└── mount_script

driver plugin

源码：https://github.com/bytedance/Elkeid/tree/main/plugins/driver

关于driver插件

driver 插件对内核模块进行管理，并利用用户态数据进行补充、过滤，最终生成不同的系统事件，支撑相关告警功能。

运行时要求

支持主流的Linux发行版，包括CentOS、RHEL、Debian、Ubuntu、RockyLinux、OpenSUSE等。支持x86-64与aarch64架构。

主机的内核版本需要在支持的的列表中，如果不在，需要单独编译并上传，详见 说明 。

快速开始

通过 elkeidup 的完整部署，此插件默认开启。

手动编译

环境要求

• Rust >= 1.48.0
• Rust x86_64-unknown-linux-musl aarch64-unknown-linux-musl 编译链
• musl-gcc

确认相关配置

• 需要确保 src/config.rs 中 DOWNLOAD_HOSTS 变量已配置为实际部署的Nginx服务地址：
  • 如果是手动部署的Server：需要确保将其配置为Nginx文件服务的地址，例如：pub const DOWNLOAD_HOSTS: &'static [&'static str] = &["http://192.168.0.1:8080"];
  • 如果是通过 elkeidup 部署的Server，可以根据部署Server机器的 ~/.elkeidup/elkeidup_config.yaml 文件获得对应配置，具体配置项为 nginx.sshhost[0].host，并将端口号设置为8080，例如：pub const DOWNLOAD_HOSTS: &'static [&'static str] = &["http://192.168.0.1:8080"];

编译

在根目录，执行：

BUILD_VERSION=1.0.0.15 bash build.sh

在编译过程中，脚本会读取 BUILD_VERSION 环境变量设置版本信息，可根据实际需要进行修改。

编译成功后，在根目录的 output 目录下，应该可以看到2个plg文件，它们分别对应不同的系统架构。

版本升级

1. 如果没有创建过客户端类型的组件，请在 Elkeid Console-组件管理 界面新建对应组件。
2. 在 Elkeid Console - 组件管理 界面，找到“elkeid-agent”条目，点击右侧“发布版本”，填写版本信息并上传对应平台与架构的文件，点击确认。
3. 在 Elkeid Console - 组件策略 界面，(如有)删除旧的“elkeid-agent”版本策略，点击“新建策略”，选中刚刚发布的版本，点击确认。后续新安装的Agent均会自升级到最新版本。
4. 在 Elkeid Console - 任务管理 界面，点击“新建任务”，选择全部主机，点击下一步，选择“同步配置”任务类型，点击确认。随后，在此页面找到刚刚创建的任务，点击运行，即可对存量旧版本Agent进行升级。

License

driver plugin is distributed under the Apache-2.0 license.

scanner

源码：https://github.com/bytedance/Elkeid/tree/main/plugins/scanner

关于 Yara Scanner 插件

Yara Scanner 使用 yara 规则对系统进程和敏感目录进行周期扫描，并使用fanotify监控敏感目录文件变动，以发现可疑静态文件（UPX/挖矿二进制/挖矿脚本/可疑脚本文件/...）。

平台兼容性

与Elkeid Agent相同。

其他

参考 https://github.com/bytedance/Elkeid/blob/main/plugins/scanner/README-zh_CN.md

collector

多种常见资产采集

源码：https://github.com/bytedance/Elkeid/tree/main/plugins/collector

关于collector插件

collector周期性采集主机上的各类资产信息，并进行关联分析，目前支持以下资产类型：

• 进程：支持对exe md5的哈希计算，后续可关联威胁情报分析，另外与容器信息进行关联，支撑后续数据溯源功能。(跨容器)
• 端口：支持tcp、udp监听端口的信息提取，以及与进程、容器信息的关联上报。另外基于sock状态及其关系，分析对外暴露服务，向上支撑主机暴露面分析功能。(跨容器)
• 账户：除了基本的账户字段外，基于弱口令字典进行端上hash碰撞检测弱口令，向上提供了Console的弱口令基线检测功能。另外，会关联分析sudoers配置，一同上报。
• 软件：支持系统软件包、pypi包、jar包，向上支撑漏洞扫描功能。(部分跨容器)
• 容器：支持docker、cri/containerd等多种运行时下的容器信息采集。
• 应用：支持数据库、消息队列、容器组件、Web服务、DevOps工具等类型的应用采集、目前支持30+中常见应用的版本、配置文件的匹配与提取。(跨容器)
• 硬件：支持网卡、磁盘等硬件信息的采集。
• 系统完整性校验：通过将软件包文件哈希与Host实际文件哈希进行对比，判断文件是否有被更改。
• 内核模块：采集基本字段，以及内存地址、依赖关系等额外字段。
• 系统服务、定时任务：兼容不同发行版下的服务及cron位置的定义，并对核心字段进行解析。

运行时要求

支持主流的Linux发行版，包括CentOS、RHEL、Debian、Ubuntu、RockyLinux、OpenSUSE等。支持x86-64与aarch64架构。

快速开始

通过 elkeidup 的完整部署，此插件默认开启。

手动编译

环境要求

• Go >= 1.18

编译

在Agent根目录，执行：

BUILD_VERSION=1.7.0.140 bash build.sh

在编译过程中，脚本会读取 BUILD_VERSION 环境变量设置版本信息，可根据实际需要进行修改。

编译成功后，在根目录的 output 目录下，应该可以看到2个deb与2个rpm文件，它们分别对应不同的系统架构。

版本升级

1. 如果没有创建过客户端类型的组件，请在 Elkeid Console-组件管理 界面新建对应组件。
2. 在 Elkeid Console - 组件管理 界面，找到“elkeid-agent”条目，点击右侧“发布版本”，填写版本信息并上传对应平台与架构的文件，点击确认。
3. 在 Elkeid Console - 组件策略 界面，(如有)删除旧的“elkeid-agent”版本策略，点击“新建策略”，选中刚刚发布的版本，点击确认。后续新安装的Agent均会自升级到最新版本。
4. 在 Elkeid Console - 任务管理 界面，点击“新建任务”，选择全部主机，点击下一步，选择“同步配置”任务类型，点击确认。随后，在此页面找到刚刚创建的任务，点击运行，即可对存量旧版本Agent进行升级。

License

collector is distributed under the Apache-2.0 license.

journal_watcher

源码：https://github.com/bytedance/Elkeid/blob/main/plugins/journal_watcher

关于Journal Watcher插件

Journal Watcher 通过读取 sshd 的日志，并进行解析，生成sshd登陆、gssapi事件。

运行时要求

支持主流的Linux发行版，包括CentOS、RHEL、Debian、Ubuntu、RockyLinux、OpenSUSE等。支持x86-64与aarch64架构。

快速开始

通过 elkeidup 的完整部署，此插件默认开启。

手动编译

环境要求

• Rust >= 1.48.0
• Rust x86_64-unknown-linux-musl aarch64-unknown-linux-musl 编译链
• musl-gcc

编译

在根目录，执行：

BUILD_VERSION=1.7.0.23 bash build.sh

在编译过程中，脚本会读取 BUILD_VERSION 环境变量设置版本信息，可根据实际需要进行修改。

编译成功后，在根目录的 output 目录下，应该可以看到2个plg文件，它们分别对应不同的系统架构。

版本升级

1. 如果没有创建过客户端类型的组件，请在 Elkeid Console-组件管理 界面新建对应组件。
2. 在 Elkeid Console - 组件管理 界面，找到“elkeid-agent”条目，点击右侧“发布版本”，填写版本信息并上传对应平台与架构的文件，点击确认。
3. 在 Elkeid Console - 组件策略 界面，(如有)删除旧的“elkeid-agent”版本策略，点击“新建策略”，选中刚刚发布的版本，点击确认。后续新安装的Agent均会自升级到最新版本。
4. 在 Elkeid Console - 任务管理 界面，点击“新建任务”，选择全部主机，点击下一步，选择“同步配置”任务类型，点击确认。随后，在此页面找到刚刚创建的任务，点击运行，即可对存量旧版本Agent进行升级。

License

journal_watcher plugin is distributed under the Apache-2.0 license.

Baseline

基线插件通过已有或自定义的基线策略对资产进行检测，来判断资产上的基线安全配置是否存在风险。基线插件每天定时扫描一次，同时也可以通过前端进行立即检查。

平台兼容性

• centos 6,7,8
• debian 8,9,10
• ubuntu 14.04-20.04
  (其余版本以及发行版理论兼容)

需要的编译环境

• Go >= 1.18

编译

rm -rf output
mkdir output
# x86_64
go build -o baseline main.go
tar -zcvf baseline-linux-x86_64.tar.gz baseline config
mv baseline-linux-x86_64.tar.gz output/
# arch64
GOARCH=arm64 go build -o baseline main.go
tar -zcvf baseline-linux-x86_64.tar.gz baseline config
mv baseline-linux-x86_64.tar.gz output/

编译成功后，在根目录的 output 目录下，应该可以看到2个deb与2个rpm文件，它们分别对应不同的系统架构。

版本升级

1. 如果没有创建过客户端类型的组件，请在 Elkeid Console-组件管理 界面新建对应组件。
2. 在 Elkeid Console - 组件管理 界面，找到“elkeid-agent”条目，点击右侧“发布版本”，填写版本信息并上传对应平台与架构的文件，点击确认。
3. 在 Elkeid Console - 组件策略 界面，(如有)删除旧的“elkeid-agent”版本策略，点击“新建策略”，选中刚刚发布的版本，点击确认。后续新安装的Agent均会自升级到最新版本。
4. 在 Elkeid Console - 任务管理 界面，点击“新建任务”，选择全部主机，点击下一步，选择“同步配置”任务类型，点击确认。随后，在此页面找到刚刚创建的任务，点击运行，即可对存量旧版本Agent进行升级。

基线配置

常规配置

基线插件的规则通过yaml文件配置，其中主要包括以下字段(建议参照config文件下实际配置对比)：

check_id: 检查项id(int)
type: "类型(英文)"
title: "标题(英文)"
description: "描述(英文)"
solution: "解决方案(英文)"
security: "安全等级(high/mid/low)"
type_cn: "类型"
title_cn: "标题"
description_cn: "描述"
solution_cn: "解决方案"
check: # 检查规则（详见自定义规则）
    rules:
    - type: "file_line_check"
        param:
        - "/etc/login.defs"
        filter: '\s*\t*PASS_MAX_DAYS\s*\t*(\d+)'
        result: '$(<=)90'

自定义规则

每个检查项配置中的"check.rules"字段即为匹配规则，下边对每个字段进行解释：

rules.type

检查方式，目前baseline插件适配的内置检测规则(src/check/rules.go)包括如下几种：
| 检测规则 | 含义 | 参数 | 返回值 | | ---- | ---- | ---- | ---- | | command_check | 运行命令行语句 | 1：命令行语句
2：特殊参数(如ignore_exit 认为当命令行报错时认为通过检测) | 命令运行结果 | file_line_check | 遍历文件，逐行匹配 | 1：文件绝对路径
2：该行的flag（用于快速筛选行，减轻正则匹配的压力）
3：该文件的注释符(默认为#) | true/false/正则筛选值 | file_permission | 检测文件权限是否符合安全配置 | 1： 文件绝对路径
2： 文件最小权限(基于8进制，如644) | true/false | if_flie_exist | 判断文件是否存在 | 1： 文件绝对路径 | true/false | file_user_group | 判断文件用户组 | 1： 文件绝对路径
2： 用户组id | true/false | file_md5_check | 判断文件MD5是否一致 | 1： 文件绝对路径
2： MD5 | true/false | func_check | 通过特殊基线规则判断 | 1： 目标规则 | true/false

rules.param

规则参数数组

rules.require

规则前提条件：一些安全基线配置可能会存在检测前提条件，如果满足了先决条件后才会存在安全隐患，如：

rules:
  - type: "file_line_check"
    require: "allow_ssh_passwd"
    param:
        - "/etc/ssh/sshd_config"
    filter: '^\s*MaxAuthTries\s*\t*(\d+)'
    result: '$(<)5'

allow_ssh_passwd: 允许用户通过ssh密码登录

rules.result

检测结果，支持int，string，bool类型结果， 其中被*$()*为特殊检测语法，以下为部分语法示例： | 检测语法 | 说明 | 示例 | 示例含义 | | ---- | ---- | ---- | ---- | | $(&&) | 包含条件 | ok$(&&)success| 结果为ok或success | | $(<=) | 常见运算符 | $(<=)4| 结果小于等于4 | | $(not) | 结果取反 | $(not)error| 结果不为error |

复杂示例：

$(<)8$(&&)$(not)2  :  目标小于8且目标不为2

check.condition

由于规则的rule可能存在多个，因此可以通过condition字段定义规则之间的关系
all: 全部规则命中,则通过检测
any: 任一规则命中,则通过检测
none: 无规则命中,则通过检测

下发任务

任务下发

{
    "baseline_id": 1200, // 基线id
    "check_id_list":[1,2,3] // 扫描检查项列表(空列表为全部扫描)
}

结果回传

{
    "baseline_id": 1200,    // 基线id
    "status": "success",    // 检测状态success|error
    "msg": "",  // 错误原因
    "check_list":[
        "check_id": 1,  // 检查项id
        "type": "",     // 类型(英文)
        "title": "",   // 检查项标题(英文)
        "description": "",  // 检查项描述(英文)
        "solution": "",     // 解决方案(英文)
        "type_cn": "",     // 类型
        "title_cn": "",     // 检查项标题
        "description_cn": "",   // 检查项描述
        "solution_cn": "",  // 解决方案
        "result": "",   // 检查结果
        "msg": "",   // 错误原因
    ]
}

检查结果：

SuccessCode 		= 1  // 成功
FailCode 		= 2  // 失败
ErrorCode 		= -1 // 其他错误
ErrorConfigWrite	= -2 // 配置编写不规范
ErrorFile		= -3 // 文件读写异常

Elkeid-HUB

Elkeid HUB 是一款由 Elkeid Team 维护的通用的规则/事件处理引擎，支持流式/离线数据处理，初衷是通过标准化的抽象语法/规则来解决复杂的数据/事件处理与外部系统联动需求。

核心组件

应用场景

• Simple HIDS

• IDS Like Scenarios

• Multiple input and output scenarios

优势

• 高性能
• 依赖极少
• 支持复杂数据处理
• 插件支持
• 支持有状态逻辑
• 支持外部系统/数据联动

在真实世界使用

• 使用 Elkeid HUB 处理 Elkeid HIDS/RASP/Sandbox 原始数据，TPS 9千万/秒。HUB 调度实例 4000+
• 99% 告警产生时间 < 0.5s
• 内部维护策略2000+

快速部署

Elkeid 完整部署 Elkeid HUB 单独部署

快速上手

Elkeid-HUB Quick Start

手册

Handbook

LICENSE (对商用不友好)

Elkeid License

Contact us && Cooperation

Elkeid HUB 社区版使用指南

适用版本

Elkeid HUB 社区版

1 概述

Elkeid Hub 是为了解决各领域内的 数据上报-实时流处理与事件处理 中的 实时处理 的需求而诞生的产品，适用于如信息安全领域中的入侵检测/事件处理与编排等场景。

2 Elkeid HUB 优势

• 高性能，且支持分布式横行扩展
• 策略编写简单易懂，相关同学只需专注数据处理本身且学习成本很低
• 支持插件，可以更好的处理复杂需求
• 部署简单，依赖少

3 Elkeid HUB 组件概述

Elkeid HUB 组件主要分为一下几种：

• Input：数据通过Input组件消费后进入到Elkeid HUB，目前支持Kafka
• Output：数据通过Output组件将Elkeid HUB中流转的数据推送到HUB之外，目前支持ES/Kafka
• RuleSet：数据处理的核心逻辑将通过RuleSet编写，如检测(支持正则，多模匹配，频率等多种检测方式)/白名单/调用Plugin等
• Plugin：用户可以自定义任意检测/响应逻辑，来满足一些复杂场景的需求，如发送告警到钉钉/飞书；与CMDB联动进行数据补充；联动Redis进行一些缓存处理等。编写完成后可以在RuleSet中调用这些插件
• Project：通过Project来构建一组Elkeid HUB逻辑，通常是由一个Input+一个或多个RuleSet+一个或多个Output组成

4 Elkeid Input/输入源

4.1 Input配置建议

• 输入源目前仅支持Kafka作为数据输入源，数据格式支持Json，任意字符分割的日志两种。
• 数据源类型建议:
  • **其他安全产品的告警日志，**HUB可以有效的对其进行二次处理，如联动其他基础组件做到对告警数据的缺失部分弥补，也可以通过自定义Action(规则引擎支持)来实现对告警进行自动化处理已完成告警-处置的闭环操作
  • 基础日志，如HTTP镜像流量/Access日志/HIDS数据等。通过规则引擎，联动模块和异常检测模块对原始数据进行安全分析/入侵检测/自动化防御等操作
  • **审计日志，**如:登陆，原始的SQL请求，敏感操作等日志。HUB可以对其进行自定义的审计功能

4.2 Input配置说明

配置字段：

配置举例(DataType: json):

InputID: wafdeny
InputName: wafdeny
InputType: kafka
DataType: json
TTL: 30
KafkaBootstrapServers: secmq1.xxx.com:9092,secmq2.xxx.com:9092,secmq3.xxx.com:9092,secmq4.xxx.com:9092
KafkaGroupId: smith-hub
KafkaOffsetReset: latest
KafkaCompression: none
KafkaWorkerSize: 3
KafkaOtherConf: ~

KafkaTopics:
  - wafdeny

5 Elkeid Output/输出

5.1 Output配置建议

目前默认HUB的默认策略报警不走Output，使用插件直接与Manager交互写入数据库，如果需要配置原始告警可以考虑配置Output

由于HUB是一款流数据处理工具，不具备存储能力，建议配置数据存储的输出源，如ES、Hadoop、SOC、SIEM等平台。目前支持的Output类型有：

• Kafka
• Elasticsearch

5.2 Output配置说明

配置字段：

OutputID: es_abnormal_dbcloud_task
OutputName: es_abnormal_dbcloud_task
OutputType: es
ESHost:
  - http://10.6.24.60:9200
ESIndex: abnormal_dbcloud_task

配置举例(kafka):

OutputID: dc_sqlhunter
OutputName: dc_sqlhunter
OutputType: kafka
KafkaBootstrapServers: 10.6.210.112:9092,10.6.210.113:9092,10.6.210.126:9092
KafkaTopics: mysqlaudit_alert
KafkaOtherConf: ~

6 Elkeid HUB RuleSet/规则集

RuleSet是HUB实现核心检测/响应动作的部分，需要根据具体业务需求来实现，下图是RuleSet在HUB中的简单工作流程：

6.1 RuleSet

HUB RuleSet是通过XML来描述的规则集

RuleSet存在两种类型，rule 和 whitelist，如下：

<root ruleset_id="test1" ruleset_name="test1" type="C">
... ....
</root>

<root ruleset_id="test2" ruleset_name="test2" type="rule" undetected_discard="true">
... ...
</root>

其中RuleSet的是固定的模式，不可随意变更，其他属性：

6.2 Rule

接下来我们来了解rule的具体语法，通常ruleset是由一个或多个rule组成的，需要注意的是多个rule之间的关系是'或'的关系，即如果一条数据可以命中其中的一条或多条rule。

<root ruleset_id="test2" ruleset_name="test2" type="rule" undetected_discard="true">

    <rule rule_id="rule_xxx_1" author="xxx" type="Detection">
        ... ...
    </rule>

    <rule rule_id="rule_xxx_2" author="xxx" type="Frequency">
        ... ...
    </rule>

</root>

一个rule的基本属性有：

我们先假设从Input传入到Ruleset的数据样例为：

{
    "data_type":"59",
    "exe":"/usr/bin/nmap",
    "uid":"0"
}

6.2.1 Detection 简单例子

<rule rule_id="detection_test_1" author="EBwill" type="Detection">
   <rule_name>detection_test_1</rule_name>
   <alert_data>True</alert_data>
   <harm_level>high</harm_level>
   <desc affected_target="test">这是一个Detection的测试</desc>
   <filter part="data_type">59</filter>
   <check_list>
       <check_node type="INCL" part="exe">nmap</check_node>
   </check_list>
</rule>

其中detection_test_1这个规则的意义是：当有数据的data_type为59，且exe中存在nmap的时候，数据继续向后传递。

6.2.2 Frequency 简单例子

<rule rule_id="frequency_test_1" author="EBwill" type="Frequency">
   <rule_name> frequency_test_1 </rule_name>
   <alert_data>True</alert_data>
   <harm_level>high</harm_level>
   <desc affected_target="test">这是一个Frequency的测试</desc>
   <filter part="data_type">59</filter>
   <check_list>
       <check_node type="INCL" part="exe">nmap</check_node>
   </check_list>
   <node_designate>
       <node part="uid"/>
   </node_designate>
   <threshold range="30" local_cache="true">10</threshold>
</rule>

其中frequency_test_1这个规则的意义是：当有数据的data_type为59，且exe中存在nmap的时候，进入到频率检测：当同一个uid，在30秒内出现了 >= 10以上行为则告警，且在这过程中使用当前HUB实例自身缓存。

我们可以看到，实际上Frequency只是比Detection多了node_designate以及threshold字段，也就是说，无论什么规则，都会需有最基础的一些字段，我们接下来就先来了解这些通用的基础字段。

6.2.3 通用字段描述

6.2.4 alert_data

我们依然以上面的Decetion的例子为例：

待检测数据：

{
    "data_type":"59",
    "exe":"/usr/bin/nmap",
    "uid":"0"
}

RuleSet：

<root ruleset_id="test2" ruleset_name="test2" type="rule" undetected_discard="true">
<rule rule_id="detection_test_1" author="EBwill" type="Detection">
   <rule_name>detection_test_1</rule_name>
   <alert_data>True</alert_data>
   <harm_level>high</harm_level>
   <desc affected_target="test">这是一个Detection的测试</desc>
   <filter part="data_type">59</filter>
   <check_list>
       <check_node type="INCL" part="exe">nmap</check_node>
   </check_list>
</rule>
</root>

如果其中alert_data为True，则该RuleSet会向后传递以下数据，会增加SMITH_ALERT_DATA字段，其中包括HIT_DATA用来描述命中规则的详情，以及RULE_INFO即规则本身的基本信息：

{
    "SMITH_ALERT_DATA":{
        "HIT_DATA":[
            "test2 exe:[INCL]: nmap"
        ],
        "RULE_INFO":{
            "AffectedTarget":"all",
            "Author":"EBwill",
            "Desc":"这是一个Detection的测试",
            "DesignateNode":null,
            "FreqCountField":"",
            "FreqCountType":"",
            "FreqHitReset":false,
            "FreqRange":0,
            "HarmLevel":"high",
            "RuleID":"test2",
            "RuleName":"detection_test_1",
            "RuleType":"Detection",
            "Threshold":""
        }
    },
    "data_type":"59",
    "exe":"/usr/bin/nmap",
    "uid":"0"
}

若alert_data为False，则会向后传递以下数据，即原始数据：

{
    "data_type":"59",
    "exe":"/usr/bin/nmap",
    "uid":"0"
}

6.2.5 check_node

check_node的基本结构如下：

<check_node type="检测类型" part="待检测路径">
   检测内容
</check_node>

6.2.5.1 检测类型

目前支持以下几种检测类型：

6.2.5.2 part

假设待检测数据为：

{
    "data":{
        "name":"EBwill",
        "number":100,
        "list":[
            "a1",
            "a2"
        ]
    },
    "class.id":"E-19"
}

对应的part描述方式如下：

data               =       "{\"name\":\"EBwill\",\"number\":100,\"list\":[\"a1\",\"a2\"]
data.name          =       "EBwill"
data.number        =       "100"
data.list.#_0      =       "a1"
data.list.#_1      =       "a2"
class\.id          =       "E-19"

需要注意的是如果待检测key中存在"."需要用""转义

6.2.5.3 高级用法之check_data_type

假设待检测数据为

{
    "stdin":"/dev/pts/1",
    "stdout":"/dev/pts/1"
}

假设我们需要检测 stdin 等于 stdout，即我们的检测内容来源于待检测数据，那么我们需要使用check_data_type="from_ori_data" 来重新定义检测内容的来源是来自于待检测数据而不是填写的内容，如下：

<check_node type="EQU" part="stdin" check_data_type="from_ori_data">stdout</check_node>

6.2.5.4 高级用法之logic_type

假设待检测数据为

{
    "data":"test1 test2 test3",
    "size": 96,
}

当我们需要检测data中是否存在 test1 或 test2 的时候，我们可以写正则来实现，也可以通过定义logic_type来实现check_node支持"AND"或者"OR"逻辑，如下：

<!-- data中存在test1 或 test2 -->
<check_node type="INCL" part="data" logic_type="or" separator="|">
    <![CDATA[test1|test2]]>
</check_node>

<!-- data中存在test1 和 test2 -->
<check_node type="INCL" part="data" logic_type="and" separator="|">
    <![CDATA[test1|test2]]>
</check_node>

其中logic_type用来描述逻辑类型，支持"and"和"or"，separator用于自定义标识切割检测数据的方式

6.2.5.5 高级用法之foreach

当我们需要对数组有较复杂的检测时可能可以通过foreach来解决。

假设待检测数据为

{
    "data_type":"12",
    "data":[
        {
            "name":"a",
            "id":"14"
        },
        {
            "name":"b",
            "id":"98"
        },
        {
            "name":"c",
            "id":"176"
        },
        {
            "name":"d",
            "id":"172"
        }
    ]
}

我们想将id > 100且顶层的data_type等于12的数据的obj筛选出来，那么可以先通过foreach进行遍历，然后再对遍历后的数据进行判断，如下：

<check_list foreach="d in data">
    <check_node type="MT" part="d.id">100</check_node>
    <check_node type="EQU" part="data_type">12</check_node>
</check_list>

则会向后传递多条数据：

{
    "data_type":"12",
    "data":[
        {
            "name":"c",
            "id":"176"
        }
    ]
}

以及

{
    "data_type":"12",
    "data":[
        {
            "name":"d",
            "id":"172"
        }
    ]
}

我们通过下图来更好的理解foreach这个高级用法

假设待检测数据为

{
    "data_type":"12",
    "data":[
        1,
        2,
        3,
        4,
        5,
        6,
        7
    ]
}

我们想筛选出data中小于5的数据，那么需要这样编写：

<check_list foreach="d in data">
    <check_node type="LT" part="d">5</check_node>
</check_list>

6.2.5.6 高级用法之cep

默认情况下，checknode之间的关系and，满足所有checknode的检测条件时，数据才会检出，当需要自定义checknode之间的关系时，可以使用cep来解决。

假设待检测数据为

{
    "data_type":"12",
    "comm":"ETNET",
    "node_name":"p23",
    "exe":"bash",
    "argv":"bash"
}

我们希望将node_name等于p23或comm等于ETNET，且exe 和 argv 等于bash ，这样的数据过滤出来，如下：

<check_list cep="(a or b) and c">
    <check_node part="comm" group="a" type="INCL" check_data_type="static">ETNET</check_node>
    <check_node part="nodename" group="b" type="INCL" check_data_type="static">p23</check_node>
    <check_node part="exe" group="c" type="INCL" check_data_type="static">bash</check_node>
    <check_node part="argv" group="c" type="INCL" check_data_type="static">bash</check_node>
</check_list>

可以为check_node声明为group ，再在cep中编写对group的条件。支持or 和 and。

6.2.6 Frequency 字段

Frequency的逻辑在check_list之后，但数据通过了filter和check_list之后，如果当前rule的类型为Frequency则会进入到Frequency的特殊检测逻辑。Frequency存在两个字段，node_designate与threshold，如下：

<node_designate>
    <node part="uid"/>
    <node part="pid"/>
</node_designate>
<threshold range="30">5</threshold>

6.2.6.1 node_designate

其中node_designate是代表group_by，上方样例的含义是对 uid 和 pid 这两个字段进行group_by。

6.2.6.2 threshold

**threshold是描述频率检测的具体检测内容：多长时间内(range)出现多少次(threshold)。**如上样例中即表达：同一uid与pid在30秒内出现5次即为检出，其中range的单位是秒。

如上图，由于仅仅在10s内就出现了5次，那么在剩下的20s内出现的全部pid=10且uid=1的数据都会告警，如下：

但是这个问题可能会导致告警数据过多，因此支持一个叫做：hit_reset的参数，使用方式如下：

<node_designate>
    <node part="uid"/>
    <node part="pid"/>
</node_designate>
<threshold range="30" hit_reset="true" >5</threshold>

当hit_reset为true时，每次满足threshold策略后，时间窗口将会重制，如下：

在频率检测的场景中，还有一种问题是性能问题，由于这种有状态的检测需要保存一些中间状态，这部分中间状态的数据我们是存在redis中，但是如果数据量过大，对redis会有一定的影响，因此我们频率检测也支持使用使用HUB自己的Local Cache来存储这些中间状态的数据，但是也会失去全局性，开启的方式是设置local_cache参数为true：

<node_designate>
    <node part="uid"/>
    <node part="pid"/>
</node_designate>
<threshold range="30" local_cache="true" hit_reset="true" >5</threshold>

失去全局性的原因是该Cache只服务于所属的HUB实例，如果是集群模式下，Local Cache并不互相共享，但是会带来一定的性能提升。

6.2.6.3 高级用法之count_type

有些情况下我们计算频率不是想计算“出现了多少次”，而会有一些其他的需求，如出现了多少类，**出现的字段内数据和是多少。**我们先来看第一个需求，出现了多少类。

假设待检测数据为

{
    "sip":"10.1.1.1",
    "sport":"6637",
    "dip":"10.2.2.2",
    "dport":"22"
}

当我们想写一个检测扫描器的规则时，我们其实往往不关心某一个IP访问了多少次其他资产，而是访问了多少不同的其他资产，当这个数据较大时，可能存在网络扫描探测的可能性，假设我们规定，3600秒内，同一IP访问的不同IP数超过100种就记录下来，那么他的频率部分规则应该这样编写：

<node_designate>
    <node part="sip"/>
</node_designate>
<threshold range="3600" count_type="classify" count_field="dip">100</threshold>

这时候count_type需要为classify，count_field则为类型计算依赖的字段，即dip。

第二个场景假设待检测数据为

{
    "sip":"10.1.1.1",
    "qps":1
}

假设我们需要筛选出3600s内qps总和大于1000的数据，那么我们可以这样编写：

<node_designate>
    <node part="sip"/>
</node_designate>
<threshold range="3600" count_type="sum" count_field="qps">1000</threshold>

count_type为空时默认计算次数，当为classify时计算的是类型，为sum时计算的是求和。

6.2.7 append

当我们想对数据进行一些增加信息的操作时，可以使用append来进行添加数据的操作，append的语法如下：

<append type="append类型" rely_part="依赖字段" append_field_name="增加字段名称">增加内容</append>