diagnostics/traffic: Add IPv6 static gateway translation support to t…

…raffic view

src/opnsense/scripts/interfaces/traffic_top.py

@@ -65,6 +65,14 @@ def local_addresses():
                 result.append(ip)
     return result
 
+def ipv6_gateways():
+    result = {}
+    sp = subprocess.run(['/usr/local/opnsense/scripts/routes/ipv6_gateways.php'], capture_output=True, text=True)
+    for line in sp.stdout.strip().split('\n'):
+        subnet, name = line.split('\t')
+        result[subnet] = name
+    return result
+
 def from_bformat(value):
     value = value.lower()
     if value.endswith('kb'):
@@ -85,6 +93,7 @@ def to_bformat(value):
 class AsyncLookup:
     def __init__(self):
         self._results = {}
+        self.all_ipv6_gateways = ipv6_gateways()
 
     async def request_ittr(self, addresses):
         self._results = {}
@@ -94,33 +103,48 @@ async def request_ittr(self, addresses):
             return
         dnsResolver.timeout = 2
         tasks = []
+        address_mapping = {}
+
         for address in addresses:
             try:
                 if ":" in address:
                     ipv6_obj = ipaddress.IPv6Address(address)
-                    reverse_ip = '.'.join(reversed(ipv6_obj.exploded.replace(":", ""))) + ".ip6.arpa"
+                    matched_gateway = self.get_gateway_for_ipv6(ipv6_obj)
+                    if matched_gateway:
+                        last_part = address.split(":")[-1]
+                        self._results[address] = f"{last_part}@{matched_gateway}"
+                    else:
+                        reverse_ip = '.'.join(reversed(ipv6_obj.exploded.replace(":", ""))) + ".ip6.arpa"
+                        tasks.append(dnsResolver.resolve(reverse_ip, "PTR"))
+                        address_mapping[reverse_ip] = address
                 else:
                     reverse_ip = '.'.join(reversed(address.split("."))) + ".in-addr.arpa"
+                    tasks.append(dnsResolver.resolve(reverse_ip, "PTR"))
+                    address_mapping[reverse_ip] = address
 
-                tasks.append(dnsResolver.resolve(reverse_ip, "PTR"))
-            except Exception as e:
-                return
+            except Exception:
+                continue
 
         responses = await asyncio.gather(*tasks, return_exceptions=True)
 
-        for response, address in zip(responses, addresses):
+        for response, reverse_ip in zip(responses, address_mapping.keys()):
+            original_address = address_mapping[reverse_ip]
             if isinstance(response, dns.resolver.Answer):
-                if ":" in address:
-                    addr = str(ipaddress.IPv6Address(address))
-                else:
-                    addr = ".".join(reversed(response.canonical_name.to_text().replace('.in-addr.arpa.', '').split('.')))
-
                 for item in response.response.answer:
                     if isinstance(item, dns.rrset.RRset) and len(item.items) > 0:
-                        self._results[addr] = str(list(item.items)[0])
+                        self._results[original_address] = str(list(item.items)[0])
 
         return self._results
 
+    def get_gateway_for_ipv6(self, ipv6_address):
+        for subnet, gateway in self.all_ipv6_gateways.items():
+            network = ipaddress.IPv6Network(subnet)
+            if network.prefixlen == 128:
+                continue
+            if network.supernet_of(ipaddress.IPv6Network(ipv6_address.exploded + "/128")):
+                return gateway
+        return None
+
     def collect(self, addresses):
         loop = asyncio.new_event_loop()
         asyncio.set_event_loop(loop)

src/opnsense/scripts/routes/ipv6_gateways.php

@@ -0,0 +1,21 @@
+#!/usr/local/bin/php
+<?php
+
+/*
+ * Returns static IPv6 routes with gateway names in TSV format.
+ * It is used for approximate IPv6 translation of dynamic addresses in traffic view (traffic_top.py).
+ */
+require_once 'config.inc';
+require_once 'util.inc';
+
+$allRoutes = get_staticroutes();
+$allSubnets = [];
+foreach ($allRoutes as $route) {
+    if (str_contains($route['network'], ':') && is_subnet($route['network'])) {
+        $allSubnets[$route['network']] = $route['gateway'];
+    }
+}
+
+foreach ($allSubnets as $network => $gateway) {
+    echo $network . "\t" . $gateway . "\n";
+}