FM / dedrads

/opt/dedrads

Name	Size	Mode	Actions
account_review/	-	0555	rm
check_software_mods/	-	0555	rm
cms_tools/	-	0555	rm
etc/	-	0555	rm
extras/	-	0555	rm
mailparse/	-	0555	rm
mysql/	-	0555	rm
nlp_scripts/	-	0555	rm
oldrads/	-	0555	rm
perl/	-	0555	rm
provision/	-	0555	rm
python/	-	0555	rm
suspended/	-	0555	rm
__pycache__/	-	0555	rm
account-review	799	0555	edit dl rm
allfw	6259	0555	edit dl rm
alp.py	16553	0555	edit dl rm
autossl_sync	6267	0555	edit dl rm
check_apache	9205	0555	edit dl rm
check_bandwidth	4546	0555	edit dl rm
check_boxtrapper	1684	0555	edit dl rm
check_cpu	12212	0555	edit dl rm
check_crons	1827	0555	edit dl rm
check_dcpumon	1973	0555	edit dl rm
check_dns	1827	0555	edit dl rm
check_domlogs	61	0555	edit dl rm
check_exim	3731	0555	edit dl rm
check_hacks	22604	0555	edit dl rm
check_imap	5249	0555	edit dl rm
check_io	6635	0555	edit dl rm
check_max_children	16341	0555	edit dl rm
check_mem	1660	0555	edit dl rm
check_misc	1510	0555	edit dl rm
check_pacct	5201	0555	edit dl rm
check_php	4344	0555	edit dl rm
check_pop3	2295	0555	edit dl rm
check_prov.py	4714	0555	edit dl rm
check_server	696	0555	edit dl rm
check_size	490	0555	edit dl rm
check_software	9849	0555	edit dl rm
check_spamd	2426	0555	edit dl rm
check_traffic	1933	0555	edit dl rm
check_user	1887	0555	edit dl rm
clean_exim.py	6390	0555	edit dl rm
clusterfix	5343	0555	edit dl rm
cmspass.py	3970	0555	edit dl rm
cms_check.py	9825	0555	edit dl rm
cms_counter.py	36734	0555	edit dl rm
cms_creds	713	0555	edit dl rm
cms_dumpdb	4485	0555	edit dl rm
cms_pw	1416	0555	edit dl rm
cpanel-api	7040	0555	edit dl rm
cpanel_postgres_manager.py	29004	0555	edit dl rm
cpmerge	4631	0555	edit dl rm
cpumon	2195	0555	edit dl rm
dcpumon.pl	1239	0555	edit dl rm
dedcheck	18127	0555	edit dl rm
dns-sync	14386	0555	edit dl rm
docroot.py	4425	0555	edit dl rm
du-tree	7450	0555	edit dl rm
envinfo.py	6633	0555	edit dl rm
exclude_rbl.py	5249	0555	edit dl rm
extract-vhost	2691	0555	edit dl rm
filescan	3566	0555	edit dl rm
find_warez	1460	0555	edit dl rm
first_setup.py	19120	0555	edit dl rm
fixwpcron.py	4822	0555	edit dl rm
fix_dns_cluster	353	0555	edit dl rm
goaccess	1217720	0555	edit dl rm
hostsfilemods	4467	0555	edit dl rm
imap_io	2323	0555	edit dl rm
innodb_converter.py	9343	0555	edit dl rm
killall911	530	0555	edit dl rm
killdns	941	0555	edit dl rm
lastcommcache.sh	1374	0555	edit dl rm
legal_lock_down.sh	7450	0555	edit dl rm
lil-cpanel	52002	0555	edit dl rm
listacct	4575	0555	edit dl rm
mailscan	4544	0555	edit dl rm
mail_sources.py	7407	0555	edit dl rm
megaclisas-status	38664	0555	edit dl rm
migrate2central.sh	14418	0555	edit dl rm
modify-account	40631	0555	edit dl rm
modsec_disable.py	11461	0555	edit dl rm
monarxctl	8912	0555	edit dl rm
msp.pl	27714	0555	edit dl rm
mysql_dstat	474	0555	edit dl rm
mysql_selector.py	73102	0555	edit dl rm
nlp	4582	0555	edit dl rm
procscrape	1863	0555	edit dl rm
quarantine	11083	0555	edit dl rm
radsfunctions.sh	863	0555	edit dl rm
recent-cp	16148	0555	edit dl rm
remote_dump	18347	0555	edit dl rm
reset_cpanel	3205	0555	edit dl rm
server-load	2192	0555	edit dl rm
show-conns	20105	0555	edit dl rm
show-conns-adv.py	54667	0555	edit dl rm
sparta.py	382498	0555	edit dl rm
sqltop	27129	0555	edit dl rm
telcheck	2411	0555	edit dl rm
temprootreset	186	0555	edit dl rm
unsuspend_user	1307	0555	edit dl rm
unsusprunner.sh	1840	0555	edit dl rm
updatednsadmin	526	0555	edit dl rm
update_spf	14972	0555	edit dl rm
upgrade-check	9216	0555	edit dl rm
vhost_data.py	6221	0555	edit dl rm
wp-xray.sh	85816	0555	edit dl rm

Edit: /opt/dedrads/show-conns-adv.py (54667B)

#!/usr/lib/rads/venv/bin/python3

import argparse
import os
import re
import shutil
import subprocess
import sys
import time
from collections import defaultdict
from datetime import datetime
from ipaddress import ip_address, ip_network
from prettytable import PrettyTable
from rads import color

ss_path = shutil.which("ss")

SKIPPED_SERVER_IPS = {'*', '127.0.0.1', '[::]', '::', '::1'}
SKIPPED_CLIENT_IPS = {'*'}

HOT_IP_THRESHOLD = 200
HOT_SERVICE_THRESHOLD = 2000
BRUTE_FORCE_MIN_UNIQUE_IPS = 30
BRUTE_FORCE_MAX_AVG_PER_IP = 10
SUBNET_FLOOD_THRESHOLD = 100
SUSPICIOUS_TCP_STATES = {'SYN-RECV', 'SYNRECV'}
DEFAULT_WATCH_INTERVAL = 5
DEFAULT_MAX_TABLES_PER_ROW = 2
MAIL_SERVICES_MAX_PER_ROW = 2
ESTIMATED_TABLE_WIDTH = 44
WEB_EXTRA_PORTS = {8080, 8443, 8000, 8888}
INTERNAL_WEB_DOMINANCE_THRESHOLD = 0.5

