Tag Archives: JMS

Token-based Authentication Plugin for ActiveMQ

This post is a part of ActiveMQ Custom Security Plugins series.

 

Similarly to how we did in case of the IP-based Authentication Plugin for ActiveMQ, in order to limit the connectivity to the ActiveMQ server based on Token (assuming the connecting client, eg. a browser through a JavaScript over STOMP protocol) is providing such token when trying to establish a connection with the broker), we’ll need to override the addConnection() method of the BrokerFilter.class.

 

For the purpose of this example, i’ll be using Redis as the data store against which i’ll be checking the Tokens of connecting clients; to make a decision whether a client is allowed to establish a connection with the broker (Token exists in Redis) or not (otherwise). To hit Redis from Java i’ll be using the Jedis driver.

 

Step1: Implementation of the plugin logic:

import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerFilter;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.command.ConnectionInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import java.util.Map;

public class TokenAuthenticationBroker extends BrokerFilter {

  private final Logger logger = LoggerFactory.getLogger(getClass());
  public final static String REDIS_KEY = "authentication:activemq:tokens";

  Map<String, String> redisConfig;

  public TokenAuthenticationBroker(Broker next, Map<String, String> redisConfig) {
    super(next);
    this.redisConfig = redisConfig;
  }

  @Override
  public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
    String host = redisConfig.get("host");
    int port = Integer.parseInt(redisConfig.get("port"));

    logger.debug("Establishing Redis connection using [host={}, port={}] ", host, port);
    Jedis jedis = new Jedis(host, port);

    String token = context.getUserName();

    logger.debug("Querying Redis using [key={}, token={}] ", REDIS_KEY, token);
    String response = jedis.hget(REDIS_KEY, token);

    if(response == null) {
      throw new SecurityException("Token not not found in the data store");
    } else {
      logger.debug("Found token [{}] belonging to user: {}. Allowing connection", token, response);
    super.addConnection(context, info);
    }
  }
}

 

As you can see in the example above, the token provided by the connecting client can be read in ActiveMQ directly from the context (by using the getUserName() method; assuming the client is sending the token as a query parameter named “username”). Having the token, next thing we need to do is to query the Redis store (under the REDIS_KEY) and check whether the token exists (hget() method invoked on jedis object/driver). Depending on the value of response, we’re making the decision whether to addConnection() or throw an SecurityException.

 

Also, after the actual plug-in logic has been implemented, the plug-in must be configured and installed. For this purpose, we need an implementation of the BrokerPlugin.class, which is used to expose the configuration of a plug-in and to install the plug-in into the ActiveMQ broker.

 

Step2: Implementation of the plugin “installer”:

import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerPlugin;
import java.util.Map;

public class TokenAuthenticationPlugin implements BrokerPlugin {

  Map<String, String> redisConfig;

  @Override
  public Broker installPlugin(Broker broker) throws Exception {
    return new TokenAuthenticationBroker(broker, redisConfig);
  }

  public Map<String, String> getRedisConfig() {
    return redisConfig;
  }

  public void setRedisConfig(Map<String, String> redisConfig) {
    this.redisConfig = redisConfig;
  }
}

 

The installPlugin() method above is used to instantiate the plug-in and return a new intercepted broker for the next plug-in in the chain. The TokenAuthenticationPlugin.class also contains getter and setter methods used to configure the TokenAuthenticationBroker. These setter and getter methods are available via a Spring beans–style XML configuration in the ActiveMQ XML configuration file (example below).

 

Step3: Configuring the plugin in activemq.xml:

// "/apache-activemq/conf/activemq.xml"
<broker brokerName="localhost" dataDirectory="${activemq.base}/data" xmlns="http://activemq.apache.org/schema/core">
  <plugins>
    <bean id="tokenAuthenticationPlugin" class="com.mycompany.mysystem.activemq.TokenAuthenticationPlugin" xmlns="http://www.springframework.org/schema/beans">
      <property name="redisConfig">
        <map>
          <entry key="host" value="localhost" />
          <entry key="port" value="6379" />
        </map>
      </property>
    </bean>
  </plugins>
</broker>

 

That’s all there is to it 🙂

 

Happy Coding!

 

 

Resources:

IP-based Authentication Plugin for ActiveMQ

To limit the connectivity to the ActiveMQ server based on IP address, we’ll need to override the addConnection() method of the BrokerFilter.class, mentioned in my initial post on ActiveMQ Custom Security Plugins.

 

Example implementation (from the book “ActiveMQ in Action”):

import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerFilter;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.command.ConnectionInfo;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class IPAuthenticationBroker extends BrokerFilter {

  List<String> allowedIPAddresses;
  Pattern pattern = Pattern.compile("^/([0-9\\.]*):(.*)");

  public IPAuthenticationBroker(Broker next, List<String> allowedIPAddresses) {
    super(next);
    this.allowedIPAddresses = allowedIPAddresses;
  }

  public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
    String remoteAddress = context.getConnection().getRemoteAddress();
    Matcher matcher = pattern.matcher(remoteAddress);
    if (matcher.matches()) {
      String ip = matcher.group(1);
        if (!allowedIPAddresses.contains(ip)) {
          throw new SecurityException("Connecting from IP address " + ip + " is not allowed" );
        }
    } else {
      throw new SecurityException("Invalid remote address " + remoteAddress);
    }
    super.addConnection(context, info);
  }
}

As you can see, the implementation above performs a simple check of the IP address using a regular expression to determine the ability to connect. If that IP address is allowed to connect, the call is delegated to the BrokerFilter.addConnection() method. If that IP address isn’t allowed to connect, an exception is thrown.

 

After the actual plug-in logic has been implemented, the plug-in must be configured and installed. For this purpose, we need an implementation of the BrokerPlugin.class, which is used to expose the configuration of a plug-in and to install the plug-in into the ActiveMQ broker.

 

import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerPlugin;
import java.util.List;

public class IPAuthenticationPlugin implements BrokerPlugin {

  List<String> allowedIPAddresses;

  public Broker installPlugin(Broker broker) throws Exception {
    return new IPAuthenticationBroker(broker, allowedIPAddresses);
  }

  public List<String> getAllowedIPAddresses() {
    return allowedIPAddresses;
  }

  public void setAllowedIPAddresses(List<String> allowedIPAddresses) {
    this.allowedIPAddresses = allowedIPAddresses;
  }
}

The installPlugin() method above is used to instantiate the plug-in and return a new intercepted broker for the next plug-in in the chain. The IPAuthenticationPlugin.class also contains getter and setter methods used to configure the IPAuthenticationBroker. These setter and getter methods are available via a Spring beans–style XML configuration in the ActiveMQ XML configuration file (example below).

 

// "\apache-activemq\conf\activemq.xml"
<broker brokerName="localhost" dataDirectory="${activemq.base}/data" xmlns="http://activemq.apache.org/schema/core">
  <plugins>
    <bean id="ipAuthenticationPlugin" class="com.mycompany.mysystem.activemq.IPAuthenticationPlugin" xmlns="http://www.springframework.org/schema/beans">
      <property name="allowedIPAddresses">
        <list>
          <value>127.0.0.1</value>
        </list>
      </property>
    </bean>
  </plugins>
</broker>

To summarize, creating custom security plugins using ActiveMQ plugin API, consists of following three steps:

  1. Implementing the plugin logic (overriding methods of the BrokerFilter.class – first code snippet above)
  2. Coding the plugin “installer” (implementing the BrokerPlugin.class – second code snippet)
  3. Configuring the plugin in activemq.xml file (Spring beans-style XML – third code snippet)

 

Happy coding!

 

 

Resources:

ActiveMQ Custom Security Plugins

With this post i’m starting a short series of articles on creating custom security plugin’s for ActiveMQ server (probably the most flexible MOM/messaging solution around; imho).

 

To get a quick overview of how powerful ActiveMQ plugin API really is, let’s start with some basic background information:

  • The flexibility of ActiveMQ plugin API comes from the BrokerFilter class
  • BrokerFilter class provides the ability to intercept many of the available broker-level operations, such as:
    • adding consumers to the broker
    • adding producers to the broker
    • committing transactions in the broker
    • adding connections to the broker
    • removing connections from the broker
  • Custom functionality can be added by extending the BrokerFilter class and overriding a method for a given operation

 

