收集 Akamai Cloud Monitor 記錄

剖析器版本：2.0

支援的國家/地區：

本文說明如何使用 Google Cloud Storage，將 Akamai Cloud Monitor 記錄檔擷取至 Google Security Operations。Cloud Function 會接收來自 Akamai Cloud Monitor 的 HTTPS POST 要求，並將 JSON 事件寫入 GCS，做為經過 gzip 壓縮的 JSONL 檔案。Akamai Cloud Monitor 是一種即時流量監控工具，適用於 Akamai 加速的網路資源，可擷取 HTTP 要求和回應中繼資料，包括用戶端 IP、要求路徑、狀態碼、快取狀態和 TLS 詳細資料。

事前準備

請確認您已完成下列事前準備事項：

  • Google SecOps 執行個體
  • 已啟用 Cloud Storage API 的 Google Cloud 專案
  • 建立及管理 GCS bucket 的權限
  • 管理 Google Cloud Storage 值區 IAM 政策的權限
  • 建立 Cloud Functions 的權限
  • Akamai Control Center 和 Property Manager 的特殊存取權

建立 Google Cloud Storage bucket

  1. 前往 Google Cloud 控制台。
  2. 選取專案或建立新專案。
  3. 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
  4. 按一下「建立值區」。

  5. 請提供下列設定詳細資料：

    設定 值
    為 bucket 命名 輸入全域不重複的名稱 (例如 akamai-cloud-monitor)
    位置類型 根據需求選擇 (區域、雙區域、多區域)
    位置 選取位置 (例如 us-central1)
    儲存空間級別 標準 (建議用於經常存取的記錄)
    存取控管 統一 (建議)
    保護工具 選用：啟用物件版本管理或保留政策

  6. 點選「建立」。

建立 Cloud Functions 服務帳戶

Cloud Function 需要具備 GCS bucket 寫入權限的服務帳戶。

建立服務帳戶

  1. 在 Google Cloud 控制台中，依序前往「IAM 與管理」>「服務帳戶」。
  2. 按一下「Create Service Account」(建立服務帳戶)。
  3. 請提供下列設定詳細資料：
    • 服務帳戶名稱：輸入 akamai-cloud-monitor-sa。
    • 服務帳戶說明：輸入 Service account for Cloud Function to collect Akamai Cloud Monitor logs。
  4. 按一下「建立並繼續」。
  5. 在「將專案存取權授予這個服務帳戶」部分：
    1. 按一下「選擇角色」。
    2. 搜尋並選取「Storage 物件管理員」。
    3. 點選「+ 新增其他角色」。
    4. 搜尋並選取「Cloud Run Invoker」。
    5. 點選「+ 新增其他角色」。
    6. 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
  6. 按一下「繼續」。
  7. 按一下 [完成]。

這些角色適用於：

  • Storage 物件管理員：將記錄檔寫入 GCS bucket，並管理狀態檔案
  • Cloud Run 叫用者：允許透過 HTTPS 叫用函式
  • Cloud Functions 叫用者：允許函式叫用

授予 GCS 值區的 IAM 權限

將 GCS bucket 的寫入權限授予服務帳戶：

  1. 依序前往「Cloud Storage」>「Buckets」。
  2. 按一下 bucket 名稱。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」。
  5. 請提供下列設定詳細資料：
    • 新增主體：輸入服務帳戶電子郵件地址 (例如 akamai-cloud-monitor-sa@PROJECT_ID.iam.gserviceaccount.com)。
    • 指派角色：選取「Storage 物件管理員」。
  6. 按一下 [儲存]。

建立 Cloud 函式來接收 Akamai 記錄

Cloud Function 會接收 Akamai Cloud Monitor 的 HTTP POST 要求，並將記錄寫入 GCS。

  1. 在 Google Cloud 控制台中，前往「Cloud Functions」。
  2. 按一下「Create function」(建立函式)。

  3. 請提供下列設定詳細資料：

    設定 值
    環境 選取「第 2 代」
    函式名稱 akamai-cloud-monitor-receiver
    區域 選取與 GCS bucket 相符的區域 (例如 us-central1)

  4. 在「Trigger」(觸發條件) 專區：

    • 觸發條件類型：選取「HTTPS」。
    • 驗證：選取「允許未經驗證的叫用」 (Akamai 會傳送未經驗證的要求)。

  5. 按一下「儲存」，儲存觸發條件設定。

  6. 展開「執行階段、建構作業、連線和安全性設定」。

  7. 在「執行階段」部分：

    • 分配的記憶體：選取「512 MiB」。
    • 逾時：輸入 600 秒 (10 分鐘)。
    • 執行階段服務帳戶：選取服務帳戶 (akamai-cloud-monitor-sa)。

  8. 在「執行階段環境變數」部分，針對下列各項點按「+ 新增變數」：

    變數名稱 範例值
    GCS_BUCKET akamai-cloud-monitor
    GCS_PREFIX akamai/cloud-monitor/json
    INGEST_TOKEN random-shared-secret (選填)

  9. 按一下「下一步」前往程式碼編輯器。

  10. 在「執行階段」下拉式選單中，選取「Python 3.12」。

