SILVER/NODEDC_PLATFORM
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(edp): provision Foundry reader grants

Browse Source
This commit is contained in:
Codex
2026-07-19 15:03:48 +03:00
parent 847a08da93
commit def9a24e0d
17 changed files with 891 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/external-data-plane/src/config.mjs
+34
View File
@@ -8,6 +8,7 @@ import {
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 foundryProvisionerApiEnabled = boolean(env.EXTERNAL_DATA_PLANE_FOUNDRY_PROVISIONING_ENABLED, false);
const managedProvisionerMaxSkewSeconds = integer(
env.EXTERNAL_DATA_PLANE_MANAGED_PROVISIONER_MAX_SKEW_SECONDS,
60,
@@ -21,6 +22,7 @@ export function readConfig(env = process.env) {
internalAccessToken: optional(env.NODEDC_INTERNAL_ACCESS_TOKEN),
provisionerApiEnabled,
managedProvisionerApiEnabled,
foundryProvisionerApiEnabled,
provisionerAccessToken: provisionerApiEnabled ? secretFile(env.EXTERNAL_DATA_PLANE_PROVISIONER_TOKEN_FILE) : "",
managedProvisionerPublicKey: managedProvisionerApiEnabled
? loadManagedProvisionerPublicKeyFile(env.EXTERNAL_DATA_PLANE_MANAGED_PROVISIONER_PUBLIC_KEY_FILE)
@@ -49,6 +51,38 @@ export function readConfig(env = process.env) {
100,
100_000,
),
foundryProvisionerPublicKey: foundryProvisionerApiEnabled
? loadManagedProvisionerPublicKeyFile(env.EXTERNAL_DATA_PLANE_FOUNDRY_PROVISIONER_PUBLIC_KEY_FILE)
: null,
foundryProvisionerServiceId: foundryProvisionerApiEnabled
? validateManagedProvisionerIdentity(
required(env.EXTERNAL_DATA_PLANE_FOUNDRY_PROVISIONER_SERVICE_ID, "EXTERNAL_DATA_PLANE_FOUNDRY_PROVISIONER_SERVICE_ID"),
"service_id",
)
: "",
foundryProvisionerKeyId: foundryProvisionerApiEnabled
? validateManagedProvisionerIdentity(
required(env.EXTERNAL_DATA_PLANE_FOUNDRY_PROVISIONER_KEY_ID, "EXTERNAL_DATA_PLANE_FOUNDRY_PROVISIONER_KEY_ID"),
"key_id",
)
: "",
foundryProvisionerAudience: foundryProvisionerApiEnabled
? validateManagedProvisionerAudience(
required(env.EXTERNAL_DATA_PLANE_FOUNDRY_PROVISIONER_AUDIENCE, "EXTERNAL_DATA_PLANE_FOUNDRY_PROVISIONER_AUDIENCE"),
)
: "",
foundryProvisionerMaxSkewMs: integer(
env.EXTERNAL_DATA_PLANE_FOUNDRY_PROVISIONER_MAX_SKEW_SECONDS,
60,
5,
300,
) * 1000,
foundryProvisionerReplayCacheMaxEntries: integer(
env.EXTERNAL_DATA_PLANE_FOUNDRY_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/reader-binding.mjs
+63
View File
@@ -11,6 +11,13 @@ const MANAGED_REQUEST_KEYS = new Set([
"generation",
"capabilityDigest",
]);
const MANAGED_CONSUMER_REQUEST_KEYS = new Set([
"allowedDataProductIds",
"expiresAt",
"generation",
"capabilityDigest",
]);
const MANAGED_CONSUMER_PLAN_KEYS = new Set(["allowedDataProductIds"]);
const SOURCE_KEYS = new Set(["tenantId", "connectionId", "providerId"]);
const SHA256_DIGEST = /^[a-f0-9]{64}$/;
@@ -82,6 +89,38 @@ export function normalizeManagedReaderBindingRequest(value) {
});
}
export function normalizeManagedConsumerReaderPlanRequest(value) {
if (!isPlainObject(value) || containsSecretLikeKey(value) || !hasOnlyKeys(value, MANAGED_CONSUMER_PLAN_KEYS)) {
throw readerError("managed_consumer_reader_plan_request_invalid");
}
const allowedDataProductIds = uniqueIdentifiers(value.allowedDataProductIds);
if (!allowedDataProductIds.length) throw readerError("managed_consumer_reader_scope_invalid");
return Object.freeze({ allowedDataProductIds: Object.freeze([...allowedDataProductIds].sort()) });
}
export function normalizeManagedConsumerReaderBindingRequest(value) {
if (!isPlainObject(value) || containsSecretLikeKey(value) || !hasOnlyKeys(value, MANAGED_CONSUMER_REQUEST_KEYS)) {
throw readerError("managed_consumer_reader_binding_request_invalid");
}
const allowedDataProductIds = uniqueIdentifiers(value.allowedDataProductIds);
if (!allowedDataProductIds.length) throw readerError("managed_consumer_reader_scope_invalid");
if (value.expiresAt !== null) throw readerError("managed_consumer_reader_must_be_durable");
const generation = Number(value.generation);
if (!Number.isSafeInteger(generation) || generation < 1 || generation > 2_147_483_647) {
throw readerError("managed_consumer_reader_generation_invalid");
}
const capabilityDigest = String(value.capabilityDigest || "").toLowerCase();
if (!SHA256_DIGEST.test(capabilityDigest)) {
throw readerError("managed_consumer_reader_capability_digest_invalid");
}
return Object.freeze({
allowedDataProductIds: Object.freeze([...allowedDataProductIds].sort()),
expiresAt: null,
generation,
capabilityDigest,
});
}
export function readerBindingRequestHash(policy) {
const canonical = JSON.stringify({
tenantId: policy.tenantId,
@@ -95,6 +134,15 @@ export function readerBindingRequestHash(policy) {
return createHash("sha256").update(canonical, "utf8").digest("hex");
}
export function consumerReaderBindingRequestHash(policy) {
return createHash("sha256").update(JSON.stringify({
allowedDataProductIds: [...policy.allowedDataProductIds].sort(),
expiresAt: null,
generation: policy.generation,
capabilityDigest: policy.capabilityDigest,
}), "utf8").digest("hex");
}
export function assertReaderProduct(binding, dataProductId, now = new Date()) {
const expired = binding?.expiresAt !== null && new Date(binding?.expiresAt) <= now;
if (!isPlainObject(binding) || binding.active !== true || expired) {
@@ -124,6 +172,21 @@ export function safeReaderBinding(binding) {
};
}
export function safeManagedConsumerReaderBinding(binding) {
return {
id: binding.id,
bindingKey: binding.bindingKey,
generation: Number(binding.generation),
allowedDataProductIds: uniqueIdentifiers(binding.allowedDataProductIds),
active: binding.active === true,
expiresAt: binding.expiresAt === null ? null : new Date(binding.expiresAt).toISOString(),
createdAt: binding.createdAt ? new Date(binding.createdAt).toISOString() : undefined,
rotatedAt: binding.rotatedAt ? new Date(binding.rotatedAt).toISOString() : undefined,
revokedAt: binding.revokedAt ? new Date(binding.revokedAt).toISOString() : undefined,
sourceScope: "resolved-server-side",
};
}
function readerError(code, status = 400) {
return Object.assign(new Error(code), { status, code });
}
services/external-data-plane/src/reader-source-scope.mjs
+41
View File
@@ -41,6 +41,47 @@ export async function resolveManagedReaderSourceConnection(db, policy) {
throw sourceScopeError("managed_reader_source_scope_ambiguous", 409);
}
export async function resolveManagedReaderScope(db, allowedDataProductIdsValue) {
const allowedDataProductIds = uniqueIdentifiers(allowedDataProductIdsValue);
if (!allowedDataProductIds.length) {
throw sourceScopeError("managed_consumer_reader_scope_request_invalid", 400);
}
const result = await db.query(
`select distinct candidate.tenant_id as "tenantId",
candidate.connection_id as "connectionId", candidate.provider_id as "providerId"
from external_data_plane_writer_bindings as candidate
where candidate.active = true
and (candidate.expires_at is null or candidate.expires_at > now())
and not exists (
select 1
from unnest($1::text[]) as requested(data_product_id)
where not exists (
select 1
from external_data_plane_writer_bindings as coverage
where coverage.tenant_id = candidate.tenant_id
and coverage.connection_id = candidate.connection_id
and coverage.provider_id = candidate.provider_id
and coverage.active = true
and (coverage.expires_at is null or coverage.expires_at > now())
and coverage.allowed_data_product_ids ? requested.data_product_id
)
)
order by candidate.tenant_id asc, candidate.provider_id asc, candidate.connection_id asc
limit 3`,
[allowedDataProductIds],
);
const candidates = result.rows
.map((row) => ({
tenantId: identifier(row.tenantId),
connectionId: identifier(row.connectionId),
providerId: identifier(row.providerId),
}))
.filter((row) => row.tenantId && row.connectionId && row.providerId);
if (candidates.length === 1) return Object.freeze(candidates[0]);
if (!candidates.length) throw sourceScopeError("managed_consumer_reader_source_scope_not_found", 409);
throw sourceScopeError("managed_consumer_reader_source_scope_ambiguous", 409);
}
export async function backfillManagedReaderSourceConnections(db) {
const unresolved = await db.query(
`select id, tenant_id as "tenantId", connection_id as "connectionId",
services/external-data-plane/src/server.mjs
+168 -1
View File
@@ -25,14 +25,18 @@ import {
} from "./managed-provisioner-auth.mjs";
import {
assertReaderProduct,
consumerReaderBindingRequestHash,
createReaderToken,
hashReaderToken,
normalizeManagedConsumerReaderBindingRequest,
normalizeManagedConsumerReaderPlanRequest,
normalizeManagedReaderBindingRequest,
normalizeReaderBindingRequest,
readerBindingRequestHash,
safeManagedConsumerReaderBinding,
safeReaderBinding,
} from "./reader-binding.mjs";
import { resolveManagedReaderSourceConnection } from "./reader-source-scope.mjs";
import { resolveManagedReaderScope, resolveManagedReaderSourceConnection } from "./reader-source-scope.mjs";
import { migrate } from "./schema.mjs";
import {
createWriterToken,
@@ -60,6 +64,19 @@ const verifyManagedProvisionerRequest = config.managedProvisionerApiEnabled
replayCache: managedProvisionerReplayCache,
})
: null;
const foundryProvisionerReplayCache = config.foundryProvisionerApiEnabled
? new ManagedProvisionerReplayCache({ maxEntries: config.foundryProvisionerReplayCacheMaxEntries })
: null;
const verifyFoundryProvisionerRequest = config.foundryProvisionerApiEnabled
? createManagedProvisionerRequestVerifier({
publicKey: config.foundryProvisionerPublicKey,
serviceId: config.foundryProvisionerServiceId,
keyId: config.foundryProvisionerKeyId,
audience: config.foundryProvisionerAudience,
maxSkewMs: config.foundryProvisionerMaxSkewMs,
replayCache: foundryProvisionerReplayCache,
})
: null;
const pool = new Pool({ connectionString: config.databaseUrl, max: config.databasePoolSize });
const app = express();
const httpServer = createServer(app);
@@ -106,6 +123,8 @@ app.get("/healthz", asyncRoute(async (_req, res) => {
managedWriterBindingLifetime: config.managedProvisionerApiEnabled ? "explicit-revoke" : "disabled",
managedReaderBindingProvisioning: config.managedProvisionerApiEnabled ? "enabled" : "disabled",
managedReaderBindingLifetime: config.managedProvisionerApiEnabled ? "explicit-revoke" : "disabled",
foundryReaderBindingProvisioning: config.foundryProvisionerApiEnabled ? "digest+server-resolved-source" : "disabled",
foundryReaderBindingLifetime: config.foundryProvisionerApiEnabled ? "explicit-revoke" : "disabled",
rawRetentionSweep: {
mode: "server-scheduled",
lastSweepAt: lastRetentionSweepAt,
@@ -378,6 +397,143 @@ app.post("/internal/data-plane/v1/writer-bindings/:bindingId/revoke", requirePro
res.json({ ok: true, writerBinding: safeWriterBinding(result.rows[0]) });
}));
app.post("/internal/data-plane/v1/consumer-reader-bindings/plan", requireFoundryProvisionerApi, asyncRoute(async (req, res) => {
const policy = normalizeManagedConsumerReaderPlanRequest(req.body);
await assertRegisteredProductIds(policy.allowedDataProductIds);
await resolveManagedReaderScope(pool, policy.allowedDataProductIds);
const dataProducts = [];
for (const dataProductId of policy.allowedDataProductIds) {
const definition = await loadDataProductDefinition(pool, dataProductId);
if (!definition || definition.active === false) throw httpError(404, "data_product_not_found");
dataProducts.push(safeDataProductDefinition(definition));
}
res.set("Cache-Control", "no-store, max-age=0");
res.json({ ok: true, sourceScope: "resolved-server-side", dataProducts });
}));
app.put("/internal/data-plane/v1/consumer-reader-bindings/by-key/:bindingKey", requireFoundryProvisionerApi, asyncRoute(async (req, res) => {
const bindingKey = requireIdentifier(req.params.bindingKey, "managed_consumer_reader_binding_key_invalid");
const policy = normalizeManagedConsumerReaderBindingRequest(req.body);
await assertRegisteredProductIds(policy.allowedDataProductIds);
const requestHash = consumerReaderBindingRequestHash(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",
source_connection_id as "sourceConnectionId", 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_reader_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_consumer_reader_binding_request_conflict");
if (binding.active !== true || binding.expiresAt !== null) {
throw httpError(409, "managed_consumer_reader_binding_generation_inactive");
}
response = { status: 200, idempotent: true, binding };
} else {
const source = await resolveManagedReaderScope(client, policy.allowedDataProductIds);
const inserted = await client.query(
`insert into external_data_plane_reader_bindings (
id, token_hash, binding_key, request_hash, generation,
tenant_id, connection_id, source_connection_id, provider_id,
allowed_data_product_ids, expires_at
) values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11)
returning id, binding_key as "bindingKey", generation,
tenant_id as "tenantId", connection_id as "connectionId",
source_connection_id as "sourceConnectionId", 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,
source.tenantId,
source.connectionId,
source.connectionId,
source.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_consumer_reader_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,
readerBinding: safeManagedConsumerReaderBinding(response.binding),
});
}));
app.post("/internal/data-plane/v1/consumer-reader-bindings/by-key/:bindingKey/generations/:generation/revoke", requireFoundryProvisionerApi, asyncRoute(async (req, res) => {
const bindingKey = requireIdentifier(req.params.bindingKey, "managed_consumer_reader_binding_key_invalid");
const generation = requirePositiveInteger(req.params.generation, "managed_consumer_reader_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",
source_connection_id as "sourceConnectionId", 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_reader_bindings
where binding_key = $1 and generation = $2
for update`,
[bindingKey, generation],
);
if (!existing.rowCount) throw httpError(404, "managed_consumer_reader_binding_not_found");
if (existing.rows[0].active === true) {
const revoked = await client.query(
`update external_data_plane_reader_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",
source_connection_id as "sourceConnectionId", 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, readerBinding: safeManagedConsumerReaderBinding(binding) });
}));
app.put("/internal/data-plane/v1/reader-bindings/by-key/:bindingKey", requireManagedProvisionerApi, asyncRoute(async (req, res) => {
const bindingKey = requireIdentifier(req.params.bindingKey, "managed_reader_binding_key_invalid");
const policy = normalizeManagedReaderBindingRequest(req.body);
@@ -918,6 +1074,17 @@ function requireManagedProvisionerApi(req, _res, next) {
}
}
function requireFoundryProvisionerApi(req, _res, next) {
if (!config.foundryProvisionerApiEnabled) return next(httpError(503, "foundry_provisioner_api_disabled"));
if (!verifyFoundryProvisionerRequest) return next(httpError(503, "foundry_provisioner_api_not_configured"));
try {
req.managedProvisionerIdentity = verifyFoundryProvisionerRequest(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);