avocado_i2n.vmnet.network module

Utility to manage local networks of vms and various topologies.

SUMMARY

Copyright: Intra2net AG

CONTENTS

The main class is the VMNetwork class and is used to perform all network related configuration for each virt test (store all its network information, offer all the network related services it needs, etc.). It can be used to test Proxy, Port forwarding, VPN, NAT, etc.

Each vm is a network node and can have one of few currently supported operating systems. For ease of defaults use it is recommended to have at least three nics, respectively with the role of host nic for local isolated connection to the host, the role of internet nic for (internet) connection to the other nodes, and the role of LAN nic for other any other connections to vm’s own LANs.

Ephemeral clients are based on RIP Linux and are temporary clients created just for the duration of a test. An arbitrary number of those can be spawned depending on test requirements and available resources.

INTERFACE

avocado_i2n.vmnet.network.BIND_DHCP_CONFIG = '/etc/dhcp/dhcpd.conf'

networking service backend paths

class avocado_i2n.vmnet.network.VMNetwork(params: Params, env: Env)[source]

Bases: object

Any VMNetwork instance can be used to connect vms in various network topologies.

It can also be used to reconfigure, ping, retrieve the session of, as well as spawn clients for each of them.

start_all_sessions() → None[source]

Start a session to each of the vm nodes.

integrate_node(node: VMNode) → None[source]

Add all interfaces and netconfigs resulting from a new vm node into the vm network.

Thus integrate the configuration into the available one.

Parameters:

node – vm node to be integrated into the network

reattach_interface(client: VM, server: VM, client_nic: str = 'internet_nic', server_nic: str = 'lan_nic', proxy_nic: str = '') → None[source]

Reconfigure a network interface of a vm reattaching it to a different interface’s network config.

Parameters:

  • client – vm whose interace will be rattached

  • server – vm whose network will the interface be attached to

  • client_nic – role of the nic of the client

  • server_nic – role of the nic of the server

  • proxy_nic – name of a proxyARP nic of the server

If the proxy_nic is defined and the second interface (server_nic) is different than the proxy_nic value, it is assumed to be and turned into a proxyarp interface whose responses are provided by the actual interface defined in the proxy_nic parameter.

Any processing related to the vm or the netconfig’s servers must be performed separately.

A typical processing for the clients is to insert a DHCP/DNS host, while a typical processing for the vm is to recreate it on top of the moved bridges with or without session.

A typical processing for the vm is to reconfigure the nic type of server_nic to PROXYARP and its IP to the IP of proxy_nic.

setup_host_services() → None[source]

Provide all necessary services like DHCP, DNS and NAT to restrict all tests locally.

setup_host_bridges() → None[source]

Set up bridges and interfaces needed to create and isolate the network.

The final network topology is derived entirely from the test parameters.

spawn_clients(server_name: str, clients_num: int, nic: str = 'lan_nic') → tuple[VM, ...][source]

Create and boot ephemeral clients for a given server.

Parameters:

  • server_name – name of the vm that plays the role of a server

  • clients_num – number of ephemeral clients to spawn

  • nic – name of the nic of the server

Returns:

generated ephemeral clients

change_network_address(netconfig: VMNetconfig, new_ip: str, new_mask: str = None) → None[source]

Change the ip of a netconfig.

More specifically of the network interface of any vm participating in it.

Parameters:

  • netconfig – netconfig to change the IP of

  • new_ip – new IP address for the netconfig

  • new_mask – new network mask for the netconfig

Note

The network must have at least one interface in order to change its address.

configure_tunnel_between_vms(name: str, vm1: VM, vm2: VM, local1: dict[str, str] = None, remote1: dict[str, str] = None, peer1: dict[str, str] = None, auth: dict[str, str] = None, apply_extra_options: dict[str, Any] = None) → None[source]

Configure a tunnel between two vms.

Parameters:

  • name – name of the tunnel

  • vm1 – left side vm of the tunnel

  • vm2 – right side vm of the tunnel

  • local1 – left local type as in tunnel constructor

  • remote1 – left remote type as in tunnel constructor

  • peer1 – left peer type as in tunnel constructor

  • auth – authentication configuration as described in the tunnel constructor

  • apply_extra_options – extra switches to apply as key exchange, firewall ruleset, etc.

configure_tunnel_on_vm(name: str, vm: VM, apply_extra_options: dict[str, Any] = None) → None[source]

