Webhooki

curl -X POST https://app.ksefshop.pl/v1/webhook_endpoints \
  -H "Authorization: Bearer $KSEF_API_KEY" \
  -H "Idempotency-Key: webhook-shop-backend" \
  -H "Content-Type: application/json" \
  -d '{
  "url": "https://example.com/ksef/webhook",
  "enabled_events": [
    "*"
  ],
  "description": "Shop backend"
}'

<?php
$payload = [
    'url' => 'https://example.com/ksef/webhook',
    'enabled_events' => ['*'],
    'description' => 'Shop backend',
];

$ch = curl_init('https://app.ksefshop.pl/v1/webhook_endpoints');
curl_setopt_array($ch, [
    CURLOPT_CUSTOMREQUEST => 'POST',
    CURLOPT_RETURNTRANSFER => true,
    CURLOPT_HTTPHEADER => [
        'Authorization: Bearer ' . getenv('KSEF_API_KEY'),
        'Idempotency-Key: webhook-shop-backend',
        'Content-Type: application/json',
    ],
    CURLOPT_POSTFIELDS => json_encode($payload, JSON_UNESCAPED_UNICODE),
]);
$body = curl_exec($ch);
$status = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
curl_close($ch);

$data = json_decode($body, true);
if ($status >= 400) {
    throw new RuntimeException($data['error']['code'] . ': ' . $data['error']['message']);
}
echo var_export($data['id'], true), ' ', var_export($data['secret'], true), PHP_EOL;

const res = await fetch("https://app.ksefshop.pl/v1/webhook_endpoints", {
  method: "POST",
  headers: {
    Authorization: `Bearer ${process.env.KSEF_API_KEY}`,
    "Idempotency-Key": "webhook-shop-backend",
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    url: "https://example.com/ksef/webhook",
    enabled_events: ["*"],
    description: "Shop backend",
  }),
});
const data = await res.json();
if (!res.ok) throw new Error(`${data.error.code}: ${data.error.message}`);
console.log(data.id, data.secret);

import os
import requests

resp = requests.post(
    "https://app.ksefshop.pl/v1/webhook_endpoints",
    headers={
        "Authorization": f"Bearer {os.environ['KSEF_API_KEY']}",
        "Idempotency-Key": "webhook-shop-backend",
    },
    json={
        "url": "https://example.com/ksef/webhook",
        "enabled_events": ["*"],
        "description": "Shop backend",
    },
    timeout=30,
)
data = resp.json()
if not resp.ok:
    raise RuntimeError(f"{data['error']['code']}: {data['error']['message']}")
print(data["id"], data["secret"])

# Wyślij do swojego odbiorcy testowe zdarzenie podpisane tak jak nasze
SECRET=whsec_shownOnlyOnceExampleValue0000
BODY='{"id":"evt_test","object":"event","type":"invoice.accepted","livemode":false,"created":1760865214,"data":{"object":{"id":"inv_test"}}}'
T=$(date +%s)
SIG=$(printf '%s.%s' "$T" "$BODY" | openssl dgst -sha256 -hmac "$SECRET" | sed 's/^.* //')
curl -X POST http://localhost:8000/ksef/webhook \
  -H "Content-Type: application/json" \
  -H "KSeF-Signature: t=$T,v1=$SIG" \
  -H "KSeF-Event-Id: evt_test" \
  -d "$BODY"

<?php
$secret = getenv('KSEF_WEBHOOK_SECRET'); // whsec_...
$payload = file_get_contents('php://input'); // surowe body, przed json_decode
$header = $_SERVER['HTTP_KSEF_SIGNATURE'] ?? '';

