diff --git a/docs/vyos.vyos.vyos_nat_module.rst b/docs/vyos.vyos.vyos_nat_module.rst index 6aba3709..7a8045ee 100644 --- a/docs/vyos.vyos.vyos_nat_module.rst +++ b/docs/vyos.vyos.vyos_nat_module.rst @@ -1,3723 +1,3723 @@ .. _vyos.vyos.vyos_nat_module: ****************** vyos.vyos.vyos_nat ****************** **NAT resource module** Version added: 6.0.0 .. contents:: :local: :depth: 1 Synopsis -------- - This module manages NAT configuration on devices running VyOS. Parameters ---------- .. raw:: html
Parameter Choices/Defaults Comments
config
dictionary
The desired configuration for the NAT resource represented as a dictionary.
nat
dictionary
Configuration for NAT rules.
cgnat
dictionary
Configuration for Carrier Grade NAT (CGNAT).
log_allocation
boolean
    Choices:
  • no
  • yes
Log CGNAT address allocations.
pool
dictionary
Configuration for CGNAT pools.
external
list / elements=dictionary
List of external NAT pools for CGNAT.
external_port_range
string
Port range to use for NAT translations in this external pool.
name
string / required
Name of the external NAT pool.
per_user_limit
dictionary
Per-user limit configuration for the external pool.
port
string
Maximum number of ports allocated per user.
range
list / elements=dictionary
List of external IP address ranges in the pool.
seq
string
Optional sequence number for this range entry.
value
string / required
IP address, prefix, or range (e.g. 203.0.113.0/24 or 203.0.113.1-203.0.113.60).
internal
list / elements=dictionary
List of internal NAT pools for CGNAT.
name
string / required
Name of the internal NAT pool.
range
list / elements=string
List of internal IP addresses or prefixes in the pool.
rule
list / elements=dictionary
List of CGNAT rules.
id
integer / required
Rule number for CGNAT.
source
dictionary
Source pool configuration for CGNAT translation.
pool
string
Source pool name to use for CGNAT translation.
translation
dictionary
Translation pool configuration for CGNAT.
pool
string
Translation pool name to use for CGNAT translation.
destination
dictionary
Configuration for destination NAT rules.
rule
list / elements=dictionary
List of destination NAT rules.
description
string
User-friendly description of the destination NAT rule.
destination
dictionary
Match criteria for destination NAT.
address
string
IP address, subnet, or range to match.
address_group
string
Address group name to match.
domain_group
string
Domain group name to match.
fqdn
string
Fully qualified domain name to match.
mac_group
string
MAC address group name to match.
network_group
string
Network group name to match.
port
string
Port number or range to match.
port_group
string
Port group name to match.
disable
boolean
    Choices:
  • no
  • yes
Disable this destination NAT rule.
exclude
boolean
    Choices:
  • no
  • yes
Exclude packets matching this rule from NAT.
id
integer / required
Rule number for destination NAT.
inbound_interface
dictionary
Match inbound interface.
group
string
Interface group to match.
name
string
Interface name to match.
load_balance
dictionary
Load balancing configuration for this NAT rule.
backend
list / elements=dictionary
List of backends to load-balance across. Weights should sum to 100.
ip
string
IP address of the backend translation target.
weight
integer
Relative weight (1-100) for this backend's share of load-balanced traffic.
hash
list / elements=string
    Choices:
  • source-address
  • destination-address
  • source-port
  • destination-port
  • random
Fields to hash on for load balancing. Mutually exclusive with translation.address.
log
boolean
    Choices:
  • no
  • yes
Log packets hitting this rule.
packet_type
string
Packet type to match.
protocol
string
Protocol to NAT (default all).
translation
dictionary
Translation configuration for destination NAT.
address
string
IP address or prefix to translate destination to.
address_mapping
string
    Choices:
  • random
  • persistent
Address mapping mode for translation.
port
string
Port number or range to translate destination port to.
port_mapping
string
    Choices:
  • random
  • none
Port mapping mode for translation.
redirect_port
string
Redirect to local port number.
source
dictionary
Configuration for source NAT rules.
rule
list / elements=dictionary
List of source NAT rules.
description
string
User-friendly description of the source NAT rule.
destination
dictionary
Destination match criteria for source NAT.
address
string
IP address, subnet, or range to match.
address_group
string
Address group name to match.
domain_group
string
Domain group name to match.
fqdn
string
Fully qualified domain name to match.
mac_group
string
MAC address group name to match.
network_group
string
Network group name to match.
port
string
Port number or range to match.
port_group
string
Port group name to match.
disable
boolean
    Choices:
  • no
  • yes