Configure a tunnel on a vm, assuming it is manually or independently configured on the other end.

Parameters:

  • name – name of the tunnel

  • vm – vm where the tunnel will be configured

  • apply_extra_options – extra switches to apply as key exchange, firewall ruleset, etc.

Raises:

exceptions.KeyError if not all tunnel parameters are present

Currently the method uses only existing tunnels.

configure_roadwarrior_vpn_on_server(name: str, server: VM, client: VM, local1: dict[str, str] = None, remote1: dict[str, str] = None, peer1: dict[str, str] = None, auth: dict[str, str] = None, apply_extra_options: dict[str, Any] = None) → None[source]

Configure a VPN connection (tunnel) on a VM to act as a VPN server.

This will allow individual clients to access it from the Internet

Arguments are similar to the ones from configure_tunnel_between_vms() with the exception of:

Parameters:

  • server – vm which will be the VPN server for roadwarrior connections

  • client – vm which will be connecting individual device

Regarding the client, only its parameters will be updated by this method.

configure_vpn_route(vms: list[VM], vpns: list[str], remote1: dict[str, str] = None, peer1: dict[str, str] = None, auth: dict[str, str] = None, extra_apply_options: dict[str, Any] = None) → None[source]

Build a set of VPN connections using VPN forwarding to gain access from one vm to another.

Arguments are similar to the ones from configure_tunnel_between_vms() with the exception of:

Parameters:

  • vms – vms to participate in the VPN route

  • vpns – VPNs over which the route will be constructed

Raises:

exceptions.TestError if #vpns < #vms - 1 or #vpns < 2 or #vms < 2

Infrastructure of point to point VPN connections must already exist.

verify_vpn_in_log(src_vm: VM, dst_vm: VM, log_vm: VM = None, require_blocked: bool = False) → None[source]

Search for the appropriate message in the vpn log file.

Parameters:

  • src_vm – source vm whose packets will be logged

  • dst_vm – destination vm whose packets will be logged

  • log_vm – vm where all packets are logged

  • require_blocked – whether to expect access message or deny message

Raises:

exceptions.TestError if the source or destination vms are not on the network

Raises:

exceptions.TestFail if the VPN packets were not logged properly

This function requires modified firewall ruleset for the vpn connection.

ping(src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None) → tuple[int, str][source]

Pings a vm from another vm to test basic ICMP connectivity.

Parameters:

  • src_vm – source vm which will ping

  • dst_vm – destination vm which will be pinged

  • dst_nic – nic of the destination vm used if necessary to obtain accessible IP

  • address – explicit IP or domain to use for pinging

Returns:

the status and output of the performed ping

If no address is provided, the IP is obtained by analyzing the network topology from src_vm to dst_vm.

If no dst_vm is provided, the ping happens directly to address.

ping_validate(src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None, timeout: int = 30) → None[source]

Pings a vm from another vm to test basic ICMP connectivity and bails on nonzero status.

Arguments are similar to the ones from ping() with the exception of:

Parameters:

timeout – number of seconds to retry the ping for as networking might not be immediately available

Raises:

exceptions.TestError if the performed ping failed

This method does not perform a refined exit status check, you can use the non-validated version and perform your own customization if you wish.

ping_all(timeout: int = 30) → None[source]

Pings all nodes from each other in order to test complete basic ICMP connectivity.

Parameters:

timeout – number of seconds to retry the ping for as networking might not be immediately available

Raises:

exceptions.TestError if a network mutual ping failed

The ping happens among all LAN members, throwing an exception if one of the pings fails.

port_connectivity(msg: str, src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None, port: int = 80, protocol: str = 'TCP') → tuple[int, str][source]

Test connectivity using a predefined port (usually in addition to pinging).

Arguments are similar to the ping() method with the exception of:

Parameters:

  • msg – probing data to be sent to the port

  • port – forwarding port to send the message to

  • protocol – protocol type (TCP, UDP or something over them)

Returns:

the result of the performed port connection attempt

port_connectivity_validate(msg: str, src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None, port: int = 80, protocol: str = 'TCP', validate_output: str = '', require_blocked: bool = False) → None[source]

Test connectivity using a predefined port (usually in addition to pinging).

Arguments are similar to the port_connectivity() method with the exception of:

Parameters:

  • validate_output – string to find in the command output and validate against

  • require_blocked – whether to expect nonzero status from the connection attempt

Raises:

exceptions.TestError if the performed port connection attempt failed

