Chiquvchi webhook'larni yuborish: qayta urinish, DLQ va SSRF himoyasi

CREATE TABLE webhook_obunalari (
  id               bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  foydalanuvchi_id bigint NOT NULL
    REFERENCES foydalanuvchilar (id) ON DELETE CASCADE,
  url       text NOT NULL
    CHECK (url ~ '^https?://' AND char_length(url) <= 2000),
  sir       text NOT NULL,   -- whsec_<base64>, bir marta ko'rsatiladi
  hodisalar text[] NOT NULL DEFAULT '{vazifa.yaratildi,
    vazifa.yangilandi,vazifa.bajarildi,vazifa.ochirildi}',
  faol            boolean     NOT NULL DEFAULT true,
  ketma_ket_xato  int         NOT NULL DEFAULT 0,
  yaratilgan_vaqt timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE webhook_yetkazishlar (
  id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  obuna_id        bigint NOT NULL
    REFERENCES webhook_obunalari (id) ON DELETE CASCADE,
  hodisa_id       text   NOT NULL,  -- webhook-id sarlavhasi
  tur             text   NOT NULL,
  tana            jsonb  NOT NULL,
  holat text NOT NULL DEFAULT 'navbatda' CHECK (holat IN
    ('navbatda', 'yuborilmoqda', 'yuborildi', 'tashlandi')),
  urinishlar      int         NOT NULL DEFAULT 0,
  keyingi_urinish timestamptz NOT NULL DEFAULT now(),
  oxirgi_status   int,
  oxirgi_xato     text,
  yaratilgan_vaqt timestamptz NOT NULL DEFAULT now(),
  yuborilgan_vaqt timestamptz,
  CONSTRAINT webhook_yetkazishlar_hodisa_key
    UNIQUE (obuna_id, hodisa_id)
);
CREATE INDEX webhook_yetkazishlar_navbat_idx
  ON webhook_yetkazishlar (keyingi_urinish)
  WHERE holat = 'navbatda';

/**
 * Hodisani shu foydalanuvchining tegishli obunalariga navbatga
 * qo'yadi. Bitta hodisa — bitta `webhook-id` (hamma obunaga bir xil).
 */
export async function hodisaQosh(
  pool: Pool,
  foydalanuvchiId: number,
  tur: string,
  malumot: unknown,
  hodisaId: string = randomUUID(),
): Promise<number> {
  const tana = {
    type: tur,
    timestamp: new Date().toISOString(),
    data: malumot,
  };
  const r = await pool.query(
    `INSERT INTO webhook_yetkazishlar (obuna_id, hodisa_id, tur, tana)
     SELECT id, $3, $2, $4 FROM webhook_obunalari
      WHERE foydalanuvchi_id = $1 AND faol AND $2 = ANY(hodisalar)
     ON CONFLICT (obuna_id, hodisa_id) DO NOTHING`,
    [foydalanuvchiId, tur, hodisaId, tana],
  );
  return r.rowCount ?? 0;
}

UPDATE webhook_yetkazishlar y
   SET holat = 'yuborilmoqda', urinishlar = urinishlar + 1,
       keyingi_urinish = now() + make_interval(secs => $2)
  FROM webhook_obunalari o
 WHERE y.id IN (
         SELECT id FROM webhook_yetkazishlar
          WHERE holat IN ('navbatda', 'yuborilmoqda')
            AND keyingi_urinish <= now()
          ORDER BY keyingi_urinish
          LIMIT $1 FOR UPDATE SKIP LOCKED)
   AND o.id = y.obuna_id
RETURNING y.id, y.obuna_id, y.hodisa_id, y.tana,
          y.urinishlar, o.url, o.sir

const javob = await undiciFetch(u, {
  method: "POST",
  headers: {
    "content-type": "application/json",
    "user-agent": "vazifalar-webhook/1",
    ...sarlavhalar(q.sir, q.hodisa_id, tana),
  },
  body: tana,
  redirect: "manual",
  dispatcher: o.agent,
  signal: AbortSignal.timeout(kutish),
});

const JADVAL_SONIYA = [
  5, 30, 120, 600, 1800, 7200, 21600, 43200, 43200,
];

