rezometz/firewall
9
0
Fork 0
Code Issues Pull requests 1 Releases Wiki Activity

NAT fonctionnel

Browse source
This commit is contained in:
Hugo Levy-Falk 2019-03-26 22:02:43 +01:00 committed by root
parent ec80954927
commit 888ceb8d20
7 changed files with 395 additions and 432 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

603
firewall.py
View file

@ -96,6 +96,8 @@ class Parser:
ip: can either be a tuple (in this case returns an IPRange), a
single IP address or a IP Network.
"""
if type(ip) in (netaddr.IPAddress, netaddr.IPNetwork, netaddr.IPRange, netaddr.IPGlob):
return ip
try:
return netaddr.IPAddress(ip, version=4)
except netaddr.core.AddrFormatError:
@ -155,6 +157,19 @@ class NetfilterSet:
NFT_TYPE = {'set', 'map'}
# A.K.A. Really, I don't hate you, so please don't hate me...
pattern = re.compile(
r"table (?P<address_family>\w+)+ (?P<table>\w+) \{\n"
r"\s*set (?P<name>\w+) \{\n"
r"\s*type (?P<type>(\w+( \. )?)+)\n"
r"(\s*flags (?P<flags>(\w+(, )?)+)\n)?"
r"(\s*elements = \{ "
r"(?P<elements>((\n?\s*)?([\w:\.-/]+( \. )?)+,?)*) "
r"\n?\s*\}\n)?"
r"\s*\}\n"
r"\s*\}"
)
def __init__(self,
name,
type_, # e.g.: ('MAC', 'IPv4')
@ -163,18 +178,11 @@ class NetfilterSet:
address_family='inet', # Manage both IPv4 and IPv6.
table_name='filter',
flags = [],
type_from=None
):
self.name = name
self.content = set()
# self.type
self.set_type(type_)
if type_from:
self.set_type_from(type_from)
self.nft_type = 'map'
self.key_filters = tuple(self.FILTERS[i] for i in self.type_from)
else:
self.nft_type = 'set'
self.filters = tuple(self.FILTERS[i] for i in self.type)
self.set_flags(flags)
# self.address_family
@ -182,10 +190,8 @@ class NetfilterSet:
self.table = table_name
sudo = ["/usr/bin/sudo"] * int(bool(use_sudo))
self.nft = [*sudo, "/usr/sbin/nft"]
if target_content and self.nft_type == 'set':
self._target_content = self.validate_set_data(target_content)
elif target_content and self.nft_type == 'map':
self._target_content = self.validate_map_data(target_content)
if target_content:
self._target_content = self.validate_data(target_content)
else:
self._target_content = set()
@ -195,14 +201,11 @@ class NetfilterSet:
@target_content.setter
def target_content(self, target_content):
self._target_content = self.validate_set_data(target_content)
self._target_content = self.validate_data(target_content)
def filter(self, elements):
return (self.filters[i](element) for i, element in enumerate(elements))
def filter_key(self, elements):
return (self.key_filters[i](element) for i, element in enumerate(elements))
def set_type(self, type_):
"""Check set type validity and store it along with a type checker."""
for element_type in type_:
@ -210,13 +213,6 @@ class NetfilterSet:
raise ValueError('Invalid type: "{}".'.format(element_type))
self.type = type_
def set_type_from(self, type_):
"""Check set type validity and store it along with a type checker."""
for element_type in type_:
if element_type not in self.TYPES:
raise ValueError('Invalid type: "{}".'.format(element_type))
self.type_from = type_
def set_address_family(self, address_family='ip'):
"""Set set addres_family, defaulting to "ip" like nftables."""
if address_family not in self.ADDRESS_FAMILIES:
@ -229,12 +225,12 @@ class NetfilterSet:
for f in flags_:
if f not in self.FLAGS:
raise ValueError('Invalid flag: "{}".'.format(f))
self.flags = set(flags_)
self.flags = set(flags_) or None
def create_in_kernel(self):
"""Create the set, removing existing set if needed."""
# Delete set if it exists with wrong type
current_set = self._get_raw_netfilter_set(parse_elements=False)
current_set = self._get_raw_netfilter(parse_elements=False)
logging.info(current_set)
if current_set is None:
self._create_new_set_in_kernel()
@ -242,21 +238,21 @@ class NetfilterSet:
self._delete_in_kernel()
self._create_new_set_in_kernel()
def _delete_in_kernel(self):
def _delete_in_kernel(self, nft_type='set'):
"""Delete the set, table and set must exist."""
CommandExec.run([
*self.nft,
'delete {nft_type} {addr_family} {table} {set_}'.format(
nft_type=self.nft_type,
addr_family=self.address_family, table=self.table,
nft_type=nft_type,
set_=self.name)
])
def _create_new_set_in_kernel(self):
def _create_new_set_in_kernel(self, nft_type='set'):
"""Create the non-existing set, creating table if needed."""
if self.flags:
nft_command = 'add {nft_type} {addr_family} {table} {set_} {{ type {type_} ; flags {flags};}}'.format(
nft_type=self.nft_type,
nft_type=nft_type,
addr_family=self.address_family,
table=self.table,
set_=self.name,
@ -265,7 +261,7 @@ class NetfilterSet:
)
else:
nft_command = 'add {nft_type} {addr_family} {table} {set_} {{ type {type_} ;}}'.format(
nft_type=self.nft_type,
nft_type=nft_type,
addr_family=self.address_family,
table=self.table,
set_=self.name,
@ -285,7 +281,7 @@ class NetfilterSet:
CommandExec.run(create_table)
CommandExec.run(create_set)
def validate_set_data(self, set_data):
def validate_data(self, set_data):
"""
Validate data, returning it or raising a ValueError.
@ -304,7 +300,170 @@ class NetfilterSet:
.format(len(errors), '",\n"'.join(map(str, errors))))
return set_
def validate_map_data(self, dict_data):
def _apply_target_content(self):
"""Change netfilter content to target set."""
current_set = self.get_netfilter_content()
if current_set is None:
raise ValueError('Cannot change "{}" netfilter set content: set '
'do not exist in "{}" "{}".'.format(
self.name, self.address_family, self.table))
to_delete = current_set - self._target_content
to_add = self._target_content - current_set
self._change_content(delete=to_delete, add=to_add)
def _change_content(self, delete=None, add=None):
todo = [tuple_ for tuple_ in (('add', add), ('delete', delete))
if tuple_[1]]
for action, elements in todo:
content = ', '.join(' . '.join(str(element) for element in tuple_)
for tuple_ in elements)
command = [
*self.nft,
'{action} element {addr_family} {table} {set_} {{{content}}}' \
.format(action=action, addr_family=self.address_family,
table=self.table, set_=self.name, content=content)
]
CommandExec.run(command)
def _get_raw_netfilter(self, parse_elements=True):
"""Return a dict describing the netfilter set matching self or None."""
_, stdout, _ = CommandExec.run_check_output(
[*self.nft, '-nn', 'list set {addr_family} {table} {set_}'.format(
addr_family=self.address_family, table=self.table,
set_=self.name)],
allowed_return_codes=(0, 1) # In case table do not exist
)
if not stdout:
return None
else:
netfilter_set = self._parse_netfilter_string(stdout)
if netfilter_set['name'] != self.name \
or netfilter_set['address_family'] != self.address_family \
or netfilter_set['table'] != self.table \
or not self.has_type(netfilter_set['type']) \
or netfilter_set.get('flags', set()) != self.flags:
raise ValueError(
'Did not get the right set, too wrong to fix. Got '
+ str(netfilter_set)
+ ("\nExpected : "
"\n\tname: {name}"
"\n\taddress_family: {family}"
"\n\ttable: {table}"
"\n\tflags: {flags}"
"\n\ttypes: {types}"
).format(
name=self.name,
family=self.address_family,
table=self.table,
flags=self.flags,
types=tuple(self.TYPES[t] for t in self.type)
)
)
if parse_elements:
if netfilter_set['raw_content']:
netfilter_set['content'] = self.validate_data((
(element.strip() for element in n_uplet.split(' . '))
for n_uplet in netfilter_set['raw_content'].split(',')))
else:
netfilter_set['content'] = set()
return netfilter_set
@classmethod
def _parse_netfilter_string(cls, set_string):
"""
Parse netfilter set definition and return set as dict.
Do not validate content type against detected set type.
Return a dict with 'name', 'address_family', 'table', 'type', 'flags',
'raw_content' keys (all strings, 'raw_content' can be None).
Raise ValueError in case of unexpected syntax.
"""
try:
values = cls.pattern.match(set_string).groupdict()
except Exception as e:
raise ValueError("Malformed expression :\n" + set_string)
return {
'address_family': values['address_family'],
'table': values['table'],
'name': values['name'],
'type': values['type'].split(' . '),
'raw_content': values['elements'],
'flags': values['flags'],
}
def get_netfilter_content(self):
"""Return current set content from netfilter."""
netfilter_set = self._get_raw_netfilter(parse_elements=True)
if netfilter_set is None:
return None
else:
return netfilter_set['content']
def has_type(self, type_):
"""Check if some type match the set's one."""
return tuple(self.TYPES[t] for t in self.type) == tuple(type_)
def manage(self):
"""Create set if needed and populate it with target content."""
self.create_in_kernel()
self._apply_target_content()
def format_type(self):
return ' . '.join(self.TYPES[i] for i in self.type)
class NetfilterMap(NetfilterSet):
# A.K.A. Again, I don't hate you, so please don't hate me...
pattern = re.compile(
r"table (?P<address_family>\w+)+ (?P<table>\w+) \{\n"
r"\s*map (?P<name>\w+) \{\n"
r"\s*type (?P<type_from>(\w+( \. )?)+) : (?P<type>\w+)\n"
r"(\s*flags (?P<flags>(\w+(, )?)+)\n)?"
r"(\s*elements = \{ "
r"(?P<elements>(\n?\s*([\w:\.-/]+( \. )?)+ : [\w:\.-/]+,?)*)"
r"\n?\s*\}\n)?"
r"\s*\}"
r"\n\s*\}"
)
def __init__(self,
name,
type_,
type_from,
target_content=None,
use_sudo=True,
address_family='inet',
table_name='filter',
flags=[]
):
super().__init__(name, type_, use_sudo=use_sudo,
address_family=address_family, table_name=table_name,
flags=flags)
self.set_type_from(type_from)
self.key_filters = tuple(self.FILTERS[i] for i in self.type_from)
if target_content:
self._target_content = self.validate_data(target_content)
else:
self._target_content = {}
def filter_key(self, elements):
return (self.key_filters[i](element) for i, element in enumerate(elements))
def set_type_from(self, type_):
"""Check set type validity and store it along with a type checker."""
for element_type in type_:
if element_type not in self.TYPES:
raise ValueError('Invalid type: "{}".'.format(element_type))
self.type_from = type_
def _delete_in_kernel(self):
"""Delete the map, table and map must exist."""
super()._delete_in_kernel(nft_type='map')
def _create_new_set_in_kernel(self):
"""Create the non-existing set, creating table if needed."""
super()._create_new_set_in_kernel(nft_type='map')
def validate_data(self, dict_data):
"""
Validate data, returning it or raising a ValueError.
@ -324,26 +483,8 @@ class NetfilterSet:
return set_
def _apply_target_content(self):
"""Change netfilter content to target set."""
if self.nft_type == 'set':
self._apply_target_content_set()
else:
self._apply_target_content_map()
def _apply_target_content_set(self):
"""Change netfilter set content to target set."""
current_set = self.get_netfilter_set_content()
if current_set is None:
raise ValueError('Cannot change "{}" netfilter set content: set '
'do not exist in "{}" "{}".'.format(
self.name, self.address_family, self.table))
to_delete = current_set - self._target_content
to_add = self._target_content - current_set
self._change_set_content(delete=to_delete, add=to_add)
def _apply_target_content_map(self):
"""Change netfilter set content to target set."""
current_map = self.get_netfilter_map_content()
"""Change netfilter map content to target map."""
current_map = self.get_netfilter_content()
if current_map is None:
raise ValueError('Cannot change "{}" netfilter map content: map '
'do not exist in "{}" "{}".'.format(
@ -356,23 +497,9 @@ class NetfilterSet:
keys_to_add.add(k)
keys_to_delete.add(k)
to_add = {k : self._target_content[k] for k in keys_to_add}
self._change_map_content(delete=keys_to_delete, add=to_add)
self._change_content(delete=keys_to_delete, add=to_add)
def _change_set_content(self, delete=None, add=None):
todo = [tuple_ for tuple_ in (('add', add), ('delete', delete))
if tuple_[1]]
for action, elements in todo:
content = ', '.join(' . '.join(str(element) for element in tuple_)
for tuple_ in elements)
command = [
*self.nft,
'{action} element {addr_family} {table} {set_} {{{content}}}' \
.format(action=action, addr_family=self.address_family,
table=self.table, set_=self.name, content=content)
]
CommandExec.run(command)
def _change_map_content(self, delete=None, add=None):
def _change_content(self, delete=None, add=None):
if delete:
content = ', '.join(' . '.join(str(element) for element in tuple_)
for tuple_ in delete)
@ -398,50 +525,7 @@ class NetfilterSet:
]
CommandExec.run(command)
def _get_raw_netfilter_set(self, parse_elements=True):
"""Return a dict describing the netfilter set matching self or None."""
_, stdout, _ = CommandExec.run_check_output(
[*self.nft, '-nn', 'list set {addr_family} {table} {set_}'.format(
addr_family=self.address_family, table=self.table,
set_=self.name)],
allowed_return_codes=(0, 1) # In case table do not exist
)
if not stdout:
return None
else:
netfilter_set = self._parse_netfilter_set_string(stdout)
if netfilter_set['name'] != self.name \
or netfilter_set['address_family'] != self.address_family \
or netfilter_set['table'] != self.table \
or not self.has_type(netfilter_set['type']) \
or netfilter_set.get('flags', set()) != self.flags:
raise ValueError(
'Did not get the right set, too wrong to fix. Got '
+ str(netfilter_set)
+ ("\nExpected : "
"\n\tname: {name}"
"\n\taddress_family: {family}"
"\n\ttable: {table}"
"\n\tflags: {flags}"
"\n\ttypes: {types}"
).format(
name=self.name,
family=self.address_family,
table=self.table,
flags=self.flags,
types=tuple(self.TYPES[t] for t in self.type)
)
)
if parse_elements:
if netfilter_set['raw_content']:
netfilter_set['content'] = self.validate_set_data((
(element.strip() for element in n_uplet.split(' . '))
for n_uplet in netfilter_set['raw_content'].split(',')))
else:
netfilter_set['content'] = set()
return netfilter_set
def _get_raw_netfilter_map(self, parse_elements=True):
def _get_raw_netfilter(self, parse_elements=True):
"""Return a dict describing the netfilter map matching self or None."""
_, stdout, _ = CommandExec.run_check_output(
[*self.nft, '-nn', 'list map {addr_family} {table} {set_}'.format(
@ -452,184 +536,177 @@ class NetfilterSet:
if not stdout:
return None
else:
netfilter_set = self._parse_netfilter_map_string(stdout)
netfilter_set = self._parse_netfilter_string(stdout)
if netfilter_set['name'] != self.name \
or netfilter_set['address_family'] != self.address_family \
or netfilter_set['table'] != self.table \
or not self.has_type(netfilter_set['type']):
or not self.has_type((netfilter_set['type_from'], netfilter_set['type'])):
raise ValueError('Did not get the right map, too wrong to fix.')
if parse_elements:
if netfilter_set['raw_content']:
netfilter_set['content'] = self.validate_map_data({
(element.strip() for element in n_uplet.split(' : ')[0].split(' . ')) :
(element.strip() for element in n_uplet.split(' : ')[1].split(' . '))
netfilter_set['content'] = self.validate_data({
(element.strip() for element in n_uplet.split(' : ')[0].split(' . ')):
n_uplet.split(' : ')[1].strip()
for n_uplet in netfilter_set['raw_content'].split(',')
})
else:
netfilter_set['content'] = {}
return netfilter_set
@staticmethod
def _parse_netfilter_set_string(set_string):
"""
Parse netfilter set definition and return set as dict.
Do not validate content type against detected set type.
Return a dict with 'name', 'address_family', 'table', 'type', 'flags',
'raw_content' keys (all strings, 'raw_content' can be None).
Raise ValueError in case of unexpected syntax.
"""
# A.K.A. Really, I don't hate you, so please don't hate me...
regexp = (
"table (?P<address_family>\w+)+ (?P<table>\w+) \{\n"
"\s*set (?P<name>\w+) \{\n"
"\s*type (?P<type>(\w+( \. )?)+)\n"
"(\s*elements = \{ "
"(?P<elements>((\n\s*)?([\w:\.]+( \. )?)+,?)*) "
"\}\n)?"
"\s*\}\n"
"\s*\}"
)
values = re.match(regexp, set_string).groupdict()
return {
'address_family': values['address_family'],
'table': values['table'],
'name': values['name'],
'type': values['type'].split(' . '),
'raw_content': values['elements'],
}
@staticmethod
def _parse_netfilter_map_string(set_string):
@classmethod
def _parse_netfilter_string(cls, set_string):
"""
Parse netfilter map definition and return map as dict.
Do not validate content type against detected map type.
Return a dict with 'name', 'address_family', 'table', 'type', 'flags'
'raw_content' keys (all strings, 'raw_content' can be None).
Raise ValueError in case of unexpected syntax.
'raw_content' and 'type_from' keys (all strings, 'raw_content' and
'flags' can be None). Raise ValueError in case of unexpected syntax.
"""
# Fragile code since using lexer / parser would be quite heavy
lines = [line.lstrip('\t ') for line in set_string.strip().splitlines()]
errors = []
# 5 lines when empty, 6 with elements = { … } + one for flags
if len(lines) not in (5, 6, 7):
errors.append('Error, expecting 5 or 6 lines for set definition, '
'got "{}".'.format(set_string))
line_iterator = iter(lines)
set_definition = {}
line = next(line_iterator).split(' ') # line #1
# 'table <address_family> <chain> {'
if len(line) != 4 or line[0] != 'table' or line[3] != '{':
errors.append(
'Cannot parse table definition, expecting "type <addr_family> '
'<table> {{", got "{}".'.format(' '.join(line)))
else:
set_definition['address_family'] = line[1]
set_definition['table'] = line[2]
line = next(line_iterator).split(' ') # line #2
# 'set <name> {'
if len(line) != 3 or line[0] != 'map' or line[2] != '{':
errors.append('Cannot parse set definition, expecting "set <name> '
'{{", got "{}".' .format(' '.join(line)))
else:
set_definition['name'] = line[1]
line, elements_type = next(line_iterator).split(' : ') # line #3
# 'type <type> [. <type>]... : <type> [. <type>]...'
line = line.split(' ')
if len(line) < 2:
errors.append(
'Cannot parse type definition, left side of \':\' is too short : %s' % line
)
type_, keys_type = line[0], line[1:]
elements_type = elements_type.split(' ')
if type_ != 'type':
errors.append(
'Cannot parse type definition, expected first word \'type\', got %s' % type_
)
elif len(elements_type) % 2 != 1 or len(keys_type) % 2 != 1 \
or any(e != '.' for e in elements_type[1::2]) \
or any(e != '.' for e in keys_type[1::2]):
errors.append(
'Cannot parse type definition, expecting "type <type> '
'[. <type>]... : <type> [. <type>]...", got "{}".'.format(' '.join(line)))
else:
set_definition['type'] = (keys_type[::2], elements_type[::2])
# here we can have the flags, if there are any
# flags <flag_1>, <flag_2>, ...
if len(lines) >= 6:
line = next(line_iterator)
if line[:5] == 'flags': # If there are actually flags
set_definition['flags'] = {f.strip() for f in line[:5].strip().split(',')}
if len(lines) >= 6:
# set is not empty, getting raw elements
if 'flags' in set_definition and len(lines) == 7: # the line unsplitted previously has been used.
line = next(line_iterator) # Unsplit line #4
print(line)
if ('flags' in set_definition and len(lines)==7) or ('flags' not in set_definition and len(lines)==6) :
if line[:13] != 'elements = { ' or line[-1] != '}':
errors.append('Cannot parse set elements, expecting "elements '
'= {{ <…>}}", got "{}".'.format(line))
else:
set_definition['raw_content'] = line[13:-1].strip()
else:
set_definition['raw_content'] = None
else:
set_definition['raw_content'] = None
# last two lines
for i in range(2):
line = next(line_iterator).split(' ')
if line != ['}']:
errors.append(
'No normal end to set definition, expecting "}}" on line '
'{}, got "{}".'.format(i+5, ' '.join(line)))
if errors:
raise ValueError('The following error(s) were encountered while '
'parsing set.\n"{}"'.format('",\n"'.join(errors)))
return set_definition
def get_netfilter_set_content(self):
"""Return current set content from netfilter."""
netfilter_set = self._get_raw_netfilter_set(parse_elements=True)
if netfilter_set is None:
return None
else:
return netfilter_set['content']
def get_netfilter_map_content(self):
"""Return current set content from netfilter."""
netfilter_set = self._get_raw_netfilter_map(parse_elements=True)
if netfilter_set is None:
return None
else:
return netfilter_set['content']
try:
values = cls.pattern.match(set_string).groupdict()
except Exception as e:
raise ValueError("Malformed expression :\n" + set_string)
return {
'address_family': values['address_family'],
'table': values['table'],
'name': values['name'],
'type': values['type'],
'type_from': values['type_from'].split(' . '),
'raw_content': values['elements'],
'flags': values['flags'],
}
def has_type(self, type_):
"""Check if some type match the set's one."""
if self.nft_type == 'set':
return tuple(self.TYPES[t] for t in self.type) == tuple(type_)
else:
return tuple(self.TYPES[t] for t in self.type) == tuple(type_[1]) and \
return tuple(self.TYPES[t] for t in self.type) == (type_[1],) and \
tuple(self.TYPES[t] for t in self.type_from) == tuple(type_[0])
def manage(self):
"""Create set if needed and populate it with target content."""
self.create_in_kernel()
self._apply_target_content()
def format_type(self):
if self.nft_type == 'set':
return ' . '.join(self.TYPES[i] for i in self.type)
else:
return ' . '.join(self.TYPES[i] for i in self.type_from) + ' : ' + ' . '.join(self.TYPES[i] for i in self.type)
def filter(self, elements):
return (self.filters[0](elements),)
def get_ip_iterable_from_str(ip):
try:
ret = netaddr.IPGlob(ip)
except netaddr.core.AddrFormatError:
try:
ret = netaddr.IPNetwork(ip)
except netaddr.core.AddrFormatError:
begin,end = ip.split('-')
ret = netaddr.IPRange(begin,end)
return ret
class NAT:
def __init__(self,
name,
range_in,
range_out,
first_port,
last_port,
use_sudo=True
):
"""Creates a NAT object for the given range of IP-Addresses.
Args:
name: name of the sets
range_in: an IPRange with the private IP address
range_out: an IPRange with the public IP address
first_port: the first port used for the nat
last_port: the last port used for the nat
use_sudo: Should the nft commands be run in sudo ?
"""
assert 0 <= first_port < last_port < 65536, (name + ": Your first_port "
"is lower than your last_port")
self.name = name
self.range_in = get_ip_iterable_from_str(range_in)
self.range_out = get_ip_iterable_from_str(range_out)
self.first_port = first_port
self.last_port = last_port
self.nb_private_by_public = self.range_in.size // self.range_out.size + 1
sudo = ["/usr/bin/sudo"] * int(bool(use_sudo))
self.nft = [*sudo, "/usr/sbin/nft"]
def create_nat_rule(self, grp, ports):
"""Create a nat rules in the form :
ip saddr @<self.name>_nat_port_<grp> ip protocol tcp snat ip saddr map @<self.name>_nat_address : <ports>
ip saddr @<self.name>_nat_port_<grp> ip protocol udp snat ip saddr map @<self.name>_nat_address : <ports>
Args:
grp: The name of the group
ports: The port range (str)
"""
CommandExec.run([
*self.nft,
"add rule ip nat {name}_nat ip saddr @{name}_nat_port_{grp} ip protocol tcp snat ip saddr map @{name}_nat_address : {ports}".format(
name=self.name,
grp=grp,
ports=ports
)
])
CommandExec.run([
*self.nft,
"add rule ip nat {name}_nat ip saddr @{name}_nat_port_{grp} ip protocol udp snat ip saddr map @{name}_nat_address : {ports}".format(
name=self.name,
grp=grp,
ports=ports
)
])
def manage(self):
"""Creates the port sets, ip map and rules
"""
ips = {}
ports = [
set() for i in range(self.nb_private_by_public)
]
for ip_out, ip in zip(
self.range_out,
range(self.range_in.first, self.range_in.last, self.nb_private_by_public)
):
range_size = self.nb_private_by_public if int(ip + self.nb_private_by_public) <= self.range_in.last else (self.range_in.last - ip)
ips[(netaddr.IPRange(ip, ip+range_size-1),)] = ip_out
for i in range(range_size):
ports[i].add((netaddr.IPAddress(ip+i),))
ip_map = NetfilterMap(
target_content=ips,
type_=('IPv4',),
name=self.name+'_nat_address',
table_name='nat',
flags=('interval',),
type_from=('IPv4',),
address_family='ip',
)
ip_map.manage()
port_range = lambda i : '-'.join([
str(int(self.first_port + i/self.nb_private_by_public * (self.last_port - self.first_port))),
str(int(self.first_port + (i+1)/self.nb_private_by_public * (self.last_port - self.first_port)-1))
])
for i, grp in enumerate(ports):
grp_set = NetfilterSet(
name=self.name+'_nat_port_'+str(i),
target_content=grp,
type_=('IPv4',),
table_name='nat',
address_family='ip',
)
grp_set.manage()
self.create_nat_rule(
str(i),
port_range(i)
)
class Firewall:
"""Manages the firewall using nftables."""

