Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

整合 Apache Kafka 與 Google SecOps

本文說明如何將 Apache Kafka 與 Google Security Operations (Google SecOps) 整合。

用途

Apache Kafka 整合功能可解決下列用途：

即時擷取安全記錄：自動擷取 Kafka 主題中的安全事件，並處理這些事件，然後匯入 Google SecOps。集中管理記錄和即時分析串流資料，並根據分析結果產生快訊。
事件導向的自動化：根據 Kafka 主題串流的特定安全事件或訊息，在 Google SecOps 中觸發自動化劇本。這有助於加快對重大事件的回應速度，例如使用者從不尋常的位置登入。
威脅情報擴充功能：從 Kafka 主題提取自訂威脅情報動態消息，擴充現有警告和案件。這可為分析師提供最新的入侵指標 (IOC) 背景資訊，並提升威脅分析的準確度。

事前準備

在 Google SecOps 中設定 Apache Kafka 整合前，請先完成下列必要條件：

Apache Kafka 伺服器：請確認您有權存取正在執行的 Apache Kafka 伺服器，且已設定必要的 Kafka 代理程式和主題。
遠端代理程式 Docker 映像檔：建立遠端代理程式時，必須使用以 Debian 為基礎的映像檔。請使用下圖確保相容性：
```
us-docker.pkg.dev/siem-ar-public/images/agent-debian:latest
```

整合參數

整合 Apache Kafka 時，需要下列參數：

參數	說明
`Kafka brokers`	必填。以半形逗號分隔的 Kafka 代理程式清單，格式為 `hostname:port`。
`Use TLS for connection`	選填。如果選取這個選項，整合服務會使用 TLS 加密進行驗證。這項參數需要憑證授權單位 (CA) 憑證。預設為停用。
`Use SASL PLAIN with TLS for connection`	選填。如果選取這個選項，整合服務會使用 SASL PLAIN 使用者名稱和密碼機制進行驗證。這個選項僅支援 TLS 加密，且需要 SASL 使用者名稱和密碼，以及 CA 憑證。預設為停用。
`CA certificate of Kafka server`	選填。用於驗證 Kafka 伺服器身分的 CA 憑證。如果啟用 SASL，則必須使用這項參數。
`Client certificate`	選填。用戶端憑證，用於與 Kafka 伺服器進行雙向 TLS 驗證。如果啟用雙向傳輸層安全標準 (TLS)，則必須提供這個參數。
`Client certificate key`	選填。與用戶端憑證對應的私密金鑰，用於雙向 TLS 驗證。如果啟用雙向傳輸層安全標準 (TLS)，則必須提供這個參數。
`Client certificate key password`	選填。用來解密用戶端憑證私密金鑰的密碼。如果啟用雙向傳輸層安全標準 (TLS)，則必須提供這個參數。
`SASL PLAIN Username`	選填。用於 Kafka 代理程式的 SASL PLAIN 驗證使用者名稱。如果啟用 SASL，則必須使用這項參數。
`SASL PLAIN Password`	選填。用於 SASL PLAIN 驗證的密碼，可連線至 Kafka 代理程式。如果啟用 SASL，則必須使用這項參數。

如需在 Google SecOps 中設定整合功能的操作說明，請參閱「設定整合功能」。

如有需要，您可以在稍後階段進行變更。設定整合執行個體後，您就可以在劇本中使用該執行個體。如要進一步瞭解如何設定及支援多個執行個體，請參閱「支援多個執行個體」。

動作

如要進一步瞭解動作，請參閱「從工作台回覆待處理動作」和「執行手動動作」。

乒乓

使用「Ping」動作測試與 Apache Kafka 的連線。

這項操作不會在 Google SecOps 實體上執行。

動作輸入內容

無

動作輸出內容

「Ping」動作會提供下列輸出內容：

動作輸出類型	可用性
案件總覽附件	無法使用
案件總覽連結	無法使用
案件總覽表格	無法使用
補充資訊表格	無法使用
JSON 結果	無法使用
輸出訊息	可用
指令碼結果。	可用

輸出訊息

「Ping」動作可能會傳回下列輸出訊息：

輸出訊息訊息說明

