Adding support for authentication on RabbitMQ

.github/workflows/test.yml

@@ -80,6 +80,9 @@ jobs:
  SDXLC_VERSION: '1.0.0'
  SDXLC_NAME: 'test-lc'
  MQ_HOST: 'localhost'
+ MQ_PORT: '5672'
+ MQ_USER: 'guest'
+ MQ_PASS: 'guest'
  MQ_NAME: 'hello'
  MQ_EXCHANGE: '' 
  SUB_QUEUE: 'test-queue'

README.md

@@ -36,8 +36,12 @@ $ docker run -it --rm --name rabbitmq \
  rabbitmq:latest
 ```
 
-Then in `env` and `docker-compose.yml` files, change `MQ_HOST` host to
-the corresponding IP address or hostname of the RabbitMQ server
+Then in `env` and `docker-compose.yml` files, change the following vars:
+
+- `MQ_HOST` host to the corresponding IP address or hostname of the RabbitMQ server
+- `MQ_PORT` to the corresponding port where the server will listen (default 5672)
+- `MQ_USER` to the corresponding username to authenticate (default guest)
+- `MQ_PASS` to the corresponding password to authenticate (default guest)
 
 
 ## Running SDX Local Controller with Docker Compose

docker-compose.yml

@@ -39,6 +39,8 @@ services:
  - MQ_NAME=hello
  - MQ_HOST=aw-sdx-monitor.renci.org
  - MQ_PORT=5672 
+ - MQ_USER=guest
+ - MQ_PASS=guest
  - DB_NAME=test-db
  - DB_CONFIG_TABLE_NAME=sdx-config-test-1
  - DOMAIN_CONTROLLER_URL=http://192.168.201.205:8088/SDX-LC/1.0.0/provision

env

@@ -5,6 +5,9 @@ export SDXLC_VERSION='1.0.0'
 export SDXLC_NAME='lc2'
 export MQ_NAME='hello'
 export MQ_HOST='aw-sdx-monitor.renci.org'
+export MQ_PORT=5672
+export MQ_USER='guest'
+export MQ_PASS='guest'
 export MQ_EXCHANGE=''
 export DB_NAME='test-db'
 export DB_CONFIG_TABLE_NAME='test-1'

swagger_server/messaging/rpc_queue_consumer.py

@@ -7,13 +7,15 @@
 import pika
 
 MQ_HOST = os.environ.get("MQ_HOST")
+MQ_PORT = os.environ.get("MQ_PORT")
+MQ_USER = os.environ.get("MQ_USER")
+MQ_PASS = os.environ.get("MQ_PASS")
+
 # subscribe to the corresponding queue
 SUB_QUEUE = os.environ.get("SUB_QUEUE")
 SUB_TOPIC = os.environ.get("SUB_TOPIC")
 SUB_EXCHANGE = os.environ.get("SUB_EXCHANGE")
 
-# hardcode for testing
-MQ_HOST = "aw-sdx-monitor.renci.org"
 SUB_QUEUE = "connection"
 SUB_TOPIC = "lc1_q1"
 SUB_EXCHANGE = "connection"
@@ -23,7 +25,11 @@ class RpcConsumer(object):
  def __init__(self, thread_queue, exchange_name):
  self.logger = logging.getLogger(__name__)
  self.connection = pika.BlockingConnection(
- pika.ConnectionParameters(host=MQ_HOST)
+ pika.ConnectionParameters(
+ host=MQ_HOST,
+ port=MQ_PORT,
+ credentials=pika.PlainCredentials(username=MQ_USER, password=MQ_PASS),
+ )
  )
 
  self.channel = self.connection.channel()
@@ -44,7 +50,11 @@ def on_request(self, ch, method, props, message_body):
  self._thread_queue.put(message_body)
 
  self.connection = pika.BlockingConnection(
- pika.ConnectionParameters(host=MQ_HOST)
+ pika.ConnectionParameters(
+ host=MQ_HOST,
+ port=MQ_PORT,
+ credentials=pika.PlainCredentials(username=MQ_USER, password=MQ_PASS),
+ )
  )
  self.channel = self.connection.channel()
 

swagger_server/messaging/rpc_queue_producer.py

@@ -8,13 +8,20 @@
 import pika
 
 MQ_HOST = os.environ.get("MQ_HOST")
+MQ_PORT = os.environ.get("MQ_PORT")
+MQ_USER = os.environ.get("MQ_USER")
+MQ_PASS = os.environ.get("MQ_PASS")
 
 
 class RpcProducer(object):
  def __init__(self, timeout, exchange_name, routing_key):
  self.logger = logging.getLogger(__name__)
  self.connection = pika.BlockingConnection(
- pika.ConnectionParameters(host=MQ_HOST)
+ pika.ConnectionParameters(
+ host=MQ_HOST,
+ port=MQ_PORT,
+ credentials=pika.PlainCredentials(username=MQ_USER, password=MQ_PASS),
+ )
  )
 
  self.channel = self.connection.channel()

swagger_server/messaging/topic_queue_consumer.py

@@ -11,6 +11,9 @@
 from swagger_server.utils.db_utils import DbUtils
 
 MQ_HOST = os.environ.get("MQ_HOST")
+MQ_PORT = os.environ.get("MQ_PORT")
+MQ_USER = os.environ.get("MQ_USER")
+MQ_PASS = os.environ.get("MQ_PASS")
 # subscribe to the corresponding queue
 SUB_QUEUE = os.environ.get("SUB_QUEUE")
 SUB_TOPIC = os.environ.get("SUB_TOPIC")
@@ -30,7 +33,11 @@ class TopicQueueConsumer(object):
  def __init__(self, thread_queue, exchange_name):
  self.logger = logging.getLogger(__name__)
  self.connection = pika.BlockingConnection(
- pika.ConnectionParameters(host=MQ_HOST)
+ pika.ConnectionParameters(
+ host=MQ_HOST,
+ port=MQ_PORT,
+ credentials=pika.PlainCredentials(username=MQ_USER, password=MQ_PASS),
+ )
  )
 
  self.channel = self.connection.channel()
@@ -54,7 +61,11 @@ def on_rpc_request(self, ch, method, props, message_body):
  self._thread_queue.put(message_body)
 
  self.connection = pika.BlockingConnection(
- pika.ConnectionParameters(host=MQ_HOST)
+ pika.ConnectionParameters(
+ host=MQ_HOST,
+ port=MQ_PORT,
+ credentials=pika.PlainCredentials(username=MQ_USER, password=MQ_PASS),
+ )
  )
  self.channel = self.connection.channel()
 

swagger_server/messaging/topic_queue_publisher.py

@@ -8,16 +8,20 @@
 import pika
 
 MQ_HOST = os.environ.get("MQ_HOST")
-
-# hardcode for testing
-MQ_HOST = "aw-sdx-monitor.renci.org"
+MQ_PORT = os.environ.get("MQ_PORT")
+MQ_USER = os.environ.get("MQ_USER")
+MQ_PASS = os.environ.get("MQ_PASS")
 
 
 class TopicQueueProducer(object):
  def __init__(self, timeout, exchange_name, routing_key):
  self.logger = logging.getLogger(__name__)
  self.connection = pika.BlockingConnection(
- pika.ConnectionParameters(host=MQ_HOST)
+ pika.ConnectionParameters(
+ host=MQ_HOST,
+ port=MQ_PORT,
+ credentials=pika.PlainCredentials(username=MQ_USER, password=MQ_PASS),
+ )
  )
 
  self.channel = self.connection.channel()