diff --git a/python/vyos/utils/process.py b/python/vyos/utils/process.py
index bd0644bc0..a2398edf7 100644
--- a/python/vyos/utils/process.py
+++ b/python/vyos/utils/process.py
@@ -1,247 +1,247 @@
# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library. If not, see <http://www.gnu.org/licenses/>.
import os
from subprocess import Popen
from subprocess import PIPE
from subprocess import STDOUT
from subprocess import DEVNULL
def popen(command, flag='', shell=None, input=None, timeout=None, env=None,
stdout=PIPE, stderr=PIPE, decode='utf-8'):
"""
popen is a wrapper helper aound subprocess.Popen
with it default setting it will return a tuple (out, err)
out: the output of the program run
err: the error code returned by the program
it can be affected by the following flags:
shell: do not try to auto-detect if a shell is required
for example if a pipe (|) or redirection (>, >>) is used
input: data to sent to the child process via STDIN
the data should be bytes but string will be converted
timeout: time after which the command will be considered to have failed
env: mapping that defines the environment variables for the new process
stdout: define how the output of the program should be handled
- PIPE (default), sends stdout to the output
- DEVNULL, discard the output
stderr: define how the output of the program should be handled
- None (default), send/merge the data to/with stderr
- PIPE, popen will append it to output
- STDOUT, send the data to be merged with stdout
- DEVNULL, discard the output
decode: specify the expected text encoding (utf-8, ascii, ...)
the default is explicitely utf-8 which is python's own default
usage:
get both stdout and stderr: popen('command', stdout=PIPE, stderr=STDOUT)
discard stdout and get stderr: popen('command', stdout=DEVNUL, stderr=PIPE)
"""
# airbag must be left as an import in the function as otherwise we have a
# a circual import dependency
from vyos import debug
from vyos import airbag
# log if the flag is set, otherwise log if command is set
if not debug.enabled(flag):
flag = 'command'
cmd_msg = f"cmd '{command}'"
debug.message(cmd_msg, flag)
use_shell = shell
stdin = None
if shell is None:
use_shell = False
if ' ' in command:
use_shell = True
if env:
use_shell = True
if input:
stdin = PIPE
input = input.encode() if type(input) is str else input
p = Popen(command, stdin=stdin, stdout=stdout, stderr=stderr,
env=env, shell=use_shell)
pipe = p.communicate(input, timeout)
pipe_out = b''
if stdout == PIPE:
pipe_out = pipe[0]
pipe_err = b''
if stderr == PIPE:
pipe_err = pipe[1]
str_out = pipe_out.decode(decode).replace('\r\n', '\n').strip()
str_err = pipe_err.decode(decode).replace('\r\n', '\n').strip()
out_msg = f"returned (out):\n{str_out}"
if str_out:
debug.message(out_msg, flag)
if str_err:
from sys import stderr
err_msg = f"returned (err):\n{str_err}"
# this message will also be send to syslog via airbag
debug.message(err_msg, flag, destination=stderr)
# should something go wrong, report this too via airbag
airbag.noteworthy(cmd_msg)
airbag.noteworthy(out_msg)
airbag.noteworthy(err_msg)
return str_out, p.returncode
def run(command, flag='', shell=None, input=None, timeout=None, env=None,
stdout=DEVNULL, stderr=PIPE, decode='utf-8'):
"""
A wrapper around popen, which discard the stdout and
will return the error code of a command
"""
_, code = popen(
command, flag,
stdout=stdout, stderr=stderr,
input=input, timeout=timeout,
env=env, shell=shell,
decode=decode,
)
return code
def cmd(command, flag='', shell=None, input=None, timeout=None, env=None,
stdout=PIPE, stderr=PIPE, decode='utf-8', raising=None, message='',
expect=[0]):
"""
A wrapper around popen, which returns the stdout and
will raise the error code of a command
raising: specify which call should be used when raising
the class should only require a string as parameter
(default is OSError) with the error code
expect: a list of error codes to consider as normal
"""
decoded, code = popen(
command, flag,
stdout=stdout, stderr=stderr,
input=input, timeout=timeout,
env=env, shell=shell,
decode=decode,
)
if code not in expect:
feedback = message + '\n' if message else ''
feedback += f'failed to run command: {command}\n'
feedback += f'returned: {decoded}\n'
feedback += f'exit code: {code}'
if raising is None:
# error code can be recovered with .errno
raise OSError(code, feedback)
else:
raise raising(feedback)
return decoded
def rc_cmd(command, flag='', shell=None, input=None, timeout=None, env=None,
stdout=PIPE, stderr=STDOUT, decode='utf-8'):
"""
A wrapper around popen, which returns the return code
of a command and stdout
% rc_cmd('uname')
(0, 'Linux')
% rc_cmd('ip link show dev eth99')
(1, 'Device "eth99" does not exist.')
"""
out, code = popen(
command, flag,
stdout=stdout, stderr=stderr,
input=input, timeout=timeout,
env=env, shell=shell,
decode=decode,
)
return code, out
def call(command, flag='', shell=None, input=None, timeout=None, env=None,
stdout=None, stderr=None, decode='utf-8'):
"""
A wrapper around popen, which print the stdout and
will return the error code of a command
"""
out, code = popen(
command, flag,
stdout=stdout, stderr=stderr,
input=input, timeout=timeout,
env=env, shell=shell,
decode=decode,
)
if out:
print(out)
return code
def process_running(pid_file):
""" Checks if a process with PID in pid_file is running """
from psutil import pid_exists
if not os.path.isfile(pid_file):
return False
with open(pid_file, 'r') as f:
pid = f.read().strip()
return pid_exists(int(pid))
def process_named_running(name: str, cmdline: str=None, timeout: int=0):
""" Checks if process with given name is running and returns its PID.
If Process is not running, return None
"""
from psutil import process_iter
def check_process(name, cmdline):
for p in process_iter(['name', 'pid', 'cmdline']):
if cmdline:
if name in p.info['name'] and cmdline in p.info['cmdline']:
return p.info['pid']
elif name in p.info['name']:
return p.info['pid']
return None
if timeout:
import time
time_expire = time.time() + timeout
while True:
tmp = check_process(name, cmdline)
if not tmp:
if time.time() > time_expire:
break
- time.sleep(0.100) # wait 250ms
+ time.sleep(0.100) # wait 100ms
continue
return tmp
else:
return check_process(name, cmdline)
return None
def is_systemd_service_active(service):
""" Test is a specified systemd service is activated.
Returns True if service is active, false otherwise.
Copied from: https://unix.stackexchange.com/a/435317 """
tmp = cmd(f'systemctl show --value -p ActiveState {service}')
return bool((tmp == 'active'))
def is_systemd_service_running(service):
""" Test is a specified systemd service is actually running.
Returns True if service is running, false otherwise.
Copied from: https://unix.stackexchange.com/a/435317 """
tmp = cmd(f'systemctl show --value -p SubState {service}')
return bool((tmp == 'running'))
diff --git a/smoketest/scripts/cli/base_vyostest_shim.py b/smoketest/scripts/cli/base_vyostest_shim.py
index efaa74fe0..112c58c09 100644
--- a/smoketest/scripts/cli/base_vyostest_shim.py
+++ b/smoketest/scripts/cli/base_vyostest_shim.py
@@ -1,142 +1,144 @@
# Copyright (C) 2021-2024 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import unittest
import paramiko
from time import sleep
from typing import Type
from vyos.configsession import ConfigSession
from vyos.configsession import ConfigSessionError
from vyos import ConfigError
from vyos.defaults import commit_lock
from vyos.utils.process import cmd
from vyos.utils.process import run
save_config = '/tmp/vyos-smoketest-save'
# This class acts as shim between individual Smoketests developed for VyOS and
# the Python UnitTest framework. Before every test is loaded, we dump the current
# system configuration and reload it after the test - despite the test results.
#
# Using this approach we can not render a live system useless while running any
# kind of smoketest. In addition it adds debug capabilities like printing the
# command used to execute the test.
class VyOSUnitTestSHIM:
class TestCase(unittest.TestCase):
# if enabled in derived class, print out each and every set/del command
# on the CLI. This is usefull to grap all the commands required to
# trigger the certain failure condition.
# Use "self.debug = True" in derived classes setUp() method
debug = False
@classmethod
def setUpClass(cls):
cls._session = ConfigSession(os.getpid())
cls._session.save_config(save_config)
if os.path.exists('/tmp/vyos.smoketest.debug'):
cls.debug = True
pass
@classmethod
def tearDownClass(cls):
# discard any pending changes which might caused a messed up config
cls._session.discard()
# ... and restore the initial state
cls._session.migrate_and_load_config(save_config)
try:
cls._session.commit()
except (ConfigError, ConfigSessionError):
cls._session.discard()
cls.fail(cls)
def cli_set(self, config):
if self.debug:
print('set ' + ' '.join(config))
self._session.set(config)
def cli_delete(self, config):
if self.debug:
print('del ' + ' '.join(config))
self._session.delete(config)
def cli_commit(self):
+ if self.debug:
+ print('commit')
self._session.commit()
# during a commit there is a process opening commit_lock, and run() returns 0
while run(f'sudo lsof -nP {commit_lock}') == 0:
sleep(0.250)
def getFRRconfig(self, string=None, end='$', endsection='^!', daemon=''):
""" Retrieve current "running configuration" from FRR """
command = f'vtysh -c "show run {daemon} no-header"'
if string: command += f' | sed -n "/^{string}{end}/,/{endsection}/p"'
out = cmd(command)
if self.debug:
import pprint
print(f'\n\ncommand "{command}" returned:\n')
pprint.pprint(out)
return out
@staticmethod
def ssh_send_cmd(command, username, password, hostname='localhost'):
""" SSH command execution helper """
# Try to login via SSH
ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh_client.connect(hostname=hostname, username=username, password=password)
_, stdout, stderr = ssh_client.exec_command(command)
output = stdout.read().decode().strip()
error = stderr.read().decode().strip()
ssh_client.close()
return output, error
# Verify nftables output
def verify_nftables(self, nftables_search, table, inverse=False, args=''):
nftables_output = cmd(f'sudo nft {args} list table {table}')
for search in nftables_search:
matched = False
for line in nftables_output.split("\n"):
if all(item in line for item in search):
matched = True
break
self.assertTrue(not matched if inverse else matched, msg=search)
def verify_nftables_chain(self, nftables_search, table, chain, inverse=False, args=''):
nftables_output = cmd(f'sudo nft {args} list chain {table} {chain}')
for search in nftables_search:
matched = False
for line in nftables_output.split("\n"):
if all(item in line for item in search):
matched = True
break
self.assertTrue(not matched if inverse else matched, msg=search)
# standard construction; typing suggestion: https://stackoverflow.com/a/70292317
def ignore_warning(warning: Type[Warning]):
import warnings
from functools import wraps
def inner(f):
@wraps(f)
def wrapped(*args, **kwargs):
with warnings.catch_warnings():
warnings.simplefilter("ignore", category=warning)
return f(*args, **kwargs)
return wrapped
return inner
diff --git a/smoketest/scripts/system/test_kernel_options.py b/smoketest/scripts/system/test_kernel_options.py
index 4666e98e7..700e4cec7 100755
--- a/smoketest/scripts/system/test_kernel_options.py
+++ b/smoketest/scripts/system/test_kernel_options.py
@@ -1,132 +1,132 @@
#!/usr/bin/env python3
#
# Copyright (C) 2020-2024 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import re
import os
import platform
import unittest
-kernel = platform.release()
+from vyos.utils.kernel import check_kmod
+kernel = platform.release()
class TestKernelModules(unittest.TestCase):
""" VyOS makes use of a lot of Kernel drivers, modules and features. The
required modules which are essential for VyOS should be tested that they are
available in the Kernel that is run. """
_config_data = None
@classmethod
def setUpClass(cls):
import gzip
from vyos.utils.process import call
super(TestKernelModules, cls).setUpClass()
CONFIG = '/proc/config.gz'
-
if not os.path.isfile(CONFIG):
- call('sudo modprobe configs')
+ check_kmod('configs')
with gzip.open(CONFIG, 'rt') as f:
cls._config_data = f.read()
def test_bond_interface(self):
# The bond/lacp interface must be enabled in the OS Kernel
for option in ['CONFIG_BONDING']:
tmp = re.findall(f'{option}=(y|m)', self._config_data)
self.assertTrue(tmp)
def test_bridge_interface(self):
# The bridge interface must be enabled in the OS Kernel
for option in ['CONFIG_BRIDGE',
'CONFIG_BRIDGE_IGMP_SNOOPING',
'CONFIG_BRIDGE_VLAN_FILTERING']:
tmp = re.findall(f'{option}=(y|m)', self._config_data)
self.assertTrue(tmp)
def test_dropmon_enabled(self):
options_to_check = [
'CONFIG_NET_DROP_MONITOR=y',
'CONFIG_UPROBE_EVENTS=y',
'CONFIG_BPF_EVENTS=y',
'CONFIG_TRACEPOINTS=y'
]
for option in options_to_check:
self.assertIn(option, self._config_data)
def test_synproxy_enabled(self):
options_to_check = [
'CONFIG_NFT_SYNPROXY',
'CONFIG_IP_NF_TARGET_SYNPROXY'
]
for option in options_to_check:
tmp = re.findall(f'{option}=(y|m)', self._config_data)
self.assertTrue(tmp)
def test_qemu_support(self):
options_to_check = [
'CONFIG_VIRTIO_BLK', 'CONFIG_SCSI_VIRTIO',
'CONFIG_VIRTIO_NET', 'CONFIG_VIRTIO_CONSOLE',
'CONFIG_VIRTIO', 'CONFIG_VIRTIO_PCI',
'CONFIG_VIRTIO_BALLOON', 'CONFIG_CRYPTO_DEV_VIRTIO',
'CONFIG_X86_PLATFORM_DEVICES'
]
for option in options_to_check:
tmp = re.findall(f'{option}=(y|m)', self._config_data)
self.assertTrue(tmp)
def test_vmware_support(self):
for option in ['CONFIG_VMXNET3']:
tmp = re.findall(f'{option}=(y|m)', self._config_data)
self.assertTrue(tmp)
def test_container_cgroup_support(self):
options_to_check = [
'CONFIG_CGROUPS', 'CONFIG_MEMCG',
'CONFIG_CGROUP_PIDS', 'CONFIG_CGROUP_BPF'
]
for option in options_to_check:
tmp = re.findall(f'{option}=(y|m)', self._config_data)
self.assertTrue(tmp)
def test_ip_routing_support(self):
options_to_check = [
'CONFIG_IP_ADVANCED_ROUTER', 'CONFIG_IP_MULTIPLE_TABLES',
'CONFIG_IP_ROUTE_MULTIPATH'
]
for option in options_to_check:
tmp = re.findall(f'{option}=(y|m)', self._config_data)
self.assertTrue(tmp)
def test_vfio(self):
options_to_check = [
'CONFIG_VFIO', 'CONFIG_VFIO_GROUP', 'CONFIG_VFIO_CONTAINER',
'CONFIG_VFIO_IOMMU_TYPE1', 'CONFIG_VFIO_NOIOMMU', 'CONFIG_VFIO_VIRQFD'
]
for option in options_to_check:
tmp = re.findall(f'{option}=(y|m)', self._config_data)
self.assertTrue(tmp)
def test_container_cpu(self):
options_to_check = [
'CONFIG_CGROUP_SCHED', 'CONFIG_CPUSETS', 'CONFIG_CGROUP_CPUACCT', 'CONFIG_CFS_BANDWIDTH'
]
for option in options_to_check:
tmp = re.findall(f'{option}=(y|m)', self._config_data)
self.assertTrue(tmp)
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/src/conf_mode/interfaces_wireless.py b/src/conf_mode/interfaces_wireless.py
index 3f01612a2..ff38c979c 100755
--- a/src/conf_mode/interfaces_wireless.py
+++ b/src/conf_mode/interfaces_wireless.py
@@ -1,308 +1,329 @@
#!/usr/bin/env python3
#
# Copyright (C) 2019-2024 VyOS maintainers and contributors
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 or later as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
from sys import exit
from re import findall
from netaddr import EUI, mac_unix_expanded
+from time import sleep
from vyos.config import Config
from vyos.configdict import get_interface_dict
from vyos.configdict import dict_merge
from vyos.configverify import verify_address
from vyos.configverify import verify_bridge_delete
from vyos.configverify import verify_mirror_redirect
from vyos.configverify import verify_vlan_config
from vyos.configverify import verify_vrf
from vyos.configverify import verify_bond_bridge_member
from vyos.ifconfig import WiFiIf
from vyos.template import render
from vyos.utils.dict import dict_search
from vyos.utils.kernel import check_kmod
from vyos.utils.process import call
+from vyos.utils.process import is_systemd_service_active
+from vyos.utils.process import is_systemd_service_running
+from vyos.utils.network import interface_exists
from vyos import ConfigError
from vyos import airbag
airbag.enable()
# XXX: wpa_supplicant works on the source interface
wpa_suppl_conf = '/run/wpa_supplicant/{ifname}.conf'
hostapd_conf = '/run/hostapd/{ifname}.conf'
hostapd_accept_station_conf = '/run/hostapd/{ifname}_station_accept.conf'
hostapd_deny_station_conf = '/run/hostapd/{ifname}_station_deny.conf'
def find_other_stations(conf, base, ifname):
"""
Only one wireless interface per phy can be in station mode -
find all interfaces attached to a phy which run in station mode
"""
old_level = conf.get_level()
conf.set_level(base)
dict = {}
for phy in os.listdir('/sys/class/ieee80211'):
list = []
for interface in conf.list_nodes([]):
if interface == ifname:
continue
# the following node is mandatory
if conf.exists([interface, 'physical-device', phy]):
tmp = conf.return_value([interface, 'type'])
if tmp == 'station':
list.append(interface)
if list:
dict.update({phy: list})
conf.set_level(old_level)
return dict
def get_config(config=None):
"""
Retrive CLI config as dictionary. Dictionary can never be empty, as at least the
interface name will be added or a deleted flag
"""
if config:
conf = config
else:
conf = Config()
base = ['interfaces', 'wireless']
ifname, wifi = get_interface_dict(conf, base)
if 'deleted' not in wifi:
# then get_interface_dict provides default keys
if wifi.from_defaults(['security', 'wep']): # if not set by user
del wifi['security']['wep']
if wifi.from_defaults(['security', 'wpa']): # if not set by user
del wifi['security']['wpa']
+ # XXX: Jinja2 can not operate on a dictionary key when it starts of with a number
+ if '40mhz_incapable' in (dict_search('capabilities.ht', wifi) or []):
+ wifi['capabilities']['ht']['fourtymhz_incapable'] = wifi['capabilities']['ht']['40mhz_incapable']
+ del wifi['capabilities']['ht']['40mhz_incapable']
+
if dict_search('security.wpa', wifi) != None:
wpa_cipher = wifi['security']['wpa'].get('cipher')
wpa_mode = wifi['security']['wpa'].get('mode')
if not wpa_cipher:
tmp = None
if wpa_mode == 'wpa':
tmp = {'security': {'wpa': {'cipher' : ['TKIP', 'CCMP']}}}
elif wpa_mode == 'wpa2':
tmp = {'security': {'wpa': {'cipher' : ['CCMP']}}}
elif wpa_mode == 'both':
tmp = {'security': {'wpa': {'cipher' : ['CCMP', 'TKIP']}}}
elif wpa_mode == 'wpa3':
# According to WiFi specs (https://www.wi-fi.org/file/wpa3-specification)
# section 3.5: WPA3-Enterprise 192-bit mode
# WiFi NICs which would be able to connect to WPA3-Enterprise managed
# networks MUST support GCMP-256.
# Reasoning: Provided that chipsets would most likely _not_ be
# "private user only", they all would come with built-in support
# for GCMP-256.
tmp = {'security': {'wpa': {'cipher' : ['CCMP', 'CCMP-256', 'GCMP', 'GCMP-256']}}}
if tmp: wifi = dict_merge(tmp, wifi)
# Only one wireless interface per phy can be in station mode
tmp = find_other_stations(conf, base, wifi['ifname'])
if tmp: wifi['station_interfaces'] = tmp
- # used in hostapt.conf.j2
+ # used in hostapd.conf.j2
wifi['hostapd_accept_station_conf'] = hostapd_accept_station_conf.format(**wifi)
wifi['hostapd_deny_station_conf'] = hostapd_deny_station_conf.format(**wifi)
return wifi
def verify(wifi):
if 'deleted' in wifi:
verify_bridge_delete(wifi)
return None
if 'physical_device' not in wifi:
raise ConfigError('You must specify a physical-device "phy"')
physical_device = wifi['physical_device']
if not os.path.exists(f'/sys/class/ieee80211/{physical_device}'):
raise ConfigError(f'Wirelss interface PHY "{physical_device}" does not exist!')
if 'type' not in wifi:
raise ConfigError('You must specify a WiFi mode')
if 'ssid' not in wifi and wifi['type'] != 'monitor':
raise ConfigError('SSID must be configured unless type is set to "monitor"!')
if wifi['type'] == 'access-point':
if 'country_code' not in wifi:
raise ConfigError('Wireless country-code is mandatory')
if 'channel' not in wifi:
raise ConfigError('Wireless channel must be configured!')
if 'capabilities' in wifi and 'he' in wifi['capabilities']:
if 'channel_set_width' not in wifi['capabilities']['he']:
raise ConfigError('Channel width must be configured!')
# op_modes drawn from:
# https://w1.fi/cgit/hostap/tree/src/common/ieee802_11_common.c?id=195cc3d919503fb0d699d9a56a58a72602b25f51#n1525
# 802.11ax (WiFi-6e - HE) can use up to 160MHz bandwidth channels
six_ghz_op_modes_he = ['131', '132', '133', '134', '135']
# 802.11be (WiFi-7 - EHT) can use up to 320MHz bandwidth channels
six_ghz_op_modes_eht = six_ghz_op_modes_he.append('137')
if 'security' in wifi and 'wpa' in wifi['security'] and 'mode' in wifi['security']['wpa']:
if wifi['security']['wpa']['mode'] == 'wpa3':
if 'he' in wifi['capabilities']:
if wifi['capabilities']['he']['channel_set_width'] in six_ghz_op_modes_he:
if 'mgmt_frame_protection' not in wifi or wifi['mgmt_frame_protection'] != 'required':
raise ConfigError('Management Frame Protection (MFP) is required with WPA3 at 6GHz! Consider also enabling Beacon Frame Protection (BFP) if your device supports it.')
if 'security' in wifi:
if {'wep', 'wpa'} <= set(wifi.get('security', {})):
raise ConfigError('Must either use WEP or WPA security!')
if 'wep' in wifi['security']:
if 'key' in wifi['security']['wep'] and len(wifi['security']['wep']) > 4:
raise ConfigError('No more then 4 WEP keys configurable')
elif 'key' not in wifi['security']['wep']:
raise ConfigError('Security WEP configured - missing WEP keys!')
elif 'wpa' in wifi['security']:
wpa = wifi['security']['wpa']
if not any(i in ['passphrase', 'radius'] for i in wpa):
raise ConfigError('Misssing WPA key or RADIUS server')
if 'radius' in wpa:
if 'server' in wpa['radius']:
for server in wpa['radius']['server']:
if 'key' not in wpa['radius']['server'][server]:
raise ConfigError(f'Misssing RADIUS shared secret key for server: {server}')
if 'capabilities' in wifi:
capabilities = wifi['capabilities']
if 'vht' in capabilities:
if 'ht' not in capabilities:
raise ConfigError('Specify HT flags if you want to use VHT!')
if {'beamform', 'antenna_count'} <= set(capabilities.get('vht', {})):
if capabilities['vht']['antenna_count'] == '1':
raise ConfigError('Cannot use beam forming with just one antenna!')
if capabilities['vht']['beamform'] == 'single-user-beamformer':
if int(capabilities['vht']['antenna_count']) < 3:
# Nasty Gotcha: see lines 708-721 in:
# https://w1.fi/cgit/hostap/tree/hostapd/hostapd.conf?h=hostap_2_10&id=cff80b4f7d3c0a47c052e8187d671710f48939e4#n708
raise ConfigError('Single-user beam former requires at least 3 antennas!')
if 'station_interfaces' in wifi and wifi['type'] == 'station':
phy = wifi['physical_device']
if phy in wifi['station_interfaces']:
if len(wifi['station_interfaces'][phy]) > 0:
raise ConfigError('Only one station per wireless physical interface possible!')
verify_address(wifi)
verify_vrf(wifi)
verify_bond_bridge_member(wifi)
verify_mirror_redirect(wifi)
# use common function to verify VLAN configuration
verify_vlan_config(wifi)
return None
def generate(wifi):
interface = wifi['ifname']
- # always stop hostapd service first before reconfiguring it
- call(f'systemctl stop hostapd@{interface}.service')
- # always stop wpa_supplicant service first before reconfiguring it
- call(f'systemctl stop wpa_supplicant@{interface}.service')
-
# Delete config files if interface is removed
if 'deleted' in wifi:
if os.path.isfile(hostapd_conf.format(**wifi)):
os.unlink(hostapd_conf.format(**wifi))
if os.path.isfile(hostapd_accept_station_conf.format(**wifi)):
os.unlink(hostapd_accept_station_conf.format(**wifi))
if os.path.isfile(hostapd_deny_station_conf.format(**wifi)):
os.unlink(hostapd_deny_station_conf.format(**wifi))
if os.path.isfile(wpa_suppl_conf.format(**wifi)):
os.unlink(wpa_suppl_conf.format(**wifi))
return None
if 'mac' not in wifi:
# http://wiki.stocksy.co.uk/wiki/Multiple_SSIDs_with_hostapd
# generate locally administered MAC address from used phy interface
with open('/sys/class/ieee80211/{physical_device}/addresses'.format(**wifi), 'r') as f:
# some PHYs tend to have multiple interfaces and thus supply multiple MAC
# addresses - we only need the first one for our calculation
tmp = f.readline().rstrip()
tmp = EUI(tmp).value
# mask last nibble from the MAC address
tmp &= 0xfffffffffff0
# set locally administered bit in MAC address
tmp |= 0x020000000000
# we now need to add an offset to our MAC address indicating this
# subinterfaces index
tmp += int(findall(r'\d+', interface)[0])
# convert integer to "real" MAC address representation
mac = EUI(hex(tmp).split('x')[-1])
# change dialect to use : as delimiter instead of -
mac.dialect = mac_unix_expanded
wifi['mac'] = str(mac)
- # XXX: Jinja2 can not operate on a dictionary key when it starts of with a number
- if '40mhz_incapable' in (dict_search('capabilities.ht', wifi) or []):
- wifi['capabilities']['ht']['fourtymhz_incapable'] = wifi['capabilities']['ht']['40mhz_incapable']
- del wifi['capabilities']['ht']['40mhz_incapable']
-
# render appropriate new config files depending on access-point or station mode
if wifi['type'] == 'access-point':
render(hostapd_conf.format(**wifi), 'wifi/hostapd.conf.j2', wifi)
render(hostapd_accept_station_conf.format(**wifi), 'wifi/hostapd_accept_station.conf.j2', wifi)
render(hostapd_deny_station_conf.format(**wifi), 'wifi/hostapd_deny_station.conf.j2', wifi)
elif wifi['type'] == 'station':
render(wpa_suppl_conf.format(**wifi), 'wifi/wpa_supplicant.conf.j2', wifi)
return None
def apply(wifi):
interface = wifi['ifname']
+ # From systemd source code:
+ # If there's a stop job queued before we enter the DEAD state, we shouldn't act on Restart=,
+ # in order to not undo what has already been enqueued. */
+ #
+ # It was found that calling restart on hostapd will (4 out of 10 cases) deactivate
+ # the service instead of restarting it, when it was not yet properly stopped
+ # systemd[1]: hostapd@wlan1.service: Deactivated successfully.
+ # Thus kill all WIFI service and start them again after it's ensured nothing lives
+ call(f'systemctl stop hostapd@{interface}.service')
+ call(f'systemctl stop wpa_supplicant@{interface}.service')
+
if 'deleted' in wifi:
- WiFiIf(interface).remove()
- else:
- # Finally create the new interface
- w = WiFiIf(**wifi)
- w.update(wifi)
-
- # Enable/Disable interface - interface is always placed in
- # administrative down state in WiFiIf class
- if 'disable' not in wifi:
- # Physical interface is now configured. Proceed by starting hostapd or
- # wpa_supplicant daemon. When type is monitor we can just skip this.
- if wifi['type'] == 'access-point':
- call(f'systemctl start hostapd@{interface}.service')
-
- elif wifi['type'] == 'station':
- call(f'systemctl start wpa_supplicant@{interface}.service')
+ WiFiIf(**wifi).remove()
+ return None
+
+ while (is_systemd_service_running(f'hostapd@{interface}.service') or \
+ is_systemd_service_active(f'hostapd@{interface}.service')):
+ sleep(0.250) # wait 250ms
+
+ # Finally create the new interface
+ w = WiFiIf(**wifi)
+ w.update(wifi)
+
+ # Enable/Disable interface - interface is always placed in
+ # administrative down state in WiFiIf class
+ if 'disable' not in wifi:
+ # Wait until interface was properly added to the Kernel
+ ii = 0
+ while not (interface_exists(interface) and ii < 20):
+ sleep(0.250) # wait 250ms
+ ii += 1
+
+ # Physical interface is now configured. Proceed by starting hostapd or
+ # wpa_supplicant daemon. When type is monitor we can just skip this.
+ if wifi['type'] == 'access-point':
+ call(f'systemctl start hostapd@{interface}.service')
+
+ elif wifi['type'] == 'station':
+ call(f'systemctl start wpa_supplicant@{interface}.service')
return None
if __name__ == '__main__':
try:
check_kmod('mac80211')
c = get_config()
verify(c)
generate(c)
apply(c)
except ConfigError as e:
print(e)
exit(1)