Use password authentication with RabbitMQ

.github/workflows/test.yml

@@ -79,6 +79,9 @@ jobs:
           SDX_VERSION: '1.0.0'
           SDX_NAME: 'sdx-controller-test'
           MQ_HOST: 'localhost'
+          MQ_PORT: '5672'
+          MQ_USER: 'guest'
+          MQ_PASS: 'guest'
           SUB_QUEUE: 'sdx-controller-test-queue'
           DB_NAME: 'sdx-controllder-test-db'
           DB_CONFIG_TABLE_NAME: 'sdx-controller-test-table'

env.template

@@ -7,6 +7,9 @@ export SDX_NAME="sdx-controller-test"
 
 # Message queue settings for SDX Controller.
 export MQ_HOST="aw-sdx-monitor.renci.org"
+export MQ_PORT=5672
+export MQ_USER="guest"
+export MQ_PASS="guest"
 export SUB_QUEUE="topo"
 
 # MongoDB settings for SDX Controller.

sdx_controller/messaging/message_queue_consumer.py

@@ -3,7 +3,10 @@
 
 import pika
 
-MQ_HOST = os.environ.get("MQ_HOST")
+MQ_HOST = os.getenv("MQ_HOST")
+MQ_PORT = os.getenv("MQ_PORT") or 5672
+MQ_USER = os.getenv("MQ_USER") or "guest"
+MQ_PASS = os.getenv("MQ_PASS") or "guest"
 
 
 class MessageQueue:
@@ -21,16 +24,33 @@ def __call__(cls, *args, **kwargs):
 
 
 class RabbitMqServerConfigure(metaclass=MetaClass):
-    def __init__(self, host=MQ_HOST, queue="hello"):
+    def __init__(
+        self,
+        host=MQ_HOST,
+        port=MQ_PORT,
+        username=MQ_USER,
+        password=MQ_PASS,
+        queue="hello",
+    ):
         self.host = host
+        self.port = port
+        self.username = username
+        self.password = password
         self.queue = queue
 
 
 class rabbitmqServer:
     def __init__(self, server):
         self.server = server
         self._connection = pika.BlockingConnection(
-            pika.ConnectionParameters(host=self.server.host)
+            pika.ConnectionParameters(
+                host=self.server.host,
+                port=self.server.port,
+                credentials=pika.PlainCredentials(
+                    username=self.server.username,
+                    password=self.server.password,
+                ),
+            )
         )
         self._channel = self._connection.channel()
         self._tem = self._channel.queue_declare(queue=self.server.queue)
@@ -54,7 +74,13 @@ def startserver(self):
 
 
 if __name__ == "__main__":
-    serverconfigure = RabbitMqServerConfigure(host=MQ_HOST, queue="hello")
+    serverconfigure = RabbitMqServerConfigure(
+        host=MQ_HOST,
+        port=MQ_PORT,
+        username=MQ_USER,
+        password=MQ_PASS,
+        queue="hello",
+    )
 
     server = rabbitmqServer(server=serverconfigure)
     server.startserver()

sdx_controller/messaging/rpc_queue_consumer.py

@@ -10,18 +10,27 @@
 from sdx_controller.handlers.lc_message_handler import LcMessageHandler
 from sdx_controller.utils.parse_helper import ParseHelper
 
-MQ_HOST = os.environ.get("MQ_HOST")
+MQ_HOST = os.getenv("MQ_HOST")
+MQ_PORT = os.getenv("MQ_PORT") or 5672
+MQ_USER = os.getenv("MQ_USER") or "guest"
+MQ_PASS = os.getenv("MQ_PASS") or "guest"
+
 # subscribe to the corresponding queue
-SUB_QUEUE = os.environ.get("SUB_QUEUE")
+SUB_QUEUE = os.getenv("SUB_QUEUE")
 
 logger = logging.getLogger(__name__)
 
 
 class RpcConsumer(object):
     def __init__(self, thread_queue, exchange_name, topology_manager):
         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()

sdx_controller/messaging/rpc_queue_producer.py

@@ -7,14 +7,21 @@
 
 import pika
 
-MQ_HOST = os.environ.get("MQ_HOST")
+MQ_HOST = os.getenv("MQ_HOST")
+MQ_PORT = os.getenv("MQ_PORT") or 5672
+MQ_USER = os.getenv("MQ_USER") or "guest"
+MQ_PASS = os.getenv("MQ_PASS") or "guest"
 
 
 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()

sdx_controller/messaging/topic_queue_consumer.py

@@ -6,18 +6,25 @@
 
 import pika
 
-MQ_HOST = os.environ.get("MQ_HOST")
+MQ_HOST = os.getenv("MQ_HOST")
+MQ_PORT = os.getenv("MQ_PORT") or 5672
+MQ_USER = os.getenv("MQ_USER") or "guest"
+MQ_PASS = os.getenv("MQ_PASS") or "guest"
 # 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")
+SUB_QUEUE = os.getenv("SUB_QUEUE")
+SUB_TOPIC = os.getenv("SUB_TOPIC")
+SUB_EXCHANGE = os.getenv("SUB_EXCHANGE")
 
 
 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()

sdx_controller/messaging/topic_queue_producer.py

@@ -6,7 +6,10 @@
 
 import pika
 
-MQ_HOST = os.environ.get("MQ_HOST")
+MQ_HOST = os.getenv("MQ_HOST")
+MQ_PORT = os.getenv("MQ_PORT") or 5672
+MQ_USER = os.getenv("MQ_USER") or "guest"
+MQ_PASS = os.getenv("MQ_PASS") or "guest"
 
 
 class TopicQueueProducer(object):
@@ -15,7 +18,11 @@ 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()