收集 Censys 日志

支持的语言：

本文档介绍了如何使用 Amazon S3 将 Censys 日志注入到 Google Security Operations。Censys 通过其 API 提供全面的攻击面管理和互联网情报。通过此集成，您可以从 Censys ASM 收集主机发现事件、风险事件和资产变更，并将它们转发到 Google SecOps 以进行分析和监控。解析器会将原始日志转换为符合 Google SecOps UDM 的结构化格式。它从原始日志消息中提取字段，执行数据类型转换，并将提取的信息映射到相应的 UDM 字段，从而使用更多上下文和标签来丰富数据。

准备工作

请确保满足以下前提条件：

  • Google SecOps 实例
  • 对 Censys ASM 的特权访问权限
  • 对 AWS（S3、IAM、Lambda、EventBridge）的特权访问权限

收集 Censys 前提条件（API 凭据）

  1. 前往 app.censys.io，登录 Censys ASM 控制台。
  2. 前往页面顶部的集成。
  3. 复制并保存您的 API 密钥和组织 ID。
  4. 请注意 API 基准网址：https://api.platform.censys.io

为 Google SecOps 配置 AWS S3 存储桶和 IAM

  1. 按照以下用户指南创建 Amazon S3 存储桶：创建存储桶
  2. 保存存储桶名称和区域以供日后参考（例如 censys-logs）。
  3. 按照以下用户指南创建用户：创建 IAM 用户。
  4. 选择创建的用户。
  5. 选择安全凭据标签页。
  6. 在访问密钥部分中，点击创建访问密钥。
  7. 选择第三方服务作为使用情形。
  8. 点击下一步。
  9. 可选：添加说明标记。
  10. 点击创建访问密钥。
  11. 点击 Download CSV file（下载 CSV 文件），保存访问密钥和不公开的访问密钥以供日后使用。
  12. 点击完成。
  13. 选择权限标签页。
  14. 在权限政策部分中，点击添加权限。
  15. 选择添加权限。
  16. 选择直接附加政策
  17. 搜索并选择 AmazonS3FullAccess 政策。
  18. 点击下一步。
  19. 点击添加权限。

为 S3 上传配置 IAM 政策和角色

  1. 在 AWS 控制台中，依次前往 IAM > 政策 > 创建政策 > JSON 标签页。

  2. 输入以下政策：

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutObjects",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::censys-logs/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::censys-logs/censys/state.json"
    }
  ]
}
    • 如果您输入了其他存储桶名称，请替换 censys-logs。

  3. 依次点击下一步 > 创建政策。

  4. 依次前往 IAM > 角色 > 创建角色 > AWS 服务 > Lambda。

  5. 附加新创建的政策和 AWSLambdaBasicExecutionRole 受管政策（用于 CloudWatch Logs 访问权限）。

  6. 将角色命名为 censys-lambda-role，然后点击创建角色。

创建 Lambda 函数

  1. 在 AWS 控制台中，依次前往 Lambda > 函数 > 创建函数。
  2. 点击从头开始创作。
  3. 提供以下配置详细信息：
设置 值
名称 censys-data-collector
运行时 Python 3.13
架构 x86_64
执行角色 censys-lambda-role

  1. 创建函数后，打开 Code 标签页，删除桩代码并输入以下代码 (censys-data-collector.py)：

    import json
import boto3
import urllib3
import gzip
import logging
import os
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Any, Optional
from urllib.parse import urlencode

# Configure logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# AWS S3 client
s3_client = boto3.client('s3')
# HTTP client
http = urllib3.PoolManager()

# Environment variables
S3_BUCKET = os.environ['S3_BUCKET']
S3_PREFIX = os.environ['S3_PREFIX']
STATE_KEY = os.environ['STATE_KEY']
CENSYS_API_KEY = os.environ['CENSYS_API_KEY']
CENSYS_ORG_ID = os.environ['CENSYS_ORG_ID']
API_BASE = os.environ.get('API_BASE', 'https://api.platform.censys.io')

