7x31
/
mailcow-dockerized
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

use 'ipaddress' to manage ips in nftables ban and unban

This commit is contained in:
Vicente 2023-03-11 15:04:47 +01:00
parent 3f87df954d
commit 8034f1bc0b
1 changed files with 6 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
data/Dockerfiles/netfilter/server.py
View File

@ -421,11 +421,8 @@ def get_ban_ip_dict(ipaddr: str, _family: str):
json_command = get_base_dict() json_command = get_base_dict()
expr_opt = [] expr_opt = []
if re.search(r'/', ipaddr): ipaddr_net = ipaddress.ip_network(ipaddr)
tmp_data = re.split(r'/', ipaddr) right_dict = {'prefix': {'addr': str(ipaddr_net.network_address), 'len': int(ipaddr_net.prefixlen) } }
right_dict = {'prefix': {'addr': tmp_data[0], 'len': int(tmp_data[1]) } }
else:
right_dict = ipaddr
left_dict = {'payload': {'protocol': _family, 'field': 'saddr'} } left_dict = {'payload': {'protocol': _family, 'field': 'saddr'} }
match_dict = {'op': '==', 'left': left_dict, 'right': right_dict } match_dict = {'op': '==', 'left': left_dict, 'right': right_dict }
@ -467,24 +464,15 @@ def get_unban_ip_dict(ipaddr:str, _family: str):
# ip currently banned # ip currently banned
rule_right = rule["right"] rule_right = rule["right"]
if isinstance(rule_right, dict): if isinstance(rule_right, dict):
current_rule_ip = rule_right["prefix"]["addr"] current_rule_ip = rule_right["prefix"]["addr"] + '/' + str(rule_right["prefix"]["len"])
current_rule_len = int(rule_right["prefix"]["len"])
else: else:
current_rule_ip = rule_right current_rule_ip = rule_right
current_rule_len = 32 if _family == 'ip' else 128 current_rule_net = ipaddress.ip_network(current_rule_ip)
# ip to ban # ip to ban
if re.search(r'/', ipaddr): candidate_net = ipaddress.ip_network(ipaddr)
tmp_data = re.split(r'/', ipaddr)
candidate_ip = tmp_data[0]
candidate_len = int(tmp_data[1])
else:
candidate_ip = ipaddr
candidate_len = 32 if _family == 'ip' else 128
if all((current_rule_ip == candidate_ip, if current_rule_net == candidate_net:
current_rule_len and candidate_len,
current_rule_len == candidate_len )):
rule_handle = _object["rule"]["handle"] rule_handle = _object["rule"]["handle"]
break break