新增函式程式碼

  1. 在「函式進入點」中輸入 main。

  2. 在內嵌程式碼編輯器中，建立兩個檔案：

    • main.py:
    import os
import json
import gzip
import io
import uuid
import datetime as dt
from google.cloud import storage
import functions_framework

GCS_BUCKET = os.environ.get("GCS_BUCKET")
GCS_PREFIX = os.environ.get("GCS_PREFIX", "akamai/cloud-monitor/json").strip("/") + "/"
INGEST_TOKEN = os.environ.get("INGEST_TOKEN")  # optional shared secret

storage_client = storage.Client()

def _write_jsonl_gz(objs: list) -> str:
    """Write JSON objects to GCS as gzipped JSONL."""
    timestamp = dt.datetime.utcnow()
    key = f"{timestamp:%Y/%m/%d}/akamai-cloud-monitor-{uuid.uuid4()}.json.gz"

    buf = io.BytesIO()
    with gzip.GzipFile(fileobj=buf, mode="w") as gz:
        for o in objs:
            gz.write((json.dumps(o, separators=(",", ":")) + "\n").encode())
    buf.seek(0)

    bucket = storage_client.bucket(GCS_BUCKET)
    blob = bucket.blob(f"{GCS_PREFIX}{key}")
    blob.upload_from_file(buf, content_type="application/json", content_encoding="gzip")

    return f"gs://{GCS_BUCKET}/{GCS_PREFIX}{key}"

def _parse_records_from_request(request) -> list:
    """Parse JSON records from HTTP request body."""
    body = request.get_data(as_text=True)

    if not body:
        return []

    try:
        data = json.loads(body)
    except Exception:
        # Accept line-delimited JSON as pass-through
        try:
            return [json.loads(line) for line in body.splitlines() if line.strip()]
        except Exception:
            return []

    if isinstance(data, list):
        return data
    if isinstance(data, dict):
        return [data]
    return []

@functions_framework.http
def main(request):
    """
    Cloud Function HTTP handler for Akamai Cloud Monitor logs.

    Args:
        request: Flask request object

    Returns:
        Tuple of (response_body, status_code, headers)
    """
    # Optional shared-secret verification via query parameter (?token=...)
    if INGEST_TOKEN:
        token = request.args.get("token")
        if token != INGEST_TOKEN:
            return ("Forbidden", 403)

    records = _parse_records_from_request(request)

    if not records:
        return ("No content", 204)

    try:
        gcs_key = _write_jsonl_gz(records)

        response = {
            "ok": True,
            "gcs_key": gcs_key,
            "count": len(records)
        }

        return (json.dumps(response), 200, {"Content-Type": "application/json"})

    except Exception as e:
        print(f"Error writing to GCS: {str(e)}")
        return (f"Internal server error: {str(e)}", 500)
    • requirements.txt:
    functions-framework==3.*
google-cloud-storage==2.*

  3. 按一下「Deploy」(部署) 即可部署函式。

  4. 等待部署作業完成 (2 到 3 分鐘)。

  5. 部署完成後，前往「觸發條件」分頁，然後複製「觸發條件網址」。您將在 Akamai 設定中使用此網址。

設定 Akamai Cloud Monitor 推送記錄

在 Akamai Control Center 中設定兩個 Property Manager 行為：「Cloud Monitor Instrumentation」，用於選取要收集的資料集；以及「Cloud Monitor Data Delivery」，用於定義接收資料的 HTTPS 端點。

  1. 登入 Akamai Control Center。
  2. 依序前往「CDN」>「Properties」。
  3. 選取要監控的資源。
  4. 按一下「編輯新版本」，即可建立新的可編輯屬性版本。
  5. 在「資源管理工具編輯器」中，按一下「新增規則」，然後選取「雲端管理」。

新增 Cloud Monitor Instrumentation 行為

  1. 在 Cloud Management 規則中，按一下「Add Behavior」(新增行為)，然後選取「Cloud Monitor Instrumentation」(Cloud Monitor 檢測)。

  2. 請提供下列設定詳細資料：

    • 已啟用：選取「開啟」。
    • 收集匯總資料：選取要收集的資料集。可用的資料集包括要求標頭、回應標頭、Cookie 和自訂欄位。

