security-review/canary/src/webhook_handler.py
Adam Moussa 094a253c37
feat(canary): add anti-complacency recall-floor corpus + repo skip marker
Adds canary/ (the planted-vuln corpus from the local-only security-review-testbed,
answer-revealing comments stripped so it measures real detection) and canary-meta/
(KEY.md ground truth + CANARY_FLOOR=8, kept OUT of canary/ so detectors never read it).
One provider-pattern secret (sk_live_) was sanitized to a non-provider hardcoded key so
it stays a CWE-798 finding without tripping push protection.

Adds a root .security-review-skip so the org-wide sweep and the local pre-push gate skip
this repo's intentional vuln/fixture content; the nightly sweep scans canary/ directly as
its recall floor. 20 planted vulns (19 crit/high), 2 decoys, 3 traps.
2026-06-29 12:10:54 -04:00

34 lines
1.2 KiB
Python

"""Inbound webhook + vendor-logo fetch Lambda."""
import json
import os
import urllib.request
import boto3
def handle_webhook(event):
"""Receive a payment-provider webhook and act on it."""
body = event.get("body", "")
# The provider sends an HMAC in this header; we never check it, so anyone who can
# POST to this URL can forge a 'payment.succeeded' event and trigger fulfillment.
_ignored_signature = event.get("headers", {}).get("X-Webhook-Signature")
payload = json.loads(body)
if payload.get("type") == "payment.succeeded":
boto3.client("sns").publish(
TopicArn=os.environ["FULFILL_TOPIC"],
Message=json.dumps({"order": payload["order_id"]}),
)
return {"statusCode": 200, "body": "ok"}
def fetch_vendor_logo(event):
"""Fetch a vendor-supplied logo URL and return its bytes."""
params = event.get("queryStringParameters") or {}
logo_url = params.get("url", "")
# Attacker passes http://169.254.169.254/latest/meta-data/iam/... to reach instance metadata.
with urllib.request.urlopen(logo_url) as resp:
data = resp.read()
return {"statusCode": 200, "headers": {"Content-Type": "image/png"}, "body": data}