socket-watchdog.py

socket-watchdog.py

import datetime
from rich.console import Console
from rich.pretty import pprint
import pandas as pd

console = Console()

debug = False


class SocketWatchdog:
    def __init__(self, host, port, proto="tcp"):
        self.last_updated = None
        self.host_ip = host
        self.port_number = port
        self.port_isopen = None
        self.port_proto = proto
        self.polling_interval = None

        self.update_socket()

    def is_port_open(self):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(1)
        try:
            s.connect((self.host_ip, self.port_number))
            if debug:
                console.print(
                    f"{self.host_ip} {self.port_number} -- OPEN", style="green"
                )
            return True
        except:
            if debug:
                console.print(
                    f"{self.host_ip} {self.port_number} -- CLOSED", style="red"
                )
            return False
        finally:
            s.close()

    def update_socket(self):
        self.port_isopen = self.is_port_open()
        self.last_updated = datetime.datetime.now()


socketList = [SocketWatchdog("127.0.0.1", 5000), SocketWatchdog("127.0.0.1", 80)]

socketState = []

for socket in socketList:
    socketState.append(
        [
            socket.host_ip,
            socket.port_number,
            socket.port_proto,
            socket.port_isopen,
            socket.last_updated,
        ]
    )

df = pd.DataFrame(
    socketState, columns=["HOST IP", "PORT", "PROTOCOL", "IS OPEN", "LAST UPDATED"]
)
pprint(df)