使用 STIX-Shifter 构建与 IBM Cloud Pak for Security 的连接器

简介

IBM Cloud Pak for Security 提供了联合搜索,用于调查和分析整个公司存在的安全洞察。执行查询时,无需将您的数据从其源中移出。Cloud Pak for Security 提供了 Universal Data Insights (UDI) 服务,这是 Cloud Pak 中的主要共享数据服务层。

任何应用程序都可以通过 UDI 服务 API 从各种共享数据源查询或读取安全数据。它使用 RESTful 原则构建,并与 STIX-Shifter 的可扩展开源软件开发工具包 (SDK) 集成,从而能够使 IBM 和全球安全性社区贡献新的转换和传输模块。开源库包含各种用于测试的安全产品、文档和 CLI 的连接器的样本和示例。

针对您的数据源,为其构建并实施 STIX 适配器,以在该数据源和 Cloud Pak for Security 之间建立连接。使用开源库

  • 连接任何储存网络安全数据的产品。
  • 将 STIX 2 模式转换为本地数据源查询。
  • 将 JSON 数据源查询结果转换为可观察对象的 STIX 捆绑包。

了解有关开发新的 STIX-Shifter 适配器和受支持的查询数据类型的更多信息。

在本教程中,您将学习如何使用 STIX-Shifter 开发您自己的 UDI 连接器。

前提条件

在开始构建 STIX-Shifter 连接器之前:

预估时间

使用 STIX-Shifter SDK 开发基本的 hello world 连接器需要 8 个小时。

步骤

遵循以下步骤来开发和测试连接器。

1. 设置开发环境

必须使用 Python 3.6 作为开发环境。

您可以使用自己的集成开发环境,如 Visual Studio Code

使用以下步骤配置和测试开发者环境:

  • 安装 Python 3.6
  • 通过打开终端窗口并输入 python -version 来验证 Python 3.6 是否已安装。
  • 派生 STIX-Shifter 项目
  • 克隆存储库并创建新的工作分支:

    从 GitHub 中的 Clone 菜单复制存储库 URL。

    在克隆存储库所在的相同终端中,将目录更改为新创建的存储库目录。

  • 创建您将在其中运行 STIX-Shifter CLI 命令的虚拟环境。这将安装 STIX-Shifter 所需的所有 Python 软件包。

    在终端中运行以下命令:

     virtualenv -p python3.6 –no-site-packages –distribute virtualenv && source virtualenv/bin/activate && pip install -r requirements-dev.txt
    
  • 通过在终端中运行以下命令来测试您的设置:

    python main.py translate dummy query '{}' "[ipv4-addr:value = '0.0.0.0']"
    

    如果所有设置均正确,您将看到已转换的查询:{'queries': ["SELECT * FROM tableName WHERE (SourceIpV4 = '0.0.0.0' OR DestinationIpV4 = '0.0.0.0')"]}

2. 构建连接器

从开源 STIX-Shifter 项目创建连接器。以派生 STIX-Shifter 项目为起始,开发新的连接器并创建工作分支。工作分支将包含新连接器所需的任何代码更改。

新连接器准备好可供审查时,提交拉取请求 (PR)。PR 请求会将所派生项目中的工作分支合并到 STIX-Shifter 项目的主分支。提交 PR 后,将对建议的更改执行代码审查。也可以使用 STIX-Shifter 的 CLI 命令在本地开发环境中对工作分支进行克隆和测试。

使用以下步骤为数据源开发 STIX-Shifter 连接器:

3. 测试连接器

测试连接器的转换和传输功能。测试对于确保连接器正常运作非常重要。STIX-Shifter 开发者指南将指导您在下面所列的开发过程的每个步骤中完成测试。

