404Forbiden

✏️ Editing: tcp.py

import agent_util
import sys
import os
import socket
import time
from datetime import datetime, timedelta

DEFAULT_TIMEOUT = 15

class TCPPlugin(agent_util.Plugin):
    textkey = "tcp"
    label = "TCP/IP port"

@classmethod
    def get_metadata(self, config):
        status = agent_util.SUPPORTED
        msg = None

data = {
            "tcp_status": {
                "label": "TCP port status",
                "options": None,
                "option_string": True,
                "status": status,
                "error_message": msg,
                "unit": "boolean",
            },
            "udp_status": {
                "label": "UDP port status",
                "options": None,
                "option_string": True,
                "status": status,
                "error_message": msg,
                "unit": "boolean",
            },
            "tcp_latency": {
                "label": "TCP port latency",
                "options": None,
                "option_string": True,
                "status": status,
                "error_message": msg,
                "unit": "secs",
            },
        }

# for Linux, add the network connection state metrics
        if "linux" in sys.platform and agent_util.which("netstat"):
            CONNECTION_STATUS = [
                "CLOSE_WAIT",
                "CLOSED",
                "ESTABLISHED",
                "FIN_WAIT_1",
                "FIN_WAIT_2",
                "LAST_ACK",
                "LISTEN",
                "SYN_RECV",
                "SYN_RECEIVED",
                "SYN_SEND",
                "TIME_WAIT",
            ]
            data["network.connections.tcp"] = {
                "label": "TCP Connections",
                "options": CONNECTION_STATUS,
                "status": status,
                "error_message": msg,
                "unit": "connections",
            }
            data["network.connections.tcpv6"] = {
                "label": "TCPv6 Connections",
                "options": CONNECTION_STATUS,
                "status": status,
                "error_message": msg,
                "unit": "connections",
            }
            data["network.connections.udp"] = {
                "label": "UDP Connections",
                "options": CONNECTION_STATUS,
                "status": status,
                "error_message": msg,
                "unit": "connections",
            }
            data["network.connections.udpv6"] = {
                "label": "UDPv6 Connections",
                "options": CONNECTION_STATUS,
                "status": status,
                "error_message": msg,
                "unit": "connections",
            }

return data

def check(self, textkey, option, config):
        # special logic for getting TCP/UDP connection counts
        if textkey.startswith("network.connections"):
            ret, output = agent_util.execute_command("netstat -tunap")

lines = output.splitlines()[1:]
            header = True
            connection_stats = {}

for line in lines:
                if line.startswith("Proto"):
                    header = False
                    continue
                elif header:
                    continue

line = line.split()
                proto = line[0].strip()
                state = line[5].strip()

if connection_stats.get(proto):
                    if connection_stats[proto].get(state):
                        connection_stats[proto][state] += 1
                    else:
                        connection_stats[proto][state] = 1
                else:
                    connection_stats[proto] = {}
                    connection_stats[proto][state] = 1

# establish a new variable for clarity
            conn_state = option

if textkey == "network.connections.tcp":
                return float(connection_stats.get("tcp", {}).get(conn_state, 0))
            elif textkey == "network.connections.tcpv6":
                return float(
                    connection_stats.get("tcp6", connection_stats.get("tcpv6", {})).get(
                        conn_state, 0
                    )
                )
            elif textkey == "network.connections.udp":
                return float(connection_stats.get("udp", {}).get(conn_state, 0))
            elif textkey == "network.connections.udpv6":
                return float(
                    connection_stats.get("udp6", connection_stats.get("udpv6", {})).get(
                        conn_state, 0
                    )
                )
            else:
                self.log.error("UNKNOWN NETWORK CONNECTION TEXTKEY- %s" % textkey)
                return None

option_parts = option.split(":")

if len(option_parts) == 2:
            ip_address = option_parts[0]
            port = int(option_parts[1])
        # default to localhost if ip not provided
        elif len(option_parts) == 1:
            ip_address = "localhost"
            port = int(option_parts[0])
        else:
            self.log.info("Invalid TCP/UDP port specification: %s" % option)
            return None

if textkey == "udp_status":
            ret, output = agent_util.execute_command(
                "netstat -unap | grep %s | grep %s" % (port, ip_address)
            )

self.log.debug("udp_status netstat output: %s" % output)

lines = output.splitlines()
            header = True

for line in lines:
                line = line.split()
                proto = line[0].strip()
                local_addr = line[3].strip()

if proto == "udp":
                    return 1

# return false
            return 0

else:
            start = time.time()
            success = None
            try:
                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                to = config.get("tcp.timeout")
                timeout = to if to else DEFAULT_TIMEOUT
                s.settimeout(timeout)
                s.connect((ip_address, port))
            except socket.timeout:
                success = False
            except Exception:
                t, e = sys.exc_info()[:2]
                self.log.exception("TCP Connection error: %s" % e)
                success = False
            else:
                s.shutdown(socket.SHUT_RDWR)
                s.close()
                success = True
            duration = time.time() - start

if textkey == "tcp_status":
                return success and 1 or 0
            elif textkey == "tcp_latency":
                return success and duration or None
            else:
                return None

Sid Gifari From Gifari Industries - BD Cyber Security Team