收集 Akamai SIEM 连接器日志
解析器版本：2.0
支持的平台：
Google SecOps
SIEM
本指南介绍了如何将 Akamai SIEM 连接器日志注入到 Google Security Operations。Akamai SIEM 集成通过 SIEM 集成 API 以 JSON 格式近乎实时地提供来自 Akamai 平台的安全事件。这些事件包括来自 Web 应用防火墙 (WAF)、Bot Manager 和其他 Akamai 安全产品的数据。由于 Akamai SIEM Integration API 基于拉取，因此计划函数会定期提取事件并将其写入对象存储，然后 Google SecOps 会提取这些事件。您可以使用以下两种受支持的方法之一来配置提取：
- 方法 1：Cloud Storage。Cloud Run 函数会定期从 Akamai SIEM Integration API 中拉取事件，并将其写入 Cloud Storage 存储桶，供 Google SecOps 提取。
- 方法 2：Amazon S3。AWS Lambda 函数会定期从 Akamai SIEM 集成 API 拉取事件，并将其写入 Amazon S3 存储桶，供 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 控制中心。
- 前往云安全部分下的安全配置。
- 打开您要收集 SIEM 数据的安全配置（以及相应版本）。
- 点击高级设置，然后展开 SIEM 集成的数据收集。
- 前往云安全部分下的安全配置。
- 打开您要收集 SIEM 数据的安全配置（以及相应版本）。
- 点击高级设置，然后展开 SIEM 集成的数据收集。
- 点击开启以启用 SIEM。
- 选择要导出哪些安全政策的数据：
- 所有安全政策：针对违反安全配置中任何或所有安全政策的事件发送 SIEM 数据。
- 特定安全政策：从下拉列表中选择一项或多项特定安全政策。
可选：如果您使用账号保护器，并希望包含未加密的用户名，请勾选包含用户名复选框。
可选：如果您想在 SIEM 事件中接收 JA4 指纹信息，请选中包含 JA4 客户端 TLS 指纹复选框。
可选：如果您想排除属于特定保护类型和操作的事件，请点击添加例外情况。选择您不希望 SIEM 收集的保护措施和相关操作。
可选：如果您想在 SIEM 事件中接收 JA4 指纹信息，请选中包含 JA4 客户端 TLS 指纹复选框。
可选：如果您想排除属于特定保护类型和操作的事件，请点击添加例外情况。选择您不希望 SIEM 收集的保护措施和相关操作。
点击保存。
复制并保存“SIEM 集成”部分中的安全配置 ID (configId)。您需要此值来进行 Lambda 配置。
复制并保存“SIEM 集成”部分中的安全配置 ID (configId)。您需要此值来进行 Lambda 配置。
为 SIEM 集成创建 Akamai API 凭据
- 登录 Akamai 控制中心。
- 依次前往账号管理 > 身份和访问权限 > 用户和 API 客户端。
- 依次前往账号管理 > 身份和访问权限 > 用户和 API 客户端。
- 点击创建 API 客户端。
- 点击高级以创建高级 API 客户端。
- 点击高级以创建高级 API 客户端。
- 提供以下配置详细信息：
- 名称：输入一个描述性名称（例如
Google SecOps SIEM Poller）。 - API 服务：选择 SIEM，并将访问权限级别设置为 READ-WRITE。
- 名称：输入一个描述性名称（例如
SecOps SIEM Poller）。 - API 服务：选择 SIEM，并将访问权限级别设置为 READ-WRITE。
- 名称：输入一个描述性名称（例如
- 点击创建 API 客户端。
- 点击下载可将凭据保存为
.edgerc文件，点击复制可复制凭据。请妥善保存以下值：- 客户端令牌
- 客户端密钥 (Client Secret)
- Access Token
- 主机（例如
akab-xxxxxxxxxxxxxxxx-xxxxxxxxxxxxxxxx.luna.akamaiapis.net）
方法 1：Cloud Storage
使用此方法将 Akamai SIEM 连接器日志收集到 Cloud Storage 存储桶中。一个 Cloud Run 函数，由 Cloud Scheduler 通过 Pub/Sub 触发，定期从 Akamai SIEM 集成 API 中拉取事件，并将其写入存储桶，供 Google SecOps 提取。
创建 Cloud Storage 存储桶
- 前往 Google Cloud 控制台。
- 选择您的项目或创建新项目。
- 前往 Cloud Storage > 存储分区。
- 点击创建存储分区。
- 提供以下配置详细信息：
- 指定存储桶的名称：输入一个全局唯一的名称（例如
akamai-siem-logs）。 - 位置类型：根据您的需求进行选择（区域、双区域或多区域）。
- 位置：选择位置（例如
us-central1）。 - 存储类别：选择标准。
- 访问权限控制：选择统一。
- 指定存储桶的名称：输入一个全局唯一的名称（例如
- 点击创建。
为 Cloud Run 函数创建服务账号
- 前往 IAM 和管理 > 服务账号。
- 点击创建服务账号。
- 提供以下配置详细信息：
- 服务账号名称：输入
akamai-siem-collector-sa。 - 服务账号说明：输入
Service account for Cloud Run function to collect Akamai SIEM Connector logs。
- 服务账号名称：输入
- 点击创建并继续。
- 授予以下角色：
- Storage Object Admin：将事件写入存储桶并管理状态文件。
- Cloud Run Invoker：允许 Pub/Sub 调用该函数。
- 点击继续，然后点击完成。
向服务账号授予对相应存储桶的访问权限
- 前往 Cloud Storage > 存储分区。
- 点击您的存储桶名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：输入服务账号电子邮件地址 (
akamai-siem-collector-sa@<PROJECT_ID>.iam.gserviceaccount.com)。 - 分配角色：选择 Storage Object Admin。
- 添加主账号：输入服务账号电子邮件地址 (
- 点击保存。
创建 Pub/Sub 主题
- 依次前往 Pub/Sub > 主题。
- 点击创建主题。
- 在主题 ID 字段中，输入
akamai-siem-trigger。 - 点击创建。
创建 Cloud Run 函数以收集日志
Cloud Run 函数由来自 Cloud Scheduler 的 Pub/Sub 消息触发。它使用 Akamai EdgeGrid 凭据对请求进行签名，从 SIEM 集成 API 中提取事件，并将这些事件写入存储桶。
- 前往 Cloud Run。
- 点击创建服务。
- 选择函数以使用内嵌代码编辑器。
- 提供以下配置详细信息：
- 服务名称：输入
akamai-siem-collector。 - 区域：选择与您的存储桶匹配的区域（例如
us-central1）。 - 运行时：选择 Python 3.12 或更高版本。
- 服务名称：输入
- 在触发器部分中：
- 点击添加触发器。
- 选择 Pub/Sub。
- 在选择 Pub/Sub 主题部分，选择
akamai-siem-trigger。 - 点击保存。
- 在身份验证部分中，选择需要进行身份验证。
- 展开容器、网络、安全性，然后在安全性标签页上将服务账号设置为
akamai-siem-collector-sa。 在容器 > 变量和密钥标签页中，添加以下环境变量：
变量名称 示例值 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将请求超时设置为
600秒，并将内存设置为 512 MiB 或更高。点击创建。服务创建完成后，系统会打开内嵌代码编辑器。
添加函数代码
- 在函数入口点字段中，输入
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点击部署，然后等待部署完成。
创建 Cloud Scheduler 作业
- 前往 Cloud Scheduler。
- 点击创建作业。
- 提供以下配置详细信息：
- 名称：输入
akamai-siem-collector-5min。 - 区域：选择与 Cloud Run 函数相同的区域。
- 频率：输入
*/5 * * * *（每 5 分钟一次）。 - 时区：选择 UTC。
- 目标类型：选择 Pub/Sub。
- 主题：选择
akamai-siem-trigger。 - 消息正文：输入
{}。
- 名称：输入
- 点击创建。
- 如需验证设置，请点击作业中的强制运行，然后检查 Cloud Run 日志和存储桶，确认事件已写入。
在 Google SecOps 中配置 Feed 以从 Cloud Storage 注入数据
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 在下一页上，点击配置单个 Feed。
- 在 Feed 名称字段中，输入 Feed 的名称（例如
Akamai SIEM Connector GCS）。 - 选择 Cloud Storage V2 作为来源类型。
- 选择 Akamai SIEM 连接器作为日志类型。
- 点击Chronicle 服务账号字段旁边的获取服务账号，然后复制服务账号电子邮件地址。
- 点击下一步。
- 为以下输入参数指定值：
- 存储分区 URI：输入
gs://akamai-siem-logs/akamai-siem/。此网址必须以尾随正斜杠 (/) 结尾。 - 来源删除选项：根据您的偏好选择删除选项。
- 文件存在时间上限：包含在过去指定天数内修改的文件。默认值为 180 天。
- 存储分区 URI：输入
- 点击下一步。
- 在最终确定界面中查看新的 Feed 配置，然后点击提交。
向 Google SecOps 服务账号授予对相应存储桶的访问权限
- 前往 Cloud Storage > 存储分区。
- 点击您的存储桶名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：粘贴从 Feed 设置中复制的 Google SecOps 服务账号电子邮件地址。
- 分配角色：选择 Storage Object Viewer（如果您选择了删除选项，则选择 Storage Object Admin）。
- 点击保存。
方法 2：Amazon S3
使用此方法可使用 AWS Lambda 和 EventBridge 将 Akamai SIEM 连接器日志收集到 Amazon S3 存储桶中。
为 Google SecOps 配置 AWS S3 存储桶和 IAM
- 按照以下用户指南创建 Amazon S3 存储桶：创建存储桶
- 保存存储桶名称和区域以供日后参考（例如
akamai-siem-logs）。 - 按照以下用户指南创建用户：创建 IAM 用户。
- 选择创建的用户。
- 选择安全凭据标签页。
- 在访问密钥部分中，点击创建访问密钥。
- 选择第三方服务作为使用情形。
- 点击下一步。
- 可选：添加说明标记。
- 点击创建访问密钥。
- 点击下载 .csv 文件以保存访问密钥和密钥，供日后参考。
- 点击完成。
- 选择权限标签页。
- 在权限政策部分中，点击添加权限。
- 选择添加权限。
- 选择直接附加政策。
- 搜索 AmazonS3FullAccess 政策。
- 选择相应政策。
- 点击下一步。
- 点击添加权限。
为 S3 上传配置 IAM 政策和角色
- 在 AWS 控制台中，依次前往 IAM > 政策 > 创建政策 > JSON 标签页。
- 复制并粘贴以下政策。
政策 JSON（将
akamai-siem-logs替换为您的存储桶名称）：{ "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 作为用例。
点击下一步。
搜索并选择政策
AkamaiSIEMtoS3Policy。搜索并选择政策
AkamaiSIEMtoS3Policy。点击下一步。
输入角色名称
AkamaiSIEMtoS3Role，然后点击创建角色。
创建 Lambda 函数
- 在 AWS 控制台中，依次前往 Lambda > 函数 > 创建函数。
- 点击从头开始创作。
提供以下配置详细信息：
设置 值 名称 AkamaiSIEMtoS3Function运行时 Python 3.13 架构 x86_64 执行角色 使用现有角色 现有角色 AkamaiSIEMtoS3Role设置 值 名称 AkamaiSIEMtoS3Function运行时 Python 3.13 架构 x86_64 执行角色 使用现有角色 现有角色 AkamaiSIEMtoS3Role点击创建函数。
创建函数后，打开 Code 标签页，删除存根并粘贴以下代码：
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 处理程序 - 获取 Akamai 事件并写入 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) }) }点击部署以保存代码。
依次前往配置 > 环境变量 > 修改。
依次前往配置 > 环境变量 > 修改。
针对以下各项点击添加环境变量：
环境变量 环境变量
| 键 | 示例值 |
|---|---|
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 |
- 点击保存。
- 选择配置标签页。
- 在常规配置面板中，点击修改。
- 将超时更改为 5 分钟（300 秒），然后点击保存。
- 选择配置标签页。
- 在常规配置面板中，点击修改。
将超时更改为 5 分钟（300 秒），然后点击保存。
创建 EventBridge 计划
- 依次前往 Amazon EventBridge > 调度器 > 创建调度。
- 提供以下配置详细信息：
- 时间安排名称：输入
AkamaiSIEMtoS3-5min。 - 安排模式：选择周期性安排。
- 安排类型：选择基于费率的安排。
- 速率表达式：输入
5，然后选择分钟。 - 时间安排名称：输入
AkamaiSIEMtoS3-5min。 - 安排模式：选择周期性安排。
- 安排类型：选择基于费率的安排。
- 速率表达式：输入
5，然后选择分钟。
- 时间安排名称：输入
- 点击下一步。
- 提供以下配置详细信息：
- 目标：选择 AWS Lambda 调用。
- Lambda 函数：选择
AkamaiSIEMtoS3Function。 - 目标：选择 AWS Lambda 调用。
- Lambda 函数：选择
AkamaiSIEMtoS3Function。
- 点击下一步。
- 点击下一步（跳过可选设置）。
- 检查并点击创建时间表。
在 Google SecOps 中配置 Feed 以注入 Akamai SIEM 连接器日志
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 在下一页上，点击配置单个 Feed。
- 为Feed 名称输入一个唯一名称（例如
Akamai SIEM Connector）。 - 点击添加新 Feed。
- 在下一页上，点击配置单个 Feed。
- 为Feed 名称输入一个唯一名称（例如
Akamai SIEM Connector）。 - 选择 Amazon S3 V2 作为来源类型。
- 选择 Akamai SIEM 连接器作为日志类型。
- 点击下一步，然后点击提交。
- 点击下一步，然后点击提交。
为以下输入参数指定值：
- S3 URI：
s3://akamai-siem-logs/akamai-siem/ - 源删除选项：根据您的偏好选择删除选项。
- 文件存在时间上限：包含在过去指定天数内修改的文件。默认值为 180 天。
- 访问密钥 ID：有权访问 S3 存储桶的用户访问密钥。
- 私有访问密钥：有权访问 S3 存储桶的用户私有密钥。
- 资产命名空间：资产命名空间。
- 注入标签：应用于此 Feed 中事件的标签。
- S3 URI：
s3://akamai-siem-logs/akamai-siem/ - 源删除选项：根据您的偏好选择删除选项。
- 文件存在时间上限：包含在过去指定天数内修改的文件。默认值为 180 天。
- 访问密钥 ID：有权访问 S3 存储桶的用户访问密钥。
- 私有访问密钥：有权访问 S3 存储桶的用户私有密钥。
- 资产命名空间：资产命名空间。
- 注入标签：应用于此 Feed 中事件的标签。
- 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 专业人士的解答。