function kechikishMs(urinish, tasodif = Math.random) {
  const s = JADVAL_SONIYA[urinish - 1];
  if (s === undefined) return undefined;
  return Math.round(s * 1000 * (0.8 + 0.4 * tasodif()));
}

console.log(kechikishMs(1, () => 0), kechikishMs(1, () => 1));
console.log(kechikishMs(4, () => 0.5));
console.log(kechikishMs(10));
const jami = JADVAL_SONIYA.reduce((a, b) => a + b, 0);
console.log(jami, "s =", (jami / 3600).toFixed(1), "soat");

4000 6000
600000
undefined
117755 s = 32.7 soat

function urlTekshir(url) {
  let u;
  try {
    u = new URL(url);
  } catch {
    throw new Error("URL noto'g'ri");
  }
  if (u.protocol !== "https:") throw new Error("faqat https");
  if (u.username || u.password) throw new Error("login/parol");
  const port = u.port ? Number(u.port) : 443;
  if (port !== 443 && port !== 80) throw new Error(`port ${port}`);
  return u.hostname;
}

for (const s of ["https://hooks.example.uz/v", "http://example.uz/h",
  "https://a:b@example.uz/h", "https://example.uz:6379/",
  "https://0x7f.1/", "https://2130706433/"]) {
  try {
    console.log("ok:", urlTekshir(s));
  } catch (e) {
    console.log("rad:", e.message);
  }
}

ok: hooks.example.uz
rad: faqat https
rad: login/parol
rad: port 6379
ok: 127.0.0.1
ok: 127.0.0.1

import { BlockList, isIP } from "node:net";

const taqiq = new BlockList();
taqiq.addSubnet("127.0.0.0", 8, "ipv4");
taqiq.addSubnet("10.0.0.0", 8, "ipv4");
taqiq.addSubnet("169.254.0.0", 16, "ipv4");
taqiq.addSubnet("192.168.0.0", 16, "ipv4");
taqiq.addSubnet("100.64.0.0", 10, "ipv4");
taqiq.addSubnet("::1", 128, "ipv6");
taqiq.addSubnet("fc00::", 7, "ipv6");

function ommaviyIpmi(ip) {
  const m = /^::ffff:(\d+\.\d+\.\d+\.\d+)$/i.exec(ip);
  if (m) return !taqiq.check(m[1], "ipv4");
  const tur = isIP(ip);
  if (tur === 0) return false;
  return !taqiq.check(ip, tur === 4 ? "ipv4" : "ipv6");
}

for (const ip of ["8.8.8.8", "127.0.0.1", "169.254.169.254",
  "100.64.0.1", "::1", "fd00::1", "::ffff:127.0.0.1", "salom"]) {
  console.log(ip.padEnd(17), ommaviyIpmi(ip));
}

8.8.8.8           true
127.0.0.1         false
169.254.169.254   false
100.64.0.1        false
::1               false
fd00::1           false
::ffff:127.0.0.1  false
salom             false

/**
 * undici Agent: har ulanishda DNS javobidagi barcha manzillar ommaviy
 * bo'lishi shart. `ruxsat` — test/lokal uchun istisno (masalan
 * 127.0.0.1 mock).
 */
export function xavfsizAgent(
  o: { ruxsat?: (ip: string) => boolean; ulanishMs?: number } = {},
): Agent {
  return new Agent({
    connect: {
      timeout: o.ulanishMs ?? 5000,
      lookup: (host: string, opts: LookupOptions, cb: LookupCb) => {
        dnsLookup(
          host,
          { ...opts, all: true },
          (err, manzillar) => {
            if (err) {
              cb(err, []);
              return;
            }
            const yomon = manzillar.find(
              (a) =>
                !ommaviyIpmi(a.address) &&
                !(o.ruxsat?.(a.address) ?? false),
            );
            if (yomon) {
              cb(
                new SsrfXato(
                  `taqiqlangan manzil: ${host} → ${yomon.address}`,
                ),
                [],
              );
              return;
            }
            if (opts.all) cb(null, manzillar);
            else {
              const b = manzillar[0];
              if (!b) {
                cb(new SsrfXato(`manzil yo'q: ${host}`), []);
                return;
              }
              cb(null, b.address, b.family);
            }
          },
        );
      },
    },
  });
}

