亚马逊AWS官方博客

基于 Amazon Security Lake 打造统一日志分析平台 方案设计与实践

1. 前言

要识别潜在的安全威胁和漏洞，客户需要启用各种资源的日志记录功能，并将这些日志集中起来，以便在分析工具中轻松访问和使用。这些数据来源包括来自本地基础设施、防火墙和端点安全解决方案的日志，在使用云服务时，还包括 Amazon Route 53、AWS CloudTrail 和 Amazon Virtual Private Cloud（Amazon VPC）等服务的日志。

客户的安全团队在定义和实现安全领域的特定方面（例如数据标准化）遇到挑战，因为这要求他们分析每个日志来源的结构和字段，定义 Schema 和映射，并提取威胁情报等数据丰富元素。

鉴于这样的需求，本方案描述了 Amazon Security Lake 如何与第三方日志分析平台集成。介绍了 Security Lake 使用到的相关云服务、支持的数据采集、支持的数据源、自定义数据和支持的外部数据订阅。给出了 Security Lake 与 Splunk 集成方案的方案优势、挑战和适用场景，并给出了的集成 Splunk 部署方案的参考步骤和示例代码。

2. 背景介绍
  1. Amazon Security Lake 是一个 Amazon 原生托管安全数据湖服务。
  2. Security Lake 将来自 Amazon 云原生和来自第三方的安全数据集中到存储在用户 Amazon 账号中的数据湖中。
  3. 数据湖由 Amazon S3 存储桶支持，用户保留对数据的所有权。
  4. Security Lake 自动从集成的 Amazon 服务和受支持的第三方产品/服务中收集安全相关的日志和事件数据。
  5. Security Lake 将摄取到的数据转换为 Open Cybersecurity Schema Framework（OCSF）数据模式，并将其存储为 Apache Parquet 文件格式。这样其他 Amazon 服务和第三方服务可以订阅存储在 Security Lake 中的数据，用于事件响应和安全数据分析。

下图为 Security Lake 技术逻辑图，可以看到，Security Lake 最大化利用的云原生服务的优势，通过调用不同云服务的组合，实现了从数据生产、消息触发、数据 ETL、数据存储、数据管理及数据消费等完整的安全数据管理方案。

以下是 Amazon Security Lake 使用到的相关云服务:

  • Amazon S3

Security Lake 将数据存储为 Amazon S3 对象。存储类和数据留存设置基于 S3 提供的选择范围。Security Lake 利用 S3 存储的 Event Notification 机制，在新对象创建时触发 Lambda 创建 Data Catalog Partition。

  • Amazon Lambda

Security Lake 使用 Lambda 函数支持原始数据的提取、转换和加载（ETL）作业，并在 Amazon Glue 中为源数据注册 partition。

  • Amazon Glue

Security Lake 使用 Glue Crawlers 来创建 Glue Data Catalog Tables 并且将新写入 S3 的数据发送到 Data Catalog。

  • Amazon Lake Formation

Security Lake 为每个源创建一个单独的 Lake Formation Table 来为 Security Lake 提供数据。Lake Formation Table 包含来自每个数据源的信息，包括 Schema、Partition 和 Data Location Information。订阅者可以选择通过 Athena 等查询服务来查询 Lake Formation Table 中的数据。

  • Amazon Event Bridge

Security Lake 通过 Event Bridge 规则来监控对应的 S3 中是否有新对象创建，并在产生新对象时向 SQS 对应消息队列中添加新消息，用于通知订阅者进行消费。

1.1 Security Lake支持的数据采集

1.1.1  Amazon 服务数据集

下表为当前 Security Lake 支持自动化采集的云原生服务数据，由于当前仍处于 preview 预览版本，故所支持的 Amazon 云服务种类相对较少。

序号 名称
1 CloudTrail
2 VPC flow logs
3 Route 53
4 Security Hub findings

订阅者能够使用 Amazon Athena，Amazon Redshift Spectrum 等云原生数据查询服务或 Splunk，Sumo Logic 等三方 SIEM 工具进行分析。

1.1.2 第三方服务数据集

