celengine

package
v0.0.0-...-23ca95c Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 3, 2026 License: AGPL-3.0 Imports: 22 Imported by: 0

Documentation

Overview

Package celengine 实现基于 CEL-Go 的实时检测规则引擎 支持从数据库加载 CEL 表达式规则，对 Kafka 消费事件进行实时评估并生成告警

Package celengine - 命令转发器 Consumer 侧通过 Redis 查找 Agent 所在 AC 实例，HTTP 转发命令

Package celengine - C4: 自动响应 规则匹配后通过 AgentCenter 下发 kill/隔离 Task

Package celengine 的端口扫描检测器 使用 Redis 滑动窗口统计同一源 IP 在时间窗口内访问的不同端口数 超过阈值时生成告警

Package celengine - 行为序列检测 使用滑动窗口 + Redis 状态机检测多步攻击链

Index

Constants

View Source 
const AttackChainCategory = "attack_chain"

AttackChainCategory 攻击链(序列)命中的告警分类，供 incident 关联识别为强信号

View Source 
const IOCHitCategory = "ioc_hit"

IOCHitCategory 威胁情报命中告警分类

Variables

This section is empty.

Functions

func CountEnabledRules 

func CountEnabledRules(db *gorm.DB) (int64, error)

CountEnabledRules 统计启用的规则数量

func IsAlertWhitelisted 

func IsAlertWhitelisted(rule *model.DetectionRule, fields map[string]string) (bool, string)

IsAlertWhitelisted 判断给定告警是否命中白名单。

返回 (whitelisted, reason)。reason 用于审计日志。

规则匹配语义：

  • RuleNamePattern 必须命中（子串匹配规则 name 或 result_id）。
  • RuleCategoryPattern（如填）必须命中规则 category。
  • ExeBasenameIn + DstIPInPrivate 是 OR 关系：任一命中即抑制。
  • 同一白名单条只有 ExeBasenameIn 或只有 DstIPInPrivate 时即只需该项命中。

func LoadRuleByID 

func LoadRuleByID(db *gorm.DB, id uint) (*model.DetectionRule, error)

LoadRuleByID 根据 ID 加载单条规则（用于调试或单规则热更新）

func LoadRulesFromDB 

func LoadRulesFromDB(db *gorm.DB) ([]model.DetectionRule, error)

LoadRulesFromDB 从数据库加载所有启用的检测规则 只加载 enabled=true 的规则，按 ID 排序保证确定性

Types

type AlertGenerator 

type AlertGenerator struct {
	// contains filtered or unexported fields
}

AlertGenerator 负责将 CEL 引擎匹配结果写入 alerts 表（去重模式）

func NewAlertGenerator 

func NewAlertGenerator(db *gorm.DB, logger *zap.Logger) *AlertGenerator

NewAlertGenerator 创建 AlertGenerator

func (*AlertGenerator) Generate 

func (g *AlertGenerator) Generate(hostID string, matchedRules []model.DetectionRule, fields map[string]string)

Generate 根据匹配的规则和事件字段生成或更新告警 去重策略：同一规则 + 同一主机合并为一条告警，累加 HitCount

告警入库前调用 IsAlertWhitelisted 过滤已知误报模式（反代上游 / 内网通信等）， 避免 nginx → backend:8888 这种业务流量被 C2 规则刷屏。

func (*AlertGenerator) GenerateFromIOC 

func (g *AlertGenerator) GenerateFromIOC(hostID, iocType, iocValue string, fields map[string]string)

GenerateFromIOC 服务端 IOC 匹配命中 → 生成/更新 ioc_hit 告警。 命中字段写入 ioc_match/ioc_type/ioc_value,前端研判可溯源命中来源。尊重白名单。

func (*AlertGenerator) GenerateFromSequence 

func (g *AlertGenerator) GenerateFromSequence(hostID string, rule model.SequenceRule, fields map[string]string)

GenerateFromSequence 为攻击链(序列)命中生成/更新告警并落 alerts 表。 攻击链是多步关联的高置信检测，以 category=attack_chain 标记，供 incident 关联识别为强 IOA 信号。 仍尊重用户研判/调优学到的 DB 白名单(同规则+主机被判误报)→ 命中即抑制，使研判学习对攻击链同样闭环。

func (*AlertGenerator) SetSIEMForwarder 

func (g *AlertGenerator) SetSIEMForwarder(f *siem.Forwarder)

SetSIEMForwarder 设置 SIEM 转发器

