# Copyright 2026 SigmaSRC, Inc. All Rights Reserved.

"""Use netstat to gather the open connections"""

if __name__ == "__main__":
    import os
    import sys
    try:
        import sigma
    except ImportError:
        sys.path.append(os.path.join(os.pardir, os.pardir))
    del sys
    del os

import string
import os
import re
import platform

from sigma.util import console

# _ipre matches ip addresses.
_ipre = r"\d+\.\d+\.\d+\.\d+"
assert re.match(_ipre, "192.168.1.6"), "_ipre is broken"

# _optipre matches an IP address or "*".
_optipre = r"(" + _ipre + r"|\*)"
assert re.match(_optipre, "192.168.1.6"), "_optipre (normal case) is broken."
assert re.match(_optipre, "*"), "_optipre (optional case) is broken."

# Match a port number or "*".
_optport = r"(\d+|\*)"
assert re.match(_optport, "123"), "_optport (normal case) is broken."
assert re.match(_optport, "*"), "_optport (optional case) is broken."

linux_linere = re.compile(
    r"(?P<proto>(tcp|udp))\s*\d+\s*\d+\s*"
    + r"(?P<localip>" + _ipre + r"):(?P<localport>\d+)\s+"
    + r"(?P<foreignip>" + _ipre + r"):(?P<foreignport>"+_optport+r")\s+"
    + r"(?P<state>[^\s]*)\s*")
assert linux_linere.match(
    "udp        0      0 0.0.0.0:111             "
    "0.0.0.0:*               LISTEN      "), "linux_linere is broken"

columns = ["proto", "localip", "localport",
           "foreignip", "foreignport", "state"]

# Break this out for unit testing.
def linux_parse(table, file):
    for line in file:
        match = linux_linere.match(line)
        if match:
            table.add(match.group(*columns))

def linux_netstat(table):
    file = os.popen("/bin/netstat --inet -na", "r")
    try:
        linux_parse(table, file)
    finally:
        file.close()

solaris_tcp_linere = re.compile(
    r"(?P<localip>" + _optipre + r")\.(?P<localport>\d+)\s+"
    + r"(?P<foreignip>" + _optipre + r")\.(?P<foreignport>" + _optport + r")"
    + r"\s+\d+\s+\d+\s+\d+\s+\d+\s+(?P<state>[^\s]*)")
assert solaris_tcp_linere.search(\
    "      *.32794              *.*                0      "
    "0 49152      0 LISTEN"), "solaris_tcp_linere is broken 1"
assert solaris_tcp_linere.search(\
    "10.0.5.201.22        10.0.5.39.34541      32480      "
    "0 47888      0 ESTABLISHED"), "solaris_tcp_linere is broken 2"

solaris_udp_linere = re.compile(
    r"(?P<localip>" + _optipre + r")\.(?P<localport>\d+)\s"
    # Bind the remaining ones to empty strings.  If all the regexps
    # bind the same variables, I can share code better.
    + r"(?P<foreignip>)(?P<foreignport>)(?P<state>)")
assert solaris_udp_linere.search(\
    "          *.111                                 "
    "Idle"), "solars_udp_linere is broken"


# Linux and Solaris have different notations for a socket that is
# bound to any IP.  Linux uses 0.0.0.0 and Solaris uses *.  This
# translates the Solaris version to the Linux version.
def zeroify_ip(ip):
    if ip == "*" or ip == "":
        return "0.0.0.0"
    else:
        return ip

def zeroify_port(port):
    if port == "" or port == "0":
        return "*"
    else:
        return port

def solaris_section(table, protocol, regexp, makefile):
    file = makefile(protocol)
    try:
        for line in file:
            match = regexp.search(line)
            if match:
                localip, localport, foreignip, foreignport, state = \
                         match.group(*columns[1:])
                table.add((protocol, zeroify_ip(localip), localport,
                           zeroify_ip(foreignip), zeroify_port(foreignport),
                           state))
    finally:
        file.close()

def solaris_file(protocol):
    return os.popen("/bin/netstat -f inet -na -P " + protocol, "r")

# Entry point for unit testing.

# The argument of makefile is the protocol (udp or tcp), and it should
# return a file object that has solaris netstat output for that
# protocol.  We need to run two netstat processes on Solaris, so it's
# a little more complicated than the Windows or Linux cases.
def solaris_parse_all(table, makefile):
    solaris_section(table, "udp", solaris_udp_linere, makefile)
    solaris_section(table, "tcp", solaris_tcp_linere, makefile)

def solaris_netstat(table):
    solaris_parse_all(table, solaris_file)

windows_linere = re.compile(
    r"\s*(?P<proto>(TCP|UDP))\s+"
    + r"(?P<localip>" + _ipre + r"):(?P<localport>\d+)\s+"
    + r"(?P<foreignip>" + _optipre + r"):(?P<foreignport>" + _optport+r")\s+"
    + r"(?P<state>[^\s]*)\s*")

assert windows_linere.match("  UDP    0.0.0.0:8082           *:*  "), \
    "windows_linere is broken for udp"

assert windows_linere.match("  TCP    10.0.5.62:139  0.0.0.0:0  LISTENING"), \
    "windows_linere is broken for tcp"

# Entry point for unit testing.
def windows_parse(table, file):
    portlist = []
    for line in file:
        match = windows_linere.match(line)
        if match:
            proto, localip, localport, foreignip, foreignport, state = \
                    match.group(*columns)
            # skip the port if listening on loopback address
            if localip == "127.0.0.1":
                continue
            if state == "LISTENING":
                state = "LISTEN"
            portlist.append((proto.lower(), zeroify_ip(localip),
                zeroify_port(localport), zeroify_ip(foreignip),
                zeroify_port(foreignport), state))
    #print portlist
    for i in range(len(portlist)):
        doadd = True
        # A TCP port is considered listening if it is in 'listen' state and optionally 
        # in 'syn_received' state
        if (portlist[i][5] == 'LISTEN'):
            for j in range (len(portlist)):
                if (portlist[j][0] == 'tcp' and portlist[j][2] == portlist[i][2] and 
                    portlist[j][5] != 'LISTEN' and portlist[j][5] != 'SYN_RECEIVED'):
                    doadd = False
                    break
        elif portlist[i][0] == 'tcp':
            doadd = False
        if doadd:
            table.add(portlist[i])

def windows_netstat(table):
    # I hope the path is right.
    file = console.popen("netstat -na", "r")
    try:
        windows_parse(table, file)
    finally:
        file.close()

def Gather(collector):
    t = collector.table("netstat", columns)
    basicos = platform.uname()[0]
    if basicos == "Linux":    
        linux_netstat(t)
    elif basicos == "SunOS":
        solaris_netstat(t)
    elif os.name == "nt":
        windows_netstat(t)

def _test():
    from sigma.gather import gathering
    gathering.test(Gather)

if __name__ == "__main__":
    _test()
