Building Budget Alert Webhook Handlers

The specific bottleneck this page solves is that a naive @app.route("/webhook") receiver is not a webhook handler, it is a liability with a URL. Bolt one together for AWS Budgets over SNS, GCP budget alerts over Pub/Sub push, and Azure Monitor action-group webhooks, and it will fail in three predictable ways: it accepts an unauthenticated POST from anyone who guesses or leaks the endpoint URL, it processes the same event two or three times because every one of these providers uses at-least-once delivery, and it blocks the HTTP response on a downstream call to Slack or PagerDuty until the provider’s own timeout fires. Those timeouts are hard limits, not tuning suggestions: SNS retries a failed HTTP delivery three times immediately, then backs off over a window that can span more than a day before giving up; a Pub/Sub push subscription’s ack deadline defaults to 10 seconds and tops out at 600 even when extended; Azure Monitor’s webhook action calls the endpoint once with roughly a 10-second timeout and no built-in retry at all. Miss any of those windows and the provider either redelivers a duplicate or drops the alert on the floor. This page builds the receiving side of Budget Alert Automation with Webhooks: a Flask handler that verifies transport-level authenticity per provider, normalizes each payload shape, deduplicates by message identity, and returns 200 in milliseconds by handing everything else to a queue.

Root Cause & Failure Modes

The three providers do not share a security model, a delivery guarantee, or a payload shape, and a handler that ignores those differences breaks in ways that are invisible until an incident forces you to look:

  • No signature or token verification. AWS Budgets alerts arrive as SNS Notification messages signed with an RSA key you can validate against a certificate AWS hosts; GCP Pub/Sub push requests carry a Google-issued OIDC bearer token; Azure Monitor’s webhook action has no native signing at all. A handler that trusts the JSON body because it “looks right” will process a forged budget breach from anyone who discovers the URL — and if that handler is wired to an automated remediation action, a forged alert can trigger a real one.
  • No idempotency. All three providers are at-least-once, not exactly-once. SNS’s retry policy alone guarantees a duplicate delivery on any transient failure, and Pub/Sub redelivers any message that is not acked inside the deadline. A handler that posts to Slack or pages on-call on every POST fires the same budget breach two, three, sometimes a dozen times for one real event, and the team learns to ignore the channel.
  • Blocking on downstream calls. If the handler synchronously calls the Slack API (rate-limited per workspace) or PagerDuty’s Events API inside the request/response cycle, any latency spike pushes the response past the provider’s window — 10 seconds for Pub/Sub’s default ack deadline, roughly 10 seconds for Azure’s webhook action — and the provider marks the delivery failed and retries, which compounds the duplicate-processing problem on top of the blocking problem.
  • Losing events on crash. With no queue between “verified” and “processed,” a process restart or an unhandled exception mid-request drops whatever alert was in flight, permanently, the moment the provider’s own retry window expires.

The fix is not more defensive code sprinkled into one handler function — it is separating verification, normalization, and dedup (all synchronous, all fast) from everything that touches the network downstream (asynchronous, queued, retried on its own schedule).

Production Pipeline Architecture

The handler runs as four phases, each with a single responsibility, and each fast enough that the slowest step never approaches a provider’s delivery timeout.

  1. Verify. Check transport-level authenticity before touching the payload’s contents: validate the SNS signature against AWS’s certificate chain, validate the Pub/Sub bearer token’s OIDC claims and audience, or compare Azure’s shared token in constant time. A request that fails verification is rejected with 403 and never reaches normalization.
  2. Normalize. Parse the provider-specific JSON — AWS’s SNS envelope wrapping a budget notification, GCP’s base64-encoded Pub/Sub message data, Azure’s common alert schema — into one AlertEvent dataclass so everything downstream is provider-agnostic.
  3. Dedup. Derive a keyed hash from the provider and the message’s own identifier, check it against a seen-set, and short-circuit with 200 if it has already been queued. This is the same content-addressed idempotency approach used for streaming ingestion in Deduplicating Streamed Billing Events, applied to alert delivery instead of billing rows.
  4. Enqueue. Put the normalized event on an in-process queue (a durable broker in production) and return 200 immediately. A separate worker thread — or a separate process entirely — owns every downstream network call. The routing logic that drains that queue into Slack and PagerDuty, including per-destination rate limiting, is covered in the sibling page Routing Budget Alerts to Slack and PagerDuty.

