FastAPI·Django에서 Python 웹훅 HMAC 검증하기

Python에서 Sume 웹훅 검증하기: timestamp.raw_body에 대한 HMAC-SHA256을 계산하고, 서명 헤더를 쉼표로 나눠 상수 시간으로 비교한 뒤 빠르게 응답하세요.

읽는 시간 6분Sume
전체 글

Python에서 Sume 웹훅을 검증하려면 원본 요청 본문을 바이트로 읽고, 서명 시크릿으로 <timestamp>.<raw_body>에 대한 HMAC-SHA256을 계산하세요. 오 분 넘게 어긋난 타임스탬프는 거부하고, x-sume-webhook-signature의 sume-v1= 항목 중 하나라도 상수 시간 비교에서 일치하면 전달을 수락합니다. FastAPI에서는 await request.body()로, Django에서는 request.body로 바이트를 얻습니다.

Sume 관련 내용은 Run 웹훅 (영문), 웹훅 (영문), 웹훅 검증, 쿡북에서, 프레임워크 관련 내용은 출처에 나열한 Python, Starlette, FastAPI, Django 문서에서 가져왔습니다. 모두 2026-09-27에 확인했습니다. Sume에는 FastAPI나 Django 전용 연동이 없고, SDK인 @sume-com/sdk는 TypeScript 클라이언트입니다. 그래서 아래 수신기는 모두 표준 라이브러리의 hmac과 hashlib으로 서명을 확인하는 일반 라우트입니다. TypeScript 수신기는 Sume 영상 실행용 서명된 웹훅에 있습니다.

Sume는 정확히 무엇에 서명하나요?

실행이든 생성 Job이든 모든 전달은 같은 방식과 같은 워크스페이스 시크릿을 쓰므로, 검증기 하나로 둘 다 처리할 수 있습니다.

Run 웹훅 (영문), 웹훅 (영문), 웹훅 검증 기준, 2026-09-27 확인.
요소값
알고리즘HMAC-SHA256, hex 인코딩
서명 대상 문자열<timestamp>.<raw_body>: 타임스탬프 헤더 값, 점, 원본 본문 순서
x-sume-webhook-timestamp초 단위 Unix 시간(예: 1785000000)
x-sume-webhook-signaturesume-v1=<hex>. 시크릿 교체 후 24시간 동안은 sume-v1=<new>,sume-v1=<old>
재전송 허용 시간이 범위를 벗어난 타임스탬프는 거부. 오 분이 무난한 기본값
시크릿대시보드 웹훅 탭, 또는 account:read로 호출하는 GET /v1/webhooks/signing-secret. SUME_COM_WEBHOOK_SIGNING_SECRET으로 저장

Python으로 검증기는 어떻게 작성하나요?

함수 하나로 두 프레임워크를 모두 처리합니다. 이 함수는 HMAC을 계산하기 전에 재전송 허용 시간을 벗어난 타임스탬프를 거부합니다. 그다음 헤더를 쉼표로 나누고 sume-v1= 항목만 남깁니다. 교체 기간에는 Sume가 유효한 시크릿마다 항목을 하나씩 최신순으로 보내므로, 헤더 전체를 비교하는 검증은 그 기간의 모든 전달에서 실패하기 때문입니다. sume-v1= 뒤의 hex 값은 hmac.compare_digest로 비교하는데, Python 문서는 이 함수가 타이밍 분석을 막도록 설계되었다고 설명합니다. 또 모든 항목을 끝까지 비교합니다.

import hashlib, hmac, os, time

SECRET = os.environ["SUME_COM_WEBHOOK_SIGNING_SECRET"].encode()
TOLERANCE_SECONDS = 300  # five minutes

