"""Read-only Linux listening sockets. Never start, stop, probe or expose apps."""
import ipaddress
import json
import os
import re
import signal
import sys
import time


def address(hex_value):
    # Linux procfs represents addresses as native-endian 32-bit words.
    packed = b"".join(int(hex_value[i:i + 8], 16).to_bytes(4, sys.byteorder)
                      for i in range(0, len(hex_value), 8))
    value = ipaddress.ip_address(packed)
    if isinstance(value, ipaddress.IPv6Address) and value.ipv4_mapped:
        value = value.ipv4_mapped
    return "all" if value.is_unspecified else "loopback" if value.is_loopback else "specific"


def parse_table(data, protocol):
    rows = []
    for line in data.splitlines()[1:]:
        fields = line.split()
        if len(fields) < 10 or fields[3] != ("0A" if protocol == "tcp" else "07"):
            continue
        local, port_hex = fields[1].rsplit(":", 1)
        port = int(port_hex, 16)
        # The MacServe tunnel itself is not a customer application.
        if not 1 <= port <= 65535 or (protocol == "udp" and port == 51820):
            continue
        if protocol == "udp" and int(fields[2].rsplit(":", 1)[1], 16) != 0:
            continue
        rows.append({"protocol": protocol, "port": port, "binding": address(local), "inode": fields[9]})
    return rows


def process_names(inodes, deadline):
    names = {}
    # comm is a short executable label; never read cmdline, environ, cwd or files.
    with os.scandir("/proc") as processes:
        for entry in processes:
            if time.monotonic() > deadline or len(names) == len(inodes):
                break
            if not entry.name.isascii() or not entry.name.isdigit():
                continue
            try:
                with open(entry.path + "/comm") as source:
                    name = source.read(64).strip()
                if not re.fullmatch(r"[A-Za-z0-9_.+ -]{1,64}", name):
                    continue
                with os.scandir(entry.path + "/fd") as descriptors:
                    for fd in descriptors:
                        if time.monotonic() > deadline:
                            return names
                        try:
                            link = os.readlink(fd.path)
                        except OSError:
                            continue
                        match = re.fullmatch(r"socket:\[(\d+)\]", link)
                        if match and match[1] in inodes:
                            names.setdefault(match[1], name)
            except OSError:
                continue
    return names


def discover():
    rows = []
    for filename, protocol in [("tcp", "tcp"), ("tcp6", "tcp"), ("udp", "udp"), ("udp6", "udp")]:
        try:
            with open("/proc/net/" + filename) as source:
                data = source.read(2 * 1024 * 1024 + 1)
            if len(data) > 2 * 1024 * 1024:
                raise ValueError("Socket table too large")
            rows.extend(parse_table(data, protocol))
        except FileNotFoundError:
            if filename.endswith("6"):
                continue  # IPv6 may be disabled in a user-managed guest.
            raise
    groups = {}
    rank = {"loopback": 0, "specific": 1, "all": 2}
    for row in rows:
        key = (row["protocol"], row["port"])
        group = groups.setdefault(key, {"protocol": row["protocol"], "port": row["port"],
                                        "binding": row["binding"], "inodes": set()})
        group["inodes"].add(row["inode"])
        if rank[row["binding"]] > rank[group["binding"]]:
            group["binding"] = row["binding"]
    selected = sorted(groups.values(), key=lambda row: (row["port"], row["protocol"]))[:64]
    inodes = {inode for row in selected for inode in row["inodes"]}
    names = process_names(inodes, time.monotonic() + 1.5) if inodes else {}
    listeners = []
    for row in selected:
        labels = sorted({names[inode] for inode in row.pop("inodes") if inode in names})
        row["processName"] = labels[0] if labels else None
        listeners.append(row)
    return {"available": True, "truncated": len(groups) > 64, "listeners": listeners}


def timed_out(_signum, _frame):
    raise TimeoutError("Socket discovery timed out")


if __name__ == "__main__":
    signal.signal(signal.SIGALRM, timed_out)
    signal.alarm(5)
    try:
        print(json.dumps(discover(), separators=(",", ":")))
    except Exception:
        # No raw subprocess/file errors or guest data in logs or API errors.
        print(json.dumps({"available": False, "truncated": False, "listeners": []}))
    finally:
        signal.alarm(0)