class CensysCollector:
    def __init__(self):
        self.headers = {
            'Authorization': f'Bearer {CENSYS_API_KEY}',
            'X-Organization-ID': CENSYS_ORG_ID,
            'Content-Type': 'application/json'
        }

    def get_last_collection_time(self) -> Optional[datetime]:
        """Get the last collection timestamp from S3 state file."""
        try:
            response = s3_client.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
            state = json.loads(response['Body'].read().decode('utf-8'))
            return datetime.fromisoformat(state.get('last_collection_time', '2024-01-01T00:00:00Z'))
        except Exception as e:
            logger.info(f"No state file found or error reading state: {e}")
            return datetime.now(timezone.utc) - timedelta(hours=1)

    def save_collection_time(self, collection_time: datetime):
        """Save the current collection timestamp to S3 state file."""
        state = {'last_collection_time': collection_time.strftime('%Y-%m-%dT%H:%M:%SZ')}
        s3_client.put_object(
            Bucket=S3_BUCKET,
            Key=STATE_KEY,
            Body=json.dumps(state),
            ContentType='application/json'
        )

    def collect_logbook_events(self, cursor: str = None) -> List[Dict[str, Any]]:
        """Collect logbook events from Censys ASM API using cursor-based pagination."""
        events = []
        url = f"{API_BASE}/v3/logbook"

        # Use cursor-based pagination as per Censys API documentation
        params = {}
        if cursor:
            params['cursor'] = cursor

        try:
            query_string = urlencode(params) if params else ''
            full_url = f"{url}?{query_string}" if query_string else url

            response = http.request('GET', full_url, headers=self.headers)

            if response.status != 200:
                logger.error(f"API request failed with status {response.status}: {response.data}")
                return []

            data = json.loads(response.data.decode('utf-8'))
            events.extend(data.get('logbook_entries', []))

            # Handle cursor-based pagination
            next_cursor = data.get('next_cursor')
            if next_cursor:
                events.extend(self.collect_logbook_events(next_cursor))

            logger.info(f"Collected {len(events)} logbook events")
            return events

        except Exception as e:
            logger.error(f"Error collecting logbook events: {e}")
            return []

    def collect_risks_events(self) -> List[Dict[str, Any]]:
        """Collect risk events from Censys ASM API."""
        events = []
        url = f"{API_BASE}/v3/risks"

        try:
            response = http.request('GET', url, headers=self.headers)

            if response.status != 200:
                logger.error(f"API request failed with status {response.status}: {response.data}")
                return []

            data = json.loads(response.data.decode('utf-8'))
            events.extend(data.get('risks', []))

            logger.info(f"Collected {len(events)} risk events")
            return events

        except Exception as e:
            logger.error(f"Error collecting risk events: {e}")
            return []

    def save_events_to_s3(self, events: List[Dict[str, Any]], event_type: str):
        """Save events to S3 in compressed NDJSON format."""
        if not events:
            return

        timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
        filename = f"{S3_PREFIX}{event_type}_{timestamp}.json.gz"

        try:
            # Convert events to newline-delimited JSON
            ndjson_content = 'n'.join(json.dumps(event, separators=(',', ':')) for event in events)

            # Compress with gzip
            gz_bytes = gzip.compress(ndjson_content.encode('utf-8'))

            s3_client.put_object(
                Bucket=S3_BUCKET,
                Key=filename,
                Body=gz_bytes,
                ContentType='application/gzip',
                ContentEncoding='gzip'
            )

            logger.info(f"Saved {len(events)} {event_type} events to {filename}")

        except Exception as e:
            logger.error(f"Error saving {event_type} events to S3: {e}")
            raise

def lambda_handler(event, context):
    """AWS Lambda handler function."""
    try:
        collector = CensysCollector()

        # Get last collection time for cursor state management
        last_collection_time = collector.get_last_collection_time()
        current_time = datetime.now(timezone.utc)

        logger.info(f"Collecting events since {last_collection_time}")

        # Collect different types of events
        logbook_events = collector.collect_logbook_events()
        risk_events = collector.collect_risks_events()

        # Save events to S3
        collector.save_events_to_s3(logbook_events, 'logbook')
        collector.save_events_to_s3(risk_events, 'risks')

        # Update state
        collector.save_collection_time(current_time)

        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Censys data collection completed successfully',
                'logbook_events': len(logbook_events),
                'risk_events': len(risk_events),
                'collection_time': current_time.strftime('%Y-%m-%dT%H:%M:%SZ')
            })
        }

    except Exception as e:
        logger.error(f"Lambda execution failed: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({
                'error': str(e)
            })
        }

  2. 依次前往配置 > 环境变量 > 修改 > 添加新的环境变量。

  3. 输入以下环境变量，并替换为您的值：

    键 示例值
    S3_BUCKET censys-logs
    S3_PREFIX censys/
    STATE_KEY censys/state.json
    CENSYS_API_KEY <your-censys-api-key>
    CENSYS_ORG_ID <your-organization-id>
    API_BASE https://api.platform.censys.io

  4. 创建函数后，请停留在其页面上（或依次打开 Lambda > 函数 > 您的函数）。

  5. 选择配置标签页。

  6. 在常规配置面板中，点击修改。

  7. 将超时更改为 5 分钟（300 秒），然后点击保存。

