读Elkeid源码(一)

如何进行用户态数据采集

2025-10-02
安全开发

文章大纲

这是读源码这个系列的开始,我想通过模仿业界成熟的安全工具,并自己模仿着造一个丐版的工具出来,从而快速学习
Elkeid是主机，容器与容器集群，Serverless 等多种工作负载安全需求的开源解决方案
Elkeid 具备以下主要能力：

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

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

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

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

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

目前第一阶段我想先模仿elkeid用户态的HIDS主机入侵检测和恶意文件识别能力

尝试

我先自己用go搭建了一个基础的hids框架
只实现了最基础的定时系统数据采集(网络,连接,内存cpu,进程),agent数据透传到server,server在用一个web看板给运营人员查看

看看Elkeid怎么做

那么就从数据采集入手,先看看Elkeid是怎么实现的
首先Elkeid的数据采集有三层 ：Collector（用户态）+ Driver（内核态）+ RASP（应用态）
同样是用户态collector,我们先看看Elkeid都采集那些信息(有完整,有用的信息的基础上,再谈识别,分析攻击的规则)
先看看Elkeid/plugins/collector下都有哪些文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
➜  collector :(main) tree                                                
.
├── app.go //应用程序识别
├── build.sh
├── container //容器环境监控
│   ├── container.go 
│   └── enum.go
├── container.go//容器环境监控
├── cron.go
├── engine
│   └── engine.go
├── go.mod
├── go.sum
├── integrity.go //系统完整性监控
├── kmod.go
├── main.go
├── net_interface.go //网络监控
├── port //网络监控
│   ├── netlink.go
│   ├── port.go
│   └── procfs.go
├── port.go
├── process //进程信息采集 
│   └── process.go 
├── process.go
├── README-zh_CN.md
├── README.md
├── rpm
│   └── rpm.go
├── service.go //系统服务监控
├── software.go
├── user.go //用户账户监控
├── utils
│   ├── file.go
│   └── user.go
├── volume.go
├── weak_password
└── zip
    └── zip.go

我将直接用于数据采集的文件或目录打上了对应的注释
大致可以分为7类

2.1 进程信息采集

数据源

1
2
3
4
/proc/[pid]/stat - 进程统计信息
/proc/[pid]/status - 进程状态详情
/proc/[pid]/cmdline - 命令行参数
/proc/[pid]/cwd - 当前工作目录

我自己的实现里读的是/proc/[pid]/comm和/proc/[pid]/cmdline
这里我去查了一下,也拿我服务器的一个进程来看了看

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
​status: ​进程状态信息的汇总。这是最重要的文件之一，包含：
	Name: 同 comm
	State: 运行状态 (R, S, D, T, Z, X 等)
	Pid, PPid, Tgid: 进程ID、父进程ID、线程组ID (主线程PID)
	Uid, Gid: 真实、有效、保存的设置用户ID和组ID
	FDSize: 已分配的文件描述符数量
	Groups: 补充组列表
		VmPeak, VmSize, VmLck, VmPin, VmHWM, VmRSS, RssAnon, RssFile, RssShmem, VmData, VmStk, VmExe, VmLib, VmPTE, VmSwap: ​内存使用情况的详细统计​ (峰值虚拟内存大小、当前虚拟内存大小、锁定内存、固定内存、峰值物理内存驻留集大小、当前物理内存驻留集大小、匿名页、文件页、共享内存页、数据段、栈段、代码段、共享库、页表项、交换空间用量等)
	Threads: 该线程组中的线程数
	SigQ: 待处理信号队列信息
	CapInh, CapPrm, CapEff, CapBnd, CapAmb: 继承的、允许的、有效的、边界和可继承的能力集。
	Seccomp: Seccomp 模式
	Cpus_allowed, Cpus_allowed_list, Mems_allowed, Mems_allowed_list: 允许的 CPU 和内存节点掩码/列表
	voluntary_ctxt_switches, nonvoluntary_ctxt_switches: 自愿和非自愿上下文切换次数
​stat: ​进程状态的原始数据。包含大量信息（PID、状态、CPU时间、优先级、nice值、线程数、启动时间、虚拟内存大小、驻留集大小等），格式紧凑但不易读。ps等工具解析此文件

