Commit 5eaeb2d9 authored by Leonard Marschke's avatar Leonard Marschke 👾
Browse files

Merge branch 'more_fuzz_tests' into 'master'

More fuzz tests

See merge request sre18/break-it!15
parents a9ac000c 2f60684a
Loading
Loading
Loading
Loading
Loading
+13 −1
Original line number Diff line number Diff line
@@ -35,5 +35,17 @@ class TestFuzzing(BaseTest):
    def test_unrecognized_command(self):
        self.assertResponse(500, 'FAIL\r\n')

    def test_small_input(self):
        fuzzy_input = 'A' * 10**1
        self.assertResponse([x for x in range(500, 510)], '{}\r\n'.format(fuzzy_input))

    def test_big_input(self):
        fuzzy_input = 'A' * 10**3
        self.assertResponse([x for x in range(500, 510)], '{}\r\n'.format(fuzzy_input))

    def test_huge_input(self):
        fuzzy_input = 'A' * 10**5
        self.assertResponse([x for x in range(500, 510)], '{}\r\n'.format(fuzzy_input))

    def test_sentry_issue_49(self):
        self.assertResponse(500, b'\xff\xf4\xff\xfd\x06' + "\r\n".encode('ASCII'))
        self.assertResponse([x for x in range(500, 510)], b'\xff\xf4\xff\xfd\x06' + "\r\n".encode('ASCII'))
+71 −0
Original line number Diff line number Diff line
@@ -103,3 +103,74 @@ class TestMail(BaseTest):
                                 'SOME_RANDOM_TEXT'])

        self.assertResponse(221, 'QUIT\r\n')

    def test_mail_many_recipients(self):
        self.assertResponse(250, 'HELO marschke.me\r\n')
        self.assertResponse(250, 'MAIL FROM:<leonard@marschke.me>\r\n')
        self.assertResponse(250, 'RCPT TO:<firstAddress@{}>\r\n'.format(self.test_server.domain))

        for i in range(10**3):
            self.assertResponse(250, 'RCPT TO:<address{}@{}>\r\n'.format(i, self.test_server.domain))

        self.assertResponse(354, 'DATA\r\n' + \
                                 'Date: Thu, 10 May 2018 10:33:29 -0100\r\n' + \
                                 'FROM: Leonard Marschke<leonard@marschke.me>\r\n' + \
                                 'Subject: Testmail\r\n' + \
                                 'To: firstaddress@{}\r\n'.format(self.test_server.domain) + \
                                 '\r\n' + \
                                 'Hi all, this is\r\n' + \
                                 'a second line.\r\n' + \
                                 'And a third.\r\n' + \
                                 'Single point incoming.\r\n' + \
                                 '..\r\n' + \
                                 'BR\r\n')
        self.assertResponse(250, '.\r\n')

        self.assertResponse(221, 'QUIT\r\n')

    def test_mail_many_mails(self):
        for _ in range(10**3):
            self.assertResponse(250, 'HELO marschke.me\r\n')
            self.assertResponse(250, 'MAIL FROM:<leonard@marschke.me>\r\n')
            self.assertResponse(250, 'RCPT TO:<firstAddress@{}>\r\n'.format(self.test_server.domain))
            self.assertResponse(250, 'RCPT TO:<secondAddress@{}>\r\n'.format(self.test_server.domain))
            self.assertResponse(354, 'DATA\r\n' + \
                                     'Date: Thu, 10 May 2018 10:33:29 -0100\r\n' + \
                                     'FROM: Leonard Marschke<leonard@marschke.me>\r\n' + \
                                     'Subject: Testmail\r\n' + \
                                     'To: firstaddress@{}\r\n'.format(self.test_server.domain) + \
                                     '\r\n' + \
                                     'Hi all, this is\r\n' + \
                                     'a second line.\r\n' + \
                                     'And a third.\r\n' + \
                                     'Single point incoming.\r\n' + \
                                     '..\r\n' + \
                                     'BR\r\n')
            self.assertResponse(250, '.\r\n')

        self.assertResponse(221, 'QUIT\r\n')

    def test_mail_huge_mails(self):
        self.assertResponse(250, 'HELO marschke.me\r\n')
        self.assertResponse(250, 'MAIL FROM:<leonard@marschke.me>\r\n')
        self.assertResponse(250, 'RCPT TO:<firstAddress@{}>\r\n'.format(self.test_server.domain))
        self.assertResponse(250, 'RCPT TO:<secondAddress@{}>\r\n'.format(self.test_server.domain))

        huge_lines = ''
        for i in range(10**3):
            huge_line = 'Line {} {}\r\n'.format(str(i), 'A' * 10**2)
            huge_lines += huge_line

        self.assertResponse(354, 'DATA\r\n' + \
                                 'Date: Thu, 10 May 2018 10:33:29 -0100\r\n' + \
                                 'FROM: Leonard Marschke<leonard@marschke.me>\r\n' + \
                                 'Subject: Testmail\r\n' + \
                                 'To: firstaddress@{}\r\n'.format(self.test_server.domain) + \
                                 '\r\n' + \
                                 '{}\r\n'.format(huge_lines) + \
                                 '..\r\n' + \
                                 'BR\r\n')

        self.assertResponse(250, '.\r\n')

        self.assertResponse(221, 'QUIT\r\n')
