Add client id and client secret based token generation

index.js

@@ -3,6 +3,7 @@
 exports.BaseChannelAuth = require('./lib/base-channel-auth')
 exports.Channel = require('./lib/channel')
 exports.ChannelAuth = require('./lib/channel-auth')
+exports.ChannelToken = require('./lib/channel-token')
 exports.ConsumerError = require('./lib/consumer-error')
 exports.PermanentAuthenticationError = require('./lib/permanent-authentication-error')
 exports.PermanentError = require('./lib/permanent-error')

lib/channel-token.js

@@ -0,0 +1,144 @@
+'use strict'
+
+var inherits = require('inherits')
+var request = require('request')
+var BaseChannelAuth = require('./base-channel-auth')
+var PermanentAuthenticationError = require('./permanent-authentication-error')
+var TemporaryAuthenticationError = require('./temporary-authentication-error')
+var util = require('./util')
+
+var LOGIN_PATH_FRAGMENT = '/iam/v1.4/token'
+
+/**
+ * @classdesc Authentication class for use with channel requests.
+ * @param {String} base - Base URL to forward authentication requests to.
+ * @param {String} clientid - clientid to supply for request authentication.
+ * @param {String} clientsecret - clientsecret to supply for request authentication.
+ * @param {Object} [options] - Additional options to supply for request
+ *   authentication.
+ * @param {String} [options.key] - Optional client private keys in PEM format.
+ *   See
+ *   {@link https://nodejs.org/api/tls.html#tls_tls_createsecurecontext_options}.
+ * @param {String} [options.cert] - Optional client cert chains in PEM format.
+ *   See
+ *   {@link https://nodejs.org/api/tls.html#tls_tls_createsecurecontext_options}.
+ * @param {String} [options.ca] - Optionally override the trusted CA
+ *   certificates used to validate the authentication server. Any string can
+ *   contain multiple PEM CAs concatenated together.
+ *   See
+ *   {@link https://nodejs.org/api/tls.html#tls_tls_createsecurecontext_options}.
+ * @param {String} [options.passphrase] - Optional shared passphrase used for a
+ *   single private key. See
+ *   {@link https://nodejs.org/api/tls.html#tls_tls_createsecurecontext_options}.
+ * @param {Boolean} [options.rejectUnauthorized=true] - If not false, the server
+ *   certificate is verified against the list of supplied CAs. See
+ *   {@link https://nodejs.org/api/tls.html#tls_tls_connect_options_callback}.
+ * @param {Function} [options.checkServerIdentity] - A callback function to
+ *   be used when checking the server's hostname against the certificate.
+ *   See
+ *   {@link https://nodejs.org/api/tls.html#tls_tls_connect_options_callback}.
+ * @implements {BaseChannelAuth}
+ * @constructor
+ */
+function ChannelToken (base, clientid, clientsecret, scope, grantType, audience, options) {
+  BaseChannelAuth.call(this)
+  this._loginRequest = request.defaults(
+    util.addTlsOptions({
+      baseUrl: base,
+      uri: LOGIN_PATH_FRAGMENT
+    }, options)
+  )
+
+  this._clientid = clientid
+  this._clientsecret = clientsecret
+  this._scope = scope
+  this._grantType = grantType
+  this._audience = audience
+  this._token = null
+
+  /**
+   * Append the current token to the `bearer` property in the supplied
+   * `requestOptions` object.
+   * @param {Object} requestOptions - The request options.
+   * @param {BaseChannelAuth~authCallback} callback - A callback to invoke
+   *   with the modified `requestOptions`.
+   * @private
+   */
+  this._addBearerAuthToken = function (requestOptions, callback) {
+    requestOptions.auth = {bearer: this._token}
+    callback(null, requestOptions)
+  }
+}
+
+inherits(ChannelToken, BaseChannelAuth)
+
+/**
+ * Authenticate the user for an HTTP channel request. The supplied callback
+ * should be invoked with the results of the authentication attempt. See
+ * {@link BaseChannelAuth~authCallback} for more information on the
+ * content provided to the callback.
+ * @param {Object} requestOptions - Options included in the HTTP channel
+ *   request.
+ * @param {BaseChannelAuth~authCallback} callback - Callback function
+ *   invoked with the results of the authentication attempt.
+ */
+ChannelToken.prototype.authenticate = function (requestOptions, callback) {
+  var that = this
+  if (this._token) {
+    // Token was acquired previously, so use it for the request
+    this._addBearerAuthToken(requestOptions, callback)
+  } else {
+    // Token was not acquired previously, so make a login request to get
+    // a token.
+    this._loginRequest(
+      {
+        oauth: {
+          clientid: this._clientid,
+          clientsecret: this._clientsecret
+        },
+        form: {
+          'grant_type': this.__grantType,
+          'scope': this.__scope,
+          'audience': this._audience,
+        }
+      },
+      function (error, response, body) {
+        if (error) {
+          callback(new TemporaryAuthenticationError(
+            'Unexpected error: ' + error.message
+          ))
+        } else if (response.statusCode === 200) {
+          if (body.access_token) {
+            // Token was acquired successfully, so set it into the options
+            // for the original request and invoke the request callback to
+            // continue.
+            that._token = body.access_token
+            that._addBearerAuthToken(requestOptions, callback)
+          } else {
+            callback(new PermanentAuthenticationError(
+              'Unable to locate access_token in login response'
+            ))
+          }
+        } else if ([401, 403].indexOf(response.statusCode) >= 0) {
+          callback(new PermanentAuthenticationError(
+            'Unauthorized ' + response.statusCode + ': ' + body
+          ))
+        } else {
+          callback(new TemporaryAuthenticationError(
+            'Unexpected status code ' + response.statusCode + ': ' +
+            JSON.stringify(body)
+          ))
+        }
+      }
+    )
+  }
+}
+
+/**
+ * Purge any credentials cached from a previous authentication.
+ */
+ChannelToken.prototype.reset = function () {
+  this._token = null
+}
+
+module.exports = ChannelToken

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@opendxl/dxl-streaming-client",
-  "version": "0.1.1",
+  "version": "0.1.2",
   "author": "McAfee, LLC",
   "description": "OpenDXL Streaming client library",
   "license": "Apache-2.0",