创建 EventBridge 计划

  1. 依次前往 Amazon EventBridge > 调度程序 > 创建计划。
  2. 提供以下配置详细信息：
    • 周期性安排：费率 (1 hour)。
    • 目标：您的 Lambda 函数 censys-data-collector。
    • 名称：censys-data-collector-1h。
  3. 点击创建时间表。

可选：为 Google SecOps 创建只读 IAM 用户和密钥

  1. 在 AWS 控制台中，依次前往 IAM > 用户 > 添加用户。
  2. 点击 Add users（添加用户）。
  3. 提供以下配置详细信息：
    • 用户：secops-reader。
    • 访问类型：访问密钥 - 以程序化方式访问。
  4. 点击创建用户。
  5. 附加最低限度的读取政策（自定义）：用户 > secops-reader > 权限 > 添加权限 > 直接附加政策 > 创建政策。

  6. 在 JSON 编辑器中，输入以下政策：

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::censys-logs/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::censys-logs"
    }
  ]
}

  7. 将名称设置为 secops-reader-policy。

  8. 依次前往创建政策 > 搜索/选择 > 下一步 > 添加权限。

  9. 依次前往安全凭据 > 访问密钥 > 创建访问密钥。

  10. 下载 CSV（这些值会输入到 Feed 中）。

在 Google SecOps 中配置 Feed 以注入 Censys 日志

  1. 依次前往 SIEM 设置> Feed。
  2. 点击 + 添加新 Feed。
  3. 在Feed 名称字段中，输入 Feed 的名称（例如 Censys logs）。
  4. 选择 Amazon S3 V2 作为来源类型。
  5. 选择 CENSYS 作为日志类型。
  6. 点击下一步。
  7. 为以下输入参数指定值：
    • S3 URI：s3://censys-logs/censys/
    • 来源删除选项：根据您的偏好设置选择删除选项。
    • 文件存在时间上限：包含在过去指定天数内修改的文件。默认值为 180 天。
    • 访问密钥 ID：有权访问 S3 存储桶的用户访问密钥。
    • 私有访问密钥：具有 S3 存储桶访问权限的用户私有密钥。
    • 资源命名空间：资源命名空间。
    • 注入标签：应用于此 Feed 中事件的标签。
  8. 点击下一步。
  9. 在最终确定界面中查看新的 Feed 配置，然后点击提交。

UDM 映射表