新增 Cloud Monitor Data Delivery 行為

  1. 在同一個 Cloud Management 規則中，按一下「Add Behavior」(新增行為)，然後選取「Cloud Monitor Data Delivery」(雲端監控資料傳送)。

  2. 請提供下列設定詳細資料：

    • 已啟用：選取「開啟」。
    • 傳送主機名稱：輸入 Cloud 函式觸發網址的主機名稱 (例如 us-central1-your-project.cloudfunctions.net)。

    • 放送網址路徑：輸入 Cloud 函式觸發網址的路徑，以及選用的查詢權杖：

      • 不含權杖：/akamai-cloud-monitor-receiver
      • 使用權杖：/akamai-cloud-monitor-receiver?token=<INGEST_TOKEN>

      將 <INGEST_TOKEN> 替換為您在 Cloud Functions 環境變數中設定的值。

  3. 按一下 [儲存]。

  4. 按一下「啟用」，先在「測試」聯播網啟用資源版本，驗證設定，然後在「正式」聯播網啟用。

擷取 Google SecOps 服務帳戶

Google SecOps 會使用專屬服務帳戶，從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「新增動態消息」。
  3. 按一下「設定單一動態饋給」。
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Akamai Cloud Monitor - GCS)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」。
  6. 選取「Akamai Cloud Monitor」做為「記錄類型」。

  7. 按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址，例如：

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  8. 複製這個電子郵件地址，以便在下一步中使用。

將 IAM 權限授予 Google SecOps 服務帳戶

Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。

  1. 依序前往「Cloud Storage」>「Buckets」。
  2. 按一下 bucket 名稱。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」。
  5. 請提供下列設定詳細資料：
    • 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
    • 指派角色：選取「Storage 物件檢視者」。

  6. 按一下 [儲存]。

在 Google SecOps 中設定資訊提供，擷取 Akamai Cloud Monitor 記錄

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「新增動態消息」。
  3. 按一下「設定單一動態饋給」。
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Akamai Cloud Monitor - GCS)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」。
  6. 選取「Akamai Cloud Monitor」做為「記錄類型」。
  7. 點選「下一步」。

  8. 指定下列輸入參數的值：

    • 儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：
    gs://akamai-cloud-monitor/akamai/cloud-monitor/json/
    • 取代：
      • akamai-cloud-monitor：您的 GCS bucket 名稱。
      • akamai/cloud-monitor/json：儲存記錄的前置路徑 (必須與 Cloud Function 中的 GCS_PREFIX 相符)。
    • 來源刪除選項：根據偏好選取刪除選項：
      • 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
      • 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
      • 刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。
    • 檔案存在時間上限：納入在過去天數內修改的檔案。預設值為 180 天。
    • 資產命名空間：akamai.cloud_monitor
    • 擷取標籤：系統會為這個動態饋給中的所有事件新增標籤 (例如 source=akamai_cloud_monitor、format=json)。

  9. 點選「下一步」。

  10. 在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

支援的 Akamai Cloud Monitor 記錄檔範例

  • JSON：

    {
"UA": "-",
"accLang": "-",
"bytes": "3929",
"cacheStatus": "1",
"cliIP": "0.0.0.0",
"cookie": "-",
"cp": "848064",
"customField": "-",
"dnsLookupTimeMSec": "-",
"errorCode": "-",
"maxAgeSec": "31536000",
"objSize": "3929",
"overheadBytes": "240",
"proto": "HTTPS",
"queryStr": "-",
"range": "-",
"referer": "-",
"reqEndTimeMSec": "4",
"reqHost": "www.example.com",
"reqId": "1ce83c03",
"reqMethod": "GET",
"reqPath": "assets/images/placeholder-tagline.png",
"reqPort": "443",
"reqTimeSec": "1622470405.760",
"rspContentLen": "3929",
"rspContentType": "image/png",
"statusCode": "200",
"tlsOverheadTimeMSec": "0",
"tlsVersion": "TLSv1.2",
"totalBytes": "4599",
"transferTimeMSec": "0",
"turnAroundTimeMSec": "0",
"uncompressedSize": "-",
"version": "1",
"xForwardedFor": "-"
}

UDM 對應表