Disable this source NAT rule.
exclude
boolean
    Choices:
  • no
  • yes
Exclude packets matching this rule from NAT.
id
integer / required
Rule number for source NAT.
load_balance
dictionary
Load balancing configuration for this NAT rule.
backend
list / elements=dictionary
List of backends to load-balance across. Weights should sum to 100.
ip
string
IP address of the backend translation target.
weight
integer
Relative weight (1-100) for this backend's share of load-balanced traffic.
hash
list / elements=string
    Choices:
  • source-address
  • destination-address
  • source-port
  • destination-port
  • random
Fields to hash on for load balancing. Mutually exclusive with translation.address.
log
boolean
    Choices:
  • no
  • yes
Log packets hitting this rule.
outbound_interface
dictionary
Match outbound interface.
group
string
Interface group to match.
name
string
Interface name to match.
packet_type
string
Packet type to match.
protocol
string
Protocol to NAT (default all).
source
dictionary
Source match criteria for source NAT.
address
string
IP address, subnet, or range to match.
address_group
string
Address group name to match.
domain_group
string
Domain group name to match.
fqdn
string
Fully qualified domain name to match.
mac_group
string
MAC address group name to match.
network_group
string
Network group name to match.
port
string
Port number or range to match.
port_group
string
Port group name to match.
translation
dictionary
Translation configuration for source NAT.
address
string
IP address or prefix to translate source to. Use masquerade to masquerade as the outbound interface address.
address_mapping
string
    Choices:
  • random
  • persistent
Address mapping mode for translation.
port
string
Port number or range to translate source port to.
port_mapping
string
    Choices:
  • random
  • none
Port mapping mode for translation.
static
dictionary
Configuration for static one-to-one NAT rules.
rule
list / elements=dictionary
List of static NAT rules.
description
string
User-friendly description of the static NAT rule.
destination
dictionary
Match criteria for static NAT.
address
string
IP address, subnet, or range to match.
id
integer / required
Rule number for static NAT.
inbound_interface
string
Inbound interface that this static NAT rule applies to.
log
boolean
    Choices:
  • no
  • yes
Log packets hitting this static NAT rule.
translation
dictionary
Translation configuration for static NAT.
address
string
IP address or prefix to translate to.
nat64
dictionary
Configuration for NAT64 (IPv6-to-IPv4) rules.
source
dictionary
Configuration for NAT64 source rules.
rule
list / elements=dictionary
List of NAT64 source rules.
description
string
User-friendly description of the NAT64 source rule.
disable
boolean
    Choices:
  • no
  • yes
Disable this NAT64 source rule.
id
integer / required
Rule number for NAT64 source rule (1-999999).
match
dictionary
Match criteria for NAT64 source rule.
mark
integer
Match on firewall mark value (1-2147483647).
source
dictionary
IPv6 source prefix to match for NAT64 translation.
prefix
string
IPv6 source prefix to match (h:h:h:h:h:h:h:h/x).
translation
dictionary
Translation configuration for NAT64 source rule.
pool
list / elements=dictionary
List of translation pools for NAT64.
address
string
IPv4 address or prefix for translation pool.
description
string
User-friendly description of the translation pool.
disable
boolean
    Choices:
  • no
  • yes
Disable this translation pool.
id
integer / required
Pool number (1-999999).
port
string
Port number or range for translation pool.
protocol
string
    Choices:
  • icmp
  • tcp
  • udp
Protocol for this translation pool entry.
nat66
dictionary
Configuration for NAT66 (IPv6-to-IPv6) rules.
destination
dictionary
Configuration for NAT66 destination rules.
rule
list / elements=dictionary
List of NAT66 destination rules.
description
string
User-friendly description of the NAT66 destination rule.
destination
dictionary
Match criteria for NAT66 destination rule.
address
string
IPv6 address or prefix to match.
port
string
Port number or range to match.
disable
boolean
    Choices:
  • no
  • yes
Disable this NAT66 destination rule.
exclude
boolean
    Choices:
  • no
  • yes
Exclude packets matching this rule from NAT66.
id
integer / required
Rule number for NAT66 destination rule.
inbound_interface
dictionary
Inbound interface to match for NAT66 destination rule.
name
string
Interface name to match.
log
boolean
    Choices:
  • no
  • yes
