enroll_fw.py

import argparse
import logging
import os
from pathlib import Path

from ironicclient.common.apiclient import exceptions as ironic_exceptions

from understack_workflows import firewall
from understack_workflows import helpers
from understack_workflows import ironic_node
from understack_workflows import netdev_reconciler
from understack_workflows.ironic.client import IronicClient

logger = logging.getLogger(__name__)

DEFAULT_FIREWALL_DRIVER = "panos"


def main() -> None:
    """Enroll a firewall, or update its metadata in place.

    Uses the Palo Alto hardware type by default, with management access in
    the node's driver_info and the mate serial in extra.

    """
    helpers.setup_logger()
    args = argument_parser().parse_args()

    node_uuid = enroll_fw(
        name=args.name,
        physical_network=args.physical_network,
        ports=netdev_reconciler.parse_ports_arg(args.ports),
        resource_class=args.resource_class,
        driver=args.driver,
        external_cmdb_id=args.external_cmdb_id,
        management_ip=args.management_ip,
        management_switch=args.management_switch,
        management_switch_port=args.management_switch_port,
        mate_serial=args.mate_serial,
        serial=args.serial,
        vendor=args.vendor,
        model=args.model,
    )

    if args.node_uuid_output:
        Path(args.node_uuid_output).write_text(node_uuid + "\n")


def _require_specific_resource_class(resource_class: str | None) -> str:
    """Return a normalized, purpose-made resource class.

    argparse(required=True) only requires the flag to be present, and the engine
    turns an empty value into the generic default.
    """
    normalized = (resource_class or "").strip()
    if not normalized or normalized.lower() == netdev_reconciler.DEFAULT_RESOURCE_CLASS:
        raise ValueError(
            "enroll-fw requires an explicit, purpose-made --resource-class; "
            f"got {resource_class!r}. The generic default is not allowed."
        )
    return normalized


def _require_management_location(
    management_switch: str, management_switch_port: str
) -> tuple[str, str]:
    """Return (switch, port) after checking both are non-empty."""
    switch = (management_switch or "").strip()
    port = (management_switch_port or "").strip()
    missing = []
    if not switch:
        missing.append("--management-switch")
    if not port:
        missing.append("--management-switch-port")
    if missing:
        raise ValueError(
            f"enroll-fw requires {' and '.join(missing)}: the management switch "
            "and port cannot be blank."
        )
    return switch, port


def enroll_fw(
    *,
    name: str,
    physical_network: str,
    ports: list[dict],
    resource_class: str,
    driver: str = DEFAULT_FIREWALL_DRIVER,
    external_cmdb_id: int | str | None = None,
    management_ip: str = "",
    management_switch: str = "",
    management_switch_port: str = "",
    mate_serial: str = "",
    serial: str = "",
    vendor: str = "",
    model: str = "",
) -> str:
    driver = (driver or "").strip()
    if not driver:
        raise ValueError("enroll-fw requires a non-empty --driver")
    if ports and not (physical_network or "").strip():
        raise ValueError("enroll-fw requires --physical-network with --ports")
    resource_class = _require_specific_resource_class(resource_class)
    management_switch, management_switch_port = _require_management_location(
        management_switch, management_switch_port
    )
    driver_info, extra, properties = firewall.firewall_metadata(
        management_ip=management_ip,
        management_switch=management_switch,
        management_switch_port=management_switch_port,
        mate_serial=mate_serial,
        serial=serial,
        vendor=vendor,
        model=model,
    )

    # Look up the node first so we can tell an in-service (active) firewall from
    # one we may (re)enroll. Only the active case is special-cased here; every
    # other state is handled by the shared engine (which rejects the states it
    # cannot enroll).
    client = IronicClient()
    node = _get_node_or_none(client, name)

    if node is not None and getattr(node, "provision_state", None) == "active":
        _update_active_firewall(
            client=client,
            node=node,
            name=name,
            physical_network=physical_network,
            ports=ports,
            resource_class=resource_class,
            driver=driver,
            external_cmdb_id=external_cmdb_id,
            driver_info=driver_info,
            extra=extra,
            properties=properties,
        )
        return node.uuid

    if not ports:
        return _enroll_by_inspection(
            client=client,
            name=name,
            resource_class=resource_class,
            driver=driver,
            external_cmdb_id=external_cmdb_id,
            driver_info=driver_info,
            extra=extra,
            properties=properties,
        )

    # New or re-enrollable node: the engine writes the metadata inside the
    # enrollment lifecycle (at create, or patched before the node becomes
    # available), so the node is never allocatable in an under-described state.
    return netdev_reconciler.enroll(
        name=name,
        physical_network=physical_network,
        ports=ports,
        resource_class=resource_class,
        driver=driver,
        external_cmdb_id=external_cmdb_id,
        driver_info=driver_info,
        extra=extra,
        properties=properties,
        assign_roles=True,
    )


