Use delegate wallet to determine content node for listen tracking (#1314

)

* Add isFromContentNode check

* Add signed message to listen recording

* Add signature params to listen recording

* Clean up

* Address comments

* Add try catch

* Clean up

* Write out math

creator-node/src/apiSigning.js

@@ -23,6 +23,36 @@ const generateTimestampAndSignature = (data, privateKey) => {
   return { timestamp, signature: signedResponse.signature }
 }
 
+// Keeps track of a cached listen signature
+// Two field object: { timestamp, signature }
+let cachedListenSignature = null
+
+/**
+ * Generates a signature for `data` if only the previous signature
+ * generated is invalid (expired). Otherwise returns an existing signature.
+ * @param {string} privateKey
+ * @returns {object} {signature, timestamp} signature data
+ */
+const generateListenTimestampAndSignature = (privateKey) => {
+  if (cachedListenSignature) {
+    const signatureTimestamp = cachedListenSignature.timestamp
+    if (signatureHasExpired(signatureTimestamp)) {
+      // If the signature has expired, remove it from the cache
+      cachedListenSignature = null
+    } else {
+      // If the signature has not expired (still valid), use it!
+      return cachedListenSignature
+    }
+  }
+  // We don't have a signature already
+  const { timestamp, signature } = generateTimestampAndSignature(
+    { data: 'listen' },
+    privateKey
+  )
+  cachedListenSignature = { timestamp, signature }
+  return { timestamp, signature }
+}
+
 /**
    * Recover the public wallet address
    * @param {*} data obj with structure {...data, timestamp}
@@ -57,6 +87,7 @@ const sortKeys = x => {
 
 module.exports = {
   generateTimestampAndSignature,
+  generateListenTimestampAndSignature,
   recoverWallet,
   sortKeys,
   MAX_SIGNATURE_AGE_MS,

creator-node/src/routes/tracks.js

@@ -29,6 +29,7 @@ const { getCID } = require('./files')
 const { decode } = require('../hashids.js')
 const RehydrateIpfsQueue = require('../RehydrateIpfsQueue')
 const DBManager = require('../dbManager')
+const { generateListenTimestampAndSignature } = require('../apiSigning.js')
 
 const readFile = promisify(fs.readFile)
 
@@ -643,9 +644,17 @@ module.exports = function (app) {
 
     if (libs.identityService) {
       req.logger.info(`Logging listen for track ${blockchainId} by ${delegateOwnerWallet}`)
+      const signatureData = generateListenTimestampAndSignature(
+        config.get('delegatePrivateKey')
+      )
       // Fire and forget listen recording
       // TODO: Consider queueing these requests
-      libs.identityService.logTrackListen(blockchainId, delegateOwnerWallet, req.ip)
+      libs.identityService.logTrackListen(
+        blockchainId,
+        delegateOwnerWallet,
+        req.ip,
+        signatureData
+      )
     }
 
     req.params.CID = fileRecord.multihash

identity-service/src/app.js

@@ -13,7 +13,12 @@ const NotificationProcessor = require('./notifications/index.js')
 const { sendResponse, errorResponseServerError } = require('./apiHelpers')
 const { fetchAnnouncements } = require('./announcements')
 const { logger, loggingMiddleware } = require('./logging')
-const { getRateLimiter, getRateLimiterMiddleware, isIPWhitelisted, getIP } = require('./rateLimiter.js')
+const {
+  getRateLimiter,
+  getRateLimiterMiddleware,
+  isIPWhitelisted,
+  getIP
+} = require('./rateLimiter.js')
 
 const DOMAIN = 'mail.audius.co'
 
@@ -127,7 +132,7 @@ class App {
         const { ip, senderIP } = getIP(req)
         const ipToCheck = senderIP || ip
         // Do not apply user-specific rate limits for any whitelisted IP
-        return isIPWhitelisted(ipToCheck)
+        return isIPWhitelisted(ipToCheck, req)
       },
       keyGenerator: function (req) {
         const trackId = req.params.id
@@ -184,15 +189,15 @@ class App {
       expiry: ONE_HOUR_IN_SECONDS * 24,
       max: config.get('rateLimitingEthRelaysPerIPPerDay'),
       skip: function (req) {
-        return isIPWhitelisted(req.ip)
+        return isIPWhitelisted(req.ip, req)
       }
     })
     const ethRelayWalletRateLimiter = getRateLimiter({
       prefix: `ethRelayWalletRateLimiter`,
       expiry: ONE_HOUR_IN_SECONDS * 24,
       max: config.get('rateLimitingEthRelaysPerWalletPerDay'),
       skip: function (req) {
-        return isIPWhitelisted(req.ip)
+        return isIPWhitelisted(req.ip, req)
       },
       keyGenerator: function (req) {
         return req.body.senderAddress

identity-service/src/audiusLibsInstance.js

@@ -42,6 +42,23 @@ class AudiusLibsWrapper {
   getAudiusLibs () {
     return this.audiusLibsInstance
   }
+
+  /**
+   * Async getter for libs. Resolves when libs is initialized.
+   */
+  async getAudiusLibsAsync () {
+    if (this.audiusLibsInstance) {
+      return this.audiusLibsInstance
+    }
+    return new Promise(resolve => {
+      const i = setInterval(() => {
+        if (this.audiusLibsInstance) {
+          clearInterval(i)
+          resolve(this.audiusLibsInstance)
+        }
+      }, 1000)
+    })
+  }
 }
 
 const audiusLibsWrapper = new AudiusLibsWrapper()