PROCESS_SERVICE_HINTS = (
    ('sshd', 'ssh'),
    ('exim', 'smtp'),
    ('dovecot', 'imap'),
    ('pop3', 'pop3'),
    ('httpd', 'http'),
    ('apache2', 'http'),
    ('nginx', 'http'),
    ('litespeed', 'http'),
    ('openlitespeed', 'http'),
    ('cpsrvd', 'cpanel'),
    ('cwpsrv', 'cwp_admin'),
    ('pure-ftpd', 'ftp'),
    ('proftpd', 'ftp'),
    ('mysqld', 'mysql'),
    ('mariadbd', 'mysql'),
    ('named', 'dns'),
    ('pdns_server', 'dns'),
)

def is_cwp_server():
    if os.path.isfile('/usr/local/cpanel/cpanel'):
        return False
    return os.path.isdir('/usr/local/cwpsrv') or os.path.isdir('/opt/cwprads')

def auth_attack_services():
    services = {'smtp', 'imap', 'pop3', 'ssh', 'webmail', 'ftp', 'mysql'}
    if is_cwp_server():
        return services | {'cwp_user', 'cwp_admin'}
    return services | {'cpanel', 'whm'}

class ServiceRegistry:
    """Maps services to ports."""

def __init__(self):
        if is_cwp_server():
            panel_user, panel_admin = 'cwp_user', 'cwp_admin'
        else:
            panel_user, panel_admin = 'cpanel', 'whm'
        self.panel_user = panel_user
        self.panel_admin = panel_admin
        self.services = {
            'tcp': {
                'http': [80, 443],
                'imap': [143, 993],
                'pop3': [110, 995],
                'smtp': [25, 465, 587],
                'ftp': [20, 21],
                'ssh': [22, 2222],
                panel_user: [2082, 2083],
                panel_admin: [2086, 2087],
                'webmail': [2095, 2096],
                'mysql': [3306],
            },
            'udp': {'dns': [53], 'ntp': [123]},
        }
        self.port_to_service_name_map = {'tcp': {}, 'udp': {}}
        for protocol, service_to_ports_map in self.services.items():
            for service_name, port_list in service_to_ports_map.items():
                for port_number in port_list:
                    self.port_to_service_name_map[protocol][
                        port_number
                    ] = service_name

# Gets a service name.
    def get_service_name(self, port_number: int, protocol: str) -> str:
        return self.port_to_service_name_map.get(protocol, {}).get(
            port_number, 'other'
        )

# Checks if port is tracked.
    def is_tracked_port(self, port_number: int, protocol: str) -> bool:
        return port_number in self.port_to_service_name_map.get(protocol, {})

# Gets all tracked ports.
    def get_all_ports(self, protocol: str) -> set:
        return set(self.port_to_service_name_map.get(protocol, {}).keys())

class ConnectionReporter:
    """Manages connection reporting."""

def __init__(self):
        self.service_registry = ServiceRegistry()
        self.command_line_args = self.parse_command_line_arguments()
        self.selected_ports_by_protocol = {'tcp': set(), 'udp': set()}
        self.requested_tcp_ports = set()
        self.requested_udp_ports = set()
        self.use_explicit_port_filter = False
        self.tcp_connections_by_port = defaultdict(lambda: defaultdict(int))
        self.udp_connections_by_port = defaultdict(lambda: defaultdict(int))
        self.tcp_states_by_port = defaultdict(lambda: defaultdict(int))
        self.udp_states_by_port = defaultdict(lambda: defaultdict(int))
        self.processed_connection_data = {}
        self.collected_client_ips_for_subnetting = []
        self.discovered_service_ports = {'tcp': set(), 'udp': set()}
        self.port_service_overrides = {}
        self.server_ips = self._detect_server_ips()
        self._validate_command_line_combinations()
        self.process_port_selection_from_args()

@staticmethod
    def _detect_server_ips():
        """Collect addresses assigned to this host (loopback + interfaces)."""
        server_ips = {'127.0.0.1', '::1'}
        ip_path = shutil.which('ip')
        if ip_path:
            try:
                result = subprocess.run(
                    [ip_path, '-o', 'addr'],
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True,
                    check=False,
                )
                if result.returncode == 0:
                    for line in result.stdout.splitlines():
                        match = re.search(r'\s(inet6?)\s([^\s/]+)', line)
                        if not match:
                            continue
                        address = match.group(2).split('%')[0]
                        try:
                            server_ips.add(str(ip_address(address)))
                        except ValueError:
                            continue
            except OSError:
                pass
        else:
            hostname_path = shutil.which('hostname')
            if hostname_path:
                try:
                    result = subprocess.run(
                        [hostname_path, '-I'],
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE,
                        text=True,
                        check=False,
                    )
                    if result.returncode == 0:
                        for address in result.stdout.split():
                            try:
                                server_ips.add(str(ip_address(address)))
                            except ValueError:
                                continue
                except OSError:
                    pass
        return server_ips

def _is_web_port(self, port_number, protocol):
        if protocol != 'tcp':
            return False
        if port_number in WEB_EXTRA_PORTS:
            return True
        return self._get_service_name(port_number, protocol) == 'http'

def _is_internal_client(self, ip_string):
        try:
            ip_object = ip_address(ip_string)
        except ValueError:
            return False
        if (
            ip_object.is_loopback
            or ip_object.is_private
            or ip_object.is_link_local
            or ip_object.is_unspecified
        ):
            return True
        return ip_string in self.server_ips

def _internal_client_reason(self, ip_string):
        try:
            ip_object = ip_address(ip_string)
        except ValueError:
            return 'internal'
        if ip_object.is_loopback:
            return 'loopback'
        if ip_string in self.server_ips:
            return 'self'
        if ip_object.is_private or ip_object.is_link_local:
            return 'private'
        return 'internal'

def _format_client_ip(self, ip_string, port_number, protocol):
        if (
            self._is_web_port(port_number, protocol)
            and self._is_internal_client(ip_string)
        ):
            return f"{ip_string} ({self._internal_client_reason(ip_string)})"
        return ip_string

