Complete VTPE Integration Example

import express from "express";
import crypto from "crypto";

const app = express();

// Middleware to capture raw body for HMAC verification
app.use(express.json({
  verify: (req, res, buf) => {
    req.rawBody = buf;
  }
}));

// Auth middleware
function authMiddleware(req, res, next) {
  const auth = req.headers["authorization"];
  if (auth !== `Bearer ${process.env.VTPE_API_SECRET}`) {
    return res.status(401).json({ error: "UNAUTHORIZED", errorMessage: "Invalid or missing API Secret" });
  }
  next();
}

// HMAC signature verification middleware
function verifySignature(req, res, next) {
  const timestamp = req.headers["x-timestamp"];
  const signature = req.headers["x-signature"];

  if (!timestamp || !signature) {
    return res.status(401).json({ error: "UNAUTHORIZED", errorMessage: "Missing signature headers" });
  }

  // Validate timestamp window (5 minutes)
  const now = Math.floor(Date.now() / 1000);
  const ts = parseInt(timestamp, 10);
  if (Math.abs(now - ts) > 300) {
    return res.status(401).json({ error: "UNAUTHORIZED", errorMessage: "Timestamp outside acceptable window" });
  }

  const payload = req.rawBody.toString("utf8") + timestamp;
  const expected = crypto
    .createHmac("sha256", process.env.VTPE_HMAC_SECRET)
    .update(payload)
    .digest("hex");

  if (!crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected))) {
    return res.status(401).json({ error: "UNAUTHORIZED", errorMessage: "Invalid signature" });
  }

  next();
}

// Product Information API
app.get("/payments/:reference", authMiddleware, (req, res) => {
  const { reference } = req.params;
  const order = db.findOrder(reference);

  if (!order) {
    return res.status(404).json({ error: "NOT_FOUND", errorMessage: "Reference not found" });
  }
  if (order.isPaid) {
    return res.status(409).json({ error: "ALREADY_PAID", errorMessage: "Payment already completed" });
  }

  return res.json({
    reference: order.reference,
    total_amount: order.totalAmount,
    currency: order.currency,
    amount_detail: order.amountDetail ?? [],
    details: order.details ?? []
  });
});

// Webhook handler
app.post("/webhooks/vtpe", authMiddleware, verifySignature, async (req, res) => {
  const { event, data } = req.body;

  if (event === "payment.initialized") {
    await db.createPendingPayment({
      paymentId: data.paymentId,
      reference: data.reference,
      amount: data.amount,
      currency: data.currency,
      status: "pending"
    });
  }

  if (event === "payment.success") {
    const alreadyProcessed = await db.getPaymentByPaymentId(data.paymentId);
    if (alreadyProcessed) {
      return res.status(200).json({ success: true });
    }

    await db.fulfillOrder({
      reference: data.reference,
      paymentId: data.paymentId,
      paidAt: data.paidAt,
      channel: data.channel
    });
  }

  if (event === "payment.fail") {
    await db.cancelOrder({
      reference: data.reference,
      paymentId: data.paymentId
    });
  }

  return res.status(200).json({ success: true });
});

app.listen(3000, () => {
  console.log("VTPE integration server running on port 3000");
});

from flask import Flask, request, jsonify
import os
import hmac
import hashlib
import time

app = Flask(__name__)

VTPE_API_SECRET = os.environ["VTPE_API_SECRET"]
VTPE_HMAC_SECRET = os.environ["VTPE_HMAC_SECRET"]

# Auth middleware
@app.before_request
def check_auth():
    if request.endpoint in ("get_payment", "handle_webhook"):
        auth = request.headers.get("Authorization", "")
        if auth != f"Bearer {VTPE_API_SECRET}":
            return jsonify({"error": "UNAUTHORIZED", "errorMessage": "Invalid or missing API Secret"}), 401

# HMAC signature verification
def verify_signature():
    timestamp = request.headers.get("X-Timestamp")
    signature = request.headers.get("X-Signature")

    if not timestamp or not signature:
        return jsonify({"error": "UNAUTHORIZED", "errorMessage": "Missing signature headers"}), 401

    # Validate timestamp window (5 minutes)
    now = int(time.time())
    ts = int(timestamp)
    if abs(now - ts) > 300:
        return jsonify({"error": "UNAUTHORIZED", "errorMessage": "Timestamp outside acceptable window"}), 401

    payload = request.get_data(as_text=True) + timestamp
    expected = hmac.new(
        VTPE_HMAC_SECRET.encode("utf-8"),
        payload.encode("utf-8"),
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        return jsonify({"error": "UNAUTHORIZED", "errorMessage": "Invalid signature"}), 401

# Product Information API
@app.route("/payments/<reference>")
def get_payment(reference):
    order = db.find_order(reference)

    if not order:
        return jsonify({"error": "NOT_FOUND", "errorMessage": "Reference not found"}), 404
    if order.is_paid:
        return jsonify({"error": "ALREADY_PAID", "errorMessage": "Payment already completed"}), 409

    return jsonify({
        "reference": order.reference,
        "total_amount": order.total_amount,
        "currency": order.currency,
        "amount_detail": order.amount_detail or [],
        "details": order.details or []
    })

# Webhook handler
@app.route("/webhooks/vtpe", methods=["POST"])
def handle_webhook():
    sig_error = verify_signature()
    if sig_error:
        return sig_error

    payload = request.get_json()
    event = payload["event"]
    data = payload["data"]

    if event == "payment.initialized":
        db.create_pending_payment(
            payment_id=data["paymentId"],
            reference=data["reference"],
            amount=data["amount"],
            currency=data["currency"],
            status="pending"
        )

    if event == "payment.success":
        if db.get_payment_by_payment_id(data["paymentId"]):
            return jsonify({"success": True}), 200

        db.fulfill_order(
            reference=data["reference"],
            payment_id=data["paymentId"],
            paid_at=data["paidAt"],
            channel=data["channel"]
        )

    if event == "payment.fail":
        db.cancel_order(
            reference=data["reference"],
            payment_id=data["paymentId"]
        )

    return jsonify({"success": True}), 200

if __name__ == "__main__":
    app.run(port=3000)