Gertjan Filarski
In het digitaal erfgoed hoor je de term: dataspace. Wat betekent dat eigenlijk? En belangrijker: wat kan het echt? Is het een nieuwe term voor beleidsmakers? Of hangt er concrete technologie onder? Uit nieuwsgierigheid ben ik op onderzoek uitgegaan.
Je vindt de experimenten in de P-322 Github dataspace-experiments repository.
In het eerste blog heb ik uitgelegd wat een dataspace is, welk probleem het probeert op te lossen; en heb ik een overzicht gegeven van de begrippen, termen, en rollen. Daar staat ook een werkende experimentele dataspace-transactie.
Onderweg naar toetsbare toegangsvoorwaarden
Mijn persoonlijke interesse zit in het inrichten van toegangsbeleid in een open data omgeving. Het heeft pas zin om dat te gaan testen als er meerdere afnemers zijn die zich bij een bronhouder melden met interesse in aangeboden data.
Daarom wil ik in dit experiment onderzoeken hoe meerdere afnemers zich tot elkaar verhouden. Daarvoor moet de code uit experiment 1 op de schop. Nu is het immers niet veel meer dan een procedureel script dat voor één provider en één consumer een aantal stappen achter elkaar uitvoert. Dat moet generieker en breder inzetbaar worden zodat ik straks makkelijker dingen kan toevoegen.
Experiment 2
Het doel is om meerdere dataspace-transacties te kunnen uitvoeren waarin meerdere consumers bij een provider data ophalen. We kunnen natuurlijk na de eerste transactie in experiment 1 de stappen voor een tweede consumer opnieuw configureren en uitvoeren, maar dat schaalt niet. Straks wil ik misschien een derde of vierde gebruiker.
Daarom splitsen we de stappen uit experiment 1 op in twee delen. Stappen 1 en 2 gaan over de opbouw van de dataset (de asset) en de aanbieding (het offer). Bij deze stappen is de provider leidend. Stappen 3 tot en met 7 worden daarentegen gestart door de consumer. Die initieert het contractonderhandelingsproces en doet een verzoek om een data-transfer. In een experiment met meerdere consumers moeten stappen 3-7 herhaalbaar zijn.
Provider
We bouwen de code daarom van een procesperspectief om naar een rolperspectief. Met de rol van provider moet je assets kunnen registreren en offers kunnen maken. Als consumer moet je de catalogus kunnen opvragen, een contractonderhandeling kunnen starten, de transfer initiëren, een EDR ophalen, en de data binnenharken.
Daarom maken we twee objecten: Provider en Consumer waarin de verschillende transactiestappen gaan landen. Stappen 1 en 2 horen bij de rol provider. Een asset en een offer bestaan onafhankelijk van wie er langskomt om ze te bekijken. Ze zijn een belofte aan de wereld - of ten minste aan de dataspace. Je maakt ze één keer aan en daarna liggen ze klaar. Dat maakt provider een logisch thuis voor die stappen: ze beschrijven wat de provider is en aanbiedt, niet wat er op een bepaald moment gebeurt.
import { Logger } from "../utils/logger.js";import { EdcManagementClient } from "./edcManagementClient.js"; export type PublicationSpec = { assetId: string; contractDefId: string; policyId: string; sourceUrl: string;}; export class Provider { readonly mgmt: EdcManagementClient; readonly log: Logger; constructor( readonly pid: string, mgmtBaseUrl: string, mgmtApiKey: string, readonly dspAddressDocker: string, readonly publicBaseHost: string, logger: Logger ) { this.mgmt = new EdcManagementClient(mgmtBaseUrl, mgmtApiKey); this.log = logger; } async ensureAsset(pub: PublicationSpec): Promise<void> { this.log.p(`Ensuring asset '${pub.assetId}' exists`); try { await this.mgmt.json( "GET", `/v3/assets/${encodeURIComponent(pub.assetId)}` ); this.log.p(`Asset '${pub.assetId}' already exists`); return; } catch (e: any) { if (!String(e.message).includes("404")) throw e; } await this.mgmt.json("POST", `/v3/assets`, { "@context": { "@vocab": "https://w3id.org/edc/v0.0.1/ns/", }, "@id": pub.assetId, "@type": "Asset", properties: { name: pub.assetId, contenttype: "application/json", }, dataAddress: { "@type": "HttpData", "https://w3id.org/edc/v0.0.1/ns/type": "HttpData", baseUrl: pub.sourceUrl, proxyPath: "true", method: "GET", }, }); this.log.p(`Asset '${pub.assetId}' created`); } async ensureContractDefinition(pub: PublicationSpec): Promise<void> { this.log.p(`Ensuring contract definition '${pub.contractDefId}' exists`); const byId = `/v3/contractdefinitions/${encodeURIComponent( pub.contractDefId )}`; try { await this.mgmt.json("GET", byId); this.log.p(`Contract definition '${pub.contractDefId}' already exists`); return; } catch (e: any) { if (!String(e.message).includes("404")) throw e; } await this.mgmt.json("POST", `/v3/contractdefinitions`, { "@context": { "@vocab": "https://w3id.org/edc/v0.0.1/ns/" }, "@id": pub.contractDefId, "@type": "ContractDefinition", accessPolicyId: pub.policyId, contractPolicyId: pub.policyId, assetsSelector: [ { "@type": "Criterion", operandLeft: "id", operator: "=", operandRight: pub.assetId, }, ], }); this.log.p(`Contract definition '${pub.contractDefId}' created`); }} 
import { Logger } from "../utils/logger.js";import { EdcManagementClient } from "./edcManagementClient.js";import { EdcTransaction } from "./edcTransaction.js";import { Provider } from "./provider.js"; export class Consumer { readonly mgmt: EdcManagementClient; readonly log: Logger; constructor( readonly pid: string, mgmtBaseUrl: string, mgmtApiKey: string, log: Logger ) { this.mgmt = new EdcManagementClient(mgmtBaseUrl, mgmtApiKey); this.log = log; } transaction(provider: Provider) { const l = this.log.scoped(`[${this.pid}]`); return new EdcTransaction(this.pid, this.mgmt, provider, l); }} Transaction
Stappen 3 tot en met 7 zijn van een andere orde. Dat zijn geen eigenschappen, maar gebeurtenissen. Ze vormen samen een tijdelijk proces tussen consumer en provider. Er wordt onderhandeld, besloten, geweigerd of toegestaan, en uiteindelijk data overgedragen.
Dat proces heeft een eigen leven: het heeft een begin, een verloop, en een einde. En is herhaalbaar: misschien wil de consumer daarna reageren op een ander offer van dezelfde provider, of een transactie aangaan met een hele nieuwe partij.
Stop je al die stappen in de consumer, dan moet je daar heel precies de staat gaan bijhouden. Het risico dat je als consumer bij provider A om een offer van provider B gaat vragen ligt dan op de loer. Dat willen we niet.
Daarnaast, kun je zien aankomen dat een consumer meerdere transacties tegelijkertijd wil uitvoeren; en een gefaalde transactie daarna opnieuw moet doen. Ook dat kan niet als we de stappen in de consumer regelen.
We brengen daarom stappen 3 tot en met 7 onder in een object EdcTransaction. Daarmee krijgt de transactie zelf gewicht. Het wordt iets wat je kunt volgen, loggen, herhalen of vergelijken. De consumer initieert transacties, maar die dragen zelf hun eigen verhaal. En juist dat verhaal wil ik begrijpen, testen en bevragen.
import { firstObj } from "../utils/collections.js";import { Logger } from "../utils/logger.js";import { EdcManagementClient } from "./edcManagementClient.js";import { Provider } from "./provider.js"; /* ------------------------------------------------------------ * Types local to the transaction * ------------------------------------------------------------ */ export type Json = | null | boolean | number | string | Json[] | { [k: string]: Json }; export type JsonObject = { [k: string]: Json };export type OfferPolicy = JsonObject; export type CatalogResult = { providerPid: string; assetId: string; offer: OfferPolicy;}; export type NegotiationResult = { negotiationId: string; agreementId: string;}; export type TransferResult = { transferProcessId: string;}; export type EdrResult = { endpointDocker: string; endpointHost: string; token: string;}; /* ------------------------------------------------------------ * EdcTransaction * ------------------------------------------------------------ */ export class EdcTransaction { constructor( readonly consumerPid: string, readonly mgmt: EdcManagementClient, readonly provider: Provider, private readonly log: ReturnType<Logger["scoped"]> ) {} /* ------------------------------------------------------------ * Helper * ------------------------------------------------------------ */ normalizeOfferForContractRequest(input: { offer: OfferPolicy; assetId: string; providerPid: string; consumerPid: string; }): JsonObject { const { offer, assetId, providerPid, consumerPid } = input; return { ...offer, "@type": "odrl:Offer", "odrl:target": { "@id": assetId }, "odrl:assigner": { "@id": providerPid }, "odrl:assignee": { "@id": consumerPid }, }; } /* ------------------------------------------------------------ * Orchestration * ------------------------------------------------------------ */ async run(authHeaderMode: "bearer" | "raw" = "raw") { this.log.s(`Transaction: Fetch catalog + offer`, { nl: true }); const cat = await this.fetchCatalog(); this.log.s(`Transaction: Contract negotiation`, { nl: true }); const neg = await this.negotiateContract(cat); this.log.s(`Transaction: Transfer process`, { nl: true }); const tr = await this.startTransfer(cat, neg); this.log.s(`Transaction: Fetch EDR`, { nl: true }); const edr = await this.fetchEdr(tr); this.log.s(`Transaction: Data access`, { nl: true }); const data = await this.accessData(edr, authHeaderMode); return { cat, neg, tr, edr, data }; } /* ------------------------------------------------------------ * Step 3 — Catalog * ------------------------------------------------------------ */ async fetchCatalog(): Promise<CatalogResult> { const l = this.log; l.c(`Requesting catalog from provider via consumer management API`); l.c(`Provider DSP address: ${this.provider.dspAddressDocker}`); l.c(`Protocol: dataspace-protocol-http`); const catalog = await this.mgmt.json("POST", `/v3/catalog/request`, { "@type": "https://w3id.org/edc/v0.0.1/ns/CatalogRequest", "https://w3id.org/edc/v0.0.1/ns/counterPartyAddress": this.provider.dspAddressDocker, "https://w3id.org/edc/v0.0.1/ns/protocol": "dataspace-protocol-http", "https://w3id.org/edc/v0.0.1/ns/querySpec": { "@type": "https://w3id.org/edc/v0.0.1/ns/QuerySpec", "https://w3id.org/edc/v0.0.1/ns/offset": 0, "https://w3id.org/edc/v0.0.1/ns/limit": 50, }, }); const providerPid = catalog?.["dspace:participantId"] ?? this.provider.pid; const dataset = firstObj<any>(catalog?.["dcat:dataset"]); if (!dataset) { l.c(`No dataset found in catalog`); throw new Error( `No dcat:dataset in catalog:\n${JSON.stringify(catalog, null, 2)}` ); } const assetId = dataset?.["@id"]; const offer = dataset?.["odrl:hasPolicy"]; if (!assetId || !offer) { throw new Error( `Missing @id or odrl:hasPolicy:\n${JSON.stringify(dataset, null, 2)}` ); } l.c(`Normalizing offer for ContractRequest`); const normalizedOffer = this.normalizeOfferForContractRequest({ offer, assetId, providerPid, consumerPid: this.consumerPid, }); return { providerPid, assetId, offer: normalizedOffer }; } /* ------------------------------------------------------------ * Step 4 — Negotiation * ------------------------------------------------------------ */ async negotiateContract(cat: CatalogResult): Promise<NegotiationResult> { const l = this.log; l.c(`Negotiating contract for asset '${cat.assetId}'`); const req = { "@context": { "@vocab": "https://w3id.org/edc/v0.0.1/ns/", edc: "https://w3id.org/edc/v0.0.1/ns/", odrl: "http://www.w3.org/ns/odrl/2/", }, "@type": "edc:ContractRequest", "edc:counterPartyAddress": this.provider.dspAddressDocker, "edc:counterPartyId": cat.providerPid, "edc:protocol": "dataspace-protocol-http", "edc:policy": cat.offer, }; const created = await this.mgmt.raw( "POST", `/v3/contractnegotiations`, JSON.stringify(req) ); const negotiationId = created?.["@id"]; if (!negotiationId) { throw new Error(`No negotiation @id returned`); } const finalized = await this.mgmt.waitForState( `/v3/contractnegotiations/${encodeURIComponent(negotiationId)}`, (b) => b?.state, "FINALIZED", 60 ); const agreementId = finalized?.contractAgreementId; if (!agreementId) { throw new Error(`No contractAgreementId on finalized negotiation`); } l.c(`Contract finalized: agreementId=${agreementId}`); return { negotiationId, agreementId }; } /* ------------------------------------------------------------ * Step 5 — Transfer * ------------------------------------------------------------ */ async startTransfer( cat: CatalogResult, neg: NegotiationResult ): Promise<TransferResult> { const l = this.log; l.c(`Starting transfer for agreement '${neg.agreementId}'`); const created = await this.mgmt.raw( "POST", `/v3/transferprocesses`, JSON.stringify({ "@context": { "@vocab": "https://w3id.org/edc/v0.0.1/ns/" }, "@type": "TransferRequest", contractId: neg.agreementId, protocol: "dataspace-protocol-http", connectorId: cat.providerPid, counterPartyAddress: this.provider.dspAddressDocker, transferType: "HttpData-PULL", }) ); const transferProcessId = created?.["@id"]; if (!transferProcessId) { throw new Error(`No transferProcessId returned`); } await this.mgmt.waitForState( `/v3/transferprocesses/${encodeURIComponent(transferProcessId)}`, (b) => b?.state, "STARTED", 60 ); return { transferProcessId }; } /* ------------------------------------------------------------ * Step 6 — EDR * ------------------------------------------------------------ */ async fetchEdr(tr: TransferResult): Promise<EdrResult> { const l = this.log; const path = `/v3/edrs/${encodeURIComponent( tr.transferProcessId )}/dataaddress`; let last: any = null; for (let i = 0; i < 60; i++) { const res = await fetch(`${this.mgmt.baseUrl}${path}`, { headers: { "X-Api-Key": this.mgmt.apiKey }, }); const text = await res.text(); try { last = JSON.parse(text); } catch { last = text; } const obj = firstObj<any>(last); if (obj?.authorization && obj?.endpoint) { const endpointHost = obj.endpoint.replace( "http://edc-provider:11005/api/public", this.provider.publicBaseHost ); l.token("consumer", "EDR token", obj.authorization); return { endpointDocker: obj.endpoint, endpointHost, token: obj.authorization, }; } await new Promise((r) => setTimeout(r, 1000)); } throw new Error(`Timed out waiting for EDR`); } /* ------------------------------------------------------------ * Step 7 — Data access * ------------------------------------------------------------ */ async accessData( edr: EdrResult, authHeaderMode: "bearer" | "raw" ): Promise<any> { const l = this.log; l.c(`Accessing provider data via EDR public endpoint`); l.c(`Endpoint: ${edr.endpointHost}/`); l.c(`Authorization mode: ${authHeaderMode}`); const auth = authHeaderMode === "bearer" ? `Bearer ${edr.token}` : edr.token; const res = await fetch(`${edr.endpointHost}/`, { headers: { Authorization: auth, Accept: "application/json", }, }); l.c(`Provider responded with HTTP ${res.status}`); const text = await res.text(); let parsed: any = text; let isJson = false; try { parsed = JSON.parse(text); isJson = true; l.c(`Response body is valid JSON`); } catch { l.c(`Response body is not JSON`); } if (!res.ok) { l.err(`Data access failed`); throw new Error( `Data access failed: HTTP ${res.status}\n${ typeof parsed === "string" ? parsed : JSON.stringify(parsed, null, 2) }` ); } l.c(`Transferred data:`); if (isJson) { if (Array.isArray(parsed)) { l.c(`Data shape: array(len=${parsed.length})`); } else if (parsed && typeof parsed === "object") { l.c(`Data shape: keys=[${Object.keys(parsed).join(", ")}]`); } else { l.c(`Data type: ${typeof parsed}`); } l.c(`Payload:\n${JSON.stringify(parsed, null, 2)}`); } else { l.c(`Payload (text):\n${String(parsed)}`); } return parsed; }} Merk op: we sorteren met EdcTransaction wel beetje voor. In experiment 2 heeft een consumer nog maar één transactie en ik had de stappen - net als bij provider - dus ook gewoon nog in de consumer kunnen stoppen.
ManagementClient
In een dataspace is niet alles een gesprek. Sommige dingen zijn gewoon administratie. Assets registreren, offers aanmaken, contracten volgen, transfers monitoren: dat gebeurt via de management-interfaces van de deelnemers. Dat zijn geen publieke gesprekken tussen partijen, maar interne handelingen waarmee een organisatie de eigen dataspace-rol bestuurt. De EDC Management API is precies dat: het bedieningspaneel van een participant.
Zoals we in experiment 1 zagen is er een Consumer Management API en een Provider Management API.
Het EdcManagementClient object is mijn eerste, minimale vertaling van dat bedieningspaneel naar code. Geen slimme logica, geen beleidskeuzes, maar een eerste meer robuuste plek om API-verzoeken te regelen, data te lezen, fouten begrijpelijk te maken en soms even geduldig te wachten tot alles is bijgewerkt.
export class EdcManagementClient { constructor(readonly baseUrl: string, readonly apiKey: string) {} async json(method: string, path: string, body?: any): Promise<any> { const res = await fetch(`${this.baseUrl}${path}`, { method, headers: { "X-Api-Key": this.apiKey, "Content-Type": "application/json", Accept: "application/json", }, body: body ? JSON.stringify(body) : undefined, }); const text = await res.text(); let parsed: any = text; try { parsed = JSON.parse(text); } catch {} if (!res.ok) { throw new Error( `HTTP ${res.status} ${method} ${this.baseUrl}${path}\n${ typeof parsed === "string" ? parsed : JSON.stringify(parsed, null, 2) }` ); } return parsed; } async raw(method: string, path: string, rawBody: string): Promise<any> { const res = await fetch(`${this.baseUrl}${path}`, { method, headers: { "X-Api-Key": this.apiKey, "Content-Type": "application/json", Accept: "application/json", }, body: rawBody, }); const text = await res.text(); let parsed: any = text; try { parsed = JSON.parse(text); } catch {} if (!res.ok) { throw new Error( `HTTP ${res.status} ${method} ${this.baseUrl}${path}\n${ typeof parsed === "string" ? parsed : JSON.stringify(parsed, null, 2) }` ); } return parsed; } async waitForState<T = any>( path: string, getState: (body: any) => string | undefined, desired: string, timeoutSeconds = 60 ): Promise<T> { const deadline = Date.now() + timeoutSeconds * 1000; let last: any = null; while (Date.now() < deadline) { const body = await this.json("GET", path); last = body; if (getState(body) === desired) return body as T; await new Promise((r) => setTimeout(r, 1000)); } throw new Error( `Timeout waiting for state '${desired}' at ${ this.baseUrl }${path}\nLast=${JSON.stringify(last, null, 2)}` ); }} Belangrijk is wat het ding níet doet. De management client onderhandelt niet, kiest geen offers en neemt geen beslissingen. Dat is het werk van de consumers, providers, en transacties. De management client zorgt er alleen voor dat die rollen op een consistente manier tegen hun eigen EDC-infrastructuur kunnen praten. Daarmee is het een soort adapter: klein en saai, maar daarom niet minder essentieel.
Het resultaat
Hoewel we nu alle code opnieuw hebben gestructureerd doet experiment 2 niet heel veel anders dan experiment 1. Het geeft vooral een steviger basis voor de volgende stappen.
import { Consumer } from "./lib/consumer.js";import { Provider, PublicationSpec } from "./lib/provider.js";import { Logger } from "./utils/logger.js"; type RunConfig = { apiKey: string; // Provider (host-facing mgmt) + (docker-facing dsp) providerPid: string; providerMgmtBaseUrl: string; providerDspAddressDocker: string; providerPublicBaseHost: string; // Publications (provider) publications: PublicationSpec[]; // Consumers consumers: Array<{ pid: string; mgmtBaseUrl: string; }>; // Step 7 authHeaderMode?: "bearer" | "raw";}; function loadRunConfig(): RunConfig { const apiKey = process.env.API_KEY ?? "SomeOtherApiKey"; return { apiKey, providerPid: process.env.PROVIDER_PID ?? "provider", providerMgmtBaseUrl: process.env.PROVIDER_MGMT ?? "http://localhost:11012/api/management", providerDspAddressDocker: process.env.PROVIDER_DSP_DOCKER ?? "http://edc-provider:11003/api/dsp", providerPublicBaseHost: process.env.PROVIDER_PUBLIC_HOST ?? "http://localhost:11015/api/public", publications: [ { assetId: process.env.ASSET_ID ?? "asset-hello-1", contractDefId: process.env.CONTRACT_DEF_ID ?? "cd-hello-1", policyId: process.env.POLICY_ID ?? "always-true", // docker-internal URL sourceUrl: process.env.SOURCE_URL ?? "http://api:7070/hello", }, ], consumers: [ { pid: process.env.CONSUMER_1_PID ?? "consumer-1", mgmtBaseUrl: process.env.CONSUMER_1_MGMT ?? "http://localhost:12012/api/management", }, { pid: process.env.CONSUMER_2_PID ?? "consumer-2", mgmtBaseUrl: process.env.CONSUMER_2_MGMT ?? "http://localhost:22012/api/management", }, ], authHeaderMode: (process.env.AUTH_HEADER_MODE as "bearer" | "raw") ?? "raw", };} export async function run() { const cfg = loadRunConfig(); const logger = new Logger(); // --- Provider setup --- const provider = new Provider( cfg.providerPid, cfg.providerMgmtBaseUrl, cfg.apiKey, cfg.providerDspAddressDocker, cfg.providerPublicBaseHost, logger.scoped("setup") ); logger.s("== Provider setup: assets + contract definitions =="); for (const pub of cfg.publications) { logger.s(`== Ensure asset '${pub.assetId}' ==`); await provider.ensureAsset(pub); logger.s(`== Ensure contract definition '${pub.contractDefId}' ==`); await provider.ensureContractDefinition(pub); } // --- Consumer transactions --- let i = 0; for (const c of cfg.consumers) { i++; const consumer = new Consumer( c.pid, c.mgmtBaseUrl, cfg.apiKey, logger.scoped(`[consumer-${i}]`) ); const tx = consumer.transaction(provider); await tx.run("raw"); } return { ok: true as const };} if (import.meta.url === `file://${process.argv[1]}`) { run().catch((err) => { console.error(err); process.exit(1); });} Na de initialisatie waarin alle containers worden gestart, configureren we het experiment: we maken de provider en consumers aan. Daarna start eerst de provider en die maakt een asset en offer aan net als in experiment 1. Vervolgens voeren we een lus uit over alle consumers. Voor iedere consumer maken we een dataspace-transactie aan en voeren die uit.
// --- Consumer transactions --- let i = 0; for (const c of cfg.consumers) { i++; const consumer = new Consumer( c.pid, c.mgmtBaseUrl, cfg.apiKey, logger.scoped(`[consumer-${i}]`) ); const tx = consumer.transaction(provider); await tx.run("raw"); }We krijgen nu tweemaal (voor consumer-1 en consumer-2) de brondata te zien:
{ "message": "Hello, Dataspace", "ts": "2026-01-07T14:47:50.462Z", "dataset": [ { "id": "a1", "title": "Example record", "license": "CC0" } ]}Maar beide transacties hebben een eigen agreementId en sleutel (de EDR-token). Iedere consumer heeft dus een eigen contractonderhandeling met de provider uitgevoerd, die heeft toegang toegestaan, en een unieke sleutel aangemaakt.
Niet spannend, maar wel geslaagd. Hoera!
Experiment 3
Na het herschrijven van code is de leukste vraag die je jezelf kan stellen: hoe maak ik dit weer kapot?
Hoewel ik eigenlijk sta te trappelen om met de toegangsvoorwaarden van de provider aan de slag te gaan - was ik zo nieuwsgierig naar het gedrag van de EDC dat ik een derde experiment heb toegevoegd. Wat nu als een kwaadwillende consumer de agreementId of de EDR-token van een andere consumer jat?
// Implements the three “reuse” attempts://// 1) Guaranteed failure: consumer-3 cannot retrieve consumer-2’s EDR from consumer-3 mgmt API// 2) Likely failure: consumer-3 cannot use consumer-2’s agreementId to start a transfer// 3) Not guaranteed: consumer-3 reusing consumer-2’s EDR token for direct HTTP access import { Consumer } from "./lib/consumer.js";import { Provider, PublicationSpec } from "./lib/provider.js";import { Logger } from "./utils/logger.js"; type RunConfig = { apiKey: string; // Provider (host-facing mgmt) + (docker-facing dsp) providerPid: string; providerMgmtBaseUrl: string; providerDspAddressDocker: string; providerPublicBaseHost: string; // Publications (provider) publications: PublicationSpec[]; // Consumers (1 and 2 run full tx) consumers: Array<{ pid: string; mgmtBaseUrl: string; }>; // Consumer-3 (reuse attempts) consumer3: { pid: string; mgmtBaseUrl: string; }; // Step 7 authHeaderMode?: "bearer" | "raw";}; function loadRunConfig(): RunConfig { const apiKey = process.env.API_KEY ?? "SomeOtherApiKey"; return { apiKey, providerPid: process.env.PROVIDER_PID ?? "provider", providerMgmtBaseUrl: process.env.PROVIDER_MGMT ?? "http://localhost:11012/api/management", providerDspAddressDocker: process.env.PROVIDER_DSP_DOCKER ?? "http://edc-provider:11003/api/dsp", providerPublicBaseHost: process.env.PROVIDER_PUBLIC_HOST ?? "http://localhost:11015/api/public", publications: [ { assetId: process.env.ASSET_ID ?? "asset-hello-1", contractDefId: process.env.CONTRACT_DEF_ID ?? "cd-hello-1", policyId: process.env.POLICY_ID ?? "always-true", // docker-internal URL sourceUrl: process.env.SOURCE_URL ?? "http://api:7070/hello", }, ], consumers: [ { pid: process.env.CONSUMER_1_PID ?? "consumer-1", mgmtBaseUrl: process.env.CONSUMER_1_MGMT ?? "http://localhost:12012/api/management", }, { pid: process.env.CONSUMER_2_PID ?? "consumer-2", mgmtBaseUrl: process.env.CONSUMER_2_MGMT ?? "http://localhost:22012/api/management", }, ], consumer3: { pid: process.env.CONSUMER_3_PID ?? "consumer-3", mgmtBaseUrl: process.env.CONSUMER_3_MGMT ?? "http://localhost:32012/api/management", }, authHeaderMode: (process.env.AUTH_HEADER_MODE as "bearer" | "raw") ?? "raw", };} export async function run() { const cfg = loadRunConfig(); const logger = new Logger(); // --- Provider setup --- const provider = new Provider( cfg.providerPid, cfg.providerMgmtBaseUrl, cfg.apiKey, cfg.providerDspAddressDocker, cfg.providerPublicBaseHost, logger.scoped("setup") ); logger.s("== Provider setup: assets + contract definitions =="); for (const pub of cfg.publications) { logger.s(`== Ensure asset '${pub.assetId}' ==`); await provider.ensureAsset(pub); logger.s(`== Ensure contract definition '${pub.contractDefId}' ==`); await provider.ensureContractDefinition(pub); } // --- Consumer transactions (1 and 2) --- const results: Array<{ consumerPid: string; r: any; // keep flexible; we mainly need r.cat/r.neg/r.tr/r.edr }> = []; for (const c of cfg.consumers) { const consumer = new Consumer( c.pid, c.mgmtBaseUrl, cfg.apiKey, logger.scoped(`[${c.pid}]`) ); const tx = consumer.transaction(provider); const r = await tx.run(cfg.authHeaderMode ?? "raw"); results.push({ consumerPid: c.pid, r }); } const r2 = results.at(1)?.r; if (!r2?.tr?.transferProcessId || !r2?.neg?.agreementId || !r2?.edr?.token) { throw new Error( `Missing consumer-2 results. Need r2.tr.transferProcessId, r2.neg.agreementId, r2.edr.token. Got:\n${JSON.stringify( r2, null, 2 )}` ); } // --- Consumer-3 reuse attempts --- const consumer3 = new Consumer( cfg.consumer3.pid, cfg.consumer3.mgmtBaseUrl, cfg.apiKey, logger.scoped(`[${cfg.consumer3.pid}]`) ); logger.s("== Reuse attempts (consumer-3) =="); // 1) Guaranteed failure: consumer-3 cannot retrieve consumer-2’s EDR from consumer-3 mgmt API logger.s( "== 1) consumer-3 fetches consumer-2 EDR via management API (should fail) ==" ); try { const tp2 = r2.tr.transferProcessId as string; await consumer3.mgmt.json( "GET", `/v3/edrs/${encodeURIComponent(tp2)}/dataaddress` ); logger.err( `Unexpected success: consumer-3 retrieved an EDR for consumer-2 transferProcessId=${tp2}` ); } catch (e: any) { logger.w(`Expected failure: ${String(e?.message ?? e)}`); } // 2) Likely failure: consumer-3 cannot use consumer-2’s agreementId to start a transfer logger.s( "== 2) consumer-3 starts transfer using consumer-2 agreementId (should fail) ==" ); try { const created = await consumer3.mgmt.raw( "POST", `/v3/transferprocesses`, JSON.stringify({ "@context": { "@vocab": "https://w3id.org/edc/v0.0.1/ns/" }, "@type": "TransferRequest", contractId: r2.neg.agreementId, protocol: "dataspace-protocol-http", connectorId: r2.cat.providerPid, counterPartyAddress: provider.dspAddressDocker, transferType: "HttpData-PULL", }) ); logger.err( `Unexpected success: consumer-3 started transfer with consumer-2 agreementId. Response:\n${JSON.stringify( created, null, 2 )}` ); } catch (e: any) { logger.w(`Expected failure: ${String(e?.message ?? e)}`); } // 3) Not guaranteed: consumer-3 reusing consumer-2’s EDR token for direct HTTP access logger.s( "== 3) consumer-3 uses consumer-2 EDR token directly (may succeed or fail) ==" ); try { const authMode = cfg.authHeaderMode ?? "raw"; const auth = authMode === "bearer" ? `Bearer ${r2.edr.token}` : r2.edr.token; const res = await fetch(`${r2.edr.endpointHost}/`, { headers: { Authorization: auth, Accept: "application/json", }, }); const text = await res.text(); let parsed: any = text; try { parsed = JSON.parse(text); } catch {} if (!res.ok) { logger.w( `Token reuse blocked (HTTP ${res.status}). Body:\n${ typeof parsed === "string" ? parsed : JSON.stringify(parsed, null, 2) }` ); } else { logger.w( `Token reuse succeeded (HTTP ${ res.status }). This means “whoever has the token can use it” in this setup.\nBody:\n${ typeof parsed === "string" ? parsed : JSON.stringify(parsed, null, 2) }` ); } } catch (e: any) { logger.w(`Token reuse attempt errored: ${String(e?.message ?? e)}`); } return { ok: true as const };} if (import.meta.url === `file://${process.argv[1]}`) { run().catch((err) => { console.error(err); process.exit(1); });} Daarvoor introduceren we in dit experiment consumer-3 die de rol van boef speelt. Dankzij de herinrichting van de code is dat nu lekker makkelijk.
Kan een boef de sleutel van een andere consumer bij de provider ophalen?
De eerste vraag is: wat gebeurd er wanneer de boef/consumer-3 achter de agreementId van consumer-2 komt en daarmee zijn EDR-token opvraagt? Dat is stap 6 in deze transactieketen. Als het goed is, geeft de provider geen antwoord want de afspraak en sleutel hangen aan een andere consumer-id.
// 1) Guaranteed failure: consumer-3 cannot retrieve consumer-2’s EDR from consumer-3 mgmt API logger.s( "== 1) consumer-3 fetches consumer-2 EDR via management API (should fail) ==" ); try { const tp2 = r2.tr.transferProcessId as string; await consumer3.mgmt.json( "GET", `/v3/edrs/${encodeURIComponent(tp2)}/dataaddress` ); logger.err( `Unexpected success: consumer-3 retrieved an EDR for consumer-2 transferProcessId=${tp2}` ); } catch (e: any) { logger.w(`Expected failure: ${String(e?.message ?? e)}`); } Het antwoord is:
[ { "message": "Object of type EndpointDataReferenceEntry with ID=019bb7a9-c6cf-7806-ae4b-4bcfbdb3e94c was not found", "type": "ObjectNotFound", "path": null, "invalidValue": null }]Mooi zo, dat lukt niet. De provider geeft simpelweg aan dat er geen sleutels zijn gevonden.
Kan een boef met de overeenkomst van een andere consumer de provider overtuigen om een sleutel aan te maken?
Vervolgens wil ik weten of onze boef eerder kan instappen en in staat is om de provider te overtuigen om met een agreementId van consumer-2 een transactie te starten waarin een EDR beschikbaar gesteld gaat worden. Dat is stap 5 in deze experimenten.
// 2) Likely failure: consumer-3 cannot use consumer-2’s agreementId to start a transfer logger.s( "== 2) consumer-3 starts transfer using consumer-2 agreementId (should fail) ==" ); try { const created = await consumer3.mgmt.raw( "POST", `/v3/transferprocesses`, JSON.stringify({ "@context": { "@vocab": "https://w3id.org/edc/v0.0.1/ns/" }, "@type": "TransferRequest", contractId: r2.neg.agreementId, protocol: "dataspace-protocol-http", connectorId: r2.cat.providerPid, counterPartyAddress: provider.dspAddressDocker, transferType: "HttpData-PULL", }) ); logger.err( `Unexpected success: consumer-3 started transfer with consumer-2 agreementId. Response:\n${JSON.stringify( created, null, 2 )}` ); } catch (e: any) { logger.w(`Expected failure: ${String(e?.message ?? e)}`); } Daar reageert de provider als volgt op:
[ { "message": "Contract agreement with id 019bb7a9-c1f6-7faf-951e-b270df39ec10 not found", "type": "InvalidRequest", "path": null, "invalidValue": null }]Zulke verzoeken worden tijdens het onderhandelingsproces door de provider geblokkeerd. Voor consumer-3 bestaat het agreementId niet en dat is precies wat we willen.
Kan een boef met de sleutel van een andere consumer data bij de provider ophalen?
De laatste vraag is misschien wel de interessantste: stel nou dat onze boef het internetverkeer tussen de provider en consumer-2 heeft afgeluisterd en de EDR-token heeft bemachtigt. Wat gebeurt er dan in stap 7 van deze experimenten? Als we de data daadwerkelijk gaan ophalen?
// 3) Not guaranteed: consumer-3 reusing consumer-2’s EDR token for direct HTTP access logger.s( "== 3) consumer-3 uses consumer-2 EDR token directly (may succeed or fail) ==" ); try { const authMode = cfg.authHeaderMode ?? "raw"; const auth = authMode === "bearer" ? `Bearer ${r2.edr.token}` : r2.edr.token; const res = await fetch(`${r2.edr.endpointHost}/`, { headers: { Authorization: auth, Accept: "application/json", }, }); const text = await res.text(); let parsed: any = text; try { parsed = JSON.parse(text); } catch {} if (!res.ok) { logger.w( `Token reuse blocked (HTTP ${res.status}). Body:\n${ typeof parsed === "string" ? parsed : JSON.stringify(parsed, null, 2) }` ); } else { logger.w( `Token reuse succeeded (HTTP ${ res.status }). This means “whoever has the token can use it” in this setup.\nBody:\n${ typeof parsed === "string" ? parsed : JSON.stringify(parsed, null, 2) }` ); } } catch (e: any) { logger.w(`Token reuse attempt errored: ${String(e?.message ?? e)}`); } In experiment 1 heb ik uitgelegd dat de transactie tussen consumer en provider over twee lagen loopt. De besturingslaag (Control Plane) doet de contractonderhandeling en verstrekt een sleutel aan de consumer. De datalaag (Data Plane) houdt zich daarna alleen met het transport van de informatie bezig: als er een geldige sleutel is, dan gaat de Data Plane aan de slag.
De kans is daarom best groot is dat het transport slaagt, en de boef toegang krijgt tot de data.
{ "message": "Hello, Dataspace", "ts": "2026-01-07T14:47:50.462Z", "dataset": [ { "id": "a1", "title": "Example record", "license": "CC0" } ]}En inderdaad, dat kan. De sleutel is geldig, de Data Plane toetst niet meer - als consumer-3 over de EDR-token beschikt dan wordt toegang verschaft.
Is dat raar? Nou, euhm... nee, eigenlijk niet. Het is immers de verantwoordelijkheid van consumer-2 om de sleutel niet te laten slingeren. Niet die van de provider. Dit is een belangrijke reden waarom de EDR-sleutels normaal gesproken een beperkte geldigheidsduur hebben.
Waar staan we nu?
In experiment 1 heb ik het proces van een dataspace-transactie getest. Die stappen zijn nu geland in een meer robuuste omgeving en klaar om verder mee te gaan experimenteren.
In de volgende experimenten ga ik kijken naar de toegangsvoorwaarden die je als provider aan een offer kunt hangen en hoe je die (eventueel automatisch) kan toetsen. Daarna heb ik nog andere experimenten gepland, maar stuur me vooral een bericht als je ideeën hebt!
De experimenten zullen niet allemaal aansluitend verschijnen. Dus blijf deze blog volgen voor updates.