Merged in natefoo/galaxy-central (pull request #392)

SSL client certificate support for LWR AMQP connections

job_conf.xml.sample_advanced

@@ -41,6 +41,16 @@
           <!-- If multiple managers configured on the LWR, specify which one
                this plugin targets. -->
           <!-- <param id="manager">_default_</param> -->
+          <!-- The AMQP client can provide an SSL client certificate (e.g. for
+               validation), the following options configure that certificate
+               (see for reference:
+                 http://kombu.readthedocs.org/en/latest/reference/kombu.connection.html
+               ). If you simply want to use SSL but not use/validate a client
+               cert, just use the ?ssl=1 query on the amqp URL instead. -->
+          <!-- <param id="amqp_connect_ssl_ca_certs">/path/to/cacert.pem</param> -->
+          <!-- <param id="amqp_connect_ssl_keyfile">/path/to/key.pem</param> -->
+          <!-- <param id="amqp_connect_ssl_certfile">/path/to/cert.pem</param> -->
+          <!-- <param id="amqp_connect_ssl_cert_reqs">cert_required</param> -->
         </plugin>
         <plugin id="cli" type="runner" load="galaxy.jobs.runners.cli:ShellJobRunner" />
         <plugin id="condor" type="runner" load="galaxy.jobs.runners.condor:CondorJobRunner" />

lib/galaxy/jobs/runners/lwr.py

@@ -40,13 +40,17 @@ class LwrJobRunner( AsynchronousJobRunner ):
     """
     runner_name = "LWRRunner"
 
-    def __init__( self, app, nworkers, transport=None, cache=None, url=None, galaxy_url=DEFAULT_GALAXY_URL ):
+    def __init__( self, app, nworkers, transport=None, cache=None, url=None, galaxy_url=DEFAULT_GALAXY_URL, **kwds ):
         """Start the job runner """
         super( LwrJobRunner, self ).__init__( app, nworkers )
         self.async_status_updates = dict()
         self._init_monitor_thread()
         self._init_worker_threads()
-        client_manager_kwargs = {'transport_type': transport, 'cache': string_as_bool_or_none(cache), "url": url}
+        amqp_connect_ssl_args = {}
+        for kwd in kwds.keys():
+            if kwd.startswith('amqp_connect_ssl_'):
+                amqp_connect_ssl_args[kwd] = kwds[kwd]
+        client_manager_kwargs = {'transport_type': transport, 'cache': string_as_bool_or_none(cache), "url": url, 'amqp_connect_ssl_args': amqp_connect_ssl_args or None}
         self.galaxy_url = galaxy_url
         self.client_manager = build_client_manager(**client_manager_kwargs)
 

lib/galaxy/jobs/runners/lwr_client/amqp_exchange.py

@@ -30,13 +30,14 @@ class LwrExchange(object):
     name _default_.
     """
 
-    def __init__(self, url, manager_name, timeout=DEFAULT_TIMEOUT):
+    def __init__(self, url, manager_name, connect_ssl=None, timeout=DEFAULT_TIMEOUT):
         """
         """
         if not kombu:
             raise Exception(KOMBU_UNAVAILABLE)
         self.__url = url
         self.__manager_name = manager_name
+        self.__connect_ssl = connect_ssl
         self.__exchange = kombu.Exchange(DEFAULT_EXCHANGE_NAME, DEFAULT_EXCHANGE_TYPE)
         self.__timeout = timeout
 
@@ -46,7 +47,7 @@ class LwrExchange(object):
 
     def consume(self, queue_name, callback, check=True, connection_kwargs={}):
         queue = self.__queue(queue_name)
-        with self.connection(self.__url, **connection_kwargs) as connection:
+        with self.connection(self.__url, ssl=self.__connect_ssl, **connection_kwargs) as connection:
             with kombu.Consumer(connection, queues=[queue], callbacks=[callback], accept=['json']):
                 while check:
                     try:
@@ -55,7 +56,7 @@ class LwrExchange(object):
                         pass
 
     def publish(self, name, payload):
-        with self.connection(self.__url) as connection:
+        with self.connection(self.__url, ssl=self.__connect_ssl) as connection:
             with pools.producers[connection].acquire() as producer:
                 key = self.__queue_name(name)
                 producer.publish(

lib/galaxy/jobs/runners/lwr_client/manager.py

@@ -13,6 +13,7 @@ from .interface import LocalLwrInterface
 from .object_client import ObjectStoreClient
 from .transport import get_transport
 from .util import TransferEventManager
+from .util import parse_amqp_connect_ssl_params
 from .destination import url_to_destination_params
 from .amqp_exchange import LwrExchange
 
@@ -76,7 +77,8 @@ class MessageQueueClientManager(object):
     def __init__(self, **kwds):
         self.url = kwds.get('url')
         self.manager_name = kwds.get("manager", "_default_")
-        self.exchange = LwrExchange(self.url, self.manager_name)
+        self.connect_ssl = parse_amqp_connect_ssl_params(kwds.get('amqp_connect_ssl_args', None))
+        self.exchange = LwrExchange(self.url, self.manager_name, self.connect_ssl)
         self.status_cache = {}
         self.callback_lock = threading.Lock()
         self.callback_thread = None

lib/galaxy/jobs/runners/lwr_client/util.py

@@ -62,6 +62,21 @@ def directory_files(directory):
                 contents.append(name)
     return contents
 
+def parse_amqp_connect_ssl_params(params):
+    ssl = None
+    rval = None
+    ssl_options = filter(lambda x: x.startswith('amqp_connect_ssl_'), params.keys())
+    if ssl_options:
+        ssl = __import__('ssl')
+        rval = {}
+    for option in ssl_options:
+        value = params.get(option)
+        option = option.replace('amqp_connect_ssl_', '', 1)
+        if option == 'cert_reqs':
+            value = getattr(ssl, value.upper())
+        rval[option] = value
+    return rval
+
 
 def filter_destination_params(destination_params, prefix):
     destination_params = destination_params or {}