通過Kafka、S3和OSS匯入日誌到Agentic SOC - Security Center - 阿里雲 - Security Center

工作流程

方案選型

在開始配置前，請根據資料來源類型、成本考量和對資料即時性的要求，選擇最合適的日誌匯入渠道。

匯入渠道	適用情境	功能特點
Kafka	需要即時或近即時日誌串流分析的情境。自建日誌平台通過 Kafka 對外提供資料。	資料即時性高，延遲低。配置和維護相對複雜。
S3 （或相容協議的Object Storage Service）	對資料即時性要求不高（分鐘級）。資料來源支援S3協議。	即時性低於 Kafka，存在分鐘級延遲。架構簡單，易於配置，成本相對可控。
OSS （阿里雲Object Storage Service）	日誌資料已儲存在阿里雲 OSS。	與阿里雲生態整合度高，配置便捷。
API	對資料即時性要求不高（分鐘級）。適用於對接不支援推送機制的第三方 SaaS 服務、雲廠商監控資料或遺留系統，無法直接寫入訊息佇列或Object Storage Service的情境。	即時性低於 Kafka，存在分鐘級延遲。拉取的速度會受限於三方服務的 API 限流規則。

投遞日誌到中間渠道

Agentic SOC匯入功能需通過 Kafka、S3 、OSS、三方服務API等渠道擷取資料。因此需要先將源端的日誌資料投遞到所選的中間渠道。

投遞方式樣本：
- OSS：將日誌資料存放區至阿里雲 OSS 儲存空間（Bucket）中。更多說明，請參見OSS快速入門。
- 華為云：通過華為雲Log Service（LTS）的日誌轉儲功能，將資料轉存至Kafka或OBS。更多說明，請參見匯入華為雲日誌資料。
- 騰訊云：通過騰訊雲Log Service （CLS）的日誌轉儲功能，將資料轉存至Kafka或COS。更多說明，請參見匯入騰訊雲日誌資料。
- Azure：利用Azure Event Hub（事件中心）對Apache Kafka協議的相容性，將事件中心視為一個Kafka服務，Agentic SOC從中擷取日誌資料。更多說明，請參見匯入Azure日誌資料。
儲存訪問資訊：請妥善保管訪問憑證（如 AccessKey、密碼）和存取點資訊（如 Endpoint、Bucket名稱）。

授權Security Center訪問三方服務

使用三方資料渠道，需授權Agentic SOC訪問其資源的許可權，以允許 Agentic SOC 讀取其中的日誌資料。

Kafka、S3（Object Storage Service）、API

訪問Security Center控制台-系統設定-功能設定，在頁面左側頂部，選擇需防護資產所在的地區：中國內地或非中國內地。
在多云配置管理頁簽，選擇多云资产後，單擊新增授权，下拉選擇IDC。在彈出的面板中進行如下配置：
說明
- 具體的配置項的數值（如訪問地址等）可參考各雲廠商的官方說明文檔。
- 目前的版本中，配置通用的 Kafka 、 S3 相容儲存（如 AWS S3、騰訊雲 COS、華為雲 OBS）或 API 時，請選擇 IDC 選項。此為介面歸類方式，不影響實際功能。
Kafka
- 廠商：選擇 Apache。
- 產品：選擇 Kafka。
- 存取點：Kafka公網訪問地址。
  
  說明
  Event Hub訪問地址為：<YOUR-NAMESPACE>.servicebus.windows.net:9093。
- 使用者名稱/密碼：Kafka的使用者名稱/密碼。
  
  說明
  Event Hub對應Kafka使用者名稱預設為$ConnectionString，密碼為主連接字串。
- 通訊協定：Kafka配置中的security.protocol，支援設定plaintext（預設）、sasl_plaintext、sasl_ssl、ssl等。
- SASL 認證機制：Kafka 配置中的 sasl.mechanism，支援設定：plain、SCRAM-SHA-256、SCRAM-SHA-512等。
S3
- 廠商：選擇 AWS-S3。
- 產品：選擇S3 。
- 存取點：Object Storage Service的訪問地址。
- Access Key Id/Secret Access Key：輸入訪問S3的存取金鑰。
API
- 廠商：選擇 Salesforce / Lark / Akamai 等。
- 產品：選擇廠商已經整合的 API 。
  - Salesforce API ：Salesforce Rest API 。
  - Lark API ：Lark Rest API 。
  - Akamai API ：Akamai Rest API 。