总的来说,从进程的status里的信息多很多,但是注意,并不是status这么多信息都会用到,仔细看了看,elkeid对进程信息的采集只记录了这几个字段

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
rec.Data.Fields["cmdline"] = cmdline
rec.Data.Fields["cwd"], _ = p.Cwd()
rec.Data.Fields["checksum"], _ = p.ExeChecksum()
rec.Data.Fields["exe_hash"], _ = p.ExeHash()
rec.Data.Fields["exe"], _ = p.Exe()
rec.Data.Fields["pid"] = p.Pid()
mapstructure.Decode(stat, &rec.Data.Fields)
mapstructure.Decode(status, &rec.Data.Fields)
mapstructure.Decode(ns, &rec.Data.Fields)
m, _ := cache.Get(5056, ns.Pid)
rec.Data.Fields["container_id"] = m["container_id"]
rec.Data.Fields["container_name"] = m["container_name"]
rec.Data.Fields["integrity"] = "true"
// only for host files
if _, ok := cache.Get(5057, rec.Data.Fields["exe"]); ok && rec.Data.Fields["container_id"] == "" {
	rec.Data.Fields["integrity"] = "false"
}
rec.Data.Fields["package_seq"] = seq
c.SendRecord(rec)

所以其实elkeid对进程的采集应该是为了做了一些可执行文件/容器的校验,篡改检测

2.2 网络信息采集

Elkeid/plugins/collector/port.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
type Port struct {
	// from inet
	Family   string `mapstructure:"family"`
	Protocol string `mapstructure:"protocol"`
	State    string `mapstructure:"state"`
	Sport    string `mapstructure:"sport"`
	Dport    string `mapstructure:"dport"`
	Sip      string `mapstructure:"sip"`
	Dip      string `mapstructure:"dip"`
	Uid      string `mapstructure:"uid"`
	Inode    string `mapstructure:"inode"`
	Username string `mapstructure:"username"`
	// from process
	Pid     string `mapstructure:"pid"`
	Exe     string `mapstructure:"exe"`
	Comm    string `mapstructure:"comm"`
	Cmdline string `mapstructure:"cmdline"`
	Psm     string `mapstructure:"psm"`
	PodName string `mapstructure:"pod_name"`
}

这个文件是用户态网络信息的核心实现逻辑,主要从两个数据源获取上述信息
首先从linux中都去协议和ip信息,这里可以得到上半部分from inet的数据

1
2
3
4
var (
	scanProto  = [2]int{unix.IPPROTO_UDP, unix.IPPROTO_TCP}
	scanFamily = [2]int{unix.AF_INET, unix.AF_INET6}
)

然后创建一个port map

1
pm := map[string]*Port{}

当系统支持 netlink socket时,会调用Elkeid/plugins/collector/netlink.go里的inetDiag()方法
利用Linux netlink socket直接与内核通信,避免了文件系统的开销

1
func inetDiag(family, proto uint8) (ret []*inetDiagResp, err error)

这个函数通过 Linux 的 NETLINK_INET_DIAG 机制获取指定协议族和协议类型的网络连接信息

1
2
3
4
5
6
var s *nl.NetlinkSocket
s, err = nl.Subscribe(unix.NETLINK_INET_DIAG)
if err != nil {
    return
}
defer s.Close()

