fix: replace webservice swanctl call human readble output with structured one.

debian-package/src/usr/lib/vpn-router/web

@@ -29,7 +29,6 @@ import hashlib
 import hmac
 import json
 import os
-import re
 import signal
 import subprocess
 import sys
@@ -49,7 +48,6 @@ LEEWAY = 60
 MAX_LIFETIME = 600
 
 TAILWIND = 'https://cdn.jsdelivr.net/npm/@tailwindcss/browser@4'
-SA_LINE = re.compile(r'^(\S+): #\d+, (\w+),')
 
 _cache = {'at': 0.0, 'data': None}
 _lock = threading.Lock()
@@ -85,13 +83,51 @@ def docker_status():
     }
 
 
-def established(connection):
-    """The number of IKE SAs of the connection, or None when swanctl cannot be asked."""
-    result = run('swanctl', '--list-sas')
+def swanctl(command):
+    """The entries of a swanctl list command, or None when swanctl cannot be asked.
+
+    The --pretty output has one item per line: "name {" opens a section, "key = [" a list
+    whose items follow one per line, "}" and "]" close them, and "key = value" is a pair
+    whose value may contain spaces. Every "... event {" section at the top holds one
+    connection or SA; each becomes an entry with its name under "name". The closing
+    "... reply {" section is left out. Values stay strings, as swanctl prints them.
+    """
+    result = run('swanctl', command, '--pretty')
     if result is None or result.returncode != 0:
         return None
-    return sum(1 for line in result.stdout.splitlines()
-               if (match := SA_LINE.match(line)) and match.group(1) == connection and match.group(2) == 'ESTABLISHED')
+    events = []
+    stack = []
+    for line in result.stdout.splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        if line in ('}', ']'):
+            stack.pop()
+        elif not stack:
+            section = {}
+            if line.endswith(' event {'):
+                events.append(section)
+            stack.append(section)
+        elif isinstance(stack[-1], list):
+            stack[-1].append(line)
+        elif line.endswith(' {'):
+            stack[-1][line[:-2]] = section = {}
+            stack.append(section)
+        elif line.endswith(' = ['):
+            stack[-1][line[:-4]] = items = []
+            stack.append(items)
+        else:
+            key, _, value = line.partition(' = ')
+            stack[-1][key] = value
+    return [dict(body, name=name) for event in events for name, body in event.items()]
+
+
+def established(connection):
+    """The number of established IKE SAs of the connection, or None when swanctl cannot be asked."""
+    sas = swanctl('--list-sas')
+    if sas is None:
+        return None
+    return sum(1 for sa in sas if sa['name'] == connection and sa.get('state') == 'ESTABLISHED')
 
 
 def wireguard_peers():
@@ -176,6 +212,17 @@ def lines_of(*args):
     return result.stdout.splitlines() if result and result.returncode == 0 else None
 
 
+def routes():
+    """The IPv4 routes as ip -j prints them, or None when ip fails."""
+    result = run('ip', '-j', '-4', 'route', 'show')
+    if result is None or result.returncode != 0:
+        return None
+    try:
+        return json.loads(result.stdout)
+    except ValueError:
+        return None
+
+
 def wireguard_details():
     """The interface and the peers of wg0, with endpoints and allowed IPs; no private or preshared key."""
     result = run('wg', 'show', 'wg0', 'dump')
@@ -202,10 +249,10 @@ def details(state):
     configuration = state.get('configuration', {})
     return {
         'configuration': configuration,
-        'network': dict(state.get('network', {}), routes=lines_of('ip', '-4', 'route', 'show')),
+        'network': dict(state.get('network', {}), routes=routes()),
         'ipsec': {
-            'connections': lines_of('swanctl', '--list-conns'),
-            'sas': lines_of('swanctl', '--list-sas'),
+            'connections': swanctl('--list-conns'),
+            'sas': swanctl('--list-sas'),
         },
         'wireguard': wireguard_details(),
         'docker': {