// chiquvchi.ts — chiquvchi webhook'lar (sql/203): hodisa → har
// obunaga yetkazish qatori → worker yuboradi (Standard Webhooks
// imzo), qayta urinish
// (eksponensial + jitter), 410 → obuna o'chadi, jadval tugasa →
// 'tashlandi' (DLQ), SSRF himoyasi (undici Agent, ulanish paytida IP
// tekshiruvi).
import { randomUUID } from "node:crypto";
import type { Pool } from "pg";
import { fetch as undiciFetch, type Agent } from "undici";
import { sarlavhalar } from "./imzo.ts";
import { urlTekshir } from "./ssrf.ts";

/**
 * Urinishlar orasidagi kechikish (soniya): Svix/Standard Webhooks
 * tavsiyasiga yaqin — jami ~ 1.5 sutka; keyin DLQ.
 */
export const JADVAL_SONIYA = [
  5, 30, 120, 600, 1800, 7200, 21600, 43200, 43200,
] as const;
export const OBUNA_OCHIRISH_CHEGARASI = 20;

/**
 * ±20 % jitter — bir vaqtda yiqilgan minglab yetkazish bir vaqtda
 * qaytib urilmasin (thundering herd).
 */
export function kechikishMs(
  urinish: number,
  tasodif: () => number = Math.random,
): number | undefined {
  const s = JADVAL_SONIYA[urinish - 1];
  if (s === undefined) return undefined;
  return Math.round(s * 1000 * (0.8 + 0.4 * tasodif()));
}

/**
 * Hodisani shu foydalanuvchining tegishli obunalariga navbatga
 * qo'yadi. Bitta hodisa — bitta `webhook-id` (hamma obunaga bir xil).
 */
export async function hodisaQosh(
  pool: Pool,
  foydalanuvchiId: number,
  tur: string,
  malumot: unknown,
  hodisaId: string = randomUUID(),
): Promise<number> {
  const tana = {
    type: tur,
    timestamp: new Date().toISOString(),
    data: malumot,
  };
  const r = await pool.query(
    `INSERT INTO webhook_yetkazishlar (obuna_id, hodisa_id, tur, tana)
     SELECT id, $3, $2, $4 FROM webhook_obunalari
      WHERE foydalanuvchi_id = $1 AND faol AND $2 = ANY(hodisalar)
     ON CONFLICT (obuna_id, hodisa_id) DO NOTHING`,
    [foydalanuvchiId, tur, hodisaId, tana],
  );
  return r.rowCount ?? 0;
}

type Qator = {
  id: string;
  obuna_id: string;
  hodisa_id: string;
  tana: unknown;
  urinishlar: number;
  url: string;
  sir: string;
};

/**
 * Vaqti kelgan yetkazishlarni oladi va yuboradi. Ko'p worker parallel
 * ishlashi mumkin (SKIP LOCKED); o'lgan worker qoldirgan
 * 'yuborilmoqda' — muddatdan keyin qayta olinadi.
 */
