path: root/plugins/modules/vyos_route_maps.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
# GNU General Public License v3.0+
from __future__ import absolute_import, division, print_function


__metaclass__ = type

DOCUMENTATION = r"""
---
module: vyos_route_maps
short_description: Manage route-map configuration on VyOS devices using REST API
description:
  - Manages route maps on VyOS via the REST API.
  - Uses REST API (C(connection=httpapi)) instead of CLI.
  - >-
    Covers the commonly used match/set fields (as documented below). VyOS's
    route-map schema is considerably larger than this (EVPN attributes,
    extended communities, RPKI matching, on-match goto/next, route-source,
    source-peer, source-vrf, and more) -- those are not modeled by this
    module and are a real, documented limitation, not an oversight.
version_added: "1.0.0"
author:
  - VyOS Community (@vyos)

options:
  config:
    description: List of route-map configurations.
    type: list
    elements: dict
    suboptions:
      route_map:
        description: Route map name.
        type: str
        required: true
      entries:
        description: Route map rules.
        type: list
        elements: dict
        suboptions:
          sequence:
            description: Rule sequence number (1-65535).
            type: int
            required: true
          action:
            description: Permit or deny.
            type: str
            choices: [permit, deny]
          description:
            description: Rule description.
            type: str
          call:
            description: Call another route map.
            type: str
          continue_sequence:
            description: Continue at a different sequence number.
            type: int
          match:
            description: Match conditions.
            type: dict
            suboptions:
              interface:
                description: Interface to match.
                type: str
              metric:
                description: Metric of route to match.
                type: int
              origin:
                description: BGP origin code to match.
                type: str
                choices: [egp, igp, incomplete]
              peer:
                description: Peer address to match.
                type: str
              protocol:
                description: Match protocol via which the route was learnt.
                type: str
                choices:
                  [
                    babel, bgp, connected, isis, kernel, ospf, ospfv3,
                    rip, ripng, static, table, vnc,
                  ]
              prefix_list:
                description: IPv4 prefix-list to match.
                type: str
              prefix_list6:
                description: IPv6 prefix-list to match.
                type: str
              ip:
                description: IPv4 next-hop match parameters.
                type: dict
                suboptions:
                  nexthop_address:
                    description: IPv4 next-hop address to match.
                    type: str
                  nexthop_prefix_list:
                    description: IPv4 next-hop prefix-list to match.
                    type: str
              ipv6:
                description: IPv6 next-hop match parameters.
                type: dict
                suboptions:
                  nexthop_address:
                    description: IPv6 next-hop address to match.
                    type: str
          set:
            description: Route parameters to set.
            type: dict
            suboptions:
              metric:
                description: Metric of route.
                type: int
              metric_type:
                description: Metric type.
                type: str
              origin:
                description: BGP origin code to set.
                type: str
                choices: [egp, igp, incomplete]
              originator_id:
                description: BGP originator ID.
                type: str
              src:
                description: Source address for route.
                type: str
              tag:
                description: Route tag value.
                type: int
              weight:
                description: BGP weight.
                type: int
              distance:
                description: Locally significant administrative distance.
                type: int
              table:
                description: Non-main kernel routing table.
                type: int
              local_preference:
                description: BGP local preference.
                type: int
              ip_next_hop:
                description: IPv4 next-hop address to set.
                type: str
              atomic_aggregate:
                description: Set the BGP atomic aggregate attribute.
                type: bool
              as_path_exclude:
                description: AS number(s) to remove from the as-path attribute.
                type: str
              as_path_prepend:
                description: AS number(s) to prepend to the as-path attribute.
                type: str
              as_path_prepend_last_as:
                description: Number of times to prepend the last AS number in the as-path.
                type: int
              aggregator:
                description: BGP aggregator attribute.
                type: dict
                suboptions:
                  as_:
                    description: AS number of an aggregation.
                    type: int
                    aliases: [as]
                  ip:
                    description: IP address of an aggregation.
                    type: str
              community:
                description: BGP community attribute.
                type: dict
                suboptions:
                  add:
                    description: Communities to add to a prefix.
                    type: list
                    elements: str
                  replace:
                    description: Communities to set for a prefix.
                    type: list
                    elements: str
                  none:
                    description: Completely remove the communities attribute from a prefix.
                    type: bool
                  delete:
                    description: Remove communities defined in a list from a prefix.
                    type: str
              large_community:
                description: BGP large community attribute.
                type: dict
                suboptions:
                  add:
                    description: Large communities to add to a prefix.
                    type: list
                    elements: str
                  replace:
                    description: Large communities to set for a prefix.
                    type: list
                    elements: str
                  none:
                    description: Completely remove the large-community attribute from a prefix.
                    type: bool
                  delete:
                    description: Remove large communities defined in a list from a prefix.
                    type: str
              ipv6_next_hop:
                description: IPv6 next-hop to set.
                type: dict
                suboptions:
                  global:
                    description: Nexthop IPv6 global address.
                    type: str
                  local:
                    description: Nexthop IPv6 local address.
                    type: str
                  peer_address:
                    description: Use the peer address (BGP only) as the nexthop.
                    type: bool
                  prefer_global:
                    description: Prefer the global address as the nexthop.
                    type: bool

  state:
    description:
      - Desired state of the route-map configuration.
      - C(merged) adds or updates entries without removing existing ones.
      - C(replaced) replaces each named route map mentioned in config.
      - C(overridden) replaces all route maps.
      - C(deleted) removes route maps. Without config removes all.
      - C(gathered) returns current configuration as structured data.
    type: str
    choices: [merged, replaced, overridden, deleted, gathered]
    default: merged

notes:
  - Requires C(ansible_connection=httpapi) with the VyOS httpapi plugin.
  - C(ansible_network_os) must be set to C(vyos.rest.vyos).
"""

