• 人工智能
• AI Agent
• 网络安全
• 后端
• 前端

本文以 AiSOC 的 Syslog 监听器为主题，讲清楚它如何在没有 webhook 的老旧安全设备与 AiSOC 之间建立直接链路：四种线格式如何被识别、租户身份为什么只能来自 ingest token、四步部署流程与实际可抄的配置，以及解析器、批处理与可靠性边界在源码层面的实现。读完后你可以独立完成一次"防火墙 syslog 直连 AiSOC"的落地，并能用/readyz与 Prometheus 指标判断链路到底是真的在收，还是看起来像在收。

为什么需要原生 Syslog 监听器

防火墙、代理、负载均衡器、老一代 IDS 以及大量自研设备只会发 syslog，永远不会长出一个 webhook。在这一能力加入之前，AiSOC 唯一的 syslog 入口是POST /v1/inbox/cef，它要求前置一个已经会说 HTTP 的转发器——也就是说设备依然无法直接触达 AiSOC，真正到达的只是转发器。这个背景在解析器包的注释里写得非常直白，见 services/ingest/internal/syslog/parse.go。

监听器的定位是"让设备自己来找 AiSOC"。它在 ingest 服务进程内直接绑定 UDP 与 TCP 套接字，解析完成后批量投递到 Kafka，因此链路上不再有额外的采集器进程。

四种线格式，以及它们如何被区分

监听器读取四种格式，并且区分它们不是美学问题，而是正确性问题：

四种都不匹配的行，会被保留为非结构化消息且文本原样保留。它绝不会被包装成一条带"凭空捏造字段"的结构化记录——一条被错误切分的行会把半条消息塞进 hostname 列，而分析师读到的将是一台并不存在的主机。

这套判定逻辑在 services/ingest/internal/syslog/parse.go 的Parse函数中按固定顺序执行，顺序本身是有含义的：

switch { case isRFC5424(rest): parseRFC5424(msg, rest) case anchorAt(rest, "LEEF:") >= 0: at := anchorAt(rest, "LEEF:") parseRFC3164Header(msg, rest[:at]) msg.Message = rest[at:] parseLEEF(msg, rest[at:]) case anchorAt(rest, "CEF:") >= 0: at := anchorAt(rest, "CEF:") parseRFC3164Header(msg, rest[:at]) msg.Message = rest[at:] parseCEF(msg, rest[at:]) case looksLikeRFC3164(rest): parseRFC3164Header(msg, rest) msg.Format = FormatRFC3164 default: msg.Format = FormatUnstructured msg.Message = rest }

几个值得注意的工程取舍：

• LEEF 与 CEF 先于 RFC 3164 判断。一条 LEEF 行通常既有 3164 头部，又有 LEEF 载荷，而载荷才是有价值的部分；如果在头部解析之后才找锚点，3164 的 tag 读取器会把LEEF当成应用名并顺带吃掉锚点。
• isRFC5424只认"1 "前缀，这是防止一条消息体恰好以数字开头的 3164 行被读成 5424 的唯一标记，见 parse.go。
• looksLikeRFC3164刻意划得很窄：要求长度至少 15 且前三字节是三字母月份加一个空格。若放宽到"含冒号的都算"，普通散文就会被拆成主机名和消息，见 parse.go。

一个监听器，一个租户

租户来自你在服务配置里放的那枚由 AiSOC 生成的 ingest token，且只来自那里。

这不是需要绕开的限制，而是设计要点。syslog 不携带任何经过认证的主体，也没有请求头，因此从线路上读到的一切都是发送方控制的：如果租户取自 hostname、应用名或结构化数据参数，那就等于开了一个攻击者零成本就能触发的跨租户写入。多租户部署的做法是每个租户跑一个监听端口。

这条约束有专门的 CI 门禁在守。scripts/check_syslog_listener.py会剔除注释后扫描sink.go，对Hostname、AppName、ProcID、MsgID、Message、StructuredData、Extension、Raw这几个发送方可控字段，检查三种 Go 语法形态是否把它们接到了租户上：直接赋值、结构体字面量字段、以及作为tmpl.Apply的第二个参数（也就是租户参数）。见 scripts/check_syslog_listener.py。门禁的自测里甚至专门构造了这三种注入变体，每一条都必须被抓到。

真正实现侧对应的代码在 services/ingest/internal/syslog/sink.go：租户 ID 只从tok.TenantID取值。

