feat(identity-webhook): live balance gate after identity verify

identity-webhook/Dockerfile

@@ -7,7 +7,7 @@ WORKDIR /app
 COPY identity-webhook/package.json identity-webhook/package-lock.json ./
 RUN npm ci --omit=dev
 
-COPY identity-webhook/keys.mjs identity-webhook/protocol.mjs identity-webhook/verifiers.mjs identity-webhook/server.mjs ./
+COPY identity-webhook/keys.mjs identity-webhook/protocol.mjs identity-webhook/verifiers.mjs identity-webhook/balance-gate.mjs identity-webhook/server.mjs ./
 
 ENV PORT=8090
 

identity-webhook/balance-gate.d.ts

@@ -0,0 +1,26 @@
+import type {
+  BalanceCheck,
+  BalanceCheckContext,
+  UsageIdentity,
+} from "./protocol.js";
+
+export function parseUsdMicros(
+  value: bigint | number | string | null | undefined,
+): bigint | null;
+
+export function createBalanceGate(options: {
+  getBalanceUsdMicros: (
+    identity: UsageIdentity,
+    ctx: BalanceCheckContext,
+  ) =>
+    | Promise<bigint | number | string | null | undefined>
+    | bigint
+    | number
+    | string
+    | null
+    | undefined;
+  minBalanceUsdMicros?: bigint | number | string;
+  expiryTtlSeconds?: number;
+  failClosed?: boolean;
+  onError?: (err: unknown, identity: UsageIdentity) => void;
+}): BalanceCheck;

identity-webhook/balance-gate.mjs

@@ -0,0 +1,149 @@
+/**
+ * Live balance/credit gate for the identity webhook.
+ *
+ * `createBalanceGate` turns a simple per-identity balance lookup into a
+ * `checkBalance` hook for `handleAuthorize` (protocol.mjs). It is verifier
+ * agnostic: whatever proved the identity (OIDC JWT, composite API key, or a
+ * plain API key), the gate reads the caller-supplied balance and rejects with
+ * the go-livepeer wire status 483 (`insufficient_balance`) when the customer is
+ * out of credit — closing the "still streaming after credit hits zero" gap that
+ * a mint-time-only gate leaves open.
+ *
+ * Balances are USD micros (1 USD = 1_000_000 micros), accepted as bigint,
+ * safe integer number, or integer string.
+ *
+ * Example (in-repo relative imports; published package exports may differ):
+ *   import { handleAuthorize } from "./protocol.mjs";
+ *   import { createBalanceGate } from "./balance-gate.mjs";
+ *
+ *   const checkBalance = createBalanceGate({
+ *     getBalanceUsdMicros: async (identity) =>
+ *       readLiveCreditBalanceUsdMicros(identity.client_id, identity.usage_subject),
+ *     expiryTtlSeconds: 30, // sets webhook expiry = now + 30s (go-livepeer AuthExpiry cache)
+ *   });
+ *   return handleAuthorize(request, { webhookSecret, endUserAuth, checkBalance });
+ */
+import {
+  REMOTE_SIGNER_ERROR_CODE,
+  REMOTE_SIGNER_HTTP_STATUS,
+  WebhookError,
+} from "./protocol.mjs";
+
+function nowSeconds() {
+  return Math.floor(Date.now() / 1000);
+}
+
+/**
+ * Coerce a USD-micros balance (bigint | safe integer number | integer string)
+ * to a bigint. Returns null for anything non-integer (including "1.5", "", null).
+ * Numbers outside Number.MAX_SAFE_INTEGER must be passed as string or bigint.
+ */
+export function parseUsdMicros(value) {
+  if (typeof value === "bigint") {
+    return value;
+  }
+  if (typeof value === "number") {
+    return Number.isSafeInteger(value) ? BigInt(value) : null;
+  }
+  if (typeof value === "string") {
+    const trimmed = value.trim();
+    if (!/^-?\d+$/.test(trimmed)) {
+      return null;
+    }
+    try {
+      return BigInt(trimmed);
+    } catch {
+      return null;
+    }
+  }
+  return null;
+}
+
+/**
+ * Build a `checkBalance` hook from a balance lookup.
+ *
+ * @param {object} options
+ * @param {(identity: import("./protocol.mjs").UsageIdentity, ctx: import("./protocol.mjs").BalanceCheckContext) => any} options.getBalanceUsdMicros
+ *   Resolve remaining balance (USD micros) for the identity. May be async.
+ *   Return null/undefined to signal "balance unknown" (see failClosed).
+ * @param {bigint | number | string} [options.minBalanceUsdMicros=1]
+ *   Minimum balance required to authorize (non-negative). Default: 1 micro.
+ * @param {number} [options.expiryTtlSeconds]
+ *   When set, sets the webhook response `expiry` to now + this many whole
+ *   seconds (also capped against the verifier expiry). go-livepeer stores that
+ *   as AuthExpiry and skips /authorize until it elapses.
+ * @param {boolean} [options.failClosed=true]
+ *   On lookup error or unknown balance: true → reject 503 billing_unavailable;
+ *   false → allow (fail open).
+ * @param {(err: unknown, identity: import("./protocol.mjs").UsageIdentity) => void} [options.onError]
+ *   Optional hook to observe lookup errors / unparseable balances.
+ * @returns {import("./protocol.mjs").BalanceCheck}
+ */
+export function createBalanceGate({
+  getBalanceUsdMicros,
+  minBalanceUsdMicros = 1n,
+  expiryTtlSeconds,
+  failClosed = true,
+  onError,
+} = {}) {
+  if (typeof getBalanceUsdMicros !== "function") {
+    throw new TypeError("createBalanceGate: getBalanceUsdMicros is required");
+  }
+  const minBalance = parseUsdMicros(minBalanceUsdMicros);
+  if (minBalance === null) {
+    throw new TypeError("createBalanceGate: minBalanceUsdMicros must be an integer");
+  }
+  if (minBalance < 0n) {
+    throw new TypeError("createBalanceGate: minBalanceUsdMicros must not be negative");
+  }
+  let ttl = null;
+  if (expiryTtlSeconds != null) {
+    ttl = Number(expiryTtlSeconds);
+    if (!Number.isInteger(ttl) || ttl <= 0) {
+      throw new TypeError("createBalanceGate: expiryTtlSeconds must be a positive integer");
+    }
+  }
+
+  const billingUnavailable = (message) =>
+    new WebhookError(message, {
+      status: REMOTE_SIGNER_HTTP_STATUS.BILLING_UNAVAILABLE,
+      code: REMOTE_SIGNER_ERROR_CODE.BILLING_UNAVAILABLE,
+    });
+
+  return async function checkBalance(ctx) {
+    let rawBalance;
+    try {
+      rawBalance = await getBalanceUsdMicros(ctx.identity, ctx);
+    } catch (err) {
+      onError?.(err, ctx.identity);
+      if (failClosed) {
+        throw billingUnavailable("billing balance lookup failed");
+      }
+      return undefined;
+    }
+
+    const balance = parseUsdMicros(rawBalance);
+    if (balance === null) {
+      onError?.(
+        new Error(`balance is not an integer micros value: ${String(rawBalance)}`),
+        ctx.identity,
+      );
+      if (failClosed) {
+        throw billingUnavailable("billing balance unavailable");
+      }
+      return undefined;
+    }
+
+    if (balance < minBalance) {
+      throw new WebhookError("insufficient balance", {
+        status: REMOTE_SIGNER_HTTP_STATUS.INSUFFICIENT_BALANCE,
+        code: REMOTE_SIGNER_ERROR_CODE.INSUFFICIENT_BALANCE,
+      });
+    }
+
+    if (ttl !== null) {
+      return { expiry: nowSeconds() + ttl };
+    }
+    return undefined;
+  };
+}