def _external_ip_counts(self, ip_counts, port_number, protocol):
        if not self._is_web_port(port_number, protocol):
            return ip_counts
        return {
            ip_string: count
            for ip_string, count in ip_counts.items()
            if not self._is_internal_client(ip_string)
        }

def _web_internal_traffic_totals(self):
        internal_total = 0
        external_total = 0
        for data_key, connection_data in self.processed_connection_data.items():
            protocol, port_string = data_key.split(':')
            port_number = int(port_string)
            if not self._is_web_port(port_number, protocol):
                continue
            for ip_string, count in connection_data['ips'].items():
                if self._is_internal_client(ip_string):
                    internal_total += count
                else:
                    external_total += count
        return internal_total, external_total

def _print_web_internal_note(self):
        internal_total, external_total = self._web_internal_traffic_totals()
        web_total = internal_total + external_total
        if web_total == 0 or internal_total == 0:
            return
        internal_share = internal_total / web_total
        if internal_share < INTERNAL_WEB_DOMINANCE_THRESHOLD:
            return
        print(self._warn_color()('WEB STACK NOTE'.center(100)))
        print(
            self._warn_color()(
                f"  {internal_total:,} of {web_total:,} HTTP connections "
                f"({internal_share:.0f}%) are from loopback, private, or "
                "this server's own IPs — typical Apache/Nginx/cPanel "
                "proxy/health-check traffic, not external clients."
            )
        )
        print()

def _validate_command_line_combinations(self):
        if self.command_line_args.watch and self.command_line_args.subnet:
            print(
                "Error: --subnet cannot be used with --watch",
                file=sys.stderr,
            )
            sys.exit(1)
        if self.command_line_args.watch is not None:
            if self.command_line_args.watch < 1:
                print("Error: watch interval must be >= 1 second", file=sys.stderr)
                sys.exit(1)

def _reset_collection_state(self):
        self.tcp_connections_by_port = defaultdict(lambda: defaultdict(int))
        self.udp_connections_by_port = defaultdict(lambda: defaultdict(int))
        self.tcp_states_by_port = defaultdict(lambda: defaultdict(int))
        self.udp_states_by_port = defaultdict(lambda: defaultdict(int))
        self.processed_connection_data = {}
        self.collected_client_ips_for_subnetting = []
        self.discovered_service_ports = {'tcp': set(), 'udp': set()}
        self.port_service_overrides = {}

# Parses command-line arguments.
    def parse_command_line_arguments(self):
        parser = argparse.ArgumentParser(
            description='Summarize TCP/UDP connection info.'
        )
        for service_name in self.service_registry.services['tcp']:
            parser.add_argument(
                f'--{service_name}',
                action='store_true',
                help=f'Show only {service_name.upper()} connections',
            )
        parser.add_argument(
            '--tcp', action='store_true', help='Show all TCP connections'
        )
        parser.add_argument(
            '--udp', action='store_true', help='Show all UDP connections'
        )
        parser.add_argument(
            '--top',
            action='store_true',
            help='Only show services ≥2000 connections or IPs >200',
        )
        parser.add_argument(
            '-s',
            '--subnet',
            action='store_true',
            help='Group connections by /24 and /16 subnets',
        )
        parser.add_argument(
            '-p',
            '--port',
            nargs='+',
            help='Specify up to 10 ports (space-separated)',
        )
        parser.add_argument(
            'ports', nargs='*', help='Deprecated: use -p or --port instead'
        )
        parser.add_argument(
            '-w',
            '--watch',
            nargs='?',
            const=DEFAULT_WATCH_INTERVAL,
            type=int,
            metavar='SECS',
            help=(
                f'Refresh every SECS seconds (default: {DEFAULT_WATCH_INTERVAL}); '
                'Ctrl+C to exit'
            ),
        )
        parser.add_argument(
            '--all-ports',
            action='store_true',
            help='Include every local TCP/UDP port with active connections',
        )
        return parser.parse_args()

# Processes selected port arguments.
    def process_port_selection_from_args(self):
        any_port_was_selected = False

if self.command_line_args.tcp:
            tcp_ports = self.service_registry.get_all_ports('tcp')
            self.selected_ports_by_protocol['tcp'].update(tcp_ports)
            self.requested_tcp_ports.update(tcp_ports)
            any_port_was_selected = True
        else:
            for service_name in self.service_registry.services['tcp']:
                if getattr(self.command_line_args, service_name, False):
                    ports = self.service_registry.services['tcp'][service_name]
                    self.selected_ports_by_protocol['tcp'].update(ports)
                    self.requested_tcp_ports.update(ports)
                    any_port_was_selected = True

if self.command_line_args.udp:
            udp_ports = self.service_registry.get_all_ports('udp')
            self.selected_ports_by_protocol['udp'].update(udp_ports)
            self.requested_udp_ports.update(udp_ports)
            any_port_was_selected = True

explicit_ports = list(self.command_line_args.port or [])
        explicit_ports.extend(self.command_line_args.ports)
        if explicit_ports:
            if len(explicit_ports) > 10:
                print("Error: max 10 ports", file=sys.stderr)
                sys.exit(1)
            for port_string in explicit_ports:
                if port_string.isdigit():
                    port_number = int(port_string)
                    self.selected_ports_by_protocol['tcp'].add(port_number)
                    self.requested_tcp_ports.add(port_number)
                    if (
                        self.command_line_args.udp
                        or self.command_line_args.top
                        or self.service_registry.is_tracked_port(
                            port_number, 'udp'
                        )
                    ):
                        self.selected_ports_by_protocol['udp'].add(
                            port_number
                        )
                        self.requested_udp_ports.add(port_number)
                    any_port_was_selected = True

if not any_port_was_selected:
            tcp_ports = self.service_registry.get_all_ports('tcp')
            self.selected_ports_by_protocol['tcp'].update(tcp_ports)
            self.requested_tcp_ports.update(tcp_ports)

if self.command_line_args.top:
            udp_ports = self.service_registry.get_all_ports('udp')
            self.selected_ports_by_protocol['udp'].update(udp_ports)
            self.requested_udp_ports.update(udp_ports)
            if (
                self.command_line_args.udp
                and not self._has_explicit_tcp_selection()
            ):
                tcp_ports = self.service_registry.get_all_ports('tcp')
                self.selected_ports_by_protocol['tcp'].update(tcp_ports)
                self.requested_tcp_ports.update(tcp_ports)

self.use_explicit_port_filter = any_port_was_selected

