core-agent-linux/qubesagent/test_firewall.py

import logging
from unittest import TestCase

import qubesagent.firewall


class DummyIptablesRestore(object):
    # pylint: disable=too-few-public-methods
    def __init__(self, worker_mock, family):
        self._worker_mock = worker_mock
        self._family = family
        self.returncode = 0

    def communicate(self, stdin=None):
        self._worker_mock.loaded_iptables[self._family] = stdin
        return ("", None)

class DummyQubesDB(object):
    def __init__(self, worker_mock):
        self._worker_mock = worker_mock
        self.entries = {}
        self.pending_watches = []

    def read(self, key):
        try:
            return self.entries[key]
        except KeyError:
            return None

    def multiread(self, prefix):
        result = {}
        for key, value in self.entries.items():
            if key.startswith(prefix):
                result[key] = value
        return result

    def list(self, prefix):
        result = []
        for key in self.entries.keys():
            if key.startswith(prefix):
                result.append(key)
        return result

    def watch(self, path):
        pass

    def read_watch(self):
        try:
            return self.pending_watches.pop(0)
        except IndexError:
            return None


class FirewallWorker(qubesagent.firewall.FirewallWorker):
    def __init__(self):
        # pylint: disable=super-init-not-called
        # don't call super on purpose - avoid connecting to QubesDB
        # super(FirewallWorker, self).__init__()
        self.qdb = DummyQubesDB(self)
        self.log = logging.getLogger('qubes.tests')

        self.init_called = False
        self.cleanup_called = False
        self.rules = {}

    def apply_rules(self, source_addr, rules):
        self.rules[source_addr] = rules

    def cleanup(self):
        self.init_called = True

    def init(self):
        self.cleanup_called = True


class IptablesWorker(qubesagent.firewall.IptablesWorker):
    '''Override methods actually modifying system state to only log what
    would be done'''

    def __init__(self):
        # pylint: disable=super-init-not-called
        # don't call super on purpose - avoid connecting to QubesDB
        # super(IptablesWorker, self).__init__()
        # copied __init__:
        self.qdb = DummyQubesDB(self)
        self.log = logging.getLogger('qubes.tests')
        self.chains = {
            4: set(),
            6: set(),
        }

        #: instead of really running `iptables`, log what would be called
        self.called_commands = {
            4: [],
            6: [],
        }
        #: rules that would be loaded with `iptables-restore`
        self.loaded_iptables = {
            4: None,
            6: None,
        }

    def run_ipt(self, family, args, **kwargs):
        self.called_commands[family].append(args)

    def run_ipt_restore(self, family, args):
        return DummyIptablesRestore(self, family)

    @staticmethod
    def dns_addresses(family=None):
        if family == 4:
            return ['1.1.1.1', '2.2.2.2']
        else:
            return ['2001::1', '2001::2']


class NftablesWorker(qubesagent.firewall.NftablesWorker):
    '''Override methods actually modifying system state to only log what
    would be done'''

    def __init__(self):
        # pylint: disable=super-init-not-called
        # don't call super on purpose - avoid connecting to QubesDB
        # super(IptablesWorker, self).__init__()
        # copied __init__:
        self.qdb = DummyQubesDB(self)
        self.log = logging.getLogger('qubes.tests')
        self.chains = {
            4: set(),
            6: set(),
        }

        #: instead of really running `nft`, log what would be loaded
        #: rules that would be loaded with `nft`
        self.loaded_rules = []

    def run_nft(self, nft_input):
        self.loaded_rules.append(nft_input)

    @staticmethod
    def dns_addresses(family=None):
        if family == 4:
            return ['1.1.1.1', '2.2.2.2']
        else:
            return ['2001::1', '2001::2']