export async function yetkazishIshi(
  pool: Pool,
  o: {
    agent: Agent;
    limit?: number;
    httpRuxsat?: boolean;
    portlar?: number[];
    kutishMs?: number;
    tasodif?: () => number;
  },
): Promise<{
  yuborildi: number;
  qayta: number;
  tashlandi: number;
}> {
  const kutish = o.kutishMs ?? 10_000;
  const r = await pool.query<Qator>(
    `UPDATE webhook_yetkazishlar y
        SET holat = 'yuborilmoqda', urinishlar = urinishlar + 1,
            keyingi_urinish = now() + make_interval(secs => $2)
       FROM webhook_obunalari o
      WHERE y.id IN (
              SELECT id FROM webhook_yetkazishlar
               WHERE holat IN ('navbatda', 'yuborilmoqda')
                 AND keyingi_urinish <= now()
               ORDER BY keyingi_urinish
               LIMIT $1 FOR UPDATE SKIP LOCKED)
        AND o.id = y.obuna_id
     RETURNING y.id, y.obuna_id, y.hodisa_id, y.tana,
               y.urinishlar, o.url, o.sir`,
    [o.limit ?? 50, kutish / 1000 + 30],
  );
  const hisob = { yuborildi: 0, qayta: 0, tashlandi: 0 };
  await Promise.all(
    r.rows.map(async (q) => {
      const tana = JSON.stringify(q.tana);
      let status: number | null = null;
      let xato: string | null = null;
      try {
        const u = urlTekshir(q.url, {
          ...(o.httpRuxsat === undefined
            ? {}
            : { httpRuxsat: o.httpRuxsat }),
          ...(o.portlar ? { portlar: o.portlar } : {}),
        });
        const javob = await undiciFetch(u, {
          method: "POST",
          headers: {
            "content-type": "application/json",
            "user-agent": "vazifalar-webhook/1",
            ...sarlavhalar(q.sir, q.hodisa_id, tana),
          },
          body: tana,
          // 3xx — muvaffaqiyat emas; yo'naltirish orqali ichki
          // manzilga o'tib ketmasin
          redirect: "manual",
          dispatcher: o.agent,
          signal: AbortSignal.timeout(kutish),
        });
        status = javob.status;
        // javob tanasi kerak emas, ulanishni bo'shatamiz
        await javob.body?.cancel();
      } catch (e) {
        const sabab =
          e instanceof Error && e.cause instanceof Error
            ? e.cause.message
            : e instanceof Error
              ? e.message
              : "";
        xato =
          e instanceof Error
            ? `${e.name}: ${sabab.slice(0, 200)}`
            : String(e);
      }
      const ok = status !== null && status >= 200 && status < 300;
      if (ok) {
        hisob.yuborildi++;
        await pool.query(
          "UPDATE webhook_yetkazishlar SET holat = 'yuborildi', " +
            "oxirgi_status = $2, oxirgi_xato = NULL, " +
            "yuborilgan_vaqt = now() WHERE id = $1",
          [q.id, status],
        );
        await pool.query(
          "UPDATE webhook_obunalari SET ketma_ket_xato = 0 WHERE " +
            "id = $1",
          [q.obuna_id],
        );
        return;
      }
      const kech =
        status === 410
          ? undefined
          : kechikishMs(q.urinishlar, o.tasodif);
      if (kech === undefined) {
        hisob.tashlandi++;
        await pool.query(
          "UPDATE webhook_yetkazishlar SET holat = 'tashlandi', " +
            "oxirgi_status = $2, oxirgi_xato = $3 WHERE id = $1",
          [q.id, status, xato],
        );
      } else {
        hisob.qayta++;
        await pool.query(
          "UPDATE webhook_yetkazishlar SET holat = 'navbatda', " +
            "oxirgi_status = $2, oxirgi_xato = $3, keyingi_urinish " +
            "= now() + make_interval(secs => $4) WHERE id = $1",
          [q.id, status, xato, kech / 1000],
        );
      }
      await pool.query(
        "UPDATE webhook_obunalari SET ketma_ket_xato = " +
          "ketma_ket_xato + 1, faol = faol AND $2 AND " +
          "ketma_ket_xato + 1 < $3 WHERE id = $1",
        [q.obuna_id, status !== 410, OBUNA_OCHIRISH_CHEGARASI],
      );
    }),
  );
  return hisob;
}

// ssrf.ts — foydalanuvchi bergan URL'ga server so'rov yuborishdan
// oldin: faqat ommaviy IP'lar.
// Tekshiruv ULANISH paytida (undici Agent `connect.lookup`): DNS
// javobi tekshiriladi va AYNAN shu IP'ga ulaniladi — DNS rebinding
// (avval ommaviy, keyin 127.0.0.1)
// oynasi yopiladi. URL'ni oldindan `dns.lookup` qilib, keyin fetch
// qilish — rebinding'ga ochiq (ikki marta so'raladi).
import {
  lookup as dnsLookup,
  type LookupAddress,
  type LookupOptions,
} from "node:dns";
import { BlockList, isIP } from "node:net";
import { Agent } from "undici";

