Merge pull request 'feat: port forwards are a firewall reader too, and only the WAN's' (#2) from feature/port-forwards-shared into main

get_port_forwards was declared on ResidentialGatewayDriver alone, as if a
port forward were a home-router feature. A firewall forwards ports just the
same (OPNsense calls it destination NAT), and netOrk asks both: is this host
reachable from the internet, which CVEs are exposed. The declaration moves to
NatVpnMixin, where the two roles already overlap, and PortForwardDict next to
NATTranslationDict.

The contract now says what counts. Destination NAT between internal networks
and rules that only exempt traffic are not port forwards: callers read every
entry as "reachable from outside". "ANY" forwards every protocol and an
external port of 0 every port -- a whole host forwarded is the most exposed
case and must not fall out for lack of a port number.

Declaration only, under TYPE_CHECKING: nothing changes at runtime.

README.md

@@ -227,6 +227,11 @@ class PfSenseDriver(FirewallDriver):
     def get_vpn_tunnels(self):
         # return Dict[str, VPNTunnelDict]
         ...
+
+    def get_port_forwards(self):
+        # return List[PortForwardDict] — forwards from the WAN only, never a
+        # redirect between internal networks (shared with home gateways)
+        ...
 ```
 
 ### Hypervisor

napalm_device_types/__init__.py

@@ -63,6 +63,7 @@ from napalm_device_types.firewall_rules import FirewallRuleMixin
 from napalm_device_types.health_metrics import HealthMetricsMixin
 from napalm_device_types.host_reboot import HostRebootMixin
 from napalm_device_types.interface_filter import InterfaceFilterMixin
+from napalm_device_types.lag import add_lag_interfaces
 from napalm_device_types.mac_acl import MacAclMixin
 from napalm_device_types.media import MediaDriver
 from napalm_device_types.nat_vpn import NatVpnMixin
@@ -101,6 +102,7 @@ __all__ = [
     "StorageDriver",
     "SwitchDriver",
     "UpdateMixin",
+    "add_lag_interfaces",
     "driver_supports_ping",
     "normalize_cidr",
     "normalize_mac",

napalm_device_types/lag.py

@@ -0,0 +1,57 @@
+# -*- coding: utf-8 -*-
+"""Logical LAG entries for ``get_interfaces()``, built from their member ports.
+
+Some switches list only physical ports, each tagged with the trunk it belongs
+to, and never the trunk itself. Turning those tags into one row per trunk is
+the same for every vendor, so it lives here once; a driver only has to set
+``trunk_group`` on member ports and, where the device says so, pass the mode.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Dict, List, Optional
+
+
+def _port_order(name: str) -> List[Any]:
+    return [int(p) if p.isdigit() else p for p in re.split(r"(\d+)", name)]
+
+
+def add_lag_interfaces(
+    interfaces: Dict[str, Dict[str, Any]],
+    lag_modes: Optional[Dict[str, str]] = None,
+) -> Dict[str, Dict[str, Any]]:
+    """Return *interfaces* plus one logical entry per ``trunk_group``.
+
+    The LAG entry is up/enabled if any member is, its speed is the members'
+    sum, and ``lag_members`` lists them in port order. A LAG the driver
+    already reported is left as it is. *interfaces* itself is not modified.
+
+    :param lag_modes: ``{lag_name: "lacp" | "trunk"}``. A LAG without a known
+        mode gets no ``lag_mode`` key rather than a guessed one.
+    """
+    result = dict(interfaces)
+    groups: Dict[str, List[str]] = {}
+    for name, iface in interfaces.items():
+        group = iface.get("trunk_group")
+        if group:
+            groups.setdefault(group, []).append(name)
+
+    for group, members in groups.items():
+        if group in result:
+            continue
+        members = sorted(members, key=_port_order)
+        lag: Dict[str, Any] = {
+            "is_up": any(interfaces[m].get("is_up") for m in members),
+            "is_enabled": any(interfaces[m].get("is_enabled") for m in members),
+            "description": f"LAG ({', '.join(members)})",
+            "last_flapped": -1.0,
+            "speed": sum(float(interfaces[m].get("speed") or 0) for m in members),
+            "mtu": -1,
+            "mac_address": "",
+            "lag_members": members,
+        }
+        if lag_modes and group in lag_modes:
+            lag["lag_mode"] = lag_modes[group]
+        result[group] = lag
+    return result

napalm_device_types/models.py

@@ -298,6 +298,21 @@ class NATTranslationDict(TypedDict):
     age: float
 
 
+class PortForwardDict(TypedDict):
+    """A port the WAN side can reach, forwarded to a host inside.
+
+    Shared by firewalls and home gateways (``NatVpnMixin.get_port_forwards``).
+    """
+
+    name: str
+    protocol: str  # "TCP" or "UDP"
+    external_port: int
+    internal_ip: str
+    internal_port: int
+    enabled: bool
+    remote_host: NotRequired[str]  # restrict forward to a specific remote source
+
+
 class SecurityZoneDict(TypedDict):
     interfaces: List[str]
     policy: str
@@ -470,16 +485,6 @@ class WANStatusDict(TypedDict):
     link_status: NotRequired[str]  # physical line state, e.g. "Up" / "Down"
 
 
-class PortForwardDict(TypedDict):
-    name: str
-    protocol: str  # "TCP" or "UDP"
-    external_port: int
-    internal_ip: str
-    internal_port: int
-    enabled: bool
-    remote_host: NotRequired[str]  # restrict forward to a specific remote source
-
-
 class HostDict(TypedDict):
     mac: str
     ip: str

napalm_device_types/nat_vpn.py

@@ -1,8 +1,8 @@
 # -*- coding: utf-8 -*-
 """Address translation and VPN tunnels.
 
-A home gateway does a subset of what a firewall does, and these two readers
-are where the sets overlap exactly.
+A home gateway does a subset of what a firewall does, and these readers are
+where the sets overlap exactly.
 
 Declared under ``if TYPE_CHECKING``: these are contracts, not placeholders.
 Nothing exists at runtime until a concrete driver implements it, so mixing
@@ -13,7 +13,7 @@ from __future__ import annotations
 
 from typing import Dict, List, TYPE_CHECKING
 
-from napalm_device_types.models import NATTranslationDict, VPNTunnelDict
+from napalm_device_types.models import NATTranslationDict, PortForwardDict, VPNTunnelDict
 
 
 class NatVpnMixin:
@@ -47,6 +47,46 @@ class NatVpnMixin:
             """
             ...
 
+        def get_port_forwards(self) -> List[PortForwardDict]:
+            """
+            Returns the port forwards that let traffic in from the WAN.
+
+            A port forward here means destination NAT on an interface facing
+            the internet: whoever reaches the external port is let through to
+            ``internal_ip``. A redirect between internal networks is
+            destination NAT as well, but it is **not** a port forward and must
+            be left out -- callers read every entry as "this host is reachable
+            from outside". So are rules that only exempt traffic from
+            redirection.
+
+            Each entry contains:
+
+            * name (string) - the rule's description/name
+            * protocol (string) - ``"TCP"`` or ``"UDP"``; a rule for both is
+              two entries. ``"ANY"`` forwards every protocol
+            * external_port (int) - the WAN-side port; the first of a range,
+              ``0`` for every port (a whole host forwarded)
+            * internal_ip (string) - the host the traffic is forwarded to
+            * internal_port (int) - the port on that host
+            * enabled (bool) - whether the rule is currently active
+            * remote_host (string, optional) - restricts the forward to a specific
+              remote source address; empty/absent means "any"
+
+            Example::
+
+                [
+                    {
+                        "name": "Webserver HTTPS",
+                        "protocol": "TCP",
+                        "external_port": 443,
+                        "internal_ip": "192.168.1.10",
+                        "internal_port": 443,
+                        "enabled": True,
+                    }
+                ]
+            """
+            ...
+
         def get_vpn_tunnels(self) -> Dict[str, VPNTunnelDict]:
             """
             Returns the status of VPN tunnels.

napalm_device_types/residential_gateway.py

@@ -6,8 +6,9 @@ wireless access point in a single consumer device (e.g. AVM FritzBox,
 ISP-supplied DSL/cable routers). This base class merges the relevant
 subsets of :class:`~napalm_device_types.firewall.FirewallDriver` and
 :class:`~napalm_device_types.access_point.AccessPointDriver` plus
-gateway-specific operations (WAN status, port forwarding, connected
-hosts).
+gateway-specific operations (WAN status, connected hosts). Port
+forwarding is shared with firewalls, in
+:class:`~napalm_device_types.nat_vpn.NatVpnMixin`.
 
 Usage::
 
@@ -25,7 +26,6 @@ from napalm_device_types.health_metrics import HealthMetricsMixin
 from napalm_device_types.dhcp import DhcpServerMixin
 from napalm_device_types.models import (
     HostDict,
-    PortForwardDict,
     RadioStatusDict,
     SSIDDict,
     WANStatusDict,
@@ -85,36 +85,6 @@ class ResidentialGatewayDriver(NatVpnMixin, HealthMetricsMixin, DhcpServerMixin,
             """
             ...
 
-        def get_port_forwards(self) -> List[PortForwardDict]:
-            """
-            Returns the configured port forwarding (port mapping) rules.
-
-            Each entry contains:
-
-            * name (string) - the rule's description/name
-            * protocol (string) - ``"TCP"`` or ``"UDP"``
-            * external_port (int) - the WAN-side port
-            * internal_ip (string) - the LAN host the traffic is forwarded to
-            * internal_port (int) - the LAN-side port
-            * enabled (bool) - whether the rule is currently active
-            * remote_host (string, optional) - restricts the forward to a specific
-              remote source address; empty/absent means "any"
-
-            Example::
-
-                [
-                    {
-                        "name": "Webserver HTTPS",
-                        "protocol": "TCP",
-                        "external_port": 443,
-                        "internal_ip": "192.168.1.10",
-                        "internal_port": 443,
-                        "enabled": True,
-                    }
-                ]
-            """
-            ...
-
         def get_hosts(self) -> List[HostDict]:
             """
             Returns the list of hosts known to the gateway (LAN clients).

tests/test_lag_interfaces.py

@@ -0,0 +1,76 @@
+"""Tests for add_lag_interfaces — one logical row per trunk group."""
+
+from napalm_device_types import add_lag_interfaces
+
+
+def _port(is_up: bool = True, is_enabled: bool = True, speed: float = 1000.0, trunk_group: str = "") -> dict:
+    port = {
+        "is_up": is_up,
+        "is_enabled": is_enabled,
+        "description": "",
+        "last_flapped": -1.0,
+        "speed": speed,
+        "mtu": -1,
+        "mac_address": "",
+    }
+    if trunk_group:
+        port["trunk_group"] = trunk_group
+    return port
+
+
+def test_adds_one_row_per_trunk_group():
+    ifaces = {
+        "1": _port(),
+        "3": _port(trunk_group="Trk3"),
+        "4": _port(trunk_group="Trk3"),
+        "10": _port(trunk_group="Trk6"),
+        "7": _port(trunk_group="Trk6"),
+    }
+    result = add_lag_interfaces(ifaces)
+
+    assert result["Trk3"]["lag_members"] == ["3", "4"]
+    # Members in natural port order, not string order ("7" before "10").
+    assert result["Trk6"]["lag_members"] == ["7", "10"]
+    assert result["Trk6"]["description"] == "LAG (7, 10)"
+    assert "Trk1" not in result
+
+
+def test_state_is_derived_from_members():
+    ifaces = {
+        "3": _port(is_up=False, speed=1000.0, trunk_group="Trk3"),
+        "4": _port(is_up=True, speed=1000.0, trunk_group="Trk3"),
+        "6": _port(is_up=False, is_enabled=False, trunk_group="Trk6"),
+    }
+    result = add_lag_interfaces(ifaces)
+
+    assert result["Trk3"]["is_up"] is True
+    assert result["Trk3"]["is_enabled"] is True
+    assert result["Trk3"]["speed"] == 2000.0
+    assert result["Trk6"]["is_up"] is False
+    assert result["Trk6"]["is_enabled"] is False
+
+
+def test_lag_mode_only_when_known():
+    """The UI reads a missing mode as "static trunk"; guessing would mislabel LACP."""
+    ifaces = {"3": _port(trunk_group="Trk3"), "6": _port(trunk_group="Trk6")}
+    result = add_lag_interfaces(ifaces, lag_modes={"Trk3": "lacp"})
+
+    assert result["Trk3"]["lag_mode"] == "lacp"
+    assert "lag_mode" not in result["Trk6"]
+
+
+def test_keeps_a_lag_the_driver_already_reported():
+    ifaces = {
+        "3": _port(trunk_group="Trk3"),
+        "Trk3": {**_port(), "description": "uplink", "lag_members": ["3"]},
+    }
+    result = add_lag_interfaces(ifaces)
+
+    assert result["Trk3"]["description"] == "uplink"
+
+
+def test_does_not_modify_its_input():
+    ifaces = {"3": _port(trunk_group="Trk3")}
+    add_lag_interfaces(ifaces)
+
+    assert list(ifaces) == ["3"]

tests/test_port_forwards.py

@@ -0,0 +1,39 @@
+"""get_port_forwards: what the WAN side may reach inside, on any gateway.
+
+The reader used to be declared on ``ResidentialGatewayDriver`` only, as if a
+port forward were a home-router feature. A firewall forwards ports just the
+same -- OPNsense calls it destination NAT -- and the two consumers that ask
+(is this host reachable from the internet, which CVEs are exposed) need the
+answer from both. The declaration therefore lives where the two roles overlap,
+next to the NAT translations reader.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+from napalm_device_types import FirewallDriver, ResidentialGatewayDriver
+from napalm_device_types.nat_vpn import NatVpnMixin
+
+
+def test_a_firewall_and_a_gateway_share_the_declaration():
+    assert issubclass(FirewallDriver, NatVpnMixin)
+    assert issubclass(ResidentialGatewayDriver, NatVpnMixin)
+    assert "def get_port_forwards(self) -> List[PortForwardDict]" in inspect.getsource(NatVpnMixin)
+
+
+def test_it_is_declared_once():
+    assert "def get_port_forwards" not in inspect.getsource(ResidentialGatewayDriver)
+
+
+def test_absent_until_a_driver_implements_it():
+    assert not hasattr(FirewallDriver, "get_port_forwards")
+    assert not hasattr(ResidentialGatewayDriver, "get_port_forwards")
+
+
+def test_the_contract_says_what_counts():
+    """A redirect between two internal networks is destination NAT too, and
+    would make an internal host look reachable from the internet."""
+    source = inspect.getsource(NatVpnMixin)
+    assert "from the WAN" in source
+    assert "between internal networks" in source