class TestIptablesWorker(TestCase):
    def setUp(self):
        super(TestIptablesWorker, self).setUp()
        self.obj = IptablesWorker()

    def test_000_chain_for_addr(self):
        self.assertEqual(
            self.obj.chain_for_addr('10.137.0.1'), 'qbs-10-137-0-1')
        self.assertEqual(
            self.obj.chain_for_addr('fd09:24ef:4179:0000::3'),
            'qbs-fd09-24ef-4179-0000--3')

    def test_001_create_chain(self):
        testdata = [
            (4, '10.137.0.1', 'qbs-10-137-0-1'),
            (6, 'fd09:24ef:4179:0000::3', 'qbs-fd09-24ef-4179-0000--3')
        ]
        for family, addr, chain in testdata:
            self.obj.create_chain(addr, chain, family)
            self.assertEqual(self.obj.called_commands[family],
                [['-N', chain],
                    ['-A', 'QBS-FORWARD', '-s', addr, '-j', chain]])

    def test_002_prepare_rules4(self):
        rules = [
            {'action': 'accept', 'proto': 'tcp',
                'dstports': '80-80', 'dst4': '1.2.3.0/24'},
            {'action': 'accept', 'proto': 'udp',
                'dstports': '443-1024', 'dsthost': 'yum.qubes-os.org'},
            {'action': 'accept', 'specialtarget': 'dns'},
            {'action': 'drop', 'proto': 'udp', 'specialtarget': 'dns'},
            {'action': 'drop', 'proto': 'icmp'},
            {'action': 'drop'},
        ]
        expected_iptables = (
            "*filter\n"
            "-A chain -d 1.2.3.0/24 -p tcp --dport 80:80 -j ACCEPT\n"
            "-A chain -d 82.94.215.165/32 -p udp --dport 443:1024 -j ACCEPT\n"
            "-A chain -d 1.1.1.1/32 -p tcp --dport 53:53 -j ACCEPT\n"
            "-A chain -d 2.2.2.2/32 -p tcp --dport 53:53 -j ACCEPT\n"
            "-A chain -d 1.1.1.1/32 -p udp --dport 53:53 -j ACCEPT\n"
            "-A chain -d 2.2.2.2/32 -p udp --dport 53:53 -j ACCEPT\n"
            "-A chain -d 1.1.1.1/32 -p udp --dport 53:53 -j DROP\n"
            "-A chain -d 2.2.2.2/32 -p udp --dport 53:53 -j DROP\n"
            "-A chain -p icmp -j DROP\n"
            "-A chain -j DROP\n"
            "COMMIT\n"
        )
        self.assertEqual(self.obj.prepare_rules('chain', rules, 4),
            expected_iptables)
        with self.assertRaises(qubesagent.firewall.RuleParseError):
            self.obj.prepare_rules('chain', [{'unknown': 'xxx'}], 4)
        with self.assertRaises(qubesagent.firewall.RuleParseError):
            self.obj.prepare_rules('chain', [{'dst6': 'a::b'}], 4)
        with self.assertRaises(qubesagent.firewall.RuleParseError):
            self.obj.prepare_rules('chain', [{'dst4': '3.3.3.3'}], 6)

    def test_003_prepare_rules6(self):
        rules = [
            {'action': 'accept', 'proto': 'tcp',
                'dstports': '80-80', 'dst6': 'a::b/128'},
            {'action': 'accept', 'proto': 'tcp',
                'dsthost': 'ripe.net'},
            {'action': 'accept', 'specialtarget': 'dns'},
            {'action': 'drop', 'proto': 'udp', 'specialtarget': 'dns'},
            {'action': 'drop', 'proto': 'icmp'},
            {'action': 'drop'},
        ]
        expected_iptables = (
            "*filter\n"
            "-A chain -d a::b/128 -p tcp --dport 80:80 -j ACCEPT\n"
            "-A chain -d 2001:67c:2e8:22::c100:68b/128 -p tcp -j ACCEPT\n"
            "-A chain -d 2001::1/128 -p tcp --dport 53:53 -j ACCEPT\n"
            "-A chain -d 2001::2/128 -p tcp --dport 53:53 -j ACCEPT\n"
            "-A chain -d 2001::1/128 -p udp --dport 53:53 -j ACCEPT\n"
            "-A chain -d 2001::2/128 -p udp --dport 53:53 -j ACCEPT\n"
            "-A chain -d 2001::1/128 -p udp --dport 53:53 -j DROP\n"
            "-A chain -d 2001::2/128 -p udp --dport 53:53 -j DROP\n"
            "-A chain -p icmp -j DROP\n"
            "-A chain -j DROP\n"
            "COMMIT\n"
        )
        self.assertEqual(self.obj.prepare_rules('chain', rules, 6),
            expected_iptables)

    def test_004_apply_rules4(self):
        rules = [{'action': 'accept'}]
        chain = 'qbs-10-137-0-1'
        self.obj.apply_rules('10.137.0.1', rules)
        self.assertEqual(self.obj.called_commands[4],
            [
                ['-N', chain],
                ['-A', 'QBS-FORWARD', '-s', '10.137.0.1', '-j', chain],
                ['-F', chain]])
        self.assertEqual(self.obj.loaded_iptables[4],
            self.obj.prepare_rules(chain, rules, 4))
        self.assertEqual(self.obj.called_commands[6], [])
        self.assertIsNone(self.obj.loaded_iptables[6])

    def test_005_apply_rules6(self):
        rules = [{'action': 'accept'}]
        chain = 'qbs-2000--a'
        self.obj.apply_rules('2000::a', rules)
        self.assertEqual(self.obj.called_commands[6],
            [
                ['-N', chain],
                ['-A', 'QBS-FORWARD', '-s', '2000::a', '-j', chain],
                ['-F', chain]])
        self.assertEqual(self.obj.loaded_iptables[6],
            self.obj.prepare_rules(chain, rules, 6))
        self.assertEqual(self.obj.called_commands[4], [])
        self.assertIsNone(self.obj.loaded_iptables[4])

    def test_006_init(self):
        self.obj.init()
        self.assertEqual(self.obj.called_commands[4],
            [['-nL', 'QBS-FORWARD']])
        self.assertEqual(self.obj.called_commands[6],
            [['-nL', 'QBS-FORWARD']])

    def test_007_cleanup(self):
        self.obj.init()
        self.obj.create_chain('1.2.3.4', 'chain-ip4-1', 4)
        self.obj.create_chain('1.2.3.6', 'chain-ip4-2', 4)
        self.obj.create_chain('2000::1', 'chain-ip6-1', 6)
        self.obj.create_chain('2000::2', 'chain-ip6-2', 6)
        # forget about commands called earlier
        self.obj.called_commands[4] = []
        self.obj.called_commands[6] = []
        self.obj.cleanup()
        self.assertEqual(self.obj.called_commands[4],
            [['-F', 'QBS-FORWARD'],
                ['-F', 'chain-ip4-1'],
                ['-X', 'chain-ip4-1'],
                ['-F', 'chain-ip4-2'],
                ['-X', 'chain-ip4-2']])
        self.assertEqual(self.obj.called_commands[6],
            [['-F', 'QBS-FORWARD'],
                ['-F', 'chain-ip6-2'],
                ['-X', 'chain-ip6-2'],
                ['-F', 'chain-ip6-1'],
                ['-X', 'chain-ip6-1']])