The in-process queue.Queue shown below is sufficient for a single-instance deployment and is the right starting point for this page, since the goal here is separating verification from dispatch, not standing up a broker. It has two real limits worth planning around before traffic grows: it does not survive a process restart, so anything queued but not yet drained at deploy time is lost, and it does not coordinate across multiple running instances behind a load balancer, so horizontal scaling requires moving to a shared broker such as SQS, a Redis list, or a Pub/Sub topic. Neither limit affects the verify/normalize/dedup phases, which are stateless per request; only the enqueue step needs to change.

Four-phase budget alert webhook handler: verify, normalize, dedup, enqueue Three provider inputs converge into one handler pipeline. AWS Budgets over SNS, GCP budget alerts over Pub/Sub push, and Azure Monitor webhook each flow into a Verify stage that checks SNS signature, Pub/Sub OIDC token, or Azure shared token respectively. Verified requests continue to Normalize, which parses each provider's distinct payload shape into one AlertEvent dataclass. Normalize flows to Dedup, which hashes the provider and message id and drops anything already seen. Dedup flows to Enqueue, which places the event on a queue and returns HTTP 200 immediately. A separate labeled path shows a worker thread draining the queue asynchronously toward Slack and PagerDuty, decoupled from the request-response cycle. AWS Budgets (SNS) GCP (Pub/Sub push) Azure Monitor webhook 1 Verify signature / JWT / shared token 2 Normalize one AlertEvent dataclass 3 Dedup HMAC(provider, message_id) 4 Enqueue return 200 in milliseconds Worker thread drains the queue asynchronously Slack / PagerDuty dispatch happens off the request-response cycle
The request handler only ever runs phases 1–4, all local CPU work with one outbound HTTPS call to fetch a certificate or validate a token. The dashed path is the worker loop that owns every call to Slack or PagerDuty, decoupled from the provider's delivery timeout.

Step-by-Step Python Implementation

The module below verifies each provider’s transport-level authenticity, handles the AWS SNS subscription-confirmation handshake, normalizes into a shared AlertEvent, deduplicates with an HMAC-keyed hash, and enqueues to a worker thread that owns all downstream I/O.

"""Hardened budget-alert webhook handler for AWS Budgets (SNS), GCP budget
alerts (Pub/Sub push), and Azure Monitor action-group webhooks. Every route
verifies, normalizes, dedups, and enqueues -- nothing downstream runs inside
the request/response cycle."""

import base64
import hashlib
import hmac
import json
import logging
import os
import queue
import threading
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone

import requests
from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from flask import Flask, jsonify, request
from google.auth.transport import requests as google_auth_requests
from google.oauth2 import id_token

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger("budget_webhook")
app = Flask(__name__)

DEDUP_HMAC_SECRET = os.environ["DEDUP_HMAC_SECRET"].encode()
AZURE_SHARED_TOKEN = os.environ["AZURE_WEBHOOK_TOKEN"]
PUBSUB_AUDIENCE = os.environ["PUBSUB_PUSH_AUDIENCE"]       # push endpoint URL on the subscription
EXPECTED_SNS_TOPIC_ARN = os.environ["EXPECTED_SNS_TOPIC_ARN"]

alert_queue: "queue.Queue" = queue.Queue(maxsize=10_000)
_seen_lock = threading.Lock()
_seen_keys: set = set()  # swap for Redis SETNX + TTL in production