def verify(raw: bytes, timestamp: str | None, header: str | None) -> bool:
    if not SECRET or not timestamp or not header:
        return False  # an empty secret would verify a forged signature
    try:
        ts = int(timestamp)
    except ValueError:
        return False
    if abs(time.time() - ts) > TOLERANCE_SECONDS:
        return False  # outside the replay window: no HMAC computed
    digest = hmac.new(SECRET, f"{ts}.".encode() + raw, hashlib.sha256).hexdigest()
    expected = digest.encode()
    matched = False
    for entry in header.split(","):  # one entry per live secret during a rotation
        version, _, value = entry.strip().partition("=")
        if version == "sume-v1" and hmac.compare_digest(value.encode(), expected):
            matched = True  # keep comparing the remaining entries
    return matched

FastAPI에서는 어떻게 받나요?

FastAPI의 Request는 Starlette에서 그대로 가져온 것으로, await request.body()는 본문을 바이트로 반환하고 헤더는 대소문자를 구분하지 않습니다. BackgroundTasks는 응답을 반환한 뒤 작업을 실행하므로 짧은 후속 작업에 맞습니다. 무거운 작업이라면 FastAPI는 메시지 큐를 쓰는 Celery 같은 더 큰 도구를 권합니다.

import json
from fastapi import BackgroundTasks, FastAPI, Request, Response

app = FastAPI()

@app.post("/hooks/sume")
async def sume_webhook(request: Request, background_tasks: BackgroundTasks):
    raw = await request.body()  # bytes, before any JSON parsing
    if not verify(raw, request.headers.get("x-sume-webhook-timestamp"),
                  request.headers.get("x-sume-webhook-signature")):
        return Response(status_code=401)
    event = json.loads(raw)
    if event.get("event") != "format.run.terminal":
        return Response(status_code=204)  # unknown event: 204, not 500
    if record_once(event["request_id"], raw):  # your insert-or-ignore
        background_tasks.add_task(handle, event)  # runs after the response
    return Response(status_code=204)

Django에서는 어떻게 받나요?

Django의 CSRF 미들웨어는 유효한 CSRF 토큰이 없는 POST에 403으로 응답하는데, Sume는 이 토큰을 보낼 수 없으므로 뷰에 @csrf_exempt를 붙이세요. HMAC 검증이 토큰의 역할을 대신합니다. request.body는 원본 바이트 문자열입니다. 다른 코드가 스트림을 읽기 전에 먼저 읽으세요. request.read() 뒤에 body에 접근하면 RawPostDataException이 발생하기 때문입니다. 크기는 문제가 되지 않습니다. Django의 기본 DATA_UPLOAD_MAX_MEMORY_SIZE는 2.5 MB로, Sume가 영수증을 인라인으로 담는 상한인 1 MiB를 넘습니다.

import json
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt

@csrf_exempt  # Sume sends no CSRF token; the HMAC authenticates the request
def sume_webhook(request):
    if request.method != "POST":
        return HttpResponse(status=405)
    raw = request.body  # raw bytestring, read before anything reads the stream
    if not verify(raw, request.headers.get("x-sume-webhook-timestamp"),
                  request.headers.get("x-sume-webhook-signature")):
        return HttpResponse(status=401)
    event = json.loads(raw)
    if event.get("event") == "format.run.terminal" and record_once(event["request_id"], raw):
        enqueue(event)  # your task queue; answer within Sume's 10 s
    return HttpResponse(status=204)

검증한 뒤 핸들러는 무엇을 해야 하나요?

전달 규칙은 어떤 언어에서든 같으며, Sume 영상 실행용 서명된 웹훅에 정리되어 있습니다. 요약하면 이벤트를 내구성 있게 기록하고 10초의 시도 시간 안에 2xx로 응답하세요. 응답하기 전에 영상을 렌더링하는 수신기는 작업하는 동안 재시도를 받으며, 시도는 총 최대 10회까지 이어지기 때문입니다. 실행은 재시도마다 반복되는 request_id로, 생성 Job은 job_id로 중복을 제거하세요. 1 MiB를 넘는 영수증이 payload: null로 도착하면 API 키로 error.result_url에서 가져오세요.

출처

관련 글

연동 카테고리의 다른 글

연동 글 전체 보기

작성자 Sume