def _has_explicit_tcp_selection(self):
        if self.command_line_args.tcp:
            return True
        for service_name in self.service_registry.services['tcp']:
            if getattr(self.command_line_args, service_name, False):
                return True
        if self.command_line_args.port or self.command_line_args.ports:
            return True
        return False

def _wants_tcp(self):
        return bool(self.requested_tcp_ports)

def _wants_udp(self):
        return bool(self.requested_udp_ports)

def _explicitly_selected_service_names(self):
        selected = set()
        for service_name in self.service_registry.services['tcp']:
            if getattr(self.command_line_args, service_name, False):
                selected.add(service_name)
        if self.command_line_args.tcp:
            selected.update(self.service_registry.services['tcp'])
        if self.command_line_args.udp:
            selected.update(self.service_registry.services['udp'])
        return selected

def _merge_discovered_ports_for_selected_services(self):
        if self.command_line_args.all_ports:
            return
        selected_services = self._explicitly_selected_service_names()
        if not selected_services:
            return
        for protocol, ports in self.discovered_service_ports.items():
            for port_number in ports:
                if self._get_service_name(port_number, protocol) in selected_services:
                    self.selected_ports_by_protocol[protocol].add(port_number)

def _get_service_name(self, port_number, protocol):
        override = self.port_service_overrides.get((protocol, port_number))
        if override:
            return override
        return self.service_registry.get_service_name(port_number, protocol)

def _is_connection_allowed(self, port_number, protocol):
        if self.command_line_args.all_ports:
            return True
        if self.use_explicit_port_filter:
            if port_number in self.selected_ports_by_protocol.get(protocol, set()):
                return True
            if port_number in self.discovered_service_ports.get(protocol, set()):
                return (
                    self._get_service_name(port_number, protocol)
                    in self._explicitly_selected_service_names()
                )
            return False
        if self.service_registry.is_tracked_port(port_number, protocol):
            return True
        if port_number in self.discovered_service_ports.get(protocol, set()):
            return True
        return False

@staticmethod
    def _service_from_process(process_field):
        process_lower = process_field.lower()
        for needle, service_name in PROCESS_SERVICE_HINTS:
            if needle in process_lower:
                return service_name
        return None

@staticmethod
    def _admin_ssh_sessions():
        sessions = []
        seen = set()
        for variable in ('SSH_CLIENT', 'SSH_CONNECTION'):
            value = os.environ.get(variable, '')
            if not value:
                continue
            parts = value.split()
            if not parts:
                continue
            client_ip = ConnectionReporter._normalize_ip(parts[0])
            server_port = ''
            if variable == 'SSH_CLIENT' and len(parts) >= 3:
                server_port = parts[2]
            elif variable == 'SSH_CONNECTION' and len(parts) >= 4:
                server_port = parts[3]
            session_key = (client_ip, server_port)
            if session_key in seen:
                continue
            seen.add(session_key)
            sessions.append({
                'ip': client_ip,
                'server_port': server_port,
            })
        return sessions

def _register_discovered_port(self, port_number, protocol, service_name=None):
        self.discovered_service_ports[protocol].add(port_number)
        if service_name:
            self.port_service_overrides[(protocol, port_number)] = service_name

def _apply_admin_session_ports(self):
        for session in self._admin_ssh_sessions():
            if not session['server_port'].isdigit():
                continue
            port_number = int(session['server_port'])
            self._register_discovered_port(port_number, 'tcp', 'ssh')

def _parse_listen_line(self, line, protocol):
        line_parts = re.split(r'\s+', line.strip())
        if line_parts[0] not in {'LISTEN', 'UNCONN'}:
            return
        try:
            local_address = re.sub(r'^::ffff:', '', line_parts[3])
            _, port_string = local_address.rsplit(':', 1)
            port_number = int(port_string)
        except (ValueError, IndexError):
            return

service_name = None
        if self.service_registry.is_tracked_port(port_number, protocol):
            tracked = self.service_registry.get_service_name(
                port_number, protocol
            )
            if tracked != 'other':
                service_name = tracked
        if not service_name:
            process_field = ''
            if 'users:(' in line:
                process_field = line[line.index('users:('):]
            service_name = self._service_from_process(process_field)
        self._register_discovered_port(port_number, protocol, service_name)

def collect_listening_ports(self):
        self.discovered_service_ports = {'tcp': set(), 'udp': set()}
        self.port_service_overrides = {}
        self._apply_admin_session_ports()
        for flags, protocol in (('-tlnp', 'tcp'), ('-ulnp', 'udp')):
            try:
                subprocess_output = subprocess.run(
                    [ss_path, flags],
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True,
                    check=True,
                )
            except subprocess.CalledProcessError:
                fallback_flags = flags.replace('p', '')
                try:
                    subprocess_output = subprocess.run(
                        [ss_path, fallback_flags],
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE,
                        text=True,
                        check=True,
                    )
                except subprocess.CalledProcessError:
                    continue
            for line in subprocess_output.stdout.splitlines():
                self._parse_listen_line(line, protocol)
        self._merge_discovered_ports_for_selected_services()

@staticmethod
    def _normalize_ip(ip_string):
        if ip_string.startswith('[') and ip_string.endswith(']'):
            return ip_string[1:-1]
        return ip_string

@staticmethod
    def _subnet_label(ip_string, prefix):
        try:
            ip_object = ip_address(ip_string)
            if (
                ip_object.is_loopback
                or ip_object.is_private
                or ip_object.is_unspecified
            ):
                return 'local'
            return str(ip_network(f"{ip_object}/{prefix}", strict=False))
        except ValueError:
            return '?'

@staticmethod
    def _warn_color():
        return getattr(color, 'yellow', color.magenta)

def _aggregate_ip_counts(self):
        ip_counts = defaultdict(int)
        for data_key, connection_data in self.processed_connection_data.items():
            protocol, port_string = data_key.split(':')
            port_number = int(port_string)
            filtered_ips = self._external_ip_counts(
                connection_data['ips'], port_number, protocol
            )
            for ip_string, count in filtered_ips.items():
                ip_counts[ip_string] += count
        return ip_counts

def _subnet_counts(self, ip_counts, prefix=24):
        subnet_counts = defaultdict(int)
        for ip_string, count in ip_counts.items():
            subnet = self._subnet_label(ip_string, prefix)
            if subnet not in {'local', '?'}:
                subnet_counts[subnet] += count
        return subnet_counts