func (*AlertGenerator) StartHostGraceReload 

func (g *AlertGenerator) StartHostGraceReload(ctx context.Context)

StartHostGraceReload 启动后台 goroutine 周期 reload 主机 created_at 快照（ctx 取消时退出）。 调用方需保证只调用一次。

func (*AlertGenerator) StartWhitelistReload 

func (g *AlertGenerator) StartWhitelistReload(ctx context.Context)

StartWhitelistReload 启动后台 goroutine 周期 reload DB 白名单（ctx 取消时退出）。 调用方需保证只调用一次。

type AutoResponder 

type AutoResponder struct {
	// contains filtered or unexported fields
}

AutoResponder 自动响应执行器

func NewAutoResponder 

func NewAutoResponder(db *gorm.DB, logger *zap.Logger) *AutoResponder

NewAutoResponder 创建自动响应执行器

func (*AutoResponder) Execute 

func (r *AutoResponder) Execute(hostID string, matchedRules []model.DetectionRule, fields map[string]string)

Execute 执行自动响应

func (*AutoResponder) SetDispatcher 

func (r *AutoResponder) SetDispatcher(dispatcher interface {
	SendCommand(agentID string, cmd interface{}) error
})

SetDispatcher 设置命令下发器

type CommandForwarder 

type CommandForwarder struct {
	// contains filtered or unexported fields
}

CommandForwarder 实现 AutoResponder 的 dispatcher 接口 通过 Redis 查找 AC 实例，HTTP 转发命令

func NewCommandForwarder 

func NewCommandForwarder(redisClient *redis.Client, logger *zap.Logger) *CommandForwarder

NewCommandForwarder 创建命令转发器

func (*CommandForwarder) SendCommand 

func (f *CommandForwarder) SendCommand(agentID string, cmd interface{}) error

SendCommand 实现 dispatcher 接口 cmd 为 map[string]interface{}，包含 data_type, data, object_name

type Engine 

type Engine struct {
	// contains filtered or unexported fields
}

Engine CEL 规则引擎 负责维护 CEL 环境、编译规则、评估事件

func New 

func New(db *gorm.DB, logger *zap.Logger) (*Engine, error)

New 创建 CEL 规则引擎实例 初始化 CEL 环境并从数据库加载规则

func NewInMemory 

func NewInMemory(logger *zap.Logger, rules []model.DetectionRule) (*Engine, error)

NewInMemory 创建内存 CEL 引擎（用于测试，不依赖数据库）

func (*Engine) CompileExpression 

func (e *Engine) CompileExpression(expression string) (cel.Program, error)

CompileExpression 编译单个 CEL 表达式为 Program（供序列检测等外部模块使用）

func (*Engine) Evaluate 

func (e *Engine) Evaluate(dataType int32, fields map[string]string) []model.DetectionRule

Evaluate 对事件进行规则评估 dataType: 事件类型（用于过滤适用规则） fields: 事件字段键值对 返回匹配的规则列表

当适用规则数 >= parallelThreshold 且 workers > 1 时，自动切换为并行求值。

func (*Engine) ProcessTreeStats 

func (e *Engine) ProcessTreeStats() (hosts, nodes int)

ProcessTreeStats 返回进程树统计信息

func (*Engine) ReloadRules 

func (e *Engine) ReloadRules() error

ReloadRules 从数据库重新加载所有启用的规则并编译 支持热更新：在规则变更后调用此方法即可生效

func (*Engine) RuleCount 

func (e *Engine) RuleCount() int

RuleCount 返回当前加载的规则数量

func (*Engine) SetWorkers 

func (e *Engine) SetWorkers(n int)

SetWorkers 设置并行求值 worker 数量（1 = 禁用并行）

func (*Engine) StartCleanup 

func (e *Engine) StartCleanup(ctx context.Context)

StartCleanup 启动进程树和事件追踪器的定期清理协程

func (*Engine) TrackerStats 

func (e *Engine) TrackerStats() (hosts, windowKeys, knownKeys int)

TrackerStats 返回事件追踪器统计信息

type EventTracker 

type EventTracker struct {
	// contains filtered or unexported fields
}

EventTracker 维护事件频率统计和首次出现追踪 用于 CEL 规则中的 recent_*_count 和 first_seen_* 变量

func NewEventTracker 

func NewEventTracker(logger *zap.Logger) *EventTracker

NewEventTracker 创建事件追踪器

func (*EventTracker) Cleanup 

func (t *EventTracker) Cleanup() int

