Integrate credentials checks in initializeCompute and startCompute handlers

.github/workflows/ci.yml

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

src/@types/commands.ts

@@ -183,6 +183,7 @@ export interface ComputeInitializeCommand extends Command {
   consumerAddress: string
   signature?: string
   maxJobDuration: number
+  policyServer?: any // object to pass to policy server
 }
 
 export interface FreeComputeStartCommand extends Command {
@@ -195,6 +196,7 @@ export interface FreeComputeStartCommand extends Command {
   output?: ComputeOutput
   resources?: ComputeResourceRequest[]
   maxJobDuration?: number
+  policyServer?: any // object to pass to policy server
 }
 export interface PaidComputeStartCommand extends FreeComputeStartCommand {
   payment: ComputePayment

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 { areKnownCredentialTypes, checkCredentials } from '../../../utils/credentials.js'
+import { PolicyServer } from '../../policyServer/index.js'
 
 export class ComputeInitializeHandler extends CommandHandler {
   validate(command: ComputeInitializeCommand): ValidateParams {
@@ -205,6 +207,39 @@ export class ComputeInitializeHandler extends CommandHandler {
               }
             }
           }
+          // check credentials (DDO level)
+          let accessGrantedDDOLevel: boolean
+          if (ddo.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() && task.policyServer) {
+              accessGrantedDDOLevel = await (
+                await new PolicyServer().checkStartCompute(
+                  ddo.id,
+                  ddo,
+                  elem.serviceId,
+                  0,
+                  elem.transferTxId,
+                  task.consumerAddress,
+                  task.policyServer
+                )
+              ).success
+            } else {
+              accessGrantedDDOLevel = areKnownCredentialTypes(ddo.credentials)
+                ? checkCredentials(ddo.credentials, task.consumerAddress)
+                : true
+            }
+            if (!accessGrantedDDOLevel) {
+              CORE_LOGGER.logMessage(`Error: Access to asset ${ddo.id} was denied`, true)
+              return {
+                stream: null,
+                status: {
+                  httpStatus: 403,
+                  error: `Error: Access to asset ${ddo.id} was denied`
+                }
+              }
+            }
+          }
           const service = AssetUtils.getServiceById(ddo, elem.serviceId)
           if (!service) {
             const error = `Cannot find service ${elem.serviceId} in DDO ${elem.documentId}`
@@ -216,6 +251,46 @@ 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() && task.policyServer) {
+              // 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().checkStartCompute(
+                    ddo.id,
+                    ddo,
+                    elem.serviceId,
+                    0,
+                    elem.transferTxId,
+                    task.consumerAddress,
+                    task.policyServer
+                  )
+                ).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 ddoInstance = DDOManager.getDDOClass(ddo)
           const { chainId: ddoChainId, nftAddress } = ddoInstance.getDDOFields()

src/components/core/compute/startCompute.ts

@@ -26,14 +26,16 @@ 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 { getNonceAsNumber, checkNonce, NonceResponse } from '../utils/nonceHandler.js'
 import { createHash } from 'crypto'
+import { PolicyServer } from '../../policyServer/index.js'
+import { areKnownCredentialTypes, checkCredentials } from '../../../utils/credentials.js'
 
 export class PaidComputeStartHandler extends CommandHandler {
   validate(command: PaidComputeStartCommand): ValidateParams {
@@ -159,6 +161,39 @@ export class PaidComputeStartHandler extends CommandHandler {
               }
             }
           }
+          // check credentials (DDO level)
+          let accessGrantedDDOLevel: boolean
+          if (ddo.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() && task.policyServer) {
+              accessGrantedDDOLevel = await (
+                await new PolicyServer().checkStartCompute(
+                  ddo.id,
+                  ddo,
+                  elem.serviceId,
+                  0,
+                  elem.transferTxId,
+                  task.consumerAddress,
+                  task.policyServer
+                )
+              ).success
+            } else {
+              accessGrantedDDOLevel = areKnownCredentialTypes(ddo.credentials)
+                ? checkCredentials(ddo.credentials, task.consumerAddress)
+                : true
+            }
+            if (!accessGrantedDDOLevel) {
+              CORE_LOGGER.logMessage(`Error: Access to asset ${ddo.id} was denied`, true)
+              return {
+                stream: null,
+                status: {
+                  httpStatus: 403,
+                  error: `Error: Access to asset ${ddo.id} was denied`
+                }
+              }
+            }
+          }
           const service = AssetUtils.getServiceById(ddo, elem.serviceId)
           if (!service) {
             const error = `Cannot find service ${elem.serviceId} in DDO ${elem.documentId}`
@@ -170,6 +205,46 @@ 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() && task.policyServer) {
+              // 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().checkStartCompute(
+                    ddo.id,
+                    ddo,
+                    elem.serviceId,
+                    0,
+                    elem.transferTxId,
+                    task.consumerAddress,
+                    task.policyServer
+                  )
+                ).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)
@@ -521,6 +596,108 @@ export class FreeComputeStartHandler extends CommandHandler {
           }
         }
       }
+      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
+            }
+          }
+        }
+        // check credentials (DDO level)
+        let accessGrantedDDOLevel: boolean
+        if (ddo.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() && task.policyServer) {
+            accessGrantedDDOLevel = await (
+              await new PolicyServer().checkStartCompute(
+                ddo.id,
+                ddo,
+                elem.serviceId,
+                0,
+                elem.transferTxId,
+                task.consumerAddress,
+                task.policyServer
+              )
+            ).success
+          } else {
+            accessGrantedDDOLevel = areKnownCredentialTypes(ddo.credentials)
+              ? checkCredentials(ddo.credentials, task.consumerAddress)
+              : true
+          }
+          if (!accessGrantedDDOLevel) {
+            CORE_LOGGER.logMessage(`Error: Access to asset ${ddo.id} was denied`, true)
+            return {
+              stream: null,
+              status: {
+                httpStatus: 403,
+                error: `Error: Access to asset ${ddo.id} 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() && task.policyServer) {
+            // 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().checkStartCompute(
+                  ddo.id,
+                  ddo,
+                  elem.serviceId,
+                  0,
+                  elem.transferTxId,
+                  task.consumerAddress,
+                  task.policyServer
+                )
+              ).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) {

src/components/policyServer/index.ts

@@ -109,6 +109,44 @@ export class PolicyServer {
     return await this.askServer(command)
   }
 
+  async checkInitializeCompute(
+    documentId: string,
+    ddo: DDO,
+    serviceId: string,
+    consumerAddress: string,
+    policyServer: any
+  ): Promise<PolicyServerResult> {
+    return await this.checkInitialize(
+      documentId,
+      ddo,
+      serviceId,
+      consumerAddress,
+      policyServer
+    )
+  }
+
+  async checkStartCompute(
+    documentId: string,
+    ddo: DDO,
+    serviceId: string,
+    fileIndex: number,
+    transferTxId: string,
+    consumerAddress: string,
+    policyServer: any
+  ): Promise<PolicyServerResult> {
+    const command = {
+      action: 'startCompute',
+      documentId,
+      ddo,
+      serviceId,
+      fileIndex,
+      transferTxId,
+      consumerAddress,
+      policyServer
+    }
+    return await this.askServer(command)
+  }
+
   async passThrough(request: any): Promise<PolicyServerResult> {
     return await this.askServer(request)
   }