def _top_subnets(self, ip_counts, prefix=24, limit=10):
        return sorted(
            self._subnet_counts(ip_counts, prefix).items(),
            key=lambda item: -item[1],
        )[:limit]

def _analyze_threats(self):
        alerts = []
        for data_key, connection_data in self.processed_connection_data.items():
            protocol, port_string = data_key.split(':')
            port_number = int(port_string)
            service_name = self._get_service_name(
                port_number, protocol
            )
            ip_counts = self._external_ip_counts(
                connection_data['ips'], port_number, protocol
            )
            if not ip_counts:
                continue

total = sum(ip_counts.values())
            unique_ips = len(ip_counts)
            top_ip, top_count = max(ip_counts.items(), key=lambda item: item[1])
            label = f"{service_name.upper()}:{port_number} ({protocol})"
            top_share = (top_count / total) * 100 if total else 0

if top_count >= HOT_IP_THRESHOLD:
                alerts.append({
                    'severity': 'high',
                    'message': (
                        f"{label}: hot source {top_ip} has {top_count} "
                        f"connections ({top_share:.0f}% of service traffic)"
                    ),
                })

if total >= HOT_SERVICE_THRESHOLD:
                alerts.append({
                    'severity': 'high',
                    'message': (
                        f"{label}: volume flood — {total} connections from "
                        f"{unique_ips} unique IPs"
                    ),
                })

if service_name in auth_attack_services():
                avg_per_ip = total / unique_ips if unique_ips else 0
                if (
                    unique_ips >= BRUTE_FORCE_MIN_UNIQUE_IPS
                    and avg_per_ip <= BRUTE_FORCE_MAX_AVG_PER_IP
                ):
                    alerts.append({
                        'severity': 'medium',
                        'message': (
                            f"{label}: possible auth brute-force — "
                            f"{unique_ips} unique IPs averaging "
                            f"{avg_per_ip:.1f} conns/IP"
                        ),
                    })

top_subnet, subnet_count = max(
                self._subnet_counts(ip_counts, 24).items(),
                key=lambda item: item[1],
                default=('', 0),
            )
            if (
                subnet_count >= SUBNET_FLOOD_THRESHOLD
                and top_subnet not in {'local', '?', ''}
            ):
                alerts.append({
                    'severity': 'medium',
                    'message': (
                        f"{label}: /24 cluster {top_subnet} accounts for "
                        f"{subnet_count} connections"
                    ),
                })

if unique_ips >= 100 and top_share < 20 and total >= 500:
                alerts.append({
                    'severity': 'medium',
                    'message': (
                        f"{label}: distributed pattern — {unique_ips} sources, "
                        f"no single IP dominates (top {top_share:.0f}%)"
                    ),
                })

return alerts

def _aggregate_states(self, protocol):
        state_counts = defaultdict(int)
        states_by_port = (
            self.tcp_states_by_port
            if protocol == 'tcp'
            else self.udp_states_by_port
        )
        for port_states in states_by_port.values():
            for state_name, count in port_states.items():
                state_counts[state_name] += count
        return state_counts

def _format_state_summary(self, state_counts, limit=4):
        if not state_counts:
            return ''
        ordered_states = sorted(
            state_counts.items(), key=lambda item: (-item[1], item[0])
        )
        return ', '.join(
            f"{state_name}:{count:,}"
            for state_name, count in ordered_states[:limit]
        )

def print_connection_state_breakdown(self):
        sections = []
        if self._wants_tcp():
            tcp_states = self._aggregate_states('tcp')
            if tcp_states:
                sections.append(('TCP connection states', tcp_states))
        if self._wants_udp():
            udp_states = self._aggregate_states('udp')
            if udp_states:
                sections.append(('UDP socket states', udp_states))

if not sections:
            return

print(color.magenta('CONNECTION STATE BREAKDOWN'.center(100)))
        for title, state_counts in sections:
            total = sum(state_counts.values())
            table = PrettyTable(['State', 'Count', '%'])
            for state_name, count in sorted(
                state_counts.items(), key=lambda item: -item[1]
            ):
                share = (count / total) * 100 if total else 0
                table.add_row([state_name, count, f'{share:.1f}%'])
            table.align['State'] = 'l'
            table.align['Count'] = 'r'
            table.align['%'] = 'r'
            print(color.cyan(title.center(100)))
            for line in table.get_string().splitlines():
                if 'SYN-RECV' in line or 'SYNRECV' in line:
                    count_match = re.search(r'\|\s+(\d+)\s+\|', line)
                    if count_match and int(count_match.group(1)) >= 50:
                        print(color.red(line))
                    else:
                        print(self._warn_color()(line))
                else:
                    print(color.cyan(line))
            suspicious = sum(
                count
                for state_name, count in state_counts.items()
                if state_name in SUSPICIOUS_TCP_STATES
            )
            if suspicious >= 50:
                print(
                    color.red(
                        f"  ! High SYN-RECV count ({suspicious:,}) — "
                        "possible SYN flood"
                    )
                )
        print()

def print_executive_summary(self):
        tcp_total = sum(
            data['total']
            for key, data in self.processed_connection_data.items()
            if key.startswith('tcp:')
        )
        udp_total = sum(
            data['total']
            for key, data in self.processed_connection_data.items()
            if key.startswith('udp:')
        )
        ip_counts = self._aggregate_ip_counts()
        unique_ips = len(ip_counts)
        total_connections = tcp_total + udp_total

print(color.cyan('=' * 100))
        print(
            color.cyan(
                f"CONNECTION SNAPSHOT  {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
                .center(100)
            )
        )
        print(
            color.cyan(
                (
                    f"Total: {total_connections:,} connections  |  "
                    f"Unique IPs: {unique_ips:,}  |  "
                    f"TCP: {tcp_total:,}  |  UDP: {udp_total:,}"
                ).center(100)
            )
        )
        print(color.cyan('=' * 100))
        print()

admin_sessions = self._admin_ssh_sessions()
        if admin_sessions:
            print(color.cyan('YOUR SSH SESSION(S)'.center(100)))
            for session in admin_sessions:
                port_label = (
                    f"port {session['server_port']}"
                    if session['server_port']
                    else 'unknown port'
                )
                line = f"  {session['ip']} connected to {port_label}"
                if session['ip'] in ip_counts:
                    print(color.cyan(f"{line} (visible below)"))
                else:
                    print(
                        color.cyan(f"{line} (")
                        + self._warn_color()("checking discovered ports")
                        + color.cyan(")")
                    )
            print()