- 存取點：輸入廠商 API 服務提供者的訪問地址。
  - Salesforce 的存取點擷取：登入 Salesforce ，點擊右上方 Setup Menu ，在左側 Quick Find 中搜尋 My Domain ，點擊搜尋結果中的My Domain 查看詳情，Current My Domain URL 就是存取點資訊。
  - Lark 的存取點擷取：固定值 https://open.larksuite.com 就是存取點資訊。
  - Akamai 的存取點擷取：登入 Akamai 控制台，在左上方功能表列搜尋 Identity & Access ，在右側搜尋結果頁中點擊進入 Identity & Access ，尋找 API 介面對應的應用賬戶資訊，點擊進入使用者詳情頁面，Host 就是存取點資訊。
- Access Key Id/Secret Access Key：輸入調用 API 的存取金鑰。
  - Salesforce 的 Access Key Id/Secret Access Key 擷取：登入 Salesforce ，點擊右上方 Setup Menu ，在左側 Quick Find 中搜尋 App Manager ，點擊搜尋結果中的 App Manager 查看 App 列表，從右側列表中尋找 API 介面對應的應用，應用 App Type 類型一般為Connected ，點擊對應應用右側的按鈕展開下拉框，點擊 View 查看詳情資訊，進入後尋找 Consumer Key（對應 Access Key Id ）和 Consumer Secret（對應 Secret Access Key ）資訊。
  - Lark 的 Access Key Id/Secret Access Key 擷取：登入 Lark 開發人員 APP 後台，找到 API 介面對應的應用，點擊進入應用詳情，點擊左側基礎資訊菜單中的憑據與基礎資訊菜單，在詳情頁尋找 App ID（對應 Access Key Id ）和 App Secret（對應 Secret Access Key ）資訊，注意 App ID 需要有調用 API 許可權，可能要聯絡 Lark 的客戶經理申請。
  - Akamai 的 Access Key Id/Secret Access Key 擷取：登入 Akamai 控制台，在左上方功能表列搜尋 Identity & Access ，在右側搜尋結果頁中點擊進入 Identity & Access ，尋找 API 介面對應的應用賬戶資訊，點擊進去使用者詳情頁面，找 Client Token（對應 Access Key Id ）和 Client Secret（對應 Secret Access Key ）資訊，注意 Client Secret 僅在 Create credential 時才會顯示，需要在建立完後記錄下來。
- Access Token ：輸入調用 API 的 Access Token，目前僅選擇廠商 Akamai 時才會需要輸入。
  - Akamai 的 Access Token 擷取：登入 Akamai 控制台，在左上方功能表列搜尋 Identity & Access ，在搜尋結果頁中點擊進入 Identity & Access ，尋找 API 介面的應用賬戶資訊，點擊進入使用者詳情頁面，尋找 Access Token 資訊。
- 裝置名稱：標識裝置的唯一性的名稱。
配置同步策略

AK服務狀態檢查：設定Security Center自動檢查存取金鑰有效性的時間間隔。可選“關閉”，表示不進行檢測。

OSS

訪問雲資源訪問授權頁面，單擊確認授權，授予系統角色訪問OSS資源的許可權。

若為RAM使用者，還需配置以下許可權。具體操作，請參見建立自訂權限原則和管理RAM使用者的許可權。

{
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["ram:PassRole", "ram:GetRole"],
      "Resource": "acs:ram:*:*:role/aliyunlogimportossrole"
    },
    {
      "Effect": "Allow",
      "Action": "oss:GetBucketWebsite",
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": "oss:ListBuckets",
      "Resource": "*"
    }
  ],
  "Version": "1"
}

建立資料匯入任務

建立資料來源

