diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 794b14dcc..78f0d7a02 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -163,7 +163,7 @@ jobs: ASSET_PURGATORY_URL: 'https://raw.githubusercontent.com/oceanprotocol/list-purgatory/main/list-assets.json' ACCOUNT_PURGATORY_URL: 'https://raw.githubusercontent.com/oceanprotocol/list-purgatory/main/list-accounts.json' - name: docker logs - run: docker logs ocean-ocean-contracts-1 && docker logs ocean-kindcluster-1 && docker logs ocean-computetodata-1 && docker logs ocean-typesense-1 + run: docker logs ocean-ocean-contracts-1 && docker logs ocean-typesense-1 if: ${{ failure() }} - uses: actions/upload-artifact@v4 with: @@ -231,7 +231,7 @@ jobs: done - name: docker logs - run: docker logs ocean-ocean-contracts-1 && docker logs ocean-kindcluster-1 && docker logs ocean-computetodata-1 && docker logs ocean-typesense-1 + run: docker logs ocean-contracts-1 && docker logs ocean-typesense-1 if: ${{ failure() }} - name: Checkout Ocean Node diff --git a/package-lock.json b/package-lock.json index 7ef338266..ce3f44a0a 100644 --- a/package-lock.json +++ b/package-lock.json @@ -34,7 +34,7 @@ "@libp2p/websockets": "^8.1.1", "@multiformats/multiaddr": "^10.2.0", "@oceanprotocol/contracts": "^2.3.0", - "@oceanprotocol/ddo-js": "^0.1.1", + "@oceanprotocol/ddo-js": "^0.1.2", "@types/lodash.clonedeep": "^4.5.7", "axios": "^1.8.4", "base58-js": "^2.0.0", @@ -3564,9 +3564,9 @@ "license": "Apache-2.0" }, "node_modules/@oceanprotocol/ddo-js": { - "version": "0.1.1", - "resolved": "https://registry.npmjs.org/@oceanprotocol/ddo-js/-/ddo-js-0.1.1.tgz", - "integrity": "sha512-RsDUiWfPjylj/xqk4HtUr3qzYzurlBLd2/YW/oP9vC7Gh0uGgJsCShu+ao7hC0Xid4y0iORhkSgiaCtshVGchQ==", + "version": "0.1.2", + "resolved": "https://registry.npmjs.org/@oceanprotocol/ddo-js/-/ddo-js-0.1.2.tgz", + "integrity": "sha512-cDhZ9yk0rpMDqnhkHL3V8eCp2DHQpSqAijlBKBORI6p1CDyb9Q0nDR1kgJObQzEsP1vCOleFkL5cN5vTYBZI3w==", "license": "Apache-2.0", "dependencies": { "@rdfjs/formats-common": "^3.1.0", diff --git a/package.json b/package.json index e232fbbd7..6ee5bc494 100644 --- a/package.json +++ b/package.json @@ -73,7 +73,7 @@ "@libp2p/websockets": "^8.1.1", "@multiformats/multiaddr": "^10.2.0", "@oceanprotocol/contracts": "^2.3.0", - "@oceanprotocol/ddo-js": "^0.1.1", + "@oceanprotocol/ddo-js": "^0.1.2", "@types/lodash.clonedeep": "^4.5.7", "axios": "^1.8.4", "base58-js": "^2.0.0", diff --git a/src/@types/commands.ts b/src/@types/commands.ts index 1649ee228..f9a963d32 100644 --- a/src/@types/commands.ts +++ b/src/@types/commands.ts @@ -16,6 +16,7 @@ import { UrlFileObject, BaseFileObject } from './fileObject' +import { PolicyServerTask } from './policyServer.js' export interface Command { command: string // command name @@ -58,7 +59,7 @@ export interface DownloadCommand extends Command { consumerAddress: string signature: string aes_encrypted_key?: string // if not present it means download without encryption - policyServer?: any // object to pass to policy server + policyServer?: PolicyServerTask // object to pass to policy server } export interface FileInfoCommand extends Command { @@ -138,7 +139,7 @@ export interface GetFeesCommand extends Command { serviceId: string consumerAddress?: string validUntil?: number // this allows a user to request a fee that is valid only for a limited period of time, less than service.timeout - policyServer?: any // object to pass to policyServer + policyServer?: PolicyServerTask // object to pass to policyServer } // admin commands export interface AdminStopNodeCommand extends AdminCommand {} @@ -189,6 +190,7 @@ export interface ComputeInitializeCommand extends Command { consumerAddress: string signature?: string maxJobDuration: number + policyServer?: PolicyServerTask // object to pass to policy server } export interface FreeComputeStartCommand extends Command { @@ -201,6 +203,7 @@ export interface FreeComputeStartCommand extends Command { output?: ComputeOutput resources?: ComputeResourceRequest[] maxJobDuration?: number + policyServer?: PolicyServerTask // object to pass to policy server metadata?: DBComputeJobMetadata } export interface PaidComputeStartCommand extends FreeComputeStartCommand { diff --git a/src/@types/policyServer.ts b/src/@types/policyServer.ts index b0b42537c..51704e41e 100644 --- a/src/@types/policyServer.ts +++ b/src/@types/policyServer.ts @@ -3,3 +3,11 @@ export interface PolicyServerResult { message?: string // error message, if any httpStatus?: number // status returned by server } + +export interface PolicyServerTask { + sessionId?: string + successRedirectUri?: string + errorRedirectUri?: string + responseRedirectUri?: string + presentationDefinitionUri?: string +} diff --git a/src/components/core/compute/initialize.ts b/src/components/core/compute/initialize.ts index 7f5b0b529..ca468fe20 100644 --- a/src/components/core/compute/initialize.ts +++ b/src/components/core/compute/initialize.ts @@ -24,13 +24,15 @@ import { validateCommandParameters } from '../../httpRoutes/validateCommands.js' import { isAddress } from 'ethers' -import { getConfiguration } from '../../../utils/index.js' +import { getConfiguration, isPolicyServerConfigured } from '../../../utils/index.js' import { sanitizeServiceFiles } from '../../../utils/util.js' import { FindDdoHandler } from '../handler/ddoHandler.js' import { isOrderingAllowedForAsset } from '../handler/downloadHandler.js' import { getNonceAsNumber } from '../utils/nonceHandler.js' import { C2DEngineDocker, getAlgorithmImage } from '../../c2d/compute_engine_docker.js' -import { DDOManager } from '@oceanprotocol/ddo-js' +import { Credentials, DDOManager } from '@oceanprotocol/ddo-js' +import { areKnownCredentialTypes, checkCredentials } from '../../../utils/credentials.js' +import { PolicyServer } from '../../policyServer/index.js' export class ComputeInitializeHandler extends CommandHandler { validate(command: ComputeInitializeCommand): ValidateParams { @@ -178,11 +180,12 @@ export class ComputeInitializeHandler extends CommandHandler { // check algo let index = 0 + const policyServer = new PolicyServer() for (const elem of [...[task.algorithm], ...task.datasets]) { const result: any = { validOrder: false } if ('documentId' in elem && elem.documentId) { result.did = elem.documentId - result.serviceId = elem.documentId + result.serviceId = elem.serviceId const ddo = await new FindDdoHandler(node).findAndFormatDdo(elem.documentId) if (!ddo) { const error = `DDO ${elem.documentId} not found` @@ -194,6 +197,12 @@ export class ComputeInitializeHandler extends CommandHandler { } } } + const ddoInstance = DDOManager.getDDOClass(ddo) + const { + chainId: ddoChainId, + nftAddress, + credentials + } = ddoInstance.getDDOFields() const isOrdable = isOrderingAllowedForAsset(ddo) if (!isOrdable.isOrdable) { CORE_LOGGER.error(isOrdable.reason) @@ -205,6 +214,39 @@ export class ComputeInitializeHandler extends CommandHandler { } } } + // check credentials (DDO level) + let accessGrantedDDOLevel: boolean + if (credentials) { + // if POLICY_SERVER_URL exists, then ocean-node will NOT perform any checks. + // It will just use the existing code and let PolicyServer decide. + if (isPolicyServerConfigured()) { + const response = await policyServer.checkStartCompute( + ddoInstance.getDid(), + ddo, + elem.serviceId, + task.consumerAddress, + task.policyServer + ) + accessGrantedDDOLevel = response.success + } else { + accessGrantedDDOLevel = areKnownCredentialTypes(credentials as Credentials) + ? checkCredentials(credentials as Credentials, task.consumerAddress) + : true + } + if (!accessGrantedDDOLevel) { + CORE_LOGGER.logMessage( + `Error: Access to asset ${ddoInstance.getDid()} was denied`, + true + ) + return { + stream: null, + status: { + httpStatus: 403, + error: `Error: Access to asset ${ddoInstance.getDid()} was denied` + } + } + } + } const service = AssetUtils.getServiceById(ddo, elem.serviceId) if (!service) { const error = `Cannot find service ${elem.serviceId} in DDO ${elem.documentId}` @@ -216,9 +258,41 @@ export class ComputeInitializeHandler extends CommandHandler { } } } + // check credentials on service level + // if using a policy server and we are here it means that access was granted (they are merged/assessed together) + if (service.credentials) { + let accessGrantedServiceLevel: boolean + if (isPolicyServerConfigured()) { + // we use the previous check or we do it again + // (in case there is no DDO level credentials and we only have Service level ones) + const response = await policyServer.checkStartCompute( + ddo.id, + ddo, + elem.serviceId, + task.consumerAddress, + task.policyServer + ) + accessGrantedServiceLevel = accessGrantedDDOLevel || response.success + } else { + accessGrantedServiceLevel = areKnownCredentialTypes(service.credentials) + ? checkCredentials(service.credentials, task.consumerAddress) + : true + } - const ddoInstance = DDOManager.getDDOClass(ddo) - const { chainId: ddoChainId, nftAddress } = ddoInstance.getDDOFields() + if (!accessGrantedServiceLevel) { + CORE_LOGGER.logMessage( + `Error: Access to service with id ${service.id} was denied`, + true + ) + return { + stream: null, + status: { + httpStatus: 403, + error: `Error: Access to service with id ${service.id} was denied` + } + } + } + } const config = await getConfiguration() const { rpc, network, chainId, fallbackRPCs } = config.supportedNetworks[ddoChainId] diff --git a/src/components/core/compute/startCompute.ts b/src/components/core/compute/startCompute.ts index 3bd77dbf6..629d7880f 100644 --- a/src/components/core/compute/startCompute.ts +++ b/src/components/core/compute/startCompute.ts @@ -26,13 +26,15 @@ import { decrypt } from '../../../utils/crypt.js' // import { verifyProviderFees } from '../utils/feesHandler.js' import { Blockchain } from '../../../utils/blockchain.js' import { validateOrderTransaction } from '../utils/validateOrders.js' -import { getConfiguration } from '../../../utils/index.js' +import { getConfiguration, isPolicyServerConfigured } from '../../../utils/index.js' import { sanitizeServiceFiles } from '../../../utils/util.js' import { FindDdoHandler } from '../handler/ddoHandler.js' // import { ProviderFeeValidation } from '../../../@types/Fees.js' import { isOrderingAllowedForAsset } from '../handler/downloadHandler.js' -import { DDOManager } from '@oceanprotocol/ddo-js' +import { Credentials, DDOManager } from '@oceanprotocol/ddo-js' import { getNonceAsNumber } from '../utils/nonceHandler.js' +import { PolicyServer } from '../../policyServer/index.js' +import { areKnownCredentialTypes, checkCredentials } from '../../../utils/credentials.js' import { generateUniqueID } from '../../database/sqliteCompute.js' export class PaidComputeStartHandler extends CommandHandler { @@ -146,13 +148,14 @@ export class PaidComputeStartHandler extends CommandHandler { } } } + const policyServer = new PolicyServer() // check algo for (const elem of [...[task.algorithm], ...task.datasets]) { console.log(elem) const result: any = { validOrder: false } if ('documentId' in elem && elem.documentId) { result.did = elem.documentId - result.serviceId = elem.documentId + result.serviceId = elem.serviceId const ddo = await new FindDdoHandler(node).findAndFormatDdo(elem.documentId) if (!ddo) { const error = `DDO ${elem.documentId} not found` @@ -164,6 +167,14 @@ export class PaidComputeStartHandler extends CommandHandler { } } } + const ddoInstance = DDOManager.getDDOClass(ddo) + const { + chainId: ddoChainId, + services, + metadata, + nftAddress, + credentials + } = ddoInstance.getDDOFields() const isOrdable = isOrderingAllowedForAsset(ddo) if (!isOrdable.isOrdable) { CORE_LOGGER.error(isOrdable.reason) @@ -175,6 +186,39 @@ export class PaidComputeStartHandler extends CommandHandler { } } } + // check credentials (DDO level) + let accessGrantedDDOLevel: boolean + if (credentials) { + // if POLICY_SERVER_URL exists, then ocean-node will NOT perform any checks. + // It will just use the existing code and let PolicyServer decide. + if (isPolicyServerConfigured()) { + const response = await policyServer.checkStartCompute( + ddo.id, + ddo, + elem.serviceId, + task.consumerAddress, + task.policyServer + ) + accessGrantedDDOLevel = response.success + } else { + accessGrantedDDOLevel = areKnownCredentialTypes(credentials as Credentials) + ? checkCredentials(credentials as Credentials, task.consumerAddress) + : true + } + if (!accessGrantedDDOLevel) { + CORE_LOGGER.logMessage( + `Error: Access to asset ${ddoInstance.getDid()} was denied`, + true + ) + return { + stream: null, + status: { + httpStatus: 403, + error: `Error: Access to asset ${ddoInstance.getDid()} was denied` + } + } + } + } const service = AssetUtils.getServiceById(ddo, elem.serviceId) if (!service) { const error = `Cannot find service ${elem.serviceId} in DDO ${elem.documentId}` @@ -186,15 +230,43 @@ export class PaidComputeStartHandler extends CommandHandler { } } } + // check credentials on service level + // if using a policy server and we are here it means that access was granted (they are merged/assessed together) + if (service.credentials) { + let accessGrantedServiceLevel: boolean + if (isPolicyServerConfigured()) { + // we use the previous check or we do it again + // (in case there is no DDO level credentials and we only have Service level ones) + const response = await policyServer.checkStartCompute( + ddoInstance.getDid(), + ddo, + elem.serviceId, + task.consumerAddress, + task.policyServer + ) + accessGrantedServiceLevel = accessGrantedDDOLevel || response.success + } else { + accessGrantedServiceLevel = areKnownCredentialTypes(service.credentials) + ? checkCredentials(service.credentials, task.consumerAddress) + : true + } + + if (!accessGrantedServiceLevel) { + CORE_LOGGER.logMessage( + `Error: Access to service with id ${service.id} was denied`, + true + ) + return { + stream: null, + status: { + httpStatus: 403, + error: `Error: Access to service with id ${service.id} was denied` + } + } + } + } const config = await getConfiguration() - const ddoInstance = DDOManager.getDDOClass(ddo) - const { - chainId: ddoChainId, - services, - metadata, - nftAddress - } = ddoInstance.getDDOFields() const { rpc, network, chainId, fallbackRPCs } = config.supportedNetworks[ddoChainId] const blockchain = new Blockchain(rpc, network, chainId, fallbackRPCs) @@ -521,6 +593,106 @@ export class FreeComputeStartHandler extends CommandHandler { } } } + const policyServer = new PolicyServer() + for (const elem of [...[task.algorithm], ...task.datasets]) { + if (!('documentId' in elem)) { + continue + } + const ddo = await new FindDdoHandler(this.getOceanNode()).findAndFormatDdo( + elem.documentId + ) + if (!ddo) { + const error = `DDO ${elem.documentId} not found` + return { + stream: null, + status: { + httpStatus: 500, + error + } + } + } + const ddoInstance = DDOManager.getDDOClass(ddo) + const { credentials } = ddoInstance.getDDOFields() + // check credentials (DDO level) + let accessGrantedDDOLevel: boolean + if (credentials) { + // if POLICY_SERVER_URL exists, then ocean-node will NOT perform any checks. + // It will just use the existing code and let PolicyServer decide. + if (isPolicyServerConfigured()) { + const response = await policyServer.checkStartCompute( + ddoInstance.getDid(), + ddo, + elem.serviceId, + task.consumerAddress, + task.policyServer + ) + accessGrantedDDOLevel = response.success + } else { + accessGrantedDDOLevel = areKnownCredentialTypes(credentials as Credentials) + ? checkCredentials(credentials as Credentials, task.consumerAddress) + : true + } + if (!accessGrantedDDOLevel) { + CORE_LOGGER.logMessage( + `Error: Access to asset ${ddoInstance.getDid()} was denied`, + true + ) + return { + stream: null, + status: { + httpStatus: 403, + error: `Error: Access to asset ${ddoInstance.getDid()} was denied` + } + } + } + } + const service = AssetUtils.getServiceById(ddo, elem.serviceId) + if (!service) { + const error = `Cannot find service ${elem.serviceId} in DDO ${elem.documentId}` + return { + stream: null, + status: { + httpStatus: 500, + error + } + } + } + // check credentials on service level + // if using a policy server and we are here it means that access was granted (they are merged/assessed together) + if (service.credentials) { + let accessGrantedServiceLevel: boolean + if (isPolicyServerConfigured()) { + // we use the previous check or we do it again + // (in case there is no DDO level credentials and we only have Service level ones) + const response = await policyServer.checkStartCompute( + ddo.id, + ddo, + service.id, + task.consumerAddress, + task.policyServer + ) + accessGrantedServiceLevel = accessGrantedDDOLevel || response.success + } else { + accessGrantedServiceLevel = areKnownCredentialTypes(service.credentials) + ? checkCredentials(service.credentials, task.consumerAddress) + : true + } + + if (!accessGrantedServiceLevel) { + CORE_LOGGER.logMessage( + `Error: Access to service with id ${service.id} was denied`, + true + ) + return { + stream: null, + status: { + httpStatus: 403, + error: `Error: Access to service with id ${service.id} was denied` + } + } + } + } + } try { const env = await engine.getComputeEnvironment(null, task.environment) if (!env) { diff --git a/src/components/core/handler/downloadHandler.ts b/src/components/core/handler/downloadHandler.ts index 22adf86fc..1e5bdb4c1 100644 --- a/src/components/core/handler/downloadHandler.ts +++ b/src/components/core/handler/downloadHandler.ts @@ -39,7 +39,7 @@ import { import { sanitizeServiceFiles } from '../../../utils/util.js' import { OrdableAssetResponse } from '../../../@types/Asset.js' import { PolicyServer } from '../../policyServer/index.js' -import { Asset, DDO, Service } from '@oceanprotocol/ddo-js' +import { Asset, Credentials, DDO, DDOManager, Service } from '@oceanprotocol/ddo-js' export const FILE_ENCRYPTION_ALGORITHM = 'aes-256-cbc' export function isOrderingAllowedForAsset(asset: Asset): OrdableAssetResponse { @@ -254,6 +254,14 @@ export class DownloadHandler extends CommandHandler { } } } + const ddoInstance = DDOManager.getDDOClass(ddo) + const { + chainId: ddoChainId, + nftAddress, + metadata, + credentials + } = ddoInstance.getDDOFields() + const policyServer = new PolicyServer() const isOrdable = isOrderingAllowedForAsset(ddo) if (!isOrdable.isOrdable) { @@ -268,7 +276,7 @@ export class DownloadHandler extends CommandHandler { } // 2. Validate ddo and credentials - if (!ddo.chainId || !ddo.nftAddress || !ddo.metadata) { + if (!ddoChainId || !nftAddress || !metadata) { CORE_LOGGER.logMessage('Error: DDO malformed or disabled', true) return { stream: null, @@ -281,33 +289,33 @@ export class DownloadHandler extends CommandHandler { // check credentials (DDO level) let accessGrantedDDOLevel: boolean - if (ddo.credentials) { + if (credentials) { // if POLICY_SERVER_URL exists, then ocean-node will NOT perform any checks. // It will just use the existing code and let PolicyServer decide. if (isPolicyServerConfigured()) { - accessGrantedDDOLevel = await ( - await new PolicyServer().checkDownload( - ddo.id, - ddo, - task.serviceId, - task.fileIndex, - task.transferTxId, - task.consumerAddress, - task.policyServer - ) - ).success + const response = await policyServer.checkDownload( + ddoInstance.getDid(), + ddo, + task.serviceId, + task.consumerAddress, + task.policyServer + ) + accessGrantedDDOLevel = response.success } else { - accessGrantedDDOLevel = areKnownCredentialTypes(ddo.credentials) - ? checkCredentials(ddo.credentials, task.consumerAddress) + accessGrantedDDOLevel = areKnownCredentialTypes(credentials as Credentials) + ? checkCredentials(credentials as Credentials, task.consumerAddress) : true } if (!accessGrantedDDOLevel) { - CORE_LOGGER.logMessage(`Error: Access to asset ${ddo.id} was denied`, true) + CORE_LOGGER.logMessage( + `Error: Access to asset ${ddoInstance.getDid()} was denied`, + true + ) return { stream: null, status: { httpStatus: 403, - error: `Error: Access to asset ${ddo.id} was denied` + error: `Error: Access to asset ${ddoInstance.getDid()} was denied` } } } @@ -315,7 +323,7 @@ export class DownloadHandler extends CommandHandler { // from now on, we need blockchain checks const config = await getConfiguration() - const { rpc, network, chainId, fallbackRPCs } = config.supportedNetworks[ddo.chainId] + const { rpc, network, chainId, fallbackRPCs } = config.supportedNetworks[ddoChainId] let provider let blockchain try { @@ -365,19 +373,14 @@ export class DownloadHandler extends CommandHandler { if (isPolicyServerConfigured()) { // we use the previous check or we do it again // (in case there is no DDO level credentials and we only have Service level ones) - accessGrantedServiceLevel = - accessGrantedDDOLevel || - (await ( - await new PolicyServer().checkDownload( - ddo.id, - ddo, - task.serviceId, - task.fileIndex, - task.transferTxId, - task.consumerAddress, - task.policyServer - ) - ).success) + const response = await policyServer.checkDownload( + ddoInstance.getDid(), + ddo, + service.id, + task.consumerAddress, + task.policyServer + ) + accessGrantedServiceLevel = accessGrantedDDOLevel || response.success } else { accessGrantedServiceLevel = areKnownCredentialTypes(service.credentials) ? checkCredentials(service.credentials, task.consumerAddress) @@ -471,26 +474,6 @@ export class DownloadHandler extends CommandHandler { } } } - // policyServer check - const policyServer = new PolicyServer() - const policyStatus = await policyServer.checkDownload( - ddo.id, - ddo, - service.id, - task.fileIndex, - task.transferTxId, - task.consumerAddress, - task.policyServer - ) - if (!policyStatus.success) { - return { - stream: null, - status: { - httpStatus: 405, - error: policyStatus.message - } - } - } try { // 7. Decrypt the url diff --git a/src/components/httpRoutes/compute.ts b/src/components/httpRoutes/compute.ts index 96b964fe3..93d1538ad 100644 --- a/src/components/httpRoutes/compute.ts +++ b/src/components/httpRoutes/compute.ts @@ -30,6 +30,7 @@ import { PROTOCOL_COMMANDS, SERVICES_API_BASE_PATH } from '../../utils/constants import { Readable } from 'stream' import { HTTP_LOGGER } from '../../utils/logging/common.js' import { LOG_LEVELS_STR } from '../../utils/logging/Logger.js' +import { PolicyServerTask } from '../../@types/policyServer.js' export const computeRoutes = express.Router() @@ -77,6 +78,7 @@ computeRoutes.post(`${SERVICES_API_BASE_PATH}/compute`, async (req, res) => { datasets: (req.body.datasets as unknown as ComputeAsset[]) || null, payment: (req.body.payment as unknown as ComputePayment) || null, resources: (req.body.resources as unknown as ComputeResourceRequest[]) || null, + policyServer: (req.query.policyServer as PolicyServerTask) || null, metadata: req.body.metadata || null, authorization: req.headers?.authorization } @@ -120,6 +122,7 @@ computeRoutes.post(`${SERVICES_API_BASE_PATH}/freeCompute`, async (req, res) => datasets: (req.body.datasets as unknown as ComputeAsset[]) || null, resources: (req.body.resources as unknown as ComputeResourceRequest[]) || null, maxJobDuration: req.body.maxJobDuration || null, + policyServer: (req.query.policyServer as PolicyServerTask) || null, metadata: req.body.metadata || null, authorization: req.headers?.authorization } diff --git a/src/components/httpRoutes/provider.ts b/src/components/httpRoutes/provider.ts index 73830570d..141e5c4d2 100644 --- a/src/components/httpRoutes/provider.ts +++ b/src/components/httpRoutes/provider.ts @@ -13,6 +13,7 @@ import { FeesHandler } from '../core/handler/feesHandler.js' import { BaseFileObject, EncryptMethod } from '../../@types/fileObject.js' import { P2PCommandResponse } from '../../@types/OceanNode.js' import { getEncryptMethodFromString } from '../../utils/crypt.js' +import { PolicyServerTask } from '../../@types/policyServer.js' export const providerRoutes = express.Router() @@ -153,7 +154,7 @@ providerRoutes.get(`${SERVICES_API_BASE_PATH}/initialize`, async (req, res) => { serviceId: (req.query.serviceId as string) || null, consumerAddress: (req.query.consumerAddress as string) || null, validUntil: parseInt(req.query.validUntil as string) || null, - policyServer: req.query.policyServer || null + policyServer: (req.query.policyServer as PolicyServerTask) || null }) if (result.stream) { const initializeREsponse = await streamToObject(result.stream as Readable) @@ -221,7 +222,7 @@ providerRoutes.get( consumerAddress: consumerAddress as string, signature: signature as string, command: PROTOCOL_COMMANDS.DOWNLOAD, - policyServer: req.query.policyServer || null, + policyServer: (req.query.policyServer as PolicyServerTask) || null, authorization: authorization as string } diff --git a/src/components/policyServer/index.ts b/src/components/policyServer/index.ts index e5d6425aa..24bb0de6c 100644 --- a/src/components/policyServer/index.ts +++ b/src/components/policyServer/index.ts @@ -69,7 +69,7 @@ export class PolicyServer { return await this.askServer(command) } - async checkInitialize( + async checkDownload( documentId: string, ddo: DDO, serviceId: string, @@ -77,7 +77,7 @@ export class PolicyServer { policyServer: any ): Promise { const command = { - action: 'initialize', + action: 'download', documentId, ddo, serviceId, @@ -87,22 +87,18 @@ export class PolicyServer { return await this.askServer(command) } - async checkDownload( + async checkStartCompute( documentId: string, ddo: DDO, serviceId: string, - fileIndex: number, - transferTxId: string, consumerAddress: string, policyServer: any ): Promise { const command = { - action: 'download', + action: 'startCompute', documentId, ddo, serviceId, - fileIndex, - transferTxId, consumerAddress, policyServer } diff --git a/src/test/data/assets.ts b/src/test/data/assets.ts index 8c3c763a5..2dae5c965 100644 --- a/src/test/data/assets.ts +++ b/src/test/data/assets.ts @@ -22,7 +22,7 @@ export const downloadAsset = { services: [ { id: 'ccb398c50d6abd5b456e8d7242bd856a1767a890b537c2f8c10ba8b8a10e6025', - type: 'download', + type: 'access', files: { files: [ { @@ -93,7 +93,7 @@ export const downloadAssetWithCredentials = { id: '', nftAddress: '', version: '4.1.0', - chainId: 80001, + chainId: 8996, metadata: { created: '2021-12-20T14:35:20Z', updated: '2021-12-20T14:35:20Z', @@ -111,7 +111,7 @@ export const downloadAssetWithCredentials = { services: [ { id: 'ccb398c50d6abd5b456e8d7242bd856a1767a890b537c2f8c10ba8b8a10e6025', - type: 'download', + type: 'access', files: { files: [ { @@ -150,12 +150,12 @@ export const downloadAssetWithCredentials = { } } -export const computeAsset = { +export const computeAssetWithCredentials = { '@context': ['https://w3id.org/did/v1'], id: '', nftAddress: '', version: '4.1.0', - chainId: 80001, + chainId: 8996, metadata: { created: '2021-12-20T14:35:20Z', updated: '2021-12-20T14:35:20Z', @@ -169,9 +169,10 @@ export const computeAsset = { termsAndConditions: true } }, + credentials: nftLevelCredentials, services: [ { - id: '1155995dda741e93afe4b1c6ced2d01734a6ec69865cc0997daf1f4db7259a36', + id: 'ccb398c50d6abd5b456e8d7242bd856a1767a890b537c2f8c10ba8b8a10e6025', type: 'compute', files: { files: [ @@ -182,6 +183,7 @@ export const computeAsset = { } ] }, + credentials: serviceLevelCredentials, datatokenAddress: '', serviceEndpoint: 'https://v4.provider.oceanprotocol.com', timeout: 86400, @@ -216,12 +218,12 @@ export const computeAsset = { } } -export const algoAsset = { +export const algoAssetWithCredentials = { '@context': ['https://w3id.org/did/v1'], id: '', nftAddress: '', version: '4.1.0', - chainId: 137, + chainId: 8996, metadata: { created: '2023-08-01T17:09:39Z', updated: '2023-08-01T17:09:39Z', @@ -245,6 +247,7 @@ export const algoAsset = { } } }, + credentials: nftLevelCredentials, services: [ { id: 'db164c1b981e4d2974e90e61bda121512e6909c1035c908d68933ae4cfaba6b0', @@ -260,8 +263,64 @@ export const algoAsset = { } ] }, + credentials: serviceLevelCredentials, timeout: 86400, + serviceEndpoint: 'https://v4.provider.oceanprotocol.com' + } + ], + stats: { + allocated: 0, + orders: 0, + price: { + value: '0' + } + }, + nft: { + address: '', + name: 'Ocean Data NFT', + symbol: 'OCEAN-NFT', + state: 5, + tokenURI: '', + owner: '', + created: '' + } +} + +export const computeAsset = { + '@context': ['https://w3id.org/did/v1'], + id: '', + nftAddress: '', + version: '4.1.0', + chainId: 8996, + metadata: { + created: '2021-12-20T14:35:20Z', + updated: '2021-12-20T14:35:20Z', + type: 'dataset', + name: 'cli fixed asset', + description: 'asset published using ocean.js cli tool', + tags: ['test'], + author: 'oceanprotocol', + license: 'https://market.oceanprotocol.com/terms', + additionalInformation: { + termsAndConditions: true + } + }, + services: [ + { + id: '1155995dda741e93afe4b1c6ced2d01734a6ec69865cc0997daf1f4db7259a36', + type: 'compute', + files: { + files: [ + { + type: 'url', + url: 'https://raw.githubusercontent.com/oceanprotocol/testdatasets/main/shs_dataset_test.txt', + method: 'GET' + } + ] + }, + datatokenAddress: '', serviceEndpoint: 'https://v4.provider.oceanprotocol.com', + timeout: 86400, compute: { allowRawAlgorithm: false, allowNetworkAccess: true, @@ -270,6 +329,77 @@ export const algoAsset = { } } ], + event: {}, + nft: { + address: '', + name: 'Ocean Data NFT', + symbol: 'OCEAN-NFT', + state: 5, + tokenURI: '', + owner: '', + created: '' + }, + purgatory: { + state: false + }, + datatokens: [] as any, + stats: { + allocated: 0, + orders: 0, + price: { + value: '0' + } + } +} + +export const algoAsset = { + '@context': ['https://w3id.org/did/v1'], + id: '', + nftAddress: '', + version: '4.1.0', + chainId: 8996, + metadata: { + created: '2023-08-01T17:09:39Z', + updated: '2023-08-01T17:09:39Z', + type: 'algorithm', + name: 'CLi Algo', + description: 'Cli algo', + author: 'OPF', + license: 'https://market.oceanprotocol.com/terms', + additionalInformation: { + termsAndConditions: true + }, + algorithm: { + language: '', + version: '0.1', + container: { + entrypoint: 'node $ALGO', + image: 'node', + tag: 'latest', + checksum: + 'sha256:1155995dda741e93afe4b1c6ced2d01734a6ec69865cc0997daf1f4db7259a36' + } + } + }, + services: [ + { + id: 'db164c1b981e4d2974e90e61bda121512e6909c1035c908d68933ae4cfaba6b0', + type: 'access', + files: { + files: [ + { + type: 'url', + method: 'GET', + url: 'https://raw.githubusercontent.com/oceanprotocol/test-algorithm/master/javascript/algo.js', + contentType: 'text/js', + encoding: 'UTF-8' + } + ] + }, + timeout: 86400, + serviceEndpoint: 'https://v4.provider.oceanprotocol.com' + } + ], stats: { allocated: 0, orders: 0, diff --git a/src/test/integration/compute.test.ts b/src/test/integration/compute.test.ts index 6c329256c..412e1f4e3 100644 --- a/src/test/integration/compute.test.ts +++ b/src/test/integration/compute.test.ts @@ -1215,7 +1215,11 @@ describe('Compute', () => { datasetDDOTest.services[0].id, oceanNode ) - expect(result).to.equal(!setTrustedAlgosEmpty) + // datasetDDOTest does not have set + // publisherTrustedAlgorithms, nor + // publisherTrustedAlgorithmPublishers + // expect the result to be true + expect(result).to.equal(true) } else expect(expectedTimeoutFailure(this.test.title)).to.be.equal(wasTimeout) } else expect(expectedTimeoutFailure(this.test.title)).to.be.equal(wasTimeout) }) diff --git a/src/test/integration/credentials.test.ts b/src/test/integration/credentials.test.ts index c04ce2da8..8f43a3457 100644 --- a/src/test/integration/credentials.test.ts +++ b/src/test/integration/credentials.test.ts @@ -51,29 +51,45 @@ import { getOceanArtifactsAdressesByChainId } from '../../utils/address.js' import { publishAsset, orderAsset } from '../utils/assets.js' -import { downloadAssetWithCredentials } from '../data/assets.js' +import { + algoAsset, + computeAssetWithCredentials, + downloadAssetWithCredentials +} from '../data/assets.js' import { ganachePrivateKeys } from '../utils/addresses.js' import { homedir } from 'os' import AccessListFactory from '@oceanprotocol/contracts/artifacts/contracts/accesslists/AccessListFactory.sol/AccessListFactory.json' assert { type: 'json' } import AccessList from '@oceanprotocol/contracts/artifacts/contracts/accesslists/AccessList.sol/AccessList.json' assert { type: 'json' } import { deployAccessListContract, getContract } from '../utils/contracts.js' +import { ComputeInitializeHandler } from '../../components/core/compute/initialize.js' +import { ComputeAlgorithm, ComputeAsset } from '../../@types/index.js' +import { ComputeGetEnvironmentsHandler } from '../../components/core/compute/environments.js' +import { ComputeInitializeCommand } from '../../@types/commands.js' -describe('Should run a complete node flow.', () => { +describe('[Credentials Flow] - Should run a complete node flow.', () => { let config: OceanNodeConfig let oceanNode: OceanNode let provider: JsonRpcProvider + let computeEnvironments: any + let firstEnv: any let publisherAccount: Signer let consumerAccounts: Signer[] let consumerAddresses: string[] let ddo: any + let computeDdo: any + let algoDdo: any let did: string + let computeDid: string + let algoDid: string const orderTxIds: string[] = [] const mockSupportedNetworks: RPCS = getMockSupportedNetworks() let previousConfiguration: OverrideEnvConfig[] + let artifactsAddresses: any + let paymentToken: string let blockchain: Blockchain let contractAcessList: Contract @@ -82,6 +98,8 @@ describe('Should run a complete node flow.', () => { before(async () => { provider = new JsonRpcProvider('http://127.0.0.1:8545') publisherAccount = (await provider.getSigner(0)) as Signer + artifactsAddresses = getOceanArtifactsAdresses() + paymentToken = artifactsAddresses.development.Ocean // override and save configuration (always before calling getConfig()) previousConfiguration = await setupEnvironment( @@ -94,7 +112,8 @@ describe('Should run a complete node flow.', () => { ENVIRONMENT_VARIABLES.AUTHORIZED_DECRYPTERS, ENVIRONMENT_VARIABLES.ALLOWED_ADMINS, ENVIRONMENT_VARIABLES.AUTHORIZED_PUBLISHERS, - ENVIRONMENT_VARIABLES.ADDRESS_FILE + ENVIRONMENT_VARIABLES.ADDRESS_FILE, + ENVIRONMENT_VARIABLES.DOCKER_COMPUTE_ENVIRONMENTS ], [ JSON.stringify(mockSupportedNetworks), @@ -105,7 +124,12 @@ describe('Should run a complete node flow.', () => { JSON.stringify([ await publisherAccount.getAddress() // signer 0 ]), - `${homedir}/.ocean/ocean-contracts/artifacts/address.json` + `${homedir}/.ocean/ocean-contracts/artifacts/address.json`, + '[{"socketPath":"/var/run/docker.sock","resources":[{"id":"disk","total":1000000000}],"storageExpiry":604800,"maxJobDuration":3600,"fees":{"' + + DEVELOPMENT_CHAIN_ID + + '":[{"feeToken":"' + + paymentToken + + '","prices":[{"id":"cpu","price":1}]}]},"free":{"maxJobDuration":60,"maxJobs":3,"resources":[{"id":"cpu","max":1},{"id":"ram","max":1000000000},{"id":"disk","max":1000000000}]}}]' ] ) ) @@ -115,6 +139,7 @@ describe('Should run a complete node flow.', () => { oceanNode = await OceanNode.getInstance(config, database) const indexer = new OceanIndexer(database, config.indexingNetworks) oceanNode.addIndexer(indexer) + await oceanNode.addC2DEngines() const rpcs: RPCS = config.supportedNetworks const chain: SupportedNetwork = rpcs[String(DEVELOPMENT_CHAIN_ID)] @@ -178,13 +203,20 @@ describe('Should run a complete node flow.', () => { }) it('should publish download datasets', async function () { - this.timeout(DEFAULT_TEST_TIMEOUT * 3) + this.timeout(DEFAULT_TEST_TIMEOUT * 5) const publishedDataset = await publishAsset( downloadAssetWithCredentials, publisherAccount ) + const publishedComputeDataset = await publishAsset( + computeAssetWithCredentials, + publisherAccount + ) + + const publishedAlgo = await publishAsset(algoAsset, publisherAccount) + did = publishedDataset.ddo.id const { ddo, wasTimeout } = await waitToIndex( did, @@ -194,6 +226,30 @@ describe('Should run a complete node flow.', () => { if (!ddo) { assert(wasTimeout === true, 'published failed due to timeout!') } + + computeDid = publishedComputeDataset.ddo.id + const resolvedComputeDdo = await waitToIndex( + computeDid, + EVENTS.METADATA_CREATED, + DEFAULT_TEST_TIMEOUT * 3 + ) + const ddoCompute = resolvedComputeDdo.ddo + const timeoutCompute = resolvedComputeDdo.wasTimeout + if (!ddoCompute) { + assert(timeoutCompute === true, 'published failed due to timeout!') + } + + algoDid = publishedAlgo.ddo.id + const resolvedAlgo = await waitToIndex( + algoDid, + EVENTS.METADATA_CREATED, + DEFAULT_TEST_TIMEOUT * 3 + ) + const algo = resolvedAlgo.ddo + const timeoutAlgo = resolvedAlgo.wasTimeout + if (!algo) { + assert(timeoutAlgo === true, 'published failed due to timeout!') + } }) it('should fetch the published ddo', async () => { @@ -201,9 +257,25 @@ describe('Should run a complete node flow.', () => { command: PROTOCOL_COMMANDS.GET_DDO, id: did } - const response = await new GetDdoHandler(oceanNode).handle(getDDOTask) + let response = await new GetDdoHandler(oceanNode).handle(getDDOTask) ddo = await streamToObject(response.stream as Readable) assert(ddo.id === did, 'DDO id not matching') + + const getComputeDDOTask = { + command: PROTOCOL_COMMANDS.GET_DDO, + id: computeDid + } + response = await new GetDdoHandler(oceanNode).handle(getComputeDDOTask) + computeDdo = await streamToObject(response.stream as Readable) + assert(computeDdo.id === computeDid, 'computeDdo id not matching') + + const getAlgoDDOTask = { + command: PROTOCOL_COMMANDS.GET_DDO, + id: algoDid + } + response = await new GetDdoHandler(oceanNode).handle(getAlgoDDOTask) + algoDdo = await streamToObject(response.stream as Readable) + assert(algoDdo.id === algoDid, 'computeDdo id not matching') }) it('should start an order for all consumers', async function () { @@ -267,6 +339,115 @@ describe('Should run a complete node flow.', () => { await doCheck() }) + it('should initializeCompute work for first consumer', async function () { + this.timeout(DEFAULT_TEST_TIMEOUT * 3) + + const consumerAddress = consumerAddresses[0] + + const dataset: ComputeAsset = { + documentId: computeDid, + serviceId: computeDdo.services[0].id + } + const algorithm: ComputeAlgorithm = { + documentId: algoDid, + serviceId: algoDdo.services[0].id + } + const getEnvironmentsTask = { + command: PROTOCOL_COMMANDS.COMPUTE_GET_ENVIRONMENTS + } + const resp = await new ComputeGetEnvironmentsHandler(oceanNode).handle( + getEnvironmentsTask + ) + computeEnvironments = await streamToObject(resp.stream as Readable) + firstEnv = computeEnvironments[0] + + const initializeComputeTask: ComputeInitializeCommand = { + datasets: [dataset], + algorithm, + environment: firstEnv.id, + payment: { + chainId: DEVELOPMENT_CHAIN_ID, + token: paymentToken + }, + maxJobDuration: 2 * 60, + consumerAddress, + command: PROTOCOL_COMMANDS.COMPUTE_INITIALIZE + } + const response = await new ComputeInitializeHandler(oceanNode).handle( + initializeComputeTask + ) + assert(response) + assert(response.stream, 'stream not present') + assert(response.status.httpStatus === 200, 'http status not 200') + expect(response.stream).to.be.instanceOf(Readable) + }) + + it('should NOT initializeCompute for second consumer - service level credentials', async function () { + this.timeout(DEFAULT_TEST_TIMEOUT * 3) + + const consumerAddress = consumerAddresses[1] + + const dataset: ComputeAsset = { + documentId: computeDid, + serviceId: computeDdo.services[0].id + } + const algorithm: ComputeAlgorithm = { + documentId: algoDid, + serviceId: algoDdo.services[0].id + } + const initializeComputeTask: ComputeInitializeCommand = { + datasets: [dataset], + algorithm, + environment: firstEnv.id, + payment: { + chainId: DEVELOPMENT_CHAIN_ID, + token: paymentToken + }, + maxJobDuration: 2 * 60, + consumerAddress, + command: PROTOCOL_COMMANDS.COMPUTE_INITIALIZE + } + const response = await new ComputeInitializeHandler(oceanNode).handle( + initializeComputeTask + ) + assert(response) + assert(response.stream === null, 'stream is present') + assert(response.status.httpStatus === 403, 'http status not 403') + }) + + it('should NOT initializeCompute for third consumer - asset level credentials', async function () { + this.timeout(DEFAULT_TEST_TIMEOUT * 3) + + const consumerAddress = consumerAddresses[2] + + const dataset: ComputeAsset = { + documentId: computeDid, + serviceId: computeDdo.services[0].id + } + const algorithm: ComputeAlgorithm = { + documentId: algoDid, + serviceId: algoDdo.services[0].id + } + const initializeComputeTask: ComputeInitializeCommand = { + datasets: [dataset], + algorithm, + environment: firstEnv.id, + payment: { + chainId: DEVELOPMENT_CHAIN_ID, + token: paymentToken + }, + maxJobDuration: 2 * 60, + consumerAddress, + command: PROTOCOL_COMMANDS.COMPUTE_INITIALIZE + } + const response = await new ComputeInitializeHandler(oceanNode).handle( + initializeComputeTask + ) + assert(response) + assert(response.stream === null, 'stream is present') + assert(response.status.httpStatus === 403, 'http status not 403') + }) + it('should not allow to download the asset for second consumer - service level credentials', async function () { this.timeout(DEFAULT_TEST_TIMEOUT * 3) diff --git a/src/test/integration/download.test.ts b/src/test/integration/download.test.ts index 64b33c78c..fd629476b 100644 --- a/src/test/integration/download.test.ts +++ b/src/test/integration/download.test.ts @@ -44,7 +44,7 @@ import { downloadAsset } from '../data/assets.js' import { genericDDO } from '../data/ddo.js' import { homedir } from 'os' -describe('Should run a complete node flow.', () => { +describe('[Download Flow] - Should run a complete node flow.', () => { let config: OceanNodeConfig let database: Database let oceanNode: OceanNode