extra_ports = []
        for protocol in ('tcp', 'udp'):
            for port_number in sorted(self.discovered_service_ports[protocol]):
                if self.service_registry.is_tracked_port(port_number, protocol):
                    continue
                service_name = self._get_service_name(port_number, protocol)
                extra_ports.append(f"{service_name}:{port_number}/{protocol}")
        if extra_ports:
            print(self._warn_color()('DISCOVERED LISTEN PORTS'.center(100)))
            print(self._warn_color()(f"  {', '.join(extra_ports[:12])}"))
            if len(extra_ports) > 12:
                print(self._warn_color()(f"  ... +{len(extra_ports) - 12} more"))
            print()

self._print_web_internal_note()

alerts = self._analyze_threats()
        if alerts:
            print(self._warn_color()('ALERTS'.center(100)))
            for alert in sorted(
                alerts, key=lambda item: item['severity'], reverse=True
            ):
                line = f"  ! {alert['message']}"
                if alert['severity'] == 'high':
                    print(color.red(line))
                else:
                    print(self._warn_color()(line))
            print()

self.print_connection_state_breakdown()

top_subnets = self._top_subnets(ip_counts, prefix=24, limit=5)
        if top_subnets:
            print(color.magenta('TOP /24 SUBNETS (all services)'.center(100)))
            subnet_table = PrettyTable(['Subnet', 'Connections', '% of total'])
            for subnet, count in top_subnets:
                share = (count / total_connections) * 100 if total_connections else 0
                subnet_table.add_row([subnet, count, f'{share:.1f}%'])
            subnet_table.align['Subnet'] = 'l'
            subnet_table.align['Connections'] = 'r'
            subnet_table.align['% of total'] = 'r'
            for line in subnet_table.get_string().splitlines():
                print(color.cyan(line))
            print()

def _require_ss(self):
        if not ss_path:
            print("Error: ss command not found", file=sys.stderr)
            sys.exit(1)

# Collects TCP connection data.
    def collect_tcp_connections(self):
        self._require_ss()
        try:
            subprocess_output = subprocess.run(
                [ss_path, "-tan"],
                stdout=subprocess.PIPE,
                check=True,
                text=True,
            )
        except subprocess.CalledProcessError:
            print("Error running: ss -tan", file=sys.stderr)
            sys.exit(1)
        for line in subprocess_output.stdout.splitlines():
            self._parse_ss_output_line(
                line, 'tcp', self.tcp_connections_by_port
            )

# Collects UDP connection data.
    def collect_udp_connections(self):
        self._require_ss()
        try:
            subprocess_output = subprocess.run(
                [ss_path, "-uan"],
                stdout=subprocess.PIPE,
                check=True,
                text=True,
            )
        except subprocess.CalledProcessError:
            print("Error running: ss -uan", file=sys.stderr)
            sys.exit(1)
        for line in subprocess_output.stdout.splitlines():
            self._parse_ss_output_line(
                line, 'udp', self.udp_connections_by_port
            )

# Parses a single connection line.
    def _parse_ss_output_line(self, line, protocol, connection_storage_dict):
        line_parts = re.split(r'\s+', line.strip())
        if len(line_parts) < 5 or line_parts[0] in {'State', 'LISTEN'}:
            return
        connection_state = line_parts[0]
        try:
            server_address_string = re.sub(r'^::ffff:', '', line_parts[3])
            client_address_string = re.sub(r'^::ffff:', '', line_parts[4])
            server_ip, server_port_string = server_address_string.rsplit(':', 1)
            client_ip, client_port_string = client_address_string.rsplit(':', 1)
        except ValueError:
            return

server_ip = self._normalize_ip(server_ip)
        client_ip = self._normalize_ip(client_ip)

if server_port_string == '*' or client_port_string == '*':
            return
        if server_ip in SKIPPED_SERVER_IPS or client_ip in SKIPPED_CLIENT_IPS:
            return
        try:
            port_number = int(server_port_string)
        except ValueError:
            return

if not self._is_connection_allowed(port_number, protocol):
            return

if self.command_line_args.subnet:
            self.collected_client_ips_for_subnetting.append(client_ip)
            return

state_storage = (
            self.tcp_states_by_port
            if protocol == 'tcp'
            else self.udp_states_by_port
        )
        state_storage[port_number][connection_state] += 1
        connection_storage_dict[port_number][client_ip] += 1

# Summarizes all connection data.
    def summarize_collected_connections(self):
        if self._wants_tcp():
            for (
                port_number,
                ip_to_count_map,
            ) in self.tcp_connections_by_port.items():
                self.processed_connection_data[f"tcp:{port_number}"] = {
                    'total': sum(ip_to_count_map.values()),
                    'ips': ip_to_count_map,
                    'states': dict(self.tcp_states_by_port[port_number]),
                }
        if self._wants_udp():
            for (
                port_number,
                ip_to_count_map,
            ) in self.udp_connections_by_port.items():
                self.processed_connection_data[f"udp:{port_number}"] = {
                    'total': sum(ip_to_count_map.values()),
                    'ips': ip_to_count_map,
                    'states': dict(self.udp_states_by_port[port_number]),
                }

# Ensures requested services appear.
    def ensure_all_requested_services_in_summary(self):
        if self._wants_tcp():
            for port_number in self.requested_tcp_ports:
                data_key = f"tcp:{port_number}"
                if data_key not in self.processed_connection_data:
                    self.processed_connection_data[data_key] = {
                        'total': 0,
                        'ips': {},
                        'states': {},
                    }
        if self._wants_udp():
            for port_number in self.requested_udp_ports:
                data_key = f"udp:{port_number}"
                if data_key not in self.processed_connection_data:
                    self.processed_connection_data[data_key] = {
                        'total': 0,
                        'ips': {},
                        'states': {},
                    }

# Prints tables side-by-side.
    def _terminal_width(self, default=100):
        try:
            return shutil.get_terminal_size().columns
        except OSError:
            return default