輸出訊息	訊息說明
`Successfully connected to the Apache Kafka server with the provided connection parameters!`	動作成功。
`Failed to connect to the Apache Kafka server! Error is ERROR_REASON`	動作失敗。請檢查伺服器連線、輸入參數或憑證。

Successfully connected to the Apache Kafka server with the provided connection parameters!

動作成功。

Failed to connect to the Apache Kafka server!
      Error is  ERROR_REASON

動作失敗。

請檢查伺服器連線、輸入參數或憑證。

指令碼結果

下表列出使用「Ping」動作時，指令碼結果輸出的值：

指令碼結果名稱	值
`is_success`	`True`或`False`

連接器

如要進一步瞭解如何在 Google SecOps 中設定連接器，請參閱「擷取資料 (連接器)」。

Apache Kafka - Messages 連接器

使用 Apache Kafka - Messages 連接器從 Apache Kafka 擷取訊息。

這個連接器會從指定的 Kafka 主題擷取訊息，並根據訊息格式以不同方式處理訊息。如果訊息是有效的 JSON 物件，連接器會擷取特定欄位，用於建立及對應快訊。如果訊息是純字串，系統會將其擷取為原始事件資料。

連接器會根據您提供的參數，處理嚴重程度對應、快訊名稱範本，以及產生專屬 ID。

JSON 嚴重程度對應

如要對應快訊嚴重程度，請指定 Apache Kafka - Messages Connector 用來取得 Severity Mapping JSON 參數嚴重程度值的欄位。連結器回應可以包含值類型，例如 integer、float 和 string。

Apache Kafka - Messages Connector 會讀取 integer 和 float 值，並根據 Google SecOps 設定對應這些值。下表顯示 integer 值與 Google SecOps 中嚴重程度的對應關係：

整數值	對應的嚴重程度
`100`	`Critical`
自 `80` 至 `100`	`High`
自 `60` 至 `80`	`Medium`
自 `40` 至 `60`	`Low`
小於 `40`	`Informational`

如果回應包含 string 值，則 Pub/Sub - Messages Connector 需要額外設定。

一開始，預設值會顯示如下：

{
    "Default": 60
}

如果對應所需的鍵值位於 event_severity JSON 鍵中，則鍵值如下所示：

"Malicious"
"Benign"
"Unknown"

如要剖析 event_severity JSON 金鑰值，並確保 JSON 物件格式正確，請依下列方式設定 Severity Mapping JSON 參數：

{
    "event_severity": {
        "Malicious": 100,
        "Unknown": 60,
        "Benign": -1
    },
    "Default": 50
}

必須提供 "Default" 值。

如果同一個 JSON 物件有多個相符項目，Apache Kafka - Messages Connector 會優先處理第一個 JSON 物件鍵。

如要處理含有 integer 或 float 值的欄位，請在 Severity Mapping JSON 參數中設定鍵和空白字串：

{
  "Default":"60",
  "integer_field": "",
  "float_field": ""
}

連接器輸入內容

Apache Kafka - Messages Connector 需要下列參數：

