Unix timestamp (seconds since epoch) when Magic Hour sent the webhook.Example:
Copy
magic-hour-event-timestamp: 1729314984
What it’s for:
Prevents replay attacks by rejecting old webhooks
Recommended tolerance: Accept webhooks within 5 minutes of current time
If timestamp is too old or too far in the future, reject the webhook
Header Case: HTTP headers are case-insensitive. Magic-Hour-Event-Signature and
magic-hour-event-signature are equivalent.
Optional but Recommended: Signature verification is not required to receive webhooks, but it’s
strongly recommended for production to ensure authenticity.
1
Extract Headers and Payload
Get the signature, timestamp, and raw JSON payload from the incoming request:
Copy
signature = request.headers.get('magic-hour-event-signature')timestamp = request.headers.get('magic-hour-event-timestamp')raw_payload = await request.body() # Raw JSON bytes, not parsed
2
Extract the headers and payload
What to do: Get three pieces of information from the incoming webhook request:
Signature header: magic-hour-event-signature
Timestamp header: magic-hour-event-timestamp
Raw JSON body: The entire request body as a string (not parsed yet)
Code example:
Copy
signature = request.headers.get('magic-hour-event-signature')timestamp = request.headers.get('magic-hour-event-timestamp')raw_body = await request.body() # Raw bytes, not parsed JSON
Critical: You must use the raw request body exactly as received. Do NOT parse the JSON and
re-stringify it, as this can change formatting and break signature verification.
3
Create the signed payload
What to do: Combine the timestamp and raw payload into a single string for signature computation.Format: {timestamp}.{raw_json_payload}Example signed payload:
# Convert raw body to string if it's bytespayload_string = raw_body.decode('utf-8') if isinstance(raw_body, bytes) else raw_body# Create signed payloadsigned_payload = f"{timestamp}.{payload_string}"
Why: The signed payload is what Magic Hour used to create the signature. You need to recreate it exactly the same way.
4
Compute your signature
What to do: Generate an HMAC-SHA256 hash using:
Key: Your webhook secret (from Magic Hour Developer Hub)
Message: The signed payload you just created
Algorithm: HMAC-SHA256 (Hash-based Message Authentication Code with SHA-256)
Copy
import hmacimport hashlibimport os# Get your webhook secret from environment variablewebhook_secret = os.getenv("MAGIC_HOUR_WEBHOOK_SECRET")def compute_signature(signed_payload: str, secret: str) -> str: """ Compute HMAC-SHA256 signature Args: signed_payload: The timestamp.payload string secret: Your webhook secret from Magic Hour Returns: 64-character hexadecimal signature string """ return hmac.new( secret.encode('utf-8'), # Convert secret to bytes signed_payload.encode('utf-8'), # Convert payload to bytes hashlib.sha256 # Use SHA256 hash ).hexdigest() # Return as hex string# Usageexpected_signature = compute_signature(signed_payload, webhook_secret)
5
Verify the signature
What to do: Compare your computed signature with the signature Magic Hour sent in the header.Important: Use a constant-time comparison to prevent timing attacks:
Copy
import hmacimport time# Get the signature from headersreceived_signature = request.headers.get('magic-hour-event-signature')# Compare signatures (constant-time comparison for security)is_valid = hmac.compare_digest(expected_signature, received_signature)if not is_valid: # Signature doesn't match - reject the webhook raise HTTPException(status_code=401, detail="Invalid signature")
Why constant-time comparison: Regular string comparison (==) can leak timing information that attackers could exploit. hmac.compare_digest() prevents this.
6
Verify the timestamp
What to do: Check that the webhook was sent recently (within 5 minutes is recommended).Why: Prevents replay attacks where someone could resend an old valid webhook.
Copy
import time# Get current timecurrent_time = int(time.time())# Check if timestamp is within 5 minutes (300 seconds)timestamp_age = abs(current_time - int(timestamp))if timestamp_age > 300: # Timestamp too old or too far in future - reject raise HTTPException(status_code=401, detail="Timestamp too old")
Adjust tolerance: You can adjust the 300-second (5-minute) window based on your needs, but keep it reasonable for security.
from fastapi import FastAPI, Request, HTTPExceptionimport hmacimport hashlibimport timeimport jsonimport osapp = FastAPI()# Get webhook secret from environmentWEBHOOK_SECRET = os.getenv("MAGIC_HOUR_WEBHOOK_SECRET")def verify_webhook(signature: str, timestamp: str, raw_body: bytes) -> bool: """ Verify webhook signature and timestamp Args: signature: The magic-hour-event-signature header value timestamp: The magic-hour-event-timestamp header value raw_body: The raw request body as bytes Returns: True if webhook is valid, False otherwise """ # Step 1: Verify timestamp is recent (within 5 minutes) current_time = int(time.time()) timestamp_age = abs(current_time - int(timestamp)) if timestamp_age > 300: # 5 minutes in seconds print(f"❌ Timestamp too old: {timestamp_age} seconds") return False # Step 2: Create signed payload # Format: "{timestamp}.{raw_json_payload}" payload_string = raw_body.decode('utf-8') signed_payload = f"{timestamp}.{payload_string}" # Step 3: Compute expected signature expected_signature = hmac.new( WEBHOOK_SECRET.encode('utf-8'), # Secret as bytes signed_payload.encode('utf-8'), # Signed payload as bytes hashlib.sha256 # SHA256 hash function ).hexdigest() # Convert to hex string # Step 4: Compare signatures (constant-time comparison) is_valid = hmac.compare_digest(signature, expected_signature) if not is_valid: print(f"❌ Signature mismatch") print(f" Received: {signature}") print(f" Expected: {expected_signature}") return is_valid@app.post("/webhook")async def secure_webhook_handler(request: Request): """Secure webhook endpoint with signature verification""" # Extract headers signature = request.headers.get('magic-hour-event-signature') timestamp = request.headers.get('magic-hour-event-timestamp') # Validate headers exist if not signature or not timestamp: raise HTTPException( status_code=401, detail="Missing security headers" ) # Get raw body for signature verification raw_body = await request.body() # Verify webhook authenticity if not verify_webhook(signature, timestamp, raw_body): raise HTTPException( status_code=401, detail="Invalid signature or timestamp" ) # Now it's safe to parse and process the JSON event = json.loads(raw_body) print(f"✅ Verified webhook: {event['type']}") # Process the event match event['type']: case 'video.completed': print("Video is ready!") # Your business logic here case 'image.completed': print("Image is ready!") # Your business logic here case _: print(f"Received: {event['type']}") return {"success": True}# Run the serverif __name__ == "__main__": import uvicorn if not WEBHOOK_SECRET: raise ValueError("MAGIC_HOUR_WEBHOOK_SECRET must be set") uvicorn.run(app, host="0.0.0.0", port=8000)