package main
import (
"fmt"
"os"
"syscall"
"go.mongodb.org/mongo-driver/bson"
)
const (
dbName string = "test"
pidColName string = "pids"
fdColName string = "fds"
fileColName string = "files"
)
// var mongoMutex sync.Mutex
var pidCol, fdCol, fileCol mongoClient
var docRes []bson.M
var err error
func deal() {
defer wg.Done()
var cooked Event
var ok bool
if err = pidCol.init(dbName, pidColName); err != nil {
fmt.Fprintf(os.Stderr, "Error while initing the mongodb: %v\n", err)
return
}
err = pidCol.InsertOne(bson.M{
"ppid": 1,
"pid": containerdPid,
"cwd": "/",
"children": []bson.M{},
"daemon": true,
})
if err != nil {
fmt.Fprintf(os.Stderr, "Error while initing the mongodb: %v\n", err)
return
}
if err = fdCol.init(dbName, fdColName); err != nil {
fmt.Fprintf(os.Stderr, "Error while initing the mongodb: %v\n", err)
return
}
if err = fileCol.init(dbName, fileColName); err != nil {
fmt.Fprintf(os.Stderr, "Error while initing the mongodb: %v\n", err)
}
fmt.Printf("Containerd: %d\n", containerdPid)
defer pidCol.Disconnect()
defer fdCol.Disconnect()
defer fileCol.Disconnect()
for {
cooked, ok = <-cookedChan
if !ok {
break
}
switch cooked.tag {
case NEWPID:
go dealNewPid(cooked)
case EXECVE:
go dealExecve(cooked)
case PIDEXIT:
go deletePid(cooked)
case FILEOPEN:
go fileOpen(cooked)
case FILEWRITE:
go fileWrite(cooked)
case FILECLOSE:
go fileClose(cooked)
case PIVOTROOT:
go pivotRoot(cooked)
}
}
}
func deletePid(cooked Event) {
// 先从老爹那里销户
pidCol.UpdateOne(bson.M{"pid": cooked.ppid}, bson.M{
"$pull": bson.M{
"children": cooked.pid,
},
})
// 在这套逻辑里，孩子是不需要收容的，因为我们根本就不看ppid来工作
// 可以去死了
pidCol.UpdateOne(bson.M{"pid": cooked.pid}, bson.M{
"$set": bson.M{
"exit_timestamp": cooked.timestamp,
"exit_code": cooked.exit_code,
"exit_signal": cooked.exit_signal,
},
})
// 理论上这里需要关闭所有文件描述符，但为了处理效率，留给后续流程
}
func dealNewPid(cooked Event) {
// 自身是否已经记录
docRes, err = pidCol.Finddoc(bson.M{"pid": cooked.pid})
if err != nil {
fmt.Fprintf(os.Stderr, "Err finding: %v\n", err)
return
}
if len(docRes) != 0 {
// 进程原本就存在，换言之别的消息先到了
// 所有先行抵达的消息必须保留execve/children字段
// 此处不再更新
// 以防把原有信息更没了
pidCol.UpdateOne(bson.M{"pid": cooked.pid}, bson.M{
"start_timestamp": cooked.timestamp,
"ppid": cooked.ppid,
"parentTgid": cooked.parentTgid,
"pid": cooked.pid,
"tgid": cooked.tgid,
"cwd": cooked.cwd,
// "execve": []bson.M{},
"args": cooked.argv,
// "children": []bson.M{},
})
} else {
// 这进程本是新修的
pidCol.InsertOne(bson.M{
"start_timestamp": cooked.timestamp,
"ppid": cooked.ppid,
"parentTgid": cooked.parentTgid,
"pid": cooked.pid,
"tgid": cooked.tgid,
"cwd": cooked.cwd,
"execve": []bson.M{},
"args": cooked.argv,
"children": []bson.M{},
})
}
pidCol.UpdateOne(bson.M{"pid": cooked.ppid}, bson.M{
"$push": bson.M{
"children": cooked.pid,
},
})
}
func dealExecve(cooked Event) {
// 父进程在不在？不在扔
docRes, err = pidCol.Finddoc(bson.M{"pid": cooked.ppid})
if err != nil || len(docRes) != 1 {
return
}
// 首先检查进程是否存在，如不存在则为之创建
docRes, err = pidCol.Finddoc(bson.M{"pid": cooked.pid})
if err != nil {
return
}
if len(docRes) == 1 {
// 自身已在，直接记录
pidCol.UpdateOne(bson.M{"pid": cooked.pid}, bson.M{
"$push": bson.M{
"execve": bson.M{
"timestamp": cooked.timestamp,
"execArgs": cooked.argv,
},
},
})
} else {
// 先fork抵达，插入
pidCol.InsertOne(bson.M{
"ppid": cooked.ppid,
"pid": cooked.pid,
"children": []bson.M{},
"execve": []bson.M{
{
"timestamp": cooked.timestamp,
"execArgs": cooked.argv,
},
},
})
}
}
func fileOpen(cooked Event) {
// 权限检查过了，不必再查
fdCol.InsertOne(bson.M{
"timestamp": cooked.timestamp,
"fileName": cooked.srcPath,
"pid": cooked.pid,
"fd": cooked.exit_code,
"flags": cooked.syscallParam,
"written": []bson.M{},
})
if cooked.syscallParam[1]&syscall.O_TRUNC != 0 {
fdCol.UpdateOne(bson.M{"pid": cooked.pid, "fd": cooked.exit_code}, bson.M{
"$push": bson.M{
"written": cooked.timestamp,
},
})
}
}
func fileClose(cooked Event) {
res, err := fdCol.FindOneAndDelete(bson.M{"pid": cooked.pid, "fd": cooked.syscallParam[0]})
if err != nil {
return
}
res["close_timestamp"] = cooked.timestamp
if err := fileCol.InsertOne(res); err != nil {
fmt.Fprintf(os.Stderr, "Err inserting files: %v\n", err)
}
}
func fileWrite(cooked Event) {
res, err := fdCol.Finddoc(bson.M{
"pid": cooked.pid,
"fd": cooked.syscallParam[0],
"close_timestamp": bson.M{"$exists": false},
})
if err != nil {
fmt.Fprintf(os.Stderr, "Err closing fd %d of pid %d: %v\n", cooked.syscallParam[0], cooked.pid, err)
}
if len(res) == 0 {
return
}
fdCol.UpdateOne(bson.M{
"pid": cooked.pid,
"fd": cooked.syscallParam[0],
"close_timestamp": bson.M{"$exists": false},
}, bson.M{"$push": bson.M{"written": cooked.timestamp}})
}
func pivotRoot(cooked Event) {
// docker的根目录信息，记录
docRes, err := pidCol.Finddoc(bson.M{"pid": cooked.pid})
if err != nil {
fmt.Fprintf(os.Stderr, "Err finding: %v\n", err)
return
}
if len(docRes) == 0 {
// fork还没到，等一下
pidCol.InsertOne(bson.M{
"start_timestamp": cooked.timestamp,
"ppid": cooked.ppid,
"pid": cooked.pid,
"rootfs": "cwd",
})
} else {
// 读取已有的工作目录
cwd := docRes[0]["cwd"]
pidCol.UpdateOne(bson.M{"pid": cooked.pid}, bson.M{
"$set": bson.M{
"rootfs": cwd,
},
})
}
}
|
