fix(identity-webhook): map allowance rejects to 483/503 + publish 0.4.2

.env.example

@@ -39,11 +39,14 @@ USAGE_SUBJECT_TYPE=api_key_user
 # Generic OIDC (raw IdP access tokens with azp/sub):
 # OIDC_ISSUER=https://your-tenant.auth0.com/
 # OIDC_AUDIENCE=https://api.your-platform.com
-# OIDC_JWKS_URI=                       # defaults to ${OIDC_ISSUER}/.well-known/jwks.json
+# OIDC_JWKS_URI=                       # optional override; else OIDC Discovery → jwks_uri
 # OIDC_CLIENT_CLAIM=azp
 # OIDC_SUBJECT_CLAIM=sub
 # OIDC_SUBJECT_TYPE=oidc_user
 # OIDC_REQUIRED_SCOPES=
+# OIDC_TOKEN_EXCHANGE_BASE_URL=        # optional; enables app_<24hex>_<secret> composite key exchange
+# OIDC_EXCHANGE_M2M_CLIENT_ID=
+# OIDC_EXCHANGE_M2M_CLIENT_SECRET=
 
 # Reference only — not read by Compose services; for minting test JWTs.
 # Canonical copy: auth0-provisioner/provision/.env.livepeer

README.md

@@ -158,11 +158,13 @@ Shared keys (`WEBHOOK_SECRET`, `KAFKA_BROKERS`, `KAFKA_GATEWAY_TOPIC`) are liste
 | `API_KEY_PREFIX` | `api_key`, optional | Default `sk_` |
 | `OIDC_ISSUER` | `oidc` | JWT issuer / JWKS host |
 | `OIDC_AUDIENCE` | `oidc` | Expected JWT audience |
-| `OIDC_JWKS_URI` | `oidc`, optional | Defaults to `${OIDC_ISSUER}/.well-known/jwks.json` |
+| `OIDC_JWKS_URI` | `oidc`, optional | Explicit JWKS override; otherwise resolved via OIDC Discovery (`${OIDC_ISSUER}/.well-known/openid-configuration` → `jwks_uri`) |
 | `OIDC_CLIENT_CLAIM` | `oidc`, optional | Tenant claim (default `azp`; Auth0 clearinghouse uses `app_client_id`) |
 | `OIDC_SUBJECT_CLAIM` | `oidc`, optional | End-user claim (default `sub`; Auth0 clearinghouse uses `external_user_id`) |
 | `OIDC_SUBJECT_TYPE` | `oidc`, optional | Default `oidc_user`; Auth0 clearinghouse uses `external_user_id` |
 | `OIDC_REQUIRED_SCOPES` | `oidc`, optional | Space- or comma-separated required scopes (e.g. `sign:job`) |
+| `OIDC_TOKEN_EXCHANGE_BASE_URL` | `oidc`, optional | Enables composite `Bearer app_<24hex>_<secret>`: RFC 8693 exchange at `/api/v1/apps/{clientId}/oidc/token`, then JWT verify |
+| `OIDC_EXCHANGE_M2M_CLIENT_ID` / `OIDC_EXCHANGE_M2M_CLIENT_SECRET` | `oidc`, optional | Optional client credentials for the exchange request (both or neither) |
 
 `PORT` is set by Compose (`8090`), not `.env`. See the `identity-webhook` block in [`.env.example`](.env.example) for Auth0 vs generic OIDC examples.
 

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;
+  reauthTtlSeconds?: number;
+  failClosed?: boolean;
+  onError?: (err: unknown, identity: UsageIdentity) => void;
+}): BalanceCheck;

identity-webhook/balance-gate.mjs