記錄欄位 UDM 對應 邏輯
accLang network.http.user_agent 如果不是「-」或空字串，則直接對應。
城市 principal.location.city 如果不是「-」或空字串，則直接對應。
cliIP principal.ip 如果不是空白字串，則直接對應。
國家/地區 principal.location.country_or_region 如果不是「-」或空字串，則直接對應。
cp additional.fields 以鍵/值組合形式對應，鍵為「cp」。
customField about.ip、about.labels、src.ip 剖析為鍵/值組合。特別處理「eIp」和「pIp」，分別對應至 src.ip 和 about.ip。其他鍵會對應為 about 中的標籤。
errorCode security_result.summary、security_result.severity 如果存在，則將 security_result.severity 設為「ERROR」，並將值對應至 security_result.summary。
geo.city principal.location.city 如果城市為「-」或空字串，則直接對應。
geo.country principal.location.country_or_region 如果國家/地區為「-」或空字串，則直接對應。
geo.lat principal.location.region_latitude 直接對應，並轉換為浮點數。
geo.long principal.location.region_longitude 直接對應，並轉換為浮點數。
geo.region principal.location.state 直接對應。
id metadata.product_log_id 如果不是空白字串，則直接對應。
message.cliIP principal.ip 如果 cliIP 是空字串，則直接對應。
message.fwdHost principal.hostname 直接對應。
message.reqHost target.hostname、target.url 用於建構 target.url 並擷取 target.hostname。
message.reqLen network.sent_bytes 直接對應，如果 totalBytes 為空或「-」，則會轉換為不帶正負號的整數。
message.reqMethod network.http.method 如果 reqMethod 是空字串，則直接對應。
message.reqPath target.url 附加至 target.url。
message.reqPort target.port 直接對應，如果 reqPort 是空字串，則轉換為整數。
message.respLen network.received_bytes 直接對應，轉換為不帶正負號的整數。
message.sslVer network.tls.version 直接對應。
message.status network.http.response_code 直接對應，如果 statusCode 為空或「-」，則轉換為整數。
message.UA network.http.user_agent 如果 UA 為「-」或空字串，則直接對應。
network.asnum additional.fields 以鍵/值組合形式對應，鍵為「asnum」。
network.edgeIP intermediary.ip 直接對應。
network.network additional.fields 以鍵/值組合形式對應，索引鍵為「network」。
network.networkType additional.fields 以鍵/值組合形式對應，鍵為「networkType」。
proto network.application_protocol 用於判斷 network.application_protocol。
queryStr target.url 如果不是「-」或空字串，則會附加至 target.url。
參照網址 network.http.referral_url、about.hostname 如果不是「-」，則直接對應。擷取的主機名稱會對應至 about.hostname。
reqHost target.hostname、target.url 用於建構 target.url 並擷取 target.hostname。
reqId metadata.product_log_id、network.session_id 如果 ID 為空字串，則直接對應。也對應至 network.session_id。
reqMethod network.http.method 如果不是空白字串，則直接對應。
reqPath target.url 如果不是「-」，則附加至 target.url。
reqPort target.port 直接對應，並轉換為整數。
reqTimeSec metadata.event_timestamp、timestamp 用於設定事件時間戳記。
開始 metadata.event_timestamp、timestamp 如果 reqTimeSec 是空白字串，可用於設定事件時間戳記。
statusCode network.http.response_code 直接對應，如果不是「-」或空字串，則轉換為整數。
tlsVersion network.tls.version 直接對應。
totalBytes network.sent_bytes 直接對應，如果不是空白或「-」，則轉換為不帶正負號的整數。
類型 metadata.product_event_type 直接對應。
UA network.http.user_agent 如果不是「-」或空字串，則直接對應。
版本 metadata.product_version 直接對應。
xForwardedFor principal.ip 如果不是「-」或空字串，則直接對應。
(剖析器邏輯) metadata.vendor_name 設為「Akamai」。
(剖析器邏輯) metadata.product_name 設為「Cloud Monitor」。
(剖析器邏輯) metadata.event_type 設為「NETWORK_HTTP」。
(剖析器邏輯) metadata.product_version 如果版本是空字串，請設為「2」。
(剖析器邏輯) metadata.log_type 設為「AKAMAI_CLOUD_MONITOR」。
(剖析器邏輯) network.application_protocol 從 proto 或 message.proto 判斷。如果兩者皆包含「HTTPS」(不區分大小寫)，則設為「HTTPS」，否則設為「HTTP」。
(剖析器邏輯) security_result.severity 如果 errorCode 為「-」或空字串，請設為「INFORMATIONAL」。
(剖析器邏輯) target.url 由通訊協定、reqHost (或 message.reqHost)、reqPath (或 message.reqPath) 和 queryStr 建構而成。

變更記錄

查看這個剖析器的變更記錄

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