收集 Cisco Application Centric Infrastructure (ACI) 日志

支持的平台：

本文档介绍了如何将 Cisco Application Centric Infrastructure (ACI) 日志注入到 Google Security Operations。解析器首先尝试使用 Grok 模式将传入的 Cisco ACI 日志处理为 syslog 消息。如果 syslog 解析失败，系统会假定消息采用 JSON 格式，并相应地进行解析。最后，它会将提取的字段映射到统一数据模型 (UDM)。

此集成支持两种方法：

  • 选项 1：通过 Bindplane 代理实现的 Syslog 格式
  • 选项 2：通过 AWS S3 使用 APIC REST API 以 JSON 格式进行迁移

每种方案都自成一体，您可以根据自己的基础架构要求和日志格式偏好单独实现。

选项 1：通过 Bindplane 代理发送 Syslog

此选项用于配置 Cisco ACI Fabric，以将 syslog 消息发送到 Bindplane 代理，然后由该代理将消息转发到 Chronicle 以供分析。

准备工作

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

  • Google SecOps 实例
  • 搭载 systemd 的 Windows 2016 或更高版本或 Linux 主机
  • 如果通过代理运行，请确保防火墙端口已根据 Bindplane 代理要求打开
  • 对 Cisco APIC 控制台的特权访问

获取 Google SecOps 注入身份验证文件

  1. 登录 Google SecOps 控制台。
  2. 依次前往 SIEM 设置 > 收集代理。
  3. 下载注入身份验证文件。将文件安全地保存在将要安装 Bindplane 的系统上。

获取 Google SecOps 客户 ID

  1. 登录 Google SecOps 控制台。
  2. 依次前往 SIEM 设置 > 个人资料。
  3. 复制并保存组织详细信息部分中的客户 ID。

安装 Bindplane 代理

按照以下说明在 Windows 或 Linux 操作系统上安装 Bindplane 代理。

Windows 安装

  1. 以管理员身份打开命令提示符或 PowerShell。

  2. 运行以下命令：

    msiexec /i "https://github.com/observIQ/bindplane-agent/releases/latest/download/observiq-otel-collector.msi" /quiet

Linux 安装

  1. 打开具有 root 或 sudo 权限的终端。

  2. 运行以下命令：

    sudo sh -c "$(curl -fsSlL https://github.com/observiq/bindplane-agent/releases/latest/download/install_unix.sh)" install_unix.sh

其他安装资源

  • 如需了解其他安装选项，请参阅此安装指南。

配置 Bindplane 代理以注入 Syslog 并将其发送到 Google SecOps

  1. 访问配置文件：

    1. 找到 config.yaml 文件。通常，它位于 Linux 上的 /etc/bindplane-agent/ 目录中或 Windows 上的安装目录中。
    2. 使用文本编辑器（例如 nano、vi 或记事本）打开该文件。

  2. 按如下方式修改 config.yaml 文件：

    receivers:
  udplog:
    # Replace the port and IP address as required
    listen_address: "0.0.0.0:514"

exporters:
  chronicle/chronicle_w_labels:
    compression: gzip
    # Adjust the path to the credentials file you downloaded in Step 1
    creds_file_path: '/path/to/ingestion-authentication-file.json'
    # Replace with your actual customer ID from Step 2
    customer_id: <CUSTOMER_ID>
    endpoint: malachiteingestion-pa.googleapis.com
    # Add optional ingestion labels for better organization
    log_type: 'CISCO_ACI'
    raw_log_field: body
    ingestion_labels:

service:
  pipelines:
    logs/source0__chronicle_w_labels-0:
      receivers:
        - udplog
      exporters:
        - chronicle/chronicle_w_labels
    • 根据基础架构的需要替换端口和 IP 地址。
    • 将 <customer_id> 替换为实际的客户 ID。
    • 将 /path/to/ingestion-authentication-file.json 更新为获取 Google SecOps 注入身份验证文件部分中保存身份验证文件的路径。

重启 Bindplane 代理以应用更改

  • 如需在 Linux 中重启 Bindplane 代理，请运行以下命令：

    sudo systemctl restart bindplane-agent

  • 如需在 Windows 中重启 Bindplane 代理，您可以使用服务控制台，也可以输入以下命令：

    net stop BindPlaneAgent && net start BindPlaneAgent

在 Cisco ACI 上配置 Syslog 转发