@dataclass(frozen=True)
class AlertEvent:
    provider: str            # "aws" | "gcp" | "azure"
    budget_name: str
    scope: str
    kind: str                 # "actual" | "forecast"
    amount: float
    threshold_pct: float
    raw_message_id: str
    received_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())


def verify_sns_signature(payload: dict) -> bool:
    """Validate an SNS message against AWS's documented canonical-string +
    RSA scheme, and reject any SigningCertURL not hosted on an SNS *.amazonaws.com
    domain -- otherwise a forged cert URL on a look-alike host would pass."""
    cert_url = payload.get("SigningCertURL", "")
    host = cert_url.split("/")[2] if "://" in cert_url else ""
    if not host.endswith(".amazonaws.com") or "sns." not in host:
        logger.warning("Rejected SNS message: untrusted cert host %s", host)
        return False

    fields = ["Message", "MessageId", "Subject", "Timestamp", "Token", "TopicArn", "Type", "SubscribeURL"]
    canonical = "".join(f"{f}\n{payload[f]}\n" for f in fields if f in payload)
    cert_pem = requests.get(cert_url, timeout=5).content
    public_key = x509.load_pem_x509_certificate(cert_pem).public_key()
    try:
        public_key.verify(
            base64.b64decode(payload["Signature"]),
            canonical.encode("utf-8"),
            padding.PKCS1v15(),
            hashes.SHA1() if payload.get("SignatureVersion", "1") == "1" else hashes.SHA256(),
        )
        return True
    except Exception:
        logger.warning("SNS signature verification failed for MessageId=%s", payload.get("MessageId"))
        return False


def verify_pubsub_jwt(auth_header: str) -> bool:
    """Verify the OIDC bearer token Pub/Sub attaches to every push request.
    The audience check matters: without it, a token minted for a different
    push endpoint would still pass signature verification."""
    if not auth_header.startswith("Bearer "):
        return False
    try:
        claims = id_token.verify_oauth2_token(
            auth_header.removeprefix("Bearer "), google_auth_requests.Request(), audience=PUBSUB_AUDIENCE
        )
        return bool(claims.get("email_verified", False))
    except ValueError as exc:
        logger.warning("Pub/Sub JWT verification failed: %s", exc)
        return False


def verify_azure_token(args) -> bool:
    """Azure Monitor's webhook action has no native request signing, so a
    shared secret travels as a query-string token and is compared in
    constant time to avoid a timing side-channel on the comparison itself."""
    return hmac.compare_digest(args.get("token", ""), AZURE_SHARED_TOKEN)


def dedup_key(provider: str, message_id: str) -> str:
    """HMAC-keyed hash rather than a plain digest, so a message id someone
    reads out of provider documentation cannot be used to forge a collision
    against the dedup store."""
    return hmac.new(DEDUP_HMAC_SECRET, f"{provider}:{message_id}".encode(), hashlib.sha256).hexdigest()


def already_seen(key: str) -> bool:
    with _seen_lock:
        if key in _seen_keys:
            return True
        _seen_keys.add(key)
        return False


def normalize_aws(message: dict) -> AlertEvent:
    return AlertEvent(
        provider="aws", budget_name=message.get("budgetName", "unknown"),
        scope=message.get("accountId", "unknown"),
        kind="forecast" if "FORECASTED" in message.get("notificationType", "") else "actual",
        amount=float(message.get("actualSpend", message.get("calculatedSpend", 0.0))),
        threshold_pct=float(message.get("thresholdPct", 100.0)),
        raw_message_id=message.get("MessageId", "unknown"),
    )


def normalize_gcp(envelope: dict) -> AlertEvent:
    data = json.loads(base64.b64decode(envelope["message"]["data"]))
    return AlertEvent(
        provider="gcp", budget_name=data.get("budgetDisplayName", "unknown"),
        scope=data.get("billingAccountId", "unknown"),
        kind="forecast" if data.get("forecastThresholdExceeded") else "actual",
        amount=float(data.get("costAmount", 0.0)),
        threshold_pct=float(data.get("alertThresholdExceeded", 1.0)) * 100,
        raw_message_id=envelope["message"]["messageId"],
    )


