Python scanner for the first free port in a range

Question

I have a distributed application (YARN), which runs a WebApp.

This application use a default port to start (8008), before I start I need to check if port is in use.

A container may run in the same virtual machine, hence port may be in use. (Max I have 4 containers in WebApp).

I created the following code which seem to work, but want to see if there are some clean ups/improvements suggested.

def port_in_use(port):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    result = sock.connect_ex(('127.0.0.1', port))
    if result == 0:
      return True
    else:
      return False


def start_dashboard():    
    base_port = os.getenv('DASHBOARD_PORT_ENV_VAR', 8008)
    scan_ports = True
    attempts = 0
    max_attempts = 10
    while(scan_ports and attempts <= max_attempts):
      if port_in_use(base_port):
         base_port += 1
         attempts += 1
      else:
         scan_ports = False
    if attempts == max_attempts:
       raise IOError('Port in use')
    dashboard.configure(port=base_port)
    dashboard.launch()

Answer 1

Your code has some incorrect assumptions.

• an application may listen on a specific address/port combination; 127.0.0.1:port can be available while *:port is not.

• an application may bind a port without listening. Connects will fail, but so will your own bind.

• a firewall or other mechanism can interfere with connections, generating false positives in your scan.

The reliable approach is to bind the port, just as your dashboard will, and then release it.

result = sock.bind(('', port))
sock.close()

---

You'll need to catch the exception and this is a good opportunity to move the whole thing into a function. That will make the start_dashboard logic cleaner and get rid of boolean loop-terminator scan_ports. Just exit the loop by returning the answer.

def next_free_port( port=1024, max_port=65535 ):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    while port <= max_port:
        try:
            sock.bind(('', port))
            sock.close()
            return port
        except OSError:
            port += 1
    raise IOError('no free ports')

def start_dashboard():

    # pass optional second parameter "max_port" here, else scan until a free one is found
    port = next_free_port( os.getenv('DASHBOARD_PORT_ENV_VAR', 8008) )

    dashboard.configure(port=port)
    dashboard.launch()

You can use netcat to make ports in-use for testing: nc -l -p 9999 will listen on port 9999; press control-C to end it.