Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
TCP Network scanner using asyncio for Python 3.7
import asyncio
import ipaddress
import re
import sys
def ipsort(t):
'''used to sort output by ipaddress, then port'''
return tuple([*map(int, t[0].split('.')), t[1]])
def eprint(*args, **kwargs):
print(*args, **kwargs, file=sys.stderr)
def parseports(portstring):
syntax: port,port-range,...
use regex to verify input validity, then create a tuple of
ports used in port scan. there definitely some room for optimization
here, but it won't matter much. go optimize the coroutines instead.
if not re.match(r'[\d\-,\s]+', portstring):
raise ValueError('Invalid port string')
ports = []
portstring = list(filter(None, portstring.split(',')))
for port in portstring:
if '-' in port:
port = [int(p) for p in port.split('-')]
except ValueError:
raise ValueError('Are you trying to scan a negative port?')
for p in range(port[0], port[1]+1):
for port in ports:
if not (-1 < port < 65536):
raise ValueError('Ports must be between 0 and 65535')
return tuple(set(ports))
def fancy_print(data, csv):
if csv:
fmt = '{},{}'
fmt = '{:<15} :{}'
for datum in data:
async def task_worker(task_queue, out_queue):
'''pull connection information from queue and attempt connection'''
while True:
ip, port, timeout = (await task_queue.get())
conn = asyncio.open_connection(ip, port)
await asyncio.wait_for(conn, timeout)
except asyncio.TimeoutError:
out_queue.put_nowait((ip, port))
async def task_master(
network: str, portrange: str, timeout: float,
task_queue: asyncio.Queue, scan_completed: asyncio.Event):
'''add jobs to a queue, up to ``MAX_NUMBER_WORKERS'' at a time'''
network = network.replace('/32', '')
# check to see if we are scanning a single host...
hosts = [str(ipaddress.ip_address(network)),]
except ValueError:
# ...or a CIDR subnet.
hosts = map(str, ipaddress.ip_network(network).hosts())
for ip in hosts:
for port in portrange:
await task_queue.put((ip, port, timeout))
async def main(network, ports=None, timeout=0.1, csv=False):
main task coroutine which manages all the other functions
if scanning over the internet, you might want to set the timeout
to around 1 second, depending on internet speed.
task_queue = asyncio.Queue(maxsize=MAX_NUMBER_WORKERS)
out_queue = asyncio.Queue()
scan_completed = asyncio.Event()
scan_completed.clear() # progress the main loop
if ports is None: # list of common-ass ports
ports = ("9,20-23,25,37,41,42,53,67-70,79-82,88,101,102,107,109-111,"
ports = parseports(ports)
# initialize task to add scan info to task queue
tasks = [asyncio.create_task(
task_master(network, ports, timeout, task_queue, scan_completed)
# initialize workers
for _ in range(MAX_NUMBER_WORKERS):
tasks.append(asyncio.create_task(task_worker(task_queue, out_queue)))
eprint('scanning . . .')
await scan_completed.wait() # wait until the task master coro is done
await task_queue.join() # wait for workers to finish
for task in tasks:
await asyncio.gather(*tasks, return_exceptions=True)
eprint('gathering output . . .')
openports = []
while out_queue.qsize():
fancy_print(openports, csv=csv)
eprint('shutting down . . .')
if __name__ == '__main__':
# import argparse?
if len(sys.argv) < 2:
'TCP Network scanner using asyncio module for Python 3.7+',
"Scan ports in ``portstring'' or common ports if blank."
'Port string syntax: port, port-range ...',
f'Usage: {sys.argv[0]} network [portstring]',
raise SystemExit
elif len(sys.argv) == 2:[1]))
else:[1], ''.join(sys.argv[2:])))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.