napalm-opnsense/tests/unit/test_port_forwards.py

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