配置带外管理合同

  1. 登录 Cisco APIC 控制台。
  2. 依次前往租户 > 管理 > 合同 > 过滤条件。
  3. 点击创建过滤器。
  4. 提供以下配置详细信息：
    • 名称：输入 syslog-udp-514。
    • 条目名称：输入 syslog。
    • EtherType：选择 IP。
    • IP 协议：选择 UDP。
    • 目标端口范围（起始）：输入 514。
    • 目标端口范围至：输入 514。
  5. 点击提交。

创建管理合同

  1. 依次前往租户> 管理 > 合同 > 标准。
  2. 点击创建合同。
  3. 提供以下配置详细信息：
    • 名称：输入 mgmt-syslog-contract。
    • 范围：选择上下文。
  4. 点击提交。
  5. 展开合同，然后点击主题。
  6. 点击创建合同主题。
  7. 提供以下配置详细信息：
    • 名称：输入 syslog-subject。
    • 同时应用两个方向：选中此选项。
  8. 点击提交。
  9. 展开相应主题，然后点击过滤条件。
  10. 点击创建过滤器绑定。
  11. 选择之前创建的 syslog-udp-514 过滤条件。
  12. 点击提交。

配置 Syslog 目标群组

  1. 依次前往管理 > 外部数据收集器 > 监控目标 > Syslog。
  2. 右键点击 Syslog，然后选择创建 Syslog 监控目标组。
  3. 提供以下配置详细信息：
    • 名称：输入 Chronicle-Syslog-Group。
    • Admin State（管理员状态）：选择 Enabled（已启用）。
    • 格式：选择 aci。
  4. 点击下一步。
  5. 在创建 Syslog 监控目标对话框中执行以下操作：
    • 名称：输入 Chronicle-BindPlane。
    • 主机：输入 Bindplane 代理服务器的 IP 地址。
    • 端口：输入 514。
    • Admin State（管理员状态）：选择 Enabled（已启用）。
    • 严重程度：选择信息（以捕获详细日志）。
  6. 点击提交。

配置监控政策

Fabric 监控政策

  1. 依次前往 Fabric > Fabric Policies > Policies > Monitoring > Common Policy。
  2. 展开 Callhome/Smart Callhome/SNMP/Syslog/TACACS。
  3. 右键点击 Syslog，然后选择 Create Syslog Source。
  4. 提供以下配置详细信息：
    • 名称：输入 Chronicle-Fabric-Syslog。
    • 审核日志：选中此复选框可包含审核事件。
    • 事件：选中此复选框可包含系统事件。
    • 故障：选中此复选框可包含故障事件。
    • 会话日志：选中此复选框可包含会话日志。
    • 目标群组：选择 Chronicle-Syslog-Group。
  5. 点击提交。

访问监控政策

  1. 依次前往 Fabric > Access Policies > Policies > Monitoring > Default Policy。
  2. 展开 Callhome/Smart Callhome/SNMP/Syslog。
  3. 右键点击 Syslog，然后选择 Create Syslog Source。
  4. 提供以下配置详细信息：
    • 名称：输入 Chronicle-Access-Syslog。
    • 审核日志：选中此复选框可包含审核事件。
    • 事件：选中此复选框可包含系统事件。
    • 故障：选中此复选框可包含故障事件。
    • 会话日志：选中此复选框可包含会话日志。
    • 目标群组：选择 Chronicle-Syslog-Group。
  5. 点击提交。

配置系统 Syslog 消息政策

  1. 依次前往 Fabric > Fabric Policies > Policies > Monitoring > Common Policy。
  2. 展开 Syslog 消息政策。
  3. 点击默认。
  4. 在设施过滤条件部分中：
    • 设施：选择默认。
    • 最低严重程度：更改为信息。
  5. 点击提交。

选项 2：通过 AWS S3 提供的 JSON

此选项使用 APIC REST API 从 Cisco ACI 交换矩阵收集 JSON 格式的事件、故障和审核日志，并将它们存储在 AWS S3 中以供 SecOps 提取。

准备工作

  • Google SecOps 实例。
  • 对 Cisco APIC 控制台的特权访问权限。
  • 对 AWS（S3、IAM、Lambda、EventBridge）的特权访问权限。

