The start of a Python port scanner

Question

My concerns:

• Is this module thread safe?
• Is there any obvious performance issues I missed?
• Is there a better container type to use for GENERATED_ADDRESSES other than a list?
• Is there any flags I can pass to socket to make the try / connect any faster?

---

#-*-coding:utf8;-*-
#qpy:3
#qpy:console

import socket
import os
import random
import struct
import time
import threading

# This is used to set a default timeout on socket
# objects.
DEFAULT_TIMEOUT = 0.30

# This is used for checking if a call to socket.connect_ex
# was successful.
SUCCESS = 0

# GENERATED_ADDRESSES is used to store the addresses that
# generate_ip generates,
# The reason i have to log the addresses
# is because i don't want generate_ip to produce
# the same address twice.
# generate_ip will keep checking GENERATED_ADDRESSES for
# each addresses generated until it produces an address
# that's not in GENERATED_ADDRESSES.
GENERATED_ADDRESSES = []

LOCK = threading.Lock()

def elapsed(func):
    def wraps(*args, **kw):
        start = time.time()
        retval = func(*args, **kw)
        stop = time.time()
        elap = stop - start
        print(func.__name__, elap)
        return retval
    return wraps


def check_port(*host_port, timeout=DEFAULT_TIMEOUT):
    ''' Try to connect to a specified host on a specified port.
    If the connection takes longer then the TIMEOUT we set we assume
    the host is down. If the connection is a success we can safely assume
    the host is up and listing on port x. If the connection fails for any
    other reason we assume the host is down and the port is closed.'''

    # Create and configure the socket.
    sock = socket.socket()
    sock.settimeout(timeout)

    # the SO_REUSEADDR flag tells the kernel to reuse a local 
    # socket in TIME_WAIT state, without waiting for its natural
    # timeout to expire.
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    # Like connect(address), but return an error indicator instead
    # of raising an exception for errors returned by the C-level connect() 
    # call (other problems, such as “host not found,” can still raise exceptions). 
    # The error indicator is 0 if the operation succeeded, otherwise the value of 
    # the errnovariable. This is useful to support, for example, asynchronous connects.
    opened = sock.connect_ex(host_port) is SUCCESS

    # Mark the socket closed. 
    # The underlying system resource (e.g. a file descriptor)
    # is also closed when all file objects from makefile() are closed.
    # Once that happens, all future operations on the socket object will fail. 
    # The remote end will receive no more data (after queued data is flushed).
    sock.close()

    # return True if port is open or False if port is closed.
    return opened


def is_private_ip(ip):
    ''' return True is ip is local and False if it is not.'''
    return ipaddress.ip_address(ip).is_private


def random_ip(**kw):

    while 1:
        _ip = socket.inet_ntoa(struct.pack('>I', random.randint(1, 0xffffffff)))
        if _ip not in GENERATED_ADDRESSES:
            ip = _ip
            # Are we running in a sub thread?
            # If so lock while appending to GENERATED_ADDRESSES.
            if threading.activeCount() > 1:
                with LOCK:
                    GENERATED_ADDRESSES.append(ip)
            # If running in the main thread we don't have to
            # lock before appending to GENERATED_ADDRESSES.
            else:
                GENERATED_ADDRESSES.append(ip)
            break

    # Only produce private addresses?
    if kw.get('private_only', 0) is True:
        # Keep generating addresses until we generate
        # a private address.
        while 1:
           if is_private_ip(ip):
               return ip

    # only produce public addresses?
    if kw.get('public_only', 0) is True:
        while 1:
            # Keep generating addresses until we
            # generate a private_address.
            if not is_private_ip(ip):
                return ip

    # Just return the addresses regardless of it public or private
    # orientation.
    return ip


while 1:
    host = random_ip(public_only=1)
    opened = check_port(host, 80)          
    if opened:
        print(host, len(GENERATED_ADDRESSES))

Answer 1

• Is there a better container type to use for GENERATED_ADDRESSES other than a list?

  You could use a set() this will save time with comparing if the ip_address is already used.

  x in set will be average of O(1) instead of O(n) for x in list, see Complexity in Python

• if kw.get('private_only', 0) is True:

  This does not work what you expect from it.

  Example

  >>> private_only = 1
  >>> private_only is True
  False
  

  That is because the is keyword also checks for the type, and will only return True if the type is a boolean. In your code it is an integer.

  This can be solved by either using the bool type by calling it like this: host = random_ip(public_only=True)

  or discarding the is True altogether since, if 1: will evaluate to True

• return ipaddress.ip_address(ip).is_private

  I think you forgot to import ipaddress.

• # only produce public addresses? has some flawed logic in it.

  This will not work, it doesn't generate new adresses so this will be stuck in a loop forever

  while 1:
      if is_private_ip(ip):
          return ip
  

  When you check for private addresses, and the address is not private, it will still return an ip. Because you always return an ip at the end of your random_ip function

  By removing the break, and checking for the private/public range in the while loop you can fix the logic.

Alternative

def random_ip(**kw):
    while True:
        _ip = socket.inet_ntoa(struct.pack('>I', random.randint(1, 0xffffffff)))
        if _ip not in GENERATED_ADDRESSES:
            ip = _ip
            GENERATED_ADDRESSES.add(ip)

        # Only produce private addresses
        if kw.get('private_only', 0):
            if is_private_ip(ip):
                yield ip

        # only produce public addresses
        elif kw.get('public_only', 0):
            if is_private_ip(ip):
                yield ip

        # Return if none is specified
        else:
            yield ip

while True:
    for host in random_ip(private_only=True):
        if check_port(host, 80) :
            print(host, len(GENERATED_ADDRESSES))

Answer 2

Your code's quite bad.

• I recomend you read and follow PEP 8, this is as your code is hard to read and not idiomatic.

• "Don't compare boolean values to True or False using ==" or is.

  Your code doesn't work correctly as 1 is True is False.

• From my POV your code doesn't need to be thread safe, I may be wrong. But using GENERATED_ADDRESSES is not safe in the slightest. Just make it a local variable. And change your algorithm.

• IMO you should generate all ips in order, not all ips in a random order. The latter just adds work.
• If you stick to keeping with a random order for IPs, I'd recomend you use a set for GENERATED_ADDRESSES. However that still may not be good enough, you should also look into random sorts too.
• You don't use elapsed, remove it.
• while 1 over for loops is dumb. Change your code to yield ips, and filter them using generators.

And so I'd change your code to:

import socket
import random
import struct

DEFAULT_TIMEOUT = 0.30
SUCCESS = 0


def check_port(host, port, timeout=DEFAULT_TIMEOUT):
    with socket.socket() as sock:
        sock.settimeout(timeout)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        opened = sock.connect_ex(host, port) is SUCCESS
        return opened


def is_private_ip(ip):
    ''' return True is ip is local and False if it is not.'''
    return ipaddress.ip_address(ip).is_private


def random_ips():
    prev_ips = {}
    while True:
        ip = random.randint(1, 0xffffffff)
        if ip not in prev_ips:
            prev_ips.add(ip)
            yield socket.inet_ntoa(struct.pack('>I', ip))


def private_only(ips):
    for ip in ips:
        if is_private_ip(ip):
            yield ip


def public_only(ips):
    for ip in ips:
        if not is_private_ip(ip):
            yield ip


ips = random_ips()
public_ips = public_only(ips)
for host in public_ips:
    opened = check_port(host, 80)
    if opened:
        print(host)