+0 −2
Original line number Diff line number Diff line
@@ -87,8 +87,6 @@ class TestSession(BaseTest):
        self.assertResponse(500, b'VRFZ\r\n')
        self.assertResponse(250, b'NOOP\r\n')

    def test_quit_operation(self):
        pass
    def test_parse_helo_correct(self):
        self.assertResponse(250, b'HELO marschke.me\r\n')

+14 −11
Original line number Diff line number Diff line
@@ -35,32 +35,35 @@ class BaseTest(unittest.TestCase):
        raise self.failureException("{} not in {}: {}".format(value, expected_list, msg))

    def assertResponse(self, expected_code, message=None):  # pylint: disable=invalid-name
        def format_message(message):
            if not message or isinstance(message, bytes):
                return message
            if message[-2:] == '\r\n':
                return message[:-2]
            return message
        def format_message(form_message):
            if not form_message or isinstance(form_message, bytes):
                return form_message
            if form_message[-2:] == '\r\n':
                form_message = form_message[:-2]
            if len(form_message) > 100:
                form_message = '{} ..[{}x].. {}'.format(form_message[:50], len(form_message)-100, form_message[-50:])

            return form_message

        if isinstance(expected_code, int):
            expected_code = [expected_code]

        if message:
            if isinstance(message, str):
                message = message.encode('ASCII')
                self.client.send(message.encode('ASCII'))
            else:
                self.client.send(message)
            message = format_message(message)

        response = ''
        while not response.endswith('\r\n'):
            try:
                byte = self.client.recv(1)
            except socket.timeout:
                raise self.failureException("Timeout with message <{}>".format(message))
                raise self.failureException("Timeout with message <{}>".format(format_message(message)))
            if byte == b'':
                raise self.failureException("No code raised, but expected {}, message <{}>".format(
                    expected_code, message))
                    expected_code, format_message(message)))
            response += byte.decode('ASCII')
        self.assertValueListEqual(int(response[:3]), expected_code, msg="Message <{}>".format(message))
        self.assertValueListEqual(int(response[:3]), expected_code, msg="Message <{}>".format(format_message(message)))
        if response[3] == '-':
            self.assertResponse(expected_code)
+2 −2
Original line number Diff line number Diff line
@@ -6,7 +6,7 @@ import time


class TestServer:
    CONNECTION_TIMEOUT = 5
    CONNECTION_TIMEOUT = 20

    def __init__(self, hostname, port, container_template, docker_port, ssl_on=False, wait_up=0.5):
        self.hostname = hostname
@@ -37,7 +37,7 @@ class TestServer:
        if not self.docker_id:
            raise AssertionError('No docker ID set!')

        subprocess.run(['docker', 'kill', self.docker_id], stdout=subprocess.PIPE)
        subprocess.run(['docker', 'kill', self.docker_id], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        self.docker_id = None

    def cleanup(self):