Cleanup 清理过期滑动窗口条目，释放内存

func (*EventTracker) Observe 

func (t *EventTracker) Observe(hostID, field, value string) (count int64, firstSeen bool)

Observe 记录一次事件观测，返回窗口内计数和是否首次出现 调用一次同时完成：检查 first_seen → 标记已知 → 记录时间戳 → 返回窗口计数

func (*EventTracker) Stats 

func (t *EventTracker) Stats() (hosts int, windowKeys int, knownKeys int)

Stats 返回追踪器统计

type HitThrottler 

type HitThrottler struct {
	// contains filtered or unexported fields
}

HitThrottler 限制 (host, rule) 维度的告警生成频率。

现状：alerts 表已对 (rule, host) 去重（resultID = cel-{ruleID}-{hostID}）， 单 alert 行只累加 HitCount。但每次命中仍会写 storyline_events、触发通知节流判断、 走 SIEM forward、调用 BDE/IForest（未来）等。当某 host 一个规则疯狂命中 （prod 实测 nginx 触发 c2_high_risk_port 单 host 31k 次），即使 alert 表 只有 1 条，下游链路仍被打爆。

Throttler 在 alert generator 入口拦截：单位时间窗内同 (host, rule) 命中 超过 burstThreshold 后，标记为 throttled，调用方应跳过重写入。

设计要点：

  • 内存 LRU + 滑动窗口，不依赖 DB
  • 容量上限 lruCapacity，超出按 LRU 淘汰
  • 计数器周期重置（每 refillWindow）
  • 线程安全

func NewHitThrottler 

func NewHitThrottler(burstThreshold int, refillWindow time.Duration, capacity int) *HitThrottler

NewHitThrottler 创建告警频率限制器。

burstThreshold: 窗口内命中超过此值开启节流（建议 50-100）。 refillWindow: 计数窗口长度（建议 1 分钟）。 throttleDur: 超阈后静默时长（建议 10 分钟，与 alert.go notifyThrottleWindow 解耦）。 capacity: LRU 上限（建议 10000，覆盖 prod 209 host × 94 rule ≈ 20k 组合）。

func (*HitThrottler) Allow 

func (t *HitThrottler) Allow(hostID, ruleID string, now time.Time) bool

Allow 返回 true 表示允许该 (hostID, ruleID) 继续生成告警；false 表示已被节流。

副作用：每次调用更新内部计数器，超阈后开启 10min 静默窗。

func (*HitThrottler) Size 

func (t *HitThrottler) Size() int

Size 返回当前 bucket 数（用于 metrics / 监控）。

type IOCMatcher 

type IOCMatcher struct {
	// contains filtered or unexported fields
}

IOCMatcher 服务端 IOC 匹配器:从 ioc_snapshots 表加载全量 IOC 到内存集,供 engine 匹配。 服务端匹配(事件本就全量到 engine),不依赖给 agent 下发 IOC。

func NewIOCMatcher 

func NewIOCMatcher(db *gorm.DB, logger *zap.Logger) *IOCMatcher

NewIOCMatcher 创建匹配器

func (*IOCMatcher) CheckHash 

func (m *IOCMatcher) CheckHash(v string) bool

func (*IOCMatcher) CheckIP 

func (m *IOCMatcher) CheckIP(v string) bool

CheckIP / CheckHash / CheckURL 判断是否命中 IOC 集

func (*IOCMatcher) CheckURL 

func (m *IOCMatcher) CheckURL(v string) bool

func (*IOCMatcher) Count 

func (m *IOCMatcher) Count() (ip, hash, url int)

Count 返回各类 IOC 数(用于日志/统计)

func (*IOCMatcher) Reload 

func (m *IOCMatcher) Reload() error

Reload 从最新快照加载 IOC 集(版本未变则跳过)

func (*IOCMatcher) StartReload 

func (m *IOCMatcher) StartReload(ctx context.Context)

StartReload 周期重载,使新增 IOC(含自有情报研判提取)无需重启即生效

type ProcessNode 

type ProcessNode struct {
	PID       string
	PPID      string
	Exe       string
	Cmdline   string
	UID       string
	StartTime time.Time
	ExitTime  *time.Time // nil = still running
	// contains filtered or unexported fields
}

ProcessNode 进程树节点

type ProcessTree 

type ProcessTree struct {
	// contains filtered or unexported fields
}

ProcessTree 按主机维护的进程树 键: hostID → pid → ProcessNode

func NewProcessTree 

func NewProcessTree(logger *zap.Logger) *ProcessTree