tok, err := s.tokens.Resolve(ctx, s.token) if err != nil { return fmt.Errorf("syslog: resolving the configured ingest token: %w", err) } if tok.TemplateID != RequiredTemplateID { return fmt.Errorf( "syslog: the configured token was minted with template %q; the listener needs one minted with %q", tok.TemplateID, RequiredTemplateID, ) } ... tenantID := tok.TenantID.String()

注意那个模板校验：如果 token 是用别的模板签发的，监听器会直接报错并拒绝整批消息，而不是猜一个模板继续走。

配置实操：四步把设备接进来

第 1 步：签发 token

使用syslog模板签发：

curl -sS -X POST https://your-aisoc/api/v1/inbox/tokens \ -H "Authorization: Bearer $AISOC_API_TOKEN" \ -H 'Content-Type: application/json' \ -d '{"template_id": "syslog", "label": "edge firewalls"}'

响应中只出现一次 token，请立刻保存。

一枚 token 覆盖全部四种格式。一台设备通常对自己的守护进程发 RFC 3164，对安全事件发 CEF；如果要求你为一台设备签发两枚 token、开两个端口，那将是任何厂商文档都不会描述出来的配置。这个决定写在 sink.go 的RequiredTemplateID常量注释里，syslog、cef-syslog、leef-syslog三个模板 ID 同时也在 API 侧的可签发白名单中，见 services/api/app/api/v1/endpoints/inbox.py——白名单缺失会让照文档操作的运维直接拿到 400。

第 2 步：打开监听器

在 ingest 服务的环境变量中：

AISOC_SYSLOG_ENABLED=true AISOC_SYSLOG_TOKEN=<the token from step 1> AISOC_SYSLOG_UDP_ADDR=0.0.0.0:5514 AISOC_SYSLOG_TCP_ADDR=0.0.0.0:5514

默认端口是5514而不是 514，这是有意为之：514 属于特权端口，绑定它需要容器具备CAP_NET_BIND_SERVICE，或在宿主机上做514到5514的端口映射。请在你的 compose 文件或 Service 中发布 UDP 与 TCP 端口，并记住 Docker 发布的 UDP 端口必须显式写出/udp。

完整的可用参数与默认值，与 services/ingest/internal/config/config.go 中的读取逻辑一一对应：

把AISOC_SYSLOG_UDP_ADDR或AISOC_SYSLOG_TCP_ADDR置为空字符串，即可只跑一种传输。两个都置空会被拒绝而不是当作无操作接受：一个绑定到"什么都没有"的监听器，和一个没人往里发消息的监听器，外观完全一致。这条校验在 listener.go 的构造函数里：

if cfg.UDPAddr == "" && cfg.TCPAddr == "" { return nil, errors.New("syslog: neither UDP nor TCP address configured; set at least one or leave the listener disabled") }

同样值得知道的是：绑定发生在New()而不是Start()。端口已被占用会变成一个运维能看见的启动错误，而不是一个退出到没人读的日志里的 goroutine。UDP 或 TCP 任一端绑定失败，构造函数会关掉已经绑好的另一端并返回带传输名与地址的错误信息，其中还提示了"低于 1024 的端口需要 capability 或容器端口映射"。

一个可直接改用的 compose 片段（仓库自带的docker-compose.yml中 ingest 服务只发布了8081与9090，并未发布 5514，需要自行追加）：

services: ingest: ports: - "8081:8080" - "9090:9090" - "5514:5514/udp" - "5514:5514/tcp" environment: AISOC_SYSLOG_ENABLED: "true" AISOC_SYSLOG_TOKEN: ${AISOC_SYSLOG_TOKEN} AISOC_SYSLOG_UDP_ADDR: "0.0.0.0:5514" AISOC_SYSLOG_TCP_ADDR: "0.0.0.0:5514"

第 3 步：把设备指过来

大多数 Linux 发送端使用如下写法：

*.* @@aisoc-ingest.internal:5514 # TCP *.* @aisoc-ingest.internal:5514 # UDP

双@表示 TCP，单@表示 UDP，这是 rsyslog 的语法约定。

第 4 步：检查/readyz

监听器会以名称出现在subscriptions下，并带上它绑定了什么、已接收多少条、最近失败在什么地方：

{ "subscriptions": { "syslog": { "attached": true, "not_resolving": false, "detail": "listening on udp 0.0.0.0:5514, tcp 0.0.0.0:5514; received 1842, rejected 0" } } }

这段 detail 由 services/ingest/main.go 的syslogDetail生成：未绑定任何传输时输出no transport bound；否则拼接listening on <传输与地址>; received N, rejected M，若有LastError再追加; last error: ...。同时attached取两种传输的或值，not_resolving取其反值。