NETLINK_INET_DIAG : 这是 Linux 内核提供的专门用于网络诊断的 netlink 协议族,允许用户空间程序直接与内核的网络子系统通信

  • 优势 : 比解析 /proc/net/* 文件更高效，数据更准确
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
req := nl.NewNetlinkRequest(nl.SOCK_DIAG_BY_FAMILY, unix.NLM_F_DUMP)
	var state uint32
	if proto == unix.IPPROTO_UDP {
		state = 7
	} else if proto == unix.IPPROTO_TCP {
		state = 10
	} else {
		err = fmt.Errorf("unsupported protocol %d", proto)
		return
	}
	req.AddData(&inetDiagReq{
		family:   family,
		protocol: proto,
		ext:      (1 << (netlink.INET_DIAG_VEGASINFO - 1)) | (1 << (netlink.INET_DIAG_INFO - 1)),
		states:   uint32(1 << state),
	})
	err = s.Send(req)
	if err != nil {
		return
	}
  • SOCK_DIAG_BY_FAMILY : 按协议族查询 socket 信息的请求类型
  • NLM_F_DUMP : 标志位，表示要获取所有匹配的记录

然后过滤出UDP 和 TCP的“连接”

  • TCP state 10 : TCP_LISTEN - TCP 监听状态
  • UDP state 7 : UDP_LISTEN - UDP 监听状态（虽然 UDP 是无连接的，但这里表示绑定到端口的 socket）
    然后AddData()构建请求数据
  • family : 地址族 (IPv4/IPv6)
  • protocol : 协议类型 (TCP/UDP)
  • ext : 扩展信息标志，请求额外的诊断信息
  • states : 状态位掩码，只获取指定状态的连接

procfs

当netlilnk socket不可用时,会从文件系统汇总采集网络信息,走入

1
ps, err = procNet(uint8(family), uint8(proto))

走入Elkeid/plugins/collector/procfs.go文件的procNet()方法
从/proc/net中去获取连接信息
没有太特别的东西,省略

进程关联

当从inetDiag()或procNet()拿到网络连接信息后,需要和进程信息就行一个关联
回到port.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
procs, err := process.Processes(false)
for _, p := range procs {
    fds, err := p.Fds()  // 获取进程的文件描述符
    for _, fd := range fds {
        // 查找 socket 类型的文件描述符
        if strings.HasPrefix(fd, "socket:[") && strings.HasSuffix(fd, "]") {
            // 提取 inode 号码
            inode := fd[8:len(fd)-1]  // 去掉 "socket:[" 和 "]"
            
            // 在 pm 中查找对应的端口信息
            if port, ok := pm[inode]; ok {
                // 关联进程信息
                port.Pid = p.Pid()
                port.Exe, _ = p.Exe()
                port.Cmdline, _ = p.Cmdline()
                port.Comm, _ = p.Comm()
                
                // 提取容器环境变量
                if envs, err := p.Envs(); err == nil {
                    // Pod 名称
                    if p, ok := envs["POD_NAME"]; ok {
                        port.PodName = p
                    }
                    // PSM 服务标识
                    if p, ok := envs["LOAD_SERVICE_PSM"]; ok {
                        port.Psm = p
                    }
                }
            }
        }
    }
}

至此,每一个网络连接我们需要的信息就获取完了

2.3 容器环境监控

Elkeid/plugins/collector/container.go
数据结构

1
2
3
4
5
6
7
8
9
10
11
type Container struct {
    ID         string  // 容器 ID
    Name       string  // 容器名称
    ImageID    string  // 镜像 ID (去除 sha256: 前缀)
    ImageName  string  // 镜像名称
    State      string  // 容器状态 (created/running/exited/unknown)
    Pid        string  // 容器主进程 PID
    Pns        string  // PID 命名空间 ID
    Runtime    string  // 运行时类型 (cri/docker)
    CreateTime string  // 创建时间戳
}

采集信息：

  • 容器基础信息：ID、名称、镜像、状态
  • 运行时信息：PID、命名空间、运行时类型
  • 时间信息：创建时间

核心接口设计

1
2
3
4
5
6
type Client interface {
    ListContainers(ctx context.Context) ([]Container, error)
    Exec(ctx context.Context, containerID string, name string, arg ...string) ([]byte, error)
    Close()
    Runtime() string
}

这个接口定义了容器客户端的标准操作：

  • ListContainers : 获取容器列表
  • Exec : 在容器内执行命令
  • Close : 关闭连接
  • Runtime : 返回运行时类型

然后就分cri和docker两种运行时实现这个Client接口了,这里我们只看docker的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
func (c *dockerClient) ListContainers(ctx context.Context) ([]Container, error) {
	containers := []Container{}
	resp, err := c.c.ContainerList(ctx, types.ContainerListOptions{All: true})
	if err != nil {
		return nil, err
	}
	for _, dockerContainer := range resp {
		container := Container{
			ID:         dockerContainer.ID,
			ImageID:    strings.TrimPrefix(dockerContainer.ImageID, "sha256:"),
			ImageName:  strings.TrimPrefix(dockerContainer.Image, "sha256:"),
			State:      dockerContainer.State,
			CreateTime: strconv.FormatInt(dockerContainer.Created, 10),
			Runtime:    c.Runtime(),
		}
		if resp, err := c.c.ContainerInspect(ctx, dockerContainer.ID); err == nil {
			container.Name = strings.TrimPrefix(resp.Name, "/")
			if container.State == StateName[int32(RUNNING)] {
				container.Pid = strconv.Itoa(resp.State.Pid)
				if resp.State.Pid > 0 {
					if p, err := process.NewProcess(container.Pid); err == nil {
						container.Pns, _ = p.Namespace("pid")
					}
				}
			}
		}
		if container.Name == "" && len(dockerContainer.Names) > 0 {
			container.Name = dockerContainer.Names[0]
		}
		containers = append(containers, container)
	}
	return containers, nil
}

连接建立用的是Docker官方的SDK,然后根据环境变量自动连接
用ContainerList() 来获取容器列表,然后遍历每个容器,使用ContainerInspect()来获取详细信息,然后resp.State.Pid直接拿到PID,然后运行时信息等等
docker官方sdk的用法网上也有很多介绍的,需要时直接查就好了,cri也是同理
go docker sdk获取容器列表

2.4 系统服务监控

Elkeid/plugins/collector/service.go
数据结构

1
2
3
4
5
6
7
8
9
type Service struct {
	Name       string `mapstructure:"name"` 
	Type       string `mapstructure:"type"` //服务类型
	Command    string `mapstructure:"command"` //启动命令
	Restart    string `mapstructure:"restart"`
	WorkingDir string `mapstructure:"working_dir"` //工作目录
	Checksum   string `mapstructure:"checksum"` //文件校验和（MD5）
	BusName    string `mapstructure:"bus_name"`
}

扫描目录

1
2
3
4
var SearchDir = []string{
	"/etc/systemd/system.control", "/run/systemd/system.control", "/run/systemd/transient",
	"/run/systemd/generator.early", "/etc/systemd/system", "/run/systemd/system",
	"/run/systemd/generator", "/usr/local/lib/systemd/system", "/usr/lib/systemd/system", "/run/systemd/generator.late"}

2.5 rpm包管理

Elkeid/plugins/collector/rpm.go
直接读取 /var/lib/rpm/Packages 数据库文件

  • 使用Berkeley DB的Hash数据库格式解析
  • 支持多种哈希算法：MD5、SHA1、SHA256、SHA512等
  • 解析包的元数据：名称、版本、架构、供应商、文件列表等

2.6 用户账户管理

Elkeid/plugins/collector/user.go

数据源 ：

  • /etc/passwd ：基本用户信息
  • /etc/shadow ：密码哈希信息
  • /var/log/wtmp ：登录历史记录
  • sudo -l ：sudo权限查询
    这里还实现了一个弱密码检测的功能,内置了弱密码字典,支持SHA256/SHA512加密验证

2.7 内核模块监控模块

Elkeid/plugins/collector/kmod.go
数据源: /proc/modules
监控字段 ：

  • name：模块名称
  • size：模块大小
  • refcount：引用计数
  • used_by：被哪些模块使用
  • state：模块状态
  • addr：内存地址

2.8 文件完整性检查模块

Elkeid/plugins/collector/integrity.go

核心功能 ：验证系统关键文件的完整性
支持的包管理器 ：

  • DPKG ：读取 .md5sums 文件进行MD5校验
  • RPM ：使用包内置的文件摘要进行校验
    策略 ：
  • 只检查二进制文件（/bin, /sbin, /usr/bin, /usr/sbin等）
  • 文件大小限制（100MB）
  • 支持多种哈希算法（MD5、SHA1、SHA256、SHA512）
  • 使用限流器控制IO速度
    检测逻辑是计算文件当前摘要值然后与包管理器记录的原始摘要对比,发现不一致时生成告警记录

后面几个比较省略,因为相对就采集数据来说,实现起来我感觉难度不高
还有几个模块对于hids来说优先级没有那么高,这里暂时不分析(volume等等)