$parts = [];
foreach (explode(',', $header) as $item) {
    [$k, $v] = array_pad(explode('=', $item, 2), 2, '');
    $parts[$k] = $v;
}
$t = (int) ($parts['t'] ?? 0);
if ($t === 0 || abs(time() - $t) > 300 || empty($parts['v1'])) {
    http_response_code(400);
    exit;
}
$expected = hash_hmac('sha256', $t . '.' . $payload, $secret);
if (!hash_equals($expected, $parts['v1'])) {
    http_response_code(400);
    exit;
}

$event = json_decode($payload, true);
// Odrzuć duplikat po $_SERVER['HTTP_KSEF_EVENT_ID'], zapisz zdarzenie, resztę zrób w tle
http_response_code(200);

import crypto from "node:crypto";
import express from "express";

const app = express();
const secret = process.env.KSEF_WEBHOOK_SECRET; // whsec_...

// express.raw: podpis liczymy z surowego body, nie z przetworzonego JSON
app.post("/ksef/webhook", express.raw({ type: "application/json" }), (req, res) => {
  const parts = Object.fromEntries(
    (req.get("KSeF-Signature") ?? "").split(",").map((p) => p.split("=", 2)),
  );
  const t = Number(parts.t);
  if (!t || Math.abs(Date.now() / 1000 - t) > 300 || !parts.v1) return res.sendStatus(400);

  const expected = crypto.createHmac("sha256", secret).update(`${t}.${req.body}`).digest("hex");
  const a = Buffer.from(expected);
  const b = Buffer.from(parts.v1);
  if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) return res.sendStatus(400);

  const event = JSON.parse(req.body);
  // Odrzuć duplikat po req.get("KSeF-Event-Id"), zapisz zdarzenie, resztę zrób w tle
  console.log(event.type, event.data.object.id);
  res.sendStatus(200);
});

app.listen(8000);

import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["KSEF_WEBHOOK_SECRET"].encode()  # whsec_...


@app.post("/ksef/webhook")
def ksef_webhook():
    payload = request.get_data()  # surowe bajty body
    header = request.headers.get("KSeF-Signature", "")
    parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)
    try:
        t = int(parts.get("t", "0"))
    except ValueError:
        abort(400)
    if not t or abs(time.time() - t) > 300 or "v1" not in parts:
        abort(400)
    expected = hmac.new(SECRET, f"{t}.".encode() + payload, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, parts["v1"]):
        abort(400)

    event = request.get_json()
    # Odrzuć duplikat po KSeF-Event-Id, zapisz zdarzenie, resztę zrób w tle
    print(event["type"], event["data"]["object"]["id"])
    return "", 200

curl "https://app.ksefshop.pl/v1/events?limit=10" \
  -H "Authorization: Bearer $KSEF_API_KEY"

<?php
$ch = curl_init('https://app.ksefshop.pl/v1/events?limit=10');
curl_setopt_array($ch, [
    CURLOPT_RETURNTRANSFER => true,
    CURLOPT_HTTPHEADER => [
        'Authorization: Bearer ' . getenv('KSEF_API_KEY'),
    ],
]);
$body = curl_exec($ch);
$status = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
curl_close($ch);

$data = json_decode($body, true);
if ($status >= 400) {
    throw new RuntimeException($data['error']['code'] . ': ' . $data['error']['message']);
}
echo count($data['data']), ' obiektów', PHP_EOL;

const res = await fetch("https://app.ksefshop.pl/v1/events?limit=10", {
  headers: {
    Authorization: `Bearer ${process.env.KSEF_API_KEY}`,
  },
});
const data = await res.json();
if (!res.ok) throw new Error(`${data.error.code}: ${data.error.message}`);
console.log(data.data.length, data.has_more);

import os
import requests

resp = requests.get(
    "https://app.ksefshop.pl/v1/events?limit=10",
    headers={
        "Authorization": f"Bearer {os.environ['KSEF_API_KEY']}",
    },
    timeout=30,
)
data = resp.json()
if not resp.ok:
    raise RuntimeError(f"{data['error']['code']}: {data['error']['message']}")
print(len(data["data"]), data["has_more"])