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