cognis-digital/cognis-connect
GitHub: cognis-digital/cognis-connect
Cognis 安全工具套件的集成 SDK，通过统一 Finding 契约和一组适配器将工具输出标准化并路由至 STIX、MISP、Sigma、Splunk、Elastic、Slack 等主流安全平台。
Stars: 0 | Forks: 0
# cognis-connect
**[Cognis suite](https://github.com/cognis-digital) 的集成 SDK。** 统一的
**Finding** 契约，每个工具都映射至此 —— 同时配备适配器，将这些
发现路由至 **STIX/TAXII、MISP、Sigma、Splunk、Elastic、Slack、Discord、webhook**，以及
你的 **edgemesh `/v1` fleet**。纯标准库，零依赖。
这正是让 300 多个工具实现真正*组合*的关键：每个工具输出 JSON，`Finding`
将其标准化，一组适配器即可将其交付到你的 SOC 现有的任何平台。
```
flowchart LR
subgraph tools["any Cognis tool (--format json)"]
MI[maritimeint]; IO[iocextract]; CV[cvecheck]; GE[geolens]; UL[uaslog]
end
tools --> F[["Finding
(the contract)"]] F --> STIX[STIX 2.1 / TAXII] F --> MISP[MISP] F --> SIG[Sigma] F --> SIEM[Splunk · Elastic] F --> CHAT[Slack · Discord · webhook] F --> AI[edgemesh /v1
summarize] classDef c fill:#6b46c1,color:#fff; class F c; ``` ## 安装 ``` pip install "git+https://github.com/cognis-digital/cognis-connect.git" ``` ## 使用 — 通过 shell 将任何工具的 JSON 直接管道传输至目标平台： ``` maritimeint locate fleet.csv --sanctions s.json --format json \ | cognis-connect emit --to stix --source maritimeint > bundle.stix.json iocextract scan log.txt --format json | cognis-connect emit --to misp --url $MISP --token $KEY cvecheck . --format json | cognis-connect emit --to sigma > rules.yml cat findings.json | cognis-connect emit --to splunk --url $HEC --token $TOK --dry-run cat findings.json | cognis-connect emit --to slack --url $SLACK_WEBHOOK cat findings.json | cognis-connect emit --to brief # analyst summary via your fleet ``` `--dry-run` 可预览确切的 HTTP 请求（方法/url/headers/body）而不实际发送 —— 因此你可以安全地构建 pipeline 并进行离线测试。 ## 使用 — 通过 Python ``` from cognis_connect import load, stix, misp, sigma, siem, notify, edgemesh findings = load("vessels.json", source="maritimeint") # normalize any tool's JSON stix.to_bundle(findings) # -> STIX 2.1 dict misp.to_event(findings) # -> MISP event print(sigma.to_rules(findings)) # -> Sigma YAML siem.send_splunk(findings, hec_url, token, dry_run=True) # preview the HEC POST notify.send_slack(findings, webhook_url) # post to Slack edgemesh.summarize(findings) # one-paragraph brief via /v1 ``` ### 契约 每个工具要么输出 `Finding`，要么通过 `normalize()` 处理其记录。Field/indicator 的别名会被自动映射；原始记录会保留在 `raw` 中： ``` from cognis_connect import normalize f = normalize({"name": "C2 beacon", "risk": "high", "ip": "203.0.113.5", "sha256": "…"}, source="iocextract") # Finding(title='C2 beacon', severity='high', source='iocextract', # indicators={'ipv4': '203.0.113.5', 'sha256': '…'}) ``` | 标准字段 | 接受的别名 | |---|---| | `title` | name, summary, msg | | `severity` | sev, level, risk, 0–4, warn/error/crit | | indicators | ip→ipv4, hostname/fqdn→domain, hash→sha256, latitude/longitude, vessel_imo→imo, … | ## 平台 | `--to` | 输出 / 目标 | |---|---| | `stix` / `taxii` | STIX 2.1 bundle（确定性 ID）/ 推送至 TAXII 2.1 collection | | `misp` | MISP Event（带类型的 attributes，威胁等级）/ 推送至 MISP 实例 | | `sigma` | Sigma 检测规则骨架（无需 pyyaml） | | `splunk` / `elastic` | Splunk HEC 事件 / Elastic `_bulk` NDJSON | | `slack` / `discord` / `webhook` | 聊天通知 / 通用 JSON webhook | | `brief` | 通过 edgemesh `/v1` fleet 生成的一段式分析师摘要 | ## 互操作性 参见 **[INTEROP.md](INTEROP.md)** —— cognis-connect 是构建该套件组合模式和参考技术栈的骨干。**300+ 工具 →** [github.com/cognis-digital](https://github.com/cognis-digital)。 ## 许可证 [COCL 1.0](LICENSE) —— Cognis 开放协作许可证。© 2026 Cognis Digital LLC。
(the contract)"]] F --> STIX[STIX 2.1 / TAXII] F --> MISP[MISP] F --> SIG[Sigma] F --> SIEM[Splunk · Elastic] F --> CHAT[Slack · Discord · webhook] F --> AI[edgemesh /v1
summarize] classDef c fill:#6b46c1,color:#fff; class F c; ``` ## 安装 ``` pip install "git+https://github.com/cognis-digital/cognis-connect.git" ``` ## 使用 — 通过 shell 将任何工具的 JSON 直接管道传输至目标平台： ``` maritimeint locate fleet.csv --sanctions s.json --format json \ | cognis-connect emit --to stix --source maritimeint > bundle.stix.json iocextract scan log.txt --format json | cognis-connect emit --to misp --url $MISP --token $KEY cvecheck . --format json | cognis-connect emit --to sigma > rules.yml cat findings.json | cognis-connect emit --to splunk --url $HEC --token $TOK --dry-run cat findings.json | cognis-connect emit --to slack --url $SLACK_WEBHOOK cat findings.json | cognis-connect emit --to brief # analyst summary via your fleet ``` `--dry-run` 可预览确切的 HTTP 请求（方法/url/headers/body）而不实际发送 —— 因此你可以安全地构建 pipeline 并进行离线测试。 ## 使用 — 通过 Python ``` from cognis_connect import load, stix, misp, sigma, siem, notify, edgemesh findings = load("vessels.json", source="maritimeint") # normalize any tool's JSON stix.to_bundle(findings) # -> STIX 2.1 dict misp.to_event(findings) # -> MISP event print(sigma.to_rules(findings)) # -> Sigma YAML siem.send_splunk(findings, hec_url, token, dry_run=True) # preview the HEC POST notify.send_slack(findings, webhook_url) # post to Slack edgemesh.summarize(findings) # one-paragraph brief via /v1 ``` ### 契约 每个工具要么输出 `Finding`，要么通过 `normalize()` 处理其记录。Field/indicator 的别名会被自动映射；原始记录会保留在 `raw` 中： ``` from cognis_connect import normalize f = normalize({"name": "C2 beacon", "risk": "high", "ip": "203.0.113.5", "sha256": "…"}, source="iocextract") # Finding(title='C2 beacon', severity='high', source='iocextract', # indicators={'ipv4': '203.0.113.5', 'sha256': '…'}) ``` | 标准字段 | 接受的别名 | |---|---| | `title` | name, summary, msg | | `severity` | sev, level, risk, 0–4, warn/error/crit | | indicators | ip→ipv4, hostname/fqdn→domain, hash→sha256, latitude/longitude, vessel_imo→imo, … | ## 平台 | `--to` | 输出 / 目标 | |---|---| | `stix` / `taxii` | STIX 2.1 bundle（确定性 ID）/ 推送至 TAXII 2.1 collection | | `misp` | MISP Event（带类型的 attributes，威胁等级）/ 推送至 MISP 实例 | | `sigma` | Sigma 检测规则骨架（无需 pyyaml） | | `splunk` / `elastic` | Splunk HEC 事件 / Elastic `_bulk` NDJSON | | `slack` / `discord` / `webhook` | 聊天通知 / 通用 JSON webhook | | `brief` | 通过 edgemesh `/v1` fleet 生成的一段式分析师摘要 | ## 互操作性 参见 **[INTEROP.md](INTEROP.md)** —— cognis-connect 是构建该套件组合模式和参考技术栈的骨干。**300+ 工具 →** [github.com/cognis-digital](https://github.com/cognis-digital)。 ## 许可证 [COCL 1.0](LICENSE) —— Cognis 开放协作许可证。© 2026 Cognis Digital LLC。
标签：Python SDK, SOC集成, 威胁情报, 安全编排与自动化, 开发者工具, 数据适配器, 逆向工具