參數	說明
`Product Field Name`	必填。儲存產品名稱的欄位名稱。產品名稱主要會影響對應。為簡化及改善連接器的對應程序，預設值會解析為程式碼參照的回退值。這個參數的任何無效輸入內容，預設都會解析為備用值。預設值為 `Product Name`。
`Event Field Name`	必填。決定事件名稱 (子類型) 的欄位名稱。預設值為 `event_type`。
`Environment Field Name`	選填。儲存環境名稱的欄位名稱。如果缺少環境欄位，連接器會使用預設值。預設值為 `""`。
`Environment Regex Pattern`	選填。要在 `Environment Field Name` 欄位中找到的值上執行的規則運算式模式。這個參數可讓您使用規則運算式邏輯，操控環境欄位。使用預設值 `.*` 擷取必要的原始 `Environment Field Name` 值。如果規則運算式模式為空值或空白，或環境值為空值，最終環境結果就是預設環境。
`Script Timeout (Seconds)`	必填。執行目前指令碼的 Python 程序逾時限制 (以秒為單位)。預設值為 `180`。
`Kafka brokers`	必填。以半形逗號分隔的 Kafka 代理程式清單，格式為 `hostname:port`。
`Use TLS for connection`	選填。如果選取這個選項，整合服務會使用 TLS 加密進行驗證。這項參數需要 CA 憑證。預設為停用。
`Use SASL PLAIN with TLS for connection`	選填。如果選取這個選項，整合服務會使用 SASL PLAIN 使用者名稱和密碼機制進行驗證。這個選項需要提供 SASL 使用者名稱和密碼。僅支援 TLS 加密，且需要憑證授權單位 (CA) 憑證。預設為停用。
`CA certificate of Kafka server`	選填。用於驗證 Kafka 伺服器身分的 CA 憑證。
`Client certificate`	選填。用戶端憑證，用於與 Kafka 伺服器進行雙向 TLS 驗證。
`Client certificate key`	選填。與用戶端憑證對應的私密金鑰，用於雙向 TLS 驗證。
`Client certificate key password`	選填。用來解密用戶端憑證私密金鑰的密碼。
`SASL PLAIN Username`	選填。用於 Kafka 代理程式的 SASL PLAIN 驗證使用者名稱。
`SASL PLAIN Password`	選填。用於 SASL PLAIN 驗證的密碼，可連線至 Kafka 代理程式。
`Topic`	必填。從中擷取事件的 Kafka 主題。
`Consumer Group ID`	選填。用於擷取事件的消費者群組 ID。如未提供任何值，系統會產生專屬 ID。
`Partitions`	選填。要擷取郵件的分割區 CSV 清單。
`Initial Offset`	選填。連接器開始從 Kafka 分區擷取訊息的位置。您可以指定正整數，從特定位移量開始，也可以使用 `earliest` 或 `latest` 值，從分割區的開頭或結尾開始擷取。
`Poll Timeout`	選填。從 Kafka 取用訊息的輪詢逾時時間 (以秒為單位)。
`Case Name Template`	選填。定義自訂案件名稱的範本。連接器會在事件中新增 `custom_case_name` 鍵。您可以使用格式為 `FIELD_NAME` 的預留位置，系統會從第一個事件的字串值填入這些預留位置。範例：`Phishing - EVENT_MAILBOX`。
`Alert Name Template`	必填。定義快訊名稱的範本。您可以使用格式為 `FIELD_NAME` 的預留位置，系統會從第一個事件的字串值填入這些預留位置。範例：`Phishing - EVENT_MAILBOX`。如果未提供值或範本無效，連接器會使用預設快訊名稱。
`Rule Generator Template`	必填。定義規則產生器的範本。您可以使用格式為 `FIELD_NAME` 的預留位置，系統會從第一個事件的字串值填入這些預留位置。範例：`Phishing - EVENT_MAILBOX`。如果未提供值或範本無效，連接器會使用預設規則產生器名稱。
`Timestamp Field`	必填。 Kafka 訊息中的欄位名稱，包含 Google SecOps 快訊時間戳記。如果時間戳記不是 Unix 紀元格式，則必須在 `Timestamp Format` 參數中定義格式。
`Timestamp Format`	選填。訊息時間戳記的格式，非 Unix 紀元時間戳記必須提供這項資訊。使用標準 Python `strftime` 格式代碼。如果時間戳記不是 Unix 紀元格式，且未設定這個參數，連接器就會失敗。
`Severity Mapping JSON`	必填。連接器使用的 JSON 物件，可從訊息中擷取嚴重程度，並對應至 Google SecOps 優先順序等級。預設值為 `{"Default": "60"}`。
`Unique ID Field`	選填。做為專屬訊息 ID 的欄位名稱。如未提供任何值，連接器會產生並使用訊息內容的 SHA-256 雜湊做為訊息 ID。
`Max Messages To Fetch`	必填。連接器在每次疊代中處理的訊息數量上限。預設值為 `100`。
`Disable Overflow`	選填。如果選取此選項，連接器會忽略 Google SecOps 溢位機制。 (預設為啟用)。
`Verify SSL`	必填。如果選取這個選項，整合服務會在連線至 Apache Kafka 伺服器時驗證 SSL 憑證。 (預設為啟用)。
`Proxy Server Address`	選填。要使用的 Proxy 伺服器位址。
`Proxy Username`	選填。用於驗證的 Proxy 使用者名稱。
`Proxy Password`	選填。用於驗證的 Proxy 密碼。

