path: root/plugins/modules/vyos_interfaces.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)

from __future__ import absolute_import, division, print_function


__metaclass__ = type

DOCUMENTATION = r"""
---
module: vyos_interfaces
short_description: Manage interface configuration on VyOS devices via REST API.
description:
  - Manages L2 interface configuration (description, MTU, speed, duplex,
    enabled, VRF assignment, VLAN sub-interfaces) on VyOS devices using the
    HTTPS REST API.
  - IP address configuration is handled by M(vyos.rest.vyos_l3_interfaces).
  - >-
    Covers 11 interface types (ethernet, bonding, loopback, tunnel,
    wireguard, vti, dummy, openvpn, pppoe, wireless, bridge), resolved
    from the device response when present, otherwise guessed from the
    interface name. The current CLI collection module documents a
    narrower scope of 5 types (ethernet, bonding, vxlan, loopback,
    vti).
version_added: "1.0.0"
author:
  - VyOS Community (@vyos)
options:
  config:
    description: List of interface configurations.
    type: list
    elements: dict
    suboptions:
      name:
        description: Full interface name (e.g. eth0, bond0, lo).
        type: str
        required: true
      description:
        description: Interface description.
        type: str
      enabled:
        description: Whether the interface is enabled. False sets the disable flag.
        type: bool
        default: true
      mtu:
        description: Interface MTU.
        type: int
      duplex:
        description: Interface duplex setting.
        type: str
        choices: [auto, full, half]
      speed:
        description: Interface speed setting.
        type: str
        choices: [auto, "10", "100", "1000", "2500", "10000"]
      vrf:
        description: VRF instance to bind this interface to.
        type: str
      vifs:
        description: 802.1Q VLAN sub-interfaces.
        type: list
        elements: dict
        suboptions:
          vlan_id:
            description: VLAN ID for this sub-interface.
            type: int
            required: true
          description:
            description: Sub-interface description.
            type: str
          enabled:
            description: Whether the sub-interface is enabled.
            type: bool
            default: true
          mtu:
            description: Sub-interface MTU.
            type: int
  state:
    description:
      - C(merged) - Merge config with existing interface settings.
      - C(replaced) - Replace config for listed interfaces.
      - C(overridden) - Replace config for all interfaces.
      - C(deleted) - Remove listed interface config or all interface config.
      - C(gathered) - Read interface config from device without changes.
    type: str
    choices: [merged, replaced, overridden, deleted, gathered]
    default: merged
seealso:
  - module: vyos.vyos.vyos_interfaces
  - module: vyos.rest.vyos_l3_interfaces
"""

EXAMPLES = r"""
- name: Merge interface configuration
  vyos.rest.vyos_interfaces:
    config:
      - name: eth0
        description: Management interface
        mtu: 1500
        enabled: true
        vrf: mgmt
        vifs:
          - vlan_id: 200
            description: VIF 200
    state: merged

- name: Disable an interface
  vyos.rest.vyos_interfaces:
    config:
      - name: eth1
        enabled: false
    state: merged

- name: Delete interface config
  vyos.rest.vyos_interfaces:
    config:
      - name: eth0
    state: deleted

- name: Gather current interface configuration
  vyos.rest.vyos_interfaces:
    state: gathered
"""

RETURN = r"""
before:
  description: Interface configuration before this module ran.
  returned: always
  type: list
after:
  description: Interface configuration after this module ran.
  returned: when changed
  type: list
commands:
  description: List of API command tuples sent to the device.
  returned: always
  type: list
gathered:
  description: Current interface configuration as structured data.
  returned: when state is gathered
  type: list
saved:
  description: Whether the config was saved after changes.
  returned: when changes are applied
  type: bool
response:
  description: Raw API response.
  returned: when changes are applied
  type: dict
"""

from ansible.module_utils.basic import AnsibleModule
from ansible_collections.vyos.rest.plugins.module_utils.vyos import (
    VyOSModule,
    cast_by_spec,
    dict_op,
    to_tag_dict,
)


_BASE = ["interfaces"]


