177 lines
6.6 KiB
Python
177 lines
6.6 KiB
Python
"""Interfaces and addresses of the router, read from the system.
|
|
|
|
Only the Python standard library is used, so the web service and its bootstrap server can use
|
|
this module too.
|
|
"""
|
|
|
|
import pathlib
|
|
import subprocess
|
|
|
|
#: The values of web.listen and of the Router Manager's listen: the interfaces a service listens on.
|
|
LISTEN_CHOICES = ('internal', 'external', 'both')
|
|
|
|
#: The interfaces a service listens on when it is not set.
|
|
DEFAULT_LISTEN = 'both'
|
|
|
|
#: The address every service also listens on, where the HTTPS proxy calls it.
|
|
LOOPBACK = '127.0.0.1'
|
|
|
|
|
|
class ConfigError(Exception):
|
|
"""A setting is present but malformed."""
|
|
|
|
|
|
def ip_output(*args):
|
|
"""The output of an ip command, or '' when it fails."""
|
|
try:
|
|
result = subprocess.run(['ip', *args], capture_output=True, text=True)
|
|
except OSError:
|
|
return ''
|
|
return result.stdout if result.returncode == 0 else ''
|
|
|
|
|
|
def detect_interfaces():
|
|
"""Work out the external and internal interface names.
|
|
|
|
Only reached in auto mode, where the operator has accepted that a wrong
|
|
guess is possible. The external interface is whichever one carries the
|
|
default route; the internal one is the only other addressed interface.
|
|
"""
|
|
tokens = ip_output('route', 'get', '1.1.1.1').split()
|
|
external = ''
|
|
for index, token in enumerate(tokens):
|
|
if token == 'dev' and index + 1 < len(tokens):
|
|
external = tokens[index + 1]
|
|
break
|
|
if not external:
|
|
raise ConfigError('auto mode: cannot determine the external interface')
|
|
|
|
candidates = [name for name in _addressed_interfaces() if name != external]
|
|
if len(candidates) != 1:
|
|
raise ConfigError(
|
|
f'auto mode: expected one internal interface, found {len(candidates)}; '
|
|
'use mode = interfaces and name them')
|
|
return external, candidates[0]
|
|
|
|
|
|
def _is_virtual(name):
|
|
"""True for an interface the kernel creates without a device behind it.
|
|
|
|
Loopback, bridges, veth pairs and tunnels live under /sys/devices/virtual,
|
|
which is where the Docker bridge is; a NIC of the virtual machine does not.
|
|
"""
|
|
link = pathlib.Path('/sys/class/net') / name
|
|
return '/devices/virtual/' in str(link.resolve())
|
|
|
|
|
|
def _addressed_interfaces():
|
|
"""Names of the NICs holding at least one IPv4 address; virtual interfaces are left out."""
|
|
names = []
|
|
for line in ip_output('-o', '-4', 'addr', 'show').splitlines():
|
|
fields = line.split()
|
|
if len(fields) >= 4 and fields[1] not in names and not _is_virtual(fields[1]):
|
|
names.append(fields[1])
|
|
return names
|
|
|
|
|
|
def interface_addresses(name, setting):
|
|
"""IPv4 addresses configured on an interface.
|
|
|
|
The interface name is configuration; its addresses are read from the
|
|
system. Looking up the addresses of a named interface is deterministic -
|
|
unlike working out which interface to use, which is a guess the package
|
|
deliberately does not make.
|
|
"""
|
|
if not name:
|
|
return []
|
|
if not (pathlib.Path('/sys/class/net') / name).exists():
|
|
raise ConfigError(f'{setting}: interface {name!r} does not exist')
|
|
|
|
result = subprocess.run(
|
|
['ip', '-o', '-4', 'addr', 'show', 'dev', name],
|
|
capture_output=True, text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
raise ConfigError(f'{setting}: cannot read addresses of {name!r}')
|
|
|
|
addresses = []
|
|
for line in result.stdout.splitlines():
|
|
fields = line.split()
|
|
if len(fields) >= 4:
|
|
addresses.append(fields[3].split('/')[0])
|
|
return addresses
|
|
|
|
|
|
def internal_address(mode, internal, int_addr):
|
|
"""The router's address on the internal network, as general.mode directs, or '' when it is not known yet.
|
|
|
|
manual takes local.int_addr; interfaces reads the address of interfaces.internal; auto also
|
|
works out which interface that is.
|
|
"""
|
|
if mode == 'manual':
|
|
return int_addr
|
|
try:
|
|
if mode == 'auto':
|
|
internal = detect_interfaces()[1]
|
|
addresses = interface_addresses(internal, 'interfaces.internal')
|
|
except ConfigError:
|
|
return ''
|
|
return addresses[0] if addresses else ''
|
|
|
|
|
|
def external_address(mode, external):
|
|
"""The router's address on the external interface, as general.mode directs, or '' when it is not known yet.
|
|
|
|
auto works out which interface that is; the other modes read the address of interfaces.external.
|
|
"""
|
|
try:
|
|
if mode == 'auto':
|
|
external = detect_interfaces()[0]
|
|
addresses = interface_addresses(external, 'interfaces.external')
|
|
except ConfigError:
|
|
return ''
|
|
return addresses[0] if addresses else ''
|
|
|
|
|
|
def _section(configuration, name):
|
|
"""A section of a configuration, or {} when it is missing or not an object."""
|
|
value = configuration.get(name)
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def listen_addresses(configuration, choice):
|
|
"""The addresses a service listens on for choice of LISTEN_CHOICES, from vpn-router.json read as configuration.
|
|
|
|
The internal address comes first; an address not known yet is left out. A choice that is not
|
|
one of LISTEN_CHOICES is DEFAULT_LISTEN. LOOPBACK is not among them; web_listen adds it.
|
|
"""
|
|
if choice not in LISTEN_CHOICES:
|
|
choice = DEFAULT_LISTEN
|
|
general, interfaces, local = (_section(configuration, name) for name in ('general', 'interfaces', 'local'))
|
|
mode = str(general.get('mode') or 'manual')
|
|
addresses = []
|
|
if choice in ('internal', 'both'):
|
|
addresses.append(internal_address(mode, str(interfaces.get('internal') or ''), str(local.get('int_addr') or '')))
|
|
if choice in ('external', 'both'):
|
|
addresses.append(external_address(mode, str(interfaces.get('external') or '')))
|
|
return [address for address in addresses if address]
|
|
|
|
|
|
def web_listen(configuration, default_port):
|
|
"""Where the web service listens, from vpn-router.json read as configuration, as ([(address, port), ...], '').
|
|
|
|
The addresses are those of the interfaces web.listen names (listen_addresses), then
|
|
LOOPBACK, the port web.port, else default_port. While web.enabled is false or no address of
|
|
an interface is known, the answer is ([], the reason).
|
|
"""
|
|
web = _section(configuration, 'web')
|
|
if web.get('enabled', True) is False:
|
|
return [], 'the web service is disabled in the configuration'
|
|
port = web.get('port', default_port)
|
|
if isinstance(port, bool) or not isinstance(port, int) or not 1 <= port <= 65535:
|
|
port = default_port
|
|
addresses = listen_addresses(configuration, web.get('listen', DEFAULT_LISTEN))
|
|
if not addresses:
|
|
return [], 'the addresses to listen on are not known yet'
|
|
return [(address, port) for address in [*addresses, LOOPBACK]], ''
|