From 3506f20606903f7cd300e78b4f61e4b7997cb384 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sat, 3 Oct 2026 16:30:38 +0200 Subject: [PATCH] feat: port forwards, read from destination NAT on the WAN get_port_forwards reads /api/firewall/d_nat/search_rule and keeps only what the contract asks for: rules on an interface with an upstream gateway (the WAN, and a second uplink as well). Internal redirects, anti-lockout rules (nordr) and rules the captive portal generates are left out -- on the first real box (OPNsense 26.7) that was 20 of 22 rules, and each would have made an internal host look reachable from the internet. Targets resolve through host/network aliases, one entry per address; an interface address or a DNS name gives no address and the rule is skipped rather than put on a guessed host. Ports resolve as numbers, the start of a range, port aliases or service names; no port is every port (0), and tcp/udp is two entries. The filtering is pure, in port_forwards.py, and the driver method does the three reads. A box without the destination-NAT API raises instead of answering "nothing forwarded", which nobody checked. --- napalm_opnsense/opnsense.py | 16 +++ napalm_opnsense/port_forwards.py | 159 ++++++++++++++++++++++ tests/unit/test_port_forwards.py | 217 +++++++++++++++++++++++++++++++ 3 files changed, 392 insertions(+) create mode 100644 napalm_opnsense/port_forwards.py create mode 100644 tests/unit/test_port_forwards.py diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index 084860f..c502d7e 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -51,6 +51,7 @@ from napalm_device_types import FingerprintRule, FirewallDriver from napalm.base.exceptions import ConnectionException, ConnectionClosedException, MergeConfigException from napalm_opnsense.ping_mixin import OPNsensePingMixin +from napalm_opnsense.port_forwards import alias_index, port_forwards, wan_interfaces class OPNsenseDriver(OPNsensePingMixin, FirewallDriver): @@ -2387,6 +2388,21 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver): return {"success": success, "output": "\n".join(lines)} + def get_port_forwards(self) -> list[dict[str, Any]]: + """Destination NAT on the WAN interfaces, as port forwards. + + Three reads: the rules, the interface overview (a WAN is an interface + with an upstream gateway) and the aliases a rule may send to. The + filtering and resolving is in :mod:`napalm_opnsense.port_forwards`. + A box without the destination-NAT API (older than its MVC rework) + raises: an empty list would claim "nothing forwarded", which nobody + checked. + """ + rules = self._get("/api/firewall/d_nat/search_rule?current=1&rowCount=-1").get("rows", []) + overview = self._get("/api/interfaces/overview/interfaces_info?current=1&rowCount=-1") + aliases = self._get("/api/firewall/alias/searchItem?current=1&rowCount=-1").get("rows", []) + return port_forwards(rules, wan_interfaces(overview), alias_index(aliases)) + def get_firewall_aliases(self) -> list[dict[str, Any]]: """Return all firewall aliases, sorted by type then name. diff --git a/napalm_opnsense/port_forwards.py b/napalm_opnsense/port_forwards.py new file mode 100644 index 0000000..635f41c --- /dev/null +++ b/napalm_opnsense/port_forwards.py @@ -0,0 +1,159 @@ +"""Destination NAT on the WAN, read as port forwards. Pure: no I/O. + +OPNsense keeps every destination-NAT rule in one list +(``/api/firewall/d_nat/search_rule``): the forwards from the internet, and +also redirects between internal networks, anti-lockout rules that only exempt +traffic (``nordr``), and rules the captive portal generates +(``is_automatic``). The contract this serves -- +``napalm_device_types.NatVpnMixin.get_port_forwards`` -- wants the first kind +only, because callers read every entry as "this host is reachable from +outside". + +What counts as the WAN is an interface with an upstream gateway, read from +the interface overview. That catches a second uplink (an LTE backup) as well +as the one named ``wan``. +""" + +from __future__ import annotations + +import ipaddress +import socket +from typing import Any, Dict, Iterable, List, Optional, Tuple + +#: Port names OPNsense accepts in a rule. ``socket.getservbyname`` knows them +#: too, but only where ``/etc/services`` exists -- a slim container has none. +WELL_KNOWN_PORTS: Dict[str, int] = { + "ftp": 21, + "ssh": 22, + "telnet": 23, + "smtp": 25, + "domain": 53, + "dns": 53, + "http": 80, + "pop3": 110, + "ntp": 123, + "imap": 143, + "snmp": 161, + "ldap": 389, + "https": 443, + "smtps": 465, + "submission": 587, + "ldaps": 636, + "imaps": 993, + "pop3s": 995, + "openvpn": 1194, + "ms-wbt-server": 3389, + "rdp": 3389, +} + +#: Alias types whose entries can be addresses. +_ADDRESS_ALIASES = ("host", "network") + +AliasIndex = Dict[str, Tuple[str, List[str]]] + + +def wan_interfaces(overview: Any) -> set: + """Identifiers of the interfaces that have an upstream gateway.""" + rows = overview.get("rows", []) if isinstance(overview, dict) else overview or [] + return {r["identifier"] for r in rows if r.get("identifier") and r.get("gateways")} + + +def alias_index(rows: Iterable[dict]) -> AliasIndex: + """``name -> (type, entries)`` from the alias list.""" + return { + r["name"]: (r.get("type", ""), [e.strip() for e in str(r.get("content", "")).splitlines() if e.strip()]) + for r in rows + if r.get("name") + } + + +def _as_address(value: str) -> Optional[str]: + try: + return str(ipaddress.ip_address(value)) + except ValueError: + return None + + +def _addresses(target: str, aliases: AliasIndex) -> List[str]: + """The addresses a rule sends to: a literal, or a host/network alias's. + + Anything else -- an interface address, a name resolved by DNS -- gives no + address, and the rule is left out rather than put on a guessed host. + """ + literal = _as_address(target) + if literal: + return [literal] + kind, entries = aliases.get(target, ("", [])) + if kind not in _ADDRESS_ALIASES: + return [] + return [a for a in (_as_address(e) for e in entries) if a] + + +def _port(value: Any, protocol: str, aliases: AliasIndex) -> Optional[int]: + """A rule's port as a number: literal, start of a range, alias or name. + + No port at all means every port, which the contract writes as 0. + """ + text = str(value).strip() + if not text: + return 0 + first = text.replace(":", "-").split("-")[0].strip() + if first.isdigit(): + return int(first) + kind, entries = aliases.get(text, ("", [])) + if kind == "port" and entries: + return _port(entries[0], protocol, aliases) + try: + return socket.getservbyname(text, protocol) + except OSError: + return WELL_KNOWN_PORTS.get(text.lower()) + + +def _faces_the_wan(rule: dict, wan: set) -> bool: + if rule.get("nordr") == "1" or rule.get("is_automatic"): + return False + return bool(set(str(rule.get("interface", "")).split(",")) & wan) + + +def _remote_host(rule: dict) -> Optional[str]: + """A source restriction; an inverted one ("all but X") restricts nothing.""" + source = str(rule.get("source.network") or "").strip() + if source in ("", "any") or rule.get("source.not") == "1": + return None + return source + + +def _forwards_of(rule: dict, aliases: AliasIndex) -> List[dict]: + protocols = [p.upper() for p in str(rule.get("protocol") or "any").split("/") if p] + target = str(rule.get("target", "")).strip() + remote = _remote_host(rule) + result: List[dict] = [] + for protocol in protocols: + external = _port(rule.get("destination.port", ""), protocol.lower(), aliases) + if external is None: + continue + local = rule.get("local-port") + internal = _port(local, protocol.lower(), aliases) if str(local or "").strip() else external + name = rule.get("descr") or f"{protocol} {external} -> {target}" + for address in _addresses(target, aliases): + entry = { + "name": name, + "protocol": protocol, + "external_port": external, + "internal_ip": address, + "internal_port": internal if internal is not None else external, + "enabled": rule.get("disabled") != "1", + } + if remote: + entry["remote_host"] = remote + result.append(entry) + return result + + +def port_forwards(rules: Iterable[dict], wan: set, aliases: AliasIndex) -> List[dict]: + """The destination-NAT rules on a WAN interface, as ``PortForwardDict`` entries.""" + result: List[dict] = [] + for rule in rules: + if _faces_the_wan(rule, wan): + result.extend(_forwards_of(rule, aliases)) + return result diff --git a/tests/unit/test_port_forwards.py b/tests/unit/test_port_forwards.py new file mode 100644 index 0000000..20e07c9 --- /dev/null +++ b/tests/unit/test_port_forwards.py @@ -0,0 +1,217 @@ +"""Destination NAT on the WAN, read as port forwards. + +OPNsense lists every destination-NAT rule in one place: the forwards from the +internet, but also redirects between internal networks, anti-lockout rules +that only exempt traffic, and rules the captive portal generates. The +contract (``NatVpnMixin.get_port_forwards``) wants the first kind only: a +caller reads each entry as "this host is reachable from outside". On the +first real box (OPNsense 26.7, 2026-10-03) that was 2 of 22 rules. + +The row shapes below follow that box's ``/api/firewall/d_nat/search_rule``, +``/api/interfaces/overview/interfaces_info`` and alias list, with the +addresses replaced. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from napalm_opnsense.opnsense import OPNsenseDriver +from napalm_opnsense.port_forwards import alias_index, port_forwards, wan_interfaces + +INTERFACES = [ + {"identifier": "lan", "description": "MGMT", "gateways": []}, + {"identifier": "wan", "description": "WAN", "gateways": ["192.0.2.1"]}, + {"identifier": "opt6", "description": "WAN4G", "gateways": ["198.51.100.1"]}, + {"identifier": "opt9", "description": "HOMEOFFICE", "gateways": []}, + {"identifier": "", "description": "Unassigned Interface"}, +] + +ALIASES = [ + {"name": "proxy_01", "type": "host", "content": "172.22.50.2"}, + {"name": "web_pair", "type": "host", "content": "10.0.0.5\n10.0.0.6"}, + {"name": "by_name", "type": "host", "content": "web.example.com"}, + {"name": "postgres", "type": "port", "content": "5432"}, +] + + +def _rule(**over) -> dict: + """One row as the API returns it; booleans are "0"/"1" strings.""" + row = { + "uuid": "u", + "disabled": "0", + "nordr": "0", + "interface": "wan", + "ipprotocol": "inet", + "protocol": "tcp", + "source.network": "any", + "source.not": "0", + "destination.network": "wanip", + "destination.not": "0", + "destination.port": "443", + "target": "proxy_01", + "local-port": "", + "descr": "Reverse proxy HTTPS", + } + row.update(over) + return row + + +def _read(*rows: dict) -> list[dict]: + return port_forwards(list(rows), wan_interfaces(INTERFACES), alias_index(ALIASES)) + + +class TestWhichInterfacesFaceTheInternet: + def test_an_interface_with_an_upstream_gateway(self): + assert wan_interfaces(INTERFACES) == {"wan", "opt6"} + + def test_the_overview_may_come_wrapped_in_rows(self): + assert wan_interfaces({"rows": INTERFACES}) == {"wan", "opt6"} + + +class TestWhatCounts: + def test_a_forward_on_the_wan(self): + assert _read(_rule()) == [ + { + "name": "Reverse proxy HTTPS", + "protocol": "TCP", + "external_port": 443, + "internal_ip": "172.22.50.2", + "internal_port": 443, + "enabled": True, + } + ] + + def test_a_second_wan_counts_too(self): + assert len(_read(_rule(interface="opt6"))) == 1 + + def test_a_rule_on_several_interfaces_counts_when_one_is_a_wan(self): + assert len(_read(_rule(interface="opt9,wan"))) == 1 + + def test_a_redirect_between_internal_networks_does_not(self): + """gw: HTTPS from HOMEOFFICE to a NAS name, sent to the proxy.""" + assert _read(_rule(interface="opt9", **{"destination.network": "HOST_NAS"})) == [] + + def test_an_exemption_from_redirection_does_not(self): + """The anti-lockout rules: ``nordr`` means "do not redirect".""" + assert _read(_rule(nordr="1")) == [] + + def test_a_generated_rule_does_not(self): + assert _read(_rule(is_automatic=True)) == [] + + def test_a_disabled_forward_is_listed_as_disabled(self): + (entry,) = _read(_rule(disabled="1")) + assert entry["enabled"] is False + + +class TestWhereItGoes: + def test_a_literal_address(self): + (entry,) = _read(_rule(target="10.0.0.9")) + assert entry["internal_ip"] == "10.0.0.9" + + def test_an_alias_with_two_hosts_is_two_forwards(self): + assert [e["internal_ip"] for e in _read(_rule(target="web_pair"))] == ["10.0.0.5", "10.0.0.6"] + + def test_an_alias_that_names_a_host_by_dns_is_skipped(self): + """No address to put the forward on; guessing one would be worse.""" + assert _read(_rule(target="by_name")) == [] + + def test_an_interface_address_is_skipped(self): + assert _read(_rule(target="opt5ip")) == [] + + def test_an_ipv6_target(self): + (entry,) = _read(_rule(ipprotocol="inet6", target="2001:db8::5")) + assert entry["internal_ip"] == "2001:db8::5" + + +class TestPorts: + @pytest.mark.parametrize( + ("value", "expected"), + [("443", 443), ("https", 443), ("http", 80), ("postgres", 5432), ("8000-8010", 8000), ("8000:8010", 8000)], + ) + def test_the_external_port(self, value, expected): + (entry,) = _read(_rule(**{"destination.port": value})) + assert entry["external_port"] == expected + + def test_the_internal_port_defaults_to_the_external_one(self): + (entry,) = _read(_rule(**{"destination.port": "80"})) + assert entry["internal_port"] == 80 + + def test_a_different_internal_port_may_come_as_a_number(self): + (entry,) = _read(_rule(**{"destination.port": "80", "local-port": 9000})) + assert entry["internal_port"] == 9000 + + def test_an_unknown_port_name_skips_the_rule(self): + assert _read(_rule(**{"destination.port": "no-such-service"})) == [] + + def test_a_whole_host_forwarded_is_kept(self): + """The most exposed case of all: every protocol, every port.""" + (entry,) = _read(_rule(protocol="any", **{"destination.port": ""})) + assert (entry["protocol"], entry["external_port"], entry["internal_port"]) == ("ANY", 0, 0) + + def test_every_port_of_one_protocol(self): + (entry,) = _read(_rule(**{"destination.port": ""})) + assert (entry["protocol"], entry["external_port"]) == ("TCP", 0) + + def test_tcp_and_udp_are_two_forwards(self): + assert [e["protocol"] for e in _read(_rule(protocol="tcp/udp"))] == ["TCP", "UDP"] + + +class TestNameAndSource: + def test_without_a_description_the_rule_is_named_after_what_it_does(self): + (entry,) = _read(_rule(descr="")) + assert entry["name"] == "TCP 443 -> proxy_01" + + def test_a_source_restriction_is_the_remote_host(self): + (entry,) = _read(_rule(**{"source.network": "203.0.113.7"})) + assert entry["remote_host"] == "203.0.113.7" + + def test_any_source_has_no_remote_host(self): + (entry,) = _read(_rule()) + assert "remote_host" not in entry + + def test_an_inverted_source_has_no_remote_host(self): + """"Everyone but X" is as good as anyone for whether it is reachable.""" + (entry,) = _read(_rule(**{"source.network": "203.0.113.7", "source.not": "1"})) + assert "remote_host" not in entry + + +class TestTheDriverMethod: + @pytest.fixture + def driver(self): + with patch("napalm_opnsense.opnsense.requests.Session"): + drv = OPNsenseDriver( + hostname="opnsense.example.com", + username="key", + password="secret", + optional_args={"verify": False}, + ) + drv.session = MagicMock() + yield drv + + def test_it_reads_rules_interfaces_and_aliases(self, driver): + seen = [] + + def fake_get(path): + seen.append(path.split("?")[0]) + if path.startswith("/api/firewall/d_nat/search_rule"): + return {"rows": [_rule()]} + if path.startswith("/api/interfaces/overview/interfaces_info"): + return {"rows": INTERFACES} + if path.startswith("/api/firewall/alias/searchItem"): + return {"rows": ALIASES} + raise AssertionError(path) + + driver._get = fake_get + assert [e["internal_ip"] for e in driver.get_port_forwards()] == ["172.22.50.2"] + assert seen == [ + "/api/firewall/d_nat/search_rule", + "/api/interfaces/overview/interfaces_info", + "/api/firewall/alias/searchItem", + ] + + def test_netork_can_tell_it_is_there(self): + """netOrk asks ``hasattr`` before it calls.""" + assert hasattr(OPNsenseDriver, "get_port_forwards")