我正在使用 SQS 发送消息,我正在本地测试以进行反复试验。
我在从下面复制的模拟消息正文中提取 json 密钥时遇到问题,我不确定为什么。当我运行处理程序时,它会运行下面的代码。
我正在尝试为此确定正确的格式。但我也在 JSON 验证器中对其进行了测试,一切都通过了。检索实际密钥(例如 policy_id)时,我下面的代码似乎有问题 我收到错误消息 KeyError: 'policy_type'.
有谁知道我如何从模拟 json 消息中提取密钥或我可以在我的模拟 json 消息中更改任何内容以使其工作?任何建议也可以提供帮助。
def parse_alert_message(sqs_message):
try:
alert = json.loads(sqs_message)
if alert['alertId'] == 'P-0':
return {'error': "Error", 'data': alert['alertId']}
parsed_alert = {
'policy_type' : alert['policyType'],
'policy_Id' : alert['policyId'],
'account' : {
'name' : alert['accountName'],
'account_id' : alert['accountId']
},
'region' : alert['resourceRegionId'],
'metadata' : alert['resource'],
'severity' : alert['severity']
}
except Exception as e:
return {'error': str(e), 'data': None}
def handler():
print(event['Records'])
这是测试 JSON 消息
"Records": [
{
"messageId": "059f36b4-87a3-44ab-83d2-661975830a7d",
"receiptHandle": "AQEBwJnKyrHigUMZj6rYigCgxlaS3SLy0a...",
"body": {"severity":"CRITICAL","alertTs": "timestamp", "data":"P-123", "resource":"resource","resourceId":"riskybucket4testing", "policyType": "config", "policyName": "PolicyNameTest", "policyId": "4daa435b-fa46-457a-9359-6a4b4a43a442", "alertRuleName":"JTB-SQS Alerts","alertRemediationImpact":"null","accountName":"jtb75-002-pan","hasFinding":"false","resourceRegionId":"us-east-1","alertRemediationCli":"null","source":"Prisma Cloud","cloudType":"aws", "accountId": "12345678", "alertId":"P-1000"},
"attributes": {
"ApproximateReceiveCount": "1",
"SentTimestamp": "1545082649183",
"SenderId": "AIDAIENQZJOLO23YVJ4VO",
"ApproximateFirstReceiveTimestamp": "1545082649185"
},
"messageAttributes": {},
"md5OfBody": "098f6bcd4621d373cade4e832627b4f6",
"eventSource": "aws:sqs",
"eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:my-queue",
"awsRegion": "us-east-2"
}
]
}