Signature Verification

import hashlib
import hmac
import time

WEBHOOK_SECRET = "your_webhook_secret"
TIMESTAMP_TOLERANCE = 300  # 5 minutes


def verify_webhook(headers: dict, body: bytes) -> bool:
    """Verify VIZOCHOK webhook signature.

    Args:
        headers: Request headers dict
        body: Raw request body bytes

    Returns:
        True if signature is valid

    Raises:
        ValueError: If signature is missing, expired, or invalid
    """
    sig_header = headers.get("x-vizochok-signature", "")
    timestamp = headers.get("x-vizochok-timestamp", "")

    # 1. Check signature format
    if not sig_header.startswith("sha256="):
        raise ValueError("Missing or malformed signature header")

    received_sig = sig_header[7:]  # Strip "sha256=" prefix

    # 2. Check timestamp (replay protection)
    if not timestamp:
        raise ValueError("Missing timestamp header")

    try:
        ts = int(timestamp)
    except ValueError:
        raise ValueError("Invalid timestamp format")

    if abs(time.time() - ts) > TIMESTAMP_TOLERANCE:
        raise ValueError("Webhook timestamp expired (possible replay attack)")

    # 3. Construct signed payload
    signed_payload = f"{timestamp}.".encode() + body

    # 4. Compute expected signature
    expected_sig = hmac.new(
        WEBHOOK_SECRET.encode(),
        signed_payload,
        hashlib.sha256,
    ).hexdigest()

    # 5. Constant-time comparison
    if not hmac.compare_digest(received_sig, expected_sig):
        raise ValueError("Invalid webhook signature")

    return True


# Usage with FastAPI
from fastapi import Request, HTTPException

async def verify_request(request: Request):
    body = await request.body()
    try:
        verify_webhook(dict(request.headers), body)
    except ValueError as e:
        raise HTTPException(status_code=401, detail=str(e))

const crypto = require('crypto');

const WEBHOOK_SECRET = 'your_webhook_secret';
const TIMESTAMP_TOLERANCE = 300; // 5 minutes

/**
 * Verify VIZOCHOK webhook signature.
 *
 * @param {Object} headers - Request headers
 * @param {Buffer} rawBody - Raw request body
 * @throws {Error} If verification fails
 */
function verifyWebhook(headers, rawBody) {
  const sigHeader = headers['x-vizochok-signature'] || '';
  const timestamp = headers['x-vizochok-timestamp'] || '';

  // 1. Check signature format
  if (!sigHeader.startsWith('sha256=')) {
    throw new Error('Missing or malformed signature header');
  }

  const receivedSig = sigHeader.slice(7);

  // 2. Check timestamp (replay protection)
  if (!timestamp) {
    throw new Error('Missing timestamp header');
  }

  const ts = parseInt(timestamp, 10);
  if (isNaN(ts)) {
    throw new Error('Invalid timestamp format');
  }

  if (Math.abs(Date.now() / 1000 - ts) > TIMESTAMP_TOLERANCE) {
    throw new Error('Webhook timestamp expired (possible replay attack)');
  }

  // 3. Construct signed payload
  const signedPayload = Buffer.concat([
    Buffer.from(`${timestamp}.`),
    rawBody,
  ]);

  // 4. Compute expected signature
  const expectedSig = crypto
    .createHmac('sha256', WEBHOOK_SECRET)
    .update(signedPayload)
    .digest('hex');

  // 5. Constant-time comparison
  if (!crypto.timingSafeEqual(
    Buffer.from(receivedSig, 'utf8'),
    Buffer.from(expectedSig, 'utf8'),
  )) {
    throw new Error('Invalid webhook signature');
  }
}

// Usage with Express
// Important: capture raw body before JSON parsing
app.use('/api/vizochok', express.json({
  verify: (req, _res, buf) => {
    req.rawBody = buf;
  },
}));

app.post('/api/vizochok/check-products', (req, res) => {
  try {
    verifyWebhook(req.headers, req.rawBody);
  } catch (err) {
    return res.status(401).json({ error: err.message });
  }
  // Handle request...
});

package webhook

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"io"
	"math"
	"net/http"
	"strconv"
	"time"
)

const (
	WebhookSecret      = "your_webhook_secret"
	TimestampTolerance = 300 // 5 minutes
)

// VerifyWebhook verifies the HMAC-SHA256 signature of a VIZOCHOK webhook.
func VerifyWebhook(r *http.Request) ([]byte, error) {
	// Read raw body
	body, err := io.ReadAll(r.Body)
	if err != nil {
		return nil, fmt.Errorf("failed to read body: %w", err)
	}

	sigHeader := r.Header.Get("X-VIZOCHOK-Signature")
	timestamp := r.Header.Get("X-VIZOCHOK-Timestamp")

	// 1. Check signature format
	if len(sigHeader) < 8 || sigHeader[:7] != "sha256=" {
		return nil, fmt.Errorf("missing or malformed signature header")
	}
	receivedSig := sigHeader[7:]

	// 2. Check timestamp (replay protection)
	if timestamp == "" {
		return nil, fmt.Errorf("missing timestamp header")
	}

	ts, err := strconv.ParseInt(timestamp, 10, 64)
	if err != nil {
		return nil, fmt.Errorf("invalid timestamp format")
	}

	diff := math.Abs(float64(time.Now().Unix() - ts))
	if diff > TimestampTolerance {
		return nil, fmt.Errorf("webhook timestamp expired (possible replay attack)")
	}

	// 3. Construct signed payload
	signedPayload := []byte(fmt.Sprintf("%s.", timestamp))
	signedPayload = append(signedPayload, body...)

	// 4. Compute expected signature
	mac := hmac.New(sha256.New, []byte(WebhookSecret))
	mac.Write(signedPayload)
	expectedSig := hex.EncodeToString(mac.Sum(nil))

	// 5. Constant-time comparison
	if !hmac.Equal([]byte(receivedSig), []byte(expectedSig)) {
		return nil, fmt.Errorf("invalid webhook signature")
	}

	return body, nil
}

// Usage in an HTTP handler
func HandleWebhook(w http.ResponseWriter, r *http.Request) {
	body, err := VerifyWebhook(r)
	if err != nil {
		http.Error(w, err.Error(), http.StatusUnauthorized)
		return
	}
	// Use body for processing...
	_ = body
}