signatureData = partnerId + "|" + transactionHash + "|" + amount + "|" + timestamp
signature     = HMAC-SHA256(apiSecret, signatureData)  → hex 소문자 64자

const crypto = require('crypto');

app.post('/webhook', (req, res) => {
  const p = req.body;                       // 평면 JSON (data 래퍼 없음)
  const secret = process.env.CRYPTOMENTS_API_SECRET;

  // 서명 대상: partnerId|transactionHash|amount|timestamp
  const data = [p.partnerId, p.transactionHash, p.amount, p.timestamp].join('|');
  const computed = crypto
    .createHmac('sha256', secret)
    .update(data)
    .digest('hex');              // ← hex (Base64 아님)

  // 타이밍 공격 방지 비교
  const ok = computed.length === (p.signature || '').length &&
    crypto.timingSafeEqual(Buffer.from(computed), Buffer.from(p.signature));

  if (!ok) {
    return res.status(401).json({ error: 'Signature mismatch' });
  }

  // eventType 으로 분기 (DEPOSIT_CONFIRMED, WITHDRAWAL_* ...)
  enqueue(p);
  res.status(200).json({ success: true });
});

import hmac
import hashlib
from flask import Flask, request

app = Flask(__name__)
API_SECRET = 'your_partner_api_secret'

@app.route('/webhook', methods=['POST'])
def webhook():
    p = request.get_json()          # 평면 JSON

    # transactionHash 는 없을 때 "" 로 오므로 그대로 사용한다
    data = '|'.join([
        p['partnerId'], p['transactionHash'],
        p['amount'], p['timestamp']
    ])
    computed = hmac.new(
        API_SECRET.encode(),
        data.encode(),
        hashlib.sha256
    ).hexdigest()          # ← hex

    if not hmac.compare_digest(computed, p.get('signature', '')):
        return {'error': 'Signature mismatch'}, 401

    return {'success': True}, 200

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.nio.charset.StandardCharsets;
import java.util.Map;

@PostMapping("/webhook")
public ResponseEntity<?> receiveWebhook(@RequestBody Map<String, Object> p) {
  try {
    String secret = System.getenv("CRYPTOMENTS_API_SECRET");

    String data = p.get("partnerId") + "|" + p.get("transactionHash")
                + "|" + p.get("amount") + "|" + p.get("timestamp");

    Mac mac = Mac.getInstance("HmacSHA256");
    mac.init(new SecretKeySpec(
      secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));

    // hex 소문자 인코딩 — Base64 아님
    StringBuilder hex = new StringBuilder();
    for (byte b : mac.doFinal(data.getBytes(StandardCharsets.UTF_8))) {
      hex.append(String.format("%02x", b));
    }

    if (!hex.toString().equals(p.get("signature"))) {
      return ResponseEntity.status(401).build();
    }

    return ResponseEntity.ok().build();
  } catch (Exception e) {
    return ResponseEntity.status(500).build();
  }
}

app.post('/webhook', async (req, res) => {
  const p = req.body;                        // 평면 JSON
  const key = `${p.eventType}:${p.transactionId}`;

  try {
    // 1. 서명 검증 (3장 참조)
    if (!verifySignature(p)) {
      return res.status(401).json({ error: 'Signature mismatch' });
    }

    // 2. 멱등 키 중복 확인
    const existing = await db.webhookEvents.findOne({ event_key: key });
    if (existing) {
      return res.status(200).json({
        success: true,
        message: 'Event already processed'
      });
    }

    // 3. 이벤트 처리 — 페이로드 최상위 필드를 그대로 사용
    await processEvent(p);

    // 4. 멱등 키 저장
    await db.webhookEvents.insertOne({
      event_key: key,
      event_type: p.eventType,
      transaction_id: p.transactionId,
      processed_at: new Date()
    });

    res.status(200).json({ success: true });
  } catch (error) {
    console.error(error);
    res.status(500).json({ error: 'Internal error' });
  }
});

const express = require('express');
const crypto = require('crypto');
const Bull = require('bull');

const app = express();
const webhookQueue = new Bull('webhooks');

app.use(express.json());

// Webhook 핸들러
app.post('/webhook', async (req, res) => {
  const signature = req.headers['x-signature'];
  const timestamp = req.headers['x-timestamp'];
  const eventId = req.headers['x-event-id'];
  const body = JSON.stringify(req.body);

  // 1. 서명 검증
  const secret = process.env.WEBHOOK_SECRET;
  const data = timestamp + '.' + body;
  const computed = crypto
    .createHmac('sha256', secret)
    .update(data)
    .digest('hex');

  if (computed !== signature) {
    return res.status(401).json({ error: 'Unauthorized' });
  }

  // 2. 중복 확인
  const exists = await db.webhookEvents.findOne({ event_id: eventId });
  if (exists) {
    return res.status(200).json({ success: true });
  }

  // 3. 큐에 추가
  await webhookQueue.add(req.body, {
    attempts: 3,
    backoff: { type: 'exponential', delay: 2000 }
  });

  // 4. 즉시 응답
  res.status(200).json({ success: true });
});

// 백그라운드 작업
webhookQueue.process(async (job) => {
  const { data } = job;
  const eventId = data.event_id;

  try {
    // 이벤트 타입별 처리
    switch (data.event_type) {
      case 'DEPOSIT_CONFIRMED':
        await handleDepositConfirmed(data.data);
        break;
      case 'WITHDRAWAL_CONFIRMED':
        await handleWithdrawalConfirmed(data.data);
        break;
      // ... 다른 이벤트 타입
    }

    // 이벤트 ID 저장
    await db.webhookEvents.insertOne({
      event_id: eventId,
      event_type: data.event_type,
      processed_at: new Date()
    });
  } catch (error) {
    console.error('Webhook processing error:', error);
    throw error; // 재시도 트리거
  }
});

app.listen(3000);

from flask import Flask, request, jsonify
from celery import Celery
import hmac
import hashlib
import json

app = Flask(__name__)
celery = Celery(app.name, broker='redis://localhost:6379')

WEBHOOK_SECRET = 'your_webhook_secret'

@app.route('/webhook', methods=['POST'])
def receive_webhook():
    signature = request.headers.get('X-Signature')
    timestamp = request.headers.get('X-Timestamp')
    event_id = request.headers.get('X-Event-Id')
    body = request.get_data(as_text=True)

    # 1. 서명 검증
    data = timestamp + '.' + body
    computed = hmac.new(
        WEBHOOK_SECRET.encode(),
        data.encode(),
        hashlib.sha256
    ).hexdigest()

    if computed != signature:
        return jsonify({'error': 'Unauthorized'}), 401

    # 2. 중복 확인
    if db.webhook_events.find_one({'event_id': event_id}):
        return jsonify({'success': True}), 200

    # 3. 비동기 처리
    webhook_data = json.loads(body)
    process_webhook.delay(webhook_data)

    # 4. 즉시 응답
    return jsonify({'success': True}), 200

@celery.task
def process_webhook(webhook_data):
    event_type = webhook_data['event_type']
    event_id = webhook_data['event_id']

    try:
        if event_type == 'DEPOSIT_CONFIRMED':
            handle_deposit_confirmed(webhook_data['data'])
        elif event_type == 'WITHDRAWAL_CONFIRMED':
            handle_withdrawal_confirmed(webhook_data['data'])

        # 이벤트 ID 저장
        db.webhook_events.insert_one({
            'event_id': event_id,
            'event_type': event_type,
            'processed_at': datetime.utcnow()
        })
    except Exception as e:
        print(f'Error processing webhook: {e}')
        raise

if __name__ == '__main__':
    app.run()