Quick IP Connection Tests With Python

Regularly my work confronts me with situations in which I need to verify that two systems can communicate with each other over a network. In case the system initiating the connection has a variety of network tools installed, such as ping, nslookup, traceroute or the (infamous) telnet, this is easy. It becomes more challenging, if you have to run such tests from container platforms, where most of your containers are stripped down to only the necessary tools – sometimes without the typical command line tools or even without a shell. If you run code as serverless function, the possibilities are even more limited.

In this post I want to document for my future self how to run a simple IP connection test with a basic Python setup or Python container without any additional packages to be installed from PyPi.

The Test

Every Python installation comes with a package called socket. This package is intended for low-level networking tasks, like in our scenario creating a socket and attempting a connection. The connect() method uses the underlying operating system’s networking facilities to establish the connection. This typically involves creating a new socket and attempting to connect to the specified remote address using that socket. The details of how this is done can vary depending on the operating system and the type of socket being used.

In an example we want to test if a particular system (a Linux machine, a cluster node, a serverless function) can create a socket and make the connection to the target system.

Let’s create a short Python program called test-connect.py:

# Import the socket package
import socket

# Import the time package, so that we can demonstrate stuff
import time 

# Define the type of socket we want to create
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# Set a timeout for the connection attempt if we don't get a response after 5 seconds
s.settimeout(5.0)

# Make the connection
s.connect(("marcbrandner.com", 443))

# Print the result
print(s)

# Sleep for 5 minutes before closing, so that we can examine the open socket
time.sleep(300)

We can run this like any other Python program:

python3 test-connect.py

and will most likely receive the following output (linebreaks added for better readability):

<socket.socket fd=3, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0,
laddr=('10.0.2.15', 34066), raddr=('185.30.32.173', 443)>

The socket was successfully created and thus printing it gives some a few details:

• The properties family and type were already defined by us, nothing new to see here.
• laddr shows the IP address of the network interface on our side, which initiated the connection. Next to it, the outgoing port number is displayed.
• raddr shows the target address the remote hostname resolved to along with the port used.

In our sample snippet, we added a sleep to keep the socket open for a while. IYou will be able to see this open socket with the netstat tool if you look for the outgoing port:

# Linux/UNIX/OSX
netstat -an | grep 34066

# Windows
netstat -an | findstr 34066

This will show:

tcp        0      0 10.0.2.15:34066         185.30.32.173:443       ESTABLISHED

Now, what if we would try to connect, but the targeted hostname does not exist or cannot be resolved by our DNS? We can test this by replacing marcbrandner.com in the example code with foo.example. Running the program once again will quickly return:

Traceback (most recent call last):
  File "/home/marc/test.py", line 6, in <module>
    s.connect(("foo.example", 443))
socket.gaierror: [Errno -3] Temporary failure in name resolution

As expected, the connection fails, because the domain foo.example cannot be resolved.

(A quick note on *.example: This one will never resolve on the Internet, as EXAMPLE is reserved top-level domain. Great for our testing purposes!)

Another type of result is generated, if we try to connect to an existing host, but to a port which does not accept connections. We can test this by changing the target port 443 in our code to 123. It will give us the following result after the timeout of 5 seconds has passed:

Traceback (most recent call last):
  File "/home/vagrant/test.py", line 5, in <module>
    s.connect(("marcbrandner.com", 123))
TimeoutError: timed out

For troubleshooting, getting these different error messages is extremely helpful. That is how easy it is!

Now on to the next topic: How can we leverage this in the context of containers?

Testing From a Docker Container

There is a multitude of scenarios, in which you may have to conduct the above test using a plain container. I have seen environments, in which it was not allowed to install Python, but allowed to spawn containers with images pulled from DockerHub. You could also find yourself in a situation, where you run containers on a managed platform (like i.e. AWS ECS), but need to test your network security rules (on AWS for example VPC Security Groups).

For this purpose, we can run above code in a short chain of commands, which spawns a temporary container running the test:

docker run --rm -i docker.io/python:alpine python -c \
    'import socket; s = socket.socket(); s.settimeout(5.0); s.connect(("marcbrandner.com", 443)); print(s)'

Notice: If you are working on Windows, remove the backslashes that escape the linebreaks.

Some explanations:

• We are using the python:alpine container image from DockerHub, because it is the smallest one available to my knowledge.
• We use flags --rm and to auto-delete the container after the Python command has finished.
• We can omit the parameters socket.AF_INET and socket.SOCK_STREAM in the socket() method, because these are set by default.

You will get the same result as in the examples above, where we ran our code from the command line.

Testing From a Container on Kubernetes

The same command chain can be run in a similarly way on Kubernetes. We only need to add --restart=Never and give the Pod of the container a mandatory name.

kubectl run -i --rm test-connect --image=docker.io/python:alpine --restart=Never -- python -c \
    'import socket; s = socket.socket(); s.settimeout(5.0); s.connect(("marcbrandner.com", 443)); print(s)'

After approx. 50 MB of Python image are pulled, a Pod will be scheduled and run the Python command. The output will be visible in Pod’s log as well as on the shell, on which you executed the command.

There we go !

Limitations

This test does not take into account stateful firewalls blocking only specific protocols (i.e., a HTTP request would be allowed by the firewall, but an attempt to create an SSH tunnel is getting suppressed).

Also, connections may fail, if they require additional authentication or client-side configuration. An example would a connection test the connection to a LDAP system. While we are able to reach it on the network level (Layer 3) it could be possible that the connections are rejected, because the client uses the wrong LDAP parameters for the connection (Layer 7).

You should also be aware that connect() does exactly what is is instructed to do. If use the parameter constant socket.AF_INET, it will only attempt to resolve to an IPv4 address. If your target does not provide an IPv4 interface, but only IPv6 (socket.AF_INET6) the test will fail. A higher-level function trying different combinations would be create_connection().

Conclusion

Python is a swiss-army knife that carrys along a bunch of helpful default packages. Small Python container images can be useful for network tests in many situations, even if you do not have direct access on the operating system level, but if you are constrained to running code from within containers or serverless functions.