如何配置EventBridge仅触发24小时内的AWS Inspector高危发现事件？

一、EventBridge原生规则无法实现动态时间筛选 #

你提到的问题是准确的：EventBridge的事件模式属于静态匹配规则，不支持动态计算当前时间并与firstObservedAt做对比，因此无法直接在EventBridge规则里过滤出过去24小时内的事件。

二、可行的解决方案 #

1. 在Lambda函数中添加时间校验与去重逻辑 #

这是最直接且易落地的方案：保持现有EventBridge规则不变，让所有符合CRITICAL+ACTIVE条件的事件触发Lambda，然后在Lambda内部完成两层校验：

• 校验firstObservedAt是否在过去24小时内
• 校验该finding是否已创建过工单（彻底避免重复）

示例Python代码片段：

import boto3
from datetime import datetime, timedelta, timezone
import os
from jira import JIRA

def lambda_handler(event, context):
    # 解析事件中的finding核心信息
    finding_detail = event['detail']
    finding_id = finding_detail['findingId']
    first_observed_str = finding_detail['firstObservedAt']
    
    # 1. 时间校验：判断是否在过去24小时内
    first_observed = datetime.fromisoformat(first_observed_str.replace('Z', '+00:00'))
    twenty_four_hours_ago = datetime.now(timezone.utc) - timedelta(hours=24)
    
    if first_observed < twenty_four_hours_ago:
        print(f"Finding {finding_id} 超过24小时，跳过处理")
        return
    
    # 2. 去重校验：用DynamoDB记录已处理的finding ID
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table(os.environ['PROCESSED_FINDINGS_TABLE'])
    
    response = table.get_item(Key={'findingId': finding_id})
    if 'Item' in response:
        print(f"Finding {finding_id} 已处理过，跳过创建工单")
        return
    
    # 3. 创建Jira工单逻辑
    jira = JIRA(server=os.environ['JIRA_SERVER'], basic_auth=(os.environ['JIRA_USER'], os.environ['JIRA_TOKEN']))
    issue_dict = {
        'project': {'key': os.environ['JIRA_PROJECT_KEY']},
        'summary': f"AWS Inspector CRITICAL风险: {finding_detail['title']}",
        'description': f"Finding ID: {finding_id}\n风险描述: {finding_detail['description']}\n关联资源: {finding_detail['resource']['resourceArn']}",
        'issuetype': {'name': 'Bug'},
    }
    new_issue = jira.create_issue(fields=issue_dict)
    
    # 4. 记录到DynamoDB，标记已处理
    table.put_item(Item={
        'findingId': finding_id, 
        'jiraIssueId': new_issue.id, 
        'processedAt': datetime.now(timezone.utc).isoformat()
    })
    
    print(f"成功创建Jira工单: {new_issue.key}")
    return

2. 定时Lambda+Inspector API查询（非实时触发） #

如果不需要实时处理风险，可改用定时触发的Lambda（比如每24小时运行一次），直接调用Inspector 2的ListFindings API，查询过去24小时内的CRITICAL+ACTIVE findings，再批量创建工单。这种方式完全绕过EventBridge的触发限制，也能天然避免重复（每次只查询指定时间范围的新发现）。

核心查询参数示例：

# 构建Inspector查询过滤器
filters = {
    'severity': [{'comparison': 'EQUALS', 'value': 'CRITICAL'}],
    'status': [{'comparison': 'EQUALS', 'value': 'ACTIVE'}],
    'firstObservedAt': [{'comparison': 'GREATER_THAN_OR_EQUALS', 'value': twenty_four_hours_ago.isoformat()}]
}

# 调用ListFindings API
inspector = boto3.client('inspector2')
response = inspector.list_findings(filterCriteria=filters, maxResults=100)

3. 用EventBridge Pipes做前置过滤 #

如果希望在事件到达业务Lambda前就过滤掉超时事件，可以使用EventBridge Pipes：

• 源选择现有EventBridge规则的事件流
• 添加一个Lambda过滤器作为中间环节，在过滤器Lambda中完成时间校验，只放行符合条件的事件到目标Lambda
• 目标Lambda专注于创建Jira工单和去重

这种方式和方案1类似，但把时间校验逻辑从业务Lambda中剥离，更符合单一职责原则。

三、额外去重建议 #

除了时间筛选，同一个Inspector finding可能会因为状态更新等原因多次触发EventBridge事件，因此必须结合finding ID的唯一性记录（比如DynamoDB、Redis）来确保不会重复创建工单，这一步是必不可少的。

内容的提问来源于stack exchange，提问作者Andy Thompson