為日誌資料建立一個專屬Agentic SOC 資料來源，若已建立請跳過此步驟。
1. 訪問Security Center控制台-Agentic SOC-管理-接入設定，在頁面左側頂部，選擇需防護資產所在的地區：中國內地或非中國內地。
2. 在資料來源頁簽，建立接收日誌的資料來源。具體操作請參見建立資料來源：日誌未接入Log Service（SLS）。
  - 來源資料源類型：可選擇CTDR專屬資料擷取通道（推薦）或使用者側Log Service。
  - 接入執行個體：建議建立日誌庫（Logstore），進行資料隔離。

在數據匯入頁簽，單擊新增數據匯入。在彈出的面板中根據資料來源類型進行如下配置：

說明

具體的配置項的數值可參考各雲廠商的官方說明文檔。

Kafka

終端節點：選擇授權Security Center訪問三方服務時填寫的Kafka公網訪問地址。
Topics：選擇Kafka中儲存日誌的Topic。

說明
Event Hub名稱即為Kafka的topic。

實值型別：日誌儲存格式，常見的關係如下：

三方儲存格式	實值型別
JSON格式	json
原始日誌格式	text

S3

終端節點：S3儲存桶訪問地址（Endpoint）。
OBS桶（Bucket）：S3儲存桶ID。
檔案路徑首碼過濾：通過檔案路徑首碼過濾S3檔案，用於準確定位待匯入的檔案。例如待匯入的檔案都在csv/目錄下，則可以指定首碼為csv/。

警告
強烈建議設定此參數。如果不設定，系統將遍曆整個 S3 Bucket。當 Bucket 內檔案數量巨大時，全量遍曆會嚴重影響匯入效率。
檔案路徑正則過濾：通過Regex篩選檔案，用於準確定位待匯入的檔案。預設為空白，表示不過濾。例如S3檔案為testdata/csv/bill.csv，可以設定Regex為(testdata/csv/)(.*)。
說明
- 推薦與檔案路徑首碼過濾一起設定，提高效率。此配置與檔案路徑首碼過濾之前為“且（and）”的關係。
- 調整Regex的方法，請參見如何調試Regex。

資料格式：檔案的解析格式。

CSV：分隔字元分割的文字檔，支援指定檔案中的首行為欄位名稱或手動指定欄位名稱。除欄位名稱外的每一行都會被解析為日誌欄位的值。
JSON：逐行讀取S3檔案，將每一行看作一個JSON對象進行解析。解析後，JSON對象中的各個欄位對應為日誌中的各個欄位。
單行文本：將S3檔案中的每一行解析為一條日誌。
跨行文本：多行模式，支援指定首行或者尾行的Regex解析日誌。

若選擇為CSV或跨行文本時，需額外設定相關參數，具體說明如下。

CSV

參數	說明
分隔字元	設定日誌的分隔字元，預設值為半形逗號（,）。
引號	CSV字串所使用的引號字元。
轉義符	配置日誌的轉義符，預設值為反斜線（\）。
日誌最大跨行數	當一條日誌跨多行時，需要指定最大行數，預設值為1。
首行作為欄位名稱	開啟開關後，將使用CSV檔案中的首行作為欄位名稱。例如提取以下CSV資料中的首行為日誌欄位的名稱。 `remote_addr,remote_user,time_local,request_time,request_length xxx5,-,11/Dec/2020:15:31:06,0,000,133,3650,404,GET xxx5,-,11/Dec/2020:15:32:06,0,000,133,3650,404,GET xxx5,-,11/Dec/2020:15:34:10,0,000,133,3650,404,GET`
跳過行數	指定跳過的日誌行數。例如設定為1，則表示從CSV檔案中的第2行開始採集日誌。

跨行文本

參數	說明
正則匹配位置	設定Regex匹配的位置，具體說明如下：首行正則：使用Regex匹配一條日誌的行首，未匹配部分為該條日誌的一部分，直到達到最大行數。尾行正則：使用Regex匹配一條日誌的行尾，未匹配部分為下一條日誌的一部分，直到達到最大行數。
Regex	根據日誌內容，設定正確的Regex。調整Regex的方法，請參見如何調試Regex。
最大行數	一條日誌最大的行數。

