Developing Features — Security-First Production Code

Summary. Language and framework: Python 3.12, FastAPI, SQLAlchemy 2.0 sync, PostgreSQL. The endpoint verifies the HMAC over the raw body before it reads any field. It rejects stale timestamps, deduplicates repeated deliveries by event ID, and marks the order paid in the same transaction that records the event.

Pre-implementation check

What am I trusting? Nothing in the body or header until the HMAC matches. order_id is read only after verification.
2.

Blast radius. A forged event could mark unpaid orders as paid, which means goods ship for free. Signature verification is the only gate.
3.

Authoritative source. The provider's signing scheme as given: "<t>.<raw body>", HMAC-SHA256, 5-minute tolerance.
4.

Failure path. Any parsing or verification error returns 401 and is never treated as a pass. A missing secret raises an error and is not skipped.
5.

Who controls the value? The sender controls the header and body, so both are verified. The secret comes from the environment.
6. Platforms. No paths or subprocess calls. Timestamps are Unix seconds, so there is no timezone issue.

Reuse ladder. Stdlib hmac.compare_digest handles the timing-safe compare. FastAPI and Pydantic already validate the payload. PostgreSQL ON CONFLICT DO NOTHING deduplicates atomically. No new dependencies.

File plan

app/payments_webhook/
├── webhook_constants.py        constants
├── webhook_types.py            Pydantic payload, outcome enum, domain errors
├── webhook_models.py           ProcessedPaymentEvent ORM model
├── signature_validation.py     parse + verify the signature header (pure)
├── webhook_repository.py       data access (event dedupe, order update)
├── webhook_service.py          business rules
└── webhook_handler.py          FastAPI route (transport only)

Code

# app/payments_webhook/webhook_constants.py
SIGNATURE_HEADER = "X-Signature"
SIGNATURE_TOLERANCE_SECONDS = 300
MAX_BODY_BYTES = 65_536
SUPPORTED_EVENT_TYPE = "payment.succeeded"
ORDER_STATUS_PAID = "paid"
WEBHOOK_SECRET_ENV = "PAYMENTS_WEBHOOK_SECRET"

# app/payments_webhook/webhook_types.py
from enum import Enum

from pydantic import BaseModel, ConfigDict, Field

class PaymentEventData(BaseModel):
    model_config = ConfigDict(frozen=True)
    order_id: str = Field(min_length=1, max_length=64)

class PaymentEvent(BaseModel):
    model_config = ConfigDict(frozen=True)
    id: str = Field(min_length=1, max_length=128)
    type: str = Field(min_length=1, max_length=64)
    data: PaymentEventData

class WebhookOutcome(str, Enum):
    PROCESSED = "processed"
    DUPLICATE = "duplicate"
    IGNORED = "ignored"

class WebhookSignatureError(Exception):
    def __init__(self, reason: str) -> None:
        super().__init__(reason)
        self.reason = reason

class OrderNotFoundError(Exception):
    def __init__(self, order_id: str) -> None:
        super().__init__(order_id)
        self.order_id = order_id

# app/payments_webhook/webhook_models.py
from datetime import datetime

from sqlalchemy import DateTime, String, func
from sqlalchemy.orm import Mapped, mapped_column

from app.db import Base

class ProcessedPaymentEvent(Base):
    __tablename__ = "processed_payment_events"

    event_id: Mapped[str] = mapped_column(String(128), primary_key=True)
    received_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )

# app/payments_webhook/signature_validation.py
import hashlib
import hmac

from app.payments_webhook.webhook_constants import SIGNATURE_TOLERANCE_SECONDS
from app.payments_webhook.webhook_types import WebhookSignatureError

def parse_signature_header(header: str) -> tuple[int, str]:
    pairs = dict(
        item.strip().split("=", 1) for item in header.split(",") if "=" in item
    )
    try:
        return int(pairs["t"]), pairs["v1"]
    except (KeyError, ValueError) as exc:
        raise WebhookSignatureError("malformed_header") from exc

def verify_signature(raw_body: bytes, header: str, secret: str, now: int) -> None:
    timestamp, received = parse_signature_header(header)
    if abs(now - timestamp) > SIGNATURE_TOLERANCE_SECONDS:
        raise WebhookSignatureError("timestamp_outside_tolerance")
    signed_payload = f"{timestamp}.".encode() + raw_body
    expected = hmac.new(secret.encode(), signed_payload, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, received):
        raise WebhookSignatureError("signature_mismatch")

# app/payments_webhook/webhook_repository.py
from sqlalchemy import update
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import Session