NewProcessTree 创建进程树

func (*ProcessTree) AncestorHas 

func (t *ProcessTree) AncestorHas(hostID, pid, exePattern string) bool

AncestorHas 检查指定主机上的进程祖先链中是否有匹配 exe 的进程 用于 CEL 自定义函数 ancestor_has(exe_pattern)

func (*ProcessTree) Cleanup 

func (t *ProcessTree) Cleanup() int

Cleanup 清理过期进程节点（已退出且超过保留时间的）

func (*ProcessTree) GetAncestorChain 

func (t *ProcessTree) GetAncestorChain(hostID, pid string) []string

GetAncestorChain 返回进程的祖先链 exe 列表（从近到远）

func (*ProcessTree) HandleExec 

func (t *ProcessTree) HandleExec(hostID string, fields map[string]string)

HandleExec 处理 process_exec 事件，将新进程加入树

func (*ProcessTree) HandleExit 

func (t *ProcessTree) HandleExit(hostID string, fields map[string]string)

HandleExit 处理 process_exit 事件，标记进程退出时间

func (*ProcessTree) Stats 

func (t *ProcessTree) Stats() (hosts int, totalNodes int)

Stats 返回进程树统计

type ResponseAction 

type ResponseAction struct {
	Type   string `json:"type"`   // kill_process, quarantine_file, block_ip
	Target string `json:"target"` // PID / 文件路径 / IP
}

ResponseAction 自动响应动作

type ResponseRule 

type ResponseRule struct {
	RuleID  uint             `json:"rule_id"`
	Actions []ResponseAction `json:"actions"`
}

ResponseRule 自动响应规则

type ScanDetector 

type ScanDetector struct {
	// contains filtered or unexported fields
}

ScanDetector 端口扫描检测器

func NewScanDetector 

func NewScanDetector(rdb *redis.Client, db *gorm.DB, logger *zap.Logger) *ScanDetector

NewScanDetector 创建端口扫描检测器 rdb 为 nil 时不启用

func (*ScanDetector) CheckIncomingConnection 

func (d *ScanDetector) CheckIncomingConnection(hostID, remoteAddr, localPort string, fields map[string]string)

CheckIncomingConnection 检查入站连接是否构成端口扫描 hostID: 被扫描主机 ID remoteAddr: 扫描源 IP localPort: 被访问端口 fields: 事件原始字段（用于告警详情）

func (*ScanDetector) StartWhitelistReload 

func (d *ScanDetector) StartWhitelistReload(ctx context.Context)

StartWhitelistReload 启动后台 goroutine 周期 reload 源 IP 白名单。 调用方需保证只调用一次；ctx 取消时退出。

type SequenceDetector 

type SequenceDetector struct {
	// contains filtered or unexported fields
}

SequenceDetector 行为序列检测器

func NewSequenceDetector 

func NewSequenceDetector(engine *Engine, db *gorm.DB, redisClient *redis.Client, logger *zap.Logger) *SequenceDetector

NewSequenceDetector 创建序列检测器 db 可为 nil（测试场景），redisClient 为 nil 时使用进程内内存状态（单副本足够）

func (*SequenceDetector) Evaluate 

func (d *SequenceDetector) Evaluate(hostID string, dataType int32, fields map[string]string) []model.SequenceRule

Evaluate 评估事件是否命中序列规则的某个步骤 返回完整匹配（所有步骤按序命中）的规则列表

func (*SequenceDetector) ReloadRules 

func (d *SequenceDetector) ReloadRules() error

ReloadRules 从数据库加载启用的序列规则并预编译

func (*SequenceDetector) RuleCount 

func (d *SequenceDetector) RuleCount() int

RuleCount 返回当前加载的序列规则数量

func (*SequenceDetector) SetRules 

func (d *SequenceDetector) SetRules(rules []model.SequenceRule) int

SetRules 设置规则（用于测试，不依赖数据库）

func (*SequenceDetector) StartReload 

func (d *SequenceDetector) StartReload(ctx context.Context)

StartReload 启动周期性规则重载，使新增/修改的序列规则无需重启即可生效

type SequenceState 

type SequenceState struct {
	RuleID       uint      `json:"rule_id"`
	HostID       string    `json:"host_id"`
	CurrentStep  int       `json:"current_step"`
	StartTime    time.Time `json:"start_time"`
	MatchedSteps []int     `json:"matched_steps"`
}

SequenceState 序列检测的中间状态（Redis 持久化）

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.