Separate Zapier, web, API, and leaf access planes with NATS authz

Zapier is one ingress. Direct web, customer API, and S2S leaf nodes are their own services. Every hop to an internal subject must pass verae.access.authz.check (default deny by plane).

packages/zappier/src/admin.ts

@@ -248,14 +248,17 @@ export function adminRouter(
       agent: typeof agent === 'string' ? agent : 'admin',
     });
     customers.save({ ...customer, balanceCents: (customer.balanceCents ?? 0) + delta });
-    natsPublish(BILLING_SUBJECTS.CREDIT_APPLIED, rec);
-    void natsAdjust({
-      customerId,
-      cents: delta,
-      reason: rec.reason,
-      agent: rec.agent,
-      kind: 'credit',
-    });
+    natsPublish(BILLING_SUBJECTS.CREDIT_APPLIED, rec, 'staff');
+    void natsAdjust(
+      {
+        customerId,
+        cents: delta,
+        reason: rec.reason,
+        agent: rec.agent,
+        kind: 'credit',
+      },
+      'staff',
+    );
     res.status(201).json(rec);
   });
 
@@ -270,7 +273,7 @@ export function adminRouter(
       res.status(404).json({ error: 'customer not found' });
       return;
     }
-    const fromNats = await natsStatement(customer.id);
+    const fromNats = await natsStatement(customer.id, 'staff');
     if (fromNats) {
       res.json({ ...fromNats, name: customer.name, tierId: customer.tierId, source: 'nats' });
       return;

packages/zappier/src/app.ts

@@ -103,7 +103,7 @@ export function buildApp(deps: AppDeps = {}): {
   const items: StoredItem[] = [];
   const credits = new CreditLedger();
   const onUsage = (e: { customerId: string; endpointId: string; cents: number }) =>
-    natsPublish(BILLING_SUBJECTS.USAGE_RECORDED, { ...e, at: new Date().toISOString() });
+    natsPublish(BILLING_SUBJECTS.USAGE_RECORDED, { ...e, at: new Date().toISOString() }, 'api');
   const hashIndex = new Map<
     string,
     { jobId: string; sha256: string; data?: string; timestamp: string }

packages/zappier/src/billing-nats.ts

@@ -1,4 +1,6 @@
-/** Internal NATS billing bus. No-op when NATS_URL is unset (tests). */
+/** Internal NATS billing bus. No-op when NATS_URL is unset (tests).
+ *  When NATS is on, every hop must pass verae.access.authz.check for a plane.
+ */
 
 export const BILLING_SUBJECTS = {
   STATEMENT_GET: 'verae.billing.statement.get',
@@ -8,6 +10,9 @@ export const BILLING_SUBJECTS = {
   CREDIT_APPLIED: 'verae.billing.credit.applied',
 };
 
+export const AUTHZ_CHECK = 'verae.access.authz.check';
+export type AccessPlane = 'zapier' | 'web' | 'api' | 'leaf' | 'staff';
+
 export type BillingStatement = {
   customerId: string;
   prepaidCents: number;
@@ -46,28 +51,79 @@ function decode(buf: Uint8Array): unknown {
   return JSON.parse(new TextDecoder().decode(buf) || '{}');
 }
 
-export async function natsStatement(customerId: string): Promise<BillingStatement | null> {
+async function authzAllow(
+  plane: AccessPlane,
+  subject: string,
+  extra: { principal?: string; kind?: string } = {},
+): Promise<boolean> {
+  const http = process.env.AUTHZ_URL;
+  if (!http && !(await nc())) return true;
+  try {
+    const c = await nc();
+    if (c) {
+      const m = await c.request(AUTHZ_CHECK, encode({ plane, subject, ...extra }), { timeout: 1500 });
+      const d = decode(m.data) as { allow?: boolean };
+      return Boolean(d.allow);
+    }
+    if (http) {
+      const r = await fetch(`${http.replace(/\/$/, '')}/check`, {
+        method: 'POST',
+        headers: { 'content-type': 'application/json' },
+        body: JSON.stringify({ plane, subject, ...extra }),
+      });
+      const d = (await r.json()) as { allow?: boolean };
+      return Boolean(d.allow);
+    }
+  } catch {
+    return false;
+  }
+  return true;
+}
+
+export async function natsStatement(
+  customerId: string,
+  plane: AccessPlane = 'web',
+): Promise<BillingStatement | null> {
   const c = await nc();
   if (!c) return null;
-  const m = await c.request(BILLING_SUBJECTS.STATEMENT_GET, encode({ customerId }), { timeout: 2000 });
+  if (!(await authzAllow(plane, BILLING_SUBJECTS.STATEMENT_GET, { principal: customerId }))) return null;
+  const m = await c.request(
+    BILLING_SUBJECTS.STATEMENT_GET,
+    encode({ customerId, plane }),
+    { timeout: 2000 },
+  );
   return decode(m.data) as BillingStatement;
 }
 
-export async function natsAdjust(payload: {
-  customerId: string;
-  cents: number;
-  reason: string;
-  agent: string;
-  kind?: string;
-}): Promise<unknown | null> {
+export async function natsAdjust(
+  payload: {
+    customerId: string;
+    cents: number;
+    reason: string;
+    agent: string;
+    kind?: string;
+  },
+  plane: AccessPlane = 'staff',
+): Promise<unknown | null> {
   const c = await nc();
   if (!c) return null;
-  const m = await c.request(BILLING_SUBJECTS.BALANCE_ADJUST, encode(payload), { timeout: 2000 });
+  if (
+    !(await authzAllow(plane, BILLING_SUBJECTS.BALANCE_ADJUST, {
+      principal: payload.agent,
+      kind: payload.kind,
+    }))
+  ) {
+    return null;
+  }
+  const m = await c.request(BILLING_SUBJECTS.BALANCE_ADJUST, encode({ ...payload, plane }), { timeout: 2000 });
   return decode(m.data);
 }
 
-export function natsPublish(subject: string, payload: unknown): void {
-  void nc().then((c) => {
-    if (c) c.publish(subject, encode(payload));
-  });
+export function natsPublish(subject: string, payload: unknown, plane: AccessPlane = 'api'): void {
+  void (async () => {
+    const c = await nc();
+    if (!c) return;
+    if (!(await authzAllow(plane, subject))) return;
+    c.publish(subject, encode({ ...(payload as object), plane }));
+  })();
 }

packages/zappier/src/portal.ts

@@ -203,7 +203,7 @@ export function portalRouter(deps: PortalDeps): Router {
   });
 
   router.get('/statement', async (req, res) => {
-    const fromNats = await natsStatement(req.customer!.id);
+    const fromNats = await natsStatement(req.customer!.id, 'web');
     if (fromNats) {
       res.json({ ...fromNats, source: 'nats' });
       return;
@@ -286,18 +286,25 @@ export function portalRouter(deps: PortalDeps): Router {
     };
     if (result.creditedCents > 0) save(deps, customer);
     if (result.creditedCents > 0) {
-      natsPublish(BILLING_SUBJECTS.PAYMENT_RECORDED, {
-        customerId: customer.id,
-        cents: result.creditedCents,
-        reason: 'reload',
-      });
-      void natsAdjust({
-        customerId: customer.id,
-        cents: result.creditedCents,
-        reason: 'reload',
-        agent: 'portal',
-        kind: 'payment',
-      });
+      natsPublish(
+        BILLING_SUBJECTS.PAYMENT_RECORDED,
+        {
+          customerId: customer.id,
+          cents: result.creditedCents,
+          reason: 'reload',
+        },
+        'web',
+      );
+      void natsAdjust(
+        {
+          customerId: customer.id,
+          cents: result.creditedCents,
+          reason: 'reload',
+          agent: 'portal',
+          kind: 'reload',
+        },
+        'web',
+      );
     }
     res.json({
       balanceCents: customer.balanceCents,