feat(p9): Capability Broker — governed egress + fail-closed gate + obligations

Tasks 9.1+9.2: contracts (CapabilityRequest/Result/Provider, PolicyObligation) + service CapabilityBroker that fails closed via opa (PolicyDeniedError, nothing sent), enforces obligations before any provider call (DENY_OUTBOUND/REVIEW → BLOCKED; MASK/REDACT → mask payload field), then routes to a CapabilityProviderRegistry (SandboxProvider default; env-gated HttpProvider overrides per channel; unknown channel fails closed). 5 tests: allow/deny/block/redact/unknown-channel. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-01 19:58:25 +05:30
parent e11964b8bc
commit c50a501f32
9 changed files with 455 additions and 0 deletions
@@ -0,0 +1,53 @@
+/**
+ * Capability Broker contract (P9) — the single governed egress boundary.
+ *
+ * IIOS prepares idempotent outbound commands; the broker gates them through
+ * policy, enforces the returned obligations, and routes to a provider. In dev the
+ * provider is a sandbox (no network); a real provider is a binding/config swap.
+ * This is the concrete realization of the `capability` platform port.
+ */
+import type { DataClass } from './events';
+import type { PolicyDecision } from './ports';
+
+/** One obligation the broker must enforce before/at egress. */
+export type PolicyObligation = PolicyDecision['obligations'][number];
+
+export type CapabilityOutcome = 'SENT' | 'BLOCKED' | 'FAILED' | 'RATE_LIMITED';
+
+export interface CapabilityRequest {
+  /** Dotted capability name, e.g. "channel.send". */
+  capability: string;
+  scopeId?: string;
+  channelType: string;
+  target: string;
+  payload: Record<string, unknown>;
+  idempotencyKey: string;
+  dataClass?: DataClass;
+}
+
+export interface CapabilityResult {
+  providerRef: string;
+  outcome: CapabilityOutcome;
+  errorCode?: string;
+  latencyMs?: number;
+}
+
+/** What a provider reports; the broker normalizes `outcome` into CapabilityResult. */
+export interface ProviderResult {
+  providerRef: string;
+  outcome: string;
+  errorCode?: string;
+  latencyMs?: number;
+}
+
+/**
+ * A pluggable egress provider. Sandbox (no network) by default; real providers
+ * (WhatsApp/SMTP/HTTP webhook) implement the same shape. Must be idempotent and
+ * must surface transport errors as FAILED rather than throwing.
+ */
+export interface CapabilityProvider {
+  readonly name: string;
+  readonly channelTypes: string[];
+  readonly capabilities: { canSend: boolean; maxPayloadBytes?: number };
+  send(req: CapabilityRequest): Promise<ProviderResult>;
+}
@@ -3,6 +3,7 @@ export * from './enums';
 export * from './ingest';
 export * from './ports';
 export * from './inference';
+export * from './capability';
 export * from './events';
 export * from './commands';
 export * from './errors';
