"""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'