opnsense/core
1
0
Fork 0
mirror of https://github.com/opnsense/core.git synced 2025-12-11 01:32:17 -06:00
Code Issues Packages Projects Releases Wiki Activity

unbound: clean up marker and size files, closes #9489 (#9491)

Browse Source 
* unbound: clean up marker and size files, closes #9489

* unbound: ditch dnsbl_available and bind around self.dnsbl

* unbound: ditch chained comparison
This commit is contained in:
Stephan de Wit 2025-12-09 15:40:28 +01:00 committed by GitHub
parent 972433b4d0
commit efe726f841
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 29 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
src/opnsense/scripts/unbound-dnsbl/lib/dnsbl.py
View File

@ -50,7 +50,6 @@ class DNSBL:
self.size_file = size_file self.size_file = size_file
self.dnsbl_mtime_cache = 0 self.dnsbl_mtime_cache = 0
self.dnsbl_update_time = 0 self.dnsbl_update_time = 0
self.dnsbl_available = False
self.dnsbl = None self.dnsbl = None
self.warn_file = "/data/dnsbl_format_warning" self.warn_file = "/data/dnsbl_format_warning"
self._context = context self._context = context
@ -64,40 +63,42 @@ class DNSBL:
t = time.time() t = time.time()
if (t - self.dnsbl_update_time) > 60: if (t - self.dnsbl_update_time) > 60:
self.dnsbl_update_time = t self.dnsbl_update_time = t
if not self._dnsbl_exists(): self._load_dnsbl()
self.dnsbl_available = False
return def _load_dnsbl(self):
last_state = (self.dnsbl is not None)
if not self._dnsbl_exists():
self.dnsbl = None
else:
fstat = os.stat(self.dnsbl_path).st_mtime fstat = os.stat(self.dnsbl_path).st_mtime
if fstat != self.dnsbl_mtime_cache: if fstat != self.dnsbl_mtime_cache:
self.dnsbl_mtime_cache = fstat self.dnsbl_mtime_cache = fstat
log_info("dnsbl_module: updating blocklist.") log_info("dnsbl_module: updating blocklist.")
self._load_dnsbl() with open(self.dnsbl_path, 'r') as f:
try:
def _load_dnsbl(self): self.dnsbl = json.load(f)
with open(self.dnsbl_path, 'r') as f: if self._context and type(self.dnsbl.get('config')) is dict:
try: if not self.dnsbl['config'].get('general'):
self.dnsbl = json.load(f) # old format, needs blocklist reload
if self._context and type(self.dnsbl.get('config')) is dict: self.dnsbl = None
if not self.dnsbl['config'].get('general'): raise ValueError("incompatible blocklist")
# old format, needs blocklist reload self._context.set_config(self.dnsbl['config'])
raise ValueError("incompatible blocklist") log_info('dnsbl_module: blocklist loaded. length is %d' % len(self.dnsbl['data']))
self._context.set_config(self.dnsbl['config']) except (json.decoder.JSONDecodeError, KeyError, ValueError) as e:
log_info('dnsbl_module: blocklist loaded. length is %d' % len(self.dnsbl['data'])) if not self.dnsbl or isinstance(e, ValueError):
with open(self.size_file, 'w') as sfile: log_err("dnsbl_module: unable to parse blocklist file: %s. Please re-apply the blocklist settings." % e)
sfile.write(str(len(self.dnsbl['data']))) open(self.warn_file, "a").close()
except (json.decoder.JSONDecodeError, KeyError, ValueError) as e: return
if not self.dnsbl or isinstance(e, ValueError): else:
log_err("dnsbl_module: unable to parse blocklist file: %s. Please re-apply the blocklist settings." % e) log_err("dnsbl_module: error parsing blocklist: %s, reusing last known state" % e)
self.dnsbl_available = False
open(self.warn_file, "a").close()
return
else:
log_err("dnsbl_module: error parsing blocklist: %s, reusing last known list" % e)
if os.path.exists(self.warn_file): if os.path.exists(self.warn_file):
os.remove(self.warn_file) os.remove(self.warn_file)
self.dnsbl_available = True if last_state != (self.dnsbl is not None):
with open(self.size_file, 'w') as sfile:
sfile.write(str(len(self.dnsbl['data'])) if self.dnsbl else '0')
def _in_network(self, client, networks): def _in_network(self, client, networks):
if not networks: if not networks:
@ -118,7 +119,7 @@ class DNSBL:
def policy_match(self, query: Query, qstate=None, orig=None): def policy_match(self, query: Query, qstate=None, orig=None):
self._update_dnsbl() self._update_dnsbl()
if not self.dnsbl_available: if not self.dnsbl:
return False return False
if not query.type in ('A', 'AAAA', 'CNAME', 'HTTPS'): if not query.type in ('A', 'AAAA', 'CNAME', 'HTTPS'):