收集 Cisco ACI APIC 前提条件（ID、API 密钥、组织 ID、令牌）

  1. 使用 HTTPS 登录 Cisco APIC 控制台。
  2. 前往管理 > AAA（在 APIC 6.0 及更高版本中）或管理 > 身份验证 > AAA（在旧版本中）。
    • 注意：从 APIC 6.0(1) 开始，AAA 菜单路径发生了变化。
  3. 创建或使用具有相应权限的现有本地用户。
  4. 复制以下详细信息并将其保存在安全的位置：
    • APIC 用户名：对监控数据具有读取权限的本地用户
    • APIC 密码：用户密码
    • APIC 网址：APIC 的 HTTPS 网址（例如 https://apic.example.com）

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

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

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

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

  3. 输入以下政策：

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

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

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

  6. 附加新创建的政策和 AWSLambdaBasicExecutionRole 管理式政策。

  7. 将角色命名为 cisco-aci-lambda-role，然后点击创建角色。

创建 Lambda 函数

  1. 在 AWS 控制台中，依次前往 Lambda > 函数 > 创建函数。
  2. 点击从头开始创作。

  3. 提供以下配置详细信息：

    • 名称：cisco-aci-events-collector
    • 运行时：Python 3.13
    • 架构：x86_64
    • 执行角色：cisco-aci-lambda-role

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

    import json
import boto3
import urllib3
import base64
from datetime import datetime, timedelta
import os
import logging

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

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

def lambda_handler(event, context):
    """
    AWS Lambda handler to fetch Cisco ACI events, faults, and audit logs and store them in S3
    """

    try:
        # Get environment variables
        s3_bucket = os.environ['S3_BUCKET']
        s3_prefix = os.environ['S3_PREFIX']
        state_key = os.environ['STATE_KEY']
        apic_url = os.environ['APIC_URL']
        apic_username = os.environ['APIC_USERNAME']
        apic_password = os.environ['APIC_PASSWORD']

        # Optional parameters
        page_size = int(os.environ.get('PAGE_SIZE', '100'))
        max_pages = int(os.environ.get('MAX_PAGES', '10'))

        logger.info(f"Starting Cisco ACI data collection for bucket: {s3_bucket}")

        # Get last run timestamp from state file
        last_timestamp = get_last_timestamp(s3_bucket, state_key)
        if not last_timestamp:
            last_timestamp = (datetime.utcnow() - timedelta(hours=1)).isoformat() + 'Z'

        # Authenticate to APIC
        session_token = authenticate_apic(apic_url, apic_username, apic_password)

        headers = {
            'Cookie': f'APIC-cookie={session_token}',
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        }

        # Data types to collect
        data_types = ['faultInst', 'eventRecord', 'aaaModLR']
        all_collected_data = []

        for data_type in data_types:
            logger.info(f"Collecting {data_type} data")
            collected_data = collect_aci_data(apic_url, headers, data_type, last_timestamp, page_size, max_pages)

            # Tag each record with its type
            for record in collected_data:
                record['_data_type'] = data_type

            all_collected_data.extend(collected_data)
            logger.info(f"Collected {len(collected_data)} {data_type} records")

        logger.info(f"Total records collected: {len(all_collected_data)}")

        # Store data in S3 if any were collected
        if all_collected_data:
            timestamp_str = datetime.utcnow().strftime('%Y%m%d_%H%M%S')
            s3_key = f"{s3_prefix}cisco_aci_events_{timestamp_str}.ndjson"

            # Convert to NDJSON format (one JSON object per line)
            ndjson_content = '\n'.join(json.dumps(record) for record in all_collected_data)

            # Upload to S3
            s3_client.put_object(
                Bucket=s3_bucket,
                Key=s3_key,
                Body=ndjson_content.encode('utf-8'),
                ContentType='application/x-ndjson'
            )

            logger.info(f"Uploaded {len(all_collected_data)} records to s3://{s3_bucket}/{s3_key}")

        # Update state file with latest timestamp from collected data
        latest_timestamp = get_latest_timestamp_from_records(all_collected_data)
        if not latest_timestamp:
            latest_timestamp = datetime.utcnow().isoformat() + 'Z'

        update_state(s3_bucket, state_key, latest_timestamp)

        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Success',
                'total_records_collected': len(all_collected_data),
                'data_types_collected': data_types
            })
        }

    except Exception as e:
        logger.error(f"Error in lambda_handler: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({
                'error': str(e)
            })
        }

