[GT-157] Drop messages from banned dns, and add tests for messages saving to the correct queue

ssm/agents.py

@@ -298,6 +298,10 @@ def run_receiver(protocol, brokers, project, token, cp, log, dn_file):
         dns = get_dns(dn_file, log)
         ssm.set_dns(dns)
 
+        log.info('Fetching banned DNs.')
+        banned_dns = get_banned_dns(log)
+        ssm.set_banned_dns(banned_dns)
+
     except Exception as e:
         log.fatal('Failed to initialise SSM: %s', e)
         log.info(LOG_BREAK)
@@ -379,3 +383,30 @@ def get_dns(dn_file, log):
 
     log.debug('%s DNs found.', len(dns))
     return dns
+
+
+def get_banned_dns(log):
+    """Retrieve the list of banned dns"""
+    banned_dns_list = []
+    try:
+        banned_dns_path = cp.get('auth', 'banned-dns')
+        banned_dns_file = os.path.normpath(os.path.expandvars(banned_dns_path))
+    except ConfigParser.NoOptionError:
+        banned_dns_file = None
+    f = None
+    try:
+        f = open(banned_dns_file, 'r')
+        lines = f.readlines()
+        for line in lines:
+            if line.isspace() or line.strip().startswith('#'):
+                continue
+            elif line.strip().startswith('/'):
+                banned_dns_list.append(line.strip())
+            else:
+                log.warning('DN in banned dns list is not in correct format: %s', line)
+    finally:
+        if f is not None:
+            f.close()
+
+    return banned_dns_list
+

ssm/ssm2.py

