diff --git a/anta/tests/stun.py b/anta/tests/stun.py new file mode 100644 index 000000000..a8e8d9eb9 --- /dev/null +++ b/anta/tests/stun.py @@ -0,0 +1,117 @@ +# Copyright (c) 2023-2024 Arista Networks, Inc. +# Use of this source code is governed by the Apache License 2.0 +# that can be found in the LICENSE file. +"""Test functions related to various STUN settings.""" + +# Mypy does not understand AntaTest.Input typing +# mypy: disable-error-code=attr-defined +from __future__ import annotations + +from ipaddress import IPv4Address +from typing import ClassVar + +from pydantic import BaseModel + +from anta.custom_types import Port +from anta.models import AntaCommand, AntaTemplate, AntaTest +from anta.tools import get_failed_logs, get_value + + +class VerifyStunClient(AntaTest): + """ + Verifies the configuration of the STUN client, specifically the IPv4 source address and port. + + Optionally, it can also verify the public address and port. + + Expected Results + ---------------- + * Success: The test will pass if the STUN client is correctly configured with the specified IPv4 source address/port and public address/port. + * Failure: The test will fail if the STUN client is not configured or if the IPv4 source address, public address, or port details are incorrect. + + Examples + -------- + ```yaml + anta.tests.stun: + - VerifyStunClient: + stun_clients: + - source_address: 172.18.3.2 + public_address: 172.18.3.21 + source_port: 4500 + public_port: 6006 + - source_address: 100.64.3.2 + public_address: 100.64.3.21 + source_port: 4500 + public_port: 6006 + ``` + """ + + name = "VerifyStunClient" + description = "Verifies the STUN client is configured with the specified IPv4 source address and port. Validate the public IP and port if provided." + categories: ClassVar[list[str]] = ["stun"] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show stun client translations {source_address} {source_port}")] + + class Input(AntaTest.Input): + """Input model for the VerifyStunClient test.""" + + stun_clients: list[ClientAddress] + + class ClientAddress(BaseModel): + """Source and public address/port details of STUN client.""" + + source_address: IPv4Address + """IPv4 source address of STUN client.""" + source_port: Port = 4500 + """Source port number for STUN client.""" + public_address: IPv4Address | None = None + """Optional IPv4 public address of STUN client.""" + public_port: Port | None = None + """Optional public port number for STUN client.""" + + def render(self, template: AntaTemplate) -> list[AntaCommand]: + """Render the template for each STUN translation.""" + return [template.render(source_address=client.source_address, source_port=client.source_port) for client in self.inputs.stun_clients] + + @AntaTest.anta_test + def test(self) -> None: + """Main test function for VerifyStunClient.""" + self.result.is_success() + + # Iterate over each command output and corresponding client input + for command, client_input in zip(self.instance_commands, self.inputs.stun_clients): + bindings = command.json_output["bindings"] + source_address = str(command.params.source_address) + source_port = command.params.source_port + + # If no bindings are found for the STUN client, mark the test as a failure and continue with the next client + if not bindings: + self.result.is_failure(f"STUN client transaction for source `{source_address}:{source_port}` is not found.") + continue + + # Extract the public address and port from the client input + public_address = client_input.public_address + public_port = client_input.public_port + + # Extract the transaction ID from the bindings + transaction_id = next(iter(bindings.keys())) + + # Prepare the actual and expected STUN data for comparison + actual_stun_data = { + "source ip": get_value(bindings, f"{transaction_id}.sourceAddress.ip"), + "source port": get_value(bindings, f"{transaction_id}.sourceAddress.port"), + } + expected_stun_data = {"source ip": source_address, "source port": source_port} + + # If public address is provided, add it to the actual and expected STUN data + if public_address is not None: + actual_stun_data["public ip"] = get_value(bindings, f"{transaction_id}.publicAddress.ip") + expected_stun_data["public ip"] = str(public_address) + + # If public port is provided, add it to the actual and expected STUN data + if public_port is not None: + actual_stun_data["public port"] = get_value(bindings, f"{transaction_id}.publicAddress.port") + expected_stun_data["public port"] = public_port + + # If the actual STUN data does not match the expected STUN data, mark the test as failure + if actual_stun_data != expected_stun_data: + failed_log = get_failed_logs(expected_stun_data, actual_stun_data) + self.result.is_failure(f"For STUN source `{source_address}:{source_port}`:{failed_log}") diff --git a/docs/api/tests.md b/docs/api/tests.md index 4b1bb4910..1db3a8bb7 100644 --- a/docs/api/tests.md +++ b/docs/api/tests.md @@ -34,6 +34,7 @@ Here are the tests that we currently provide: - [SNMP](tests.snmp.md) - [Software](tests.software.md) - [STP](tests.stp.md) +- [STUN](tests.stun.md) - [System](tests.system.md) - [VLAN](tests.vlan.md) - [VXLAN](tests.vxlan.md) diff --git a/docs/api/tests.stun.md b/docs/api/tests.stun.md new file mode 100644 index 000000000..b4274e9a7 --- /dev/null +++ b/docs/api/tests.stun.md @@ -0,0 +1,20 @@ +--- +anta_title: ANTA catalog for STUN tests +--- + + +::: anta.tests.stun + options: + show_root_heading: false + show_root_toc_entry: false + show_bases: false + merge_init_into_class: false + anta_hide_test_module_description: true + show_labels: true + filters: + - "!test" + - "!render" diff --git a/examples/tests.yaml b/examples/tests.yaml index 6697e2350..e4445aa67 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -358,6 +358,18 @@ anta.tests.stp: - 10 - 20 +anta.tests.stun: + - VerifyStunClient: + stun_clients: + - source_address: 172.18.3.2 + public_address: 172.18.3.21 + source_port: 4500 + public_port: 6006 + - source_address: 100.64.3.2 + public_address: 100.64.3.21 + source_port: 4500 + public_port: 6006 + anta.tests.system: - VerifyUptime: minimum: 86400 diff --git a/mkdocs.yml b/mkdocs.yml index 784ade5a2..0050640ac 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -199,6 +199,7 @@ nav: - Services: api/tests.services.md - SNMP: api/tests.snmp.md - STP: api/tests.stp.md + - STUN: api/tests.stun.md - Software: api/tests.software.md - System: api/tests.system.md - VXLAN: api/tests.vxlan.md diff --git a/tests/units/anta_tests/test_stun.py b/tests/units/anta_tests/test_stun.py new file mode 100644 index 000000000..2c873650c --- /dev/null +++ b/tests/units/anta_tests/test_stun.py @@ -0,0 +1,176 @@ +# Copyright (c) 2023-2024 Arista Networks, Inc. +# Use of this source code is governed by the Apache License 2.0 +# that can be found in the LICENSE file. +"""Test inputs for anta.tests.stun.py.""" + +from __future__ import annotations + +from typing import Any + +from anta.tests.stun import VerifyStunClient +from tests.lib.anta import test # noqa: F401; pylint: disable=W0611 + +DATA: list[dict[str, Any]] = [ + { + "name": "success", + "test": VerifyStunClient, + "eos_data": [ + { + "bindings": { + "000000010a64ff0100000000": { + "sourceAddress": {"ip": "100.64.3.2", "port": 4500}, + "publicAddress": {"ip": "192.64.3.2", "port": 6006}, + } + } + }, + { + "bindings": { + "000000040a64ff0100000000": { + "sourceAddress": {"ip": "172.18.3.2", "port": 4500}, + "publicAddress": {"ip": "192.18.3.2", "port": 6006}, + } + } + }, + { + "bindings": { + "000000040a64ff0100000000": { + "sourceAddress": {"ip": "172.18.4.2", "port": 4500}, + "publicAddress": {"ip": "192.18.4.2", "port": 6006}, + } + } + }, + { + "bindings": { + "000000040a64ff0100000000": { + "sourceAddress": {"ip": "172.18.6.2", "port": 4500}, + "publicAddress": {"ip": "192.18.6.2", "port": 6006}, + } + } + }, + ], + "inputs": { + "stun_clients": [ + {"source_address": "100.64.3.2", "public_address": "192.64.3.2", "source_port": 4500, "public_port": 6006}, + {"source_address": "172.18.3.2"}, + {"source_address": "172.18.4.2", "source_port": 4500, "public_address": "192.18.4.2"}, + {"source_address": "172.18.6.2", "source_port": 4500, "public_port": 6006}, + ] + }, + "expected": {"result": "success"}, + }, + { + "name": "failure-incorrect-public-ip", + "test": VerifyStunClient, + "eos_data": [ + { + "bindings": { + "000000010a64ff0100000000": { + "sourceAddress": {"ip": "100.64.3.2", "port": 4500}, + "publicAddress": {"ip": "192.64.3.2", "port": 6006}, + } + } + }, + { + "bindings": { + "000000040a64ff0100000000": { + "sourceAddress": {"ip": "172.18.3.2", "port": 4500}, + "publicAddress": {"ip": "192.18.3.2", "port": 6006}, + } + } + }, + ], + "inputs": { + "stun_clients": [ + {"source_address": "100.64.3.2", "public_address": "192.164.3.2", "source_port": 4500, "public_port": 6006}, + {"source_address": "172.18.3.2", "public_address": "192.118.3.2", "source_port": 4500, "public_port": 6006}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "For STUN source `100.64.3.2:4500`:\nExpected `192.164.3.2` as the public ip, but found `192.64.3.2` instead.", + "For STUN source `172.18.3.2:4500`:\nExpected `192.118.3.2` as the public ip, but found `192.18.3.2` instead.", + ], + }, + }, + { + "name": "failure-no-client", + "test": VerifyStunClient, + "eos_data": [ + {"bindings": {}}, + {"bindings": {}}, + ], + "inputs": { + "stun_clients": [ + {"source_address": "100.64.3.2", "public_address": "192.164.3.2", "source_port": 4500, "public_port": 6006}, + {"source_address": "172.18.3.2", "public_address": "192.118.3.2", "source_port": 4500, "public_port": 6006}, + ] + }, + "expected": { + "result": "failure", + "messages": ["STUN client transaction for source `100.64.3.2:4500` is not found.", "STUN client transaction for source `172.18.3.2:4500` is not found."], + }, + }, + { + "name": "failure-incorrect-public-port", + "test": VerifyStunClient, + "eos_data": [ + {"bindings": {}}, + { + "bindings": { + "000000040a64ff0100000000": { + "sourceAddress": {"ip": "172.18.3.2", "port": 4500}, + "publicAddress": {"ip": "192.18.3.2", "port": 4800}, + } + } + }, + ], + "inputs": { + "stun_clients": [ + {"source_address": "100.64.3.2", "public_address": "192.164.3.2", "source_port": 4500, "public_port": 6006}, + {"source_address": "172.18.3.2", "public_address": "192.118.3.2", "source_port": 4500, "public_port": 6006}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "STUN client transaction for source `100.64.3.2:4500` is not found.", + "For STUN source `172.18.3.2:4500`:\n" + "Expected `192.118.3.2` as the public ip, but found `192.18.3.2` instead.\n" + "Expected `6006` as the public port, but found `4800` instead.", + ], + }, + }, + { + "name": "failure-all-type", + "test": VerifyStunClient, + "eos_data": [ + {"bindings": {}}, + { + "bindings": { + "000000040a64ff0100000000": { + "sourceAddress": {"ip": "172.18.3.2", "port": 4500}, + "publicAddress": {"ip": "192.18.3.2", "port": 4800}, + } + } + }, + ], + "inputs": { + "stun_clients": [ + {"source_address": "100.64.3.2", "public_address": "192.164.3.2", "source_port": 4500, "public_port": 6006}, + {"source_address": "172.18.4.2", "public_address": "192.118.3.2", "source_port": 4800, "public_port": 6006}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "STUN client transaction for source `100.64.3.2:4500` is not found.", + "For STUN source `172.18.4.2:4800`:\n" + "Expected `172.18.4.2` as the source ip, but found `172.18.3.2` instead.\n" + "Expected `4800` as the source port, but found `4500` instead.\n" + "Expected `192.118.3.2` as the public ip, but found `192.18.3.2` instead.\n" + "Expected `6006` as the public port, but found `4800` instead.", + ], + }, + }, +]