feat(data-plane): add signed managed writer bindings

services/external-data-plane/src/config.mjs

@@ -1,14 +1,54 @@
 import { readFileSync } from "node:fs";
+import {
+  loadManagedProvisionerPublicKeyFile,
+  validateManagedProvisionerAudience,
+  validateManagedProvisionerIdentity,
+} from "./managed-provisioner-auth.mjs";
 
 export function readConfig(env = process.env) {
   const provisionerApiEnabled = boolean(env.EXTERNAL_DATA_PLANE_PROVISIONING_ENABLED, false);
+  const managedProvisionerApiEnabled = boolean(env.EXTERNAL_DATA_PLANE_MANAGED_PROVISIONING_ENABLED, false);
+  const managedProvisionerMaxSkewSeconds = integer(
+    env.EXTERNAL_DATA_PLANE_MANAGED_PROVISIONER_MAX_SKEW_SECONDS,
+    60,
+    5,
+    300,
+  );
   const config = {
     port: integer(env.PORT, 18106, 1, 65535),
     databaseUrl: required(env.EXTERNAL_DATA_PLANE_DATABASE_URL, "EXTERNAL_DATA_PLANE_DATABASE_URL"),
     databasePoolSize: integer(env.EXTERNAL_DATA_PLANE_DATABASE_POOL_SIZE, 10, 1, 50),
     internalAccessToken: optional(env.NODEDC_INTERNAL_ACCESS_TOKEN),
     provisionerApiEnabled,
+    managedProvisionerApiEnabled,
     provisionerAccessToken: provisionerApiEnabled ? secretFile(env.EXTERNAL_DATA_PLANE_PROVISIONER_TOKEN_FILE) : "",
+    managedProvisionerPublicKey: managedProvisionerApiEnabled
+      ? loadManagedProvisionerPublicKeyFile(env.EXTERNAL_DATA_PLANE_MANAGED_PROVISIONER_PUBLIC_KEY_FILE)
+      : null,
+    managedProvisionerServiceId: managedProvisionerApiEnabled
+      ? validateManagedProvisionerIdentity(
+        required(env.EXTERNAL_DATA_PLANE_MANAGED_PROVISIONER_SERVICE_ID, "EXTERNAL_DATA_PLANE_MANAGED_PROVISIONER_SERVICE_ID"),
+        "service_id",
+      )
+      : "",
+    managedProvisionerKeyId: managedProvisionerApiEnabled
+      ? validateManagedProvisionerIdentity(
+        required(env.EXTERNAL_DATA_PLANE_MANAGED_PROVISIONER_KEY_ID, "EXTERNAL_DATA_PLANE_MANAGED_PROVISIONER_KEY_ID"),
+        "key_id",
+      )
+      : "",
+    managedProvisionerAudience: managedProvisionerApiEnabled
+      ? validateManagedProvisionerAudience(
+        required(env.EXTERNAL_DATA_PLANE_MANAGED_PROVISIONER_AUDIENCE, "EXTERNAL_DATA_PLANE_MANAGED_PROVISIONER_AUDIENCE"),
+      )
+      : "",
+    managedProvisionerMaxSkewMs: managedProvisionerMaxSkewSeconds * 1000,
+    managedProvisionerReplayCacheMaxEntries: integer(
+      env.EXTERNAL_DATA_PLANE_MANAGED_PROVISIONER_REPLAY_CACHE_MAX_ENTRIES,
+      10_000,
+      100,
+      100_000,
+    ),
     rawRetentionDays: integer(env.EXTERNAL_DATA_PLANE_RAW_RETENTION_DAYS, 14, 1, 3650),
     maxBatchBytes: integer(env.EXTERNAL_DATA_PLANE_MAX_BATCH_BYTES, 5 * 1024 * 1024, 1024, 50 * 1024 * 1024),
     maxFactsPerPublish: integer(env.EXTERNAL_DATA_PLANE_MAX_FACTS_PER_PUBLISH, 5000, 1, 100_000),

services/external-data-plane/src/data-product-delivery.mjs

@@ -2,7 +2,7 @@ import { createHash, randomUUID } from "node:crypto";
 import {
   DATA_PRODUCT_PATCH_SCHEMA_VERSION,
   DATA_PRODUCT_SNAPSHOT_SCHEMA_VERSION,
-} from "@nodedc/external-provider-contract";
+} from "@nodedc/external-provider-contract/data-plane";
 
 export async function loadDataProductDefinition(db, dataProductId, { activeOnly = true } = {}) {
   const result = await db.query(

services/external-data-plane/src/managed-provisioner-auth.mjs

@@ -0,0 +1,318 @@
+import { createHash, createPublicKey, timingSafeEqual, verify as verifySignature } from "node:crypto";
+import { closeSync, constants as fsConstants, fstatSync, lstatSync, openSync, readFileSync } from "node:fs";
+import { isAbsolute } from "node:path";
+
+export const MANAGED_PROVISIONER_SIGNATURE_SCHEMA = "nodedc.external-data-plane.managed-provisioner-request/v1";
+
+export const MANAGED_PROVISIONER_HEADERS = Object.freeze({
+  serviceId: "x-nodedc-engine-service-id",
+  keyId: "x-nodedc-engine-key-id",
+  audience: "x-nodedc-request-audience",
+  timestamp: "x-nodedc-request-timestamp",
+  nonce: "x-nodedc-request-nonce",
+  bodySha256: "x-nodedc-content-sha256",
+  signature: "x-nodedc-request-signature",
+});
+
+const ISO_TIMESTAMP = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/;
+const BASE64URL = /^[A-Za-z0-9_-]+$/;
+const IDENTITY = /^[a-z][a-z0-9._:-]{2,127}$/i;
+const AUDIENCE = /^[a-z][a-z0-9._:/-]{2,255}$/i;
+
+export function loadManagedProvisionerPublicKeyFile(pathValue) {
+  const path = String(pathValue ?? "").trim();
+  if (!path) throw configError("external_data_plane_managed_provisioner_public_key_file_required");
+  if (!isAbsolute(path)) throw configError("external_data_plane_managed_provisioner_public_key_file_must_be_absolute");
+
+  let stat;
+  try {
+    stat = lstatSync(path);
+  } catch {
+    throw configError("external_data_plane_managed_provisioner_public_key_file_unreadable");
+  }
+  if (!stat.isFile() || stat.isSymbolicLink()) {
+    throw configError("external_data_plane_managed_provisioner_public_key_file_must_be_regular");
+  }
+  if ((stat.mode & 0o022) !== 0) {
+    throw configError("external_data_plane_managed_provisioner_public_key_file_permissions_invalid");
+  }
+  if (stat.size < 1 || stat.size > 8192) {
+    throw configError("external_data_plane_managed_provisioner_public_key_file_size_invalid");
+  }
+
+  let descriptor;
+  try {
+    descriptor = openSync(path, fsConstants.O_RDONLY | fsConstants.O_NOFOLLOW);
+  } catch {
+    throw configError("external_data_plane_managed_provisioner_public_key_file_unreadable");
+  }
+  let pem;
+  try {
+    const openedStat = fstatSync(descriptor);
+    if (!openedStat.isFile()) {
+      throw configError("external_data_plane_managed_provisioner_public_key_file_must_be_regular");
+    }
+    if ((openedStat.mode & 0o022) !== 0) {
+      throw configError("external_data_plane_managed_provisioner_public_key_file_permissions_invalid");
+    }
+    if (openedStat.size < 1 || openedStat.size > 8192) {
+      throw configError("external_data_plane_managed_provisioner_public_key_file_size_invalid");
+    }
+    pem = readFileSync(descriptor, "utf8");
+  } catch (error) {
+    if (String(error?.code || "").startsWith("external_data_plane_managed_provisioner_")) throw error;
+    throw configError("external_data_plane_managed_provisioner_public_key_file_unreadable");
+  } finally {
+    closeSync(descriptor);
+  }
+  const normalizedPem = pem.replace(/\r\n/g, "\n").trim();
+  if (!/^-----BEGIN PUBLIC KEY-----\n[A-Za-z0-9+/=\n]+\n-----END PUBLIC KEY-----$/.test(normalizedPem)) {
+    throw configError("external_data_plane_managed_provisioner_public_key_file_format_invalid");
+  }
+
+  let publicKey;
+  try {
+    publicKey = createPublicKey(normalizedPem);
+  } catch {
+    throw configError("external_data_plane_managed_provisioner_public_key_file_format_invalid");
+  }
+  if (publicKey.type !== "public" || publicKey.asymmetricKeyType !== "ed25519") {
+    throw configError("external_data_plane_managed_provisioner_public_key_algorithm_invalid");
+  }
+  return publicKey;
+}
+
+export function validateManagedProvisionerIdentity(value, name) {
+  const normalized = String(value ?? "").trim();
+  if (!IDENTITY.test(normalized)) throw configError(`external_data_plane_managed_provisioner_${name}_invalid`);
+  return normalized;
+}
+
+export function validateManagedProvisionerAudience(value) {
+  const normalized = String(value ?? "").trim();
+  if (!AUDIENCE.test(normalized)) throw configError("external_data_plane_managed_provisioner_audience_invalid");
+  return normalized;
+}
+
+export function managedProvisionerSigningPayload({
+  audience,
+  serviceId,
+  keyId,
+  method,
+  path,
+  timestamp,
+  nonce,
+  bodySha256,
+}) {
+  return JSON.stringify({
+    schemaVersion: MANAGED_PROVISIONER_SIGNATURE_SCHEMA,
+    audience,
+    serviceId,
+    keyId,
+    method,
+    path,
+    timestamp,
+    nonce,
+    bodySha256,
+  });
+}
+
+export function sha256RawBody(rawBody) {
+  if (!Buffer.isBuffer(rawBody)) throw new TypeError("raw_body_buffer_required");
+  return createHash("sha256").update(rawBody).digest("hex");
+}
+
+export class ManagedProvisionerReplayCache {
+  constructor({ maxEntries }) {
+    if (!Number.isInteger(maxEntries) || maxEntries < 1) throw new TypeError("replay_cache_max_entries_invalid");
+    this.maxEntries = maxEntries;
+    this.entries = new Map();
+  }
+
+  consume(key, expiresAt, now = Date.now()) {
+    this.prune(now);
+    const existing = this.entries.get(key);
+    if (existing !== undefined && existing > now) {
+      throw authError(409, "managed_provisioner_request_replayed");
+    }
+    if (this.entries.size >= this.maxEntries) {
+      throw authError(503, "managed_provisioner_replay_cache_exhausted");
+    }
+    this.entries.set(key, expiresAt);
+  }
+
+  prune(now = Date.now()) {
+    for (const [key, expiresAt] of this.entries) {
+      if (expiresAt <= now) this.entries.delete(key);
+    }
+  }
+}
+
+export function createManagedProvisionerRequestVerifier({
+  publicKey,
+  serviceId,
+  keyId,
+  audience,
+  maxSkewMs,
+  replayCache,
+  now = Date.now,
+}) {
+  if (publicKey?.type !== "public" || publicKey?.asymmetricKeyType !== "ed25519") {
+    throw configError("external_data_plane_managed_provisioner_public_key_algorithm_invalid");
+  }
+  if (!IDENTITY.test(serviceId)) throw configError("external_data_plane_managed_provisioner_service_id_invalid");
+  if (!IDENTITY.test(keyId)) throw configError("external_data_plane_managed_provisioner_key_id_invalid");
+  if (!AUDIENCE.test(audience)) throw configError("external_data_plane_managed_provisioner_audience_invalid");
+  if (!Number.isInteger(maxSkewMs) || maxSkewMs < 1) {
+    throw configError("external_data_plane_managed_provisioner_max_skew_invalid");
+  }
+  if (!(replayCache instanceof ManagedProvisionerReplayCache)) {
+    throw configError("external_data_plane_managed_provisioner_replay_cache_invalid");
+  }
+
+  return function verifyManagedProvisionerRequest(req) {
+    const requestServiceId = exactHeader(req, MANAGED_PROVISIONER_HEADERS.serviceId);
+    const requestKeyId = exactHeader(req, MANAGED_PROVISIONER_HEADERS.keyId);
+    const requestAudience = exactHeader(req, MANAGED_PROVISIONER_HEADERS.audience);
+    const timestamp = exactHeader(req, MANAGED_PROVISIONER_HEADERS.timestamp);
+    const nonce = exactHeader(req, MANAGED_PROVISIONER_HEADERS.nonce);
+    const declaredBodySha256 = exactHeader(req, MANAGED_PROVISIONER_HEADERS.bodySha256);
+    const encodedSignature = exactHeader(req, MANAGED_PROVISIONER_HEADERS.signature);
+    if (hasHeader(req, "authorization")) {
+      throw authError(401, "managed_provisioner_authorization_header_forbidden");
+    }
+
+    if (!safeEqualText(requestServiceId, serviceId)
+      || !safeEqualText(requestKeyId, keyId)
+      || !safeEqualText(requestAudience, audience)) {
+      throw authError(401, "managed_provisioner_identity_mismatch");
+    }
+
+    const timestampMs = parseTimestamp(timestamp);
+    const verificationTime = Number(now());
+    if (!Number.isFinite(verificationTime) || Math.abs(verificationTime - timestampMs) >= maxSkewMs) {
+      throw authError(401, "managed_provisioner_timestamp_outside_window");
+    }
+    const nonceBytes = decodeCanonicalBase64Url(nonce, null, "managed_provisioner_nonce_invalid");
+    if (nonceBytes.length < 16 || nonceBytes.length > 96) throw authError(401, "managed_provisioner_nonce_invalid");
+    if (!/^[a-f0-9]{64}$/.test(declaredBodySha256)) {
+      throw authError(401, "managed_provisioner_body_hash_invalid");
+    }
+
+    const method = exactMethod(req);
+    const path = exactPath(req);
+    const rawBody = exactRawBody(req);
+    const actualBodySha256 = sha256RawBody(rawBody);
+    if (!safeEqualText(declaredBodySha256, actualBodySha256)) {
+      throw authError(401, "managed_provisioner_body_hash_mismatch");
+    }
+
+    const signature = decodeCanonicalBase64Url(encodedSignature, 64, "managed_provisioner_signature_invalid");
+    const payload = managedProvisionerSigningPayload({
+      audience: requestAudience,
+      serviceId: requestServiceId,
+      keyId: requestKeyId,
+      method,
+      path,
+      timestamp,
+      nonce,
+      bodySha256: declaredBodySha256,
+    });
+    if (!verifySignature(null, Buffer.from(payload, "utf8"), publicKey, signature)) {
+      throw authError(401, "managed_provisioner_signature_invalid");
+    }
+
+    const replayKey = `${requestServiceId}\0${requestKeyId}\0${nonce}`;
+    replayCache.consume(replayKey, timestampMs + maxSkewMs, verificationTime);
+    return Object.freeze({ serviceId: requestServiceId, keyId: requestKeyId });
+  };
+}
+
+function exactHeader(req, name) {
+  const rawHeaders = req?.rawHeaders;
+  if (Array.isArray(rawHeaders)) {
+    const values = [];
+    for (let index = 0; index < rawHeaders.length; index += 2) {
+      if (String(rawHeaders[index]).toLowerCase() === name) values.push(rawHeaders[index + 1]);
+    }
+    if (values.length !== 1 || typeof values[0] !== "string" || !values[0]) {
+      throw authError(401, "managed_provisioner_header_invalid");
+    }
+    return values[0];
+  }
+
+  const value = req?.headers?.[name];
+  if (typeof value !== "string" || !value) throw authError(401, "managed_provisioner_header_invalid");
+  return value;
+}
+
+function hasHeader(req, name) {
+  if (Array.isArray(req?.rawHeaders)) {
+    for (let index = 0; index < req.rawHeaders.length; index += 2) {
+      if (String(req.rawHeaders[index]).toLowerCase() === name) return true;
+    }
+    return false;
+  }
+  return Object.hasOwn(req?.headers || {}, name);
+}
+
+function exactMethod(req) {
+  const method = String(req?.method ?? "");
+  if (!/^[A-Z]{3,10}$/.test(method)) throw authError(401, "managed_provisioner_method_invalid");
+  return method;
+}
+
+function exactPath(req) {
+  const path = String(req?.originalUrl ?? "");
+  if (!path.startsWith("/") || path.length > 2048 || /[\u0000-\u0020#]/.test(path)) {
+    throw authError(401, "managed_provisioner_path_invalid");
+  }
+  return path;
+}
+
+function exactRawBody(req) {
+  if (req?.rawBodyCaptured === true && Buffer.isBuffer(req.rawBody)) return req.rawBody;
+  const contentLength = req?.headers?.["content-length"];
+  const transferEncoding = req?.headers?.["transfer-encoding"];
+  if ((contentLength === undefined || contentLength === "0") && transferEncoding === undefined) {
+    return Buffer.alloc(0);
+  }
+  throw authError(401, "managed_provisioner_raw_body_unavailable");
+}
+
+function parseTimestamp(value) {
+  if (!ISO_TIMESTAMP.test(value)) throw authError(401, "managed_provisioner_timestamp_invalid");
+  const timestampMs = Date.parse(value);
+  if (!Number.isFinite(timestampMs) || new Date(timestampMs).toISOString() !== value) {
+    throw authError(401, "managed_provisioner_timestamp_invalid");
+  }
+  return timestampMs;
+}
+
+function decodeCanonicalBase64Url(value, expectedBytes, errorCode) {
+  if (!BASE64URL.test(value)) throw authError(401, errorCode);
+  let decoded;
+  try {
+    decoded = Buffer.from(value, "base64url");
+  } catch {
+    throw authError(401, errorCode);
+  }
+  if ((expectedBytes !== null && decoded.length !== expectedBytes) || decoded.toString("base64url") !== value) {
+    throw authError(401, errorCode);
+  }
+  return decoded;
+}
+
+function safeEqualText(left, right) {
+  const leftBuffer = Buffer.from(String(left), "utf8");
+  const rightBuffer = Buffer.from(String(right), "utf8");
+  return leftBuffer.length === rightBuffer.length && timingSafeEqual(leftBuffer, rightBuffer);
+}
+
+function authError(status, code) {
+  return Object.assign(new Error(code), { status, code });
+}
+
+function configError(code) {
+  return Object.assign(new Error(code), { code });
+}

services/external-data-plane/src/schema.mjs

@@ -65,6 +65,9 @@ export async function migrate(pool) {
     create table if not exists external_data_plane_writer_bindings (
       id uuid primary key,
       token_hash text not null unique,
+      binding_key text,
+      request_hash text,
+      generation integer not null default 1,
       tenant_id text not null,
       connection_id text not null,
       provider_id text not null,
@@ -74,6 +77,7 @@ export async function migrate(pool) {
       created_at timestamptz not null default now(),
       rotated_at timestamptz,
       revoked_at timestamptz,
+      check (generation > 0),
       check (
         case when jsonb_typeof(allowed_data_product_ids) = 'array'
           then jsonb_array_length(allowed_data_product_ids) > 0
@@ -82,6 +86,37 @@ export async function migrate(pool) {
       )
     )
   `);
+  await pool.query("alter table external_data_plane_writer_bindings add column if not exists binding_key text");
+  await pool.query("alter table external_data_plane_writer_bindings add column if not exists request_hash text");
+  await pool.query("alter table external_data_plane_writer_bindings add column if not exists generation integer not null default 1");
+  await pool.query(`
+    do $$
+    begin
+      if not exists (
+        select 1 from pg_constraint
+        where conrelid = 'external_data_plane_writer_bindings'::regclass
+          and conname = 'external_data_plane_writer_bindings_generation_positive_ck'
+      ) then
+        alter table external_data_plane_writer_bindings
+          add constraint external_data_plane_writer_bindings_generation_positive_ck
+          check (generation > 0);
+      end if;
+      if not exists (
+        select 1 from pg_constraint
+        where conrelid = 'external_data_plane_writer_bindings'::regclass
+          and conname = 'external_data_plane_writer_bindings_managed_metadata_ck'
+      ) then
+        alter table external_data_plane_writer_bindings
+          add constraint external_data_plane_writer_bindings_managed_metadata_ck
+          check (
+            (binding_key is null and request_hash is null)
+            or (binding_key is not null and request_hash ~ '^[a-f0-9]{64}$')
+          );
+      end if;
+    end
+    $$
+  `);
+  await pool.query("create unique index if not exists external_data_plane_writer_bindings_managed_key_idx on external_data_plane_writer_bindings (binding_key, generation) where binding_key is not null");
   await pool.query("create index if not exists external_data_plane_writer_bindings_active_idx on external_data_plane_writer_bindings (active, expires_at)");
 
   await pool.query(`

services/external-data-plane/src/server.mjs

@@ -2,7 +2,7 @@ import express from "express";
 import { createHash, randomUUID, timingSafeEqual } from "node:crypto";
 import { createServer } from "node:http";
 import { Pool } from "pg";
-import { validateDataProductPublish, validateIntakeBatch } from "@nodedc/external-provider-contract";
+import { validateDataProductPublish, validateIntakeBatch } from "@nodedc/external-provider-contract/data-plane";
 import { readConfig } from "./config.mjs";
 import {
   loadDataProductDefinition,
@@ -18,6 +18,10 @@ import {
 import { normalizeDataProductDefinition, safeDataProductDefinition } from "./data-product-policy.mjs";
 import { reconcileDataProductDefinitions } from "./definitions.mjs";
 import { assertBatchTimeBounds, rawRetentionExpiry } from "./intake-policy.mjs";
+import {
+  createManagedProvisionerRequestVerifier,
+  ManagedProvisionerReplayCache,
+} from "./managed-provisioner-auth.mjs";
 import {
   assertReaderProduct,
   createReaderToken,
@@ -31,11 +35,26 @@ import {
   hashWriterToken,
   materializeDataProductPublish,
   materializeWriterBoundBatch,
+  normalizeManagedWriterBindingRequest,
   normalizeWriterBindingRequest,
   safeWriterBinding,
+  writerBindingRequestHash,
 } from "./writer-binding.mjs";
 
 const config = readConfig();
+const managedProvisionerReplayCache = config.managedProvisionerApiEnabled
+  ? new ManagedProvisionerReplayCache({ maxEntries: config.managedProvisionerReplayCacheMaxEntries })
+  : null;
+const verifyManagedProvisionerRequest = config.managedProvisionerApiEnabled
+  ? createManagedProvisionerRequestVerifier({
+    publicKey: config.managedProvisionerPublicKey,
+    serviceId: config.managedProvisionerServiceId,
+    keyId: config.managedProvisionerKeyId,
+    audience: config.managedProvisionerAudience,
+    maxSkewMs: config.managedProvisionerMaxSkewMs,
+    replayCache: managedProvisionerReplayCache,
+  })
+  : null;
 const pool = new Pool({ connectionString: config.databaseUrl, max: config.databasePoolSize });
 const app = express();
 const httpServer = createServer(app);
@@ -47,7 +66,19 @@ let shuttingDown = false;
 let shutdownPromise = null;
 
 app.disable("x-powered-by");
-app.use(express.json({ limit: config.maxBatchBytes }));
+app.use((req, _res, next) => {
+  req.rawBody = Buffer.alloc(0);
+  req.rawBodyCaptured = false;
+  next();
+});
+app.use(express.json({
+  limit: config.maxBatchBytes,
+  inflate: false,
+  verify(req, _res, buffer) {
+    req.rawBody = Buffer.from(buffer);
+    req.rawBodyCaptured = true;
+  },
+}));
 
 app.get("/healthz", asyncRoute(async (_req, res) => {
   await pool.query("select 1");
@@ -59,10 +90,12 @@ app.get("/healthz", asyncRoute(async (_req, res) => {
     providerLogic: "absent",
     commandTransport: "absent",
     writerBindings: "supported",
+    managedWriterBindings: "digest+idempotent-generation",
     readerBindings: "supported",
     dataProductDelivery: "snapshot+durable-patch",
     legacyIntake: config.legacyIntakeEnabled ? "migration-only" : "disabled",
-    writerBindingProvisioning: config.provisionerApiEnabled ? "enabled" : "disabled",
+    legacyCapabilityProvisioning: config.provisionerApiEnabled ? "enabled" : "disabled",
+    managedWriterBindingProvisioning: config.managedProvisionerApiEnabled ? "enabled" : "disabled",
     rawRetentionSweep: {
       mode: "server-scheduled",
       lastSweepAt: lastRetentionSweepAt,
@@ -134,6 +167,128 @@ app.post("/internal/data-plane/v1/intake/writer-bound", requireLegacyIntake, req
   res.status(result.idempotent ? 200 : 201).json({ ok: true, ...result });
 }));
 
+app.put("/internal/data-plane/v1/writer-bindings/by-key/:bindingKey", requireManagedProvisionerApi, asyncRoute(async (req, res) => {
+  const bindingKey = requireIdentifier(req.params.bindingKey, "managed_writer_binding_key_invalid");
+  const policy = normalizeManagedWriterBindingRequest(req.body, {
+    maxTtlDays: config.writerBindingMaxTtlDays,
+  });
+  await assertRegisteredProductIds(policy.allowedDataProductIds);
+  const requestHash = writerBindingRequestHash(policy);
+  const client = await pool.connect();
+  let response;
+  try {
+    await client.query("begin");
+    await client.query("select pg_advisory_xact_lock(hashtextextended($1, 0))", [bindingKey]);
+    const existing = await client.query(
+      `select id, binding_key as "bindingKey", request_hash as "requestHash", generation,
+         tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
+         allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
+         active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"
+       from external_data_plane_writer_bindings
+       where binding_key = $1 and generation = $2
+       for update`,
+      [bindingKey, policy.generation],
+    );
+    if (existing.rowCount) {
+      const binding = existing.rows[0];
+      if (binding.requestHash !== requestHash) {
+        throw httpError(409, "managed_writer_binding_request_conflict");
+      }
+      if (binding.active !== true || new Date(binding.expiresAt) <= new Date()) {
+        throw httpError(409, "managed_writer_binding_generation_inactive");
+      }
+      response = { status: 200, idempotent: true, binding };
+    } else {
+      const inserted = await client.query(
+        `insert into external_data_plane_writer_bindings (
+          id, token_hash, binding_key, request_hash, generation,
+          tenant_id, connection_id, provider_id, allowed_data_product_ids, expires_at
+        ) values ($1, $2, $3, $4, $5, $6, $7, $8, $9::jsonb, $10)
+        returning id, binding_key as "bindingKey", generation,
+          tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
+          allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
+          active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
+        [
+          randomUUID(),
+          policy.capabilityDigest,
+          bindingKey,
+          requestHash,
+          policy.generation,
+          policy.tenantId,
+          policy.connectionId,
+          policy.providerId,
+          JSON.stringify(policy.allowedDataProductIds),
+          policy.expiresAt,
+        ],
+      );
+      response = { status: 201, idempotent: false, binding: inserted.rows[0] };
+    }
+    await client.query("commit");
+  } catch (error) {
+    await client.query("rollback");
+    if (error?.code === "23505") {
+      throw httpError(409, "managed_writer_binding_capability_digest_conflict");
+    }
+    throw error;
+  } finally {
+    client.release();
+  }
+  res.set("Cache-Control", "no-store, max-age=0");
+  res.status(response.status).json({
+    ok: true,
+    idempotent: response.idempotent,
+    writerBinding: safeWriterBinding(response.binding),
+  });
+}));
+
+app.post("/internal/data-plane/v1/writer-bindings/by-key/:bindingKey/generations/:generation/revoke", requireManagedProvisionerApi, asyncRoute(async (req, res) => {
+  const bindingKey = requireIdentifier(req.params.bindingKey, "managed_writer_binding_key_invalid");
+  const generation = requirePositiveInteger(req.params.generation, "managed_writer_binding_generation_invalid");
+  const client = await pool.connect();
+  let binding;
+  let idempotent;
+  try {
+    await client.query("begin");
+    await client.query("select pg_advisory_xact_lock(hashtextextended($1, 0))", [bindingKey]);
+    const existing = await client.query(
+      `select id, binding_key as "bindingKey", generation,
+         tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
+         allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
+         active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"
+       from external_data_plane_writer_bindings
+       where binding_key = $1 and generation = $2
+       for update`,
+      [bindingKey, generation],
+    );
+    if (!existing.rowCount) throw httpError(404, "managed_writer_binding_not_found");
+    if (existing.rows[0].active === true) {
+      const revoked = await client.query(
+        `update external_data_plane_writer_bindings
+         set active = false, revoked_at = now()
+         where binding_key = $1 and generation = $2 and active = true
+         returning id, binding_key as "bindingKey", generation,
+           tenant_id as "tenantId", connection_id as "connectionId", provider_id as "providerId",
+           allowed_data_product_ids as "allowedDataProductIds", expires_at as "expiresAt",
+           active, created_at as "createdAt", rotated_at as "rotatedAt", revoked_at as "revokedAt"`,
+        [bindingKey, generation],
+      );
+      binding = revoked.rows[0];
+      idempotent = false;
+    } else {
+      binding = existing.rows[0];
+      idempotent = true;
+    }
+    await client.query("commit");
+  } catch (error) {
+    await client.query("rollback");
+    throw error;
+  } finally {
+    client.release();
+  }
+  res.set("Cache-Control", "no-store, max-age=0");
+  res.json({ ok: true, idempotent, writerBinding: safeWriterBinding(binding) });
+}));
+
 app.post("/internal/data-plane/v1/writer-bindings", requireProvisionerApi, asyncRoute(async (req, res) => {
   const policy = normalizeWriterBindingRequest(req.body, {
     maxTtlDays: config.writerBindingMaxTtlDays,
@@ -172,7 +327,7 @@ app.post("/internal/data-plane/v1/writer-bindings/:bindingId/rotate", requirePro
   const result = await pool.query(
     `update external_data_plane_writer_bindings
      set token_hash = $2, rotated_at = now()
-     where id = $1 and active = true and expires_at > now()
+     where id = $1 and binding_key is null and active = true and expires_at > now()
      returning id, tenant_id as "tenantId", connection_id as "connectionId",
        provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
        expires_at as "expiresAt", active, created_at as "createdAt",
@@ -188,7 +343,7 @@ app.post("/internal/data-plane/v1/writer-bindings/:bindingId/revoke", requirePro
   const result = await pool.query(
     `update external_data_plane_writer_bindings
      set active = false, revoked_at = now()
-     where id = $1 and active = true
+     where id = $1 and binding_key is null and active = true
      returning id, tenant_id as "tenantId", connection_id as "connectionId",
        provider_id as "providerId", allowed_data_product_ids as "allowedDataProductIds",
        expires_at as "expiresAt", active, created_at as "createdAt",
@@ -564,6 +719,21 @@ function requireLegacyIntake(_req, _res, next) {
 
 function requireProvisionerApi(req, _res, next) {
   if (!config.provisionerApiEnabled) return next(httpError(503, "provisioner_api_disabled"));
+  return requireProvisionerCredential(req, next);
+}
+
+function requireManagedProvisionerApi(req, _res, next) {
+  if (!config.managedProvisionerApiEnabled) return next(httpError(503, "managed_provisioner_api_disabled"));
+  if (!verifyManagedProvisionerRequest) return next(httpError(503, "managed_provisioner_api_not_configured"));
+  try {
+    req.managedProvisionerIdentity = verifyManagedProvisionerRequest(req);
+    return next();
+  } catch (error) {
+    return next(error);
+  }
+}
+
+function requireProvisionerCredential(req, next) {
   if (!config.provisionerAccessToken) return next(httpError(503, "provisioner_api_not_configured"));
   const value = bearerToken(req);
   if (!value || !safeEqual(value, config.provisionerAccessToken)) return next(httpError(401, "provisioner_unauthorized"));
@@ -697,6 +867,14 @@ function requireIdentifier(value, code) {
   return normalized;
 }
 
+function requirePositiveInteger(value, code) {
+  const normalized = String(value ?? "");
+  if (!/^[1-9]\d*$/.test(normalized)) throw httpError(400, code);
+  const number = Number(normalized);
+  if (!Number.isSafeInteger(number) || number > 2_147_483_647) throw httpError(400, code);
+  return number;
+}
+
 function parseCursor(value) {
   const normalized = String(value ?? "");
   if (!/^(?:0|[1-9]\d*)$/.test(normalized)) throw httpError(400, "data_product_cursor_invalid");

services/external-data-plane/src/writer-binding.mjs

@@ -4,7 +4,15 @@ const IDENTIFIER = /^[a-z][a-z0-9._:-]{2,127}$/;
 const TOKEN_PREFIX = "ndc_edpwb_";
 const SECRET_LIKE_KEY = /(token|secret|password|access[_-]?token|refresh[_-]?token|api[_-]?key)/i;
 const BINDING_REQUEST_KEYS = new Set(["source", "allowedDataProductIds", "expiresAt"]);
+const MANAGED_BINDING_REQUEST_KEYS = new Set([
+  "source",
+  "allowedDataProductIds",
+  "expiresAt",
+  "generation",
+  "capabilityDigest",
+]);
 const BINDING_SOURCE_KEYS = new Set(["tenantId", "connectionId", "providerId"]);
+const SHA256_DIGEST = /^[a-f0-9]{64}$/;
 
 /**
  * Produces an opaque, high-entropy capability. The plaintext is returned only
@@ -59,6 +67,57 @@ export function normalizeWriterBindingRequest(value, { now = new Date(), maxTtlD
   });
 }
 
+/**
+ * Validates the zero-touch control-plane form. The native credential secret is
+ * generated and stored inside Engine; EDP receives only its SHA-256 digest.
+ * `generation` makes a retry address the same immutable binding generation.
+ */
+export function normalizeManagedWriterBindingRequest(value, { now = new Date(), maxTtlDays = 90 } = {}) {
+  if (!isPlainObject(value) || !isPlainObject(value.source)) {
+    throw writerBindingError("managed_writer_binding_request_invalid");
+  }
+  if (containsSecretLikeKey(value)) {
+    throw writerBindingError("managed_writer_binding_secret_material_forbidden");
+  }
+  if (!hasOnlyKeys(value, MANAGED_BINDING_REQUEST_KEYS) || !hasOnlyKeys(value.source, BINDING_SOURCE_KEYS)) {
+    throw writerBindingError("managed_writer_binding_request_fields_invalid");
+  }
+
+  const scope = normalizeWriterBindingRequest({
+    source: value.source,
+    allowedDataProductIds: value.allowedDataProductIds,
+    expiresAt: value.expiresAt,
+  }, { now, maxTtlDays });
+  const generation = Number(value.generation);
+  if (!Number.isSafeInteger(generation) || generation < 1 || generation > 2_147_483_647) {
+    throw writerBindingError("managed_writer_binding_generation_invalid");
+  }
+  const capabilityDigest = String(value.capabilityDigest || "").toLowerCase();
+  if (!SHA256_DIGEST.test(capabilityDigest)) {
+    throw writerBindingError("managed_writer_binding_capability_digest_invalid");
+  }
+
+  return Object.freeze({
+    ...scope,
+    allowedDataProductIds: Object.freeze([...scope.allowedDataProductIds].sort()),
+    generation,
+    capabilityDigest,
+  });
+}
+
+export function writerBindingRequestHash(policy) {
+  const canonical = JSON.stringify({
+    tenantId: policy.tenantId,
+    connectionId: policy.connectionId,
+    providerId: policy.providerId,
+    allowedDataProductIds: [...policy.allowedDataProductIds].sort(),
+    expiresAt: new Date(policy.expiresAt).toISOString(),
+    generation: policy.generation,
+    capabilityDigest: policy.capabilityDigest,
+  });
+  return createHash("sha256").update(canonical, "utf8").digest("hex");
+}
+
 /**
  * Converts a caller-provided, deliberately unscoped intake envelope into the
  * canonical scoped form. Caller scope is rejected, never trusted or merged.
@@ -136,6 +195,8 @@ export function materializeDataProductPublish(value, binding, definition, dataPr
 export function safeWriterBinding(binding) {
   return {
     id: binding.id,
+    ...(binding.bindingKey ? { bindingKey: binding.bindingKey } : {}),
+    ...(Number.isInteger(Number(binding.generation)) ? { generation: Number(binding.generation) } : {}),
     tenantId: binding.tenantId,
     connectionId: binding.connectionId,
     providerId: binding.providerId,