转换

  • 使用 python main.py translate query '{}'将 STIX 模式转换为本机查询。

    python main.py translate dummy query '{}' "[network-traffic:src_port NOT = 37020 AND network-traffic:dst_port != 635] OR [ipv4-addr:value = '333.333.333.0'] OR [url:value = 'www.example.com']" '{"validate_pattern": "true"}'
    

    validate_pattern 选项对提交的模式执行强制验证。这项检查会在模式进入转换逻辑之前执行。格式错误的模式将抛出错误。

  • 使用 python main.py translate results 将 JSON 结果转换为 STIX 对象。

    identity 对象表示数据源,在 UDI 流期间会自动包含该对象,但在 CLI 中进行测试时必须手动包含该对象。JSON 结果通常来自数据源的 API 结果调用,但必须包含测试有效负载,CLI 才能对转换为 STIX 的结果进行测试。

    python main.py translate dummy results '{"type: "identity","id": "identity–f431f809-377b-45e0-aa1c-6a4751cae5ff", "name": "DummyAdapter","identity_class": "system"}' '[ { "SourcePort": 1234, "EventCount": 1, "Magnitude": 4, "DestinationIpV4": "0.0.0.0", "SourceIpV4": "9.21.123.112", "DestinationPort": 5678, "UserName": "root", "StartTime": 1540312200111}]"{"stix_validator": true}'
    

传输

传输命令包括 pingquerystatusresultsdeleteis_async

  • Ping 通过 API 尝试与数据源进行连接。

    python main.py transmit dummy '{"host":"some.fake.address.com", "port":"12345"}' '{"auth":{"username": "some_user_name", "password": "some password"}}' ping
    
  • Query 通过 API 将查询提交到数据源。

    python main.py transmit dummy ''{"host":"some.fake.address.com", "port":"12345"}' '{"auth":{"username": "some_user_name", "password": "some password"}}' query "SELECT * FROM tableName"
    
  • Status 检查查询的当前状态(如果查询是异步的)

    python main.py transmit dummy '{"host":"some.fake.address.com", "port":"12345"}' '{"auth":{"username": "some_user_name", "password": "some password"}}' status"b5751e36-1699-4d57-9f7e-a020b4f91176"
    
  • Results 通过 API 获取查询结果。页面和偏移量数字定义了所返回总结果的范围(子集)。

    python main.py transmit dummy '{"host":"some.fake.address.com", "port":"12345"}' '{"auth":{"username": "some_user_name", "password": "some password"}}' results"b5751e36-1699-4d57-9f7e-a020b4f91176"1 3
    
  • Delete 通过 API 删除查询(如果受数据源支持)。

    python main.py transmit dummy '{"host":"some.fake.address.com", "port":"12345"}' '{"auth":{"username": "some_user_name", "password": "some password"}}' delete "b5751e36-1699-4d57-9f7e-a020b4f91176"
    
  • Is_async 返回 true/false 值,用于指示数据源是否使用异步查询。

    python main.py transmit dummy '{"host":"some.fake.address.com", "port":"12345"}' '{"auth":{"username": "some_user_name", "password": "some password"}}' is_async
    

执行

execute 命令提供了测试整个转换——传输流的方法。从 CLI 调用 execute 时,将发生以下情况:

  • 首先验证提交的模式,然后将其转换为本地查询。
  • 本地查询传输到数据源。
  • 执行 is_asynccheck。如果结果为 true,将传输状态检查以确定查询是否完成。
  • 从数据源获取(传输)查询结果。如果数据源是同步的,此操作将立即执行,否则会根据查询 ID 来查找结果。
  • 结果转换为 STIX 网络可观察对象的捆绑包。
  • 对每个已转换对象运行 STIX 验证。

结束语

在本教程中,您了解了如何使用 STIX-Shifter 构建连接器。如果您希望了解更多信息,可查看 GitHub 上的 STIX-Shifter 开源项目中 Open Cybersecurity Alliance 下的其他资源。想要就本主题与专家互动,可填写 Open Cybersecurity Alliance Slack Invite Form 以加入 Slack 频道。

本文翻译自:Build a connector to IBM Cloud Pak for Security with STIX-Shifter(2020-06-26)