identity-webhook/balance-gate.test.mjs

@@ -0,0 +1,155 @@
+import assert from "node:assert/strict";
+import { describe, it } from "node:test";
+import { createBalanceGate, parseUsdMicros } from "./balance-gate.mjs";
+import { WebhookError } from "./protocol.mjs";
+
+const identity = {
+  issuer: "http://webhook.test",
+  client_id: "tenant-a",
+  usage_subject: "user-1",
+  usage_subject_type: "external_user_id",
+};
+
+function ctx(overrides = {}) {
+  return { identity, expiry: 2000000000, payload: {}, request: new Request("http://x/authorize"), ...overrides };
+}
+
+describe("parseUsdMicros", () => {
+  it("accepts bigint, integer number, and integer string", () => {
+    assert.equal(parseUsdMicros(5n), 5n);
+    assert.equal(parseUsdMicros(5), 5n);
+    assert.equal(parseUsdMicros("5"), 5n);
+    assert.equal(parseUsdMicros("  42 "), 42n);
+    assert.equal(parseUsdMicros("0"), 0n);
+    assert.equal(parseUsdMicros("-3"), -3n);
+  });
+
+  it("rejects non-integer / junk values as null", () => {
+    assert.equal(parseUsdMicros(null), null);
+    assert.equal(parseUsdMicros(undefined), null);
+    assert.equal(parseUsdMicros(""), null);
+    assert.equal(parseUsdMicros("1.5"), null);
+    assert.equal(parseUsdMicros("abc"), null);
+    assert.equal(parseUsdMicros(1.5), null);
+    assert.equal(parseUsdMicros(Number.MAX_SAFE_INTEGER + 1), null);
+    assert.equal(parseUsdMicros(Number.POSITIVE_INFINITY), null);
+    assert.equal(parseUsdMicros(Number.NaN), null);
+  });
+
+  it("accepts large balances via string or bigint", () => {
+    const big = "9007199254740993"; // MAX_SAFE_INTEGER + 2
+    assert.equal(parseUsdMicros(big), BigInt(big));
+    assert.equal(parseUsdMicros(BigInt(big)), BigInt(big));
+  });
+});
+
+describe("createBalanceGate", () => {
+  it("throws on missing getBalanceUsdMicros", () => {
+    assert.throws(() => createBalanceGate({}), /getBalanceUsdMicros is required/);
+  });
+
+  it("validates minBalanceUsdMicros and expiryTtlSeconds", () => {
+    assert.throws(
+      () => createBalanceGate({ getBalanceUsdMicros: async () => 0n, minBalanceUsdMicros: "x" }),
+      /minBalanceUsdMicros must be an integer/,
+    );
+    assert.throws(
+      () => createBalanceGate({ getBalanceUsdMicros: async () => 0n, minBalanceUsdMicros: -1 }),
+      /minBalanceUsdMicros must not be negative/,
+    );
+    assert.throws(
+      () => createBalanceGate({ getBalanceUsdMicros: async () => 0n, minBalanceUsdMicros: "-5" }),
+      /minBalanceUsdMicros must not be negative/,
+    );
+    assert.throws(
+      () => createBalanceGate({ getBalanceUsdMicros: async () => 0n, expiryTtlSeconds: 0 }),
+      /expiryTtlSeconds must be a positive integer/,
+    );
+    assert.throws(
+      () => createBalanceGate({ getBalanceUsdMicros: async () => 0n, expiryTtlSeconds: 1.5 }),
+      /expiryTtlSeconds must be a positive integer/,
+    );
+  });
+
+  it("allows a positive balance", async () => {
+    const gate = createBalanceGate({ getBalanceUsdMicros: async () => "1" });
+    assert.equal(await gate(ctx()), undefined);
+  });
+
+  it("rejects zero balance with 483 insufficient_balance", async () => {
+    const gate = createBalanceGate({ getBalanceUsdMicros: async () => 0n });
+    await assert.rejects(gate(ctx()), (err) => {
+      assert.ok(err instanceof WebhookError);
+      assert.equal(err.status, 483);
+      assert.equal(err.code, "insufficient_balance");
+      return true;
+    });
+  });
+
+  it("rejects a negative balance", async () => {
+    const gate = createBalanceGate({ getBalanceUsdMicros: async () => "-100" });
+    await assert.rejects(gate(ctx()), /insufficient balance/);
+  });
+
+  it("honors a custom minBalanceUsdMicros threshold", async () => {
+    const gate = createBalanceGate({
+      getBalanceUsdMicros: async () => 500n,
+      minBalanceUsdMicros: 1000n,
+    });
+    await assert.rejects(gate(ctx()), (err) => err.status === 483);
+  });
+
+  it("passes identity to the lookup", async () => {
+    let seen;
+    const gate = createBalanceGate({
+      getBalanceUsdMicros: async (id) => {
+        seen = id;
+        return 10n;
+      },
+    });
+    await gate(ctx());
+    assert.equal(seen.client_id, "tenant-a");
+    assert.equal(seen.usage_subject, "user-1");
+  });
+
+  it("caps expiry via expiryTtlSeconds", async () => {
+    const gate = createBalanceGate({
+      getBalanceUsdMicros: async () => 10n,
+      expiryTtlSeconds: 30,
+    });
+    const before = Math.floor(Date.now() / 1000);
+    const result = await gate(ctx());
+    assert.ok(result.expiry >= before + 30 && result.expiry <= before + 31);
+  });
+
+  it("fails closed with 503 on lookup error by default", async () => {
+    const errors = [];
+    const gate = createBalanceGate({
+      getBalanceUsdMicros: async () => {
+        throw new Error("openmeter down");
+      },
+      onError: (err) => errors.push(err),
+    });
+    await assert.rejects(gate(ctx()), (err) => {
+      assert.equal(err.status, 503);
+      assert.equal(err.code, "billing_unavailable");
+      return true;
+    });
+    assert.equal(errors.length, 1);
+  });
+
+  it("fails open when failClosed is false", async () => {
+    const gate = createBalanceGate({
+      getBalanceUsdMicros: async () => {
+        throw new Error("openmeter down");
+      },
+      failClosed: false,
+    });
+    assert.equal(await gate(ctx()), undefined);
+  });
+
+  it("treats an unparseable balance as billing_unavailable when failing closed", async () => {
+    const gate = createBalanceGate({ getBalanceUsdMicros: async () => "not-a-number" });
+    await assert.rejects(gate(ctx()), (err) => err.status === 503);
+  });
+});