diff --git a/CLAUDE.md b/CLAUDE.md index d4219a5..92b3eb2 100755 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -168,7 +168,7 @@ uses directly. `destroyProviders()` tears providers down on exit (next to `stopP - **Publish** (`publish`, `publishAlgo`): read a JSON DDO file, then `createAssetUtil` with `asset.indexedMetadata.nft.name/symbol` and `asset.services[0].files.files`. `--encrypt` (default `true`) controls DDO encryption. See `metadata/*.json` for the expected DDO shape. - **Edit** (`editAsset`): resolve the DDO via `waitForIndexer`, shallow-merge the top-level keys from the update JSON into the asset, then `updateAssetMetadata`. - **allowAlgo / disallowAlgo**: mutate `services[0].compute.publisherTrustedAlgorithms` (checks signer is the NFT owner and the service is a `compute` service; computes container + files checksums via `ProviderInstance.checkDidFiles` / `getHash`) and re-publish metadata. (`disallowAlgo` exists on `Commands` but is not registered as a CLI command.) -- **Download/consume** (`download`): resolve DDO → `orderAsset` (buys a datatoken) → `tx.wait()` → `ProviderInstance.getDownloadUrl` → `downloadFile` (streams to disk, filename from `content-disposition` when present). +- **Download/consume** (`download`): resolve DDO → look up the target service by id (errors if the `serviceId` is not in the DDO, instead of silently falling back to `services[0]`) → for **DDO version ≥ 5.0.0** run a provider-initialize step (`Commands.initializeProvider` → `ProviderInstance.initialize`, plus an SSI/policy-server verification via `ProviderInstance.initializePSVerification` when `SSI_WALLET_API` is set) then fetch the policy-server object (`getPolicyServerOBJ`, which now returns `null` when the node reports the policy server is not configured) → `orderAsset` (buys a datatoken) → `tx.wait()` → `ProviderInstance.getDownloadUrl` → `downloadFile` (streams to disk, filename from `content-disposition` when present). Each step catches its own error, prints an actionable message, and returns rather than throwing. ### Compute flow diff --git a/README.md b/README.md index 343fff6..1b6cb1d 100644 --- a/README.md +++ b/README.md @@ -335,7 +335,9 @@ Notes when switching nodes: (Order of `--did` and `--folder` does not matter.) - **Rules:** - serviceId is optional. If omitted, the CLI defaults to the first available download service. + serviceId is optional. If omitted, the CLI defaults to the first service listed in the DDO (`services[0]`). If you pass a `serviceId` that does not exist in the DDO, the command now fails fast with a clear error instead of silently ordering the first service. + + For **v5 DDOs** (version ≥ 5.0.0) the download first runs a provider-initialization step against the asset's service endpoint. When `SSI_WALLET_API` is set (see the env vars above) this also performs the SSI / policy-server verification flow; when the target node reports it has no policy server configured, that step is skipped automatically. --- diff --git a/src/commands.ts b/src/commands.ts index cc088b5..83bee58 100644 --- a/src/commands.ts +++ b/src/commands.ts @@ -30,6 +30,7 @@ import { ConfigHelper, Datatoken, ProviderInstance, + ProviderInitialize, amountToUnits, getHash, orderAsset, @@ -58,6 +59,7 @@ import chalk from "chalk"; import { getPolicyServerOBJ, getPolicyServerOBJs, + isPolicyServerConfigured, isVersionGte, } from "./policyServerHelper.js"; import { @@ -440,6 +442,57 @@ export class Commands { } else console.log(util.inspect(resolvedDDO, false, null, true)); } + private async initializeProvider( + asset: Asset, + serviceId: string, + accountId: string, + providerUrl: string, + ): Promise { + // Only run SSI/policy-server verification when a wallet is configured AND + // the node confirms it has a policy server. This mirrors getPolicyServerOBJ's + // skip behavior, so a download against a node without a policy server + // proceeds instead of failing in initializePSVerification. + if ( + process.env.SSI_WALLET_API?.trim() && + (await isPolicyServerConfigured(providerUrl)) + ) { + const command = { + documentId: asset.id, + serviceId, + consumerAddress: accountId, + policyServer: { + sessionId: "", + successRedirectUri: "", + errorRedirectUri: "", + responseRedirectUri: "", + presentationDefinitionUri: "", + }, + }; + const initializePs = await ProviderInstance.initializePSVerification( + providerUrl, + this.signer, + command, + ); + if (!initializePs?.success) { + throw new Error( + `Provider initialization failed: ${initializePs?.error || "Policy Server verification failed"}`, + ); + } + } + try { + return await ProviderInstance.initialize( + asset.id, + serviceId, + 0, + accountId, + providerUrl, + ); + } catch (error: unknown) { + const message = error instanceof Error ? error.message : String(error); + throw new Error(message.replace(/^Error:\s*/i, ""), { cause: error }); + } + } + public async download(args: string[]) { const did = args[1]; const dataDdo = await this.aquarius.waitForIndexer( @@ -460,20 +513,40 @@ export class Commands { const ddoInstance = DDOManager.getDDOClass(dataDdo); const { services, version } = ddoInstance.getDDOFields(); const serviceId = args[3] ? args[3] : services[0].id; + const service = services.find((s) => s.id === serviceId); + if (!service) { + console.error( + chalk.red(`Service ID "${serviceId}" not found in DDO ${did}.`), + ); + return; + } + let policyServer = null; - try { - if (isVersionGte(version, "5.0.0")) { + if (isVersionGte(version, "5.0.0")) { + try { + await this.initializeProvider( + dataDdo, + serviceId, + await this.signer.getAddress(), + service.serviceEndpoint || this.oceanNodeUrl, + ); + } catch (error: unknown) { + const message = error instanceof Error ? error.message : String(error); + console.error(chalk.red("Error initializing Provider:"), message); + return; + } + try { policyServer = await getPolicyServerOBJ( dataDdo, serviceId, this.signer, this.oceanNodeUrl, ); + } catch (error: unknown) { + const message = error instanceof Error ? error.message : String(error); + console.error(chalk.red("Error getting Policy Server Object:"), message); + return; } - } catch (error) { - throw new Error("Error getting Policy Server Object: " + error.message, { - cause: error, - }); } const datatoken = new Datatoken( this.signer, @@ -482,21 +555,28 @@ export class Commands { ); // Order the same service that policy retrieval and getDownloadUrl target. const serviceIndex = services.findIndex((s) => s.id === serviceId); - const tx = await this.orderWithRetry(() => - orderAsset( - dataDdo, - this.signer, - this.config, - datatoken, - this.oceanNodeUrl, - undefined, // consumerAddress - undefined, // consumeMarketOrderFee - undefined, // providerFees - undefined, // consumeMarketFixedSwapFee - undefined, // datatokenIndex - serviceIndex < 0 ? 0 : serviceIndex, - ), - ); + let tx; + try { + tx = await this.orderWithRetry(() => + orderAsset( + dataDdo, + this.signer, + this.config, + datatoken, + this.oceanNodeUrl, + undefined, // consumerAddress + undefined, // consumeMarketOrderFee + undefined, // providerFees + undefined, // consumeMarketFixedSwapFee + undefined, // datatokenIndex + serviceIndex < 0 ? 0 : serviceIndex, + ), + ); + } catch (error: unknown) { + const message = error instanceof Error ? error.message : String(error); + console.error(chalk.red("Error ordering asset:"), message); + return; + } if (!tx) { console.error( diff --git a/src/policyServerHelper.ts b/src/policyServerHelper.ts index a674b38..98247c2 100644 --- a/src/policyServerHelper.ts +++ b/src/policyServerHelper.ts @@ -13,6 +13,32 @@ import { import axios from "axios"; import { Signer } from "ethers"; +// Bounded timeout for the node `status` probe. Without it an unresponsive node +// would hang the probe (and any download/compute waiting on it) indefinitely. +const PS_STATUS_PROBE_TIMEOUT_MS = 10_000; + +/** + * Probe whether the target node has a policy server configured, via the + * `status` directCommand. Returns `true` only when the node explicitly reports + * `isPSConfigured === true`. On a `false` report, a probe error, or a timeout it + * returns `false`, so callers can skip policy-server verification and proceed + * rather than hanging or hard-failing on an unresponsive node. + */ +export async function isPolicyServerConfigured( + providerUrl: string, +): Promise { + try { + const statusResponse = await axios.post( + `${providerUrl}/directCommand`, + { command: "status" }, + { timeout: PS_STATUS_PROBE_TIMEOUT_MS }, + ); + return statusResponse.data?.isPSConfigured === true; + } catch { + return false; + } +} + // Semver-aware "version >= minimum" comparison (numeric, dot-separated). Avoids // the lexicographic pitfalls of comparing version strings directly (e.g. // '5.10.0' < '5.9.0' as strings). A missing/empty version is treated as below @@ -274,13 +300,35 @@ export function extractURLSearchParams( return params; } +/** + * Resolve the policy-server object for a single asset/service. + * + * Returns `null` when policy-server support is unavailable — i.e. the node + * reports it has no policy server configured (`isPSConfigured !== true`) — so + * callers must treat `null` as "no policy server" and proceed without one. A + * probe that fails or times out falls through to the normal flow instead of + * masking a real error with `null`. + */ export async function getPolicyServerOBJ( ddo: Asset, serviceId: string, signer: Signer, providerUrl: string, -): Promise { +): Promise { try { + try { + const statusResponse = await axios.post( + `${providerUrl}/directCommand`, + { command: "status" }, + { timeout: PS_STATUS_PROBE_TIMEOUT_MS }, + ); + if (statusResponse.data?.isPSConfigured !== true) { + return null; + } + } catch { + // Node did not answer the status probe; fall through and attempt the + // normal flow rather than masking a real error with a null. + } const accountId = await signer.getAddress(); const presentationResult = await requestCredentialPresentation( ddo, @@ -380,6 +428,16 @@ export async function getPolicyServerOBJ( } } +/** + * Resolve policy-server objects for a set of datasets plus an optional + * algorithm (compute flows). + * + * Returns `null` when policy-server support is unavailable for the job — any + * entry below DDO v5, or any entry whose per-asset lookup yields `null` (node + * has no policy server configured). Callers must treat `null` as "no policy + * server" and pass it straight through to the provider (which accepts a + * nullable `policyServer`). + */ export async function getPolicyServerOBJs( ddos: { documentId: string; @@ -410,6 +468,9 @@ export async function getPolicyServerOBJs( signer, providerUrl, ); + if (!result) { + return null; + } results.push({ ...result, documentId: ddo.documentId, @@ -430,6 +491,9 @@ export async function getPolicyServerOBJs( signer, providerUrl, ); + if (!algoResult) { + return null; + } results.push({ ...algoResult, documentId: algo.documentId,