收集 Akamai SIEM 連接器記錄
剖析器版本：2.0
支援的國家/地區：
Google SecOps
SIEM
本指南說明如何將 Akamai SIEM Connector 記錄檔擷取至 Google Security Operations。Akamai SIEM 整合功能會透過 SIEM 整合 API，以 JSON 格式近乎即時地提供 Akamai 平台中的安全事件。這些事件包括網頁應用程式防火牆 (WAF)、Bot Manager 和其他 Akamai 安全性產品的資料。由於 Akamai SIEM Integration API 是以提取為基礎，因此排程函式會定期擷取事件，並將其寫入物件儲存空間，供 Google SecOps 擷取。您可以透過下列任一支援方法設定擷取作業：
- 方法 1：Cloud Storage。Cloud Run 函式會定期從 Akamai SIEM Integration API 提取事件，並將事件寫入 Cloud Storage bucket，供 Google SecOps 擷取。
- 方法 2：Amazon S3。AWS Lambda 函式會定期從 Akamai SIEM Integration API 提取事件，並將事件寫入 Amazon S3 bucket，供 Google SecOps 擷取。
事前準備
請確認您已完成下列事前準備事項：
兩種方法皆適用：
- Google SecOps 執行個體。
- 具備「管理 SIEM」使用者角色，可享有 Akamai Control Center 的特殊存取權。
方法 1 的額外必要條件 (Cloud Storage)：
- 已啟用 Cloud Storage、Cloud Run、Pub/Sub 和 Cloud Scheduler API 的 Google Cloud 專案。
- 建立及管理 Cloud Storage 值區、服務帳戶、Cloud Run 函式、Pub/Sub 主題和 Cloud Scheduler 工作的權限。
方法 2 的其他必要條件 (Amazon S3)：
- AWS (S3、Identity and Access Management、Lambda、EventBridge) 的具備權限存取權。
在 Akamai Control Center 中啟用 SIEM 整合
- 登入 Akamai Control Center。
- 前往「Cloud Security」部分下方的「Security Configurations」。
- 開啟要收集 SIEM 資料的安全性設定 (和適當版本)。
- 按一下「進階設定」，然後展開「SIEM 整合的資料收集」。
- 前往「Cloud Security」部分下方的「Security Configurations」。
- 開啟要收集 SIEM 資料的安全性設定 (和適當版本)。
- 按一下「進階設定」，然後展開「SIEM 整合的資料收集」。
- 按一下「開啟」即可啟用 SIEM。
- 選擇要匯出資料的安全政策：
- 所有安全性政策：針對違反安全性設定中任何或所有安全性政策的事件，傳送 SIEM 資料。
- 特定安全性政策：從下拉式選單中選取一或多項特定安全性政策。
選用：如果您使用帳戶保護功能，並想加入未加密的「使用者名稱」，請勾選「加入使用者名稱」核取方塊。
選用：如要在 SIEM 事件中接收 JA4 指紋資訊，請勾選「Include the JA4 Client TLS Fingerprint」(包含 JA4 用戶端 TLS 指紋) 核取方塊。
選用：如要排除屬於特定保護措施類型和動作的事件，請按一下「新增例外狀況」。選取您不希望 SIEM 收集的防護措施和相關動作。
選用：如要在 SIEM 事件中接收 JA4 指紋資訊，請勾選「Include the JA4 Client TLS Fingerprint」(包含 JA4 用戶端 TLS 指紋) 核取方塊。
選用：如要排除屬於特定保護措施類型和動作的事件，請按一下「新增例外狀況」。選取您不希望 SIEM 收集的防護措施和相關動作。
按一下 [儲存]。
從 SIEM 整合部分複製並儲存「安全性設定 ID」 (configId)。您需要這個值才能設定 Lambda。
從 SIEM 整合部分複製並儲存「安全性設定 ID」 (configId)。您需要這個值才能設定 Lambda。
為 SIEM 整合建立 Akamai API 憑證
- 登入 Akamai Control Center。
- 依序前往「帳戶管理員」>「身分與存取權」>「使用者和 API 用戶端」。
- 依序前往「帳戶管理員」>「身分與存取權」>「使用者和 API 用戶端」。
- 按一下「建立 API 用戶端」。
- 按一下「進階」，建立進階 API 用戶端。
- 按一下「進階」，建立進階 API 用戶端。
- 請提供下列設定詳細資料：
- 名稱：輸入描述性名稱 (例如
Google SecOps SIEM Poller)。 - API 服務：選取「SIEM」，並將存取層級設為「READ-WRITE」。
- 名稱：輸入描述性名稱 (例如
SecOps SIEM Poller)。 - API 服務：選取「SIEM」，並將存取層級設為「READ-WRITE」。
- 名稱：輸入描述性名稱 (例如
- 按一下「建立 API 用戶端」。
- 按一下「下載」將憑證儲存為
.edgerc檔案，或按一下「複製」複製憑證。安全地儲存下列值：- 用戶端憑證
- 用戶端密碼
- 存取權杖
- 主機 (例如
akab-xxxxxxxxxxxxxxxx-xxxxxxxxxxxxxxxx.luna.akamaiapis.net)
方法 1：Cloud Storage
使用這個方法，將 Akamai SIEM 連接器記錄收集到 Cloud Storage bucket。Cloud Run 函式會透過 Pub/Sub 由 Cloud Scheduler 觸發，定期從 Akamai SIEM Integration API 提取事件，並寫入值區，供 Google SecOps 擷取。
建立 Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下「建立值區」。
- 請提供下列設定詳細資料：
- 為 bucket 命名：輸入全域不重複的名稱 (例如
akamai-siem-logs)。 - 位置類型：根據需求選擇 (區域、雙區域或多區域)。
- 位置：選取位置 (例如
us-central1)。 - 儲存空間級別：選取「Standard」。
- 存取控管：選取「統一」。
- 為 bucket 命名：輸入全域不重複的名稱 (例如
- 點選「建立」。
為 Cloud Run 函式建立服務帳戶
- 依序前往「IAM 與管理」>「服務帳戶」。
- 按一下「建立服務帳戶」。
- 請提供下列設定詳細資料：
- 服務帳戶名稱：輸入
akamai-siem-collector-sa。 - 服務帳戶說明：輸入
Service account for Cloud Run function to collect Akamai SIEM Connector logs。
- 服務帳戶名稱：輸入
- 點按「Create and continue」(建立並繼續)。
- 授予下列角色：
- Storage 物件管理員：將事件寫入 bucket，並管理狀態檔案。
- Cloud Run 叫用者：允許 Pub/Sub 叫用函式。
- 依序點選「繼續」和「完成」。
授予服務帳戶 bucket 存取權
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱。
- 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (
akamai-siem-collector-sa@<PROJECT_ID>.iam.gserviceaccount.com)。 - 指派角色：選取「Storage 物件管理員」。
- 新增主體：輸入服務帳戶電子郵件地址 (
- 按一下 [儲存]。
建立 Pub/Sub 主題
- 前往「Pub/Sub」>「主題」。
- 按一下「建立主題」。
- 在「Topic ID」(主題 ID) 欄位中輸入
akamai-siem-trigger。 - 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式是由 Cloud Scheduler 的 Pub/Sub 訊息觸發。這項功能會使用 Akamai EdgeGrid 憑證簽署要求、從 SIEM 整合 API 擷取事件，並將事件寫入 Bucket。
- 前往 Cloud Run。
- 按一下「Create service」(建立服務)。
- 選取「函式」即可使用內嵌程式碼編輯器。
- 請提供下列設定詳細資料：
- 服務名稱：輸入
akamai-siem-collector。 - Region (區域)：選取與值區相符的區域 (例如
us-central1)。 - 執行階段：選取 Python 3.12 以上版本。
- 服務名稱：輸入
- 在「Trigger」(觸發條件) 區段：
- 按一下 [Add trigger] (新增觸發條件)。
- 選取「Pub/Sub」。
- 在「Select a Pub/Sub topic」(選取 Pub/Sub 主題) 中，選取
akamai-siem-trigger。 - 按一下 [儲存]。
- 在「驗證」部分，選取「需要驗證」。
- 展開「Containers, Networking, Security」，然後在「Security」分頁中，將「服務帳戶」設為
akamai-siem-collector-sa。 在「Containers」>「Variables & Secrets」分頁中，新增下列環境變數：
變數名稱 範例值 GCS_BUCKETakamai-siem-logsGCS_PREFIXakamai-siem/STATE_KEYakamai-siem/state.jsonAKAMAI_HOSTakab-xxxxxxxx.luna.akamaiapis.netAKAMAI_CLIENT_TOKENakab-client-token-xxxxAKAMAI_CLIENT_SECRETyour-client-secretAKAMAI_ACCESS_TOKENakab-access-token-xxxxAKAMAI_CONFIG_ID12345LIMIT10000WINDOW_SECONDS3600HTTP_TIMEOUT110MAX_PAGES20將「Request timeout」(要求逾時) 設為
600秒，並將「Memory」(記憶體) 設為 512 MiB 以上。點按「Create」(建立)。服務建立完成後，系統會開啟內嵌程式碼編輯器。
新增函式程式碼
- 在「函式進入點」欄位中輸入
main。 在內嵌程式碼編輯器中，將
main.py的內容替換成下列程式碼：import functions_framework from google.cloud import storage import json import os import time import uuid import requests from akamai.edgegrid import EdgeGridAuth storage_client = storage.Client() GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'akamai-siem/') STATE_KEY = os.environ.get('STATE_KEY', 'akamai-siem/state.json') AKAMAI_HOST = os.environ.get('AKAMAI_HOST') CLIENT_TOKEN = os.environ.get('AKAMAI_CLIENT_TOKEN') CLIENT_SECRET = os.environ.get('AKAMAI_CLIENT_SECRET') ACCESS_TOKEN = os.environ.get('AKAMAI_ACCESS_TOKEN') CONFIG_ID = os.environ.get('AKAMAI_CONFIG_ID') LIMIT = int(os.environ.get('LIMIT', '10000')) WINDOW_SEC = int(os.environ.get('WINDOW_SECONDS', '3600')) HTTP_TIMEOUT = int(os.environ.get('HTTP_TIMEOUT', '110')) MAX_PAGES = int(os.environ.get('MAX_PAGES', '20')) def _session(): """Create a requests session that signs requests with Akamai EdgeGrid.""" session = requests.Session() session.auth = EdgeGridAuth( client_token=CLIENT_TOKEN, client_secret=CLIENT_SECRET, access_token=ACCESS_TOKEN, ) return session def _load_state(bucket): """Load the saved offset from GCS.""" try: blob = bucket.blob(STATE_KEY) if blob.exists(): return json.loads(blob.download_as_text()) except Exception as e: print(f'Warning: could not load state: {e}') return {} def _save_state(bucket, state): """Persist the offset to GCS for the next run.""" try: bucket.blob(STATE_KEY).upload_from_string( json.dumps(state, separators=(',', ':')), content_type='application/json', ) except Exception as e: print(f'Warning: could not save state: {e}') def _fetch(session, offset, from_ts): """Fetch one batch of events. Use offset when available, otherwise a time window. The Akamai SIEM API does not allow combining the offset parameter with from/to. """ url = f'https://{AKAMAI_HOST}/siem/v1/configs/{CONFIG_ID}' params = {'limit': LIMIT} if offset: params['offset'] = offset else: params['from'] = int(from_ts) resp = session.get( url, params=params, timeout=HTTP_TIMEOUT, headers={'Accept': 'application/json'}, ) resp.raise_for_status() return resp.text def _parse_ndjson(text): """Split the NDJSON response into event lines and the trailing context object. Each line is a JSON security event; the last line is a context object that contains the offset to use for the next request. """ lines = [ln for ln in text.splitlines() if ln.strip()] if not lines: return [], {} try: context = json.loads(lines[-1]) except ValueError: return lines, {} if isinstance(context, dict) and 'offset' in context: return lines[:-1], context return lines, {} def _write_events(bucket, event_lines, page_num): """Write a batch of raw JSON events to GCS, one event per line.""" ts_path = time.strftime('%Y/%m/%d', time.gmtime()) uniq = f'{int(time.time() * 1e6)}_{uuid.uuid4().hex[:8]}' key = f'{GCS_PREFIX}{ts_path}/akamai_siem_p{page_num:03d}_{uniq}.json' bucket.blob(key).upload_from_string( '\n'.join(event_lines), content_type='application/x-ndjson', ) return key @functions_framework.cloud_event def main(cloud_event): """Triggered by Pub/Sub to pull Akamai SIEM events and write them to GCS.""" if not all([GCS_BUCKET, AKAMAI_HOST, CLIENT_TOKEN, CLIENT_SECRET, ACCESS_TOKEN, CONFIG_ID]): print('Error: missing required environment variables') return bucket = storage_client.bucket(GCS_BUCKET) state = _load_state(bucket) offset = state.get('offset') from_ts = float(state.get('last_to_ts') or (time.time() - WINDOW_SEC)) session = _session() pages = 0 total = 0 try: while pages < MAX_PAGES: text = _fetch(session, offset, from_ts) events, context = _parse_ndjson(text) if events: _write_events(bucket, events, pages + 1) total += len(events) new_offset = context.get('offset') if new_offset: offset = new_offset pages += 1 # Stop when this window is drained (fewer than LIMIT events returned). if not new_offset or len(events) < LIMIT: break state['offset'] = offset state['last_to_ts'] = time.time() state['last_successful_run'] = time.time() _save_state(bucket, state) print(f'Retrieved {total} events across {pages} pages') except Exception as e: print(f'Error collecting Akamai SIEM logs: {e}') raise使用以下內容建立
requirements.txt檔案：functions-framework==3.* google-cloud-storage==2.* requests>=2.31.0 edgegrid-python>=1.3.1按一下「Deploy」(部署)，然後等待部署作業完成。
建立 Cloud Scheduler 工作
- 前往 Cloud Scheduler。
- 按一下「Create job」(建立工作)。
- 請提供下列設定詳細資料：
- 「Name」(名稱)：輸入
akamai-siem-collector-5min。 - 區域：選取與 Cloud Run 函式相同的區域。
- 頻率：輸入
*/5 * * * *(每五分鐘)。 - 時區：選取「UTC」。
- 目標類型：選取「Pub/Sub」。
- 主題：選取
akamai-siem-trigger。 - 郵件內文：輸入
{}。
- 「Name」(名稱)：輸入
- 點選「建立」。
- 如要驗證設定，請按一下作業的「強制執行」，然後檢查 Cloud Run 記錄和 bucket，確認事件已寫入。
在 Google SecOps 中設定動態饋給，從 Cloud Storage 擷取資料
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 在下一個頁面中，按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Akamai SIEM Connector GCS)。 - 選取「Cloud Storage V2」做為「來源類型」。
- 選取「Akamai SIEM 連接器」做為「記錄類型」。
- 按一下「Chronicle 服務帳戶」欄位旁的「取得服務帳戶」，然後複製服務帳戶電子郵件地址。
- 點選「下一步」。
- 指定下列輸入參數的值：
- 儲存空間值區 URI：輸入
gs://akamai-siem-logs/akamai-siem/。這個網址的結尾必須是正斜線 (/)。 - 來源刪除選項：根據偏好設定選取刪除選項。
- 檔案存在時間上限：納入在過去天數內修改的檔案。預設值為 180 天。
- 儲存空間值區 URI：輸入
- 點選「下一步」。
- 在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。
授予 Google SecOps 服務帳戶值區存取權
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱。
- 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：貼上動態消息設定中的 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」 (如果您選擇刪除選項，請選取「Storage 物件管理員」)。
- 按一下 [儲存]。
方法 2：Amazon S3
使用這個方法，透過 AWS Lambda 和 EventBridge 將 Akamai SIEM 連接器記錄檔收集到 Amazon S3 值區。
為 Google SecOps 設定 AWS S3 值區和 IAM
- 按照這份使用者指南建立 Amazon S3 bucket：建立 bucket
- 請儲存 bucket 的「名稱」和「區域」，以供日後參考 (例如
akamai-siem-logs)。 - 請按照這份使用者指南建立使用者：建立 IAM 使用者。
- 選取建立的「使用者」。
- 選取「安全憑證」分頁標籤。
- 在「Access Keys」部分中，按一下「Create Access Key」。
- 選取「第三方服務」做為「用途」。
- 點選「下一步」。
- 選用：新增說明標記。
- 按一下「建立存取金鑰」。
- 按一下「Download .csv file」(下載 .csv 檔案)，儲存「Access Key」(存取金鑰) 和「Secret Access Key」(私密存取金鑰)，以供日後參考。
- 按一下 [完成]。
- 選取「權限」分頁標籤。
- 在「權限政策」部分中，按一下「新增權限」。
- 選取「新增權限」。
- 選取「直接附加政策」。
- 搜尋「AmazonS3FullAccess」AmazonS3FullAccess政策。
- 選取政策。
- 點選「下一步」。
- 按一下「Add permissions」。
設定 S3 上傳的身分與存取權管理政策和角色
- 在 AWS 控制台中，依序前往「IAM」>「Policies」>「Create policy」>「JSON」分頁標籤。
- 複製並貼上下列政策。
政策 JSON (將
akamai-siem-logs替換為您的 bucket 名稱)：{ "Version": "2012-10-17", "Statement": [ { "Sid": "AllowPutObjects", "Effect": "Allow", "Action": "s3:PutObject", "Resource": "arn:aws:s3:::akamai-siem-logs/*" }, { "Sid": "AllowGetStateObject", "Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::akamai-siem-logs/akamai-siem/state.json" } ] }{ "Version": "2012-10-17", "Statement": [ { "Sid": "AllowPutObjects", "Effect": "Allow", "Action": "s3:PutObject", "Resource": "arn:aws:s3:::akamai-siem-logs/*" }, { "Sid": "AllowGetStateObject", "Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::akamai-siem-logs/akamai-siem/state.json" } ] }點選「下一步」。
輸入政策名稱
AkamaiSIEMtoS3Policy，然後按一下「建立政策」。依序前往「IAM」>「角色」>「建立角色」。
選取「AWS 服務」。
選取「Lambda」Lambda做為用途。
點選「下一步」。
搜尋並選取政策
AkamaiSIEMtoS3Policy。搜尋並選取政策
AkamaiSIEMtoS3Policy。點選「下一步」。
輸入角色名稱
AkamaiSIEMtoS3Role，然後按一下「建立角色」。
建立 Lambda 函式
- 在 AWS 控制台中，依序前往「Lambda」>「Functions」>「Create function」。
- 按一下「從頭開始撰寫」。
請提供下列設定詳細資料：
設定 值 名稱 AkamaiSIEMtoS3Function執行階段 Python 3.13 架構 x86_64 執行角色 使用現有角色 現有角色 AkamaiSIEMtoS3Role設定 值 名稱 AkamaiSIEMtoS3Function執行階段 Python 3.13 架構 x86_64 執行角色 使用現有角色 現有角色 AkamaiSIEMtoS3Role按一下「Create function」(建立函式)。
建立函式後，開啟「程式碼」分頁，刪除虛設常式並貼上下列程式碼：
import json import boto3 import os import urllib3 import hmac import hashlib import base64 from datetime import datetime from urllib.parse import urlparse, urljoin ```python import json import boto3 import os import urllib3 import hmac import hashlib import base64 from datetime import datetime from urllib.parse import urlparse, urljoin # Configuration from environment variables S3_BUCKET = os.environ['S3_BUCKET'] S3_PREFIX = os.environ.get('S3_PREFIX', 'akamai-siem/') STATE_KEY = os.environ.get('STATE_KEY', 'akamai-siem/state.json') # Configuration from environment variables S3_BUCKET = os.environ['S3_BUCKET'] S3_PREFIX = os.environ.get('S3_PREFIX', 'akamai-siem/') STATE_KEY = os.environ.get('STATE_KEY', 'akamai-siem/state.json') AKAMAI_HOST = os.environ['AKAMAI_HOST'] AKAMAI_CLIENT_TOKEN = os.environ['AKAMAI_CLIENT_TOKEN'] AKAMAI_CLIENT_SECRET = os.environ['AKAMAI_CLIENT_SECRET'] AKAMAI_ACCESS_TOKEN = os.environ['AKAMAI_ACCESS_TOKEN'] AKAMAI_CONFIG_IDS = os.environ['AKAMAI_CONFIG_IDS'].split(',') AKAMAI_HOST = os.environ['AKAMAI_HOST'] AKAMAI_CLIENT_TOKEN = os.environ['AKAMAI_CLIENT_TOKEN'] AKAMAI_CLIENT_SECRET = os.environ['AKAMAI_CLIENT_SECRET'] AKAMAI_ACCESS_TOKEN = os.environ['AKAMAI_ACCESS_TOKEN'] AKAMAI_CONFIG_IDS = os.environ['AKAMAI_CONFIG_IDS'].split(',') LIMIT = int(os.environ.get('LIMIT', '10000')) LIMIT = int(os.environ.get('LIMIT', '10000')) s3_client = boto3.client('s3') http = urllib3.PoolManager() s3_client = boto3.client('s3') http = urllib3.PoolManager() def load_state(): """Load offset state from S3""" try: response = s3_client.get_object(Bucket=S3_BUCKET, Key=STATE_KEY) return json.loads(response['Body'].read().decode('utf-8')) except s3_client.exceptions.NoSuchKey: return {} except Exception as e: print(f"Error loading state: {e}") return {} def load_state(): """Load offset state from S3""" try: response = s3_client.get_object(Bucket=S3_BUCKET, Key=STATE_KEY) return json.loads(response['Body'].read().decode('utf-8')) except s3_client.exceptions.NoSuchKey: return {} except Exception as e: print(f"Error loading state: {e}") return {} def save_state(state): """Save offset state to S3""" try: s3_client.put_object( Bucket=S3_BUCKET, Key=STATE_KEY, Body=json.dumps(state, indent=2).encode('utf-8'), ContentType='application/json' ) except Exception as e: print(f"Error saving state: {e}") def save_state(state): """Save offset state to S3""" try: s3_client.put_object( Bucket=S3_BUCKET, Key=STATE_KEY, Body=json.dumps(state, indent=2).encode('utf-8'), ContentType='application/json' ) except Exception as e: print(f"Error saving state: {e}") def make_edgegrid_auth_header(url, method='GET'): """Create EdgeGrid authentication header""" timestamp = datetime.utcnow().strftime('%Y%m%dT%H:%M:%S+0000') nonce = base64.b64encode(os.urandom(16)).decode('utf-8') parsed_url = urlparse(url) relative_url = parsed_url.path if parsed_url.query: relative_url += '?' + parsed_url.query auth_header = f'EG1-HMAC-SHA256 ' \ f'client_token={AKAMAI_CLIENT_TOKEN};' \ f'access_token={AKAMAI_ACCESS_TOKEN};' \ f'timestamp={timestamp};' \ f'nonce={nonce};' data_to_sign = '\t'.join([ method, parsed_url.scheme, parsed_url.netloc, relative_url, '', # Request body for GET '', # No additional headers ]) signing_key = hmac.new( AKAMAI_CLIENT_SECRET.encode('utf-8'), timestamp.encode('utf-8'), hashlib.sha256 ).digest() auth_signature = base64.b64encode( hmac.new( signing_key, (data_to_sign + auth_header).encode('utf-8'), hashlib.sha256 ).digest() ).decode('utf-8') return auth_header + f'signature={auth_signature}' def make_edgegrid_auth_header(url, method='GET'): """Create EdgeGrid authentication header""" timestamp = datetime.utcnow().strftime('%Y%m%dT%H:%M:%S+0000') nonce = base64.b64encode(os.urandom(16)).decode('utf-8') parsed_url = urlparse(url) relative_url = parsed_url.path if parsed_url.query: relative_url += '?' + parsed_url.query auth_header = f'EG1-HMAC-SHA256 ' \ f'client_token={AKAMAI_CLIENT_TOKEN};' \ f'access_token={AKAMAI_ACCESS_TOKEN};' \ f'timestamp={timestamp};' \ f'nonce={nonce};' data_to_sign = '\t'.join([ method, parsed_url.scheme, parsed_url.netloc, relative_url, '', # Request body for GET '', # No additional headers ]) signing_key = hmac.new( AKAMAI_CLIENT_SECRET.encode('utf-8'), timestamp.encode('utf-8'), hashlib.sha256 ).digest() auth_signature = base64.b64encode( hmac.new( signing_key, (data_to_sign + auth_header).encode('utf-8'), hashlib.sha256 ).digest() ).decode('utf-8') return auth_header + f'signature={auth_signature}' def fetch_akamai_events(config_id, offset=None): """Fetch events from Akamai SIEM API""" base_url = f'https://{AKAMAI_HOST}' endpoint = f'/siem/v1/configs/{config_id}' params = f'limit={LIMIT}' if offset: params += f'&offset={offset}' url = f'{base_url}{endpoint}?{params}' try: headers = { 'Authorization': make_edgegrid_auth_header(url) } response = http.request('GET', url, headers=headers, timeout=120) if response.status != 200: print(f"Error response {response.status}: {response.data.decode('utf-8')}") return [], offset # Parse multi-JSON response (newline-delimited JSON) lines = response.data.decode('utf-8').strip().split('\n') events = [] new_offset = offset for line in lines: if not line.strip(): continue try: obj = json.loads(line) # Check if this is offset context (metadata object with offset) if 'offset' in obj and ('total' in obj or 'responseContext' in obj): new_offset = obj.get('offset') continue # This is an event events.append(obj) except json.JSONDecodeError as e: print(f"Warning: Failed to parse line: {e}") continue return events, new_offset except Exception as e: print(f"Error fetching events for config {config_id}: {e}") return [], offset def fetch_akamai_events(config_id, offset=None): """Fetch events from Akamai SIEM API""" base_url = f'https://{AKAMAI_HOST}' endpoint = f'/siem/v1/configs/{config_id}' params = f'limit={LIMIT}' if offset: params += f'&offset={offset}' url = f'{base_url}{endpoint}?{params}' try: headers = { 'Authorization': make_edgegrid_auth_header(url) } response = http.request('GET', url, headers=headers, timeout=120) if response.status != 200: print(f"Error response {response.status}: {response.data.decode('utf-8')}") return [], offset # Parse multi-JSON response (newline-delimited JSON) lines = response.data.decode('utf-8').strip().split('\n') events = [] new_offset = offset for line in lines: if not line.strip(): continue try: obj = json.loads(line) # Check if this is offset context (metadata object with offset) if 'offset' in obj and ('total' in obj or 'responseContext' in obj): new_offset = obj.get('offset') continue # This is an event events.append(obj) except json.JSONDecodeError as e: print(f"Warning: Failed to parse line: {e}") continue return events, new_offset except Exception as e: print(f"Error fetching events for config {config_id}: {e}") return [], offset def lambda_handler(event, context): """Lambda handler - fetches Akamai events and writes to S3""" print(f"Starting Akamai SIEM fetch at {datetime.utcnow().isoformat()}Z") state = load_state() total_events = 0 for config_id in AKAMAI_CONFIG_IDS: config_id = config_id.strip() if not config_id: continue print(f"Fetching events for config: {config_id}") current_offset = state.get(config_id) events, new_offset = fetch_akamai_events(config_id, current_offset) if events: print(f"Fetched {len(events)} events for config {config_id}") # Write events to S3 as newline-delimited JSON timestamp = datetime.utcnow().strftime('%Y%m%d_%H%M%S') s3_key = f'{S3_PREFIX}{config_id}/{timestamp}.json' payload = '\n'.join(json.dumps(event) for event in events) try: s3_client.put_object( Bucket=S3_BUCKET, Key=s3_key, Body=payload.encode('utf-8'), ContentType='application/json' ) print(f"Wrote {len(events)} events to s3://{S3_BUCKET}/{s3_key}") # Update offset only after successful write if new_offset: state[config_id] = new_offset total_events += len(events) except Exception as e: print(f"Error writing to S3: {e}") else: print(f"No new events for config {config_id}") # Save updated state save_state(state) return { 'statusCode': 200, 'body': json.dumps({ 'message': f'Successfully processed {total_events} events', 'configs_processed': len(AKAMAI_CONFIG_IDS) }) }def lambda_handler(event, context): """Lambda handler - fetches Akamai events and writes to S3""" print(f"Starting Akamai SIEM fetch at {datetime.utcnow().isoformat()}Z")
state = load_state() total_events = 0 for config_id in AKAMAI_CONFIG_IDS: config_id = config_id.strip() if not config_id: continue print(f"Fetching events for config: {config_id}") current_offset = state.get(config_id) events, new_offset = fetch_akamai_events(config_id, current_offset) if events: print(f"Fetched {len(events)} events for config {config_id}") # Write events to S3 as newline-delimited JSON timestamp = datetime.utcnow().strftime('%Y%m%d_%H%M%S') s3_key = f'{S3_PREFIX}{config_id}/{timestamp}.json' payload = '\n'.join(json.dumps(event) for event in events) try: s3_client.put_object( Bucket=S3_BUCKET, Key=s3_key, Body=payload.encode('utf-8'), ContentType='application/json' ) print(f"Wrote {len(events)} events to s3://{S3_BUCKET}/{s3_key}") # Update offset only after successful write if new_offset: state[config_id] = new_offset total_events += len(events) except Exception as e: print(f"Error writing to S3: {e}") else: print(f"No new events for config {config_id}") # Save updated state save_state(state) return { 'statusCode': 200, 'body': json.dumps({ 'message': f'Successfully processed {total_events} events', 'configs_processed': len(AKAMAI_CONFIG_IDS) }) }按一下「Deploy」(部署) 即可儲存程式碼。
依序前往「Configuration」>「Environment variables」>「Edit」。
依序前往「Configuration」>「Environment variables」>「Edit」。
針對下列各項，按一下「新增環境變數」：
環境變數 環境變數
| 鍵 | 範例值 |
|---|---|
S3_BUCKET |
akamai-siem-logs |
S3_PREFIX |
akamai-siem/ |
STATE_KEY |
akamai-siem/state.json |
AKAMAI_HOST |
akab-xxxxxxxx.luna.akamaiapis.net |
AKAMAI_CLIENT_TOKEN |
your-client-token |
AKAMAI_CLIENT_SECRET |
your-client-secret |
AKAMAI_ACCESS_TOKEN |
your-access-token |
AKAMAI_CONFIG_IDS |
12345,67890 |
LIMIT |
10000 |
| 鍵 | 範例值 |
S3_BUCKET |
akamai-siem-logs |
S3_PREFIX |
akamai-siem/ |
STATE_KEY |
akamai-siem/state.json |
AKAMAI_HOST |
akab-xxxxxxxx.luna.akamaiapis.net |
AKAMAI_CLIENT_TOKEN |
your-client-token |
AKAMAI_CLIENT_SECRET |
your-client-secret |
AKAMAI_ACCESS_TOKEN |
your-access-token |
AKAMAI_CONFIG_IDS |
12345,67890 |
LIMIT |
10000 |
- 按一下 [儲存]。
- 選取「設定」分頁標籤。
- 在「一般設定」面板中，按一下「編輯」。
- 將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒))，然後按一下「Save」(儲存)。
- 選取「設定」分頁標籤。
- 在「一般設定」面板中，按一下「編輯」。
將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒))，然後按一下「Save」(儲存)。
建立 EventBridge 排程
- 依序前往「Amazon EventBridge」>「Scheduler」>「Create schedule」。
- 請提供下列設定詳細資料：
- 時間表名稱：輸入
AkamaiSIEMtoS3-5min。 - 排程模式：選取「週期性排程」。
- 時間表類型：選取「以費率為準的時間表」。
- 費率運算式：輸入
5並選取「分鐘」。 - 時間表名稱：輸入
AkamaiSIEMtoS3-5min。 - 排程模式：選取「週期性排程」。
- 時間表類型：選取「以費率為準的時間表」。
- 費率運算式：輸入
5並選取「分鐘」。
- 時間表名稱：輸入
- 點選「下一步」。
- 請提供下列設定詳細資料：
- 目標：選取「AWS Lambda Invoke」。
- Lambda 函式：選取
AkamaiSIEMtoS3Function。 - 目標：選取「AWS Lambda Invoke」。
- Lambda 函式：選取
AkamaiSIEMtoS3Function。
- 點選「下一步」。
- 按一下「下一步」 (略過選用設定)。
- 檢查並點選「建立排程」。
在 Google SecOps 中設定資訊提供，擷取 Akamai SIEM 連接器記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 在下一個頁面中，按一下「設定單一動態饋給」。
- 輸入專屬的動態饋給名稱 (例如
Akamai SIEM Connector)。 - 按一下「新增動態消息」。
- 在下一個頁面中，按一下「設定單一動態饋給」。
- 輸入專屬的動態饋給名稱 (例如
Akamai SIEM Connector)。 - 選取「Amazon S3 V2」做為「來源類型」。
- 選取「Akamai SIEM 連接器」做為「記錄類型」。
- 依序點按「繼續」和「提交」。
- 依序點按「繼續」和「提交」。
指定下列輸入參數的值：
- S3 URI：
s3://akamai-siem-logs/akamai-siem/ - 來源刪除選項：根據偏好設定選取刪除選項。
- 檔案存在時間上限：納入在過去天數內修改的檔案。預設值為 180 天。
- 存取金鑰 ID：具有 S3 值區存取權的使用者存取金鑰。
- 存取密鑰：具有 S3 bucket 存取權的使用者私密金鑰。
- 資產命名空間：資產命名空間。
- 擷取標籤：套用至這個動態饋給事件的標籤。
- S3 URI：
s3://akamai-siem-logs/akamai-siem/ - 來源刪除選項：根據偏好設定選取刪除選項。
- 檔案存在時間上限：納入在過去天數內修改的檔案。預設值為 180 天。
- 存取金鑰 ID：具有 S3 值區存取權的使用者存取金鑰。
- 存取密鑰：具有 S3 bucket 存取權的使用者私密金鑰。
- 資產命名空間：資產命名空間。
- 擷取標籤：套用至這個動態饋給事件的標籤。
- S3 URI：
依序點按「繼續」和「提交」。
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
accept_language |
additional.fields |
已合併 |
additional_policy_key |
additional.fields |
已合併 |
attackData_label |
additional.fields |
已合併 |
attackData_ruleActions_label |
additional.fields |
已合併 |
attackData_ruleData_label |
additional.fields |
已合併 |
attackData_ruleSelectors_label |
additional.fields |
已合併 |
attackdata_ruleMessages_label |
additional.fields |
已合併 |
attackdata_ruleTags_label |
additional.fields |
已合併 |
attackdata_ruleVersions_label |
additional.fields |
已合併 |
botData_label |
additional.fields |
已合併 |
continent_label |
additional.fields |
已合併 |
dur_label |
additional.fields |
已合併 |
etag_label |
additional.fields |
已合併 |
format_label |
additional.fields |
已合併 |
httpMessage_query_label |
additional.fields |
已合併 |
httpMessage_response_headers_label |
additional.fields |
已合併 |
identity_label |
additional.fields |
已合併 |
identityv2_label |
additional.fields |
已合併 |
mime_version_label |
additional.fields |
已合併 |
regionCode_label |
additional.fields |
已合併 |
req_id_label |
additional.fields |
已合併 |
requestHeaders_label |
additional.fields |
已合併 |
responseSegment_label |
additional.fields |
已合併 |
ruleAction_list |
additional.fields |
已合併 |
ruleData_list |
additional.fields |
已合併 |
ruleMessages_list |
additional.fields |
已合併 |
ruleSelectors_list |
additional.fields |
已合併 |
ruleTags_list |
additional.fields |
已合併 |
ruleVersions_list |
additional.fields |
已合併 |
rules_list |
additional.fields |
已合併 |
same_site |
additional.fields |
已合併 |
telemetryType_label |
additional.fields |
已合併 |
Date |
metadata.event_timestamp |
已剖析為 EEE, dd MMM yyyy HH:mm:ss ZZZ |
expires |
metadata.event_timestamp |
已剖析為 EEE, dd-MMM-yyyy HH:mm:ss ZZZ |
httpMessage.start |
metadata.event_timestamp |
已剖析為 UNIX |
event_type |
metadata.event_type |
直接對應 |
attackData.configId |
metadata.product_log_id |
直接對應 |
version |
metadata.product_version |
直接對應 |
protocol |
network.application_protocol |
直接對應 |
MS-ASProtocolVersion |
network.application_protocol_version |
直接對應 |
httpMessage.method |
network.http.method |
直接對應 |
httpMessage.status |
network.http.response_code |
已重新命名/對應 |
User-Agent |
network.http.user_agent |
直接對應 |
httpMessage.bytes |
network.sent_bytes |
已重新命名/對應 |
ASP.NET_SessionId |
network.session_id |
直接對應 |
httpMessage.requestId |
network.session_id |
直接對應 |
httpMessage.tls |
network.tls.version |
直接對應 |
attackData.clientIP |
principal.asset.ip |
已合併 |
attackData.clientIP |
principal.ip |
已合併 |
geo.city |
principal.location.city |
直接對應 |
geo.country |
principal.location.country_or_region |
直接對應 |
geo.asn |
principal.location.name |
直接對應 |
additional_content_Security_Policy |
principal.resource.attribute.labels |
已合併 |
DeviceType |
principal.resource.name |
直接對應 |
DeviceId |
principal.resource.product_object_id |
直接對應 |
cache_label |
security_result.about.labels |
已合併 |
accept_encoding_label |
security_result.about.resource.attribute.labels |
已合併 |
accept_label |
security_result.about.resource.attribute.labels |
已合併 |
connection_label |
security_result.about.resource.attribute.labels |
已合併 |
content_encoding |
security_result.about.resource.attribute.labels |
已合併 |
content_length_label |
security_result.about.resource.attribute.labels |
已合併 |
content_type_label |
security_result.about.resource.attribute.labels |
已合併 |
req_content_length_label |
security_result.about.resource.attribute.labels |
已合併 |
req_content_type_label |
security_result.about.resource.attribute.labels |
已合併 |
desc |
security_result.description |
直接對應 |
CP_IsMobile_label |
security_result.detection_fields |
已合併 |
X-BEServer_label |
security_result.detection_fields |
已合併 |
X-CalculatedBETarget_label |
security_result.detection_fields |
已合併 |
X-Content-Type-Options_label |
security_result.detection_fields |
已合併 |
X-DiagInfo_label |
security_result.detection_fields |
已合併 |
X-FEServer_label |
security_result.detection_fields |
已合併 |
X-MS-BackOffDuration_label |
security_result.detection_fields |
已合併 |
apiId_label |
security_result.detection_fields |
已合併 |
backend_label |
security_result.detection_fields |
已合併 |
p3p_label |
security_result.detection_fields |
已合併 |
type_label |
security_result.detection_fields |
已合併 |
vary_label |
security_result.detection_fields |
已合併 |
attackData.policyId |
security_result.summary |
直接對應 |
Host |
target.asset.hostname |
直接對應 |
httpMessage.host |
target.asset.hostname |
直接對應 |
path |
target.file.full_path |
直接對應 |
Host |
target.hostname |
直接對應 |
httpMessage.host |
target.hostname |
直接對應 |
httpMessage.port |
target.port |
直接對應 |
Cmd |
target.process.command_line |
直接對應 |
httpMessage.path |
target.url |
直接對應 |
User |
target.user.userid |
直接對應 |
| 不適用 | metadata.event_type |
常數：GENERIC_EVENT |
| 不適用 | metadata.product_name |
常數：AKAMAI_SIEM_CONNECTOR |
| 不適用 | metadata.vendor_name |
常數：AKAMAI_SIEM_CONNECTOR |
| 不適用 | principal.resource.resource_type |
常數：DEVICE |
變更記錄
還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。