Python program to check if a network port is open
In this Python programming article, you are going to learn how to check if a particular port is open. We are using socket.socket() and socket.connect_ex() to check if a network port is open. Before going into programming first let’s understand what is Socket Programming and some basic terminology of socket programming.
Before we go forward, we first need to know about socket programming.
What is Socket Programming
Socket Programming is a way of connecting two nodes on a network in order to communicate with each other. One socket or node listens to a particular port at an IP address and the other socket reaches out to the other port for creating a connection. In simple terms, they are called client-server programming. They are the backbones behind web browsing.
Python has a socket library for implementing socket programming. Let’s make a simple socket
import socket as sock s = sock.socket(sock.AF_INET, sock.SOCK_STREAM)
Here we made a socket and passed two parameters. The first parameter is AF_INET refers to address-family ipv4 and the second parameter SOCK_STREAM refers to the connection-oriented TCP protocol.
Checking if the network port is open using Python programming
Below are given steps to perform our task in Python:
1) First import the socket library as sock.
2) Create a socket object by calling sock.socket(family, type) with the family set to sock.AF_INET and the type set to SOCK_STREAM.
3) Create a destination as a tuple containing the IP address and the desired port number.
4) To check port is open call sock.connect_ex(destination) which returns 0 if the port is open, 1 if the port is close.
5) After checking the port is open, then socket by using socket.close().
import socket as sock create_socket = sock.socket(sock.AF_INET, sock.SOCK_STREAM) destination = ("127.0.0.1", 80) result = create_socket.connect_ex(destination) if result == 0: print("Port is open") else: print("Port is not open") create_socket.close() betrcode / README.md
Using Python to check if remote port is open and accessible.
- Open a Python prompt
- Run the script in is_port_open.py
- Call isOpen with a host and a port parameter. Example: isOpen(«www.google.com», 80)
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
| import socket |
| def isOpen ( ip , port ): |
| s = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) |
| try : |
| s . connect (( ip , int ( port ))) |
| s . shutdown ( 2 ) |
| return True |
| except : |
| return False |
Maybe improve script to take command line arguments.
Simple and precise! From what I can tell, however, this does not seem to be working. Take for example the following:
This will never finish. Should I be doing something different?
Could it because of the timeout?
If that’s the case then maybe could add socket.settimeout()?
for i in range(10): try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(1) s.connect((ip, port)) print ip + " is UP" except: time.sleep(1) finally: s.shutdown(socket.SHUT_RDWR) s.close() for i in range(10): try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(1) s.connect((ip, port)) print ip + " is UP" except: time.sleep(1) finally: s.shutdown(socket.SHUT_RDWR) s.close() This doesn’t work, please check code before replying. If the connection fails, the finally block runs, but s is not connected, therefore shutdown fails.
Full example, tested with a retry and delay between retries:
#!/usr/bin/python import socket import time ip = "google.com" port = 443 retry = 5 delay = 10 timeout = 3 def isOpen(ip, port): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(timeout) try: s.connect((ip, int(port))) s.shutdown(socket.SHUT_RDWR) return True except: return False finally: s.close() def checkHost(ip, port): ipup = False for i in range(retry): if isOpen(ip, port): ipup = True break else: time.sleep(delay) return ipup if checkHost(ip, port): print ip + " is UP" Another option for anyone interested is I’ve incorporated the above into a Python app with argument parsing that uses SMTP to send alerts. It also runs in docker: https://github.com/Fmstrat/server-monitor
This also now support UDP.
Cheers for sharing this code. It worked really well for me.
Yes an old thread but still usefull. Can anyone shed some light as to how you can have multiple hosts that it polls until the app is exited?
as in..
domain1.com host id UP
domain2.com host id UP
domain3.com host id DOWN
domain4.com host id UP
#!/usr/bin/python3
import sys
import socket
import time
ip = «domain.com»
port = 443
retry = 1
delay = 1
timeout = 2
def isOpen(ip, port):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
s.connect((ip, int(port)))
s.shutdown(2)
return True
except:
return False
def checkHost(ip, port):
ipup = False
for i in range(retry):
if isOpen(ip, port):
ipup = True
break
else:
time.sleep(delay)
return ipup
if checkHost(ip, port):
print ip + » port » + str(port) + u» is \u001b[32;1mUP!\u001b[0m»
else:
print ip + » port » + str(port) + u» is \u001b[31;1mDOWN!\u001b[0m»
async def wait_host_port(host, port, duration=10, delay=2): """Repeatedly try if a port on a host is open until duration seconds passed Parameters ---------- host : str host ip address or hostname port : int port number duration : int, optional Total duration in seconds to wait, by default 10 delay : int, optional delay in seconds between each try, by default 2 Returns ------- awaitable bool """ tmax = time.time() + duration while time.time() tmax: try: _reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=5) writer.close() await writer.wait_closed() return True except: if delay: await asyncio.sleep(delay) return False
async def wait_host_port(host, port, duration=10, delay=2): """Repeatedly try if a port on a host is open until duration seconds passed Parameters ---------- host : str host ip address or hostname port : int port number duration : int, optional Total duration in seconds to wait, by default 10 delay : int, optional delay in seconds between each try, by default 2 Returns ------- awaitable bool """ tmax = time.time() + duration while time.time() tmax: try: _reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=5) writer.close() await writer.wait_closed() return True except: if delay: await asyncio.sleep(delay) return False
Thanks this is EXACTLY what I was looking for.
async def wait_host_port(host, port, duration=10, delay=2): """Repeatedly try if a port on a host is open until duration seconds passed Parameters ---------- host : str host ip address or hostname port : int port number duration : int, optional Total duration in seconds to wait, by default 10 delay : int, optional delay in seconds between each try, by default 2 Returns ------- awaitable bool """ tmax = time.time() + duration while time.time() tmax: try: _reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=5) writer.close() await writer.wait_closed() return True except: if delay: await asyncio.sleep(delay) return False
While reading this, I had tears of joy. Many thanks!
Very good, replace ping command,haha!
Thank you very much. Works out of the box 🙂
Small note, this can be sped up by using connect_ex instead of connect . The first call will not raise any exceptions (except for No host found , but that’s standard) and is faster than the second one. Example:
def isOpen(ip, port): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(5) try: is_open = s.connect_ex((ip, int(port))) == 0 # True if open, False if not if is_open: s.shutdown(socket.SHUT_RDWR) except Exception: is_open = False s.close() return is_open
async def wait_host_port(host, port, duration=10, delay=2): """Repeatedly try if a port on a host is open until duration seconds passed Parameters ---------- host : str host ip address or hostname port : int port number duration : int, optional Total duration in seconds to wait, by default 10 delay : int, optional delay in seconds between each try, by default 2 Returns ------- awaitable bool """ tmax = time.time() + duration while time.time() tmax: try: _reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=5) writer.close() await writer.wait_closed() return True except: if delay: await asyncio.sleep(delay) return False
Thank you for the code. It works, but in python3.6, I get an exception like this ‘StreamWriter’ object has no attribute ‘wait_closed’ . Could you tell me how to fix it. Please
Footer
You can’t perform that action at this time.