@@ -59,4 +59,4 @@
       "mocha"
     ]
   }
-}
+}

sample/basic/basic-consume-client-id-example.js

@@ -0,0 +1,91 @@
+'use strict'
+
+var fs = require('fs')
+var common = require('../common')
+var client = common.require('@opendxl/dxl-streaming-client')
+var Channel = client.Channel
+var ChannelToken = client.ChannelToken
+
+// Change these below to match the appropriate details for your
+// channel connection.
+var CHANNEL_URL = 'http://127.0.0.1:50080'
+var CHANNEL_CLIENT_ID = 'me'
+var CHANNEL_CLIENT_SECRET = 'secret'
+var CHANNEL_SCOPE = ''
+var CHANNEL_GRANT_TYPE = ''
+var CHANNEL_AUDIENCE = ''
+var CHANNEL_CONSUMER_GROUP = 'sample_consumer_group'
+var CHANNEL_TOPIC_SUBSCRIPTIONS = [
+  'case-mgmt-events',
+  'my-topic',
+  'topic-abc123']
+
+// Path to a CA bundle file containing certificates of trusted CAs. The CA
+// bundle is used to validate that the certificate of the server being connected
+// to was signed by a valid authority. If set to an empty string, the server
+// certificate is not validated.
+var VERIFY_CERTIFICATE_BUNDLE = ''
+
+// This constant controls the frequency (in seconds) at which the channel 'run'
+// call below polls the streaming service for new records.
+var WAIT_BETWEEN_QUERIES = 5
+
+// Read the contents of the CA bundle file into a string if one was specified
+// above for the VERIFY_CERTIFICATE_BUNDLE constant.
+var CA_BUNDLE_TEXT =
+  VERIFY_CERTIFICATE_BUNDLE ? fs.readFileSync(
+    VERIFY_CERTIFICATE_BUNDLE) : null
+
+// Add TLS-related options to the supplied options object. This is used to
+// supply the CA bundle file content for server verification.
+var addTlsOptions = function (options) {
+  options = options || {}
+  if (CA_BUNDLE_TEXT) {
+    options.ca = CA_BUNDLE_TEXT
+    options.rejectUnauthorized = true
+  } else {
+    options.rejectUnauthorized = false
+  }
+  return options
+}
+
+// Create a new channel object
+var channel = new Channel(CHANNEL_URL,
+  addTlsOptions({
+    auth: new ChannelToken(CHANNEL_URL, CHANNEL_CLIENT_ID,
+      CHANNEL_CLIENT_SECRET,
+      CHANNEL_SCOPE,
+      CHANNEL_GRANT_TYPE,
+      CHANNEL_AUDIENCE,
+      addTlsOptions()),
+    consumerGroup: CHANNEL_CONSUMER_GROUP
+  })
+)
+
+// Consume records indefinitely
+channel.run(
+  // The function below is called back upon by the 'run' method when
+  // records are received from the channel.
+  function (payloads) {
+    // Print the payloads which were received. 'payloads' is an array of
+    // objects extracted from the records received from the channel.
+    console.log('Received payloads: ' + JSON.stringify(payloads, null, 4))
+    // Return 'True' in order for the 'run' call to continue attempting to
+    // consume records.
+    return true
+  },
+  {
+    doneCallback:
+      // The function below is called if the 'run' is stopped. If the 'run' is
+      // stopped due to an error, the 'runError' object contains a 'Error' type
+      // object.
+      function (runError) {
+        if (runError) {
+          console.log('Run error, exiting: ' + runError.message)
+          channel.destroy()
+        }
+      },
+    waitBetweenQueries: WAIT_BETWEEN_QUERIES,
+    topics: CHANNEL_TOPIC_SUBSCRIPTIONS
+  }
+)