EXAMPLES = r"""
- name: Merge route map configuration
  vyos.rest.vyos_route_maps:
    config:
      - route_map: RM-TEST-EXPORT-POLICY
        entries:
          - sequence: 10
            action: permit
            match:
              peer: 192.0.2.32
            set:
              metric: 5
              as_path_exclude: "111"
              aggregator:
                as_: 100
    state: merged

- name: Delete all route maps
  vyos.rest.vyos_route_maps:
    state: deleted

- name: Delete a specific route map
  vyos.rest.vyos_route_maps:
    config:
      - route_map: RM-TEST-EXPORT-POLICY
    state: deleted

- name: Gather current route map configuration
  vyos.rest.vyos_route_maps:
    state: gathered
"""

RETURN = r"""
before:
  description: Route map configuration before this module ran.
  returned: always
  type: list
after:
  description: Route map 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 route map configuration as structured data.
  returned: when state is gathered
  type: list
saved:
  description: Whether the config was saved after changes.
  returned: when changed
  type: bool
response:
  description: Raw API response.
  returned: always
  type: dict
"""

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


_BASE = ["policy", "route-map"]


def _derive_key_field(options_spec):
    """The field identifying each entry in a named-list section is
    never inferable from a generic walk alone -- but it doesn't need
    to be hand-declared either: both named-list sections in this
    argspec (route maps, rules) already mark exactly one suboption
    required=True. 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=None):
    """A list of dicts, each identified by key_field's value, becomes a
    device dict keyed by that value -- the one structural mechanic
    every named-list section in this module needs. entry_transform
    supplies whatever else is genuinely irreducible for a given section
    (a nested reshape) -- defaulting to the generic recursive walker.
    """
    entry_transform = entry_transform or autoclean
    result = {}
    for item in items or []:
        if not item.get(key_field):
            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=None, key_cast=None):
    entry_transform = entry_transform or from_device
    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())
    ]


# ---------------------------------------------------------------------------
# match.prefix_list / match.prefix_list6 -- confirmed genuine structural
# insertions: the argspec has these as flat fields, but the device
# nests them two levels down (ip.address.prefix-list /
# ipv6.address.prefix-list).
#
# match.ip.nexthop_address / nexthop_prefix_list, match.ipv6.
# nexthop_address -- confirmed the device nests these ONE level
# deeper than the argspec (ip.nexthop.address / ip.nexthop.prefix-list),
# under a "nexthop" node the argspec doesn't have (commented out in
# vyos-1x itself, T3304/T3976, since a plain leaf there would collide
# with the node).
# ---------------------------------------------------------------------------


def _match_to_device(match):
    if not match:
        return {}
    exclude = {"prefix_list", "prefix_list6", "ip", "ipv6"}
    device = autoclean({k: v for k, v in match.items() if k not in exclude})

    if match.get("prefix_list"):
        device.setdefault("ip", {}).setdefault("address", {})["prefix-list"] = match["prefix_list"]
    if match.get("prefix_list6"):
        ipv6_addr = device.setdefault("ipv6", {}).setdefault("address", {})
        ipv6_addr["prefix-list"] = match["prefix_list6"]

    ip = match.get("ip") or {}
    if ip.get("nexthop_address"):
        device.setdefault("ip", {}).setdefault("nexthop", {})["address"] = ip["nexthop_address"]
    if ip.get("nexthop_prefix_list"):
        ip_nh = device.setdefault("ip", {}).setdefault("nexthop", {})
        ip_nh["prefix-list"] = ip["nexthop_prefix_list"]

    ipv6 = match.get("ipv6") or {}
    if ipv6.get("nexthop_address"):
        device.setdefault("ipv6", {}).setdefault("nexthop", {})["address"] = ipv6["nexthop_address"]

    return device


def _match_from_device(data):
    if not data:
        return {}
    ip_raw = data.get("ip") or {}
    ipv6_raw = data.get("ipv6") or {}
    exclude = {"ip", "ipv6"}
    entry = from_device({k: v for k, v in data.items() if k not in exclude})

    prefix_list = (ip_raw.get("address") or {}).get("prefix-list")
    if prefix_list:
        entry["prefix_list"] = prefix_list
    prefix_list6 = (ipv6_raw.get("address") or {}).get("prefix-list")
    if prefix_list6:
        entry["prefix_list6"] = prefix_list6

    ip_nexthop = ip_raw.get("nexthop") or {}
    ip_sub = {}
    if ip_nexthop.get("address"):
        ip_sub["nexthop_address"] = ip_nexthop["address"]
    if ip_nexthop.get("prefix-list"):
        ip_sub["nexthop_prefix_list"] = ip_nexthop["prefix-list"]
    if ip_sub:
        entry["ip"] = ip_sub

    ipv6_nexthop = ipv6_raw.get("nexthop") or {}
    if ipv6_nexthop.get("address"):
        entry["ipv6"] = {"nexthop_address": ipv6_nexthop["address"]}

    return entry


# ---------------------------------------------------------------------------
# set.as_path_* -- confirmed structural collapse: three flat argspec
# keys (as_path_exclude/prepend/prepend_last_as) collapse onto one
# nested device node (as-path.{exclude,prepend,prepend-last-as}) with
# different sub-key names -- no mechanical transform gets from
# "as_path_exclude" to that shape.
#
# set.community / set.large_community / set.ipv6_next_hop are fully
# generic once modeled as real nested dicts (confirmed against schema:
# community/large-community are add/replace/none/delete nodes;
# ipv6-next-hop is global/local/peer-address/prefer-global) -- no
# entry-transform needed for them at all, the top-level community_to_
# device call handles them via ordinary recursion.
# ---------------------------------------------------------------------------
_AS_PATH_FIELDS = {
    "as_path_exclude": "exclude",
    "as_path_prepend": "prepend",
    "as_path_prepend_last_as": "prepend-last-as",
}


# Both renames in this module are position-specific -- confirmed
# against vyos-1x: "as" and "continue" are Python keywords and can't
# be used as dict() kwargs at all, so "as_"/"continue_sequence" are
# unavoidable argspec names, renamed to the device's real leaf names
# "as"/"continue". Neither fits a shared flat rename map: "as_" is
# nested inside "aggregator" specifically, and "continue_sequence" is
# a rule-level field, not a set-level one -- each is handled directly
# at its own point below instead.


def _set_to_device(setv):
    if not setv:
        return {}
    exclude = set(_AS_PATH_FIELDS) | {"aggregator"}
    device = autoclean({k: v for k, v in setv.items() if k not in exclude})

    as_path = {
        device_key: setv[arg_key]
        for arg_key, device_key in _AS_PATH_FIELDS.items()
        if setv.get(arg_key) is not None
    }
    if as_path:
        device["as-path"] = as_path

    agg = setv.get("aggregator")
    if agg:
        agg_device = autoclean({k: v for k, v in agg.items() if k != "as_"})
        if agg.get("as_") is not None:
            agg_device["as"] = agg["as_"]
        if agg_device:
            device["aggregator"] = agg_device

    return device


def _set_from_device(data):
    if not data:
        return {}
    as_path_raw = data.get("as-path") or {}
    agg_raw = data.get("aggregator") or {}
    exclude = {"as-path", "aggregator"}
    entry = from_device({k: v for k, v in data.items() if k not in exclude})

    for arg_key, device_key in _AS_PATH_FIELDS.items():
        if as_path_raw.get(device_key) is not None:
            entry[arg_key] = as_path_raw[device_key]
    if "as_path_prepend_last_as" in entry:
        entry["as_path_prepend_last_as"] = int(entry["as_path_prepend_last_as"])

    if agg_raw:
        agg_entry = from_device({k: v for k, v in agg_raw.items() if k != "as"})
        if agg_raw.get("as") is not None:
            agg_entry["as_"] = int(agg_raw["as"])
        if agg_entry:
            entry["aggregator"] = agg_entry

    return entry


# ---------------------------------------------------------------------------
# Rules (keyed by sequence) and route maps (keyed by name) -- both are
# named-list sections like any other in this collection, so they go
# through the same _keyed_list_to_device/_keyed_list_from_device
# mechanic as everything else, with key_field derived from ARGSPEC
# rather than hand-declared, instead of the hand-rolled loops this had
# before. _ROUTE_MAP_KEY/_RULE_KEY are derived after ARGUMENT_SPEC is
# built (near the bottom of this file) since they need it to exist.
# ---------------------------------------------------------------------------


def _rule_entry_to_device(rest):
    exclude = {"match", "set", "continue_sequence"}
    device = autoclean({k: v for k, v in rest.items() if k not in exclude})
    if rest.get("continue_sequence") is not None:
        device["continue"] = rest["continue_sequence"]
    if rest.get("match"):
        m = _match_to_device(rest["match"])
        if m:
            device["match"] = m
    if rest.get("set"):
        s = _set_to_device(rest["set"])
        if s:
            device["set"] = s
    return device


def _rule_entry_from_device(data):
    data = dict(data or {})
    continue_raw = data.pop("continue", None)
    match_raw = data.pop("match", None)
    set_raw = data.pop("set", None)
    entry = from_device(data)
    if continue_raw is not None:
        entry["continue_sequence"] = int(continue_raw)
    match = _match_from_device(match_raw)
    if match:
        entry["match"] = match
    setv = _set_from_device(set_raw)
    if setv:
        entry["set"] = setv
    return entry


def _route_map_entry_to_device(rest):
    entries = rest.get("entries") or []
    if not entries:
        return {}
    return {"rule": _keyed_list_to_device(entries, _RULE_KEY, _rule_entry_to_device)}


def _route_map_entry_from_device(data):
    rule_raw = (data or {}).get("rule")
    if not rule_raw:
        return {"entries": []}
    entries = _keyed_list_from_device(rule_raw, _RULE_KEY, _rule_entry_from_device, key_cast=int)
    # _keyed_list_from_device sorts by the raw device key as a string,
    # which orders sequence numbers wrong across a digit-count boundary
    # (e.g. "10" < "9" lexicographically) -- re-sort numerically now
    # that key_cast has already converted each key to a real int.
    return {"entries": sorted(entries, key=lambda e: e[_RULE_KEY])}


def _want_to_device(config):
    with_entries = [rm for rm in (config or []) if rm.get("entries")]
    return _keyed_list_to_device(with_entries, _ROUTE_MAP_KEY, _route_map_entry_to_device)


def get_running_config(vyos):
    raw = vyos.get_config(_BASE) or {}
    if isinstance(raw, dict) and "route-map" in raw:
        return raw["route-map"] or {}
    return raw


def _device_to_argspec(raw):
    if not raw:
        return []
    return _keyed_list_from_device(raw, _ROUTE_MAP_KEY, _route_map_entry_from_device)


def _seed_route_map_placeholders(want, have):
    """dict_op's fallback guesses a translated device key whenever a
    want key is missing from have entirely (a brand-new route map or
    rule). That guess is correct for a schema field name but wrong for
    a route-map name, which is an opaque value that may legitimately
    contain an underscore (confirmed against vyos-1x: "Name of
    route-map can only contain alpha-numeric letters, hyphen and
    underscores") -- confirmed as a real bug via direct reproduction,
    the same class found in vyos_snmp_server's "admin_user" case:
    "my_route_map" was silently becoming "my-route-map" in the
    generated command on first creation.

    Seeds an empty placeholder into have (mutated in place) for every
    route-map name present in want but not yet in have, using the
    exact verbatim value -- dict_op's own unmodified exact-match lookup
    then finds it directly and never reaches its guessing fallback.
    Also seeds each rule's own tag-node level with None (not {}), since
    a rule with no other fields set is a presence-only entry -- seeding
    {} there would make dict_op think it already matches and skip
    emitting the needed set command (the same mistake caught and fixed
    once already this session).
    """
    for rm_name, rm_val in (want or {}).items():
        rm_have = have.setdefault(rm_name, {})
        if not isinstance(rm_have, dict):
            continue
        rule_want = (rm_val or {}).get("rule") or {}
        if rule_want:
            rule_have = rm_have.setdefault("rule", {})
            if isinstance(rule_have, dict):
                for seq in rule_want:
                    if seq not in rule_have:
                        rule_have[seq] = None


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

    if state == "deleted":
        if not config:
            return [("delete", _BASE)] if raw_have else []
        cmds = []
        for rm in config:
            if rm.get("route_map") in raw_have:
                cmds.append(("delete", _BASE + [rm["route_map"]]))
        return cmds

    want = _want_to_device(config)
    norm_have = _want_to_device(_device_to_argspec(raw_have))
    _seed_route_map_placeholders(want, norm_have)

    commands = []
    if state == "overridden":
        commands += dict_op(want, norm_have, _BASE, op="purge")
    elif state == "replaced":
        want_names = {rm.get("route_map") for rm in config if rm.get("route_map")}
        for name in want_names:
            section_want = want.get(name, {})
            section_have = norm_have.get(name, {})
            commands += dict_op(section_want, section_have, _BASE + [name], op="purge")
    commands += dict_op(want, norm_have, _BASE, op="set")
    return commands


ARGUMENT_SPEC = dict(
    config=dict(
        type="list",
        elements="dict",
        options=dict(
            route_map=dict(type="str", required=True),
            entries=dict(
                type="list",
                elements="dict",
                options=dict(
                    sequence=dict(type="int", required=True),
                    action=dict(type="str", choices=["permit", "deny"]),
                    description=dict(type="str"),
                    call=dict(type="str"),
                    continue_sequence=dict(type="int"),
                    match=dict(
                        type="dict",
                        options=dict(
                            interface=dict(type="str"),
                            metric=dict(type="int"),
                            origin=dict(type="str", choices=["egp", "igp", "incomplete"]),
                            peer=dict(type="str"),
                            protocol=dict(
                                type="str",
                                choices=[
                                    "babel",
                                    "bgp",
                                    "connected",
                                    "isis",
                                    "kernel",
                                    "ospf",
                                    "ospfv3",
                                    "rip",
                                    "ripng",
                                    "static",
                                    "table",
                                    "vnc",
                                ],
                            ),
                            prefix_list=dict(type="str"),
                            prefix_list6=dict(type="str"),
                            ip=dict(
                                type="dict",
                                options=dict(
                                    nexthop_address=dict(type="str"),
                                    nexthop_prefix_list=dict(type="str"),
                                ),
                            ),
                            ipv6=dict(
                                type="dict",
                                options=dict(
                                    nexthop_address=dict(type="str"),
                                ),
                            ),
                        ),
                    ),
                    set=dict(
                        type="dict",
                        options=dict(
                            metric=dict(type="int"),
                            metric_type=dict(type="str"),
                            origin=dict(type="str", choices=["egp", "igp", "incomplete"]),
                            originator_id=dict(type="str"),
                            src=dict(type="str"),
                            tag=dict(type="int"),
                            weight=dict(type="int"),
                            distance=dict(type="int"),
                            table=dict(type="int"),
                            local_preference=dict(type="int"),
                            ip_next_hop=dict(type="str"),
                            atomic_aggregate=dict(type="bool"),
                            as_path_exclude=dict(type="str"),
                            as_path_prepend=dict(type="str"),
                            as_path_prepend_last_as=dict(type="int"),
                            aggregator=dict(
                                type="dict",
                                options=dict(
                                    as_=dict(type="int", aliases=["as"]),
                                    ip=dict(type="str"),
                                ),
                            ),
                            community=dict(
                                type="dict",
                                options=dict(
                                    add=dict(type="list", elements="str"),
                                    replace=dict(type="list", elements="str"),
                                    none=dict(type="bool"),
                                    delete=dict(type="str"),
                                ),
                            ),
                            large_community=dict(
                                type="dict",
                                options=dict(
                                    add=dict(type="list", elements="str"),
                                    replace=dict(type="list", elements="str"),
                                    none=dict(type="bool"),
                                    delete=dict(type="str"),
                                ),
                            ),
                            ipv6_next_hop=dict(
                                type="dict",
                                options={
                                    "global": dict(type="str"),
                                    "local": dict(type="str"),
                                    "peer_address": dict(type="bool"),
                                    "prefer_global": dict(type="bool"),
                                },
                            ),
                        ),
                    ),
                ),
            ),
        ),
    ),
    state=dict(
        default="merged",
        choices=["merged", "replaced", "overridden", "deleted", "gathered"],
    ),
)

_ENTRY_OPTIONS = ARGUMENT_SPEC["config"]["options"]["entries"]["options"]
_ROUTE_MAP_KEY = _derive_key_field(ARGUMENT_SPEC["config"]["options"])
_RULE_KEY = _derive_key_field(_ENTRY_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 rm in have:
        for entry in rm.get("entries") or []:
            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)

    if commands:
        response = vyos.apply_commands(commands)
        saved = vyos.save_config()
        after_raw = get_running_config(vyos)
        after = _device_to_argspec(after_raw)
        for rm in after:
            for entry in rm.get("entries") or []:
                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()