14
nat.nft
View file

@ -10,8 +10,20 @@ table ip nat {
chain postrouting {
type nat hook postrouting priority 100
meta oifname != $if_supelec return
meta iifname vmap {
$if_adherent : jump adherent_nat,
$if_admin : jump admin_nat,
$if_federez : jump federez_nat,
$if_aloes : jump aloes_nat,
$if_prerezotage : jump prerezotage_nat
}
counter
# ip saddr 10.0.0.0/8 snat to 193.48.225.3
meta oifname $if_supelec snat to 193.48.225.3
snat to 193.48.225.3
}

159
nat.py
View file

@ -24,88 +24,18 @@ from configparser import ConfigParser
import netaddr
from firewall import NetfilterSet
from firewall import NAT
CONFIG = ConfigParser()
CONFIG.read('config.ini')
def get_ip_iterable_from_str(ip):
try:
ret = netaddr.IPGlob(ip)
except netaddr.core.AddrFormatError:
try:
ret = netaddr.IPNetwork(ip)
except netaddr.core.AddrFormatError:
begin,end = ip.split('-')
ret = netaddr.IPRange(begin,end)
return ret
def create_nat(name, range_in, range_out, first_port, last_port):
"""Create two nftables tables for the nat:
- <name>_address : which link a (or a range of) local address to a
public address;
- <name>_port : which links a local address to a
range of ports.
Args:
name: name of the sets
range_in: an IPRange with the private IP address
range_out: an IPRange with the public IP address
first_port: the first port used for the nat
last_port: the last port used for the nat
Returns:
(<name>_address, <name>_port) which are NetfilterSet
"""
assert 0 <= first_port < last_port < 65536, (name + ": Your first_port "
"is lower than your last_port")
nb_private_by_public = range_in.size // range_out.size
nb_port_by_ip = (last_port - first_port + 1) // nb_private_by_public
ports = {}
ips = {}
port = first_port
for ip in range_in:
ports[(str(netaddr.IPAddress(ip)),)] = ("%d-%d" % (port, min(port+nb_port_by_ip, 65535)),)
port += nb_port_by_ip + 1
if port >= last_port:
port = first_port
ip = range_in.first
for ip_out in range_out:
ips[('-'.join([
str(netaddr.IPAddress(ip)),
str(netaddr.IPAddress(ip+nb_private_by_public))
]),)] = (str(ip_out),)
ip += nb_private_by_public + 1
return (
NetfilterSet(
target_content=ips,
type_=('IPv4',),
name=name+'_nat_address',
table_name='nat',
flags=('interval',),
type_from=('IPv4',)
),
NetfilterSet(
target_content=ports,
type_=('port',),
name=name+'_nat_port',
table_name='nat',
flags=('interval',),
type_from=('IPv4',)
),
)
def create_nat_adherent():
range_in = get_ip_iterable_from_str(CONFIG['NAT']['range_in_adherent'])
range_out = get_ip_iterable_from_str(CONFIG['NAT']['range_out_adherent'])
range_in = CONFIG['NAT']['range_in_adherent']
range_out = CONFIG['NAT']['range_out_adherent']
first_port = int(CONFIG['NAT']['first_port_adherent'])
last_port = int(CONFIG['NAT']['last_port_adherent'])
return create_nat(
return NAT(
'adherent',
range_in,
range_out,
@ -115,11 +45,11 @@ def create_nat_adherent():
def create_nat_federez():
range_in = get_ip_iterable_from_str(CONFIG['NAT']['range_in_federez'])
range_out = get_ip_iterable_from_str(CONFIG['NAT']['range_out_federez'])
first_port = CONFIG['NAT']['first_port_federez']
last_port = CONFIG['NAT']['last_port_federez']
return create_nat(
range_in = CONFIG['NAT']['range_in_federez']
range_out = CONFIG['NAT']['range_out_federez']
first_port = int(CONFIG['NAT']['first_port_federez'])
last_port = int(CONFIG['NAT']['last_port_federez'])
return NAT(
'federez',
range_in,
range_out,
@ -129,11 +59,11 @@ def create_nat_federez():
def create_nat_aloes():
range_in = get_ip_iterable_from_str(CONFIG['NAT']['range_in_aloes'])
range_out = get_ip_iterable_from_str(CONFIG['NAT']['range_out_aloes'])
first_port = CONFIG['NAT']['first_port_aloes']
last_port = CONFIG['NAT']['last_port_aloes']
return create_nat(
range_in = CONFIG['NAT']['range_in_aloes']
range_out = CONFIG['NAT']['range_out_aloes']
first_port = int(CONFIG['NAT']['first_port_aloes'])
last_port = int(CONFIG['NAT']['last_port_aloes'])
return NAT(
'aloes',
range_in,
range_out,
@ -143,11 +73,11 @@ def create_nat_aloes():
def create_nat_admin():
range_in = get_ip_iterable_from_str(CONFIG['NAT']['range_in_admin'])
range_out = get_ip_iterable_from_str(CONFIG['NAT']['range_out_admin'])
first_port = CONFIG['NAT']['first_port_admin']
last_port = CONFIG['NAT']['last_port_admin']
return create_nat(
range_in = CONFIG['NAT']['range_in_admin']
range_out = CONFIG['NAT']['range_out_admin']
first_port = int(CONFIG['NAT']['first_port_admin'])
last_port = int(CONFIG['NAT']['last_port_admin'])
return NAT(
'admin',
range_in,
range_out,
@ -156,46 +86,23 @@ def create_nat_admin():
)
def create_nat_prerezotage():
range_in = get_ip_iterable_from_str(CONFIG['NAT']['range_in_prerezotage'])
range_out = get_ip_iterable_from_str(CONFIG['NAT']['range_out_prerezotage'])
first_port = CONFIG['NAT']['first_port_prerezotage']
last_port = CONFIG['NAT']['last_port_prerezotage']
return create_nat(
'prerezotage',
range_in,
range_out,
first_port,
last_port
)
def main():
logging.info("Creating adherent nat...")
address, port = create_nat_adherent()
address.manage()
port.manage()
nat_adherent = create_nat_adherent()
nat_adherent.manage()
logging.info("Done.")
logging.info("Creating federez nat...")
nat_federez = create_nat_federez()
nat_federez.manage()
logging.info("Done.")
logging.info("Creating aloes nat...")
aloes_nat = create_nat_aloes()
aloes_nat.manage()
logging.info("Done.")
logging.info("Creating admin nat...")
admin_nat = create_nat_admin()
admin_nat.manage()
logging.info("Done.")
#logging.info("Creating federez nat...")
#address, port = create_nat_federez()
#address.manage()
#port.manage()
#logging.info("Done.")
#logging.info("Creating aloes nat...")
#address, port = create_nat_aloes()
#address.manage()
#port.manage()
#logging.info("Done.")
#logging.info("Creating admin nat...")
#address, port = create_nat_admin()
#address.manage()
#port.manage()
#logging.info("Done.")
#logging.info("Creating prerezotage nat...")
#address, port = create_nat_prerezotage()
#address.manage()
#port.manage()
#logging.info("Done.")
if __name__=='__main__':

18
zones/adherent.nft
View file

@ -22,18 +22,12 @@ table ip nat {
type ipv4_addr: ipv4_addr
flags interval
}
# exemple: 10.69.0.1 : 11135-12834
# On peut aussi ajouter dynamiquement des éléments :
# nft add element nat adherent_nat_port {10.69.0.1 : 11135-12834}
# Sauf qu'on peut pas faire de maps d'intervalles (seules les clés peuvent en être)
# du coup je vois rien d'autre à faire que de modifier à la volée les règles...
map adherent_nat_port {
type ipv4_addr: inet_service
flags interval
}
chain adh_nat {
ip protocol tcp snat ip saddr map @adherent_nat_address : ip saddr map @adherent_nat_port
ip protocol udp snat ip saddr map @adherent_nat_address : ip saddr map @adherent_nat_port
chain adherent_nat {
# Le script crée 32 sets et y répartit les adresses privées, puis
# crée 2*32 lignes correspondantes de la forme de celles qui suivent
# en matchant sur les sets précédents...
# ip saddr @adherent_nat_grp1 ip protocol tcp snat ip saddr map @adherent_nat_address : 11135-12834
# ip saddr @adherent_nat_grp1 ip protocol udp snat ip saddr map @adherent_nat_address : 11135-12834
}
}

11
zones/admin.nft
View file

@ -22,16 +22,7 @@ table nat {
type ipv4_addr: ipv4_addr
flags interval
}
# exemple: 10.7.0.1 : 11135-12834
# On peut aussi ajouter dynamiquement des éléments :
# nft add element nat federez_nat_port {10.7.0.1 : 11135-12834}
map admin_nat_port {
type ipv4_addr: inet_service
flags interval
}
chain adm_nat {
ip protocol tcp snat ip saddr map @admin_nat_address : ip saddr map @admin_nat_port
ip protocol udp snat ip saddr map @admin_nat_address : ip saddr map @admin_nat_port
chain admin_nat {
}
}

9
zones/aloes.nft
View file

@ -24,16 +24,7 @@ table nat {
type ipv4_addr: ipv4_addr
flags interval
}
# exemple: 10.66.0.1 : 11135-12834
# On peut aussi ajouter dynamiquement des éléments :
# nft add element nat federez_nat_port {10.66.0.1 : 11135-12834}
map aloes_nat_port {
type ipv4_addr: inet_service
flags interval
}
chain aloes_nat {
ip protocol tcp snat ip saddr map @aloes_nat_address : ip saddr map @aloes_nat_port
ip protocol udp snat ip saddr map @aloes_nat_address : ip saddr map @aloes_nat_port
}
}

9
zones/federez.nft
View file

@ -22,16 +22,7 @@ table nat {
type ipv4_addr: ipv4_addr
flags interval
}
# exemple: 10.20.0.1 : 11135-12834
# On peut aussi ajouter dynamiquement des éléments :
# nft add element nat federez_nat_port {10.20.0.1 : 11135-12834}
map federez_nat_port {
type ipv4_addr: inet_service
flags interval
}
chain federez_nat {
ip protocol tcp snat ip saddr map @federez_nat_address : ip saddr map @federez_nat_port
ip protocol udp snat ip saddr map @federez_nat_address : ip saddr map @federez_nat_port
}
}