_IFACE_TYPE_PREFIX = {
    "eth": "ethernet",
    "bond": "bonding",
    "lo": "loopback",
    "tun": "tunnel",
    "wg": "wireguard",
    "vti": "vti",
    "dum": "dummy",
    "vtun": "openvpn",
    "ppp": "pppoe",
    "wlan": "wireless",
    "br": "bridge",
}


def _guess_iface_type(name):
    for prefix, itype in _IFACE_TYPE_PREFIX.items():
        if name.startswith(prefix):
            return itype
    return "ethernet"


def _resolve_iface_type(name, raw_have):
    """Prefer the real type from the device's own raw response
    (organized by type at the top level) over a name-prefix guess --
    only fall back to guessing for a brand-new interface that doesn't
    exist on the device yet.
    """
    for itype, ifaces in (raw_have or {}).items():
        if name in to_tag_dict(ifaces):
            return itype
    return _guess_iface_type(name)


def _iface_base(name, raw_have):
    return _BASE + [_resolve_iface_type(name, raw_have), name]


_DEVICE_RENAMES = {
    "vifs": "vif",
}


_ENABLED_FIELD = "enabled"
_DISABLE_DEVICE_KEY = "disable"


def _derive_key_field(options_spec):
    """The field identifying each entry in a keyed-list section is
    never inferable from a generic walk alone -- but it doesn't need
    to be hand-declared either: every such section in this argspec
    already marks exactly one suboption required=True (you can't
    create a VIF without a vlan_id). Deriving it here means the key
    field is asserted to exist by the argspec itself, not duplicated
    in a place that could drift out of sync with it.
    """
    required = [k for k, spec in options_spec.items() if spec.get("required")]
    if len(required) != 1:
        raise ValueError(
            "expected exactly one required suboption to serve as the key field, "
            "found: {0}".format(required),
        )
    return required[0]


def _keyed_list_to_device(items, key_field, entry_transform):
    result = {}
    for item in items or []:
        if item.get(key_field) is None:
            continue
        rest = {k: v for k, v in item.items() if k != key_field}
        result[str(item[key_field])] = entry_transform(rest)
    return result


def _keyed_list_from_device(raw, key_field, entry_transform, key_cast=None):
    key_cast = key_cast or (lambda k: k)
    return [
        {key_field: key_cast(key), **entry_transform(data or {})}
        for key, data in sorted(to_tag_dict(raw).items())
    ]


def _spec_to_device(value, options_spec):
    if not isinstance(value, dict):
        return value
    result = {}
    for arg_key, sub_spec in options_spec.items():
        if arg_key == _ENABLED_FIELD:
            if value.get(arg_key) is False:
                result[_DISABLE_DEVICE_KEY] = {}
            continue

        val = value.get(arg_key)
        if val is None or val is False:
            continue

        device_key = _DEVICE_RENAMES.get(arg_key, arg_key.replace("_", "-"))
        sub_type = sub_spec.get("type")
        sub_options = sub_spec.get("options")

        if sub_type == "dict" and sub_options:
            converted = _spec_to_device(val, sub_options)
            if converted:
                result[device_key] = converted
        elif sub_type == "list" and sub_options:
            key_field = _derive_key_field(sub_options)
            result[device_key] = _keyed_list_to_device(
                val,
                key_field,
                lambda rest, spec=sub_options: _spec_to_device(rest, spec),
            )
        elif val is True:
            result[device_key] = {}
        elif sub_type == "list":
            result[device_key] = list(val)
        else:
            result[device_key] = val
    return result