identity-service/src/rateLimiter.js

@@ -3,13 +3,28 @@ const RedisStore = require('rate-limit-redis')
 const config = require('./config.js')
 const rateLimit = require('express-rate-limit')
 const express = require('express')
+const { isIPFromContentNode } = require('./utils/contentNodeIPCheck')
 const redisClient = new Redis(config.get('redisPort'), config.get('redisHost'))
 
 const DEFAULT_EXPIRY = 60 * 60 // one hour in seconds
 
-const isIPWhitelisted = (ip) => {
+const isIPWhitelisted = (ip, req) => {
+  // If the IP is either something in the regex whitelist or it is from
+  // a known content node, return true
   const whitelistRegex = config.get('rateLimitingListensIPWhitelist')
-  return whitelistRegex && !!ip.match(whitelistRegex)
+  const isWhitelisted = whitelistRegex && !!ip.match(whitelistRegex)
+
+  let isFromContentNode = false
+  try {
+    isFromContentNode = isIPFromContentNode(ip, req)
+  } catch (e) {
+    // Log out and continue if for some reason signature validation threw
+    req.logger.error(e)
+  }
+
+  // Don't return early so we can see logs for both paths
+  req.logger.info(`isIPWhitelisted - isWhitelisted: ${isWhitelisted}, isFromContentNode: ${isFromContentNode}`)
+  return isWhitelisted || isFromContentNode
 }
 
 const getIP = (req) => {
@@ -45,7 +60,7 @@ const getIP = (req) => {
   // either the actual user or a content node
   const senderIP = headers[headers.length - 2]
 
-  if (isIPWhitelisted(senderIP)) {
+  if (isIPWhitelisted(senderIP, req)) {
     const forwardedIP = headers[headers.length - 3]
     if (!forwardedIP) {
       req.logger.debug(`_getIP: content node sent a req that was missing a forwarded-for header, using IP: ${senderIP}, Forwarded-For: ${forwardedFor}`)
@@ -145,4 +160,9 @@ const getRateLimiterMiddleware = () => {
   return router
 }
 
-module.exports = { getIP, isIPWhitelisted, getRateLimiter, getRateLimiterMiddleware }
+module.exports = {
+  getIP,
+  isIPWhitelisted,
+  getRateLimiter,
+  getRateLimiterMiddleware
+}

identity-service/src/utils/apiSigning.js

@@ -0,0 +1,67 @@
+const Web3 = require('web3')
+const web3 = new Web3()
+
+// TODO: This is copied from the same code path in content node
+// and should be standardized across this file as well as the method in libs
+// Do not modify this file without touching the other!
+
+/**
+ * Max age of signature in milliseconds
+ * Set to 5 minutes
+ */
+const MAX_SIGNATURE_AGE_MS = 5 * 60 * 1000
+
+/**
+ * Generate the timestamp and signature for api signing
+ * @param {object} data
+ * @param {string} privateKey
+ */
+const generateTimestampAndSignature = (data, privateKey) => {
+  const timestamp = new Date().toISOString()
+  const toSignObj = { ...data, timestamp }
+  // JSON stringify automatically removes white space given 1 param
+  const toSignStr = JSON.stringify(sortKeys(toSignObj))
+  const toSignHash = web3.utils.keccak256(toSignStr)
+  const signedResponse = web3.eth.accounts.sign(toSignHash, privateKey)
+
+  return { timestamp, signature: signedResponse.signature }
+}
+
+/**
+ * Recover the public wallet address
+ * @param {object} data obj with structure {...data, timestamp}
+ * @param {string} signature signature generated with signed data
+ */
+const recoverWallet = (data, signature) => {
+  let structuredData = JSON.stringify(sortKeys(data))
+  const hashedData = web3.utils.keccak256(structuredData)
+  const recoveredWallet = web3.eth.accounts.recover(hashedData, signature)
+
+  return recoveredWallet
+}
+
+/**
+ * Returns boolean indicating if provided timestamp is older than MAX_SIGNATURE_AGE
+ * @param {string} signatureTimestamp unix timestamp string when signature was generated
+ */
+const signatureHasExpired = (signatureTimestamp) => {
+  const signatureTimestampDate = new Date(signatureTimestamp)
+  const currentTimestampDate = new Date()
+  const signatureAge = currentTimestampDate - signatureTimestampDate
+
+  return (signatureAge >= MAX_SIGNATURE_AGE_MS)
+}
+
+const sortKeys = x => {
+  if (typeof x !== 'object' || !x) { return x }
+  if (Array.isArray(x)) { return x.map(sortKeys) }
+  return Object.keys(x).sort().reduce((o, k) => ({ ...o, [k]: sortKeys(x[k]) }), {})
+}
+
+module.exports = {
+  generateTimestampAndSignature,
+  recoverWallet,
+  sortKeys,
+  MAX_SIGNATURE_AGE_MS,
+  signatureHasExpired
+}

identity-service/src/utils/contentNodeIPCheck.js

@@ -0,0 +1,78 @@
+const { logger } = require('../logging')
+const { recoverWallet, signatureHasExpired } = require('./apiSigning')
+const audiusLibsWrapper = require('../audiusLibsInstance')
+
+const FIND_CONTENT_NODES_INTERVAL_MS = 10 * 60 * 1000
+
+const KNOWN_CONTENT_NODE_IP_ADDRESSES = new Set([])
+let KNOWN_CONTENT_NODE_WALLETS = new Set([])
+
+/**
+ * Poll for content nodes and memoizes their delegate owner wallets
+ */
+const findContentNodes = async () => {
+  const libs = await audiusLibsWrapper.getAudiusLibsAsync()
+  const { ethContracts, ethWeb3Manager } = libs
+  const nodes = await ethContracts.getServiceProviderList('content-node')
+  const toChecksumAddress = ethWeb3Manager.getWeb3().utils.toChecksumAddress
+  KNOWN_CONTENT_NODE_WALLETS = new Set(
+    nodes.map(node => toChecksumAddress(node.delegateOwnerWallet))
+  )
+  logger.info(`findContentNodes - Known wallets: ${[...KNOWN_CONTENT_NODE_WALLETS]}`)
+}
+
+findContentNodes()
+setInterval(findContentNodes, FIND_CONTENT_NODES_INTERVAL_MS)
+
+/**
+ * Find whether a given IP belongs to a registered content node.
+ * If the ip is from an already known content node IP address, return true
+ * Otherwise, if the request has a signature and timestamp, recover the signing
+ * wallet and determine whether it is a registered content node
+ * @param {string} ip
+ * @param {Request} req
+ * @param {Set<string>} knownContentNodeWallets
+ * @param {Set<string>} knownContentNodeIPAddresses
+ */
+const _isIPFromContentNode = (ip, req, knownContentNodeWallets, knownContentNodeIPAddresses) => {
+  if (knownContentNodeIPAddresses.has(ip)) {
+    return true
+  }
+
+  if (!req.body || !req.body.signature || !req.body.timestamp) {
+    return false
+  }
+
+  const { signature, timestamp } = req.body
+
+  const hasExpired = signatureHasExpired(timestamp)
+  if (hasExpired) {
+    return false
+  }
+
+  req.logger.info(`isIPFromContentNode - Recovering signature: ${signature}, timestamp: ${timestamp}`)
+  req.logger.info(`isIPFromContentNode - Known wallets: ${[...knownContentNodeWallets]}, ips: ${[...knownContentNodeIPAddresses]}`)
+  const wallet = recoverWallet({
+    data: 'listen',
+    timestamp
+  }, signature)
+
+  req.logger.info(`isIPFromContentNode - Recovered wallet: ${wallet}`)
+  if (knownContentNodeWallets.has(wallet)) {
+    req.logger.info(`isIPFromContentNode - Was from content node`)
+    knownContentNodeIPAddresses.add(ip)
+    return true
+  }
+  req.logger.info(`isIPFromContentNode - Was not from content node`)
+  return false
+}
+
+const isIPFromContentNode = (ip, req) => {
+  return _isIPFromContentNode(ip, req, KNOWN_CONTENT_NODE_WALLETS, KNOWN_CONTENT_NODE_IP_ADDRESSES)
+}
+
+module.exports = {
+  // Exposed for testing purposes only
+  _isIPFromContentNode,
+  isIPFromContentNode
+}