@@ -0,0 +1,55 @@
+import { describe, it, expect } from 'vitest';
+import { makeFakePorts } from '@insignia/iios-testkit';
+import { PolicyDeniedError, type CapabilityRequest } from '@insignia/iios-contracts';
+import { CapabilityBroker } from './capability.broker';
+import { CapabilityProviderRegistry } from './capability.registry';
+import type { FakePorts } from '@insignia/iios-testkit';
+
+// No DB — the broker is pure governance over an in-memory provider registry.
+const req = (over: Partial<CapabilityRequest> = {}): CapabilityRequest => ({
+  capability: 'channel.send',
+  scopeId: 'scope1',
+  channelType: 'WEBHOOK',
+  target: 'user@x',
+  payload: { text: 'hello' },
+  idempotencyKey: 'k1',
+  ...over,
+});
+const broker = (ports: FakePorts) => new CapabilityBroker(ports, new CapabilityProviderRegistry());
+
+describe('CapabilityBroker (P9 — governed egress)', () => {
+  it('allow + no obligations → provider SENT', async () => {
+    const r = await broker(makeFakePorts()).invoke(req());
+    expect(r.outcome).toBe('SENT');
+    expect(r.providerRef).toContain('sim-WEBHOOK');
+  });
+
+  it('fails closed: opa deny → PolicyDeniedError, provider never called', async () => {
+    const ports = makeFakePorts();
+    ports.opa.deny('no egress');
+    await expect(broker(ports).invoke(req())).rejects.toBeInstanceOf(PolicyDeniedError);
+  });
+
+  it('DENY_OUTBOUND obligation → BLOCKED, provider never called', async () => {
+    const ports = makeFakePorts();
+    ports.opa.setDecision({ allow: true, obligations: [{ kind: 'DENY_OUTBOUND', reason: 'user opted out' }] });
+    const r = await broker(ports).invoke(req());
+    expect(r.outcome).toBe('BLOCKED');
+    expect(r.errorCode).toBe('DENY_OUTBOUND');
+  });
+
+  it('REDACT obligation → provider receives masked payload', async () => {
+    const ports = makeFakePorts();
+    ports.opa.setDecision({ allow: true, obligations: [{ kind: 'REDACT', target: 'text', reason: 'PII' }] });
+    const registry = new CapabilityProviderRegistry();
+    let seen: unknown;
+    registry.register({ name: 'spy', channelTypes: ['WEBHOOK'], capabilities: { canSend: true }, async send(r) { seen = r.payload; return { providerRef: 'spy-1', outcome: 'SENT' }; } });
+    const r = await new CapabilityBroker(ports, registry).invoke(req());
+    expect(r.outcome).toBe('SENT');
+    expect((seen as { text: string }).text).toBe('[REDACTED]');
+  });
+
+  it('unknown channel → fails closed (throws, no silent egress)', async () => {
+    await expect(broker(makeFakePorts()).invoke(req({ channelType: 'CARRIER_PIGEON' }))).rejects.toThrow();
+  });
+});
@@ -0,0 +1,57 @@
+import { Inject, Injectable } from '@nestjs/common';
+import type { CapabilityRequest, CapabilityResult, IiosPlatformPorts } from '@insignia/iios-contracts';
+import { PLATFORM_PORTS } from '../platform/platform-ports';
+import { decideOrThrow } from '../platform/fail-closed';
+import { CapabilityProviderRegistry } from './capability.registry';
+
+/**
+ * The single governed egress boundary (P9). Every outbound action flows through
+ * here: it fails closed on policy (nothing sent unless explicitly allowed),
+ * enforces the policy's obligations BEFORE a provider is ever called, then routes
+ * to the channel's provider. It is the concrete realization of the `capability`
+ * platform port — a real broker replacing the permissive P1 stub.
+ */
+@Injectable()
+export class CapabilityBroker {
+  constructor(
+    @Inject(PLATFORM_PORTS) private readonly ports: IiosPlatformPorts,
+    private readonly registry: CapabilityProviderRegistry,
+  ) {}
+
+  async invoke(req: CapabilityRequest): Promise<CapabilityResult> {
+    // 1. Fail-closed policy gate. Deny/timeout ⇒ PolicyDeniedError, nothing sent.
+    const decision = await decideOrThrow(this.ports, {
+      action: `iios.capability.${req.capability}`,
+      scopeId: req.scopeId,
+      channelType: req.channelType,
+      target: req.target,
+    });
+
+    // 2. Enforce obligations BEFORE touching a provider.
+    for (const ob of decision.obligations) {
+      if (ob.kind === 'DENY_OUTBOUND' || ob.kind === 'REVIEW') {
+        return { providerRef: 'blocked', outcome: 'BLOCKED', errorCode: ob.kind };
+      }
+    }
+    const payload = this.applyMasks(req.payload, decision.obligations);
+
+    // 3. Route to the channel's provider (unknown channel fails closed in the registry).
+    const provider = this.registry.forChannel(req.channelType);
+    const result = await provider.send({ ...req, payload });
+
+    // 4. Normalize the provider's outcome.
+    const outcome = result.outcome === 'SENT' ? 'SENT' : result.outcome === 'RATE_LIMITED' ? 'RATE_LIMITED' : 'FAILED';
+    return { providerRef: result.providerRef, outcome, errorCode: result.errorCode, latencyMs: result.latencyMs };
+  }
+
+  private applyMasks(
+    payload: Record<string, unknown>,
+    obligations: { kind: string; target?: string }[],
+  ): Record<string, unknown> {
+    const masks = obligations.filter((o) => (o.kind === 'MASK' || o.kind === 'REDACT') && o.target);
+    if (masks.length === 0) return payload;
+    const out = { ...payload };
+    for (const m of masks) out[m.target as string] = '[REDACTED]';
+    return out;
+  }
+}
@@ -0,0 +1,14 @@
+import { Module } from '@nestjs/common';
+import { CapabilityProviderRegistry } from './capability.registry';
+import { CapabilityBroker } from './capability.broker';
+
+/**
+ * The governed egress boundary (P9 slice 1). Exports the broker + registry so the
+ * outbound layer routes all sends through policy + obligations + a pluggable
+ * provider. PLATFORM_PORTS (for the opa gate) comes from the @Global PlatformModule.
+ */
+@Module({
+  providers: [CapabilityProviderRegistry, CapabilityBroker],
+  exports: [CapabilityBroker, CapabilityProviderRegistry],
+})
+export class CapabilityModule {}
@@ -0,0 +1,40 @@
+import { Injectable, NotFoundException } from '@nestjs/common';
+import type { CapabilityProvider } from '@insignia/iios-contracts';
+import { SandboxProvider } from './sandbox.provider';
+import { HttpProvider } from './http.provider';
+
+const DEFAULT_CHANNELS = ['WEBHOOK', 'EMAIL', 'WHATSAPP', 'PORTAL'];
+
+/**
+ * Maps a channelType to the provider that executes egress for it. Sandbox by
+ * default; if `IIOS_PROVIDER_URL_<CHANNELTYPE>` is set, a real HttpProvider
+ * overrides the sandbox for that channel (the "flip the binding" swap). Unknown
+ * channels fail closed — no silent egress path.
+ */
+@Injectable()
+export class CapabilityProviderRegistry {
+  private readonly byChannel = new Map<string, CapabilityProvider>();
+
+  constructor() {
+    for (const ch of DEFAULT_CHANNELS) this.register(new SandboxProvider([ch]));
+    for (const ch of DEFAULT_CHANNELS) {
+      const url = process.env[`IIOS_PROVIDER_URL_${ch}`];
+      if (url) this.register(new HttpProvider(ch, url));
+    }
+  }
+
+  register(provider: CapabilityProvider): void {
+    for (const ch of provider.channelTypes) this.byChannel.set(ch, provider);
+  }
+
+  forChannel(channelType: string): CapabilityProvider {
+    const p = this.byChannel.get(channelType);
+    if (!p) throw new NotFoundException(`no egress provider registered for channel: ${channelType}`);
+    return p;
+  }
+
+  /** Ops view: which provider backs each channel. */
+  bindings(): Record<string, string> {
+    return Object.fromEntries([...this.byChannel.entries()].map(([ch, p]) => [ch, p.name]));
+  }
+}
@@ -0,0 +1,36 @@
+import type { CapabilityProvider, CapabilityRequest, ProviderResult } from '@insignia/iios-contracts';
+
+/**
+ * A real egress provider: POSTs the outbound command to a configured URL. Activated
+ * only when `IIOS_PROVIDER_URL_<CHANNELTYPE>` is set — proving "flip the binding →
+ * it really sends" without hardcoding any specific vendor. A transport failure is
+ * surfaced as FAILED (never thrown), per the adapter doctrine.
+ */
+export class HttpProvider implements CapabilityProvider {
+  readonly name = 'http';
+  readonly channelTypes: string[];
+  readonly capabilities = { canSend: true };
+
+  constructor(
+    channelType: string,
+    private readonly url: string,
+  ) {
+    this.channelTypes = [channelType];
+  }
+
+  async send(req: CapabilityRequest): Promise<ProviderResult> {
+    const started = Date.now();
+    try {
+      const res = await fetch(this.url, {
+        method: 'POST',
+        headers: { 'content-type': 'application/json', 'idempotency-key': req.idempotencyKey },
+        body: JSON.stringify({ channelType: req.channelType, target: req.target, payload: req.payload }),
+      });
+      const latencyMs = Date.now() - started;
+      if (!res.ok) return { providerRef: `http-${res.status}`, outcome: 'FAILED', errorCode: `HTTP_${res.status}`, latencyMs };
+      return { providerRef: `http-${req.channelType}-${req.idempotencyKey}`, outcome: 'SENT', latencyMs };
+    } catch (err) {
+      return { providerRef: 'http-error', outcome: 'FAILED', errorCode: (err as Error).message.slice(0, 60), latencyMs: Date.now() - started };
+    }
+  }
+}
@@ -0,0 +1,21 @@
+import type { CapabilityProvider, CapabilityRequest, ProviderResult } from '@insignia/iios-contracts';
+
+/**
+ * Default egress provider: a sandbox sink that records a deterministic providerRef
+ * and never touches the network (mirrors the P5 adapter sandbox). Idempotent by
+ * construction — the same request just yields another sim ref, never throws.
+ */
+export class SandboxProvider implements CapabilityProvider {
+  readonly name = 'sandbox';
+  readonly channelTypes: string[];
+  readonly capabilities = { canSend: true };
+  private seq = 0;
+
+  constructor(channelTypes: string[]) {
+    this.channelTypes = channelTypes;
+  }
+
+  async send(req: CapabilityRequest): Promise<ProviderResult> {
+    return { providerRef: `sim-${req.channelType}-${++this.seq}`, outcome: 'SENT', latencyMs: 1 };
+  }
+}