from app.orders.models import Order
from app.payments_webhook.webhook_constants import ORDER_STATUS_PAID
from app.payments_webhook.webhook_models import ProcessedPaymentEvent

class PaymentWebhookRepository:
    def __init__(self, session: Session) -> None:
        self._session = session

    def record_event_if_new(self, event_id: str) -> bool:
        stmt = (
            insert(ProcessedPaymentEvent)
            .values(event_id=event_id)
            .on_conflict_do_nothing(index_elements=["event_id"])
        )
        return self._session.execute(stmt).rowcount == 1

    def mark_order_paid(self, order_id: str) -> bool:
        stmt = update(Order).where(Order.id == order_id).values(status=ORDER_STATUS_PAID)
        return self._session.execute(stmt).rowcount == 1

# app/payments_webhook/webhook_service.py
from typing import Protocol

from app.payments_webhook.webhook_constants import SUPPORTED_EVENT_TYPE
from app.payments_webhook.webhook_types import OrderNotFoundError, PaymentEvent, WebhookOutcome

class PaymentEventStore(Protocol):
    def record_event_if_new(self, event_id: str) -> bool: ...
    def mark_order_paid(self, order_id: str) -> bool: ...

class PaymentWebhookService:
    def __init__(self, store: PaymentEventStore) -> None:
        self._store = store

    def handle(self, event: PaymentEvent) -> WebhookOutcome:
        if event.type != SUPPORTED_EVENT_TYPE:
            return WebhookOutcome.IGNORED
        if not self._store.record_event_if_new(event.id):
            return WebhookOutcome.DUPLICATE
        if not self._store.mark_order_paid(event.data.order_id):
            raise OrderNotFoundError(event.data.order_id)
        return WebhookOutcome.PROCESSED

# app/payments_webhook/webhook_handler.py
import logging
import os
import time

from fastapi import APIRouter, Depends, Header, HTTPException, Request
from pydantic import ValidationError
from sqlalchemy.orm import Session

from app.db import get_session
from app.payments_webhook.signature_validation import verify_signature
from app.payments_webhook.webhook_constants import (
    MAX_BODY_BYTES, SIGNATURE_HEADER, WEBHOOK_SECRET_ENV,
)
from app.payments_webhook.webhook_repository import PaymentWebhookRepository
from app.payments_webhook.webhook_service import PaymentWebhookService
from app.payments_webhook.webhook_types import (
    OrderNotFoundError, PaymentEvent, WebhookSignatureError,
)

router = APIRouter()
logger = logging.getLogger(__name__)

async def read_raw_body(request: Request) -> bytes:
    body = await request.body()
    if len(body) > MAX_BODY_BYTES:
        raise HTTPException(status_code=413)
    return body

def get_webhook_secret() -> str:
    secret = os.environ.get(WEBHOOK_SECRET_ENV)
    if not secret:
        raise RuntimeError(f"{WEBHOOK_SECRET_ENV} is not set")
    return secret

@router.post("/v1/webhooks/payments")
def receive_payment_webhook(
    raw_body: bytes = Depends(read_raw_body),
    signature: str = Header(default="", alias=SIGNATURE_HEADER),
    secret: str = Depends(get_webhook_secret),
    session: Session = Depends(get_session),
) -> dict[str, str]:
    start = time.monotonic()
    try:
        verify_signature(raw_body, signature, secret, now=int(time.time()))
    except WebhookSignatureError as exc:
        logger.warning("payment_webhook.signature_invalid", extra={"reason": exc.reason})
        raise HTTPException(status_code=401) from exc

    try:
        event = PaymentEvent.model_validate_json(raw_body)
    except ValidationError as exc:
        logger.warning("payment_webhook.payload_invalid", extra={"error_count": exc.error_count()})
        raise HTTPException(status_code=400) from exc

    service = PaymentWebhookService(PaymentWebhookRepository(session))
    try:
        with session.begin():
            outcome = service.handle(event)
    except OrderNotFoundError as exc:
        logger.error(
            "payment_webhook.order_not_found",
            extra={"event_id": event.id, "order_id": exc.order_id},
        )
        raise HTTPException(status_code=404) from exc

    logger.info(
        "payment_webhook.handled",
        extra={
            "event_id": event.id,
            "order_id": event.data.order_id,
            "outcome": outcome.value,
            "duration_ms": int((time.monotonic() - start) * 1000),
        },
    )
    return {"status": outcome.value}

Migration (Alembic, same PR). Create processed_payment_events (event_id varchar(128) PRIMARY KEY, received_at timestamptz NOT NULL DEFAULT now()). The primary key is the unique index that dedupe relies on. orders.id is already the primary key, so the update needs no new index.

