Using Python to check if remote port is open and accessible.
Taken from http://snipplr.com/view/19639/test-if-an-ipport-is-open/
Usage:
| 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 |
Simple and precise! From what I can tell, however, this does not seem to be working. Take for example the following:
isOpen('www.google.com', 79)
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()?
set timeout explicitly
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()
set timeout explicitly
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
What Im using so far...
#!/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"
An async version:
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
An async version:
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.
An async version:
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
An async version:
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
Small note, this can be sped up by using
connect_exinstead ofconnect. The first call will not raise any exceptions (except forNo host found, but that's standard) and is faster than the second one. Example:
Thanks for sharing this Oscar.
An async version:
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 FalseThank 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
the .wait_closed() method was added in python 3.7
Maybe improve script to take command line arguments.