import logging
import operator
from unittest import TestCase
from unittest.mock import patch
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.user_script_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
def run_user_script(self):
self.user_script_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()
self.subprocess_patch = patch('subprocess.call')
self.subprocess_mock = self.subprocess_patch.start()
def tearDown(self):
self.subprocess_patch.stop()
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-09-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],
['-I', '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 147.75.32.69/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 REJECT "
"--reject-with icmp-admin-prohibited\n"
"-A chain -d 2.2.2.2/32 -p udp --dport 53:53 -j REJECT "
"--reject-with icmp-admin-prohibited\n"
"-A chain -p icmp -j REJECT "
"--reject-with icmp-admin-prohibited\n"
"-A chain -j REJECT "
"--reject-with icmp-admin-prohibited\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 REJECT "
"--reject-with icmp6-adm-prohibited\n"
"-A chain -d 2001::2/128 -p udp --dport 53:53 -j REJECT "
"--reject-with icmp6-adm-prohibited\n"
"-A chain -p icmpv6 -j REJECT "
"--reject-with icmp6-adm-prohibited\n"
"-A chain -j REJECT "
"--reject-with icmp6-adm-prohibited\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],
['-I', '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],
['-I', '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], [
['-F', 'QBS-FORWARD'],
['-A', 'QBS-FORWARD', '!', '-i', 'vif+', '-j', 'RETURN'],
['-A', 'QBS-FORWARD', '-j', 'DROP']])
self.assertEqual(self.obj.called_commands[6], [
['-F', 'QBS-FORWARD'],
['-A', 'QBS-FORWARD', '!', '-i', 'vif+', '-j', 'RETURN'],
['-A', 'QBS-FORWARD', '-j', 'DROP']])
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][0]] +
sorted(self.obj.called_commands[4][1:], key=operator.itemgetter(1)),
[['-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][0]] +
sorted(self.obj.called_commands[6][1:], key=operator.itemgetter(1)),
[['-F', 'QBS-FORWARD'],
['-F', 'chain-ip6-1'],
['-X', 'chain-ip6-1'],
['-F', 'chain-ip6-2'],
['-X', 'chain-ip6-2']])
class TestNftablesWorker(TestCase):
def setUp(self):
super(TestNftablesWorker, self).setUp()
self.obj = NftablesWorker()
self.subprocess_patch = patch('subprocess.call')
self.subprocess_mock = self.subprocess_patch.start()
def tearDown(self):
self.subprocess_patch.stop()
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 { 147.75.32.69/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 reject with icmp type admin-prohibited\n'
' ip protocol icmp reject with icmp type admin-prohibited\n'
' reject with icmp type admin-prohibited\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 reject with icmpv6 type admin-prohibited\n'
' ip6 nexthdr icmpv6 icmpv6 type 128 reject with icmpv6 type '
'admin-prohibited\n'
' reject with icmpv6 type admin-prohibited\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'
' policy drop;\n'
' ct state established,related accept\n'
' meta iifname != "vif*" accept\n'
' }\n'
'}\n'
'table ip6 qubes-firewall {\n'
' chain forward {\n'
' type filter hook forward priority 0;\n'
' policy drop;\n'
' ct state established,related accept\n'
' meta iifname != "vif*" accept\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
self.subprocess_patch = patch('subprocess.call')
self.subprocess_mock = self.subprocess_patch.start()
def tearDown(self):
self.subprocess_patch.stop()
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'}])
@patch('os.path.isfile')
@patch('os.access')
@patch('subprocess.call')
def test_run_user_script(self, mock_subprocess, mock_os_access,
mock_os_path_isfile):
mock_os_path_isfile.return_value = False
mock_os_access.return_value = False
super(FirewallWorker, self.obj).run_user_script()
self.assertFalse(mock_subprocess.called)
mock_os_path_isfile.return_value = True
mock_os_access.return_value = False
super(FirewallWorker, self.obj).run_user_script()
self.assertFalse(mock_subprocess.called)
mock_os_path_isfile.return_value = True
mock_os_access.return_value = True
super(FirewallWorker, self.obj).run_user_script()
mock_subprocess.assert_called_once_with(
['/rw/config/qubes-firewall-user-script'])
def test_main(self):
self.obj.main()
self.assertTrue(self.obj.init_called)
self.assertTrue(self.obj.cleanup_called)
self.assertTrue(self.obj.user_script_called)
self.assertEqual(set(self.obj.rules.keys()), self.obj.list_targets())
# rules content were already tested