def normalize_azure(payload: dict) -> AlertEvent:
    ctx = payload.get("data", {}).get("context", payload)
    return AlertEvent(
        provider="azure", budget_name=ctx.get("name", "unknown"),
        scope=ctx.get("resourceGroupName", ctx.get("subscriptionId", "unknown")),
        kind="actual", amount=float(ctx.get("amount", 0.0)),
        threshold_pct=float(ctx.get("threshold", 100.0)),
        raw_message_id=payload.get("data", {}).get("essentials", {}).get("alertId", str(time.time())),
    )


@app.route("/webhooks/aws-budgets", methods=["POST"])
def aws_budgets():
    payload = request.get_json(force=True, silent=True) or {}
    if not verify_sns_signature(payload):
        return jsonify(error="signature verification failed"), 403
    if payload.get("TopicArn") != EXPECTED_SNS_TOPIC_ARN:
        logger.warning("Valid signature, unexpected TopicArn=%s", payload.get("TopicArn"))
        return jsonify(error="unexpected topic"), 403

    # A new subscription stays PendingConfirmation, and SNS never delivers a
    # real Notification, until this endpoint GETs SubscribeURL exactly once.
    if payload.get("Type") == "SubscriptionConfirmation":
        requests.get(payload["SubscribeURL"], timeout=5)
        logger.info("Confirmed SNS subscription for %s", payload.get("TopicArn"))
        return jsonify(status="subscribed"), 200
    if payload.get("Type") != "Notification":
        return jsonify(status="ignored"), 200

    key = dedup_key("aws", payload["MessageId"])
    if already_seen(key):
        return jsonify(status="duplicate"), 200

    message = json.loads(payload["Message"])
    message["MessageId"] = payload["MessageId"]
    alert_queue.put_nowait(normalize_aws(message))
    return jsonify(status="queued"), 200


@app.route("/webhooks/gcp-budgets", methods=["POST"])
def gcp_budgets():
    if not verify_pubsub_jwt(request.headers.get("Authorization", "")):
        return jsonify(error="jwt verification failed"), 403

    envelope = request.get_json(force=True, silent=True) or {}
    message_id = envelope.get("message", {}).get("messageId", "unknown")
    key = dedup_key("gcp", message_id)
    if already_seen(key):
        return jsonify(status="duplicate"), 200

    alert_queue.put_nowait(normalize_gcp(envelope))
    return jsonify(status="queued"), 200


@app.route("/webhooks/azure-budgets", methods=["POST"])
def azure_budgets():
    if not verify_azure_token(request.args):
        return jsonify(error="token verification failed"), 403

    payload = request.get_json(force=True, silent=True) or {}
    message_id = payload.get("data", {}).get("essentials", {}).get("alertId", "unknown")
    key = dedup_key("azure", message_id)
    if already_seen(key):
        return jsonify(status="duplicate"), 200

    alert_queue.put_nowait(normalize_azure(payload))
    return jsonify(status="queued"), 200


def _drain_queue_worker() -> None:
    """Owns every downstream network call, so the three routes above never
    block on Slack or PagerDuty inside the request/response cycle."""
    while True:
        event = alert_queue.get()
        logger.info("Dispatching %s alert: budget=%s scope=%s amount=%.2f",
                    event.provider, event.budget_name, event.scope, event.amount)
        # real implementation hands off to the routing logic in
        # Routing Budget Alerts to Slack and PagerDuty
        alert_queue.task_done()


if __name__ == "__main__":
    threading.Thread(target=_drain_queue_worker, daemon=True).start()
    app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 8080)))

Verification & Testing