編碼格式：待匯入的S3檔案的編碼格式。支援GBK、UTF-8。
壓縮格式：根據設定S3資料壓縮格式，由系統自動檢測。
檔案修改時間過濾：從任務啟動時刻回溯30分鐘作為起始時間，匯入此後修改的所有檔案（含未來新增檔案）。
檢查新檔案周期：用於設定匯入任務定期掃描新檔案的時間間隔，任務將按此時間間隔自動掃描並匯入新增的檔案。

OSS

重要

目前僅支援匯入5 GB以內的OSS檔案，壓縮檔大小按照壓縮後的大小計算。

OSS地區：OSS資料存放區地區。

說明
跨地區訪問會產生公網流量費用，此部分費用由OSS服務收取，具體說明請參見費用說明。
Bucket：待匯入的OSS檔案所在的Bucket。
檔案路徑首碼過濾：通過Regex篩選檔案，用於準確定位待匯入的檔案。例如待匯入的檔案都在csv/目錄下，則可以指定首碼為csv/。

警告
建議設定此參數，如果不設定，系統遍曆整個OSS Bucket。若當檔案數量巨大時，遍曆整個Bucket會影響匯入效率併產生不必要的費用。
檔案路徑正則過濾：通過Regex篩選檔案，用於準確定位待匯入的檔案。預設為空白，表示不過濾。例如OSS檔案為testdata/csv/bill.csv，您可以設定Regex為(testdata/csv/)(.*)。
說明
- 推薦與檔案路徑首碼過濾一起設定，提高效率。此配置與檔案路徑首碼過濾之前為“且（and）”的關係。
- 調整Regex的方法，請參見如何調試Regex。
檔案修改時間過濾：從任務啟動時刻回溯30分鐘作為起始時間，匯入此後修改的所有檔案（含未來新增檔案）。

資料格式：檔案的解析格式。

CSV：分隔字元分割的文字檔，支援指定檔案中的首行為欄位名稱或手動指定欄位名稱。除欄位名稱外的每一行都會被解析為日誌欄位的值。
單行JSON：逐行讀取OSS檔案，將每一行看做一個JSON對象進行解析。解析後，JSON對象中的各個欄位對應為日誌中的各個欄位。
單行文本：將OSS檔案中的每一行解析為一條日誌。
跨行文本：多行模式，支援指定首行或者尾行的Regex解析日誌。
ORC：ORC檔案格式，無需任何配置，自動解析成日誌格式。
Parquet：Parquet格式，無需任何配置，自動解析成日誌格式。
阿里雲OSS訪問日誌：阿里雲OSS訪問日誌格式。更多資訊，請參見日誌轉存。
阿里雲CDN下載日誌：阿里雲CDN下載日誌格式。更多資訊，請參見快速入門。

若選擇為CSV或跨行文本時，需額外設定相關參數，具體說明如下。

CSV

參數	說明
分隔字元	設定日誌的分隔字元，預設值為半形逗號（,）。
引號	CSV字串所使用的引號字元。
轉義符	配置日誌的轉義符，預設值為反斜線（\）。
日誌最大跨行數	當一條日誌跨多行時，需要指定最大行數，預設值為1。
首行作為欄位名稱	開啟開關後，將使用CSV檔案中的首行作為欄位名稱。例如提取以下CSV資料中的首行為日誌欄位的名稱。 `remote_addr,remote_user,time_local,request_time,request_length xxx5,-,11/Dec/2020:15:31:06,0,000,133,3650,404,GET xxx5,-,11/Dec/2020:15:32:06,0,000,133,3650,404,GET xxx5,-,11/Dec/2020:15:34:10,0,000,133,3650,404,GET`
跳過行數	指定跳過的日誌行數。例如設定為1，則表示從CSV檔案中的第2行開始採集日誌。

跨行文本

參數	說明
正則匹配位置	設定Regex匹配的位置，具體說明如下：首行正則：使用Regex匹配一條日誌的行首，未匹配部分為該條日誌的一部分，直到達到最大行數。尾行正則：使用Regex匹配一條日誌的行尾，未匹配部分為下一條日誌的一部分，直到達到最大行數。
Regex	根據日誌內容，設定正確的Regex。調整Regex的方法，請參見如何調試Regex。
最大行數	一條日誌最大的行數。