下表为当前已可与 Security Lake 集成的产品清单，能够以模块化方式将安全数据转换并输入 Security Lake。可以看到下表的大部分为 SaaS 产品。

序号 名称
1 Accenture – MxDR
2 Aqua Security
3 Barracuda – Email Protection
4 Booz Allen Hamilton
5 ChaosSearch
6 Cisco Security – Secure Firewall
7 Claroty – xDome
8 CMD Solutions
9 Confluent – Amazon S3 Sink Connector
10 Cribl – Stream
11 CrowdStrike – Falcon Data Replicator
12 CyberArk – Unified Identify Security Platform
13 Darktrace – Cyber AI Loop
14 Datadog
15 Deloitte – MXDR Cyber Analytics and AI Engine (CAE)
16 Devo
17 DXC – SecMon
18 Eviden—AIsaac（以前 Atos)
19 ExtraHop – Reveal(x) 360
20 Falcosidekick
21 Gigamon – Application Metadata Intelligence
22 IBM – QRadar
23 Infosys
24 Insbuilt
25 Kyndryl – AIOps
26 Lacework – Polygraph
27 Laminar
28 Monad
29 NETSCOUT – Omnis Cyber Intelligence
30 Netskope – CloudExchange
31 New Relic ONE
32 Okta – Workforce Identity Cloud
33 Orca – Cloud Security Platform
34 Palo Alto Networks – Prisma Cloud
35 Ping Identity – PingOne
36 PwC – Fusion center
37 Rapid7 – InsightIDR
38 RipJar – Labyrinth for Threat Investigations
39 Securonix
40 SentinelOne
41 Sentra – Data Lifecyle Security Platform
42 SOC Prime
43 Splunk
44 Stellar Cyber
45 Sumo Logic
46 Swimlane – Turbine
47 Tanium
48 TCS
49 Tines – No-code security automation
50 Torq – Enterprise Security Automation Platform
51 Trellix – XDR
52 Trend Micro – CloudOne
53 Uptycs – Uptycs XDR
54 Vectra AI – Vectra Detect for AWS
55 VMware Aria Automation for Secure Clouds
56 Wazuh
57 Wipro
58 Wiz – CNAPP
59 Zscaler – Zscaler Posture Control

1.1.3 自定义数据源

要向 Security Lake 添加其他三方或者自定义数据源，必须满足以下要求：

  1. 数据源目的地
    • 三方或者自定义数据源必须能够以 S3 对象的形式存储到包含自定义前缀的指定存储路径下，从而将数据写入 Security Lake。例如： 
      example-prefix/region=us-west-2/accountId=123456789012/eventHour=2022072112/
    • 对于包含多个类别数据的源，需要按不同的 OCSF Event Class 作为单独的源。Security Lake 将会创建一个 IAM Role，允许自定义源写入 S3 存储桶中的指定位置。
  1. 存储格式

从自定义源收集的每个 S3 对象都应格式化为 Apache Parquet 文件。

  1. 数据格式

一个 Parquet 格式对象中的每条记录应符合 OCSF 数据模式，并属于同一个 Event Class。

1.2 Security Lake 支持的外部数据订阅

目前已有部分主流 XDR（Extended Detection and Response） / MxDR （Managed Extended Detection and Response） / SIEM（Security Information and Event Management） 类产品支持订阅 Amazon Security Lake 数据，可以看到，主要分为托管安全分析服务，SaaS 安全分析产品和少数 On-premise 安全分析产品服务。

序号 名称
1 Accenture MxDR
2 Atos AIsaac platform
3 Datadog Cloud SIEM
4 Deloitte MxDR
5 DXC
6 IBM Security QRadar XDR
7 Kyndryl
8 Onica by Rackspace
10 PwC Fraud Fusion Center
11 Rapid7 InsightIDR
12 Securonix Next-Gen SIEM and XDR
13 SentinelOne Singularity™ XDR
14 Splunk
15 Sumo Logic
16 Trellix

2 Amazon Security Lake 集成 Splunk 方案设计

2.1 方案优势

