<filename>tests/unit/test_plugins/test_relay_eval.py

"""Test for oa.plugins.relay_eval Plugin"""


import unittest

try:
    from unittest.mock import patch, Mock, MagicMock, call
except ImportError:
    from mock import patch, Mock, MagicMock, call

import oa.plugins.relay_eval
import oa.message


class TestRelayEvalBase(unittest.TestCase):
    def setUp(self):
        unittest.TestCase.setUp(self)
        self.global_data = {}
        self.local_data = {}
        self.mock_ctxt = MagicMock()
        self.mock_msg = MagicMock(msg={}, trusted_relays=[],
                                  untrusted_relays=[])
        self.plugin = oa.plugins.relay_eval.RelayEval(self.mock_ctxt)
        self.plugin.set_local = lambda m, k, v: self.local_data.__setitem__(k,
                                                                            v)
        self.plugin.get_local = lambda m, k: self.local_data.__getitem__(k)
        self.plugin.set_global = self.global_data.__setitem__
        self.plugin.get_global = self.global_data.__getitem__
        self.mock_ruleset = MagicMock()

    def tearDown(self):
        unittest.TestCase.tearDown(self)
        patch.stopall()


class TestEvalRules(TestRelayEvalBase):
    def test_check_for_numeric_helo(self):
        self.mock_msg.untrusted_relays = [{u"helo": u"127.0.0.1"}]
        result = self.plugin.check_for_numeric_helo(self.mock_msg)
        self.assertFalse(result)

        self.mock_msg.untrusted_relays = [{u"helo": u"172.16.17.32"}]
        result = self.plugin.check_for_numeric_helo(self.mock_msg)
        self.assertTrue(result)

    def test_check_for_illegal_ip(self):
        result = self.plugin.check_for_illegal_ip(self.mock_msg)
        self.assertFalse(result)

    def test_check_all_trusted(self):
        self.mock_msg.trusted_relays = [{u"ip": u"127.0.0.1"}]
        self.mock_msg.untrusted_relays = []
        result = self.plugin.check_all_trusted(self.mock_msg)
        self.assertTrue(result)

        self.mock_msg.trusted_relays = [{u"ip": u"127.0.0.1"}]
        self.mock_msg.untrusted_relays = [{u"ip": u"172.16.17.32"}]
        result = self.plugin.check_all_trusted(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_sender_no_reverse(self):
        self.mock_msg.trusted_relays = [{u"ip": u"127.0.0.1"}]
        self.mock_msg.untrusted_relays = []
        result = self.plugin.check_for_sender_no_reverse(self.mock_msg)
        self.assertFalse(result)

        self.mock_msg.untrusted_relays = [{u"rdns": u"badrdns"}]
        result = self.plugin.check_for_sender_no_reverse(self.mock_msg)
        self.assertFalse(result)

        self.mock_msg.untrusted_relays = [{u"rdns": u"test.example.com",
                                           u"ip": u"127.0.0.1"}]
        result = self.plugin.check_for_sender_no_reverse(self.mock_msg)
        self.assertFalse(result)

        self.mock_msg.untrusted_relays = [{u"rdns": u"test.example.com",
                                           u"ip": u"172.16.17.32"}]
        result = self.plugin.check_for_sender_no_reverse(self.mock_msg)
        self.assertTrue(result)

    def test_check_for_from_domain_in_received_headers(self):
        mock_get_addr_header = Mock(return_value=["<EMAIL>"])
        self.mock_msg.attach_mock(mock_get_addr_header, 'get_addr_header')
        result = self.plugin.check_for_from_domain_in_received_headers(
            self.mock_msg, 'example.org', 'true')
        self.assertFalse(result)

        self.mock_msg.trusted_relays = [{u"rdns": u"example.com",
                                         u"by": u"example.com"}]
        result = self.plugin.check_for_from_domain_in_received_headers(
            self.mock_msg, 'example.com', 'true')
        self.assertTrue(result)

    def test_helo_ip_mismatch(self):
        self.mock_msg.untrusted_relays = [{u"helo": u"172.16.17.32",
                                           u"ip": u"192.168.127.12"}]
        result = self.plugin.helo_ip_mismatch(self.mock_msg)
        self.assertTrue(result)

        self.mock_msg.untrusted_relays = [{u"helo": u"172.16.17.32",
                                           u"ip": u"192.168.127.12"}]

        result = self.plugin.helo_ip_mismatch(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_no_rdns_dotcom_helo(self):
        self.mock_msg.untrusted_relays = [{u"ip": u"172.16.17.32", u"rdns": u"",
                                           u"helo": u"subdomain.lycos.com"}]
        result = self.plugin.check_for_no_rdns_dotcom_helo(self.mock_msg)
        self.assertTrue(result)

        self.mock_msg.untrusted_relays = [{u"ip": u"172.16.17.32", u"rdns": u"some",
                                           u"helo": u"subdomain.lycos.com"}]
        result = self.plugin.check_for_no_rdns_dotcom_helo(self.mock_msg)
        self.assertFalse(result)

    def test_hostname_to_domain(self):
        hostname = u"subdomain.example.com"
        result = self.plugin.hostname_to_domain(hostname)
        self.assertEqual(result, u"example.com")
        hostname = u"172.16.17.32"
        result = self.plugin.hostname_to_domain(hostname)
        self.assertEqual(result, hostname)

    def test_check_for_forged_received(self):
        self.mock_msg.untrusted_relays = [{u"ip": u"172.16.17.32",
                                           u"rdns": u"test.example.com",
                                           u"by": u"example.com",
                                           u"helo": u"2192.168.127.12"}]
        self.plugin._check_for_forged_received(self.mock_msg)
        mismatch_ip_helo = self.plugin.get_global("mismatch_ip_helo")
        self.assertEqual(mismatch_ip_helo, 1)

    def test_check_for_forged_received_mismatch_from(self):
        self.mock_msg.untrusted_relays = [{u"ip": u"172.16.17.32",
                                           u"rdns": u"test.example.com",
                                           u"by": u"example.com",
                                           u"helo": u"2192.168.127.12"},
                                          {u"ip": u"172.16.17.32",
                                           u"rdns": u"test.example.com",
                                           u"by": u"example.com",
                                           u"helo": u"2192.168.127.12"}]
        self.plugin._check_for_forged_received(self.mock_msg)
        mismatch_ip_helo = self.plugin.get_global("mismatch_ip_helo")
        mismatch_from = self.plugin.get_global("mismatch_from")
        self.assertEqual(mismatch_ip_helo, 2)
        self.assertEqual(mismatch_from, 1)

    def test_check_for_forged_received_trail(self):
        self.mock_msg.untrusted_relays = [{u"ip": u"172.16.17.32",
                                           u"rdns": u"test.example.com",
                                           u"by": u"example.com",
                                           u"helo": u"172.16.17.32"}
                                          for _ in range(3)]
        result = self.plugin.check_for_forged_received_trail(self.mock_msg)
        self.assertTrue(result)

    def test_check_for_forged_received_ip_helo(self):
        self.mock_msg.untrusted_relays = [{u"ip": u"172.16.17.32",
                                           u"rdns": u"test.example.com",
                                           u"by": u"example.com",
                                           u"helo": u"22.33.44.55"}]
        result = self.plugin.check_for_forged_received_ip_helo(self.mock_msg)
        self.assertTrue(result)

