Add Mux webhook verification

[heff]

Passing on the request. Alex Steel may have more context if needed.

[sasp1]

Any update on this issue or time horizon for when this could be implemented?

[philcluff]

Hi @sasp1, we don't have a timeline on this at the moment, as this SDK is code generated, we have to work to do to create a pattern that works well for non-generated code in the same repository.

Thanks.

[ryan-alley]

It looks like this was added? Am I missing something?

https://docs.mux.com/guides/video/verify-webhook-signatures

[philcluff]

@ryan-alley we currently do not provide code in this package to verify webhook signatures, but the functionality is present on the platform.

Thanks!

[ryan-alley]

I'm a real goober - forgot I was just looking at the python package! It's been a long week. I'll be creating an authorizer for the webhooks in Python next month and can submit a pull request after I finished. ETA end-of-April. Will follow the above documentation to implement.

[pfcodes]

Bump

[Meess]

Anyone stumbling upon here looking for the way to handle this, independent of framework:

import hashlib
import hmac
from typing import Literal, TypedDict

class MuxWebhookValidationResponse(TypedDict):
    success: bool
    status: Literal[200, 401]
    message: str
    
def validate_webhook_signature(
    body: bytes,
    secret: str,
    mux_signature_header: str | None = None
) -> MuxWebhookValidationResponse:
    if not mux_signature_header:
        return {
            'success': False,
            'status': 401,
            'message': 'No Mux signature found in request headers',
        }

    # Parse the Mux-Signature header which looks like:
    # t=9285601944,v1=90b75d1180d735dd8d796d81123dfd5d932fd17df63d4d9871fd38da3a2c3e9
    try:
        signature_parts = dict(part.split('=') for part in mux_signature_header.split(','))
        signature = signature_parts.get('v1')
        t = signature_parts.get('t')
        if not signature or not t:
            return {
                'success': False,
                'status': 401,
                'message': 'Invalid Mux signature format',
            }
    except Exception:
        return {
            'success': False,
            'status': 401,
            'message': 'Failed to parse Mux signature',
        }

    payload = f'{t}.{body.decode("utf-8")}'

    # Create the expected signature
    expected_signature = hmac.new(
        key=secret.encode('utf-8'), msg=payload.encode('utf-8'), digestmod=hashlib.sha256
    ).hexdigest()

    # Compare signatures
    if not hmac.compare_digest(signature, expected_signature):
        return {
            'success': False,
            'status': 401,
            'message': 'Invalid Mux signature',
        }

    return {
        'success': True,
        'status': 200,
        'message': 'Valid Mux signature',
    }

If you're using Django with DRF, DRF already parses the orignal body which is difficult to retrieve afterwards. Hence I would suggest a django endpoint, and disable csrf for your webhook endpoint. E.g. for Django:

from rest_framework.response import Response
from django.views.decorators.csrf import csrf_exempt
from django.conf import settings
from django.http import HttpRequest
from .mux import validate_webhook_signature

@csrf_exempt
def mux_webhook(request: HttpRequest):
    valid = validate_webhook_signature(request.body, settings.MUX_WEBHOOK_SIGNING_SECRET, request.headers.get('Mux-Signature'))
    if not valid['success']:
        return Response({'error': valid['message']}, status=valid['status'])
       
       ... your code to handle the event ...

Or switch to Node, their node sdk has really good support with build in validation, however this python sdk is not in a ver usable state atm. But if you have to use python, then this is how to validate their signature for the webhook, or atleast sets you up with a working example.