Feat/stage

package.json

@@ -1,6 +1,6 @@
 {
   "name": "ocean-node",
-  "version": "3.2.2",
+  "version": "3.2.21",
   "description": "Ocean Node is used to run all core services in the Ocean stack",
   "author": "Ocean Protocol Foundation",
   "license": "Apache-2.0",

src/@types/OceanNode.ts

@@ -143,6 +143,7 @@ export interface OceanNodeConfig {
   httpCertPath?: string
   httpKeyPath?: string
   enableBenchmark?: boolean
+  isPSConfigured: boolean
   persistentStorage?: PersistentStorageConfig
 }
 
@@ -185,6 +186,7 @@ export interface OceanNodeStatus {
   version: string
   http: boolean
   p2p: boolean
+  isPSConfigured: boolean
   provider: OceanNodeProvider[]
   indexer: OceanNodeIndexer[]
   escrowAddress: AddressPerChain

src/OceanNode.ts

@@ -23,6 +23,7 @@ import { createPersistentStorage } from './components/persistentStorage/createPe
 import { PersistentStorageFactory } from './components/persistentStorage/PersistentStorageFactory.js'
 import { isAddress, FallbackProvider, ethers } from 'ethers'
 import { create256Hash } from './utils/crypt.js'
+import { isPolicyServerConfigured } from './utils/config.js'
 
 export interface RequestLimiter {
   requester: string | string[] // IP address or peer ID
@@ -255,6 +256,7 @@ export class OceanNode {
   }
 
   public getConfig(): OceanNodeConfig {
+    this.config.isPSConfigured = isPolicyServerConfigured()
     return this.config
   }
 

src/components/core/handler/downloadHandler.ts

@@ -313,6 +313,11 @@ export class DownloadHandler extends CommandHandler {
     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.
+      CORE_LOGGER.logMessage('Checking DDO level credentials', true)
+      CORE_LOGGER.logMessage(
+        'isPolicyServerConfigured: ' + isPolicyServerConfigured(),
+        true
+      )
       if (isPolicyServerConfigured()) {
         const response = await policyServer.checkDownload(
           ddoInstance.getDid(),
@@ -321,14 +326,24 @@ export class DownloadHandler extends CommandHandler {
           task.consumerAddress,
           task.policyServer
         )
+        CORE_LOGGER.logMessage(
+          'Policy server response: ' + JSON.stringify(response),
+          true
+        )
         accessGrantedDDOLevel = response.success
       } else {
+        CORE_LOGGER.logMessage('Checking DDO level credentials INTERNAL', true)
+        CORE_LOGGER.logMessage(
+          'isPolicyServerConfigured: ' + isPolicyServerConfigured(),
+          true
+        )
         accessGrantedDDOLevel = await checkCredentials(
           task.consumerAddress,
           credentials as Credentials,
           await blockchain.getSigner()
         )
       }
+      CORE_LOGGER.logMessage('DDO level credentials check complete', true)
       if (!accessGrantedDDOLevel) {
         CORE_LOGGER.logMessage(
           `Error: Access to asset ${ddoInstance.getDid()} was denied`,

src/components/core/handler/feesHandler.ts

@@ -16,7 +16,10 @@ import { ProviderInitialize } from '../../../@types/Fees.js'
 import { getNonce } from '../utils/nonceHandler.js'
 import { streamToString } from '../../../utils/util.js'
 import { isOrderingAllowedForAsset } from './downloadHandler.js'
-import { DDOManager } from '@oceanprotocol/ddo-js'
+import { Credentials, DDOManager } from '@oceanprotocol/ddo-js'
+import { checkCredentials } from '../../../utils/credentials.js'
+import { isPolicyServerConfigured } from '../../../utils/index.js'
+import { PolicyServer } from '../../policyServer/index.js'
 
 export class FeesHandler extends CommandHandler {
   validate(command: GetFeesCommand): ValidateParams {
@@ -60,14 +63,164 @@ export class FeesHandler extends CommandHandler {
       }
     }
     const ddoInstance = DDOManager.getDDOClass(ddo)
-    const { services } = ddoInstance.getDDOFields() as any
+    const {
+      chainId: ddoChainId,
+      credentials,
+      services
+    } = ddoInstance.getDDOFields() as any
     const service = services.find((what: any) => what.id === task.serviceId)
     if (!service) {
       errorMsg = 'Invalid serviceId'
     }
-    if (service.type === 'compute') {
+    if (service?.type === 'compute') {
       errorMsg = 'Use the initializeCompute endpoint to initialize compute jobs'
     }
+
+    if (errorMsg) {
+      PROVIDER_LOGGER.logMessageWithEmoji(
+        errorMsg,
+        true,
+        GENERIC_EMOJIS.EMOJI_CROSS_MARK,
+        LOG_LEVELS_STR.LEVEL_ERROR
+      )
+      return {
+        stream: null,
+        status: {
+          httpStatus: 500,
+          error: errorMsg
+        }
+      }
+    }
+
+    const policyServer = new PolicyServer()
+    if (credentials || service.credentials) {
+      if (!task.consumerAddress) {
+        return {
+          stream: null,
+          status: {
+            httpStatus: 400,
+            error: 'consumerAddress is required to check credentials'
+          }
+        }
+      }
+
+      let signer
+      if (!isPolicyServerConfigured()) {
+        const oceanNode = this.getOceanNode()
+        const supportedNetwork = oceanNode.getConfig().supportedNetworks?.[ddoChainId]
+        if (!supportedNetwork) {
+          return {
+            stream: null,
+            status: {
+              httpStatus: 400,
+              error: `Fees handler: Unsupported chain id ${ddoChainId}`
+            }
+          }
+        }
+
+        const blockchain = oceanNode.getBlockchain(supportedNetwork.chainId)
+        if (!blockchain) {
+          return {
+            stream: null,
+            status: {
+              httpStatus: 400,
+              error: `Fees handler: Blockchain instance not available for chain ${supportedNetwork.chainId}`
+            }
+          }
+        }
+
+        const { ready, error } = await blockchain.isNetworkReady()
+        if (!ready) {
+          return {
+            stream: null,
+            status: {
+              httpStatus: 400,
+              error: `Fees handler: ${error}`
+            }
+          }
+        }
+        signer = await blockchain.getSigner()
+      }
+
+      let accessGrantedDDOLevel = false
+      PROVIDER_LOGGER.info(
+        `Checking credentials for consumer address: ${task.consumerAddress} AND credentials: ${JSON.stringify(credentials)}`
+      )
+      if (credentials) {
+        if (isPolicyServerConfigured()) {
+          PROVIDER_LOGGER.info(
+            `Checking credentials for consumer address on policy server: ${task.consumerAddress}`
+          )
+          const response = await policyServer.initializePSVerification(
+            ddoInstance.getDid(),
+            ddo,
+            task.serviceId,
+            task.consumerAddress,
+            task.policyServer
+          )
+          accessGrantedDDOLevel = response.success
+        } else {
+          PROVIDER_LOGGER.info(
+            `Checking credentials for consumer address internal: ${task.consumerAddress}`
+          )
+          accessGrantedDDOLevel = await checkCredentials(
+            task.consumerAddress,
+            credentials as Credentials,
+            signer
+          )
+        }
+      }
+
+      PROVIDER_LOGGER.info(`credentials: ${JSON.stringify(credentials)}`)
+      PROVIDER_LOGGER.info(`accessGrantedDDOLevel: ${accessGrantedDDOLevel}`)
+
+      if (credentials && !accessGrantedDDOLevel) {
+        const error = `Error: Access to asset ${ddoInstance.getDid()} was denied`
+        PROVIDER_LOGGER.error(error)
+        return {
+          stream: null,
+          status: { httpStatus: 403, error }
+        }
+      }
+
+      if (service.credentials) {
+        let accessGrantedServiceLevel: boolean
+        if (isPolicyServerConfigured()) {
+          const response = await policyServer.initializePSVerification(
+            ddoInstance.getDid(),
+            ddo,
+            service.id,
+            task.consumerAddress,
+            task.policyServer
+          )
+          accessGrantedServiceLevel = accessGrantedDDOLevel || response.success
+        } else {
+          PROVIDER_LOGGER.info(
+            'Checking credentials for consumer address internal for service: ' +
+              task.consumerAddress
+          )
+          accessGrantedServiceLevel = await checkCredentials(
+            task.consumerAddress,
+            service.credentials,
+            signer
+          )
+        }
+        PROVIDER_LOGGER.info(
+          `service.credentials: ${JSON.stringify(service.credentials)}`
+        )
+        PROVIDER_LOGGER.info(`accessGrantedServiceLevel: ${accessGrantedServiceLevel}`)
+
+        if (!accessGrantedServiceLevel) {
+          const error = `Error: Access to service with id ${service.id} was denied`
+          PROVIDER_LOGGER.error(error)
+          return {
+            stream: null,
+            status: { httpStatus: 403, error }
+          }
+        }
+      }
+    }
+
     const now = new Date().getTime() / 1000
     let validUntil = service.timeout === 0 ? 0 : now + service.timeout // first, make it service default
     if (task.validUntil && !isNaN(task.validUntil)) {

src/components/core/utils/nonceHandler.ts

@@ -200,16 +200,23 @@ async function validateNonceAndSignature(
     ['bytes'],
     [ethers.hexlify(ethers.toUtf8Bytes(message))]
   )
-  const messageHashBytes = ethers.toBeArray(consumerMessage)
+  const messageHashBytes = ethers.getBytes(consumerMessage)
+  const legacyMessageHashBytes = ethers.toBeArray(consumerMessage)
 
   // Try EOA signature validation
   try {
     const addressFromHashSignature = ethers.verifyMessage(consumerMessage, signature)
     const addressFromBytesSignature = ethers.verifyMessage(messageHashBytes, signature)
+    const addressFromLegacyBytesSignature = ethers.verifyMessage(
+      legacyMessageHashBytes,
+      signature
+    )
     if (
       ethers.getAddress(addressFromHashSignature)?.toLowerCase() ===
         ethers.getAddress(consumer)?.toLowerCase() ||
       ethers.getAddress(addressFromBytesSignature)?.toLowerCase() ===
+        ethers.getAddress(consumer)?.toLowerCase() ||
+      ethers.getAddress(addressFromLegacyBytesSignature)?.toLowerCase() ===
         ethers.getAddress(consumer)?.toLowerCase()
     ) {
       return { valid: true }

src/components/core/utils/statusHandler.ts

@@ -128,6 +128,7 @@ export async function status(
       version: getPackageVersion(),
       http: config.hasHttp,
       p2p: config.hasP2P,
+      isPSConfigured: config.isPSConfigured,
       provider: [],
       indexer: [],
       escrowAddress: {},
@@ -152,6 +153,7 @@ export async function status(
   nodeStatus.platform.freemem = os.freemem()
   nodeStatus.platform.loadavg = os.loadavg()
   nodeStatus.uptime = process.uptime()
+  nodeStatus.isPSConfigured = config.isPSConfigured
 
   // depends on request
   if (detailed) {

src/test/integration/credentials.test.ts

@@ -31,6 +31,7 @@ import {
 } from '../../utils/index.js'
 import { DownloadHandler } from '../../components/core/handler/downloadHandler.js'
 import { GetDdoHandler } from '../../components/core/handler/ddoHandler.js'
+import { ProviderFees } from '../../components/core/utils/feesHandler.js'
 
 import { Readable } from 'stream'
 import { OceanNodeConfig } from '../../@types/OceanNode.js'
@@ -306,28 +307,47 @@ describe('**********         [Credentials Flow] - Should run a complete node flo
   it('should start an order for all consumers', async function () {
     this.timeout(DEFAULT_TEST_TIMEOUT * 3)
 
+    const providerFeesFactory = new ProviderFees(oceanNode)
+    const createTestProviderFees = async (asset: any) => {
+      const service = asset.services[0]
+      const now = Date.now() / 1000
+      const validUntil = service.timeout === 0 ? 0 : now + service.timeout
+      const providerFees = await providerFeesFactory.createProviderFee(
+        asset,
+        service,
+        validUntil
+      )
+      assert(providerFees, 'provider fees could not be created')
+      return providerFees
+    }
+
+    const providerFees = await createTestProviderFees(ddo)
     for (let i = 0; i < 3; i++) {
       const orderTxReceipt = await orderAsset(
         ddo,
         0,
         consumerAccounts[i],
         consumerAddresses[i],
         publisherAccount,
-        oceanNode
+        oceanNode,
+        providerFees
       )
       assert(orderTxReceipt, `order transaction for consumer ${i} failed`)
       const txHash = orderTxReceipt.hash
       assert(txHash, `transaction id not found for consumer ${i}`)
       orderTxIds.push(txHash)
     }
+
+    const providerFeesWithMatchAll = await createTestProviderFees(ddoWithMatchAll)
     for (let i = 0; i < 3; i++) {
       const orderTxReceipt = await orderAsset(
         ddoWithMatchAll,
         0,
         consumerAccounts[i],
         consumerAddresses[i],
         publisherAccount,
-        oceanNode
+        oceanNode,
+        providerFeesWithMatchAll
       )
       assert(orderTxReceipt, `order transaction for consumer ${i} failed`)
       const txHash = orderTxReceipt.hash

src/utils/config/builder.ts

@@ -20,6 +20,7 @@ import { CONFIG_LOGGER } from '../logging/common.js'
 import { LOG_LEVELS_STR, GENERIC_EMOJIS } from '../logging/Logger.js'
 import { OceanNodeConfigSchema } from './schemas.js'
 import { ENV_TO_CONFIG_MAPPING } from './constants.js'
+import { isPolicyServerConfigured } from '../config.js'
 import { fileURLToPath } from 'url'
 import lodash from 'lodash'
 
@@ -281,6 +282,7 @@ export function buildMergedConfig(): OceanNodeConfig {
   config.c2dClusters = buildC2DClusters(
     config.dockerComputeEnvironments as C2DDockerConfig[]
   )
+  config.isPSConfigured = isPolicyServerConfigured()
 
   return config as OceanNodeConfig
 }