def _tables_per_row(self, table_count, preferred=DEFAULT_MAX_TABLES_PER_ROW):
        width_based = max(1, self._terminal_width() // ESTIMATED_TABLE_WIDTH)
        return max(1, min(table_count, preferred, width_based))

@staticmethod
    def _chunk_tables(tables_with_row_data, max_per_row):
        for start in range(0, len(tables_with_row_data), max_per_row):
            yield tables_with_row_data[start:start + max_per_row]

def _print_service_title(self, service_title, service_totals):
        total_connections = service_totals['connections']
        service_is_hot = total_connections >= HOT_SERVICE_THRESHOLD

title_parts = [
            f"{service_title.upper()}",
            f"{total_connections:,} conns",
            f"{service_totals['unique_ips']:,} unique IPs",
        ]
        internal_web = service_totals.get('internal_web_connections', 0)
        if internal_web > 0:
            title_parts.append(f"{internal_web:,} internal/web-stack")
        top_subnet = service_totals.get('top_subnet', '')
        state_summary = service_totals.get('state_summary', '')
        if top_subnet:
            title_parts.append(f"top /24: {top_subnet}")
        if state_summary:
            title_parts.append(f"states: {state_summary}")
        title_string = ' | '.join(title_parts)
        if service_is_hot:
            print(color.red(title_string.center(100)))
        else:
            print(color.magenta(title_string.center(100)))

def _print_table_row_chunk(self, tables_with_row_data):
        list_of_table_line_lists = []
        list_of_high_traffic_ip_sets = []
        for table, rows, port_number, protocol in tables_with_row_data:
            list_of_table_line_lists.append(table.get_string().splitlines())
            list_of_high_traffic_ip_sets.append({
                ip
                for ip, count in rows
                if count >= HOT_IP_THRESHOLD
                and not (
                    self._is_web_port(port_number, protocol)
                    and self._is_internal_client(ip)
                )
            })

max_height = max(
            len(lines) for lines in list_of_table_line_lists if lines
        )

for lines in list_of_table_line_lists:
            if not lines:
                continue
            table_width = len(lines[0])
            while len(lines) < max_height:
                lines.append(' ' * table_width)

for i in range(max_height):
            line_parts = []
            for current_table_index, lines in enumerate(
                list_of_table_line_lists
            ):
                line = lines[i]
                is_header = i == 1
                is_border = line.strip().startswith('+') or not line.strip()

if is_border:
                    line_parts.append(color.cyan(line))
                elif is_header:
                    line_parts.append(color.magenta(line))
                else:
                    is_hot_line = any(
                        self._line_contains_ip(line, hot_ip)
                        for hot_ip in list_of_high_traffic_ip_sets[
                            current_table_index
                        ]
                    )
                    if is_hot_line:
                        line_parts.append(color.red(line))
                    else:
                        line_parts.append(line)
            print("  ".join(line_parts))

def print_service_tables_side_by_side(
        self,
        service_title,
        tables_with_row_data,
        service_totals=None,
        max_per_row=None,
    ):
        if not tables_with_row_data:
            return

if service_totals is None:
            displayed_connections = sum(
                sum(count for _, count in rows)
                for _, rows, _, _ in tables_with_row_data
            )
            service_totals = {
                'connections': displayed_connections,
                'unique_ips': len(
                    {
                        ip
                        for _, rows, _, _ in tables_with_row_data
                        for ip, _ in rows
                    }
                ),
            }

if max_per_row is None:
            max_per_row = self._tables_per_row(len(tables_with_row_data))

self._print_service_title(service_title, service_totals)
        table_chunks = list(
            self._chunk_tables(tables_with_row_data, max_per_row)
        )
        for chunk_index, chunk in enumerate(table_chunks):
            if chunk_index > 0:
                print()
            self._print_table_row_chunk(chunk)
        print()

@staticmethod
    def _line_contains_ip(line, ip):
        return f'| {ip} ' in line or f'| {ip}|' in line

def _service_totals(self, ports_data, protocol='tcp'):
        ip_counts = defaultdict(int)
        state_counts = defaultdict(int)
        total_connections = 0
        internal_web_connections = 0
        for port_number, info in ports_data.items():
            total_connections += info['total']
            for ip_string, count in info['ips'].items():
                if (
                    self._is_web_port(port_number, protocol)
                    and self._is_internal_client(ip_string)
                ):
                    internal_web_connections += count
                else:
                    ip_counts[ip_string] += count
            for state_name, count in info.get('states', {}).items():
                state_counts[state_name] += count

top_subnet = ''
        subnet_totals = self._subnet_counts(ip_counts, 24)
        if subnet_totals:
            top_subnet = max(subnet_totals.items(), key=lambda item: item[1])[0]

return {
            'connections': total_connections,
            'external_connections': total_connections - internal_web_connections,
            'internal_web_connections': internal_web_connections,
            'unique_ips': len(ip_counts),
            'top_subnet': top_subnet if top_subnet not in {'local', '?'} else '',
            'state_summary': self._format_state_summary(state_counts),
        }

def _build_service_tables(self, service_name, ports_data, protocol='tcp'):
        tables_to_print = []
        for port_number, info in ports_data.items():
            if info['total'] <= 0:
                continue
            service_total = info['total']
            rows = sorted(info['ips'].items(), key=lambda x: -x[1])[:10]
            table = PrettyTable(
                [
                    f"{service_name.upper()}:{port_number}",
                    "Count",
                    "%",
                ]
            )
            for ip_string, count in rows:
                share = (count / service_total) * 100
                table.add_row([
                    self._format_client_ip(ip_string, port_number, protocol),
                    count,
                    f'{share:.1f}%',
                ])
            table.align = "l"
            table.align["Count"] = "r"
            table.align["%"] = "r"
            tables_to_print.append((table, rows, port_number, protocol))
        return tables_to_print

def _should_print_service(self, ports_data, protocol='tcp'):
        total_connections = sum(d['total'] for d in ports_data.values())
        if total_connections <= 0:
            return False
        any_ip_over_200 = any(
            count > 200
            for port_number, data in ports_data.items()
            for ip_string, count in self._external_ip_counts(
                data['ips'], port_number, protocol
            ).items()
        )
        return not self.command_line_args.top or (
            total_connections >= 2000 or any_ip_over_200
        )

def _print_service_if_needed(
        self, service_title, ports_data, tables_to_print, max_per_row=None,
        protocol='tcp',
    ):
        if tables_to_print and self._should_print_service(
            ports_data, protocol=protocol
        ):
            self.print_service_tables_side_by_side(
                service_title,
                tables_to_print,
                service_totals=self._service_totals(
                    ports_data, protocol=protocol
                ),
                max_per_row=max_per_row,
            )

# Reports on service connections.
    def generate_connection_report_tables(self):
        service_map_by_protocol = defaultdict(
            lambda: defaultdict(lambda: {'total': 0, 'ips': {}})
        )
        for data_key, connection_data in self.processed_connection_data.items():
            protocol, port_string = data_key.split(':')
            port_number = int(port_string)
            service_name = self._get_service_name(
                port_number, protocol
            )
            service_map_by_protocol[(protocol, service_name)][port_number] = {
                'total': connection_data['total'],
                'ips': connection_data['ips'],
                'states': connection_data.get('states', {}),
            }

service_print_order_by_protocol = {
            'tcp': [
                'http',
                self.service_registry.panel_user,
                self.service_registry.panel_admin,
                'smtp',
                'imap',
                'pop3',
                'webmail',
                'ssh',
                'ftp',
                'mysql',
            ],
            'udp': ['dns', 'ntp'],
        }

for protocol in ['tcp', 'udp']:
            if protocol == 'udp' and not self._wants_udp():
                continue

service_print_order = service_print_order_by_protocol[protocol]
            i = 0
            while i < len(service_print_order):
                service_name = service_print_order[i]

# Group mail services (TCP only)
                if protocol == 'tcp' and service_name == 'imap':
                    combined_mail_services = ['imap', 'pop3', 'webmail']
                    tables_to_print = []
                    ports_data = {}
                    for sub_service_name in combined_mail_services:
                        sub_service_ports_data = service_map_by_protocol.get(
                            (protocol, sub_service_name), {}
                        )
                        ports_data.update(sub_service_ports_data)
                        tables_to_print.extend(
                            self._build_service_tables(
                                sub_service_name,
                                sub_service_ports_data,
                                protocol=protocol,
                            )
                        )

self._print_service_if_needed(
                        'MAIL SERVICES',
                        ports_data,
                        tables_to_print,
                        max_per_row=MAIL_SERVICES_MAX_PER_ROW,
                        protocol=protocol,
                    )

i += len(combined_mail_services)
                    continue

ports_data = service_map_by_protocol.get(
                    (protocol, service_name), {}
                )
                if ports_data:
                    tables_to_print = self._build_service_tables(
                        service_name, ports_data, protocol=protocol
                    )
                    self._print_service_if_needed(
                        service_name,
                        ports_data,
                        tables_to_print,
                        protocol=protocol,
                    )
                i += 1

other_ports_data = service_map_by_protocol.get(
                (protocol, 'other'), {}
            )
            if other_ports_data:
                tables_to_print = self._build_service_tables(
                    'other', other_ports_data, protocol=protocol
                )
                self._print_service_if_needed(
                    'other',
                    other_ports_data,
                    tables_to_print,
                    protocol=protocol,
                )

# Summarizes subnet connections.
    def generate_subnet_summary_tables(self):
        slash_24_subnet_counts = defaultdict(int)
        slash_16_subnet_counts = defaultdict(int)
        for ip_string in self.collected_client_ips_for_subnetting:
            try:
                ip_address_object = ip_address(ip_string)
                if (
                    ip_address_object.is_loopback
                    or ip_address_object.is_private
                    or ip_address_object.is_unspecified
                ):
                    continue
                slash_24_subnet_counts[
                    str(ip_network(f"{ip_address_object}/24", strict=False))
                ] += 1
                slash_16_subnet_counts[
                    str(ip_network(f"{ip_address_object}/16", strict=False))
                ] += 1
            except ValueError:
                continue

for table_title_label, subnet_counts_data in (
            ("Top /24 Subnets", slash_24_subnet_counts),
            ("Top /16 Subnets", slash_16_subnet_counts),
        ):
            if not subnet_counts_data:
                continue

# Filter rows based on the --top flag
            all_rows_sorted = sorted(
                subnet_counts_data.items(), key=lambda kv: -kv[1]
            )
            if self.command_line_args.top:
                rows_to_display = [
                    row for row in all_rows_sorted if row[1] > 200
                ]
            else:
                rows_to_display = all_rows_sorted[:20]

if not rows_to_display:
                continue

table = PrettyTable(["Subnet", "Count"])
            table.title = table_title_label

high_traffic_row_indices = {
                i for i, row in enumerate(rows_to_display) if row[1] > 200
            }
            for row in rows_to_display:
                table.add_row(row)

table.align["Subnet"] = "l"
            table.align["Count"] = "r"

lines = table.get_string().splitlines()
            for i, line in enumerate(lines):
                # Line 1 is title, Line 3 is header text
                if i in {1, 3}:
                    print(color.magenta(line))
                # Data rows start at index 5
                elif i >= 5 and (i - 5) in high_traffic_row_indices:
                    print(color.red(line))
                else:
                    print(color.cyan(line))
            print()

def _run_report_once(self):
        self._require_ss()
        self.collect_listening_ports()

if self.command_line_args.subnet:
            if self._wants_tcp():
                self.collect_tcp_connections()
            if self._wants_udp():
                self.collect_udp_connections()
            self.generate_subnet_summary_tables()
            return

if self._wants_udp():
            self.collect_udp_connections()

if self._wants_tcp():
            self.collect_tcp_connections()

self.summarize_collected_connections()
        self.ensure_all_requested_services_in_summary()
        self.print_executive_summary()
        self.generate_connection_report_tables()

@staticmethod
    def _clear_screen():
        if sys.stdout.isatty():
            print('\033[2J\033[H', end='')

def run_watch_mode(self):
        interval = self.command_line_args.watch
        print(
            color.cyan(
                f"Watch mode: refreshing every {interval}s (Ctrl+C to exit)"
            )
        )
        try:
            while True:
                self._clear_screen()
                self._reset_collection_state()
                self._run_report_once()
                time.sleep(interval)
        except KeyboardInterrupt:
            print(color.cyan('\nWatch mode stopped.'))

# Main execution function.
    def run(self):
        if self.command_line_args.watch is not None:
            self.run_watch_mode()
            return
        self._run_report_once()

if __name__ == "__main__":
    ConnectionReporter().run()

cmd: