Integrate credentials checks in initializeCompute and startCompute handlers (#937)

* Integrate credentials checks for service and asset levels in initializeCompute and startCompute handlers. Added tests.

* skip lint.

* Update ci docker logs.

* Fix lint.

* Added credentials check for free start compute. Use download endpoint from policy server.

* log error.

* Check if did is provided on free start compute.

* cleanup logs.

* Integrate commands for compute from policy server.

* Fix condition in the test.

* Debug result.

* cleanup.

* print ddo test.

* Updated test.

* cleanup function.

* remove initialize compute command. Update initializeCompute handler.

* Enhance code.

* Define dedicated type for policy server when it comes on the handler task.

* Fix type of policy server on http requests.

* Fix commands usage for policy server.

* Fix arguments.

* Fix review.

* Use ddo.js for DDO fields manipulation.

* Update for free start compute.

* Update datasets samples.

* Update branch for cli.

* Fix datasets samples.

* Fix samples and ci.

* Update ddo.js.

* Update lock.

* Updated ddo schemas version.

* Fix samples.

* Add engines for test.

* Bring all tests back.

Author: mariacarmina

.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

package-lock.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",

package.json

@@ -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 {

src/@types/commands.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
+}

src/@types/policyServer.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]