from jsonschema import ValidationError as JsonSchemaError
from ...exceptions import ValidationError
from ..base.backend import BaseBackend
from ..vxlan.vxlan_wireguard import VxlanWireguard
from ..wireguard.wireguard import Wireguard
from ..zerotier.zerotier import ZeroTier
from . import converters
from .parser import OpenWrtParser, config_path, packages_pattern
from .renderer import OpenWrtRenderer
from .schema import schema
class OpenWrt(BaseBackend):
"""
OpenWRT / LEDE Configuration Backend
"""
schema = schema
converters = [
converters.General,
converters.Ntp,
converters.Led,
converters.Interfaces,
converters.Routes,
converters.Rules,
converters.Switch,
converters.Radios,
converters.Wireless,
converters.OpenVpn,
converters.WireguardPeers,
converters.ZeroTier,
converters.Default,
]
parser = OpenWrtParser
renderer = OpenWrtRenderer
list_identifiers = ['name', 'config_value', 'id']
[docs]
def __init__(
self, config=None, native=None, templates=None, context=None, dsa=True
):
"""
:param config: ``dict`` containing a valid **NetJSON** configuration dictionary
:param native: ``str`` or file object representing a native configuration that will
be parsed and converted to a **NetJSON** configuration dictionary
:param templates: ``list`` containing **NetJSON** configuration dictionaries that
will be used as a base for the main config
:param context: ``dict`` containing configuration variables
:param dsa: ``bool`` flag to switch between OpenWrt configuration syntax.
``True`` generates configuration in OpenWrt >21 syntax.
``False`` generates configuration in OpenWrt <= 19 syntax.
:raises TypeError: raised if ``config`` is not of type ``dict`` or if
``templates`` is not of type ``list``
"""
self.dsa = dsa
super().__init__(config, native, templates, context)
def validate(self):
self._validate_radios()
super().validate()
# When VLAN filtering is enabled on a "bridge" interfaces,
# primary VLAN ID can be set for only one VLAN.
for index, interface in enumerate(self.config.get('interfaces', [])):
pvid_mapping = []
if interface.get('type') != 'bridge':
continue
for vlan in interface.get('vlan_filtering', []):
for port in vlan.get('ports', []):
if port.get('primary_vid', False):
if port['ifname'] in pvid_mapping:
raise ValidationError(
JsonSchemaError(
f'Invalid configuration triggered by "#/interfaces/{index}"'
' says: Primary VID can be set only one VLAN for a port.'
)
)
pvid_mapping.append(port['ifname'])
def _generate_contents(self, tar):
"""
Adds configuration files to tarfile instance.
:param tar: tarfile instance
:returns: None
"""
uci = self.render(files=False)
# create a list with all the packages (and remove empty entries)
packages = packages_pattern.split(uci)
if '' in packages:
packages.remove('')
# create an UCI file for each configuration package used
for package in packages:
lines = package.split('\n')
package_name = lines[0]
text_contents = '\n'.join(lines[2:])
self._add_file(
tar=tar,
name='{0}{1}'.format(config_path, package_name),
contents=text_contents,
)
[docs]
@classmethod
def wireguard_auto_client(cls, **kwargs):
data = Wireguard.auto_client(**kwargs)
config = {
'interfaces': [
{
'name': data['interface_name'],
'type': 'wireguard',
'private_key': data['client']['private_key'],
'port': data['client']['port'],
# Default values for Wireguard Interface
'mtu': 1420,
'nohostroute': False,
'fwmark': '',
'ip6prefix': [],
'addresses': [],
'network': '',
}
],
'wireguard_peers': [
{
'interface': data['interface_name'],
'public_key': data['server']['public_key'],
'allowed_ips': data['server']['allowed_ips'],
'endpoint_host': data['server']['endpoint_host'],
'endpoint_port': data['server']['endpoint_port'],
# Default values for Wireguard Peers
'preshared_key': '',
'persistent_keepalive': 60,
'route_allowed_ips': True,
}
],
}
if data['client']['ip_address']:
config['interfaces'][0]['addresses'] = [
{
'proto': 'static',
'family': 'ipv4',
'address': data['client']['ip_address'],
'mask': 32,
},
]
return config
[docs]
@classmethod
def vxlan_wireguard_auto_client(cls, **kwargs):
config = cls.wireguard_auto_client(**kwargs)
vxlan_config = VxlanWireguard.auto_client(**kwargs)
vxlan_interface = {
'name': vxlan_config['name'],
'type': 'vxlan',
'vtep': vxlan_config['server_ip_address'],
'port': 4789,
'vni': vxlan_config['vni'],
'tunlink': config['interfaces'][0]['name'],
# Default values for VXLAN interface
'rxcsum': True,
'txcsum': True,
'mtu': 1280,
'ttl': 64,
'mac': '',
'disabled': False,
'network': '',
}
config['interfaces'].append(vxlan_interface)
return config
[docs]
@classmethod
def zerotier_auto_client(cls, **kwargs):
data = ZeroTier.auto_client(**kwargs)
return {'zerotier': [data]}
def _validate_radios(self):
# We use "hwmode" or "band" property of "radio" configuration
# to predict the radio frequency. If both of these
# properties are absent from the configuration, then channels
# are used to predict the radio frequency. If the channel is
# set to "auto" (0) in the configuration, then netjsonconfig
# cannot predict the radio frequency. Thus, raises an error.
for radio in self.config.get('radios', []):
if radio['protocol'] not in ['802.11n', '802.11ax']:
continue
if (
radio.get('band') is None
and radio.get('hwmode') is None
and radio.get('channel') == 0
):
raise JsonSchemaError(
'"channel" cannot be set to "auto" when'
' "hwmode" or "band" property is not configured.'
)