連接器規則

連接器支援 Proxy。

連接器快訊

下表說明 Apache Kafka 訊息欄位如何對應至 Google SecOps 快訊欄位：

Siemplify 警報欄位	Apache Kafka 訊息欄位
`SourceSystemName`	由架構填入。
`TicketId`	專屬 ID 欄位的值或郵件的 SHA-256 雜湊。
`DisplayId`	`ApacheKafka_{unique id or hash}_{connector identifier}`
`Name`	`Alert Name Template` 產生的值。
`Reason`	不適用
`Description`	不適用
`DeviceVendor`	硬式編碼：`Apache Kafka`
`DeviceProduct`	備用值：`Message`
`Priority`	對應自 `Severity Mapping JSON` 參數。
`RuleGenerator`	`Rule Generator Template` 產生的值。
`SourceGroupingIdentifier`	不適用
`StartTime`	從 `Timestamp Field` 轉換而來。
`EndTime`	從 `Timestamp Field` 轉換而來。
`Siemplify Alert - Extensions`	不適用
`Siemplify Alert - Attachments`	不適用

連接器事件

以下是連接器事件的範例：

{
  "notificationConfigName": "organizations/ORGANIZATION_ID/notificationConfigs/soar_connector_CONNECTOR_ID_toxic_notifications_config",
  "finding": {
    "name": "organizations/ORGANIZATION_ID/sources/SOURCE_ID/findings/FINDING_ID",
    "parent": "organizations/ORGANIZATION_ID/sources/SOURCE_ID",
    "resourceName": "//compute.googleapis.com/projects/PROJECT_ID/global/firewalls/FIREWALL_ID",
    "state": "ACTIVE",
    "category": "OPEN_NETBIOS_PORT",
    "externalUri": "https://console.cloud.google.com/networking/firewalls/details/default-allow-rdp?project=PROJECT_ID",
    "sourceProperties": {
      "Recommendation": "Restrict the firewall rules at: https://console.cloud.google.com/networking/firewalls/details/default-allow-rdp?project=PROJECT_ID",
      "ExceptionInstructions": "Add the security mark \"allow_open_netbios_port\" to the asset with a value of \"true\" to prevent this finding from being activated again.",
      "Explanation": "Firewall rules that allow connections from all IP addresses on TCP ports 137-139 or UDP ports 137-139 may expose NetBIOS services to attackers.",
      "ScannerName": "FIREWALL_SCANNER",
      "ResourcePath": [
        "projects/PROJECT_ID/",
        "folders/FOLDER_ID_1/",
        "folders/FOLDER_ID_2/",
        "organizations/ORGANIZATION_ID/"
      ],
      "ExposedService": "NetBIOS",
      "OpenPorts": {
        "TCP": [
          137.0,
          138.0,
          139.0
        ],
        "UDP": [
          137.0,
          138.0,
          139.0
        ]
      },
      "compliance_standards": {
        "iso": [
          {
            "ids": [
              "A.13.1.1"
            ]
          }
        ],
        "pci": [
          {
            "ids": [
              "1.2.1"
            ]
          }
        ],
        "nist": [
          {
            "ids": [
              "SC-7"
            ]
          }
        ]
      },
      "ReactivationCount": 4.0
    },
    "securityMarks": {
      "name": "organizations/ORGANIZATION_ID/sources/SOURCE_ID/findings/FINDING_ID/securityMarks",
      "marks": {
        "USER_ID": "SECURITY_MARK"
      }
    },
    "eventTime": "2024-08-30T14:44:37.973090Z",
    "createTime": "2024-06-24T07:08:54.777Z",
    "propertyDataTypes": {
      "ResourcePath": {
        "listValues": {
          "propertyDataTypes": [
            {
              "primitiveDataType": "STRING"
            }
          ]
        }
      },
      "ReactivationCount": {
        "primitiveDataType": "NUMBER"
      },
      "Explanation": {
        "primitiveDataType": "STRING"
      },
      "ExposedService": {
        "primitiveDataType": "STRING"
      },
      "ScannerName": {
        "primitiveDataType": "STRING"
      }
    }
  }
}

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。