Port Scanner Client and Server

portchecker.py

import typer
from loguru import logger as log
import socket
from multiprocessing import Pool

DEFAULT_MESSAGE = "Thank you sir, may I have another"

PORTS = [
    443
]
MAX_REQUESTS = 5
BIND_ADDR = "0.0.0.0"  # All Addresses

app = typer.Typer()


def create_socket(TCP_PORT, hostname):
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind((hostname, TCP_PORT))
    server_socket.listen(1)
    return server_socket


def setup_process_listener(port, hostname):
    server = create_socket(port, hostname)
    while True:
        connection, address = server.accept()
        buf = connection.recv(1024)
        log.info(f"Address: {address}:{port}, Was sent buffer: {buf}")
        connection.close()


@app.command()
def server(hostname: str = BIND_ADDR):
    number_ports = len(PORTS)
    with Pool(processes=number_ports) as pool:
        args = [(port, hostname) for port in PORTS]
        log.info(f"Started port listeners: {PORTS}")
        pool.starmap(setup_process_listener, args)


@app.command()
def client(target:str=None, message:str=DEFAULT_MESSAGE):
    if target is None:
        target="127.0.0.1"
    log.info(f"Starting to iterate through ports of host {target}...")
    for port in PORTS:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(5)        
        try:
            sock.connect((target, port))
            sock.sendall(message.encode())
            log.info(f"Connected and sent succesfully: {target}:{port}")
        except socket.timeout as e:
            log.warning(f"socket.timeout error: port={port}, {e}")
        except TimeoutError as e:
            log.warning(f"Timeout error: port={port}, {e}")
        except ConnectionRefusedError as e:
            log.warning(f"Connection refused: port={port}, {e}")

if __name__ == "__main__":
    app()