Splunk 是在网络安全领域业界领先的 SIEM 产品，在 Gartner 安全信息和事件管理（Security Information and Event Management，SIEM）魔力象限中连续第九年被评为领跑者。通过集成 Amazon Security Lake，两个能够发挥各自优势，帮助用户降本增效，高效进行安全数据管理和分析。

  1. 云原生服务数据高效采集

Amazon Security Lake 最大化发挥了云原生服务的优势，能够通过极简单的流程将云原生服务数据进行快速收集，存储和解析归类。

  1. 平台级对接方式，快速便捷

Splunk 通过 Splunk Add-on for Amazon Security Lake ，能够轻松地将 Amazon Security Lake 数据导入到 Splunk 中。

  1. 快速构建云上威胁发现能力

借助 Splunk 安全分析能力和开箱即用的安全检测规则，建立端到端的威胁建模和检测能力。

  1. 安全数据管理降本增效

通过 Amazon Security Lake，对数据进行分区和分类。将出于审查目的而需要进行记录和备份的大量冷数据/非经常性搜索数据进行事件分类和压缩存储，将需要分析的网络/安全/业务数据进行对应的订阅消费，从而达到降本增效的目的。

2.2 挑战

  1. 三方/自定义数据源 ETL 复杂度高

用户持有不同品牌类型的 On-premise 软硬件产品，包含大量不同来源，不同格式的机器数据，使用 Amazon Security Lake 进行统一存储和管理，需要对数据进行 OCSF 模式化，对数据的 ETL 开发有较高要求，将耗费大量人工开发和维护成本。

  1. 当前支持的云原生数据较少

当前 Preview 版本的 Amazon Security Lake 仅支持 CloudTrail，VPC flow logs，Route 53，Security Hub findings 这四种云服务数据，对云上安全事件的分析和调查覆盖面不全，希望后续能够优先将 Guard Duty，EKS，EC2 OS logs 等关键性数据进行自动化纳管。

2.3 适用场景

  1. 在 Amazon 上持有部分关键业务，对公有云安全及云上用户活动有迫切的分析和安全建设需求的用户。
  2. Splunk License 用量较大，尝试通过基于高性价比存储方案对数据分析/管理成本进行控制的用户。
  3. 云上资产情况复杂，人员/经验有限。通过三方安全数据分析平台进行建设的经验成本和难度较大，寻找云上安全数据湖/数据仓库方案的用户。

3 Amazon Security Lake + Splunk 方案部署

3.1 角色创建

创建执行角色 AmazonSecurityLakeMetaStoreManager

