Handling Canvas Live Events Webhooks in Python

Canvas Live Events push a stream of learning activity — grade changes, submissions, enrollments, asset access — to an HTTPS endpoint the moment each event occurs, delivered through Canvas Data Services as Caliper-shaped JSON. Standing up a receiver looks trivial until the details bite: the endpoint is public, so it must authenticate every request; delivery is at-least-once, so it must deduplicate; and a slow response earns a redelivery, so it must acknowledge fast and process later. This guide builds a minimal, correct receiver in Python that verifies the HMAC-SHA256 signature over the raw body, parses the Live Events envelope, stores each event idempotently by event_id, tokenizes identifiers at the FERPA boundary, and returns 200 immediately. It is the concrete implementation of the entity model and delivery semantics laid out in Webhook Event Ingestion.

Prerequisites

Confirm each of these before wiring up the receiver — the script assumes all of them are in place.

• Python 3.10+ (the code uses str | None unions and modern type hints).
• flask==3.* installed in the virtualenv (any WSGI framework works; Flask keeps the example short). The hmac and hashlib modules are standard library.
• A registered Canvas Live Events subscription through Canvas Data Services, delivering to your public HTTPS endpoint, with the shared signing secret exported as CANVAS_LIVE_EVENTS_SECRET.
• A separate EVENT_ID_SALT secret for tokenization, exported from your secrets manager — never hard-coded and never the same value as the signing secret.
• A publicly reachable HTTPS URL (behind a load balancer or tunnel in development); Canvas will not deliver to plain HTTP.
• Upstream payload shape: a JSON body with a metadata block (event_name, an event id, and a timestamp) and a body block carrying the event-specific fields, including a user_id and course_id. Canvas signs the raw request body with HMAC-SHA256 and sends the digest in a request header.

Step-by-step implementation

1. Read the raw request body before anything else. The signature is computed over the exact bytes Canvas sent, so you must capture request.get_data() and verify against that — re-serializing a parsed dict changes whitespace and key order and breaks the hash. Never call request.json first.

2. Verify the HMAC-SHA256 signature in constant time. Recompute hmac.new(secret, raw_body, sha256).hexdigest() and compare it to the header value with hmac.compare_digest, never ==. A constant-time compare denies a timing side channel that would otherwise leak the expected signature byte by byte.

3. Bound a replay window. Read the event timestamp from the metadata and reject anything older than a few minutes, so a valid request captured off the wire cannot be replayed later. Signature validity plus freshness is the full authenticity check.

4. Parse the Caliper / Live Events envelope only after verifying. With authenticity established, json.loads the body and pull event_name, the event id, occurred_at, and the event-specific fields out of the metadata/body structure into your flat envelope.

5. Deduplicate by event_id. Delivery is at-least-once, so check a durable uniqueness constraint (a unique index in production; a set here) before doing any work. A repeat returns 200 immediately with no side effect — acknowledging a duplicate is correct, because Canvas only stops retrying once it sees a 2xx.

6. Tokenize identifiers at the FERPA boundary. Replace user_id and any login_id with a salted SHA-256 surrogate before the envelope is stored or logged, following the cross-LMS student ID mapping tokenization contract, so no raw identifier is ever persisted.

7. Acknowledge fast, process asynchronously. Enqueue the tokenized envelope to a durable inbox and return 200 in milliseconds. The heavy work — recomputing grades, upserting the warehouse — runs in a separate worker draining the queue, exactly as async polling for grade syncs defers long jobs off the request thread.

Complete runnable code block

import hashlib
import hmac
import json
import os
import time
from datetime import datetime, timezone

from flask import Flask, request

app = Flask(__name__)

# Secrets injected from a manager; never hard-coded, and never the same value.
SIGNING_SECRET = os.environ["CANVAS_LIVE_EVENTS_SECRET"].encode()
ID_SALT = os.environ["EVENT_ID_SALT"].encode()
REPLAY_WINDOW_S = 300                       # reject events older than 5 minutes
PII_FIELDS = ("user_id", "login_id", "sis_user_id")

# Stand-in for a durable uniqueness constraint + queue in production.
_seen: set[str] = set()
_inbox: list[dict] = []


def tokenize(value: str | int) -> str:
    """Irreversible FERPA-safe surrogate; a stable 'stu_' token."""
    return "stu_" + hashlib.sha256(ID_SALT + str(value).encode()).hexdigest()[:24]


def signature_ok(raw_body: bytes, supplied: str | None) -> bool:
    """Constant-time HMAC-SHA256 check over the exact received bytes."""
    if not supplied:
        return False
    expected = hmac.new(SIGNING_SECRET, raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, supplied)