def authenticate_apic(apic_url, username, password):
    """
    Authenticate to APIC and return session token
    """
    login_url = f"{apic_url}/api/aaaLogin.json"
    login_data = {
        "aaaUser": {
            "attributes": {
                "name": username,
                "pwd": password
            }
        }
    }

    response = http.request(
        'POST',
        login_url,
        body=json.dumps(login_data).encode('utf-8'),
        headers={'Content-Type': 'application/json'},
        timeout=30
    )

    if response.status != 200:
        raise RuntimeError(f"APIC authentication failed: {response.status} {response.data[:256]!r}")

    response_data = json.loads(response.data.decode('utf-8'))
    token = response_data['imdata'][0]['aaaLogin']['attributes']['token']
    logger.info("Successfully authenticated to APIC")
    return token

def collect_aci_data(apic_url, headers, data_type, last_timestamp, page_size, max_pages):
    """
    Collect data from APIC REST API with pagination
    """
    all_data = []
    page = 0

    while page < max_pages:
        # Build API URL with pagination and time filters
        api_url = f"{apic_url}/api/class/{data_type}.json"
        params = [
            f'page-size={page_size}',
            f'page={page}',
            f'order-by={data_type}.created|asc'
        ]

        # Add time filter for all data types to prevent duplicates
        time_attr = 'created'
        if last_timestamp:
            params.append(f'query-target-filter=gt({data_type}.{time_attr},"{last_timestamp}")')

        full_url = f"{api_url}?{'&'.join(params)}"

        logger.info(f"Fetching {data_type} page {page} from APIC")

        # Make API request
        response = http.request('GET', full_url, headers=headers, timeout=60)

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

        data = json.loads(response.data.decode('utf-8'))
        records = data.get('imdata', [])

        if not records:
            logger.info(f"No more {data_type} records found")
            break

        # Extract the actual data from APIC format
        extracted_records = []
        for record in records:
            if data_type in record:
                extracted_records.append(record[data_type])

        all_data.extend(extracted_records)
        page += 1

        # If we got less than page_size records, we've reached the end
        if len(records) < page_size:
            break

    return all_data

def get_last_timestamp(bucket, state_key):
    """
    Get the last run timestamp from S3 state file
    """
    try:
        response = s3_client.get_object(Bucket=bucket, Key=state_key)
        state_data = json.loads(response['Body'].read().decode('utf-8'))
        return state_data.get('last_timestamp')
    except s3_client.exceptions.NoSuchKey:
        logger.info("No state file found, starting from 1 hour ago")
        return None
    except Exception as e:
        logger.warning(f"Error reading state file: {str(e)}")
        return None

def get_latest_timestamp_from_records(records):
    """
    Get the latest timestamp from collected records to prevent missing events
    """
    if not records:
        return None

    latest = None
    latest_time = None

    for record in records:
        try:
            # Handle both direct attributes and nested structure
            attrs = record.get('attributes', record)
            created = attrs.get('created')
            modTs = attrs.get('modTs')  # Fallback for some object types

            timestamp = created or modTs
            if timestamp:
                if latest_time is None or timestamp > latest_time:
                    latest_time = timestamp
                    latest = record
        except Exception as e:
            logger.debug(f"Error parsing timestamp from record: {e}")
            continue

    return latest_time

def update_state(bucket, state_key, timestamp):
    """
    Update the state file with the current timestamp
    """
    try:
        state_data = {
            'last_timestamp': timestamp,
            'updated_at': datetime.utcnow().isoformat() + 'Z'
        }

        s3_client.put_object(
            Bucket=bucket,
            Key=state_key,
            Body=json.dumps(state_data).encode('utf-8'),
            ContentType='application/json'
        )

        logger.info(f"Updated state file with timestamp: {timestamp}")

    except Exception as e:
        logger.error(f"Error updating state file: {str(e)}")

  5. 依次前往配置 > 环境变量。

  6. 依次点击修改 > 添加新的环境变量。

  7. 输入以下提供的环境变量，并将其替换为您的值。

    • S3_BUCKET：cisco-aci-logs
    • S3_PREFIX：cisco-aci-events/
    • STATE_KEY：cisco-aci-events/state.json
    • APIC_URL: https://apic.example.com
    • APIC_USERNAME：<your-apic-username>
    • APIC_PASSWORD：<your-apic-password>
    • PAGE_SIZE：100（可选，用于控制分页大小）
    • MAX_PAGES：10（可选，限制每次运行提取的总页数）

  8. 创建函数后，请停留在其页面上（或依次打开 Lambda > 函数 > cisco-aci-events-collector）。

  9. 选择配置标签页。

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

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

