Skip to content

Commit

Permalink
feat(anta.tests): Added testcase to verify stun client (#578)
Browse files Browse the repository at this point in the history
Co-authored-by: gmulocher <[email protected]>
  • Loading branch information
MaheshGSLAB and gmuloc authored Apr 12, 2024
1 parent 245c6f9 commit afe12ca
Show file tree
Hide file tree
Showing 6 changed files with 327 additions and 0 deletions.
117 changes: 117 additions & 0 deletions anta/tests/stun.py
Original file line number Diff line number Diff line change
@@ -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}")
1 change: 1 addition & 0 deletions docs/api/tests.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
20 changes: 20 additions & 0 deletions docs/api/tests.stun.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
---
anta_title: ANTA catalog for STUN tests
---
<!--
~ 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.
-->

::: 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"
12 changes: 12 additions & 0 deletions examples/tests.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions mkdocs.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
176 changes: 176 additions & 0 deletions tests/units/anta_tests/test_stun.py
Original file line number Diff line number Diff line change
@@ -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.",
],
},
},
]

0 comments on commit afe12ca

Please sign in to comment.