const taqiq = new BlockList();
for (const [tarmoq, prefiks] of [
  ["0.0.0.0", 8],
  ["10.0.0.0", 8],
  ["100.64.0.0", 10],
  ["127.0.0.0", 8],
  ["169.254.0.0", 16],
  ["172.16.0.0", 12],
  ["192.0.0.0", 24],
  ["192.0.2.0", 24],
  ["192.168.0.0", 16],
  ["198.18.0.0", 15],
  ["198.51.100.0", 24],
  ["203.0.113.0", 24],
  ["224.0.0.0", 4],
  ["240.0.0.0", 4],
] as const)
  taqiq.addSubnet(tarmoq, prefiks, "ipv4");
for (const [tarmoq, prefiks] of [
  ["::", 128],
  ["::1", 128],
  ["fc00::", 7],
  ["fe80::", 10],
  ["ff00::", 8],
  ["2001:db8::", 32],
  ["64:ff9b::", 96],
] as const)
  taqiq.addSubnet(tarmoq, prefiks, "ipv6");

/** IPv4-mapped IPv6 (::ffff:127.0.0.1) ham tekshiriladi. */
export function ommaviyIpmi(ip: string): boolean {
  const m = /^::ffff:(\d+\.\d+\.\d+\.\d+)$/i.exec(ip);
  const v4 = m?.[1];
  if (v4) return !taqiq.check(v4, "ipv4");
  const tur = isIP(ip);
  if (tur === 0) return false;
  return !taqiq.check(ip, tur === 4 ? "ipv4" : "ipv6");
}

export class SsrfXato extends Error {
  readonly code = "SSRF_TAQIQ";
  constructor(xabar: string) {
    super(xabar);
    this.name = "SsrfXato";
  }
}

/**
 * URL shakli: faqat http(s), foydalanuvchi:parol yo'q, port — 80/443
 * yoki ruxsat etilgan. IP literal bo'lsa — darhol tekshiriladi.
 */
export function urlTekshir(
  url: string,
  o: { httpRuxsat?: boolean; portlar?: number[] } = {},
): URL {
  let u: URL;
  try {
    u = new URL(url);
  } catch {
    throw new SsrfXato("URL noto'g'ri");
  }
  if (
    u.protocol !== "https:" &&
    !(o.httpRuxsat && u.protocol === "http:")
  )
    throw new SsrfXato("faqat https");
  if (u.username || u.password)
    throw new SsrfXato("URL ichida login/parol");
  const port = u.port
    ? Number(u.port)
    : u.protocol === "https:"
      ? 443
      : 80;
  if (!(o.portlar ?? [80, 443]).includes(port))
    throw new SsrfXato(`port ${String(port)} taqiqlangan`);
  const host = u.hostname.replace(/^\[|\]$/g, "");
  if (isIP(host) && !ommaviyIpmi(host))
    throw new SsrfXato(`taqiqlangan IP: ${host}`);
  return u;
}

✔ imzo: standardwebhooks 1.1.1 bilan ikki tomonlama mos (5.4168ms)
✔ imzo: o'zgartirilgan tana, boshqa sir, eski/kelajak vaqt, sarlavhasiz (0.5404ms)
✔ SSRF: IP ro'yxati va URL qoidalari (6.2427ms)
✔ kechikish jadvali: jitter ±20 %, oxirida undefined (DLQ) (0.2222ms)
▶ chiquvchi yetkazish (Postgres + lokal qabul qiluvchi)
  ✔ yuborildi: imzo qabul qiluvchida tekshiriladi; hodisa_id bir xil bo'lsa takror navbatga tushmaydi (119.3479ms)
  ✔ 500 → qayta (urinishlar, keyingi vaqt); 302 → muvaffaqiyat EMAS; 410 → obuna o'chadi; jadval tugasa — DLQ (34.6685ms)
  ✔ SSRF: localhost URL (ruxsat ro'yxatisiz agent) — ulanishdan oldin to'xtatiladi (11.2884ms)
✔ chiquvchi yetkazish (Postgres + lokal qabul qiluvchi) (3221.3033ms)
ℹ tests 7
ℹ suites 1
ℹ pass 7
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 7927.0156

M='feat(webhook): chiquvchi yetkazish — jurnal, SKIP LOCKED,'
M+=' jitter, DLQ, 410, SSRF agent'
git commit -m "$M"