Using the ActiveMQ plugins API is one way to approach broker security; used often for requirements (security, among others) that can’t be met using either:

  • ActiveMQ’s native Simple Authentication Plugin (which handles credentials directly in XML configuration file or in a properties file)
    or
  • JAAS-based pluggable security modules (JAAS stands for Java Authentication and Authorization Service). What is worth mention is that ActiveMQ comes with JAAS-based implementations of some modules that can authenticate users using properties files, LDAP, and SSL certificates; which will be enough for many use cases.

 

OK, having said the above, let’s move on and study following example implementations:

 

 

Resources:

MOM and JMS primer (ActiveMQ example)

I’ve been reading recently a very good book on Message-oriented middleware (MOM) broker server Apache ActiveMQ which is an open source implementation of the Java Message Service (JMS) spec. and makes for a reliable hub in any message-oriented enterprise application and integrates beautifully with Java EE containers, ESBs, and other JMS providers.

The book “ActiveMQ in Action” (authored by Bruce Snyder, Dejan Bosanac, and Rob Davies) starts from the anatomy of a JMS message and moves quickly through connectors, message persistence, authentication, and authorization.

Please find below few notes i made while reading:

 

Standard Message Headers:

JMSDestination

  • Header Field: JMSDestination
  • Set By: JMS provider
  • Set mode: auto
  • Meaning: where to send the message
  • Example: queue://mycompany.1.0.notifications.rs.queue

JMSDeliveryMode

  • Header Field: JMSDeliveryMode
  • Set By: JMS provider
  • Set mode: auto
  • Meaning: persistent or non-persistent*
  • Example: Persistent

JMSExpiration

  • Header Field: JMSExpiration
  • Set By: JMS provider
  • Set mode: auto
  • Meaning: when should the message expire (in ms)
  • Example: 0 (never)

JMSPriority

  • Header Field: JMSPriority
  • Set By: JMS provider
  • Set mode: auto
  • Meaning: priority 0-9 (0-4 normal, 5-9 expedited)
  • Example: 4 (standard)

JMSMessageID

  • Header Field: JMSDestination
  • Set By: JMS provider
  • Set mode: auto
  • Meaning: identifier of the message
  • Example: ID:POL-MPRZYDATEK-63606-1370265295842-1:2:1:1:1

JMSTimestamp

  • Header Field: JMSTimestamp
  • Set By: JMS provider
  • Set mode: auto
  • Meaning: when the event occurred
  • Example: 2013-06-03 15:15:27:196 CEST

JMSCorrelationID

  • Header Field: JMSCorrelationID
  • Set By: JMS Client
  • Set mode: manual
  • Meaning: link one message with another (eg. conversation-style msg. ex.)

JMSReplyTo

  • Header Field: JMSReplyTo
  • Set By: JMS Client
  • Set mode: manual
  • Meaning: where (destination) is the reply expected

JMSType

  • Header Field: JMSType
  • Set By: JMS Client
  • Set mode: manual
  • Meaning: used by few vendors to semantically identify the message type

JMSRedelivered

  • Header Field: JMSRedelivered
  • Set By: JMS provider
  • Set mode: auto
  • Meaning: if set to true, it is likely that the msg was delivered but not acknowledged
  • Example: false

* Persistent messages are intended to survive system failures of the JMS provider (the message server). Persistent messages are written to disk as soon as the message server receives them from the JMS client. After the message is persisted to disk the message server can then attempt to deliver the message to its intended consumer. As the messaging server delivers the message to the consumers it keeps track of which consumers successfully receive the message. If the JMS provider fails while delivering the message, the message server will pick up where it left off following a recovery. Persistent messages are delivered once-and-only-once. Persistent messages incur
more overhead due to the need to store the message, and value reliability over performance.
* Nonpersistent messages are not written to disk when they are received by the message server, so if the JMS provider fails, the message will be lost. In general nonpersistent messages perform better than persistent messages. They are delivered more quickly and require less system resources on the message server. However, nonpersistent messages should only be used when a loss of messages due to a JMS provider failures is not an issue.

 

Request/reply messaging pattern

  • is an asynchronous back-and-forth conversational pattern utilizing either the PTP domain or the pub/sub domain through a combination of the JMSReplyTo and JMSCorrelationID message headers and temporary destinations.
  • The JMSReplyTo specifies the destination where a reply should be sent, and the JMSCorrelationID in the reply message specifies the JMSMessageID of the request message.
  • These headers are used to link the reply message(s) to the original request message.
  • Temporary destinations are those that are created only for the duration of a connection and can only be consumed from by the connection that created them.
  • These restrictions make temporary destinations useful for request/reply.

 

