Search code examples
pythonvalidationfastapiwebhookszoom-sdk

Zoom Webhook Validation using Python


I've make an endpoint for Zoom to validate the webhook URL, but this is the output Unauthorized request to Zoom Webhook sample.

this is my python code:

@router.post('/webhook')
async def webhook(request: Request):
    headers = dict(request.headers)
    body = await request.json()
    
    print(headers)
    print(body)
    # construct the message string
    message = f"v0:{headers['x-zm-request-timestamp']}:{body}"
    # hash the message string with your Webhook Secret Token and prepend the version semantic
    hash_for_verify = hmac.new(ZOOM_SECRET_TOKEN.encode(), message.encode(), hashlib.sha256).hexdigest()
    signature = f"v0={hash_for_verify}"

    # validating the request came from Zoom
    if headers['x-zm-signature'] == signature:

        # Zoom validating you control the webhook endpoint
        if body['event'] == 'endpoint.url_validation':
            hash_for_validate = hmac.new(ZOOM_SECRET_TOKEN.encode(), body['payload']['plainToken'].encode(), hashlib.sha256).hexdigest()

            response = {
                'message': {
                    'plainToken': body['payload']['plainToken'],
                    'encryptedToken': hash_for_validate
                },
                'status': 200
            }

            print(response['message'])
            return response['message']
        else:
            response = {'message': 'Authorized request to Zoom Webhook sample.', 'status': 200}

            print(response['message'])
            return response

            # business logic here, example make API request to Zoom or 3rd party

    else:
        response = {'message': 'Unauthorized request to Zoom Webhook sample.', 'status': 401}

        print(response['message'])
        return response

I've tried to change the hash encoding but same problem


Solution

  • this the final code. I've fix it by myself

    @router.post('/webhook')
    async def webhook(request: Request):
        headers = dict(request.headers)
        body = await request.json()
        print(headers)
        print(body)
    
        if 'payload' in body and 'plainToken' in body['payload']:
            secret_token = ZOOM_SECRET_TOKEN.encode("utf-8")
            plaintoken = body['payload']['plainToken']
            mess = plaintoken.encode("utf-8")
            has = hmac.new(secret_token, mess, hashlib.sha256).digest()
            hexmessage = has.hex()
    
            response = {
                'message': {
                    'plainToken': plaintoken,
                    'encryptedToken': hexmessage
                }
            }
            print(response['message'])
            return response['message']
        else:
            return {'error': 'Invalid payload'}