{	
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowWriteLambdaInsights",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": [
        "arn:Amazon:logs:*:*:log-group:/Amazon/lambda-insights:*"
      ]
    },
    {
      "Sid": "AllowCreateCloudWatchLogGroup",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup"
      ],
      "Resource": [
        "*"
      ]
    },
    {
      "Sid": "AllowCreateAmazonCloudWatchLogGroup",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup"
      ],
      "Resource": [
        "arn:Amazon:logs:*:*:*"
      ]
    },
    {
      "Sid": "AllowWriteLambdaApplicationLogs",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": [
        "arn:Amazon:logs:*:*:log-group:/Amazon/lambda/*"
      ]
    },
    {
      "Sid": "AllowGlueManage",
      "Effect": "Allow",
      "Action": [
                "glue:CreatePartition",
                "glue:CreatePartitionIndex",
                "glue:GetPartition",
                "glue:GetPartitionIndexes",
                "glue:GetPartitions",
                "glue:UpdateTable",
                "glue:GetTable",
                "glue:GetDatabase",
                "glue:CreateDatabase",
                "glue:CreateTable",
                "glue:GetTables",
                "glue:BatchCreatePartition"
      ],
      "Resource": [
        "arn:Amazon:glue:*:*:table/Amazon_security_lake_glue_db*/*",
        "arn:Amazon:glue:*:*:database/Amazon_security_lake_glue_db*",
        "arn:Amazon:glue:*:*:catalog",
        "arn:Amazon:glue:*:*:database/default"
      ]
    },
     {
           "Sid": "AllowToReadFromSqs",
           "Effect": "Allow",
           "Action": [
               "sqs:ReceiveMessage",
               "sqs:DeleteMessage",
               "sqs:GetQueueAttributes"
           ],
           "Resource": [
               "arn:Amazon:sqs:*:{Security Lake delegated administrator account ID}:*"
           ]
       }
  ]
}

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "AllowLambdaAndSecurityLake",
            "Effect": "Allow",
            "Principal": {
                "Service": [
                    "securitylake.AmazonAmazon.com",
                    "lambda.AmazonAmazon.com"
                ]
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

3.2 定义集体目标

  • 日志和事件来源建议选择特定的日志和事件来源
  • 选择区域选择指定区域

（这两步后续可以按需要调整）

3.3 定义目标

选择汇总区域，如果选择了多个可用区，可以将所有的数据汇总到一个可用区。通过设置存储类别，选择合适的存储。

3.4 总览

创建完成后，Lambda 会将所选的日志源转换为 parquet 格式存储在相应可用区的 S3 桶中。

3.5 数据查询

可以通过 Athena 查询 Security Lake 自动构建的 Glue 爬取 S3 后构建的数据库（由于数据量少，需要等待一段时间）。并在 Lake Formation 进行统一的权限管理。

如果需要通过 Athena 查看数据，则需要给用户添加查询数据的权限。

查询数据测试

创建 SQS 订阅

来到自动创建的 Role，本次为 AmazonSecurityLake-7c9fe261-d191-4df7-a6fd-5654fc52ab0c，修改信任关系，删除外部 ID，修改 ARN 为所需的用户账号

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "1",
            "Effect": "Allow",
            "Principal": {
                "Amazon": "arn:Amazon:iam::xxxxxxxxxxx:user/xxxxxxx"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

导航到用户下载密钥。

3.6 Splunk add-on 部署

安装 Add-On for Amazon Security Lake 之前需要先安装 Python for Scientific Computing (Linux 64-bit)

搜索应用

安装完成后配置账户

配置 IAM role

结果

3.7 自定义数据源

参考要求：https://docs.Amazon.Amazon.com/security-lake/latest/userguide/custom-sources.html

根据 https://schema.ocsf.io/classes?extensions= 划分类型，并统一日志格式

参考 https://github.com/ocsf/examples

创建 IAM 角色用于 Glue 获取 S3 数据到 Lakeformation

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "glue.AmazonAmazon.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

添加 Amazon 托管策略 AmazonGlueServiceRole

添加内联策略

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "S3WriteRead",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:PutObject"
            ],
            "Resource": [
                "arn:Amazon:s3:::Amazon-security-data-lake-ap-northeast-1-ythj2ww5kon2gd4evah1bpall/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "kms:GenerateDataKey",
                "kms:Decrypt"
            ],
            "Resource": "*"
        }
    ]
}

创建自定义来源

填写账户，填写角色和事件类别，创建自定义数据源，用户需要有 Lakeformation 的 catalog 权限

创建完成后，会自动生成一个 Glue 爬虫用于爬取数据

且在 S3 会生成/ext/Scurity_Finding 路径，如：

持续向该路径写入符合要求的数据即可。

这里我用 python 模拟数据的写入：

import pandas as pd
import warnings
import uuid
from getTime import TimeTools
import json
import random
import boto3
warnings.filterwarnings('ignore')
df = pd.DataFrame(columns=['attacks', 'finding', 'malware', 'resource', 'state', 'state_id', 'vulnerabilities', 'activity_id', 'category_uid', 'class_uid', 'time', 'message', 'metadata', 'severity_id', 'activity_name', 'category_name', 'class_name', 'end_time', 'enrichments', 'observables', 'severity', 'status_code'])
with open("/home/ndmiao/parquet/securityFindings.json", "r") as file:
    data = json.load(file)

severity_id = {0: 'Unknown',1: 'Informational',2: 'Low',3: 'Medium',4: 'High',5: 'Critical',6: 'Fatal',99: 'Other'}
state_id = {0: 'Unknown',1:'New', 2:'In Progress', 3:'Suppressed', 4:'Resolved',99:'Other'}
status_id = {0: 'Unknown', 1: 'Success', 2:'Failure', 99:'Other'}
activity_id = {0: 'Unknown', 1: 'Generate', 2:'Update', 99:'Other'}