class TestNftablesWorker(TestCase):
    def setUp(self):
        super(TestNftablesWorker, self).setUp()
        self.obj = NftablesWorker()

    def test_000_chain_for_addr(self):
        self.assertEqual(
            self.obj.chain_for_addr('10.137.0.1'), 'qbs-10-137-0-1')
        self.assertEqual(
            self.obj.chain_for_addr('fd09:24ef:4179:0000::3'),
            'qbs-fd09-24ef-4179-0000--3')

    def expected_create_chain(self, family, addr, chain):
        return (
            'table {family} qubes-firewall {{\n'
            '  chain {chain} {{\n'
            '  }}\n'
            '  chain forward {{\n'
            '    {family} saddr {addr} jump {chain}\n'
            '  }}\n'
            '}}\n'.format(family=family, addr=addr, chain=chain))

    def test_001_create_chain(self):
        testdata = [
            (4, '10.137.0.1', 'qbs-10-137-0-1'),
            (6, 'fd09:24ef:4179:0000::3', 'qbs-fd09-24ef-4179-0000--3')
        ]
        for family, addr, chain in testdata:
            self.obj.create_chain(addr, chain, family)
        self.assertEqual(self.obj.loaded_rules,
            [self.expected_create_chain('ip', '10.137.0.1', 'qbs-10-137-0-1'),
             self.expected_create_chain(
                 'ip6', 'fd09:24ef:4179:0000::3', 'qbs-fd09-24ef-4179-0000--3'),
             ])

    def test_002_prepare_rules4(self):
        rules = [
            {'action': 'accept', 'proto': 'tcp',
                'dstports': '80-80', 'dst4': '1.2.3.0/24'},
            {'action': 'accept', 'proto': 'udp',
                'dstports': '443-1024', 'dsthost': 'yum.qubes-os.org'},
            {'action': 'accept', 'specialtarget': 'dns'},
            {'action': 'drop', 'proto': 'udp', 'specialtarget': 'dns'},
            {'action': 'drop', 'proto': 'icmp'},
            {'action': 'drop'},
        ]
        expected_nft = (
            'flush chain ip qubes-firewall chain\n'
            'table ip qubes-firewall {\n'
            '  chain chain {\n'
            '    ip protocol tcp ip daddr 1.2.3.0/24 tcp dport 80 accept\n'
            '    ip protocol udp ip daddr { 82.94.215.165/32 } '
            'udp dport 443-1024 accept\n'
            '    ip daddr { 1.1.1.1/32, 2.2.2.2/32 } tcp dport 53 accept\n'
            '    ip daddr { 1.1.1.1/32, 2.2.2.2/32 } udp dport 53 accept\n'
            '    ip protocol udp ip daddr { 1.1.1.1/32, 2.2.2.2/32 } udp dport '
            '53 drop\n'
            '    ip protocol icmp drop\n'
            '    drop\n'
            '  }\n'
            '}\n'
        )
        self.assertEqual(self.obj.prepare_rules('chain', rules, 4),
            expected_nft)
        with self.assertRaises(qubesagent.firewall.RuleParseError):
            self.obj.prepare_rules('chain', [{'unknown': 'xxx'}], 4)
        with self.assertRaises(qubesagent.firewall.RuleParseError):
            self.obj.prepare_rules('chain', [{'dst6': 'a::b'}], 4)
        with self.assertRaises(qubesagent.firewall.RuleParseError):
            self.obj.prepare_rules('chain', [{'dst4': '3.3.3.3'}], 6)

    def test_003_prepare_rules6(self):
        rules = [
            {'action': 'accept', 'proto': 'tcp',
                'dstports': '80-80', 'dst6': 'a::b/128'},
            {'action': 'accept', 'proto': 'tcp',
                'dsthost': 'ripe.net'},
            {'action': 'accept', 'specialtarget': 'dns'},
            {'action': 'drop', 'proto': 'udp', 'specialtarget': 'dns'},
            {'action': 'drop', 'proto': 'icmp', 'icmptype': '128'},
            {'action': 'drop'},
        ]
        expected_nft = (
            'flush chain ip6 qubes-firewall chain\n'
            'table ip6 qubes-firewall {\n'
            '  chain chain {\n'
            '    ip6 nexthdr tcp ip6 daddr a::b/128 tcp dport 80 accept\n'
            '    ip6 nexthdr tcp ip6 daddr { 2001:67c:2e8:22::c100:68b/128 } '
            'accept\n'
            '    ip6 daddr { 2001::1/128, 2001::2/128 } tcp dport 53 accept\n'
            '    ip6 daddr { 2001::1/128, 2001::2/128 } udp dport 53 accept\n'
            '    ip6 nexthdr udp ip6 daddr { 2001::1/128, 2001::2/128 } '
            'udp dport 53 drop\n'
            '    ip6 nexthdr icmpv6 icmpv6 type 128 drop\n'
            '    drop\n'
            '  }\n'
            '}\n'
        )
        self.assertEqual(self.obj.prepare_rules('chain', rules, 6),
            expected_nft)

    def test_004_apply_rules4(self):
        rules = [{'action': 'accept'}]
        chain = 'qbs-10-137-0-1'
        self.obj.apply_rules('10.137.0.1', rules)
        self.assertEqual(self.obj.loaded_rules,
            [self.expected_create_chain('ip', '10.137.0.1', chain),
             self.obj.prepare_rules(chain, rules, 4),
             ])

    def test_005_apply_rules6(self):
        rules = [{'action': 'accept'}]
        chain = 'qbs-2000--a'
        self.obj.apply_rules('2000::a', rules)
        self.assertEqual(self.obj.loaded_rules,
            [self.expected_create_chain('ip6', '2000::a', chain),
             self.obj.prepare_rules(chain, rules, 6),
             ])

    def test_006_init(self):
        self.obj.init()
        self.assertEqual(self.obj.loaded_rules,
        [
            'table ip qubes-firewall {\n'
            '  chain forward {\n'
            '    type filter hook forward priority 0;\n'
            '  }\n'
            '}\n'
            'table ip6 qubes-firewall {\n'
            '  chain forward {\n'
            '    type filter hook forward priority 0;\n'
            '  }\n'
            '}\n'
        ])

    def test_007_cleanup(self):
        self.obj.init()
        self.obj.create_chain('1.2.3.4', 'chain-ip4-1', 4)
        self.obj.create_chain('1.2.3.6', 'chain-ip4-2', 4)
        self.obj.create_chain('2000::1', 'chain-ip6-1', 6)
        self.obj.create_chain('2000::2', 'chain-ip6-2', 6)
        # forget about commands called earlier
        self.obj.loaded_rules = []
        self.obj.cleanup()
        self.assertEqual(self.obj.loaded_rules,
            ['delete table ip qubes-firewall\n'
             'delete table ip6 qubes-firewall\n',
             ])