日志字段 UDM 映射 逻辑
assetId read_only_udm.principal.asset.hostname 如果 assetId 字段不是 IP 地址，则会映射到 principal.asset.hostname。
assetId read_only_udm.principal.asset.ip 如果 assetId 字段是 IP 地址，则会映射到 principal.asset.ip。
assetId read_only_udm.principal.hostname 如果 assetId 字段不是 IP 地址，则会映射到 principal.hostname。
assetId read_only_udm.principal.ip 如果 assetId 字段是 IP 地址，则会映射到 principal.ip。
associatedAt read_only_udm.security_result.detection_fields.value associatedAt 字段映射到 security_result.detection_fields.value。
autonomousSystem.asn read_only_udm.additional.fields.value.string_value autonomousSystem.asn 字段会转换为字符串，并映射到 additional.fields.value.string_value，键为“autonomousSystem_asn”。
autonomousSystem.bgpPrefix read_only_udm.additional.fields.value.string_value autonomousSystem.bgpPrefix 字段映射到 additional.fields.value.string_value，键为“autonomousSystem_bgpPrefix”。
横幅 read_only_udm.principal.resource.attribute.labels.value 横幅字段映射到键为“banner”的 principal.resource.attribute.labels.value。
云 read_only_udm.metadata.vendor_name 云字段映射到 metadata.vendor_name。
comments.refUrl read_only_udm.network.http.referral_url comments.refUrl 字段映射到 network.http.referral_url。
data.cve read_only_udm.additional.fields.value.string_value data.cve 字段映射到键为“data_cve”的 additional.fields.value.string_value。
data.cvss read_only_udm.additional.fields.value.string_value data.cvss 字段映射到具有键“data_cvss”的 additional.fields.value.string_value。
data.ipAddress read_only_udm.principal.asset.ip 如果 data.ipAddress 字段不等于 assetId 字段，则会映射到 principal.asset.ip。
data.ipAddress read_only_udm.principal.ip 如果 data.ipAddress 字段不等于 assetId 字段，则会将其映射到 principal.ip。
data.location.city read_only_udm.principal.location.city 如果 location.city 字段为空，则将 data.location.city 字段映射到 principal.location.city。
data.location.countryCode read_only_udm.principal.location.country_or_region 如果 location.country 字段为空，则 data.location.countryCode 字段会映射到 principal.location.country_or_region。
data.location.latitude read_only_udm.principal.location.region_coordinates.latitude 如果 location.coordinates.latitude 和 location.geoCoordinates.latitude 字段为空，则将 data.location.latitude 字段转换为浮点数，并映射到 principal.location.region_coordinates.latitude。
data.location.longitude read_only_udm.principal.location.region_coordinates.longitude 如果 location.coordinates.longitude 和 location.geoCoordinates.longitude 字段为空，则将 data.location.longitude 字段转换为浮点数，并映射到 principal.location.region_coordinates.longitude。
data.location.province read_only_udm.principal.location.state 如果 location.province 字段为空，则 data.location.province 字段会映射到 principal.location.state。
data.mailServers read_only_udm.additional.fields.value.list_value.values.string_value data.mailServers 数组中的每个元素都会映射到单独的 additional.fields 条目，其中键为“Mail Servers”，value.list_value.values.string_value 设置为相应元素的值。
data.names.forwardDns[].name read_only_udm.network.dns.questions.name data.names.forwardDns 数组中的每个元素都会映射到单独的 network.dns.questions 条目，并将 name 字段设置为相应元素的 name 字段。
data.nameServers read_only_udm.additional.fields.value.list_value.values.string_value data.nameServers 数组中的每个元素都会映射到单独的 additional.fields 条目，其中键为“Name nameServers”，value.list_value.values.string_value 设置为相应元素的值。
data.protocols[].transportProtocol read_only_udm.network.ip_protocol 如果 data.protocols[].transportProtocol 字段是 TCP、EIGRP、ESP、ETHERIP、GRE、ICMP、IGMP、IP6IN4、PIM、UDP 或 VRRP 之一，则会映射到 network.ip_protocol。
data.protocols[].transportProtocol read_only_udm.principal.resource.attribute.labels.value data.protocols[].transportProtocol 字段映射到 principal.resource.attribute.labels.value，键为“data_protocols {index}”。
http.request.headers[].key, http.request.headers[].value.headers.0 read_only_udm.network.http.user_agent 如果 http.request.headers[].key 字段为“User-Agent”，则相应的 http.request.headers[].value.headers.0 字段会映射到 network.http.user_agent。
http.request.headers[].key, http.request.headers[].value.headers.0 read_only_udm.network.http.parsed_user_agent 如果 http.request.headers[].key 字段为“User-Agent”，则相应的 http.request.headers[].value.headers.0 字段会被解析为用户代理字符串并映射到 network.http.parsed_user_agent。
http.request.headers[].key, http.request.headers[].value.headers.0 read_only_udm.principal.resource.attribute.labels.key、read_only_udm.principal.resource.attribute.labels.value 对于 http.request.headers 数组中的每个元素，key 字段会映射到 principal.resource.attribute.labels.key，value.headers.0 字段会映射到 principal.resource.attribute.labels.value。
http.request.uri read_only_udm.principal.asset.hostname 提取 http.request.uri 字段的主机名部分，并将其映射到 principal.asset.hostname。
http.request.uri read_only_udm.principal.hostname 提取 http.request.uri 字段的主机名部分，并将其映射到 principal.hostname。
http.response.body read_only_udm.principal.resource.attribute.labels.value http.response.body 字段映射到 principal.resource.attribute.labels.value，键为“http_response_body”。
http.response.headers[].key, http.response.headers[].value.headers.0 read_only_udm.target.hostname 如果 http.response.headers[].key 字段为“Server”，则相应的 http.response.headers[].value.headers.0 字段会映射到 target.hostname。
http.response.headers[].key, http.response.headers[].value.headers.0 read_only_udm.principal.resource.attribute.labels.key、read_only_udm.principal.resource.attribute.labels.value 对于 http.response.headers 数组中的每个元素，key 字段会映射到 principal.resource.attribute.labels.key，而 value.headers.0 字段会映射到 principal.resource.attribute.labels.value。
http.response.statusCode read_only_udm.network.http.response_code http.response.statusCode 字段会转换为整数并映射到 network.http.response_code。
ip read_only_udm.target.asset.ip ip 字段已映射到 target.asset.ip。
ip read_only_udm.target.ip ip 字段映射到 target.ip。
isSeed read_only_udm.additional.fields.value.string_value isSeed 字段会转换为字符串，并映射到键为“isSeed”的 additional.fields.value.string_value。
location.city read_only_udm.principal.location.city location.city 字段会映射到 principal.location.city。
location.continent read_only_udm.additional.fields.value.string_value location.continent 字段映射到键为“location_continent”的 additional.fields.value.string_value。
location.coordinates.latitude read_only_udm.principal.location.region_coordinates.latitude location.coordinates.latitude 字段转换为浮点数，并映射到 principal.location.region_coordinates.latitude。
location.coordinates.longitude read_only_udm.principal.location.region_coordinates.longitude location.coordinates.longitude 字段会转换为浮点数，并映射到 principal.location.region_coordinates.longitude。
location.country read_only_udm.principal.location.country_or_region location.country 字段会映射到 principal.location.country_or_region。
location.geoCoordinates.latitude read_only_udm.principal.location.region_coordinates.latitude 如果 location.coordinates.latitude 字段为空，则将 location.geoCoordinates.latitude 字段转换为浮点数，并映射到 principal.location.region_coordinates.latitude。
location.geoCoordinates.longitude read_only_udm.principal.location.region_coordinates.longitude 如果 location.coordinates.longitude 字段为空，则将 location.geoCoordinates.longitude 字段转换为浮点数，并映射到 principal.location.region_coordinates.longitude。
location.postalCode read_only_udm.additional.fields.value.string_value location.postalCode 字段映射到键为“Postal code”的 additional.fields.value.string_value。
location.province read_only_udm.principal.location.state location.province 字段映射到 principal.location.state。
操作 read_only_udm.security_result.action_details 操作字段会映射到 security_result.action_details。
perspectiveId read_only_udm.principal.group.product_object_id perspectiveId 字段映射到 principal.group.product_object_id。
端口 read_only_udm.principal.port 端口字段会转换为整数并映射到 principal.port。
risks[].severity、risks[].title read_only_udm.security_result.category_details 将 risks[].severity 字段与 risks[].title 字段串联，并映射到 security_result.category_details。
serviceName read_only_udm.network.application_protocol 如果 serviceName 字段为“HTTP”或“HTTPS”，则会映射到 network.application_protocol。
sourceIp read_only_udm.principal.asset.ip sourceIp 字段映射到 principal.asset.ip。
sourceIp read_only_udm.principal.ip sourceIp 字段映射到 principal.ip。
时间戳 read_only_udm.metadata.event_timestamp 时间戳字段会被解析为时间戳，并映射到 metadata.event_timestamp。
transportFingerprint.id read_only_udm.metadata.product_log_id transportFingerprint.id 字段转换为字符串并映射到 metadata.product_log_id。
transportFingerprint.raw read_only_udm.additional.fields.value.string_value transportFingerprint.raw 字段映射到 additional.fields.value.string_value，键为“transportFingerprint_raw”。
类型 read_only_udm.metadata.product_event_type type 字段会映射到 metadata.product_event_type。
- read_only_udm.metadata.product_name 值“CENSYS_ASM”已分配给 metadata.product_name。
- read_only_udm.metadata.vendor_name 值“CENSYS”已分配给 metadata.vendor_name。
- read_only_udm.metadata.event_type 事件类型根据特定字段的存在情况来确定：如果 has_princ_machine_id 和 has_target_machine 为 true 且 has_network_flow 为 false，则为 NETWORK_CONNECTION；如果 has_network_flow 为 true，则为 NETWORK_DNS；如果 has_princ_machine_id 为 true，则为 STATUS_UPDATE；否则为 GENERIC_EVENT。

需要更多帮助？从社区成员和 Google SecOps 专业人士那里获得解答。