Through the Looking Glass: A Mad Tea-Party
This is post 3 in a seven-part series on what the Anchore Enterprise API makes possible for container security teams.
The fastest way to react to a new Critical CVE is to not have to ask. When a vulnerability is disclosed, a policy evaluation fails, or a container image finishes analysing, you want your tooling to respond on its own — the moment Anchore knows about it, not the next time your cron job wakes up. Anchore Enterprise’s webhook notifications make that loop fast and direct.
Anchore Events are the platform’s internal log of everything notable that happens — image analyses completing, policy evaluations changing, vulnerabilities being identified, system errors being logged — and webhook notifications are how you subscribe to the slice that matters to you. Events don’t arrive on your schedule — they arrive on theirs. The Mad Tea-Party in Wonderland is the canonical scene for this: Alice walks up to a table already in progress, where things happen when they happen and not before. Anchore’s notifications work the same way, and your job as the receiver is to be ready when the cup is passed.
Anchore Enterprise supports outbound webhooks that fire when key events occur. This makes it straightforward to build event-driven workflows that connect Anchore to the rest of your tooling — ticket systems, deployment pipelines, Slack channels, SIEMs, or anything else that accepts an HTTP request.
Setting the Table: Configuring a Webhook
Before any guest can sit down, the Hatter has to set out the cups. Webhooks in Anchore Enterprise are managed through the notifications API and work the same way: first register a webhook endpoint configuration (the cup), then attach a selector that defines which events should be delivered to it (who gets a seat at that end of the table).
Register your webhook endpoint configuration:
curl -s -u _api_key:<your-api-key> \
-X POST "https://wonderland.example.com/v2/notifications/endpoints/webhook/configurations" \
-H "Content-Type: application/json" \
-d '{
"name": "tea-party-receiver",
"url": "https://hatter-table.wonderland.example.com/anchore/webhook",
"verify_ssl": true
}'The response will include a uuid for the newly created configuration. Use that to attach a selector that maps policy evaluation events to this endpoint:
curl -s -u _api_key:<your-api-key> \
-X POST "https://wonderland.example.com/v2/notifications/endpoints/webhook/configurations/<uuid>/selectors" \
-H "Content-Type: application/json" \
-d '{
"scope": "account",
"event": {
"level": "*",
"resource_type": "image_digest",
"type": "user.image.policy_eval.*"
}
}'The event filter uses a structured <category>.<subcategory>.<event> format and supports wildcards, so user.image.policy_eval.* captures all policy evaluation outcomes for images in your account. You can retrieve the full list of supported event types from the /v2/event_types endpoint on your deployment. The scope field controls breadth — account limits events to your own account, while global (admin only) captures events across the entire system.
Anchore will POST to your registered URL whenever a matching event fires. Before putting it into production, you can verify your endpoint is reachable using the built-in test endpoint:
curl -s -u _api_key:<your-api-key> \
"https://wonderland.example.com/v2/notifications/endpoints/webhook/configurations/<uuid>/test"What’s in the Cup: Understanding the Payload
Before you write logic against the events, it’s worth looking at what’s actually in the cup. When Anchore fires a webhook, it POSTs a JSON payload to your registered URL with the following structure:
{
"id": "211c5fff5641456d935e60f905c46a66",
"type": "user.image.analysis_update",
"level": "info",
"message": "Image analysis complete",
"details": {},
"timestamp": "2025-10-01T09:08:54.749806",
"resource": {
"account_name": "mad-hatter-team",
"type": "image_digest",
"id": "sha256:<digest>"
},
"source": {
"request_id": null,
"service_name": "catalog",
"host_id": "anchore-enterprise-catalog-5654fb8d84-5ln8r",
"base_url": "http://anchore-enterprise-catalog.anchore.svc.cluster.local:8082"
}
}The key fields for building a receiver are type (the event that fired, matching the pattern you set in your selector), level (info, warn, or error), and resource.id (the identifier for the affected resource — for image events, this will be the image digest). The details object carries event-specific additional data that varies by event type.
Is Your Watch Right? Testing Your Receiver
The Hatter’s watch was two days wrong, even with butter in the works. Before wiring real logic to the events Anchore sends, prove your receiver is reachable and the payloads look the way you expect.
For a one-off smoke test, point your webhook configuration at webhook.site — it gives you a disposable URL and a live view of every request that arrives, no code required. That’s enough to confirm the event is firing and to eyeball the payload shape.
When you need to run logic against the events — for example to see exactly what details contains across multiple event types, since its content varies and isn’t fully enumerated in the documentation — stand up a minimal local logger. Here’s a Flask service that accepts any POST and logs the full payload:
import json
import logging
from flask import Flask, request, jsonify
app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
@app.route("/anchore/webhook", methods=["POST"])
def webhook():
payload = request.get_json(force=True)
logging.info(json.dumps(payload, indent=2))
return jsonify({"status": "received"}), 200
if __name__ == "__main__":
app.run(port=5000)Point your webhook configuration at this endpoint, trigger a few events, and inspect what arrives before writing production logic against it.
Pouring the Tea: Building a Webhook Receiver in Python
Once you know what’s in the cup, doing something with it is straightforward. The example below routes events by type, extracts the image digest from resource.id, and queries the API for the full vulnerability list:
import logging
import requests
from flask import Flask, request, jsonify
>app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
ANCHORE_URL = "https://wonderland.example.com/v2"
AUTH = ("_api_key", "<your-api-key>")
>def get_vulnerabilities(digest):
resp = requests.get(
f"{ANCHORE_URL}/images/{digest}/vuln/all",
auth=AUTH,
)
resp.raise_for_status()
return resp.json().get("vulnerabilities", [])
@app.route("/anchore/webhook", methods=["POST"])
def webhook():
payload = request.get_json(force=True)
event_type = payload.get("type", "")
level = payload.get("level", "")
resource = payload.get("resource", {})
digest = resource.get("id")
if not digest or resource.get("type") != "image_digest":
logging.info(f"Ignoring non-image event: {event_type}")
return jsonify({"status": "ignored"}), 200
logging.info(f"Event: {event_type} | Level: {level} | Digest: {digest}")
if "analysis_update" in event_type and level == "info":
vulns = get_vulnerabilities(digest)
critical = [v for v in vulns if v.get("severity") == "Critical"]
if critical:
logging.warning(
f"{len(critical)} Critical vulnerabilities in {digest}"
)
# open a ticket, page on-call, block promotion, etc.
else:
logging.info(
f"No Critical vulnerabilities in {digest} "
"— eligible for promotion"
)
# trigger promotion pipeline, notify Slack, update CMDB, etc.
return jsonify({"status": "received"}), 200
if __name__ == "__main__":
app.run(port=5000)The notification tells you something happened and which container image it happened to; the API call tells you what was found. The logic in the if critical branch is entirely yours — open a Jira ticket, post a formatted message to Slack, block a deployment pipeline, or trigger a rollback. Anchore fires the event and holds the data; your service decides what happens next.
“Move Down, Move Down!”: Transforming and Forwarding to a SIEM
Whenever the Hatter ran out of clean cups, everyone shifted one seat to the right and used the next person’s place. A natural extension of the webhook receiver follows the same idea: take the enriched data — the Anchore event combined with the vulnerability findings you’ve just fetched — and pass it along to a SIEM or other downstream system in a normalized format. This is the transform-and-forward pattern: receive the raw event, enrich it with context from the API, reshape it into the structure your tooling expects, and send it on.
The transformation step is where you decide what the downstream system needs to know. A SIEM doesn’t need the full Anchore vulnerability object for every finding — it typically needs a structured event with enough context to correlate, alert, and drive investigation. Here’s a function that builds that normalized event from the Anchore webhook payload and vulnerability data:
from collections import Counter
from datetime import datetime, timezone
def build_siem_event(payload, vulns):
resource = payload.get("resource", {})
severity_counts = Counter(
v.get("severity", "Unknown") for v in vulns
)
critical_findings = [
{
"cve": v.get("vuln"),
"severity": v.get("severity"),
"package": v.get("package"),
"package_version": v.get("package_version"),
"fix_available": v.get("fix") not in (None, "None"),
"fix_version": (
v.get("fix") if v.get("fix") not in (None, "None") else None
),
}
for v in vulns if v.get("severity") == "Critical"
]
return {
"event_id": payload.get("id"),
"event_type": payload.get("type"),
"level": payload.get("level"),
"timestamp": payload.get("timestamp"),
"forwarded_at": datetime.now(timezone.utc).isoformat(),
"image": {
"digest": resource.get("id"),
"account": resource.get("account_name"),
},
"vulnerability_summary": {
"total": len(vulns),
"critical": severity_counts.get("Critical", 0),
"high": severity_counts.get("High", 0),
"medium": severity_counts.get("Medium", 0),
"low": severity_counts.get("Low", 0),
"negligible": severity_counts.get("Negligible", 0),
},
"critical_findings": critical_findings,
"source": payload.get("source", {}),
}With a normalized event in hand, forwarding it is a straightforward HTTP POST to whatever endpoint your tooling exposes. The example below targets an Elasticsearch index using the _doc endpoint — a common SIEM ingestion pattern — but the same approach applies to Microsoft Sentinel, Datadog, or any platform that accepts JSON over HTTP:
import os
import requests
ES_URL = "https://looking-glass-search.wonderland.example.com:9200"
ES_INDEX = "anchore-events"
ES_API_KEY = os.environ["ELASTIC_API_KEY"]
def forward_to_siem(event):
resp = requests.post(
f"{ES_URL}/{ES_INDEX}/_doc",
json=event,
headers={"Authorization": f"ApiKey {ES_API_KEY}"},
timeout=5,
)
resp.raise_for_status()
logging.info(f"Forwarded event {event['event_id']} to SIEM")Bringing it all together, the updated webhook route chains the full pipeline — receive, enrich, transform, forward:
@app.route("/anchore/webhook", methods=["POST"])
def webhook():
payload = request.get_json(force=True)
resource = payload.get("resource", {})
digest = resource.get("id")
if not digest or resource.get("type") != "image_digest":
logging.info(
f"Ignoring non-image event: {payload.get('type')}"
)
return jsonify({"status": "ignored"}), 200
# Enrich: fetch vulnerability data from Anchore
vulns = get_vulnerabilities(digest)
# Transform: build a normalized SIEM event
event = build_siem_event(payload, vulns)
# Forward: send to SIEM
try:
forward_to_siem(event)
except Exception as e:
logging.error(f"Failed to forward event to SIEM: {e}")
return jsonify({"status": "received"}), 200The try/except around the forwarding call is intentional — you always want to return 200 to Anchore regardless of what happens downstream. If your SIEM endpoint is temporarily unavailable, that shouldn’t cause Anchore to retry the notification indefinitely.
This pattern — receive, enrich, transform, forward — is the foundation for connecting Anchore to virtually any downstream system. Swap out forward_to_siem for a function that opens a Jira ticket, posts to a Slack channel, or writes to an S3 bucket, and the rest of the pipeline stays the same.
Up Next
Anchore’s notifications API turns a system you query into a system that talks back. Combined with the SBOM and vulnerability data from the previous post, you have everything you need to build container security workflows that are both deeply informed and fully automated.
Next in the series: Humpty Dumpty — Custom Reporting and GraphQL via the API. Where REST gives you what each endpoint chose to hand back, GraphQL lets you ask for exactly the shape of data you want — “when I use a word,” said Humpty Dumpty, “it means just what I choose it to mean.” We’ll work through Anchore’s embedded GraphQL subsystem and build custom reports against it.
If you’re an Anchore Enterprise customer looking to build with the API, the Customer Success team is the fastest way to get unblocked — reach out through the Anchore Support Portal. If you’re not a customer yet but want to see what any of this looks like against your own container images, request a demo and we’ll walk you through it.
