napalm-opnsense/napalm_opnsense/port_forwards.py

"""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 collections.abc import Iterable
from typing import Any

#: 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) -> str | None:
    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) -> int | None:
    """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) -> str | None:
    """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