@@ -90,6 +90,7 @@ def __init__(self, hosts_and_ports, qpath, cert, key, dest=None, listen=None,
         self._dest = dest
 
         self._valid_dns = []
+        self._banned_dns = []
         self._pidfile = pidfile
 
         # Used to differentiate between STOMP and AMS methods
@@ -189,6 +190,10 @@ def set_dns(self, dn_list):
         """Set the list of DNs which are allowed to sign incoming messages."""
         self._valid_dns = dn_list
 
+    def set_banned_dns(self, banned_dn_list):
+        """Set the list of banned dns, so their messages can be dropped."""
+        self._banned_dns = banned_dn_list
+
     ##########################################################################
     # Methods called by stomppy
     ##########################################################################
@@ -283,8 +288,10 @@ def _handle_msg(self, text):
         Namely:
         - decrypt if necessary
         - verify signature
+        - send an error message if the message wasn't sent from a valid DN
         - Return plain-text message, signer's DN and an error/None.
         """
+
         if text is None or text == '':
             warning = 'Empty text passed to _handle_msg.'
             log.warning(warning)
@@ -307,10 +314,23 @@ def _handle_msg(self, text):
             log.error(error)
             return None, None, error
 
-        if signer not in self._valid_dns:
+        # If the message has been sent from a banned DN,
+        # set a specific error message that can be
+        # checked for later.
+        if signer in self._banned_dns:
+            warning = 'Signer is in the banned DNs list'
+            log.warning(warning)
+            return None, signer, warning
+
+        # Else, if the signer is not in valid DNs list,
+        # but also not a banned dn,
+        # set a specific error message
+        elif signer not in self._valid_dns:
             warning = 'Signer not in valid DNs list: %s' % signer
             log.warning(warning)
             return None, signer, warning
+
+        # Else, the message has been sent from a valid DN
         else:
             log.info('Valid signer: %s', signer)
 
@@ -320,9 +340,15 @@ def _save_msg_to_queue(self, body, empaid):
         """Extract message contents and add to the accept or reject queue."""
         extracted_msg, signer, err_msg = self._handle_msg(body)
         try:
+            # If the warning states the message was sent from a banned DN,
+            # don't send the message to the reject queue.
+            # Instead, drop the message (don't send it to any queue)
+            if err_msg == "Signer is in the banned DNs list":
+                log.info("Message dropped as was sent from a banned dn: %s", signer)
+
             # If the message is empty or the error message is not empty
             # then reject the message.
-            if extracted_msg is None or err_msg is not None:
+            elif extracted_msg is None or err_msg is not None:
                 if signer is None:  # crypto failed
                     signer = 'Not available.'
                 elif extracted_msg is not None:

test/test_ssm.py

@@ -4,11 +4,17 @@
 import shutil
 import tempfile
 import unittest
-from subprocess import call
+import dirq
+from subprocess import call, Popen, PIPE
+import logging
 
 from ssm.message_directory import MessageDirectory
 from ssm.ssm2 import Ssm2, Ssm2Exception
 
+logging.basicConfig()
+
+schema = {"body": "string", "signer": "string",
+          "empaid": "string?", "error": "string?"}
 
 class TestSsm(unittest.TestCase):
     '''
@@ -134,6 +140,225 @@ def test_ssm_init_non_dirq(self):
         # Assert the outbound queue is of the expected type.
         self.assertTrue(isinstance(ssm._outq, MessageDirectory))
 
+class TestMsgToQueue(unittest.TestCase):
+    '''
+    Class used for testing how messages are sent to queues based
+    upon the DN that sent them.
+    '''
+
+    def setUp(self):
+        # Create temporary directory for message queues and pidfiles
+        self.dir_path = tempfile.mkdtemp()
+
+        """Set up a test directory and certificates."""
+        self._tmp_dir = tempfile.mkdtemp(prefix='ssm')
+
+        # Below is not currently being used, and was an alternate method
+        # I would like to keep it until the test is working
+        """
+        # Some functions require the hardcoded expired certificate and
+        # key to be files.
+        key_fd, self._key_path = tempfile.mkstemp(prefix='key',
+                                                  dir=self._tmp_dir)
+        os.close(key_fd)
+        with open(self._key_path, 'w') as key:
+            key.write(TEST_KEY)
+
+        cert_fd, self._expired_cert_path = tempfile.mkstemp(prefix='cert',
+                                                            dir=self._tmp_dir)
+        os.close(cert_fd)
+        with open(self._expired_cert_path, 'w') as cert:
+            cert.write(TEST_CERT_FILE)
+
+        valid_dn_file, self.valid_dn_path = tempfile.mkstemp(
+            prefix='valid', dir=self._tmp_dir)
+        os.close(valid_dn_file)
+        with open(self.valid_dn_path, 'w') as dn:
+            dn.write('/test/dn')
+
+        # Create a new certificate using the hardcoded key.
+        # The subject has been hardcoded so that the generated
+        # certificate subject matches the subject of the hardcoded,
+        # expired, certificate at the bottom of this file.
+        # 2 days used so that verify_cert_date doesn't think it expires soon.
+        call(['openssl', 'req', '-x509', '-nodes', '-days', '2', '-new',
+              '-key', self._key_path, '-out', TEST_CERT_FILE,
+              '-subj', '/C=UK/O=STFC/OU=SC/CN=Test Cert'])
+        """
+
+        self._brokers = [('not.a.broker', 123)]
+        self._capath = '/not/a/path'
+        self._check_crls = False
+        self._pidfile = self._tmp_dir + '/pidfile'
+
+        self._listen = '/topic/test'
+        self._dest = '/topic/test'
+
+        self._msgdir = tempfile.mkdtemp(prefix='msgq')
+
+
+    def test_banned_dns_not_saved_to_queue(self):
+        '''
+        Test that messages sent from banned dns are dropped
+        and not sent to the accept or reject queue.
+        '''
+
+        # we need to setup an incoming and reject queue to make sure
+        # the banned messages aren't passed into either of them
+        in_q = dirq.queue.Queue(os.path.join(self._tmp_dir, 'incoming'),
+                                schema=schema)
+
+        re_q = dirq.queue.Queue(os.path.join(self._tmp_dir, 'reject'),
+                                schema=schema)
+
+        # create a list of fake valid dns that will send the messages
+        # to make sure these aren't sent to the reject queue
+        valid_dns = ("/C=UK/O=eScience/OU=CLRC/L=RAL/CN=valid-1.esc.rl.ac.uk",
+                     "/C=UK/O=eScience/OU=CLRC/L=RAL/CN=valid-2.esc.rl.ac.uk",
+                     "/C=UK/O=eScience/OU=CLRC/L=RAL/CN=valid-3.esc.rl.ac.uk")
+
+        # create a list of fake banned dns that feature in the dn list
+        # these should be dropped, and not sent to a queue
+        banned_dns = ("/C=UK/O=eScience/OU=CLRC/L=RAL/CN=banned-1.esc.rl.ac.uk",
+                      "/C=UK/O=eScience/OU=CLRC/L=RAL/CN=banned-2.esc.rl.ac.uk",
+                      "/C=UK/O=eScience/OU=CLRC/L=RAL/CN=banned-3.esc.rl.ac.uk")
+
+        # to check the valid dns aren't sent to the reject queue
+        # for each dn in the valid dn list
+        # method needs to be passed message and empaid
+        # empaid can just be 1
+        # message needs to contain the body, a signer dn from the dn list,
+        # and an error msg (none)
+        for dn in valid_dns:
+            # create a key/cert pair
+            call(['openssl', 'req', '-x509', '-nodes', '-days', '2',
+                  '-newkey', 'rsa:2048', '-keyout', TEST_KEY_FILE,
+                  '-out', TEST_CERT_FILE, '-subj', dn])
+
+            # Set up an openssl-style CA directory, containing the
+            # self-signed certificate as its own CA certificate, but with its
+            # name as <hash-of-subject-DN>.0.
+            p1 = Popen(['openssl', 'x509', '-subject_hash', '-noout'],
+                       stdin=PIPE, stdout=PIPE, stderr=PIPE,
+                       universal_newlines=True)
+
+            with open(TEST_CERT_FILE, 'r') as test_cert:
+                cert_string = test_cert.read()
+
+            hash_name, _unused_error = p1.communicate(cert_string)
+
+            self.ca_certpath = os.path.join(TEST_CA_DIR, hash_name.strip() + '.0')
+            with open(self.ca_certpath, 'w') as ca_cert:
+                ca_cert.write(cert_string)
+
+            empaid = "1"
+            message_valid = {"body": """APEL-summary-job-message: v0.2
+                    Site: RAL-LCG2
+                    Month: 3
+                    Year: 2010
+                    GlobalUserName: """ + dn + """
+                    VO: atlas
+                    VOGroup: /atlas
+                    VORole: Role=production
+                    WallDuration: 234256
+                    CpuDuration: 2345
+                    NumberOfJobs: 100
+                    %%
+                    Site: RAL-LCG2
+                    Month: 4
+                    Year: 2010
+                    GlobalUserName: """ + dn + """
+                    VO: atlas
+                    VOGroup: /atlas
+                    VORole: Role=production
+                    WallDuration: 234256
+                    CpuDuration: 2345
+                    NumberOfJobs: 100
+                    %%""",
+                  "signer": dn, "empaid": empaid, "error": ""}
+
+            ssm = Ssm2(self._brokers, self._msgdir, TEST_CERT_FILE,
+                        TEST_KEY_FILE, dest=self._dest, listen=self._listen,
+                        capath=self.ca_certpath)
+            ssm._save_msg_to_queue(message_valid["body"], empaid)
+
+            # check the valid message hasn't been sent to the reject queue
+            self.assertEquals(re_q.count(), 0)
+
+            # check the valid message reached the incoming queue
+            self.assertEquals(in_q.count(), 1)
+
+
+        # to check the banned dns aren't sent to a queue
+        # for each dn in the banned dn list
+        # method needs to be passed message and empaid
+        # empaid can just be 1
+        # message needs to contain the body, a signer dn from the dn list,
+        # and an error msg (Signer is in the banned DNs list)
+        for dn in banned_dns:
+            # create a key/cert pair
+            call(['openssl', 'req', '-x509', '-nodes', '-days', '2',
+                  '-newkey', 'rsa:2048', '-keyout', TEST_KEY_FILE,
+                  '-out', TEST_CERT_FILE, '-subj', dn])
+
+            # Set up an openssl-style CA directory, containing the
+            # self-signed certificate as its own CA certificate, but with its
+            # name as <hash-of-subject-DN>.0.
+            p1 = Popen(['openssl', 'x509', '-subject_hash', '-noout'],
+                       stdin=PIPE, stdout=PIPE, stderr=PIPE,
+                       universal_newlines=True)
+
+            with open(TEST_CERT_FILE, 'r') as test_cert:
+                cert_string = test_cert.read()
+
+            hash_name, _unused_error = p1.communicate(cert_string)
+
+            self.ca_certpath = os.path.join(TEST_CA_DIR, hash_name.strip() + '.0')
+            with open(self.ca_certpath, 'w') as ca_cert:
+                ca_cert.write(cert_string)
+
+
+            empaid = "1"
+            message_banned = {"body": """APEL-summary-job-message: v0.2
+                    Site: RAL-LCG2
+                    Month: 3
+                    Year: 2010
+                    GlobalUserName: """ + dn + """
+                    VO: atlas
+                    VOGroup: /atlas
+                    VORole: Role=production
+                    WallDuration: 234256
+                    CpuDuration: 2345
+                    NumberOfJobs: 100
+                    %%
+                    Site: RAL-LCG2
+                    Month: 4
+                    Year: 2010
+                    GlobalUserName: """ + dn + """
+                    VO: atlas
+                    VOGroup: /atlas
+                    VORole: Role=production
+                    WallDuration: 234256
+                    CpuDuration: 2345
+                    NumberOfJobs: 100
+                    %%""",
+                  "signer": dn, "empaid": empaid, "error": "Signer is in the banned DNs list"}
+
+            ssm = Ssm2(self._brokers, self._msgdir, TEST_CERT_FILE,
+                        TEST_KEY_FILE, dest=self._dest, listen=self._listen,
+                        capath=self.ca_certpath)
+            ssm._save_msg_to_queue(message_banned["body"], empaid)
+
+            # check the banned message hasn't been sent to the reject queue
+            self.assertEquals(re_q.count(), 0)
+
+            # check the banned message hasn't been sent to the incoming queue
+            self.assertEquals(in_q.count(), 0)
+
+
+TEST_KEY_FILE = '/tmp/test.key'
+
+TEST_CA_DIR='/tmp'
 
 TEST_CERT_FILE = '/tmp/test.crt'