def _device_to_spec(raw, options_spec):
    if not raw or not isinstance(raw, dict):
        return {}
    have_idx = {k.replace("-", "_"): k for k in raw}
    result = {}

    if _ENABLED_FIELD in options_spec and _DISABLE_DEVICE_KEY in raw:
        result[_ENABLED_FIELD] = False

    for arg_key, sub_spec in options_spec.items():
        if arg_key == _ENABLED_FIELD:
            continue
        device_key = _DEVICE_RENAMES.get(arg_key, arg_key.replace("_", "-"))
        orig_key = device_key if device_key in raw else have_idx.get(arg_key)
        if orig_key is None:
            continue
        raw_val = raw[orig_key]
        sub_type = sub_spec.get("type")
        sub_options = sub_spec.get("options")

        if sub_type == "dict" and sub_options:
            converted = _device_to_spec(raw_val, sub_options)
            if converted:
                result[arg_key] = converted
        elif sub_type == "list" and sub_options:
            key_field = _derive_key_field(sub_options)
            key_cast = int if sub_options[key_field].get("type") == "int" else None
            entries = _keyed_list_from_device(
                raw_val,
                key_field,
                lambda d, spec=sub_options: _device_to_spec(d, spec),
                key_cast=key_cast,
            )
            if entries:
                result[arg_key] = entries
        elif sub_type == "list":
            if raw_val:
                result[arg_key] = sorted(to_tag_dict(raw_val).keys())
        elif isinstance(raw_val, dict) and not raw_val:
            result[arg_key] = True
        else:
            result[arg_key] = raw_val
    return result


def get_running_config(vyos):
    """VyOS's REST API collapses a single-child tag node to a plain
    string (or a list for multiple) -- normalizing through to_tag_dict
    unconditionally means callers always receive a genuine dict.
    """
    return to_tag_dict(vyos.get_config(_BASE) or {})


def _device_to_argspec(raw):
    result = []
    for itype, ifaces in sorted((raw or {}).items()):
        for name, data in sorted(to_tag_dict(ifaces).items()):
            entry = {"name": name}
            entry.update(_device_to_spec(data or {}, _ENTRY_OPTIONS))
            result.append(entry)
    return result


def _shadow_vif_entries(want_device, have_device):
    """Ensure want_device has a (possibly empty) placeholder for every
    VLAN ID present in have_device's own "vif" dict, so a dict_op
    purge recurses into each VIF individually rather than treating the
    whole "vif" key, or any single VLAN entry, as one unit.
    """
    have_vifs = have_device.get("vif")
    if not have_vifs:
        return want_device
    shadowed = dict(want_device)
    want_vifs = dict(shadowed.get("vif") or {})
    for vlan_id in have_vifs:
        want_vifs.setdefault(vlan_id, {})
    shadowed["vif"] = want_vifs
    return shadowed


def _purge_commands(want_device, have_device, base):
    return dict_op(_shadow_vif_entries(want_device, have_device), have_device, base, op="purge")


def _entry_to_device(entry, options_spec):
    """to-device conversion for a keyed entry whose own key field
    (e.g. "name" for an interface, same role "vlan_id" plays for a
    VIF) is present in the input but must never be treated as a
    regular child leaf -- it identifies the entry itself and is
    already expressed in the API path (_iface_base), not a field to
    set/purge under it. _keyed_list_to_device already strips a VIF's
    "vlan_id" the same way before conversion; interface entries need
    the same treatment here since build_commands handles the top
    level manually rather than through that helper.
    """
    key_field = _derive_key_field(options_spec)
    rest = {k: v for k, v in (entry or {}).items() if k != key_field}
    return _spec_to_device(rest, options_spec)


def _scoped_purge_commands(name, have_entry, raw_have):
    """Remove every field this module manages for one interface --
    scoped to this module's own fields only, never a whole-subtree
    delete for the VIF container shared with vyos_l3_interfaces.
    """
    have_device = _entry_to_device(have_entry, _ENTRY_OPTIONS)
    base = _iface_base(name, raw_have)
    return _purge_commands({}, have_device, base)


def _enabled_leaves(device, allowed_vlan_ids=None):
    """allowed_vlan_ids restricts the VIF portion to VLAN IDs the task
    actually listed. The interface itself needs no such restriction --
    it's always "listed" by virtue of appearing in want at all -- but
    an unlisted VIF was never mentioned by the task, and merged must
    never touch it. Confirmed real bug otherwise: a VIF the task
    doesn't reference at all (or a listed interface whose vifs simply
    omits it) would still have its stale "disable" leaf removed,
    silently re-enabling a VLAN the administrator deliberately
    disabled. Pass None (the want side, where every VIF present
    already is one the task listed) to skip this restriction.
    """
    result = {}
    if _DISABLE_DEVICE_KEY in device:
        result[_DISABLE_DEVICE_KEY] = device[_DISABLE_DEVICE_KEY]
    vif = device.get("vif")
    if vif:
        vif_result = {
            vlan_id: {_DISABLE_DEVICE_KEY: v[_DISABLE_DEVICE_KEY]}
            for vlan_id, v in vif.items()
            if _DISABLE_DEVICE_KEY in v
            and (allowed_vlan_ids is None or vlan_id in allowed_vlan_ids)
        }
        if vif_result:
            result["vif"] = vif_result
    return result


