-
-
Notifications
You must be signed in to change notification settings - Fork 3
Description
Hey!
I'd like to propose adding IPv6 capabilities to the https://github.com/dmulyalin/nornir-salt/blob/master/nornir_salt/plugins/tasks/tcp_ping.py file.
This is my suggestion:
-
added the function socketType() -- this will use the python ipaddress package to determine if a valid IPv4 or IPv6 address was entered. If there was a valid address it will return back either socket.AF_INET or socket.AF_INET6 and raise an exception if an invalid IP was entered.
-
Adding the two lines below will open the socket on either IPv4 or IPv6 depending on the type of host that was entered in
socket_inet = socketType(host)
s = socket.socket(socket_inet)
I tested this out in normal python, but did not actually do a test of it yet within nornir_salt, i don't have a full development zone for nornir_salt to test this out in yet.
(I'm not sure if this type of logic will be required elsewhere)
Thanks!
"""
tcp_ping
########
Tests connection to a TCP port trying to establish a three way
handshake. Useful for network discovery or testing.
tcp_ping sample usage
=====================
Sample code to run ``tcp_ping`` task::
import pprint
from nornir import InitNornir
from nornir_salt.plugins.tasks import tcp_ping
from nornir_salt.plugins.functions import ResultSerializer
nr = InitNornir(config_file="config.yaml")
result = NornirObj.run(
task=tcp_ping,
ports=[22]
)
result_dictionary = ResultSerializer(result)
pprint.pprint(result_dictionary)
# prints:
#
# {'IOL1': {'tcp_ping': {22: True}},
# 'IOL2': {'tcp_ping': {22: True}}}
tcp_ping returns
================
Returns dictionary of port numbers as keys with True/False as values
tcp_ping reference
==================
.. autofunction:: nornir_salt.plugins.tasks.tcp_ping.tcp_ping
"""
import logging
import socket
from typing import Optional, List
from nornir.core.task import Result, Task
from ipaddress import ip_address, IPv4Address
log = logging.getLogger(__name__)
def socketType(IP: str):
try:
return socket.AF_INET if type(ip_address(IP)) is IPv4Address else socket.AF_INET6
except ValueError:
return Exception(f"Invalid format for {IP} provided.")
def tcp_ping(
task: Task, ports: List[int] = None, timeout: int = 1, host: Optional[str] = None
) -> Result:
"""
:param ports: list of int, optional, tcp ports to ping, defaults to host's port or 22
:param timeout: int, optional, connection timeout, defaults to 1
:param host: string, optional, address to TCP ping, defaults to hosts' ``hostname`` value
:returns: dictionary of port numbers as keys with True/False as values
"""
ports = ports or []
if not ports:
ports = [int(task.host.port) if task.host.port else 22]
elif isinstance(ports, int):
ports = [ports]
if isinstance(ports, list):
if not all(isinstance(port, int) for port in ports):
raise ValueError("Invalid value for 'ports'")
else:
raise ValueError("Invalid value for 'ports'")
host = host or task.host.hostname
result = {}
for port in ports:
socket_inet = socketType(host)
s = socket.socket(socket_inet)
s.settimeout(timeout)
try:
status = s.connect_ex((host, port))
if status == 0:
connection = True
else:
connection = False
except (socket.gaierror, socket.timeout, socket.error):
connection = False
finally:
s.close()
result[port] = connection
return Result(host=task.host, result=result)