@@ -0,0 +1,144 @@
+/**
+ * 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,
+ * integer number, or integer string.
+ *
+ * Example:
+ *   import { handleAuthorize } from "@pymthouse/clearinghouse-identity-webhook/protocol";
+ *   import { createBalanceGate } from "@pymthouse/clearinghouse-identity-webhook/balance-gate";
+ *
+ *   const checkBalance = createBalanceGate({
+ *     getBalanceUsdMicros: async (identity) =>
+ *       readLiveCreditBalanceUsdMicros(identity.client_id, identity.usage_subject),
+ *     reauthTtlSeconds: 30, // re-check at least every 30s mid-stream
+ *   });
+ *   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 | integer number | integer string) to a
+ * bigint. Returns null for anything non-integer (including "1.5", "", null).
+ */
+export function parseUsdMicros(value) {
+  if (typeof value === "bigint") {
+    return value;
+  }
+  if (typeof value === "number") {
+    return Number.isInteger(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.js").UsageIdentity, ctx: import("./protocol.js").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. Default: 1 micro (any positive credit).
+ * @param {number} [options.reauthTtlSeconds]
+ *   When set, caps the returned expiry to now + this, forcing go-livepeer to
+ *   call back and re-check the balance at least this often.
+ * @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.js").UsageIdentity) => void} [options.onError]
+ *   Optional hook to observe lookup errors / unparseable balances.
+ * @returns {import("./protocol.js").BalanceCheck}
+ */
+export function createBalanceGate({
+  getBalanceUsdMicros,
+  minBalanceUsdMicros = 1n,
+  reauthTtlSeconds,
+  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");
+  }
+  let ttl = null;
+  if (reauthTtlSeconds != null) {
+    ttl = Number(reauthTtlSeconds);
+    if (!Number.isFinite(ttl) || ttl <= 0) {
+      throw new TypeError("createBalanceGate: reauthTtlSeconds must be a positive number");
+    }
+  }
+
+  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,134 @@
+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);
+  });
+});
+
+describe("createBalanceGate", () => {
+  it("throws on missing getBalanceUsdMicros", () => {
+    assert.throws(() => createBalanceGate({}), /getBalanceUsdMicros is required/);
+  });
+
+  it("validates minBalanceUsdMicros and reauthTtlSeconds", () => {
+    assert.throws(
+      () => createBalanceGate({ getBalanceUsdMicros: async () => 0n, minBalanceUsdMicros: "x" }),
+      /minBalanceUsdMicros must be an integer/,
+    );
+    assert.throws(
+      () => createBalanceGate({ getBalanceUsdMicros: async () => 0n, reauthTtlSeconds: 0 }),
+      /reauthTtlSeconds must be a positive number/,
+    );
+  });
+
+  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 reauthTtlSeconds", async () => {
+    const gate = createBalanceGate({
+      getBalanceUsdMicros: async () => 10n,
+      reauthTtlSeconds: 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);
+  });
+});

identity-webhook/legacy-env.test.mjs

@@ -173,10 +173,27 @@ describe("pymthouse embedded flow (handleAuthorize + legacy config)", () => {
     assert.equal(body.identity.usage_subject_type, "external_user_id");
   });
 
-  it("authorizes a composite app_*.pmth_* via mocked exchange", async () => {
-    const { token, jwks } = await setupPymthouseJwt();
+  it("authorizes a composite app_*_* via mocked exchange", async () => {
+    const { publicKey, privateKey } = await generateKeyPair("RS256");
+    const jwk = await exportJWK(publicKey);
+    jwk.kid = "pymthouse-test";
+    jwk.alg = "RS256";
+    jwk.use = "sig";
+    const jwks = createLocalJWKSet({ keys: [jwk] });
+    const clientId = "app_abc123abc123abc123abcdef";
+    const token = await new SignJWT({
+      client_id: clientId,
+      external_user_id: "user-456",
+      scope: "sign:job",
+    })
+      .setProtectedHeader({ alg: "RS256", kid: "pymthouse-test" })
+      .setIssuer(ISSUER)
+      .setAudience(defaultSignerWebhookJwtAudience(ISSUER))
+      .setIssuedAt()
+      .setExpirationTime("5m")
+      .sign(privateKey);
+
     const { createOidcVerifier } = await import("./verifiers.mjs");
-    const clientId = "app_abc123";
     const fetchImpl = async (input) => {
       const url = String(input);
       assert.ok(url.includes(`/api/v1/apps/${clientId}/oidc/token`));
@@ -205,14 +222,16 @@ describe("pymthouse embedded flow (handleAuthorize + legacy config)", () => {
         "content-type": "application/json",
       },
       body: JSON.stringify({
-        headers: { Authorization: [`Bearer ${clientId}.pmth_deadbeef`] },
+        headers: {
+          Authorization: [`Bearer ${clientId}_pmth_deadbeef`],
+        },
       }),
     });
 
     const response = await handleAuthorize(request, config);
     assert.equal(response.status, 200);
     const body = await response.json();
     assert.equal(body.status, 200);
-    assert.equal(body.auth_id, "app_abc123:user-456");
+    assert.equal(body.auth_id, `${clientId}:user-456`);
   });
 });