def build_commands(config, raw_have, state):
    raw_have = raw_have or {}
    config = config or []

    have_list = _device_to_argspec(raw_have)
    have_by_name = {e["name"]: e for e in have_list}
    want_by_name = {e["name"]: e for e in config if e.get("name")}

    if state == "deleted":
        cmds = []
        targets = (
            have_by_name
            if not config
            else {n: have_by_name[n] for n in want_by_name if n in have_by_name}
        )
        for name, have_entry in targets.items():
            cmds += _scoped_purge_commands(name, have_entry, raw_have)
        return cmds

    commands = []
    if state == "overridden":
        for name in set(have_by_name) - set(want_by_name):
            commands += _scoped_purge_commands(name, have_by_name[name], raw_have)

    for name, want_entry in want_by_name.items():
        have_entry = have_by_name.get(name, {})
        want_device = _entry_to_device(want_entry, _ENTRY_OPTIONS)
        have_device = _entry_to_device(have_entry, _ENTRY_OPTIONS)
        base = _iface_base(name, raw_have)

        if state in ("replaced", "overridden"):
            commands += _purge_commands(want_device, have_device, base)
        else:
            want_enabled = _enabled_leaves(want_device)
            have_enabled = _enabled_leaves(
                have_device,
                allowed_vlan_ids=(want_device.get("vif") or {}).keys(),
            )
            commands += _purge_commands(want_enabled, have_enabled, base)
        commands += dict_op(want_device, have_device, base, op="set")

    return commands


ARGUMENT_SPEC = dict(
    config=dict(
        type="list",
        elements="dict",
        options=dict(
            name=dict(type="str", required=True),
            description=dict(type="str"),
            enabled=dict(type="bool", default=True),
            mtu=dict(type="int"),
            duplex=dict(type="str", choices=["auto", "full", "half"]),
            speed=dict(type="str", choices=["auto", "10", "100", "1000", "2500", "10000"]),
            vrf=dict(type="str"),
            vifs=dict(
                type="list",
                elements="dict",
                options=dict(
                    vlan_id=dict(type="int", required=True),
                    description=dict(type="str"),
                    enabled=dict(type="bool", default=True),
                    mtu=dict(type="int"),
                ),
            ),
        ),
    ),
    state=dict(
        type="str",
        default="merged",
        choices=["merged", "replaced", "overridden", "deleted", "gathered"],
    ),
)

_ENTRY_OPTIONS = ARGUMENT_SPEC["config"]["options"]
_VIF_OPTIONS = _ENTRY_OPTIONS["vifs"]["options"]


def main():
    module = AnsibleModule(ARGUMENT_SPEC, supports_check_mode=True)
    vyos = VyOSModule(module)

    state = module.params["state"]
    config = module.params.get("config") or []

    raw_have = get_running_config(vyos)
    have = _device_to_argspec(raw_have)
    for entry in have:
        cast_by_spec(entry, _ENTRY_OPTIONS)

    if state == "gathered":
        module.exit_json(changed=False, gathered=have)

    commands = build_commands(config, raw_have, state)

    if module.check_mode:
        module.exit_json(changed=bool(commands), commands=commands, before=have, after=have)

    if commands:
        response = vyos.apply_commands(commands)
        saved = vyos.save_config()
        after_raw = get_running_config(vyos)
        after = _device_to_argspec(after_raw)
        for entry in after:
            cast_by_spec(entry, _ENTRY_OPTIONS)
        module.exit_json(
            changed=True,
            before=have,
            after=after,
            commands=commands,
            saved=saved,
            response=response,
        )

    module.exit_json(changed=False, before=have, after=have, commands=[])


if __name__ == "__main__":
    main()