def _get_node_or_none(client: IronicClient, name: str):
    """Return the node named ``name``, or None if it does not exist yet."""
    try:
        return client.get_node(name)
    except ironic_exceptions.NotFound:
        return None


def _enroll_by_inspection(
    *,
    client: IronicClient,
    name: str,
    resource_class: str,
    driver: str,
    external_cmdb_id: int | str | None,
    driver_info: dict,
    extra: dict,
    properties: dict,
) -> str:
    """Enroll a firewall without ports and let inspection find and label them.

    The node is provided only once it has an outside and an inside port, so a
    router can never adopt it unlabelled.
    """
    if external_cmdb_id not in (None, ""):
        extra = {**extra, "external_cmdb_id": external_cmdb_id}
    node, created = netdev_reconciler.find_or_create_netdev_node(
        client=client,
        name=name,
        resource_class=resource_class,
        driver=driver,
        driver_info=driver_info,
        extra=extra,
        properties=properties,
    )
    if node.provision_state != "manageable":
        ironic_node.transition(node, target_state="manage", expected_state="manageable")
    if not created:
        netdev_reconciler.update_node_metadata(
            client=client,
            node=node,
            resource_class=resource_class,
            driver_info=driver_info,
            extra=extra,
            properties=properties,
        )

    logger.info("[node:%s] Inspecting to discover and label the ports", node.uuid)
    ironic_node.transition(node, target_state="inspect", expected_state="manageable")
    found = {getattr(port, "category", None) for port in client.list_ports(node.uuid)}
    if not found.issuperset(netdev_reconciler.PORT_ROLES):
        raise RuntimeError(
            f"[node:{node.uuid}] Inspection did not give the node an outside and "
            "an inside port; leaving it manageable"
        )

    ironic_node.transition(node, target_state="provide", expected_state="available")
    return node.uuid


def _update_active_firewall(
    *,
    client: IronicClient,
    node,
    name: str,
    physical_network: str,
    ports: list[dict],
    resource_class: str,
    driver: str,
    external_cmdb_id: int | str | None,
    driver_info: dict,
    extra: dict,
    properties: dict,
) -> None:
    """Update firewall metadata on an in-service (active) node, in place.

    An active firewall is carrying live traffic, so its base configuration must
    not change here. We re-validate the submitted ports and compare them
    read-only against the node's real ports.
    If everything matches we patch only driver_info/extra/properties.
    """
    node_driver = getattr(node, "driver", None)
    if node_driver != driver:
        raise RuntimeError(
            f"Node {name} ({node.uuid}) is active with driver {node_driver!r}; "
            f"refusing to enroll it as {driver!r}"
        )

    node_resource_class = getattr(node, "resource_class", None)
    if node_resource_class != resource_class:
        raise RuntimeError(
            f"[node:{node.uuid}] Node is active with resource_class "
            f"{node_resource_class!r}; refusing to change it to "
            f"{resource_class!r} on an in-service firewall"
        )

    # Without ports there is nothing to compare: inspection owns them.
    if ports:
        _reject_structural_drift(
            client=client,
            node=node,
            name=name,
            physical_network=physical_network,
            netdev_ports=netdev_reconciler.build_netdev_ports(ports),
        )

    active_extra = dict(extra)
    if external_cmdb_id not in (None, ""):
        active_extra["external_cmdb_id"] = external_cmdb_id

    logger.info(
        "[node:%s] Node is active; ports/physical_network match the request, "
        "updating firewall metadata only",
        node.uuid,
    )
    firewall.apply_node_metadata(client, node, driver_info, active_extra, properties)


