Qubes/core-agent-linux
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

firewall: put DNS resolving into its own function

Browse Source
This commit is contained in:
3hhh 2021-05-18 17:17:28 +02:00
parent 795bec8038
commit 50e448e23a
No known key found for this signature in database
GPG Key ID: EB03A691DB2F0833
1 changed files with 22 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
qubesagent/firewall.py
View File

@ -127,6 +127,26 @@ class FirewallWorker(object):
rules.append({'action': policy})
return rules
def resolve_dns(self, fqdn, family):
"""
Resolve the given FQDN via DNS.
:param fqdn: FQDN
:param family: 4 or 6 for IPv4 or IPv6
:return: see socket.getaddrinfo()
:raises: RuleParseError
"""
try:
addrinfo = socket.getaddrinfo(fqdn, None,
(socket.AF_INET6 if family == 6 else socket.AF_INET))
except socket.gaierror as e:
raise RuleParseError('Failed to resolve {}: {}'.format(
fqdn, str(e)))
except UnicodeError as e:
raise RuleParseError('Invalid destination {}: {}'.format(
fqdn, str(e)))
return addrinfo
def update_dns_info(self, source, dns):
"""
Write resolved DNS addresses back to QubesDB. This can be useful
@ -339,12 +359,7 @@ class IptablesWorker(FirewallWorker):
elif 'dst6' in rule:
dsthosts = [rule['dst6']]
elif 'dsthost' in rule:
try:
addrinfo = socket.getaddrinfo(rule['dsthost'], None,
(socket.AF_INET6 if family == 6 else socket.AF_INET))
except socket.gaierror as e:
raise RuleParseError('Failed to resolve {}: {}'.format(
rule['dsthost'], str(e)))
addrinfo = self.resolve_dns(rule['dsthost'], family)
dsthosts = set(item[4][0] + fullmask for item in addrinfo)
ret_dns[rule['dsthost']] = dsthosts
else:
@ -652,15 +667,7 @@ class NftablesWorker(FirewallWorker):
elif 'dst6' in rule:
nft_rule += ' ip6 daddr {}'.format(rule['dst6'])
elif 'dsthost' in rule:
try:
addrinfo = socket.getaddrinfo(rule['dsthost'], None,
(socket.AF_INET6 if family == 6 else socket.AF_INET))
except socket.gaierror as e:
raise RuleParseError('Failed to resolve {}: {}'.format(
rule['dsthost'], str(e)))
except UnicodeError as e:
raise RuleParseError('Invalid destination {}: {}'.format(
rule['dsthost'], str(e)))
addrinfo = self.resolve_dns(rule['dsthost'], family)
dsthosts = set(item[4][0] + fullmask for item in addrinfo)
nft_rule += ' {} daddr {{ {} }}'.format(ip_match,
', '.join(dsthosts))