sample/basic/basic-produce-client-id-example.js

@@ -0,0 +1,89 @@
+'use strict'
+
+var fs = require('fs')
+var common = require('../common')
+var Buffer = common.require('safe-buffer').Buffer
+var client = common.require('@opendxl/dxl-streaming-client')
+var Channel = client.Channel
+var ChannelToken = client.ChannelToken
+
+// Change these below to match the appropriate details for your
+// channel connection.
+var CHANNEL_URL = 'http://127.0.0.1:50080'
+var CHANNEL_CLIENT_ID = 'me'
+var CHANNEL_CLIENT_SECRET = 'secret'
+var CHANNEL_SCOPE = ''
+var CHANNEL_GRANT_TYPE = ''
+var CHANNEL_AUDIENCE = ''
+var CHANNEL_TOPIC = 'my-topic'
+// Path to a CA bundle file containing certificates of trusted CAs. The CA
+// bundle is used to validate that the certificate of the server being connected
+// to was signed by a valid authority. If set to an empty string, the server
+// certificate is not validated.
+var VERIFY_CERTIFICATE_BUNDLE = ''
+
+// Read the contents of the CA bundle file into a string if one was specified
+// above for the VERIFY_CERTIFICATE_BUNDLE constant.
+var CA_BUNDLE_TEXT =
+  VERIFY_CERTIFICATE_BUNDLE ? fs.readFileSync(
+    VERIFY_CERTIFICATE_BUNDLE) : null
+
+// Add TLS-related options to the supplied options object. This is used to
+// supply the CA bundle file content for server verification.
+var addTlsOptions = function (options) {
+  options = options || {}
+  if (CA_BUNDLE_TEXT) {
+    options.ca = CA_BUNDLE_TEXT
+    options.rejectUnauthorized = true
+  } else {
+    options.rejectUnauthorized = false
+  }
+  return options
+}
+
+// Create the message payload to be included in a record
+var messagePayload = {
+  message: 'Hello from OpenDXL'
+}
+
+// Create the full payload with records to produce to the channel
+var channelPayload = {
+  records: [
+    {
+      routingData: {
+        topic: CHANNEL_TOPIC,
+        shardingKey: ''
+      },
+      message: {
+        headers: {},
+        // Convert the message payload from an object to a base64-encoded
+        // string.
+        payload: Buffer.from(JSON.stringify(messagePayload)).toString('base64')
+      }
+    }
+  ]
+}
+
+// Create a new channel object
+var channel = new Channel(CHANNEL_URL,
+  addTlsOptions({
+    auth: new ChannelToken(CHANNEL_URL, CHANNEL_CLIENT_ID,
+      CHANNEL_CLIENT_SECRET,
+      CHANNEL_SCOPE,
+      CHANNEL_GRANT_TYPE,
+      CHANNEL_AUDIENCE, addTlsOptions())
+  })
+)
+
+// Produce the payload records to the channel
+channel.produce(
+  channelPayload,
+  function (error) {
+    channel.destroy()
+    if (error) {
+      console.log('Error : ' + error)
+    } else {
+      console.log('Succeeded.')
+    }
+  }
+)