def _reject_structural_drift(
    *,
    client: IronicClient,
    node,
    name: str,
    physical_network: str,
    netdev_ports: list,
) -> None:
    """Refuse if any requested port would create/update on the node.

    Reuses the engine's read-only port planner: a non-None plan means the port
    is missing or differs (MAC, switch, interface, physical_network, ...).
    """
    node_ports = list(client.list_ports(node.uuid))
    ports_by_mac = {(p.address or "").lower(): p for p in node_ports}
    ports_by_name = {p.name: p for p in node_ports if p.name}

    for port in netdev_ports:
        plan = netdev_reconciler.plan_netdev_port(
            node=node,
            node_name=name,
            physical_network=physical_network,
            port=port,
            existing=ports_by_name.get(f"{name}:{port.label}"),
            ports_by_mac=ports_by_mac,
            # This path assigns no roles, so a live firewall's outside/inside
            # labels must not count as drift.
            manage_category=False,
        )
        if plan is not None:
            raise RuntimeError(
                f"[node:{node.uuid}] Node is active; refusing to change ports/"
                f"physical_network on an in-service firewall (port "
                f"{name}:{port.label} would be {plan['kind']}d). Reconcile the "
                "base configuration through a maintenance window, not a metadata "
                "update."
            )

    netdev_reconciler.warn_orphan_ports(node, name, node_ports, netdev_ports)


def argument_parser():
    parser = argparse.ArgumentParser(
        prog=os.path.basename(__file__),
        description="Enroll a firewall (defaults to the panos hardware type)",
    )
    parser.add_argument(
        "--node-uuid-output",
        help="Write node UUID to this file after successful enrollment",
    )
    parser.add_argument("--name", required=True, help="Ironic node name")
    parser.add_argument(
        "--physical-network",
        default="",
        help="Port physical_network. Required with --ports.",
    )
    parser.add_argument(
        "--ports",
        default="[]",
        help="JSON array of ports (same format as enroll-netdev). 'switch' must "
        "be the switch FQDN (e.g. n11-22-1.dfw3.rackspace.net) so the Nautobot "
        "sync can resolve the cable. Optional per-port 'bios_name' (the device "
        "interface name); defaults to the label. At least two ports: the "
        "lowest-numbered interface becomes the router's outside NIC (gateway) "
        "and the next one its inside NIC (subnets). Leave it out to have "
        "inspection discover the ports over LLDP and label them.",
    )
    parser.add_argument(
        "--resource-class",
        required=True,
        help="Ironic resource class (required: use a purpose-made pool, not "
        "the generic default, or the router flavor may adopt wrong hardware)",
    )
    parser.add_argument(
        "--driver",
        default=DEFAULT_FIREWALL_DRIVER,
        help=(
            "Ironic hardware type to enroll the node as "
            f"(default: {DEFAULT_FIREWALL_DRIVER}). Must match an existing "
            "node's driver."
        ),
    )
    parser.add_argument(
        "--external-cmdb-id",
        type=helpers.int_or_str,
        required=False,
        default="",
        help="CMDB ID",
    )
    parser.add_argument(
        "--management-ip",
        required=False,
        default="",
        help="Management IP -> driver_info.management_ip",
    )
    parser.add_argument(
        "--management-switch",
        required=True,
        help="Management switch name -> driver_info.management_switch",
    )
    parser.add_argument(
        "--management-switch-port",
        required=True,
        help="Management switch port -> driver_info.management_switch_port",
    )
    parser.add_argument(
        "--mate-serial",
        required=False,
        default="",
        help="HA mate serial number -> extra.mate_serial",
    )
    parser.add_argument(
        "--serial",
        required=False,
        default="",
        help="Device's own serial number -> extra.serial (Nautobot sync)",
    )
    parser.add_argument(
        "--vendor",
        required=False,
        default="Palo Alto",
        help="Device vendor -> properties.vendor (Nautobot sync)",
    )
    parser.add_argument(
        "--model",
        required=False,
        default="",
        help="Device model, e.g. PA-1410 -> properties.model (Nautobot sync)",
    )
    return parser


if __name__ == "__main__":
    main()