接收端连续数小时没有消息是完全正常的，因此"没有消息"什么也证明不了——这正是就绪状态报告计数器、而不是只报一个布尔值的原因。

一条消息之后发生了什么

这个分流很重要。Category 2 会被无条件提升为告警，所以把防火墙每一条例行守护进程日志都归成 Security Finding，就等于把每一行都变成告警——恰恰是这个平台要消除的噪声。一条普通 syslog 行只有在检测规则命中时才成为告警；而 CEF 或 LEEF 行会被提升，是因为发这种格式的设备自己已经判定这是一条安全事件。

映射关系定义在 services/ingest/internal/syslog/sink.go 的templateForFormat：

var templateForFormat = map[Format]string{ FormatRFC5424: "syslog", FormatRFC3164: "syslog", FormatUnstructured: "syslog", FormatCEF: "cef-syslog", FormatLEEF: "leef-syslog", }

三个模板文件都随二进制走，可分别查看：

• services/ingest/internal/normalizer/templates/syslog.yaml ——class_uid: 1007，字段映射把hostname落到device.hostname、app_name落到actor.process.name、proc_id落到actor.process.pid、raw落到raw_data，并在常量里标记source_type: syslog。
• services/ingest/internal/normalizer/templates/cef-syslog.yaml ——class_uid: 2001，把DeviceVendor/DeviceProduct落到metadata.product.*，src/dst/spt/dpt、suser/duser落到端点与用户字段。
• services/ingest/internal/normalizer/templates/leef-syslog.yaml ——class_uid: 2001，额外映射proto、action、policy、identSrc等 QRadar 常见键。

模板还承担严重度换算,——这是四种格式里最容易出错的一处：

• syslog 的严重度阶梯是 0 最严重、7 最轻，而 OCSF 的阶梯方向相反，因此syslog.yaml做的是"反转"而非"重新分级"：emerg: 6、alert: 5、crit: 5、err: 4、warning: 3、notice: 2、info: 1、debug: 1。
• CEF 的Severity是 0–10 且 10 最严重，cef-syslog.yaml用十档加Low/Medium/High/Very-High的字符值一起映射。
• LEEF 的sev同样是 1–10、1 最轻，leef-syslog.yaml对应映射到五档并额外接受Critical。

原始行会逐字保留在每个事件的raw_data上。这里每一次解析都是有损摘要，而设备发出了一些解析器读错的内容，本身就是一条发现——所以原文不能丢。

解析器内部的几个关键点

理解下面这些细节，能让你在排查"字段为什么不对"时少走很多弯路。

<PRI>是可选的。直接写 TCP socket 的设备经常省略它，因此缺失不算解析失败。缺失时解析器显式写入user.notice（priority 13、severity code 5、facilityuser），而不是留下一个会被读成kern.emerg——最严重的取值——的零值。见 parse.go。

结构化数据按键展平。RFC 5424 的 SD-ELEMENT 被展平成sdid@enterprise.param形式的键，再在Fields()里以sd.前缀暴露给模板：out["sd."+key] = value。CEF/LEEF 的扩展键则并入顶层而不是嵌套，因为现有cef-syslog模板就是在顶层取src、suser、act这些键；嵌套会意味着一种格式要两份模板。见 parse.go。

SD 切分用括号计数而不是正则。参数值里的]可以转义为\]，rsyslog 自己的origin元素还会塞进一个带方括号的x-infoURL。如果不识别引号与转义，扫描会永远停在 depth 1，整个 SD 块静默地落进消息体，而消息看起来还很像样。见 parse.go。

RFC 3164 的 TAG 切分加了双重条件。只有当冒号出现在 RFC 允许的 32 字节标签长度之内、且冒号之前没有空格时，才把冒号当分隔符。缺任一条件，Error: connection refused就会变成一个应用名。见 parse.go。

CEF 扩展不能用空格切。msg=connection refused src=10.0.0.1这样的尾部，按空格切就把消息切碎了。实现是扫描到下一个未转义的=，回退到该键之前的空格，以此结束上一个值——这是在没有 schema 的前提下唯一能正确读这个格式的办法，见 services/ingest/internal/syslog/kv.go。头部字段按|切分时也遵守\|转义，且要求至少 7 个字段，否则不标记为 CEF（宁可不认，也不产出一条"标记为 CEF 却厂商为空"的记录）。

LEEF 2.0 的分隔符字段会被解析。该字段可能是字面字符，也可能是xHH/x09形式给出的十六进制码。若假定是 tab，那么每对键值在一行以^分隔的数据里都会挤进同一个键，见 kv.go。

