Authenticate PolicyServerPassthrough / initializePSVerification, and stop logging credentials

docs/PolicyServer.md

@@ -134,3 +134,49 @@ Called whenever a new decrypt command is received by Ocean Node
   "policyServer": {}
 }
 ```
+
+## Passthrough and caller identity
+
+`POST /api/services/PolicyServerPassthrough` lets a caller send an arbitrary payload
+straight to the PolicyServer, and `POST /api/services/initializePSVerification` starts an
+`initiate` flow. Both are **authenticated by Ocean Node**: the caller must supply either an
+`Authorization` header carrying an auth token, or a `nonce` + `signature` pair, together
+with `consumerAddress`. Unauthenticated requests get a `401` and never reach the
+PolicyServer.
+
+Because the passthrough body is forwarded verbatim, Ocean Node **overwrites** the identity
+fields after it has verified the caller. The payload the PolicyServer receives therefore
+always carries:
+
+| field           | set by     | meaning                                                            |
+| --------------- | ---------- | ------------------------------------------------------------------ |
+| consumerAddress | Ocean Node | the address this node verified — trustworthy, not caller-controlled |
+| authorization   | Ocean Node | the caller's auth token, relayed as received                       |
+| nonce           | Ocean Node | the caller's nonce (already consumed by this node)                 |
+| signature       | Ocean Node | the caller's signature, so the PolicyServer can re-verify it       |
+| ddo             | Ocean Node | the DDO resolved from `documentId`, or `null` if not found         |
+| nodeAddress     | Ocean Node | the address of the node making the request                         |
+
+Everything else in the payload — including `action` — is caller-supplied and must be
+treated as untrusted input.
+
+Each endpoint is its own command, and the command string is part of the signed message, so a
+signature is scoped to one endpoint and cannot be replayed against the other:
+
+| endpoint                              | command                    | signed message                                        |
+| ------------------------------------- | -------------------------- | ----------------------------------------------------- |
+| `/api/services/PolicyServerPassthrough`  | `PolicyServerPassthrough`  | `consumerAddress + nonce + "PolicyServerPassthrough"` |
+| `/api/services/initializePSVerification` | `PolicyServerInitialize`   | `consumerAddress + nonce + "PolicyServerInitialize"`  |
+
+A PolicyServer can independently recompute and verify either one. Note that for
+`initializePSVerification` the credentials arrive nested inside the `policyServer` object
+rather than at the top level.
+
+> **A caller controls `action`.** A passthrough payload can claim `"action": "download"` or
+> `"action": "startCompute"` and look much like the ones Ocean Node itself sends for those
+> commands. The `consumerAddress` is trustworthy, but the action is not — do not grant a
+> passthrough request the same authority as a node-initiated one.
+
+> **The caller's auth token leaves the node.** `authorization` is relayed to the
+> PolicyServer so it can run its own checks, so `POLICY_SERVER_URL` should be an HTTPS
+> endpoint the operator controls.

src/@types/commands.ts

@@ -323,13 +323,20 @@ export interface JobStatus {
 
 export interface PolicyServerPassthroughCommand extends Command {
   policyServerPassthrough?: any
+  // caller identity, verified by this node before anything is forwarded to the policy server.
+  // either the inherited "authorization" token, or consumerAddress + nonce + signature
+  consumerAddress?: string
+  nonce?: string
+  signature?: string
 }
 
 export interface PolicyServerInitializeCommand extends Command {
   documentId?: string
   serviceId?: string
   consumerAddress?: string
   policyServer?: any
+  nonce?: string
+  signature?: string
 }
 
 export interface CreateAuthTokenCommand extends Command {

src/components/core/handler/coreHandlersRegistry.ts

@@ -11,7 +11,10 @@ import {
 } from './ddoHandler.js'
 import { DownloadHandler } from './downloadHandler.js'
 import { FileInfoHandler } from './fileInfoHandler.js'
-import { PolicyServerPassthroughHandler } from './policyServer.js'
+import {
+  PolicyServerPassthroughHandler,
+  PolicyServerInitializeHandler
+} from './policyServer.js'
 import { EncryptHandler, EncryptFileHandler } from './encryptHandler.js'
 import { FeesHandler } from './feesHandler.js'
 import { BaseHandler, CommandHandler } from './handler.js'
@@ -129,6 +132,10 @@ export class CoreHandlersRegistry {
       PROTOCOL_COMMANDS.POLICY_SERVER_PASSTHROUGH,
       new PolicyServerPassthroughHandler(node)
     )
+    this.registerCoreHandler(
+      PROTOCOL_COMMANDS.POLICY_SERVER_INITIALIZE,
+      new PolicyServerInitializeHandler(node)
+    )
 
     this.registerCoreHandler(PROTOCOL_COMMANDS.VALIDATE_DDO, new ValidateDDOHandler(node))
     this.registerCoreHandler(

src/components/core/handler/policyServer.ts

@@ -4,6 +4,7 @@ import {
   PolicyServerInitializeCommand
 } from '../../../@types/commands.js'
 import { Readable } from 'stream'
+import { isAddress } from 'ethers'
 import { CommandHandler } from './handler.js'
 import {
   ValidateParams,
@@ -20,7 +21,21 @@ export class PolicyServerPassthroughHandler extends CommandHandler {
       return buildInvalidRequestMessage(
         'Invalid Request: missing policyServerPassthrough field!'
       )
-    const validation = validateCommandParameters(command, []) // all optional? weird
+    // we inject fields into this object below, so it has to be a keyed object. arrays are
+    // objects too, and would be forwarded as {"0":..,"1":..} with no action
+    if (
+      typeof command.policyServerPassthrough !== 'object' ||
+      Array.isArray(command.policyServerPassthrough)
+    )
+      return buildInvalidRequestMessage(
+        'Invalid Request: "policyServerPassthrough" must be an object!'
+      )
+    const validation = validateCommandParameters(command, ['consumerAddress'])
+    if (validation.valid && !isAddress(command.consumerAddress)) {
+      return buildInvalidRequestMessage(
+        'Parameter : "consumerAddress" is not a valid web3 address'
+      )
+    }
     return validation
   }
 
@@ -29,6 +44,17 @@ export class PolicyServerPassthroughHandler extends CommandHandler {
     if (this.shouldDenyTaskHandling(validationResponse)) {
       return validationResponse
     }
+    // same auth contract as startCompute: an authorization token, or nonce + signature
+    const authValidationResponse = await this.validateTokenOrSignature(
+      task.authorization,
+      task.consumerAddress,
+      task.nonce,
+      task.signature,
+      task.command
+    )
+    if (authValidationResponse.status.httpStatus !== 200) {
+      return authValidationResponse
+    }
     task.policyServerPassthrough.ddo = null
     // resolve DDO first
     try {
@@ -41,6 +67,13 @@ export class PolicyServerPassthroughHandler extends CommandHandler {
         `PolicyServerPassthroughHandler: DDO not found for documentId ${task.policyServerPassthrough.documentId}: ${error.message}`
       )
     }
+    // the passthrough payload is forwarded verbatim, so every identity field has to be
+    // (re)written here, after validation. otherwise a caller could forge consumerAddress
+    // and impersonate the typed actions (download, startCompute, ...)
+    task.policyServerPassthrough.consumerAddress = authValidationResponse.consumerAddress
+    task.policyServerPassthrough.authorization = task.authorization
+    task.policyServerPassthrough.nonce = task.nonce
+    task.policyServerPassthrough.signature = task.signature
     // policyServer check
     const policyServer = new PolicyServer()
     const policyStatus = await policyServer.passThrough(task.policyServerPassthrough)
@@ -71,7 +104,12 @@ export class PolicyServerInitializeHandler extends CommandHandler {
       'documentId',
       'serviceId',
       'consumerAddress'
-    ]) // all optional? weird
+    ])
+    if (validation.valid && !isAddress(command.consumerAddress)) {
+      return buildInvalidRequestMessage(
+        'Parameter : "consumerAddress" is not a valid web3 address'
+      )
+    }
     return validation
   }
 
@@ -80,6 +118,17 @@ export class PolicyServerInitializeHandler extends CommandHandler {
     if (this.shouldDenyTaskHandling(validationResponse)) {
       return validationResponse
     }
+    // same auth contract as startCompute: an authorization token, or nonce + signature
+    const authValidationResponse = await this.validateTokenOrSignature(
+      task.authorization,
+      task.consumerAddress,
+      task.nonce,
+      task.signature,
+      task.command
+    )
+    if (authValidationResponse.status.httpStatus !== 200) {
+      return authValidationResponse
+    }
     // resolve DDO first
     try {
       const database = await this.getOceanNode().getDatabase()
@@ -98,12 +147,19 @@ export class PolicyServerInitializeHandler extends CommandHandler {
       }
       // policyServer check
       const policyServer = new PolicyServer()
+      // forward the address this node actually verified, plus the caller credentials,
+      // so the policy server can run its own additional checks
       const policyStatus = await policyServer.initializePSVerification(
         task.documentId,
         ddo,
         task.serviceId,
-        task.consumerAddress,
-        task.policyServer
+        authValidationResponse.consumerAddress,
+        {
+          ...task.policyServer,
+          authorization: task.authorization,
+          nonce: task.nonce,
+          signature: task.signature
+        }
       )
       if (!policyStatus.success) {
         return {

src/components/httpRoutes/policyServer.ts

@@ -21,6 +21,10 @@ PolicyServerPassthroughRoute.post(
       const response = await new PolicyServerPassthroughHandler(req.oceanNode).handle({
         command: PROTOCOL_COMMANDS.POLICY_SERVER_PASSTHROUGH,
         policyServerPassthrough: req.body.policyServerPassthrough,
+        consumerAddress: req.body.consumerAddress,
+        nonce: req.body.nonce,
+        signature: req.body.signature,
+        authorization: req.headers?.authorization,
         caller: req.caller
       })
       if (response.stream) {
@@ -49,11 +53,14 @@ PolicyServerPassthroughRoute.post(
     )
     try {
       const response = await new PolicyServerInitializeHandler(req.oceanNode).handle({
-        command: PROTOCOL_COMMANDS.POLICY_SERVER_PASSTHROUGH,
+        command: PROTOCOL_COMMANDS.POLICY_SERVER_INITIALIZE,
         documentId: req.body.documentId,
         serviceId: req.body.serviceId,
         consumerAddress: req.body.consumerAddress,
         policyServer: req.body.policyServer,
+        nonce: req.body.nonce,
+        signature: req.body.signature,
+        authorization: req.headers?.authorization,
         caller: req.caller
       })
       if (response.stream) {

src/components/httpRoutes/validateCommands.ts

@@ -2,6 +2,7 @@ import { PROTOCOL_COMMANDS, SUPPORTED_PROTOCOL_COMMANDS } from '../../utils/cons
 import { P2PCommandResponse } from '../../@types/OceanNode.js'
 import { Command } from '../../@types/commands.js'
 import { CORE_LOGGER } from '../../utils/logging/common.js'
+import { isDefined } from '../../utils/util.js'
 import { ReadableString } from '../P2P/handlers.js'
 
 export type ValidateParams = {
@@ -10,6 +11,69 @@ export type ValidateParams = {
   status?: number
 }
 
+// credentials present on (almost) any command. these must never reach the logs, on any
+// command, since they are what authorizes the request in the first place
+const SENSITIVE_COMMAND_FIELDS = [
+  'authorization', // auth token (JWT), usually taken from the Authorization header
+  'signature', // consumer signature authorizing this command
+  'aes_encrypted_key', // download: encrypted key material
+  'encryptedDockerRegistryAuth' // compute: encrypted docker registry credentials
+]
+
+// "token" is an ERC20 address on most commands (escrow, compute payment) and only a
+// credential on the auth-token commands, so it is redacted per-command instead
+const SENSITIVE_TOKEN_COMMANDS: string[] = [
+  PROTOCOL_COMMANDS.INVALIDATE_AUTH_TOKEN,
+  PROTOCOL_COMMANDS.VALIDATE_AUTH_TOKEN
+]
+
+const REDACTED = '[REDACTED]'
+
+/**
+ * Returns a copy of the payload with every credential field redacted, at any depth.
+ *
+ * This must not mutate its input: the clone the caller hands us can be a *shallow* copy
+ * (the fallback path below), so nested objects are still shared with the real command and
+ * the handlers still need the actual credentials. So we rebuild containers instead of
+ * writing into them.
+ *
+ * Only plain objects and arrays are walked - Buffers, typed arrays, Dates, streams and the
+ * like are passed through untouched.
+ */
+function redactSensitiveFields(
+  value: any,
+  redactToken: boolean,
+  seen: WeakSet<object>
+): any {
+  if (value === null || typeof value !== 'object') {
+    return value
+  }
+  if (seen.has(value)) {
+    return '[CIRCULAR]'
+  }
+  if (Array.isArray(value)) {
+    seen.add(value)
+    const copy = value.map((item) => redactSensitiveFields(item, redactToken, seen))
+    seen.delete(value)
+    return copy
+  }
+  const proto = Object.getPrototypeOf(value)
+  if (proto !== Object.prototype && proto !== null) {
+    return value
+  }
+  seen.add(value)
+  const copy: any = {}
+  for (const [key, item] of Object.entries(value)) {
+    if (SENSITIVE_COMMAND_FIELDS.includes(key) || (redactToken && key === 'token')) {
+      copy[key] = isDefined(item) ? REDACTED : item
+    } else {
+      copy[key] = redactSensitiveFields(item, redactToken, seen)
+    }
+  }
+  seen.delete(value)
+  return copy
+}
+
 // add others when we add suppor
 
 // request level validation, just check if we have a "command" field and its a supported one
@@ -56,6 +120,15 @@ export function validateCommandParameters(
     logCommandData.rawData = []
   }
 
+  // never log the caller's credentials, whatever the command is. credentials also show up
+  // nested (the free-form "policyServer" / "policyServerPassthrough" blobs), so this walks
+  // the whole payload
+  logCommandData = redactSensitiveFields(
+    logCommandData,
+    SENSITIVE_TOKEN_COMMANDS.includes(commandStr),
+    new WeakSet()
+  )
+
   CORE_LOGGER.info(
     `Checking received command data for Command "${commandStr}": ${JSON.stringify(
       logCommandData,