创建 EventBridge 计划

  1. 依次前往 Amazon EventBridge > 调度器 > 创建调度。
  2. 提供以下配置详细信息：
    • 周期性安排：费率 (15 minutes)。
    • 目标：您的 Lambda 函数 cisco-aci-events-collector。
    • 名称：cisco-aci-events-collector-15m。
  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:::cisco-aci-logs/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::cisco-aci-logs"
    }
  ]
}

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

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

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

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

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

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

UDM 映射表

日志字段 UDM 映射 逻辑
@timestamp read_only_udm.metadata.event_timestamp 值取自原始日志字段“@timestamp”，并解析为时间戳。
aci_tag read_only_udm.metadata.product_log_id 值取自原始日志字段“aci_tag”。
cisco_timestamp - 未映射。
DIP read_only_udm.target.ip 值取自原始日志字段“DIP”。
DPort read_only_udm.target.port 值取自原始日志字段“DPort”，并转换为整数。
说明 read_only_udm.security_result.description 值取自原始日志字段“description”。
fault_cause read_only_udm.additional.fields.value.string_value 值取自原始日志字段“fault_cause”。该键设置为“故障原因”。
主机名 read_only_udm.principal.hostname 值取自原始日志字段“hostname”。
lifecycle_state read_only_udm.metadata.product_event_type 值取自原始日志字段“lifecycle_state”。
log.source.address - 未映射。
logstash.collect.host - 未映射。
logstash.collect.timestamp read_only_udm.metadata.collected_timestamp 值取自原始日志字段“logstash.collect.timestamp”，并解析为时间戳。
logstash.ingest.host read_only_udm.intermediary.hostname 值取自原始日志字段“logstash.ingest.host”。
logstash.irm_environment read_only_udm.additional.fields.value.string_value 值取自原始日志字段“logstash.irm_environment”。该键设置为“IRM_Environment”。
logstash.irm_region read_only_udm.additional.fields.value.string_value 值取自原始日志字段“logstash.irm_region”。该键设置为“IRM_Region”。
logstash.irm_site read_only_udm.additional.fields.value.string_value 值取自原始日志字段“logstash.irm_site”。该键设置为“IRM_Site”。
logstash.process.host read_only_udm.intermediary.hostname 值取自原始日志字段“logstash.process.host”。
消息 - 未映射。
message_class - 未映射。
message_code - 未映射。
message_content - 未映射。
message_dn - 未映射。
message_type read_only_udm.metadata.product_event_type 该值取自原始日志字段“message_type”，并移除了方括号。
node_link read_only_udm.principal.process.file.full_path 值取自原始日志字段“node_link”。
PktLen read_only_udm.network.received_bytes 值取自原始日志字段“PktLen”，并转换为无符号整数。
计划 - 未映射。
Proto read_only_udm.network.ip_protocol 值取自原始日志字段“Proto”，转换为整数，并映射到相应的 IP 协议名称（例如，6 -> TCP）。
SIP read_only_udm.principal.ip 值取自原始日志字段“SIP”。
SPort read_only_udm.principal.port 值取自原始日志字段“SPort”，并转换为整数。
syslog_facility - 未映射。
syslog_facility_code - 未映射。
syslog_host read_only_udm.principal.ip, read_only_udm.observer.ip 值取自原始日志字段“syslog_host”。
syslog_prog - 未映射。
syslog_severity read_only_udm.security_result.severity_details 值取自原始日志字段“syslog_severity”。
syslog_severity_code read_only_udm.security_result.severity 该值取自原始日志字段“syslog_severity_code”，并映射到相应的严重程度级别：5、6、7 -> INFORMATIONAL；3、4 -> MEDIUM；0、1、2 -> HIGH。
syslog5424_pri - 未映射。
Vlan-Id read_only_udm.principal.resource.id 值取自原始日志字段“Vlan-Id”。
- read_only_udm.metadata.event_type 逻辑：如果存在“SIP”或“hostname”，并且存在“Proto”，则设置为“NETWORK_CONNECTION”。否则，如果存在“SIP”“hostname”或“syslog_host”，则设置为“STATUS_UPDATE”。否则，设置为“GENERIC_EVENT”。
- read_only_udm.metadata.log_type 逻辑：设置为“CISCO_ACI”。
- read_only_udm.metadata.vendor_name 逻辑：设置为“Cisco”。
- read_only_udm.metadata.product_name 逻辑：设置为“ACI”。

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