Tests

# tests/payments_webhook/test_signature_validation.py
import hashlib
import hmac

import pytest

from app.payments_webhook.signature_validation import verify_signature
from app.payments_webhook.webhook_types import WebhookSignatureError

SECRET = "whsec_test"
BODY = b'{"id":"evt_1","type":"payment.succeeded","data":{"order_id":"ord_1"}}'
NOW = 1_760_000_000

def sign(body: bytes, ts: int, secret: str = SECRET) -> str:
    mac = hmac.new(secret.encode(), f"{ts}.".encode() + body, hashlib.sha256).hexdigest()
    return f"t={ts},v1={mac}"

def test_should_accept_when_signature_matches():
    verify_signature(BODY, sign(BODY, NOW), SECRET, now=NOW)

def test_should_reject_when_body_is_tampered():
    with pytest.raises(WebhookSignatureError, match="signature_mismatch"):
        verify_signature(BODY.replace(b"ord_1", b"ord_2"), sign(BODY, NOW), SECRET, now=NOW)

def test_should_reject_when_timestamp_is_older_than_tolerance():
    with pytest.raises(WebhookSignatureError, match="timestamp_outside_tolerance"):
        verify_signature(BODY, sign(BODY, NOW - 301), SECRET, now=NOW)

@pytest.mark.parametrize("header", ["", "t=abc,v1=00", "v1=00", "garbage"])
def test_should_reject_when_header_is_malformed(header):
    with pytest.raises(WebhookSignatureError, match="malformed_header"):
        verify_signature(BODY, header, SECRET, now=NOW)

# tests/payments_webhook/test_webhook_service.py
import pytest

from app.payments_webhook.webhook_service import PaymentWebhookService
from app.payments_webhook.webhook_types import OrderNotFoundError, PaymentEvent, WebhookOutcome

class FakeStore:
    def __init__(self, existing_orders: set[str]) -> None:
        self.events: set[str] = set()
        self.orders = existing_orders
        self.paid: list[str] = []

    def record_event_if_new(self, event_id: str) -> bool:
        if event_id in self.events:
            return False
        self.events.add(event_id)
        return True

    def mark_order_paid(self, order_id: str) -> bool:
        if order_id not in self.orders:
            return False
        self.paid.append(order_id)
        return True

def event(event_type: str = "payment.succeeded", order_id: str = "ord_1") -> PaymentEvent:
    return PaymentEvent(id="evt_1", type=event_type, data={"order_id": order_id})

def test_should_mark_order_paid_when_event_is_new():
    store = FakeStore({"ord_1"})
    assert PaymentWebhookService(store).handle(event()) is WebhookOutcome.PROCESSED
    assert store.paid == ["ord_1"]

def test_should_not_pay_twice_when_event_is_redelivered():
    store = FakeStore({"ord_1"})
    service = PaymentWebhookService(store)
    service.handle(event())
    assert service.handle(event()) is WebhookOutcome.DUPLICATE
    assert store.paid == ["ord_1"]

def test_should_ignore_when_event_type_is_unsupported():
    store = FakeStore({"ord_1"})
    assert PaymentWebhookService(store).handle(event("payment.refunded")) is WebhookOutcome.IGNORED
    assert store.paid == []

def test_should_raise_when_order_does_not_exist():
    with pytest.raises(OrderNotFoundError):
        PaymentWebhookService(FakeStore(set())).handle(event(order_id="ord_missing"))

Validation still to run. None of this was executed here. Run ruff check . && ruff format --check . && mypy app/payments_webhook && pytest tests/payments_webhook. Add one integration test that goes through the route with a real PostgreSQL instance. ON CONFLICT and the rollback on OrderNotFoundError can't be tested with the fake store.

Decisions & trade-offs

Verify before parsing. The HMAC is checked on the untouched raw body, and Pydantic runs only after it passes. Parsing first would mean acting on unauthenticated data.

Dedupe and update in one transaction. If the order update fails, the event row rolls back too. The provider's retry can then succeed, for example when the order commit was racing the webhook.

Database-level dedupe. ON CONFLICT DO NOTHING on the primary key is atomic across workers. An in-memory or check-then-insert approach would let two simultaneous deliveries both pay.

404 for an unknown order. It returns non-2xx so the provider retries and the error log alerts someone. The alternative, returning 200 and dropping the event, would hide a real inconsistency.

• Unsupported types return 200 "ignored". The provider doesn't need to retry events we deliberately don't handle.