Log packets hitting this NAT66 destination rule.
protocol
string
Protocol to match.
source
dictionary
Source match criteria for NAT66 destination rule.
address
string
IPv6 source address or prefix to match.
port
string
Source port number or range to match.
translation
dictionary
Translation configuration for NAT66 destination rule.
address
string
IPv6 address or prefix to translate destination to.
port
string
Port number or range to translate destination port to.
source
dictionary
Configuration for NAT66 source rules.
rule
list / elements=dictionary
List of NAT66 source rules.
description
string
User-friendly description of the NAT66 source rule.
destination
dictionary
Destination match criteria for NAT66 source rule.
port
string
Destination port number or range to match.
prefix
string
IPv6 destination prefix to match (h:h:h:h:h:h:h:h/x).
disable
boolean
    Choices:
  • no
  • yes
Disable this NAT66 source rule.
exclude
boolean
    Choices:
  • no
  • yes
Exclude packets matching this rule from NAT66.
id
integer / required
Rule number for NAT66 source rule.
log
boolean
    Choices:
  • no
  • yes
Log packets hitting this NAT66 source rule.
outbound_interface
dictionary
Outbound interface to match for NAT66 source rule.
name
string
Interface name to match.
protocol
string
Protocol to match.
source
dictionary
Source match criteria for NAT66 source rule.
port
string
Source port number or range to match.
prefix
string
IPv6 source prefix to match (h:h:h:h:h:h:h:h/x).
translation
dictionary
Translation configuration for NAT66 source rule.
address
string
IPv6 address or prefix to translate source to. Use masquerade to masquerade as the outbound interface address.
port
string
Port number or range to translate source port to.
running_config
string
This option is used only with state parsed.
The value of this option should be the output received from the VyOS device by executing the command show configuration commands | match 'nat'.
The state parsed reads the configuration from show configuration commands | match 'nat' and transforms it into Ansible structured data as per the module argspec. The value is then returned in the parsed key within the result.
The state replaced replaces only the provided configuration, while overridden removes any existing NAT configuration not specified in config.
state
string
    Choices:
  • deleted
  • merged ←
  • overridden
  • replaced
  • gathered
  • rendered
  • parsed
The state the configuration should be left in.

Notes ----- .. note:: - - Tested against VyOS 1.3.8, 1.4.2, the upcoming 1.5, and the rolling release of spring 2025. + - Tested against VyOS 1.4.3 and 1.5.0. - This module works with connection ``network_cli``. Examples -------- .. code-block:: yaml # Using merged - configure CGNAT - name: Merge CGNAT configuration vyos.vyos.vyos_nat: config: nat: cgnat: log_allocation: true pool: external: - name: ext-pool-1 external_port_range: "10000-20000" per_user_limit: port: "200" range: - value: 203.0.113.0/24 internal: - name: int-pool-1 range: - 10.0.0.0/24 rule: - id: 1 source: pool: int-pool-1 translation: pool: ext-pool-1 state: merged # Using merged - configure destination NAT - name: Merge destination NAT rule vyos.vyos.vyos_nat: config: nat: destination: rule: - id: 100 description: "Web server NAT" protocol: tcp log: true destination: address: 198.51.100.10 port: "80" translation: address: 192.168.1.10 port: "8080" state: merged # Using merged - configure source NAT - name: Merge source NAT rule vyos.vyos.vyos_nat: config: nat: source: rule: - id: 200 description: "Outbound masquerade" protocol: tcp log: true outbound_interface: name: eth0 translation: address: masquerade state: merged # Using merged - configure static NAT - name: Merge static NAT rule vyos.vyos.vyos_nat: config: nat: static: rule: - id: 300 description: "Static mapping" inbound_interface: eth2 destination: address: 198.51.100.20 translation: address: 192.168.1.20 log: true state: merged # Using merged - configure NAT64 - name: Merge NAT64 source rule vyos.vyos.vyos_nat: config: nat64: source: rule: - id: 10 description: "NAT64 example" source: prefix: 2001:db8::/96 match: mark: 100 translation: pool: - id: 1 address: 192.168.100.10 port: "1-65535" protocol: udp state: merged # Using merged - configure NAT66 - name: Merge NAT66 destination rule vyos.vyos.vyos_nat: config: nat66: destination: rule: - id: 20 description: "NAT66 DNAT" protocol: tcp inbound_interface: name: eth1 destination: address: 2001:db8::1 translation: address: 2001:db8:1::10 port: "8443" state: merged # Using replaced - replace specific NAT rules - name: Replace destination NAT rule vyos.vyos.vyos_nat: config: nat: destination: rule: - id: 100 description: "Replaced web server NAT" protocol: tcp destination: address: 198.51.100.10 port: "443" translation: address: 192.168.1.10 port: "8443" state: replaced # Using overridden - override entire NAT configuration - name: Override entire NAT configuration vyos.vyos.vyos_nat: config: nat: destination: rule: - id: 100 description: "Only rule after override" protocol: tcp destination: address: 198.51.100.10 port: "80" translation: address: 192.168.1.10 port: "8080" state: overridden # Using deleted - delete all NAT configuration - name: Delete all NAT configuration vyos.vyos.vyos_nat: state: deleted # Using deleted - delete specific NAT rules - name: Delete specific NAT rules vyos.vyos.vyos_nat: config: nat: destination: rule: - id: 100 source: rule: - id: 200 nat64: source: rule: - id: 10 state: deleted # Using gathered - name: Gather NAT configuration from device vyos.vyos.vyos_nat: state: gathered # Using rendered - name: Render NAT configuration offline vyos.vyos.vyos_nat: config: nat: destination: rule: - id: 100 description: "Rendered rule" protocol: tcp destination: address: 198.51.100.10 port: "80" translation: address: 192.168.1.10 port: "8080" state: rendered # Using parsed - name: Parse NAT configuration from file vyos.vyos.vyos_nat: running_config: "{{ lookup('file', './nat_config.cfg') }}" state: parsed Return Values ------------- Common return values are documented `here `_, the following are the fields unique to this module: .. raw:: html
Key Returned Description
after
dictionary
when changed
The resulting configuration after module execution.