class TestFirewallWorker(TestCase):
    def setUp(self):
        self.obj = FirewallWorker()
        rules = {
            '10.137.0.1': {
                'policy': 'accept',
                '0000': 'proto=tcp dstports=80-80 action=drop',
                '0001': 'proto=udp specialtarget=dns action=accept',
                '0002': 'proto=udp action=drop',
            },
            '10.137.0.2': {'policy': 'accept'},
            # no policy
            '10.137.0.3': {'0000': 'proto=tcp action=accept'},
            # no action
            '10.137.0.4': {
                'policy': 'drop',
                '0000': 'proto=tcp'
            },
        }
        for addr, entries in rules.items():
            for key, value in entries.items():
                self.obj.qdb.entries[
                    '/qubes-firewall/{}/{}'.format(addr, key)] = value

    def test_read_rules(self):
        expected_rules1 = [
            {'proto': 'tcp', 'dstports': '80-80', 'action': 'drop'},
            {'proto': 'udp', 'specialtarget': 'dns', 'action': 'accept'},
            {'proto': 'udp', 'action': 'drop'},
            {'action': 'accept'},
        ]
        expected_rules2 = [
            {'action': 'accept'},
        ]
        self.assertEqual(self.obj.read_rules('10.137.0.1'), expected_rules1)
        self.assertEqual(self.obj.read_rules('10.137.0.2'), expected_rules2)
        with self.assertRaises(qubesagent.firewall.RuleParseError):
            self.obj.read_rules('10.137.0.3')
        with self.assertRaises(qubesagent.firewall.RuleParseError):
            self.obj.read_rules('10.137.0.4')


    def test_list_targets(self):
        self.assertEqual(self.obj.list_targets(), set(['10.137.0.{}'.format(x)
            for x in range(1, 5)]))

    def test_is_ip6(self):
        self.assertTrue(self.obj.is_ip6('2000::abcd'))
        self.assertTrue(self.obj.is_ip6('2000:1:2:3:4:5:6:abcd'))
        self.assertFalse(self.obj.is_ip6('10.137.0.1'))

    def test_handle_addr(self):
        self.obj.handle_addr('10.137.0.2')
        self.assertEqual(self.obj.rules['10.137.0.2'], [{'action': 'accept'}])
        # fallback to block all
        self.obj.handle_addr('10.137.0.3')
        self.assertEqual(self.obj.rules['10.137.0.3'], [{'action': 'drop'}])
        self.obj.handle_addr('10.137.0.4')
        self.assertEqual(self.obj.rules['10.137.0.4'], [{'action': 'drop'}])


    def test_main(self):
        self.obj.main()
        self.assertTrue(self.obj.init_called)
        self.assertTrue(self.obj.cleanup_called)
        self.assertEqual(set(self.obj.rules.keys()), self.obj.list_targets())
        # rules content were already tested