Treat the handler’s guarantees as assertions, not assumptions:

  • Signature tampering. POST a captured SNS payload with one byte of Message flipped and assert verify_sns_signature returns False and the route responds 403. Do the same for a Bearer token signed for a different audience against verify_pubsub_jwt.
  • Idempotency assertion. POST the identical SNS Notification twice. The second call must return status: duplicate and alert_queue.qsize() must not have grown, proving the dedup key — not luck — suppressed it.
  • Subscription confirmation. Mock requests.get and POST a SubscriptionConfirmation payload; assert the mock was called once with the exact SubscribeURL and that no event reached the queue.
  • Response-time budget. Time a full request from POST to 200 in a load test and assert it stays under a few hundred milliseconds even with a cold _seen_keys set — anything approaching Pub/Sub’s 10-second ack deadline means a downstream call has leaked into the request path.

Common Pitfalls Checklist

  • Skipping the TopicArn check after signature verification. A validly-signed SNS message from an unrelated topic in the same AWS account still passes RSA verification — fix by asserting TopicArn equals the specific expected ARN, not just any trusted signer.
  • Forgetting the SNS subscription confirmation handshake. A subscription left in PendingConfirmation never delivers a real notification and fails silently with no application-visible error — fix by handling Type == "SubscriptionConfirmation" explicitly and GET-ing SubscribeURL.
  • Doing the Slack or PagerDuty call inside the route function. One slow API call pushes the response past the provider’s delivery window, which triggers a retry and doubles the duplicate-handling burden — fix by enqueuing and returning 200 immediately, as the worker thread does here.
  • Returning 200 even when normalization throws. Swallowing a parse failure with a blanket try/except hides a malformed payload from the provider’s own retry and dead-letter handling — fix by letting normalization errors propagate to a 500 so Pub/Sub or SNS retries (or dead-letters) it instead of silently losing it.
  • Relying on the in-process _seen_keys set past a single instance. It does not survive a restart and does not coordinate across multiple running workers — fix by backing dedup with Redis SETNX plus a TTL, or the same store used in Deduplicating Streamed Billing Events.

Frequently Asked Questions

Why does the handler enqueue instead of processing the alert inline?

Because every provider’s delivery guarantee is tied to a response-time window: Pub/Sub’s push ack deadline defaults to 10 seconds, and Azure Monitor’s webhook action times out in roughly the same window. A synchronous call to Slack or PagerDuty inside the request risks exceeding that window under any latency spike, which causes the provider to mark the delivery failed and retry — turning a slow response into a duplicate-processing problem on top of a latency problem. Enqueuing and returning 200 immediately keeps the acknowledgment fast regardless of how long the downstream dispatch takes.

What happens if the AWS SNS subscription confirmation step is skipped?

SNS keeps the subscription in PendingConfirmation indefinitely and never delivers a real Notification message to the endpoint — there is no error, no retry, just silence. The handler must detect Type: "SubscriptionConfirmation" on the first message SNS sends and issue a GET request to the SubscribeURL it contains exactly once to move the subscription to Confirmed.

Why verify the Pub/Sub JWT’s audience instead of just checking a shared secret?

A shared secret embedded in the URL or a header can leak through logs, proxies, or a misconfigured monitoring tool, and it does not prove the request actually came from Pub/Sub. Verifying the OIDC token’s signature and audience claim against the exact push endpoint URL proves both that Google issued the token and that it was minted specifically for this subscription, so a token replayed from a different push endpoint is rejected even if an attacker somehow obtained it.

How should the dedup store be implemented in production instead of the in-memory set shown here?

Back it with a store that has both atomic check-and-set semantics and a TTL — Redis SET key value NX EX <seconds> is the standard choice, since NX makes the insert conditional on the key not already existing and the TTL bounds how long a duplicate can still be caught without the store growing forever. This is the same pattern used for deduplicating streamed billing rows in Deduplicating Streamed Billing Events; the dedup key differs (message identity here, a billing-row content hash there) but the storage mechanics are identical.

Up: Budget Alert Automation with Webhooks · Home: Cloud Cost Optimization & FinOps Automation