Sample:
This output will always be in the same format as the module argspec.
before
dictionary
when state is merged, replaced, overridden or deleted
The configuration prior to the module execution.

Sample:
This output will always be in the same format as the module argspec.
commands
list
when state is merged, replaced, overridden or deleted
The set of commands pushed to the remote device.

Sample:
["set nat destination rule 100 description 'Web server NAT'", 'set nat destination rule 100 protocol tcp', 'set nat destination rule 100 inbound-interface name eth2', 'set nat destination rule 100 destination address 198.51.100.10', 'set nat destination rule 100 translation address 192.168.1.10', 'delete nat source rule 200']
gathered
dictionary
when state is gathered
Facts about the network resource gathered from the remote device as structured data.

Sample:
This output will always be in the same format as the module argspec.
parsed
dictionary
when state is parsed
The device native config provided in running_config option parsed into structured data as per module argspec.

Sample:
This output will always be in the same format as the module argspec.
rendered
list
when state is rendered
The provided configuration in the task rendered in device-native format (offline).

Sample:
["set nat destination rule 100 description 'Web server NAT'", 'set nat destination rule 100 protocol tcp', 'set nat destination rule 100 inbound-interface name eth2', 'set nat destination rule 100 destination address 198.51.100.10', 'set nat destination rule 100 translation address 192.168.1.10']


Status ------ Authors ~~~~~~~ - Evgeny Molotkov (@omnom62) diff --git a/plugins/module_utils/network/vyos/config/nat/nat.py b/plugins/module_utils/network/vyos/config/nat/nat.py index e0cbd4dd..f21e5bf6 100644 --- a/plugins/module_utils/network/vyos/config/nat/nat.py +++ b/plugins/module_utils/network/vyos/config/nat/nat.py @@ -1,568 +1,578 @@ # -*- 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) 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._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(): 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 {} + have_lb = have.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" - ), - ) + if self.state == "merged": + if 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" + ), + ) + if want_trans_addr is not None and have_lb: + self._module.fail_json( + msg=( + "Cannot add translation.address to a rule that already has load_balance 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: weight = wb.get("weight") if weight is None: self._module.fail_json(msg="load_balance.backend entries require 'weight'") self.addcmd(dict(ctx, ip=ip, weight=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/tests/integration/targets/vyos_nat/tasks/cli.yaml b/tests/integration/targets/vyos_nat/tasks/cli.yaml index f4ca537e..a71206c0 100644 --- a/tests/integration/targets/vyos_nat/tasks/cli.yaml +++ b/tests/integration/targets/vyos_nat/tasks/cli.yaml @@ -1,19 +1,20 @@ --- - name: Collect all cli test cases ansible.builtin.find: paths: "{{ role_path }}/tests/cli" patterns: "{{ testcase }}\\.yaml$" use_regex: true delegate_to: localhost + register: test_cases - name: Set test_items ansible.builtin.set_fact: test_items: "{{ test_cases.files | map(attribute='path') | list }}" - name: Run test case (connection=ansible.netcommon.network_cli) ansible.builtin.include_tasks: "{{ test_case_to_run }}" vars: ansible_connection: ansible.netcommon.network_cli with_items: "{{ test_items }}" loop_control: loop_var: test_case_to_run