Distinguishing message durability from message persistence

  • Message durability can only be achieved with the pub/sub domain. When clients connect to a topic, they can do so using a durable or a nondurable subscription
    • Durable subscription—A durable subscription is infinite. It’s registered with the topic subscription to tell the JMS provider to preserve the subscription state in the event that the subscriber disconnects. If a durable subscriber disconnects the JMS provider will hold all messages until that subscriber connects again or until the subscriber explicitly unsubscribes from the topic.
    • Nondurable subscription—A nondurable subscription is finite. It’s registered with the topic subscription to tell the JMS provider to not preserve the subscription state in the event that the subscriber disconnects. If a subscriber disconnects, the JMS provider won’t hold any messages during the disconnection period
    • Message persistence is independent of the message domain. Message persistence is a quality of service property used to indicate the JMS application’s ability to handle missing messages in the event of a JMS provider failure

 

Four types of message persistence:

  • KahaDB
    • recommended message store for general-purpose messages since ActiveMQ version 5.3 is KahaDB.
    • a file-based message store that combines a transactional journal, for reliable message storage and recovery, with good performance and scalability
    • the structure of the KahaDB store has been streamlined especially for the requirements of a message broker.
    • The KahaDB message store uses a transactional log for its indexes and only uses one index file for all its destinations.
    • It’s been used in production environments with 10,000 active connections, each connection having a separate queue
  • AMQ
    • like KahaDB, is a combination of a transactional journal for reliable persistence (to survive system crashes) and high-performance indexes, which makes this store the best option when message throughput is the main requirement for an application.
    • But because it uses two separate files for every index, and there’s an index per destination, the AMQ message store shouldn’t be used if you intend to use thousands of queues per broker.
    • Also, recovery can be slow if the ActiveMQ broker isn’t shut down cleanly. This is because all the indexes need to be rebuilt, which requires the broker to traverse all its data logs to accurately build the indexes again.
  • JDBC
    • JDBC persistence is definitely not superior in performance to KahaDB nor AMQ
    • the use of a shared database is particularly useful for making a redundant master/slave topology out of multiple brokers. When a group of ActiveMQ brokers is configured to use a shared database, they’ll all try to connect and grab a lock in the lock table, but only one will succeed and become the master. The remaining brokers will be slaves, and will be in a wait state, not accepting client connections until the master fails.
  • memory
    • memory message store holds all persistent messages in memory. No active caching is involved, so you have to be careful that both the JVM and the memory limits you set for the broker are large enough to accommodate all the messages that may exist in this message store at one time.
    • The memory message store can be useful if you know that the broker will only store a finite amount of messages, which will typically be consumed quickly

 

Caching messages in the broker for consumers

  • Although one of the most important aspects of message persistence is that the messages will survive in long-term storage, there are a number of cases where messages are required to be available for consumers that were disconnected from the broker, but persisting the messages in a data store is too slow
  • ActiveMQ supports the caching of messages for these types of systems using message caching in the broker by using something called a subscription recovery policy. This configurable policy is used for deciding which types of messages should be cached, how many, and for how long
  • subscription recovery policies:
    • fixed (memory) size subscription recovery policy
    • fixed count (messages) subscription recovery policy
    • query-based subscription recovery policy – Caches only messages that match the query
    • timed subscription recovery policy (The time in milliseconds to keep messages in the cache
    • LAST IMAGE SUBSCRIPTION RECOVERY POLICY (only last message)
    • NO SUBSCRIPTION RECOVERY POLICY (disables message caching for topics)

 

Authentication

  • Simple authentication plug-in—Handles credentials directly in the XML configuration file or in a properties file
  • JAAS authentication plug-in—Implements the JAAS API and provides a more powerful and customizable authentication solution (JAAS provides pluggable authentication, which means ActiveMQ will use the same authentication API regardless of the technique used to verify user credentials (a text file, a relational database, LDAP, and so on)
  • custom plug-in for handling security (BrokerFilter implementation)

 

Authorization

  • operation-level
    • Read—The ability to receive messages from the destination
    • Write—The ability to send messages to the destination
    • Admin—The ability to administer the destination
  • message-level
    • remote address
    • message headers

 

Hope you find this short summary useful.