sin/samba-mirror
1
0
Fork 0
mirror of https://github.com/samba-team/samba.git synced 2024-12-22 13:34:15 +03:00
Code

gp: Ensure Firewalld preforms proper cleanup

Browse Source 
Now uses gp_applier to ensure old settings are
properly cleaned up.

Signed-off-by: David Mulder <dmulder@samba.org>
Reviewed-by: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>
This commit is contained in:
David Mulder 2023-01-06 14:48:12 -07:00 committed by Andrew Bartlett
parent 7db3b63e76
commit 6ac22de749
2 changed files with 44 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
python/samba/gp/gp_firewalld_ext.py
View File

@ -16,10 +16,9 @@
import os
from subprocess import Popen, PIPE
from hashlib import blake2b
from shutil import which
import json
from samba.gp.gpclass import gp_pol_ext
from samba.gp.gpclass import gp_pol_ext, gp_applier
from samba.gp.util.logging import log
def firewall_cmd(*args):
@ -41,16 +40,19 @@ def rule_segment_parse(name, rule_segment):
return '%s %s ' % (name,
' '.join(['%s=%s' % (k, v) for k, v in rule_segment.items()]))
class gp_firewalld_ext(gp_pol_ext):
class gp_firewalld_ext(gp_pol_ext, gp_applier):
def __str__(self):
return 'Security/Firewalld'
def apply_zone(self, zone):
def apply_zone(self, guid, zone):
zone_attrs = []
ret = firewall_cmd('--permanent', '--new-zone=%s' % zone)[0]
if ret != 0:
log.error('Failed to add new zone', zone)
else:
self.gp_db.store(str(self), 'zone:%s' % zone, zone)
attribute = 'zone:%s' % zone
self.cache_add_attribute(guid, attribute, zone)
zone_attrs.append(attribute)
# Default to matching the interface(s) for the default zone
ret, out = firewall_cmd('--list-interfaces')
if ret != 0:
@ -60,8 +62,10 @@ class gp_firewalld_ext(gp_pol_ext):
'--add-interface=%s' % interface.decode())
if ret != 0:
log.error('Failed to set interfaces for zone', zone)
return zone_attrs
def apply_rules(self, rule_dict):
def apply_rules(self, guid, rule_dict):
rule_attrs = []
for zone, rules in rule_dict.items():
for rule in rules:
if 'rule' in rule:
@ -88,50 +92,60 @@ class gp_firewalld_ext(gp_pol_ext):
if ret != 0:
log.error('Failed to add firewall rule', rule_parsed)
else:
rhash = blake2b(rule_parsed.encode()).hexdigest()
self.gp_db.store(str(self), 'rule:%s:%s' % (zone, rhash),
rule_parsed)
rhash = self.generate_value_hash(rule_parsed)
attribute = 'rule:%s:%s' % (zone, rhash)
self.cache_add_attribute(guid, attribute, rule_parsed)
rule_attrs.append(attribute)
return rule_attrs
def unapply(self, guid, attribute, value):
if attribute.startswith('zone'):
ret = firewall_cmd('--permanent',
'--delete-zone=%s' % value)[0]
if ret != 0:
log.error('Failed to remove zone', value)
else:
self.cache_remove_attribute(guid, attribute)
elif attribute.startswith('rule'):
_, zone, _ = attribute.split(':')
ret = firewall_cmd('--permanent', '--zone=%s' % zone,
'--remove-rich-rule', value)[0]
if ret != 0:
log.error('Failed to remove firewall rule', value)
else:
self.cache_remove_attribute(guid, attribute)
def apply(self, applier_func, *args):
return applier_func(*args)
def process_group_policy(self, deleted_gpo_list, changed_gpo_list):
for guid, settings in deleted_gpo_list:
self.gp_db.set_guid(guid)
if str(self) in settings:
for attribute, value in settings[str(self)].items():
if attribute.startswith('zone'):
ret = firewall_cmd('--permanent',
'--delete-zone=%s' % value)[0]
if ret != 0:
log.error('Failed to remove zone', value)
else:
self.gp_db.delete(str(self), attribute)
elif attribute.startswith('rule'):
_, zone, _ = attribute.split(':')
ret = firewall_cmd('--permanent', '--zone=%s' % zone,
'--remove-rich-rule', value)[0]
if ret != 0:
log.error('Failed to remove firewall rule', value)
else:
self.gp_db.delete(str(self), attribute)
self.gp_db.commit()
self.unapply(guid, attribute, value)
for gpo in changed_gpo_list:
if gpo.file_sys_path:
section = 'Software\\Policies\\Samba\\Unix Settings\\Firewalld'
self.gp_db.set_guid(gpo.name)
pol_file = 'MACHINE/Registry.pol'
path = os.path.join(gpo.file_sys_path, pol_file)
pol_conf = self.parse(path)
if not pol_conf:
continue
attrs = []
for e in pol_conf.entries:
if e.keyname.startswith(section):
if e.keyname.endswith('Rules'):
self.apply_rules(json.loads(e.data))
attrs.extend(self.apply(self.apply_rules, gpo.name,
json.loads(e.data)))
elif e.keyname.endswith('Zones'):
if e.valuename == '**delvals.':
continue
self.apply_zone(e.data)
self.gp_db.commit()
attrs.extend(self.apply(self.apply_zone, gpo.name,
e.data))
# Cleanup all old zones and rules from this GPO
self.clean(gpo.name, keep=attrs)
def rsop(self, gpo):
output = {}

1
selftest/knownfail.d/gpo
View File

@ -1,4 +1,3 @@
^samba.tests.gpo.samba.tests.gpo.GPOTests.test_gp_firewalld_ext
^samba.tests.gpo.samba.tests.gpo.GPOTests.test_gp_firefox_ext
^samba.tests.gpo.samba.tests.gpo.GPOTests.test_gp_motd
^samba.tests.gpo.samba.tests.gpo.GPOTests.test_vgp_motd