def is_fresh(occurred_at: str) -> bool:
    """Bound a replay window against the event's own timestamp."""
    try:
        ts = datetime.fromisoformat(occurred_at.replace("Z", "+00:00"))
    except (ValueError, AttributeError):
        return False
    age = (datetime.now(timezone.utc) - ts).total_seconds()
    return abs(age) <= REPLAY_WINDOW_S


def to_envelope(payload: dict) -> dict:
    """Flatten the Caliper/Live Events shape into a canonical envelope."""
    meta, body = payload.get("metadata", {}), payload.get("body", {})
    clean = {k: v for k, v in body.items() if k not in PII_FIELDS}
    for f in PII_FIELDS:
        if body.get(f) is not None:
            clean[f"{f}_token"] = tokenize(body[f])       # raw id never kept
    return {
        "event_id": meta["event_id"],
        "event_type": meta.get("event_name", "unknown"),
        "occurred_at": meta.get("event_time"),
        "received_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "partition_key": clean.get("course_id", "unknown"),
        "payload": clean,
    }


@app.post("/canvas/live-events")
def receive():
    raw = request.get_data()                              # RAW bytes, not parsed
    supplied_sig = request.headers.get("X-Canvas-Signature")
    if not signature_ok(raw, supplied_sig):
        return {"error": "invalid signature"}, 401        # never enters the inbox

    payload = json.loads(raw)                             # parse only AFTER verify
    envelope = to_envelope(payload)

    if not is_fresh(envelope["occurred_at"] or ""):
        return {"error": "stale event outside replay window"}, 401

    if envelope["event_id"] in _seen:
        return {"status": "duplicate"}, 200               # at-least-once absorbed

    _seen.add(envelope["event_id"])
    _inbox.append(envelope)                               # durable queue in prod
    return {"status": "enqueued"}, 200                    # ack fast; process async


if __name__ == "__main__":
    app.run(port=8080)

Verification and output validation

Exercise the receiver against a synthetic signed request before pointing Canvas at it:

• Valid signature accepted. Sign a sample body with the secret, POST it, and assert a 200 with {"status": "enqueued"}. Confirm the envelope in _inbox carries user_id_token and no raw user_id.
• Tampered body rejected. Flip one byte of the body after signing and assert a 401 — the recomputed HMAC must diverge.
• Duplicate collapses. Send the same signed request twice and assert the second returns {"status": "duplicate"} with len(_inbox) unchanged; at-least-once delivery must be a no-op.
• Stale event rejected. Set event_time to an hour ago and assert 401 — the replay window must reject it even with a valid signature.
• No PII leaked. Assert "user_id" not in _inbox[0]["payload"] and that every *_token value starts with stu_ and is 28 characters (stu_ plus 24 hex).

Troubleshooting

• Every request returns 401 despite a correct secret. You verified against parsed JSON, not the raw body. Capture request.get_data() and hash those bytes; re-serializing changes whitespace and key order and invalidates the HMAC.
• Signature check passes locally but fails in production. A proxy or WSGI layer re-encoded or gzip-decoded the body before your handler saw it. Verify against the exact bytes delivered, and disable body rewriting on the load balancer.
• Duplicate events reach the warehouse. The _seen set is per-process and resets on restart or across replicas. Back deduplication with a durable unique constraint on event_id in shared storage, not in-memory state.
• Valid recent events rejected as stale. Server clock skew pushes now - occurred_at past the window. Sync the host with NTP and widen REPLAY_WINDOW_S modestly; clock skew, not a slow network, is the usual cause.
• Canvas keeps redelivering the same event. Your handler is too slow and Canvas times out before the 200, treating it as failed. Move all processing behind the inbox and return 200 in milliseconds — a duplicate you already have should still ack 200, not error.
• Large payloads truncated or rejected. A body-size limit on the framework or proxy clips big events. Raise MAX_CONTENT_LENGTH and the upstream limit to comfortably exceed the largest Live Event, and reject cleanly rather than parsing a partial body.

Related

• Webhook Event Ingestion — the parent entity model, delivery semantics, and durable-inbox architecture this receiver implements.
• Async Polling for Grade Syncs — the pull counterpart that reconciles events this feed might drop.
• Incremental Sync State Management — the last-seen cursor that backstops webhook loss.
• Error Retry Logic for Sync Jobs — retry budgets and the dead-letter path for poison events the worker cannot process.

Part of: Webhook Event Ingestion