This method does not perform a refined exit status check, you can use the non-validated version and perform your own customization if you wish.

http_connectivity(src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None, port: int = 80, protocol: str = 'HTTP') → tuple[int, str][source]

Test connectivity using an HTTP port and protocol.

Arguments are similar to the port_connectivity() method.

Raises:

exceptions.TestError if inappropriate protocol was given

http_connectivity_validate(src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None, port: int = 80, protocol: str = 'HTTP', require_blocked: bool = False) → None[source]

Test connectivity using an HTTP port and protocol.

Arguments are similar to the port_connectivity() method.

Raises:

exceptions.TestError if inappropriate protocol was given

https_connectivity(src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None, port: int = 443, protocol: str = 'HTTPS') → tuple[int, str][source]

Test connectivity using an HTTPS port and protocol.

Arguments are similar to the port_connectivity() method.

Raises:

exceptions.TestError if inappropriate protocol was given

https_connectivity_validate(src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None, port: int = 443, protocol: str = 'HTTPS', require_blocked: bool = False) → None[source]

Test connectivity using an HTTPS port and protocol.

Arguments are similar to the port_connectivity() method.

Raises:

exceptions.TestError if inappropriate protocol was given

ssh_connectivity(src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None, port: int = 22, protocol: str = 'SSH') → tuple[int, str][source]

Test connectivity using an SSH port and protocol.

Arguments are similar to the port_connectivity() method.

Raises:

exceptions.TestError if inappropriate protocol was given

ssh_connectivity_validate(src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None, port: int = 22, protocol: str = 'SSH', require_blocked: bool = False) → None[source]

Test connectivity using an SSH port and protocol.

Arguments are similar to the port_connectivity() method.

Raises:

exceptions.TestError if inappropriate protocol was given

ssh_hostname(src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', timeout: int = 10) → str[source]

Get the host name of a vm from any other vm in the vm net using the SSH protocol.

Parameters:

  • src_vm – source vm with the SSH client

  • dst_vm – destination vm with the SSH server

  • dst_nic – nic of the destination vm used if necessary to obtain accessible IP

  • timeout – timeout for the SSH connection

Returns:

the hostname of the SSH server

This tests the TCP connectivity and verifies it leads to the correct machine.

scp_files(src_path: str, dst_path: str, src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', timeout: int = 10) → None[source]

Copy files securely where built-in methods like vm.copy_files_to() fail.

Parameters:

  • src_path – source path for the securely copied files

  • dst_path – destination path for the securely copied files

  • src_vm – source vm with the ssh client

  • dst_vm – destination vm with the ssh server

  • dst_nic – nic of the destination vm used if necessary to obtain accessible IP

  • timeout – timeout for the SSH connection

Raises:

exceptions.TestFail if the files couldn’t be copied

The paths src_path and dst_path must be strings, possibly with a wildcard.

ftp_connectivity(msg: str, file: str | None, src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None, port: int = 21) → tuple[int, str][source]

Send file request to an FTP destination port and address and verify it was received.

Arguments are similar to the port_connectivity() method with the exception of:

Parameters:

file – file to retrieve containing the test data or none if sent directly

Raises:

exceptions.TestError if inappropriate protocol was given

ftp_connectivity_validate(msg: str, file: str | None, src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None, port: int = 21, require_blocked: bool = False) → None[source]

Send file request to an FTP destination port and address and verify it was received.

Arguments are similar to the port_connectivity() method with the exception of:

Parameters:

file – file to retrieve containing the test data or none if sent directly

Raises:

exceptions.TestError if inappropriate protocol was given

tftp_connectivity(msg: str, file: str | None, src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None, port: int = 69) → tuple[int, str][source]

Send file request to an TFTP destination port and address and verify it was received.

Arguments are similar to the port_connectivity() method with the exception of:

Parameters:

file – file to retrieve containing the test data or none if sent directly

Raises:

exceptions.TestError if inappropriate protocol was given

tftp_connectivity_validate(msg: str, file: str | None, src_vm: VM, dst_vm: VM, dst_nic: str = 'lan_nic', address: str = None, port: int = 69, require_blocked: bool = False) → None[source]

Send file request to an TFTP destination port and address and verify it was received.

Arguments are similar to the port_connectivity() method with the exception of:

Parameters:

file – file to retrieve containing the test data or none if sent directly

Raises:

exceptions.TestError if inappropriate protocol was given