def get_data():
    timeNow = int(TimeTools().time_to_stamp(TimeTools().beijing_now()) * 10 * 10 * 10)
    severity_id_key = random.choice(list(severity_id.keys()))
    state_id_key = random.choice(list(state_id.keys()))
    status_id_key = random.choice(list(status_id.keys()))
    activity_id_key = random.choice(list(activity_id.keys()))
    type_uid = 2001 * 100 + activity_id_key
    return timeNow, severity_id_key, state_id_key, status_id_key, activity_id_key, type_uid


if __name__ == '__main__':
    random_number = random.randint(1, 10)
    for i in range(random_number):
        timeNow, severity_id_key, state_id_key, status_id_key, activity_id_key, type_uid = get_data()
        data['finding']['created_time'], data['finding']['first_seen_time'], data['finding']['last_seen_time'] = timeNow, timeNow, timeNow
        data['time'], data['metadata']['original_time'], data['metadata']['modified_time'], data['end_time'] = timeNow, timeNow, timeNow, timeNow
        data['severity_id'] = severity_id_key
        data['severity'] = severity_id[severity_id_key]
        data['state_id'] = state_id_key
        data['state'] = state_id[state_id_key]
        data['activity_id'] = activity_id_key
        data['activity_name'] = activity_id[activity_id_key]
        df = df.append(data, ignore_index=True)
    fileName = "{}.parquet".format(uuid.uuid4())
    df.to_parquet("/home/ndmiao/parquet/"+fileName)
    eventHour = TimeTools().time_to_str(TimeTools().beijing_now(), 2)
    path = 'ext/Security_Finding/region=ap-northeast-1/accountId=119596848433/eventHour={}/{}'.format(eventHour, fileName)
    # 指定 Amazon 认证凭证
    Amazon_access_key_id = 'xxxxxxxxx'
    Amazon_secret_access_key = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'

    # 创建 Boto3 Session 对象
    session = boto3.Session(
        Amazon_access_key_id=Amazon_access_key_id,
        Amazon_secret_access_key=Amazon_secret_access_key,
    )

    # 创建 S3 客户端
    s3 = session.client('s3', region_name='ap-northeast-1')

    # 将文件上传到 S3 存储桶中
    bucket_name = 'Amazon-security-data-lake-ap-northeast-1-ythj2ww5kon2gd4evah1bpall'
    s3.upload_file("/home/ndmiao/parquet/"+fileName, bucket_name, path)
    os.remove(fileName)

数据查看

由于 Security Lake 并不直接提供第三方源的 SQS 数据订阅，需要订阅第三方的数据源需要修改如下内容。订阅所用的 IAM Role 需要加上 S3 对应路径的读取权限：

在 S3 上添加对应路径的事件触发

EventBridge 事件模式添加

4 总结

本文对 Security Lake 做了介绍，并给出了实践案例参考。对解决第三方日志集中收集管理的问题，有较好的指导。随着和更多的第三方产品的集成，可以更有效的帮助到用户。

本篇作者

叶傲斌

华讯网络安全与数据分析团队架构师，从事多年安全咨询、安全运营和应急响应工作，同时拥有丰富的公有云安全架构设计经验，为国内外大型制造业、金融、零售企业提供安全合规咨询、安全规划、安全运营体系建设及托管安全运营服务。

林达

亚马逊云科技解决方案架构师，负责基于亚马逊云科技的云计算方案的咨询和架构设计。拥有超过15年的基础架构经验和安全经验，参与过多个 IT 项目的规划和设计，对于 IT 行业有深度理解和见解。

陈琪

亚马逊云科技资深解决方案架构师，超过 10 年企业数字化工作经验，涉及 IT 咨询、产品管理、软件研发等领域。长期从事企业数字化架构咨询，具有丰富的企业客户数字化转型实践经验，尤其在安全合规、数据分析、网络等技术领域具有深刻洞察。现负责亚马逊云科技全国战略合作伙伴的技术路线制定、差异化能力建设、创新性方案构建。