79 lines
4.3 KiB
Python
79 lines
4.3 KiB
Python
"""Strict, deliberately small WireGuard client configuration grammar."""
|
|
import base64
|
|
import ipaddress
|
|
import re
|
|
|
|
class ConfigError(ValueError):
|
|
pass
|
|
|
|
|
|
def parse_config(text):
|
|
if not isinstance(text, str) or len(text.encode('utf-8')) > 16384:
|
|
raise ConfigError('Die Konfiguration darf höchstens 16 KiB groß sein.')
|
|
sections = {}
|
|
current = None
|
|
for line in text.splitlines():
|
|
line = line.split('#', 1)[0].strip()
|
|
if not line:
|
|
continue
|
|
if line.startswith('['):
|
|
if line not in ('[Interface]', '[Peer]') or line in sections:
|
|
raise ConfigError('Genau ein Interface und ein Peer werden unterstützt.')
|
|
current = sections.setdefault(line, {})
|
|
continue
|
|
if current is None or '=' not in line:
|
|
raise ConfigError('Ungültiges WireGuard-Dateiformat.')
|
|
name, value = (part.strip() for part in line.split('=', 1))
|
|
allowed = {'PrivateKey', 'Address', 'DNS', 'MTU', 'ListenPort'} if current is sections.get('[Interface]') else {'PublicKey', 'PresharedKey', 'AllowedIPs', 'Endpoint', 'PersistentKeepalive'}
|
|
if name not in allowed or name in current:
|
|
raise ConfigError('Unbekannte oder doppelte Direktive. Hooks, Table und SaveConfig sind nicht erlaubt.')
|
|
if not value or any(ord(c) < 32 for c in value):
|
|
raise ConfigError('Leerer oder ungültiger Konfigurationswert.')
|
|
current[name] = value
|
|
interface, peer = sections.get('[Interface]', {}), sections.get('[Peer]', {})
|
|
if not {'PrivateKey', 'Address'} <= interface.keys() or not {'PublicKey', 'AllowedIPs', 'Endpoint'} <= peer.keys():
|
|
raise ConfigError('PrivateKey, Address, PublicKey, AllowedIPs und Endpoint sind erforderlich.')
|
|
for section, key in ((interface, 'PrivateKey'), (peer, 'PublicKey'), (peer, 'PresharedKey')):
|
|
if key in section:
|
|
try:
|
|
decoded = base64.b64decode(section[key], validate=True)
|
|
if len(decoded) != 32 or not any(decoded):
|
|
raise ValueError()
|
|
except ValueError:
|
|
raise ConfigError('Ein WireGuard-Schlüssel hat ein ungültiges Format.') from None
|
|
try:
|
|
addresses = [ipaddress.ip_interface(v.strip()) for v in interface['Address'].split(',')]
|
|
networks = [ipaddress.ip_network(v.strip(), strict=False) for v in peer['AllowedIPs'].split(',')]
|
|
if len(addresses) != 1 or addresses[0].version != 4 or any(n.version != 4 for n in networks):
|
|
raise ConfigError('Diese Version unterstützt eine IPv4-Tunneladresse und IPv4-AllowedIPs.')
|
|
if addresses[0].ip.is_loopback or addresses[0].ip.is_unspecified or addresses[0].ip.is_multicast:
|
|
raise ValueError()
|
|
if len(networks) > 32:
|
|
raise ValueError()
|
|
endpoint, port = peer['Endpoint'].rsplit(':', 1)
|
|
if not re.fullmatch(r'[A-Za-z0-9](?:[A-Za-z0-9.-]{0,251}[A-Za-z0-9])?', endpoint):
|
|
raise ValueError()
|
|
if not 1 <= int(port) <= 65535:
|
|
raise ValueError()
|
|
mtu = int(interface.get('MTU', '1420'))
|
|
keepalive = int(peer.get('PersistentKeepalive', '25'))
|
|
listen = int(interface.get('ListenPort', '0'))
|
|
if not 576 <= mtu <= 9000 or not 0 <= keepalive <= 65535 or not 0 <= listen <= 65535:
|
|
raise ValueError()
|
|
except (ValueError, KeyError) as exc:
|
|
if isinstance(exc, ConfigError):
|
|
raise
|
|
raise ConfigError('Ungültige Adresse, Endpoint, Port, MTU oder Keepalive.') from None
|
|
warnings = ['DNS wird nicht übernommen; die Container-DNS-Auflösung bleibt bestehen.'] if 'DNS' in interface else []
|
|
return dict(interface=interface, peer=peer, address=str(addresses[0].ip), allowed_ips=[str(n) for n in networks], mtu=mtu, keepalive=keepalive, listen=listen, warnings=warnings)
|
|
|
|
|
|
def wireguard_text(config):
|
|
"""Never executed as shell or wg-quick input; excludes all hooks and routes."""
|
|
lines = ['[Interface]', 'PrivateKey = ' + config['interface']['PrivateKey'], 'ListenPort = ' + str(config['listen']), '[Peer]']
|
|
for key in ('PublicKey', 'PresharedKey', 'Endpoint'):
|
|
if key in config['peer']:
|
|
lines.append(key + ' = ' + config['peer'][key])
|
|
lines.extend(['AllowedIPs = ' + ', '.join(config['allowed_ips']), 'PersistentKeepalive = ' + str(config['keepalive'])])
|
|
return '\n'.join(lines) + '\n'
|