Python实现Zoom Webhook URL验证时返回未授权请求的问题排查

问题 #

搭建Zoom Webhook URL验证端点时，始终返回错误：Unauthorized request to Zoom Webhook sample.，以下是使用FastAPI编写的Python代码：

@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

尝试修改哈希编码后问题依旧，请求排查未授权错误原因。

排查思路与修复方案 #

出现Unauthorized错误的核心原因是签名验证不通过，以下是具体问题点和修复方法：

1. 错误使用解析后的JSON对象构造签名字符串（致命问题） #

代码中直接将解析后的Python字典body拼接到签名字符串中，这会导致字符串格式与Zoom发送的原始请求体完全不一致——Python字典的字符串表示用单引号，而原始JSON用双引号，再加上缩进、空格等差异，最终生成的哈希值必然和Zoom的签名不匹配。

修复方法：获取原始请求体字节数据，用其字符串形式构造签名：

# 替换原有的body = await request.json()
raw_body = await request.body()
body = json.loads(raw_body)  # 仍需解析JSON处理业务逻辑，但签名用原始字节

# 构造签名字符串时使用原始body的UTF-8编码字符串
message = f"v0:{timestamp}:{raw_body.decode('utf-8')}"

2. HTTP Header大小写兼容问题 #

HTTP协议中Header名称不区分大小写，但部分Web框架（如FastAPI）会将Header转换为小写形式，直接通过headers['x-zm-request-timestamp']或headers['x-zm-signature']取值可能抛出KeyError，导致签名逻辑中断。

修复方法：用get方法兼容不同大小写的Header：

timestamp = request.headers.get('x-zm-request-timestamp') or request.headers.get('X-Zm-Request-Timestamp')
zoom_signature = request.headers.get('x-zm-signature') or request.headers.get('X-Zm-Signature')

if not timestamp or not zoom_signature:
    return {'message': 'Missing required headers', 'status': 400}

3. 时间戳过期校验（可选但建议） #

Zoom的签名机制要求请求时间戳与当前时间偏差不能超过5分钟（防止重放攻击），若服务器时间与Zoom服务器时间偏差过大，也会导致验证失败。

修复方法：添加时间戳校验逻辑：

import time

current_timestamp = int(time.time())
if abs(current_timestamp - int(timestamp)) > 300:  # 5分钟=300秒
    return {'message': 'Timestamp expired', 'status': 401}

完整修复后的代码示例 #

import hmac
import hashlib
import json
import time
from fastapi import Request, APIRouter

router = APIRouter()
ZOOM_SECRET_TOKEN = "你的Webhook密钥"

@router.post('/webhook')
async def webhook(request: Request):
    # 获取关键Header，兼容大小写
    timestamp = request.headers.get('x-zm-request-timestamp') or request.headers.get('X-Zm-Request-Timestamp')
    zoom_signature = request.headers.get('x-zm-signature') or request.headers.get('X-Zm-Signature')
    
    if not timestamp or not zoom_signature:
        response = {'message': 'Missing required headers', 'status': 400}
        print(response['message'])
        return response
    
    # 校验时间戳是否过期
    current_timestamp = int(time.time())
    if abs(current_timestamp - int(timestamp)) > 300:
        response = {'message': 'Timestamp expired', 'status': 401}
        print(response['message'])
        return response
    
    # 获取原始请求体，用于签名验证
    raw_body = await request.body()
    body = json.loads(raw_body)
    
    print(dict(request.headers))
    print(body)
    
    # 构造签名字符串
    message = f"v0:{timestamp}:{raw_body.decode('utf-8')}"
    hash_for_verify = hmac.new(ZOOM_SECRET_TOKEN.encode(), message.encode(), hashlib.sha256).hexdigest()
    signature = f"v0={hash_for_verify}"

    # 验证签名
    if zoom_signature == signature:
        if body['event'] == 'endpoint.url_validation':
            hash_for_validate = hmac.new(ZOOM_SECRET_TOKEN.encode(), body['payload']['plainToken'].encode(), hashlib.sha256).hexdigest()
            response_msg = {
                'plainToken': body['payload']['plainToken'],
                'encryptedToken': hash_for_validate
            }
            print(response_msg)
            return response_msg
        else:
            response = {'message': 'Authorized request to Zoom Webhook sample.', 'status': 200}
            print(response['message'])
            return response
    else:
        response = {'message': 'Unauthorized request to Zoom Webhook sample.', 'status': 401}
        print(response['message'])
        return response

内容的提问来源于stack exchange，提问作者Huzdaifah