单行上限 64 KiB。MaxLineBytes = 64 * 1024：RFC 5424 要求接收端至少支持 2048 字节，64 KiB 远高于任何设备的实际输出，又远低于恶意发送方耗尽连接缓冲区所需的量。

部署前必须知道的限制

• TCP 分帧两种都读：RFC 6587 的八位组计数（123 <34>1 ...）与换行分隔。八位组计数是消息内部含换行时仍能存活的那一种，如果你的发送端支持，优先选它。实现上每条消息通过窥探 1 个字节决定分帧方式，若首个字节不是数字则退回换行分帧——所以一条恰好以数字开头的非分帧行不会被误当成长度前缀吞掉整个流。长度超过 64 KiB 或数字位数超过 10 位，会直接关闭连接而不是按这个长度分配内存。
• 投递失败即丢消息。syslog 没有可以扣住的确认，Kafka 不可达时该批消息被丢弃，计入aisoc_ingest_syslog_rejected_total并写进/readyz的 detail。这一层无法做到无损；对"丢失不可接受"的来源，请改用 HTTP inbox 路由。
• RFC 3164 时间戳没有年份。接收端会补上年份，并留一天余量，使 1 月 1 日刚过午夜到达的行归于它被写下的那一年而不是下一年。原始文本仍保留在raw_data中。
• 连接数有上限（AISOC_SYSLOG_MAX_CONNECTIONS，默认 256）。超过上限的连接被立即拒绝而不是挂着不放，这样发送方能发现自己没送进去，而不是误以为在正常投递。
• 空闲 TCP 连接会被关闭。读超时默认 5 分钟（Config.ReadTimeout的withDefaults），否则一个不告而别、没有发 FIN 的转发器会一直占着它的槽位直到进程重启。

批处理是另一个容易被忽略但影响很大的点。syslog 是一行一条到达的，每条都发一条 Kafka 记录正是让接收端垮掉的原因，所以按BatchSize（默认 500）与FlushInterval（默认 2 秒）双条件触发：规模触发解决吞吐，时间触发保证一个安静来源的那一条消息不会被无限期扣住。测试 services/ingest/internal/syslog/listener_test.go 同时验证了这两条路径，因为修好其中一个通常会弄坏另一个。

用指标与门禁判断链路是否真的在工作

监听器暴露三个指标，定义在 listener.go：

reason的取值来自源码中的埋点，排查时非常有指向性：read_error（UDP 读错误，通常由 ICMP 端口不可达等瞬时情况导致，循环继续）、too_many_connections（触达连接上限）、frame_error（分帧失败，连接被关闭）、parse_error（解析失败）、publish_error（投递失败，按批内条数累加）。

另外，空行不会被计入拒绝：ErrEmpty被单独区分出来，因为空的 UDP 数据报是保活而不是故障，把它算成畸形会让畸形行指标失去意义。

除了运行时指标，仓库还用一个静态门禁守住"这个功能确实被接线"这四件事，它用ast与纯文本解析、不需要 Go 工具链也能在 lint 任务里跑：

1. 四种格式真的被区分——要求每个常量都出现在某处Format = ...赋值中。常量存在不够，因为它是声明在一个文件、赋值在另一个文件里的。
2. 被构造且被启动——main.go必须调用syslog.New(、必须调用syslogListener.Start(、必须注册RegisterSubscription("syslog"。
3. 租户来自凭据——sink.go必须出现tok.TenantID，且不得从发送方可控字段派生租户。
4. 依赖的模板存在且可签发——三个 YAML 必须存在，且都在 API 的ALLOWED_TEMPLATE_IDS中。

对应实现见 scripts/check_syslog_listener.py，自测部分会逐条注入违规并要求每条都被抓到。

排错清单

小结

这个监听器真正的价值不在"支持四种格式"这句话本身，而在它围绕这句承诺做的一串防御性选择：格式靠格式自身定义的标记判定、判不出来就保留原文而不是猜；租户只能从凭据来，从线路读到的一切都视为不可信；绑定放在构造函数里让端口冲突成为可见的启动错误；就绪状态报告计数器与最近错误，因为"没有消息"对一个 syslog 接收端本来就是常态；投递失败宁可丢并计数，也不把批次留在内存里等到进程被杀。理解这些取舍之后，/readyz的那一行 detail 才真正读得懂，也才能在防火墙、代理与老 IDS 这些永远不会长 webhook 的设备上，把它们的日志稳定地接进 AiSOC 的检测与调查流水线。

• 人工智能
• AI Agent
• 网络安全
• 后端
• 前端