firewall: always use policy 'drop'
There is a problem with having separate default action ("policy") and rules because it isn't possible to set both of them atomically at the same time. To solve this problem, always have policy 'drop' (as a safe default), but by default have a single rule with action 'accept' Fixes QubesOS/qubes-issues#2869
This commit is contained in:
parent
919841635b
commit
842efb577d
@ -368,21 +368,23 @@ class Firewall(object):
|
|||||||
self.vm = vm
|
self.vm = vm
|
||||||
#: firewall rules
|
#: firewall rules
|
||||||
self.rules = []
|
self.rules = []
|
||||||
#: default action
|
|
||||||
self.policy = None
|
|
||||||
|
|
||||||
if load:
|
if load:
|
||||||
self.load()
|
self.load()
|
||||||
|
|
||||||
|
@property
|
||||||
|
def policy(self):
|
||||||
|
''' Default action - always 'drop' '''
|
||||||
|
return Action('drop')
|
||||||
|
|
||||||
def __eq__(self, other):
|
def __eq__(self, other):
|
||||||
if isinstance(other, Firewall):
|
if isinstance(other, Firewall):
|
||||||
return self.policy == other.policy and self.rules == other.rules
|
return self.rules == other.rules
|
||||||
return NotImplemented
|
return NotImplemented
|
||||||
|
|
||||||
def load_defaults(self):
|
def load_defaults(self):
|
||||||
'''Load default firewall settings'''
|
'''Load default firewall settings'''
|
||||||
self.rules = []
|
self.rules = [Rule(None, action='accept')]
|
||||||
self.policy = Action('accept')
|
|
||||||
|
|
||||||
def clone(self, other):
|
def clone(self, other):
|
||||||
'''Clone firewall settings from other instance.
|
'''Clone firewall settings from other instance.
|
||||||
@ -390,7 +392,6 @@ class Firewall(object):
|
|||||||
|
|
||||||
:param other: other :py:class:`Firewall` instance
|
:param other: other :py:class:`Firewall` instance
|
||||||
'''
|
'''
|
||||||
self.policy = other.policy
|
|
||||||
rules = []
|
rules = []
|
||||||
for rule in other.rules:
|
for rule in other.rules:
|
||||||
new_rule = Rule()
|
new_rule = Rule()
|
||||||
@ -422,9 +423,9 @@ class Firewall(object):
|
|||||||
policy_v1 = xml_root.get('policy')
|
policy_v1 = xml_root.get('policy')
|
||||||
assert policy_v1 in ('allow', 'deny')
|
assert policy_v1 in ('allow', 'deny')
|
||||||
if policy_v1 == 'allow':
|
if policy_v1 == 'allow':
|
||||||
self.policy = Action('accept')
|
policy = Action('accept')
|
||||||
else:
|
else:
|
||||||
self.policy = Action('drop')
|
policy = Action('drop')
|
||||||
|
|
||||||
def _translate_action(key):
|
def _translate_action(key):
|
||||||
if xml_root.get(key, policy_v1) == 'allow':
|
if xml_root.get(key, policy_v1) == 'allow':
|
||||||
@ -439,7 +440,7 @@ class Firewall(object):
|
|||||||
action=_translate_action('icmp'),
|
action=_translate_action('icmp'),
|
||||||
proto=Proto.icmp))
|
proto=Proto.icmp))
|
||||||
|
|
||||||
if self.policy == Action.accept:
|
if policy == Action.accept:
|
||||||
rule_action = Action.drop
|
rule_action = Action.drop
|
||||||
else:
|
else:
|
||||||
rule_action = Action.accept
|
rule_action = Action.accept
|
||||||
@ -447,11 +448,11 @@ class Firewall(object):
|
|||||||
for element in xml_root:
|
for element in xml_root:
|
||||||
rule = Rule.from_xml_v1(element, rule_action)
|
rule = Rule.from_xml_v1(element, rule_action)
|
||||||
self.rules.append(rule)
|
self.rules.append(rule)
|
||||||
|
if policy == Action.accept:
|
||||||
|
self.rules.append(Rule(None, action='accept'))
|
||||||
|
|
||||||
def load_v2(self, xml_root):
|
def load_v2(self, xml_root):
|
||||||
'''Load new (Qubes >= 4.0) firewall XML format'''
|
'''Load new (Qubes >= 4.0) firewall XML format'''
|
||||||
self.policy = Action(xml_root.findtext('policy'))
|
|
||||||
|
|
||||||
xml_rules = xml_root.find('rules')
|
xml_rules = xml_root.find('rules')
|
||||||
for xml_rule in xml_rules:
|
for xml_rule in xml_rules:
|
||||||
rule = Rule(xml_rule)
|
rule = Rule(xml_rule)
|
||||||
@ -464,10 +465,6 @@ class Firewall(object):
|
|||||||
|
|
||||||
xml_root = lxml.etree.Element('firewall', version=str(2))
|
xml_root = lxml.etree.Element('firewall', version=str(2))
|
||||||
|
|
||||||
xml_policy = lxml.etree.Element('policy')
|
|
||||||
xml_policy.text = str(self.policy)
|
|
||||||
xml_root.append(xml_policy)
|
|
||||||
|
|
||||||
xml_rules = lxml.etree.Element('rules')
|
xml_rules = lxml.etree.Element('rules')
|
||||||
for rule in self.rules:
|
for rule in self.rules:
|
||||||
if rule.expire:
|
if rule.expire:
|
||||||
|
@ -463,17 +463,17 @@ class TC_10_Firewall(qubes.tests.QubesTestCase):
|
|||||||
def test_000_defaults(self):
|
def test_000_defaults(self):
|
||||||
fw = qubes.firewall.Firewall(self.vm, False)
|
fw = qubes.firewall.Firewall(self.vm, False)
|
||||||
fw.load_defaults()
|
fw.load_defaults()
|
||||||
self.assertEqual(fw.policy, 'accept')
|
self.assertEqual(fw.policy, 'drop')
|
||||||
self.assertEqual(fw.rules, [])
|
self.assertEqual(fw.rules, [qubes.firewall.Rule(None, action='accept')])
|
||||||
|
|
||||||
def test_001_save_load_empty(self):
|
def test_001_save_load_empty(self):
|
||||||
fw = qubes.firewall.Firewall(self.vm, True)
|
fw = qubes.firewall.Firewall(self.vm, True)
|
||||||
self.assertEqual(fw.policy, 'accept')
|
self.assertEqual(fw.policy, 'drop')
|
||||||
self.assertEqual(fw.rules, [])
|
self.assertEqual(fw.rules, [qubes.firewall.Rule(None, action='accept')])
|
||||||
fw.save()
|
fw.save()
|
||||||
fw.load()
|
fw.load()
|
||||||
self.assertEqual(fw.policy, 'accept')
|
self.assertEqual(fw.policy, 'drop')
|
||||||
self.assertEqual(fw.rules, [])
|
self.assertEqual(fw.rules, [qubes.firewall.Rule(None, action='accept')])
|
||||||
|
|
||||||
def test_002_save_load_rules(self):
|
def test_002_save_load_rules(self):
|
||||||
fw = qubes.firewall.Firewall(self.vm, True)
|
fw = qubes.firewall.Firewall(self.vm, True)
|
||||||
@ -485,13 +485,13 @@ class TC_10_Firewall(qubes.tests.QubesTestCase):
|
|||||||
qubes.firewall.Rule(None, action='accept', specialtarget='dns'),
|
qubes.firewall.Rule(None, action='accept', specialtarget='dns'),
|
||||||
]
|
]
|
||||||
fw.rules.extend(rules)
|
fw.rules.extend(rules)
|
||||||
fw.policy = qubes.firewall.Action.drop
|
|
||||||
fw.save()
|
fw.save()
|
||||||
self.assertTrue(os.path.exists(os.path.join(
|
self.assertTrue(os.path.exists(os.path.join(
|
||||||
self.vm.dir_path, self.vm.firewall_conf)))
|
self.vm.dir_path, self.vm.firewall_conf)))
|
||||||
fw = qubes.firewall.Firewall(TestVM(), True)
|
fw = qubes.firewall.Firewall(TestVM(), True)
|
||||||
self.assertEqual(fw.policy, qubes.firewall.Action.drop)
|
self.assertEqual(fw.policy, qubes.firewall.Action.drop)
|
||||||
self.assertEqual(fw.rules, rules)
|
self.assertEqual(fw.rules,
|
||||||
|
[qubes.firewall.Rule(None, action='accept')] + rules)
|
||||||
|
|
||||||
def test_003_load_v1(self):
|
def test_003_load_v1(self):
|
||||||
xml_txt = """<QubesFirewallRules dns="allow" icmp="allow"
|
xml_txt = """<QubesFirewallRules dns="allow" icmp="allow"
|
||||||
@ -524,8 +524,7 @@ class TC_10_Firewall(qubes.tests.QubesTestCase):
|
|||||||
dstports=67, expire=1373300257),
|
dstports=67, expire=1373300257),
|
||||||
qubes.firewall.Rule(None, action='accept', specialtarget='dns'),
|
qubes.firewall.Rule(None, action='accept', specialtarget='dns'),
|
||||||
]
|
]
|
||||||
fw.rules.extend(rules)
|
fw.rules = rules
|
||||||
fw.policy = qubes.firewall.Action.drop
|
|
||||||
fw.save()
|
fw.save()
|
||||||
rules.pop(2)
|
rules.pop(2)
|
||||||
fw = qubes.firewall.Firewall(self.vm, True)
|
fw = qubes.firewall.Firewall(self.vm, True)
|
||||||
@ -539,8 +538,7 @@ class TC_10_Firewall(qubes.tests.QubesTestCase):
|
|||||||
qubes.firewall.Rule(None, action='accept', proto='udp'),
|
qubes.firewall.Rule(None, action='accept', proto='udp'),
|
||||||
qubes.firewall.Rule(None, action='accept', specialtarget='dns'),
|
qubes.firewall.Rule(None, action='accept', specialtarget='dns'),
|
||||||
]
|
]
|
||||||
fw.rules.extend(rules)
|
fw.rules = rules
|
||||||
fw.policy = qubes.firewall.Action.drop
|
|
||||||
expected_qdb_entries = {
|
expected_qdb_entries = {
|
||||||
'policy': 'drop',
|
'policy': 'drop',
|
||||||
'0000': 'action=drop proto=icmp',
|
'0000': 'action=drop proto=icmp',
|
||||||
|
Loading…
Reference in New Issue
Block a user