編碼格式：待匯入的OSS檔案的編碼格式。支援GBK、UTF-8。
壓縮格式：根據設定OSS資料壓縮格式，由系統自動檢測。
檢查新檔案周期：用於設定匯入任務定期掃描新檔案的時間間隔，任務將按此時間間隔自動掃描並匯入新增的檔案。

API

產品：選擇一個產品的RestAPI。
客戶端標識符：選擇該產品對應的用戶端標識符，如果下拉式清單沒有可選的用戶端標識符，需要點擊右側 “去配置多雲資產授權” 連結去管理用戶端標識符。
採集劇本模版：選擇該產品對應的採集劇本模版。
劇本名稱：填寫劇本名稱。
劇本參數設定：填寫劇本參數。（僅部分劇本才需要設定）
匯入頻率：填寫日誌匯入周期，用於設定匯入任務定期通過API查詢新增日誌的時間間隔，任務將按此時間間隔自動查詢並拉取新增的日誌。

配置目標資料來源
- 資料來源名稱：選擇步驟1建立的資料來源。
- 目標Logstore：根據步驟1設定的日誌庫（Logstore）。
單擊確定儲存配置：匯入配置完成後，Security Center將自動從資料匯入渠道拉取日誌。

分析匯入資料

資料成功接入後，需配置相應的解析和檢測規則，才能讓Security Center對這些日誌進行分析。

建立新的接入策略

參考產品接入，建立新的接入策略，配置如下：
- 資料來源：選擇資料匯入任務中配置的目標資料來源。
- 標準化規則：Agentic SOC提供了適配部分雲產品的內建標準化規則，可直接選用。
- 標準化方式：當接入日誌標準化為警示日誌時，僅支援“即時消費”方式標準化。
配置威脅檢測規則

根據安全需求，在規則管理中啟用或建立日誌檢測規則，才能對日誌進行分析、產出警示並產生安全事件。具體操作，請參見檢測規則。

費用與成本

Agentic SOC 與 SLS 計費：費用的承擔主體取決於選擇的資料存放區方式。

說明

Agentic SOC計費說明，請參見Agentic SOC訂用帳戶、Agentic SOC隨用隨付。

Log Service（SLS）計費說明，請參見計費概述。

資料來源類型

Agentic SOC 計費項目

SLS 計費項目

特別說明

CTDR專屬資料擷取通道

日誌接入費用。
日誌儲存、寫入費用。

說明

以上均消耗日志接入流量。

除日誌儲存和寫入外的其他費用（如公網流量等）。

由Agentic SOC 建立並管理 SLS 資源，因此日誌庫儲存和寫入費用由Agentic SOC出賬。

使用者側Log Service

日誌接入費用，消耗日志接入流量。

日誌相關所有費用（包含日誌儲存和寫入、公網流量費用等）。

日誌資源全部由Log Service（SLS）管理，因此日誌相關費用，均由SLS出賬。

OSS相關費用：從 OSS 匯入資料時，OSS需收取流量費用和請求費用，關計費項目的定價詳情，請參見OSS定價。

成本計算公式

欄位	說明
`N`	每天匯入的檔案個數。
`T`	每天匯入的總資料量，單位：GB。
`p_read`	每GB資料的流量費用。同地區匯入時，會產生OSS的內網流出流量，該流量免費。跨地區匯入時，會產生OSS的外網流出流量。
`p_put`	每萬次的Put類型請求費用。 Log Service使用ListObjects介面擷取目標Bucket中的檔案清單。該介面在OSS側按照Put類型請求收費。另外，該介面每次最多返回1000條資料，因此如果您有100萬個新增檔案，需要進行1,000,000/1000=1000次請求。
`p_get`	每萬次的GET類型請求費用。
`M`	新檔案檢查周期，單位：分鐘。您可以在建立資料匯入配置時，設定檢查新檔案周期參數。

計費案例

三方費用：
- 騰訊云：騰訊雲Log Service-計費概述、騰訊雲COS-計費概述。
- 華為云：華為雲Log Service-計費概述、華為雲Kafka-計費概述、華為雲OBS-計費概述。
- Azure：事件中心定價。