一次扫描为多条检测规则筛选候选
Suricata 的 ac 后端把多条规则的模式构建为 Aho–Corasick 自动机,用预先展开的状态转移表扫描载荷,并把命中的规则编号交给后续检测。
工程问题#
网络检测规则可能包含大量不同的内容模式。如果对每条规则分别扫描同一段载荷,会重复读取相同字节;但仅命中一段字符串也不等于整条规则已经满足。
实际实现#
选择 mpm-algo: ac 时,MPM 的 Search 入口指向 SCACSearch。准备阶段先建 goto 和 failure 表,再把回退结果展开为完整 delta 转移表;扫描时每个字节据当前状态查表,读取输出模式并将关联规则编号加入 PrefilterRuleStore。大小写敏感模式还需核对原文,offset、depth 和重复候选也有独立处理。
处理流程#
- 将规则预筛选模式构建为前缀状态图,按广度优先次序计算 failure 与输出集合
- 将失败回退预先折入完整状态转移表
- 载荷预筛选器通过 MPM 接口调用 ac 扫描函数,逐字节转移状态
- 检查命中模式的约束并收集规则编号,后续检测继续判断整条规则
源码片段#
固定版本 · 原文摘录117static void PrefilterPktPayload(DetectEngineThreadCtx *det_ctx,118 Packet *p, const void *pectx)119{120 SCEnter();121122 const MpmCtx *mpm_ctx = (MpmCtx *)pectx;123 if (p->payload_len < mpm_ctx->minlen)124 SCReturn;125126 (void)mpm_table[mpm_ctx->mpm_type].Search(mpm_ctx,127 &det_ctx->mtc, &det_ctx->pmq,128 p->payload, p->payload_len);129130 PREFILTER_PROFILING_ADD_BYTES(det_ctx, p->payload_len);131}132133int PrefilterPktPayloadRegister(DetectEngineCtx *de_ctx,134 SigGroupHead *sgh, MpmCtx *mpm_ctx)135{136 return PrefilterAppendPayloadEngine(de_ctx, sgh,137 PrefilterPktPayload, mpm_ctx, NULL, "payload");载荷达到最短模式长度才交给配置的 MPM Search,结果写入当前线程的预筛选规则集合;这里给出算法在检测管线中的实际入口。
1196void MpmACRegister(void)1197{1198 mpm_table[MPM_AC].name = "ac";1199 mpm_table[MPM_AC].InitCtx = SCACInitCtx;1200 mpm_table[MPM_AC].InitThreadCtx = SCACInitThreadCtx;1201 mpm_table[MPM_AC].DestroyCtx = SCACDestroyCtx;1202 mpm_table[MPM_AC].DestroyThreadCtx = SCACDestroyThreadCtx;1203 mpm_table[MPM_AC].AddPattern = SCACAddPatternCS;1204 mpm_table[MPM_AC].AddPatternNocase = SCACAddPatternCI;1205 mpm_table[MPM_AC].Prepare = SCACPreparePatterns;1206 mpm_table[MPM_AC].Search = SCACSearch;1207 mpm_table[MPM_AC].PrintCtx = SCACPrintInfo;名字 ac 的后端明确把 Prepare 和 Search 绑定到 SCAC 实现,连接配置与扫描函数。
691static void SCACPrepareStateTable(MpmCtx *mpm_ctx)692{693 SCACCtx *ctx = (SCACCtx *)mpm_ctx->ctx;694695 /* create the 0th state in the goto table and output_table */696 SCACInitNewState(mpm_ctx);697698 SCACDetermineLevel1Gap(mpm_ctx);699700 /* create the goto table */701 SCACCreateGotoTable(mpm_ctx);702 /* create the failure table */703 SCACCreateFailureTable(mpm_ctx);704 /* create the final state(delta) table */705 SCACCreateDeltaTable(mpm_ctx);706 /* club the output state presence with delta transition entries */707 SCACClubOutputStatePresenceWithDeltaTable(mpm_ctx);708709 /* club nocase entries */710 SCACInsertCaseSensitiveEntriesForPatterns(mpm_ctx);先构建 goto、failure 与 delta 表,再合入输出状态标记;扫描阶段无需逐次执行 failure 链回退。
499 while (!SCACStateQueueIsEmpty(q)) {500 /* pick up every state from the queue and add failure transitions */501 r_state = SCACDequeue(q);502 for (ascii_code = 0; ascii_code < 256; ascii_code++) {503 int32_t temp_state = ctx->goto_table[r_state][ascii_code];504 if (temp_state == SC_AC_FAIL)505 continue;506 SCACEnqueue(q, temp_state);507 state = ctx->failure_table[r_state];508509 while(ctx->goto_table[state][ascii_code] == SC_AC_FAIL)510 state = ctx->failure_table[state];511 ctx->failure_table[temp_state] = ctx->goto_table[state][ascii_code];512 SCACClubOutputStates(temp_state, ctx->failure_table[temp_state],513 mpm_ctx);514 }515 }按队列次序处理状态,沿 failure 查找可用转移,并把失败状态的输出集合并入当前状态,保留后缀模式命中。
1001 if (ctx->state_count < 32767) {1002 register SC_AC_STATE_TYPE_U16 state = 0;1003 SC_AC_STATE_TYPE_U16 (*state_table_u16)[256] = ctx->state_table_u16;1004 for (i = 0; i < buflen; i++) {1005 state = state_table_u16[state & 0x7FFF][u8_tolower(buf[i])];1006 if (state & 0x8000) {1007 uint32_t no_of_entries = ctx->output_table[state & 0x7FFF].no_of_entries;1008 uint32_t *pids = ctx->output_table[state & 0x7FFF].pids;1009 uint32_t k;1010 for (k = 0; k < no_of_entries; k++) {1011 if (pids[k] & AC_CASE_MASK) {1012 uint32_t lower_pid = pids[k] & AC_PID_MASK;1013 const SCACPatternList *pat = &pid_pat_list[lower_pid];1014 const int offset = i - pat->patlen + 1;10151016 if (offset < (int)pat->offset || (pat->depth && i > pat->depth))1017 continue;10181019 if (SCMemcmp(pat->cs, buf + offset, pat->patlen) != 0)1020 {1021 /* inside loop */1022 continue;逐字节查 delta 表,检查输出标记;大小写敏感模式还要对原文字节比较,因而完整搜索成本不只有查表次数。
1032 const SCACPatternList *pat = &pid_pat_list[pids[k]];1033 const int offset = i - pat->patlen + 1;10341035 if (offset < (int)pat->offset || (pat->depth && i > pat->depth))1036 continue;10371038 if (bitarray[pids[k] / 8] & (1 << (pids[k] % 8))) {1039 ;1040 } else {1041 bitarray[pids[k] / 8] |= (1 << (pids[k] % 8));1042 PrefilterAddSids(pmq, pat->sids, pat->sids_size);1043 }1044 matches++;1045 }对不要求大小写原文复核的分支检查 offset/depth,用位数组去重已加入的模式,再把关联的规则编号加入预筛选集合。
六段原文串起管线入口、后端注册、自动机构建、failure 处理、字节转移和候选收集。没有把多模式字符串命中写成整条检测规则或告警判定。
收益与取舍#
工程分析带来的收益
- 多个模式共享同一次输入扫描,避免每个模式独立从头搜索
- 把失败跳转的计算移到准备阶段,扫描主循环以直接查表推进状态
付出的代价
- S 个状态、256 字节字母表的稠密 delta 表占 O(256S) 空间,另有输出列表、模式和规则关联信息;增加规则会带来准备与内存成本
- 状态转移阶段随输入长度 n 线性增长,但输出遍历、大小写原文核对和候选规则数量仍会增加实际工作量,不能把整个检测阶段一概写成 O(n)
- 命中仅用于预筛选;复杂规则仍需要后续匹配,预筛选不是完整入侵检测语义
可比较的选择
逐模式 KMP每个模式状态更简单,但 k 个模式独立扫描可能需要约 O(kn) 的扫描工作;适合模式数量少或需要独立生命周期的场景。
Suricata 的其他 MPM 后端Hyperscan、ac-ks 等使用不同实现和平台条件;需要针对规则、流量、CPU 与内存评估,不能由算法名称推断统一性能排名。
版本与适用范围#
对应 Suricata suricata-7.0.8,明确限定 mpm-algo: ac;不将所有部署或其他 AC 变体归为本实现。片段分析数据包载荷的预筛选入口,SCACSearch 本身从状态 0 开始,不负责跨 TCP 包的重组。16 位扫描分支的条件是 state_count < 32767,其最高位记录输出标志;更大状态数走 32 位分支,不能沿用文件旧注释把阈值写成 2^16。
参考来源#
源码、官方文档与原始论文src/detect-engine-payload.c · PrefilterPktPayload · 载荷预筛选入口
src/util-mpm-ac.c · MpmACRegister / SCACPrepareStateTable / SCACSearch
src/detect-engine-mpm.c · PatternMatchDefaultMatcher · mpm-algo 选择