diff --git a/plugins/module_utils/network/vyos/config/nat/nat.py b/plugins/module_utils/network/vyos/config/nat/nat.py index 3d633094..9b1f1f1d 100644 --- a/plugins/module_utils/network/vyos/config/nat/nat.py +++ b/plugins/module_utils/network/vyos/config/nat/nat.py @@ -1,549 +1,565 @@ # -*- 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 from copy import deepcopy from ansible_collections.ansible.netcommon.plugins.module_utils.network.common.rm_base.resource_module import ( ResourceModule, ) from ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.facts.facts import Facts from ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.rm_templates.nat import ( NatTemplate, ) from ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.utils.utils import combine class Nat(ResourceModule): """The vyos_nat config class""" def __init__(self, module): super(Nat, self).__init__( empty_fact_val={}, facts_module=Facts(module), module=module, resource="nat", tmplt=NatTemplate(), ) self.parsers = [] def execute_module(self): if self.state not in ["parsed", "gathered"]: self.generate_commands() self.run_commands() return self.result def generate_commands(self): wantd = deepcopy(self.want) haved = deepcopy(self.have) if self.state == "merged": - wantd = combine(haved, wantd, recursive=True, list_merge="append_rp") + wantd = combine(haved, wantd, recursive=True) if self.state == "deleted": if not wantd: for nat_type in haved: self.commands.append(f"delete {nat_type}") return self._list_to_dict(wantd) self._list_to_dict(haved) self._delete_nat_objects(wantd, haved, only_missing=False) return self._list_to_dict(wantd) self._list_to_dict(haved) if self.state == "replaced": self._delete_nat_objects(wantd, haved, only_missing=False) self._set_commands(wantd, haved) elif self.state == "overridden": self._delete_nat_objects(wantd, haved, only_missing=True) self._delete_nat_objects(wantd, haved, only_missing=False) self._set_commands(wantd, haved) else: self._set_commands(wantd, haved) self.commands = list(dict.fromkeys(self.commands)) # ------------------------------------------------------------------------- # List → keyed dict conversion # ------------------------------------------------------------------------- def _list_to_dict(self, config): nat = config.get("nat", {}) cgnat = nat.get("cgnat", {}) pool = cgnat.get("pool", {}) for ptype in ("external", "internal"): entries = pool.get(ptype) if isinstance(entries, list): pool[ptype] = {item["name"]: item for item in entries} rules = cgnat.get("rule") if isinstance(rules, list): cgnat["rule"] = {r["id"]: r for r in rules} for section in ("destination", "source", "static"): rules = nat.get(section, {}).get("rule") if isinstance(rules, list): nat[section]["rule"] = {r["id"]: r for r in rules} nat64 = config.get("nat64", {}) rules = nat64.get("source", {}).get("rule") if isinstance(rules, list): nat64["source"]["rule"] = {r["id"]: r for r in rules} for rule in nat64["source"]["rule"].values(): pools = rule.get("translation", {}).get("pool") if isinstance(pools, list): rule["translation"]["pool"] = {p["id"]: p for p in pools} nat66 = config.get("nat66", {}) for section in ("destination", "source"): rules = nat66.get(section, {}).get("rule") if isinstance(rules, list): nat66[section]["rule"] = {r["id"]: r for r in rules} # ------------------------------------------------------------------------- # Top-level dispatch # ------------------------------------------------------------------------- def _set_commands(self, wantd, haved): self._compare_cgnat_global(wantd, haved) self._compare_cgnat_pools(wantd, haved) self._compare_cgnat_rules(wantd, haved) for section in ("destination", "source", "static"): self._compare_nat_rules("nat", section, wantd, haved) self._compare_nat_rules("nat64", "source", wantd, haved) for section in ("destination", "source"): self._compare_nat_rules("nat66", section, wantd, haved) self.commands = list(dict.fromkeys(self.commands)) # ------------------------------------------------------------------------- # Delete helpers # ------------------------------------------------------------------------- def _delete_nat_objects(self, wantd, haved, only_missing=False): """ Generate delete commands for NAT objects. only_missing=False: delete objects present in both want and have (when different) only_missing=True: delete objects present in have but absent from want """ for nat_type in haved: want_nat = wantd.get(nat_type, {}) have_nat = haved[nat_type] if only_missing and nat_type not in wantd: self.commands.append(f"delete {nat_type}") continue for section in have_nat: want_section = want_nat.get(section, {}) have_section = have_nat[section] if only_missing and section not in want_nat: self.commands.append( f"delete {nat_type} {section.replace('_', '-')}", ) continue if section == "cgnat": for pool_type in ("external", "internal"): want_pools = want_section.get("pool", {}).get(pool_type, {}) have_pools = have_section.get("pool", {}).get(pool_type, {}) for name in have_pools: if only_missing and name not in want_pools: self.commands.append( f"delete {nat_type} cgnat pool {pool_type} {name}", ) elif not only_missing and name in want_pools: if self.state == "deleted" or want_pools[name] != have_pools[name]: self.commands.append( f"delete {nat_type} cgnat pool {pool_type} {name}", ) want_rules = want_section.get("rule", {}) have_rules = have_section.get("rule", {}) for rid in have_rules: if only_missing and rid not in want_rules: self.commands.append(f"delete {nat_type} cgnat rule {rid}") elif not only_missing and rid in want_rules: if self.state == "deleted" or want_rules[rid] != have_rules[rid]: self.commands.append(f"delete {nat_type} cgnat rule {rid}") else: want_rules = want_section.get("rule", {}) have_rules = have_section.get("rule", {}) cli_section = section.replace("_", "-") for rid in have_rules: if only_missing and rid not in want_rules: self.commands.append( f"delete {nat_type} {cli_section} rule {rid}", ) elif not only_missing and rid in want_rules: if self.state == "deleted" or want_rules[rid] != have_rules[rid]: self.commands.append( f"delete {nat_type} {cli_section} rule {rid}", ) # ------------------------------------------------------------------------- # CGNAT # ------------------------------------------------------------------------- def _compare_cgnat_global(self, wantd, haved): if self.state in ("replaced", "overridden") and not wantd.get("nat", {}).get("cgnat"): return w = wantd.get("nat", {}).get("cgnat", {}).get("log_allocation") h = haved.get("nat", {}).get("cgnat", {}).get("log_allocation") if bool(w) != bool(h): self.addcmd( {"nat": {"cgnat": {"log_allocation": True}}}, "cgnat_log_allocation", not bool(w), ) def _compare_cgnat_pools(self, wantd, haved): want_ext = wantd.get("nat", {}).get("cgnat", {}).get("pool", {}).get("external", {}) have_ext = haved.get("nat", {}).get("cgnat", {}).get("pool", {}).get("external", {}) want_int = wantd.get("nat", {}).get("cgnat", {}).get("pool", {}).get("internal", {}) have_int = haved.get("nat", {}).get("cgnat", {}).get("pool", {}).get("internal", {}) scope = self.state in ("replaced", "overridden") ext_names = set(want_ext) if scope else set(want_ext) | set(have_ext) int_names = set(want_int) if scope else set(want_int) | set(have_int) for name in ext_names: w = want_ext.get(name, {}) h = have_ext.get(name, {}) if scope and w != h: h = {} self._compare_external_pool(name, w, h) for name in int_names: w = want_int.get(name, {}) h = have_int.get(name, {}) if scope and w != h: h = {} self._compare_internal_pool(name, w, h) def _compare_external_pool(self, name, want, have): w = want.get("external_port_range") h = have.get("external_port_range") if w != h: if w: self.addcmd({"name": name, "range": w}, "cgnat_pool_external_port_range", False) elif self.state in ("replaced", "overridden"): self.addcmd({"name": name, "range": h}, "cgnat_pool_external_port_range", True) w = want.get("per_user_limit", {}).get("port") h = have.get("per_user_limit", {}).get("port") if w != h: if w: self.addcmd({"name": name, "limit": w}, "cgnat_pool_external_per_user", False) elif self.state in ("replaced", "overridden"): self.addcmd({"name": name, "limit": h}, "cgnat_pool_external_per_user", True) want_ranges = {(r["value"] if isinstance(r, dict) else r): r for r in want.get("range", [])} have_ranges = {(r["value"] if isinstance(r, dict) else r): r for r in have.get("range", [])} for val, rng in want_ranges.items(): - if val not in have_ranges: + existing = have_ranges.get(val) + if existing is None or existing != rng: seq = rng.get("seq") if isinstance(rng, dict) else None self.addcmd( {"name": name, "range": val, "seq": seq}, "cgnat_pool_external_range", False, ) if self.state in ("replaced", "overridden"): for val in have_ranges: if val not in want_ranges: self.addcmd({"name": name, "range": val}, "cgnat_pool_external_range", True) def _compare_internal_pool(self, name, want, have): want_ranges = set(want.get("range", [])) have_ranges = set(have.get("range", [])) for rng in want_ranges - have_ranges: self.addcmd({"name": name, "range": rng}, "cgnat_pool_internal_range", False) if self.state in ("replaced", "overridden"): for rng in have_ranges - want_ranges: self.addcmd({"name": name, "range": rng}, "cgnat_pool_internal_range", True) def _compare_cgnat_rules(self, wantd, haved): want_rules = wantd.get("nat", {}).get("cgnat", {}).get("rule", {}) have_rules = haved.get("nat", {}).get("cgnat", {}).get("rule", {}) rids = ( set(want_rules) if self.state in ("replaced", "overridden") else set(want_rules) | set(have_rules) ) for rid in rids: w = want_rules.get(rid, {}) h = have_rules.get(rid, {}) if self.state in ("replaced", "overridden") and w != h: h = {} w_src = w.get("source", {}).get("pool") h_src = h.get("source", {}).get("pool") if w_src != h_src: if w_src: self.addcmd({"id": rid, "pool": w_src}, "cgnat_rule_source_pool", False) elif self.state in ("replaced", "overridden"): self.addcmd({"id": rid, "pool": h_src}, "cgnat_rule_source_pool", True) w_tr = w.get("translation", {}).get("pool") h_tr = h.get("translation", {}).get("pool") if w_tr != h_tr: if w_tr: self.addcmd({"id": rid, "pool": w_tr}, "cgnat_rule_translation_pool", False) elif self.state in ("replaced", "overridden"): self.addcmd({"id": rid, "pool": h_tr}, "cgnat_rule_translation_pool", True) # ------------------------------------------------------------------------- # NAT / NAT64 / NAT66 rules # ------------------------------------------------------------------------- def _compare_nat_rules(self, nat_type, section, wantd, haved): want_rules = wantd.get(nat_type, {}).get(section, {}).get("rule", {}) have_rules = haved.get(nat_type, {}).get(section, {}).get("rule", {}) rids = ( set(want_rules) if self.state in ("replaced", "overridden") else set(want_rules) | set(have_rules) ) for rid in rids: w = want_rules.get(rid, {}) h = have_rules.get(rid, {}) if self.state in ("replaced", "overridden") and w != h: h = {} if w == h and self.state != "rendered": continue self._compare_rule(nat_type, section, rid, w, h) def _compare_rule(self, nat_type, section, rid, want, have): ctx = {"nat": nat_type, "type": section, "id": rid} + want_lb = want.get("load_balance") or {} + want_trans_addr = (want.get("translation") or {}).get("address") + have_trans_addr = (have.get("translation") or {}).get("address") + if want_lb and want_trans_addr is not None: + self._module.fail_json( + msg="translation.address and load_balance are mutually exclusive", + ) + if self.state == "merged" and want_lb and have_trans_addr is not None: + self._module.fail_json( + msg=( + "Cannot add load_balance to a rule that already has translation.address with " + "state=merged; use state=replaced or state=overridden" + ), + ) + for field in set(want) | set(have): if field == "inbound_interface": continue val = want.get(field) if field in want else have.get(field) if isinstance(val, bool): self._cmp_bool(want, have, field, ctx, f"nat_type_{field}") elif isinstance(val, str): self._cmp_scalar(want, have, field, ctx, f"nat_type_{field}") self._cmp_interface(want, have, ctx, nat_type, section) self._cmp_outbound_interface(want, have, ctx) for atype in ("destination", "source"): self._cmp_addr_sub(want, have, atype, ctx) self._cmp_translation(want, have, ctx) self._cmp_match_mark(want, have, ctx) self._cmp_nat64_pools(want, have, ctx) self._cmp_load_balance(want, have, ctx) # ------------------------------------------------------------------------- # Field-level helpers # ------------------------------------------------------------------------- def _cmp_scalar(self, want, have, field, ctx, parser): w = want.get(field) h = have.get(field) if w != h: if w is not None: self.addcmd(dict(ctx, **{field: w}), parser, False) elif self.state in ("replaced", "overridden"): self.addcmd(dict(ctx, **{field: h}), parser, True) def _cmp_bool(self, want, have, field, ctx, parser): w = bool(want.get(field)) h = bool(have.get(field)) if w != h: if w: self.addcmd(dict(ctx), parser, False) elif self.state in ("replaced", "overridden"): self.addcmd(dict(ctx), parser, True) def _cmp_interface(self, want, have, ctx, nat_type, section): iface_w = want.get("inbound_interface") iface_h = have.get("inbound_interface") if iface_w == iface_h: return if nat_type == "nat" and section == "static": if iface_w: self.addcmd(dict(ctx, value=iface_w), "nat_static_inbound_interface", False) elif self.state in ("replaced", "overridden"): self.addcmd(dict(ctx, value=iface_h), "nat_static_inbound_interface", True) return iface_w = iface_w or {} iface_h = iface_h or {} if nat_type == "nat": parser_name = "nat_inbound_interface_name" parser_group = "nat_inbound_interface_group" else: parser_name = "nat6x_inbound_interface" parser_group = "nat6x_inbound_interface" if iface_w.get("name") != iface_h.get("name"): if iface_w.get("name"): self.addcmd(dict(ctx, value=iface_w["name"]), parser_name, False) elif self.state in ("replaced", "overridden"): self.addcmd(dict(ctx, value=iface_h["name"]), parser_name, True) if nat_type == "nat" and iface_w.get("group") != iface_h.get("group"): if iface_w.get("group"): self.addcmd(dict(ctx, value=iface_w["group"]), parser_group, False) elif self.state in ("replaced", "overridden"): self.addcmd(dict(ctx, value=iface_h["group"]), parser_group, True) def _cmp_outbound_interface(self, want, have, ctx): iface_w = want.get("outbound_interface") or {} iface_h = have.get("outbound_interface") or {} if iface_w.get("name") != iface_h.get("name"): if iface_w.get("name"): self.addcmd(dict(ctx, value=iface_w["name"]), "nat_type_outbound_interface", False) elif self.state in ("replaced", "overridden"): self.addcmd(dict(ctx, value=iface_h["name"]), "nat_type_outbound_interface", True) if iface_w.get("group") != iface_h.get("group"): if iface_w.get("group"): self.addcmd( dict(ctx, value=iface_w["group"]), "nat_type_outbound_interface_group", False, ) elif self.state in ("replaced", "overridden"): self.addcmd( dict(ctx, value=iface_h["group"]), "nat_type_outbound_interface_group", True, ) def _cmp_addr_sub(self, want, have, atype, ctx): sub_w = want.get(atype) or {} sub_h = have.get(atype) or {} if sub_w == sub_h: return changed = {k: v for k, v in sub_w.items() if sub_h.get(k) != v} removed = { k: v for k, v in sub_h.items() if k not in sub_w and self.state in ("replaced", "overridden") } if changed: self.addcmd(dict(ctx, atype=atype, sub=changed), "nat_type_address", False) if removed: self.addcmd(dict(ctx, atype=atype, sub=removed), "nat_type_address", True) def _cmp_translation(self, want, have, ctx): trans_w = want.get("translation") or {} trans_h = have.get("translation") or {} if trans_w == trans_h: return changed = {k: v for k, v in trans_w.items() if k != "pool" and trans_h.get(k) != v} removed = { k: v for k, v in trans_h.items() if k != "pool" and k not in trans_w and self.state in ("replaced", "overridden") } if changed: self.addcmd(dict(ctx, translation=changed), "nat_type_translation_address", False) if removed: self.addcmd(dict(ctx, translation=removed), "nat_type_translation_address", True) def _cmp_match_mark(self, want, have, ctx): w = want.get("match", {}).get("mark") h = have.get("match", {}).get("mark") if w != h: if w is not None: self.addcmd(dict(ctx, mark=w), "nat64_match_mark", False) elif self.state in ("replaced", "overridden"): self.addcmd(dict(ctx, mark=h), "nat64_match_mark", True) def _cmp_nat64_pools(self, want, have, ctx): want_pools = want.get("translation", {}).get("pool", {}) have_pools = have.get("translation", {}).get("pool", {}) if isinstance(want_pools, list): want_pools = {p["id"]: p for p in want_pools} if isinstance(have_pools, list): have_pools = {p["id"]: p for p in have_pools} for pid in set(want_pools) | set(have_pools): wp = want_pools.get(pid, {}) hp = have_pools.get(pid, {}) if wp == hp: continue changed = {k: v for k, v in wp.items() if k != "id" and hp.get(k) != v} removed = { k: v for k, v in hp.items() if k != "id" and k not in wp and self.state in ("replaced", "overridden") } if changed: self.addcmd( dict(ctx, pool_id=pid, pool=changed), "nat64_translation_pool", False, ) if removed: self.addcmd( dict(ctx, pool_id=pid, pool=removed), "nat64_translation_pool", True, ) def _cmp_load_balance(self, want, have, ctx): lb_w = want.get("load_balance") or {} lb_h = have.get("load_balance") or {} want_hash = set(lb_w.get("hash") or []) have_hash = set(lb_h.get("hash") or []) for h in want_hash - have_hash: self.addcmd(dict(ctx, value=h), "nat_type_lb_hash", False) if self.state in ("replaced", "overridden"): for h in have_hash - want_hash: self.addcmd(dict(ctx, value=h), "nat_type_lb_hash", True) want_backends = lb_w.get("backend", []) have_backends = lb_h.get("backend", []) if isinstance(want_backends, list): want_backends = {b["ip"]: b for b in want_backends} if isinstance(have_backends, list): have_backends = {b["ip"]: b for b in have_backends} for ip in set(want_backends) | set(have_backends): wb = want_backends.get(ip, {}) hb = have_backends.get(ip, {}) if wb == hb: continue if wb: self.addcmd(dict(ctx, ip=ip, weight=wb.get("weight")), "nat_type_lb_backend", False) elif self.state in ("replaced", "overridden"): self.addcmd(dict(ctx, ip=ip, weight=hb.get("weight")), "nat_type_lb_backend", True) diff --git a/plugins/module_utils/network/vyos/facts/nat/nat.py b/plugins/module_utils/network/vyos/facts/nat/nat.py index 3c264473..73d6238f 100644 --- a/plugins/module_utils/network/vyos/facts/nat/nat.py +++ b/plugins/module_utils/network/vyos/facts/nat/nat.py @@ -1,174 +1,174 @@ # -*- 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 import re from ansible_collections.ansible.netcommon.plugins.module_utils.network.common import utils from ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.argspec.nat.nat import ( NatArgs, ) from ansible_collections.vyos.vyos.plugins.module_utils.network.vyos.rm_templates.nat import ( NatTemplate, ) class NatFacts(object): """The vyos nat facts class""" def __init__(self, module, subspec="config", options="options"): self._module = module self.argument_spec = NatArgs.argument_spec def get_config(self, connection): - return connection.get("show configuration commands | match 'nat'") + return connection.get("show configuration commands | match 'set nat'") def populate_facts(self, connection, ansible_facts, data=None): facts = {} config_lines = [] if not data: data = self.get_config(connection) for resource in data.splitlines(): config_lines.append(re.sub(r"'([^']*)'", r"\1", resource)) nat_parser = NatTemplate(lines=config_lines, module=self._module) objs = nat_parser.parse() objs = self._normalise(objs) ansible_facts["ansible_network_resources"].pop("nat", None) params = utils.remove_empties( nat_parser.validate_config(self.argument_spec, {"config": objs}, redact=True), ) if params.get("config"): facts["nat"] = params["config"] ansible_facts["ansible_network_resources"].update(facts) return ansible_facts def _deep_merge(self, base, override): for k, v in override.items(): if k in base and isinstance(base[k], dict) and isinstance(v, dict): self._deep_merge(base[k], v) elif k in base and isinstance(base[k], list) and isinstance(v, list): for entry in v: if entry not in base[k]: base[k].append(entry) else: base[k] = v return base def _merge_rule_list(self, rules): merged = {} for item in rules: rid = item["id"] if rid not in merged: merged[rid] = {"id": rid} for k, v in item.items(): if k == "id": continue if isinstance(v, list): existing = merged[rid].setdefault(k, []) for entry in v: if entry not in existing: existing.append(entry) elif isinstance(v, dict): merged[rid].setdefault(k, {}) self._deep_merge(merged[rid][k], v) else: merged[rid][k] = v return list(merged.values()) def _merge_pool_list(self, pools): merged = {} for item in pools: name = item["name"] if name not in merged: merged[name] = {"name": name} for k, v in item.items(): if k == "name": continue if k == "range" and isinstance(v, list): existing = merged[name].setdefault(k, []) existing.extend(v) if v and isinstance(v[0], dict): merged[name][k] = self._merge_range_list(existing) else: merged[name][k] = list(dict.fromkeys(existing)) elif isinstance(v, list): merged[name].setdefault(k, []) for val in v: if val not in merged[name][k]: merged[name][k].append(val) elif isinstance(v, dict): merged[name].setdefault(k, {}) self._deep_merge(merged[name][k], v) else: merged[name][k] = v return list(merged.values()) def _merge_range_list(self, ranges): """Merge external pool range entries by value, preserving seq.""" merged = {} for entry in ranges: if isinstance(entry, dict): key = entry.get("value") or entry.get("address", "") if not key: continue if key not in merged: merged[key] = {"value": key} if entry.get("seq"): merged[key]["seq"] = entry["seq"] else: if entry not in merged: merged[entry] = {"value": entry} return list(merged.values()) def _normalise(self, objs): for nat_type in ["nat", "nat64", "nat66"]: nat = objs.get(nat_type) if not nat: continue for section in ["destination", "source", "static", "cgnat"]: if section not in nat: continue rules = nat[section].get("rule") if isinstance(rules, list): nat[section]["rule"] = self._merge_rule_list(rules) nat[section]["rule"].sort(key=lambda x: x.get("id", 0)) if "cgnat" in nat and "pool" in nat["cgnat"]: pool = nat["cgnat"]["pool"] for ptype in ["external", "internal"]: if ptype in pool and isinstance(pool[ptype], list): pool[ptype] = self._merge_pool_list(pool[ptype]) if nat_type == "nat64": for rule in nat.get("source", {}).get("rule", []): pools = rule.get("translation", {}).get("pool") if pools and isinstance(pools, list): rule["translation"]["pool"] = self._merge_rule_list(pools) rule["translation"]["pool"].sort(key=lambda x: x.get("id", 0)) self._cast_ports(objs) return objs def _cast_ports(self, obj): """Recursively cast known integer port/seq fields to str.""" if isinstance(obj, dict): for k, v in obj.items(): if k in ("port", "seq") and isinstance(v, int): obj[k] = str(v) else: self._cast_ports(v) elif isinstance(obj, list): for item in obj: self._cast_ports(item) diff --git a/plugins/module_utils/network/vyos/utils/utils.py b/plugins/module_utils/network/vyos/utils/utils.py index 78445c02..62b65b17 100644 --- a/plugins/module_utils/network/vyos/utils/utils.py +++ b/plugins/module_utils/network/vyos/utils/utils.py @@ -1,309 +1,315 @@ # -*- coding: utf-8 -*- # Copyright 2019 Red Hat # GNU General Public License v3.0+ # (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) # utils from __future__ import absolute_import, division, print_function __metaclass__ = type from ansible.module_utils.basic import missing_required_lib try: import ipaddress HAS_IPADDRESS = True except ImportError: HAS_IPADDRESS = False def search_obj_in_list(name, lst, key="name"): if lst: for item in lst: if item[key] == name: return item return None def get_interface_type(interface): """Gets the type of interface""" if interface.startswith("eth"): return "ethernet" elif interface.startswith("bond"): return "bonding" elif interface.startswith("vti"): return "vti" elif interface.startswith("lo"): return "loopback" elif interface.startswith("vtun"): return "openvpn" elif interface.startswith("wg"): return "wireguard" elif interface.startswith("tun"): return "tunnel" elif interface.startswith("br"): return "bridge" elif interface.startswith("dum"): return "dummy" def get_interface_with_vif(interface): """Gets virtual interface if any or return as is""" vlan = None interface_real = interface if "." in interface: interface_real, vlan = interface.split(".") if vlan is not None: interface_real = interface_real + " vif " + vlan return interface_real def dict_delete(base, comparable): """ This function generates a dict containing key, value pairs for keys that are present in the `base` dict but not present in the `comparable` dict. :param base: dict object to base the diff on :param comparable: dict object to compare against base :returns: new dict object with key, value pairs that needs to be deleted. """ to_delete = dict() for key in base: if isinstance(base[key], dict): sub_diff = dict_delete(base[key], comparable.get(key, {})) if sub_diff: to_delete[key] = sub_diff else: if key not in comparable: to_delete[key] = base[key] return to_delete def diff_list_of_dicts(want, have): diff = [] set_w = set(tuple(d.items()) for d in want) set_h = set(tuple(d.items()) for d in have) difference = set_w.difference(set_h) for element in difference: diff.append(dict((x, y) for x, y in element)) return diff def get_lst_diff_for_dicts(want, have, lst): """ This function generates a list containing values that are only in want and not in list in have dict :param want: dict object to want :param have: dict object to have :param lst: list the diff on :return: new list object with values which are only in want. """ if not have: diff = want.get(lst) or [] else: want_elements = want.get(lst) or {} have_elements = have.get(lst) or {} diff = list_diff_want_only(want_elements, have_elements) return diff def get_lst_same_for_dicts(want, have, lst): """ This function generates a list containing values that are common for list in want and list in have dict :param want: dict object to want :param have: dict object to have :param lst: list the comparison on :return: new list object with values which are common in want and have. """ diff = None if want and have: want_list = want.get(lst) or {} have_list = have.get(lst) or {} diff = [i for i in want_list and have_list if i in have_list and i in want_list] return diff def list_diff_have_only(want_list, have_list): """ This function generated the list containing values that are only in have list. :param want_list: :param have_list: :return: new list with values which are only in have list """ if have_list and not want_list: diff = have_list elif not have_list: diff = None else: diff = [i for i in have_list + want_list if i in have_list and i not in want_list] return diff def list_diff_want_only(want_list, have_list): """ This function generated the list containing values that are only in want list. :param want_list: :param have_list: :return: new list with values which are only in want list """ if have_list and not want_list: diff = None elif not have_list: diff = want_list else: diff = [i for i in have_list + want_list if i in want_list and i not in have_list] return diff def search_dict_tv_in_list(d_val1, d_val2, lst, key1, key2): """ This function return the dict object if it exist in list. :param d_val1: :param d_val2: :param lst: :param key1: :param key2: :return: """ obj = next( (item for item in lst if item[key1] == d_val1 and item[key2] == d_val2), None, ) if obj: return obj else: return None def key_value_in_dict(have_key, have_value, want_dict): """ This function checks whether the key and values exist in dict :param have_key: :param have_value: :param want_dict: :return: """ for key, value in want_dict.items(): if key == have_key and value == have_value: return True return False def is_dict_element_present(dict, key): """ This function checks whether the key is present in dict. :param dict: :param key: :return: """ for item in dict: if item == key: return True return False def get_ip_address_version(address): """ This function returns the version of IP address :param address: IP address :return: """ if not HAS_IPADDRESS: raise Exception(missing_required_lib("ipaddress")) try: address = unicode(address) except NameError: address = str(address) version = ipaddress.ip_address(address.split("/")[0]).version return version def get_route_type(address): """ This function returns the route type based on IP address :param address: :return: """ version = get_ip_address_version(address) if version == 6: return "route6" elif version == 4: return "route" def _bool_to_str(val): """ This function converts the bool value into string. :param val: bool value. :return: enable/disable. """ return "enable" if str(val) == "True" else "disable" if str(val) == "False" else val def _is_w_same(w, h, key): """ This function checks whether the key value is same in desired and target config dictionary. :param w: base config. :param h: target config. :param key:attribute name. :return: True/False. """ return True if h and key in h and h[key] == w[key] else False def _in_target(h, key): """ This functi checks whether the target exist and key present in target config. :param h: target config. :param key: attribute name. :return: True/False. """ return True if h and key in h else False def in_target_not_none(h, key): """ This function checks whether the target exist,key present in target config, and the value is not None. :param h: target config. :param key: attribute name. :return: True/False. """ return True if h and key in h and h[key] is not None else False def combine(a, b, recursive=False, list_merge="replace"): """ Merge two dictionaries (shallow or deep). :param a: dict :param b: dict :param recursive: bool, deep merge - :param list_merge: str, only 'replace' is supported (default Ansible behavior) + :param list_merge: str, accepted for call-site compatibility only -- + NOT implemented. Any list value in `b` always replaces the + corresponding list in `a` wholesale, regardless of what this + parameter is set to. There is no item-level list merging here; + callers needing that must convert lists to ID-keyed dicts + themselves before calling combine() (see Nat._list_to_dict()). """ + if not isinstance(a, dict) or not isinstance(b, dict): raise ValueError("combine expects two dictionaries") result = a.copy() for k, v in b.items(): if recursive and k in result and isinstance(result[k], dict) and isinstance(v, dict): result[k] = combine(result[k], v, recursive=True, list_merge=list_merge) else: result[k] = v return result