# Copyright 2026 SigmaSRC, Inc. All Rights Reserved.

"""Run various commands and gather the results."""

import os
import sys
import logging
import platform

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

from sigma.util import config

_maxlength = 4000
_truncatestring = "\n(Output truncated to " + str(_maxlength) + " characters)"

logger = logging.getLogger("gather.gather_command")

# Console programs do NOT emit UTF-8: on Windows they use the OEM code page
# (CP_OEMCP, e.g. cp850/cp437/cp858) and on Unix the locale codec. Under Python
# UTF-8 Mode the default text decode is strict UTF-8, so reading such output with
# a bare os.popen(...).read() raises UnicodeDecodeError on the first non-ASCII
# byte (e.g. an accented user name in `qwinsta` on a French/German host) and
# aborts the whole 'command' gatherer. Decode with the platform's real console
# codec -- "oem" on Windows -- and replace anything that still won't decode, so a
# single odd byte can never lose all command output.
_console_encoding = "oem" if sys.platform == "win32" else "utf-8"


def popen_text(command):
    """Run a shell command and return its combined output as text, decoded with
    the console code page (errors replaced). Never raises -- a failing command
    yields ""."""
    import subprocess
    try:
        p = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE,
                             stderr=subprocess.STDOUT,
                             encoding=_console_encoding, errors="replace")
        out, _ = p.communicate()
        return out or ""
    except Exception:
        logger.exception("command failed: %r", command)
        return ""


# For each command, we have:
# *  a short name that gets incorporated into the attribute.  If the
#    short name is "ps", we the attribute "Command_ps" contains the
#    command we actually ran, and "Output_ps" contains the output from the
#    command.
# *  The command to run.
_linuxcommands = {
    # It's /bin/date on Linux, /usr/bin/date on Solaris.
    "date": "/bin/date",
    "arp": "/sbin/arp",
    "df": "/bin/df -kl",
    "mounts": "/bin/cat /proc/mounts",
    "dmesg": "/bin/dmesg",
    "finger": "/usr/bin/finger",
    # If I pass -a to ifconfig, both Solaris and Linux
    # can use this.  Linux ifconfig doesn't care.
    "ifconfig": "/sbin/ifconfig -a",
    "ipcs": "/usr/bin/ipcs",
    "last": "/usr/bin/last -50",
    "lsmod": "/sbin/lsmod",
    # Active connections
    "netstata": "/bin/netstat -anp",
    "netstati": "/bin/netstat -i",
    "netstatr": "/bin/netstat -rn",
    "nfsstat": "/usr/sbin/nfsstat",
    "ps": "/bin/ps -ef",
    "rpcinfo": "/usr/sbin/rpcinfo -p",
    "showmounta": "/usr/sbin/showmount -a",
    "showmounte": "/usr/sbin/showmount -e",
    "top": "/usr/bin/top -b -n 1",
    "uname": "/bin/uname -a",
    "uptime": "/usr/bin/uptime",
    "w": "/usr/bin/w"
     }

# The Solaris commands are the same as the linux commands, with
# a few exceptions.
_solariscommands = _linuxcommands.copy()
_solariscommands["arp"] = "/usr/sbin/arp -a"
_solariscommands["date"] = "/usr/bin/date"
_solariscommands["mounts"] = "/bin/cat /etc/mnttab"
del _solariscommands["lsmod"]
_solariscommands["netstata"] = "/usr/bin/netstat -an"
_solariscommands["netstati"] = "/usr/bin/netstat -i"
_solariscommands["netstatr"] = "/usr/bin/netstat -rn"
_solariscommands["nfsstat"] = "/usr/bin/nfsstat"
_solariscommands["rpcinfo"] = "/usr/bin/rpcinfo -p"
del _solariscommands["top"]
for dn in ["/usr/local/bin", "/opt/sfw/bin"]:
    fn = os.path.join(dn, "top")
    if os.path.exists(fn):
        _solariscommands["top"] = fn + " -b -n 1"
        break
_solariscommands["showrev"] = "/usr/bin/showrev -p | nawk '{print $2}'"

# The HP-UX commands are the same as the Solaris commands, mostly.
_hpuxcommands = _solariscommands.copy()
_hpuxcommands["dmesg"] = "/usr/sbin/dmesg"
del _hpuxcommands["ifconfig"]
del _hpuxcommands["showrev"]

# The OS X commands derive from Linux.
_osxcommands = _linuxcommands.copy()
_osxcommands["arp"] = "/usr/sbin/arp -a"
_osxcommands["showmounte"] = "/usr/bin/showmount -e"
_osxcommands["showmounta"] = "/usr/bin/showmount -a"
del _osxcommands["showmounte"], _osxcommands["showmounta"] # XXX hangs?
del _osxcommands["lsmod"]
del _osxcommands["ipcs"]
_osxcommands["dmesg"] = "/sbin/dmesg"
_osxcommands["netstata"] = "/usr/sbin/netstat -an"
_osxcommands["netstati"] = "/usr/sbin/netstat -in"
_osxcommands["netstatr"] = "/usr/sbin/netstat -rn"
_osxcommands["uname"] = "/usr/bin/uname -a"
_osxcommands["ps"] = "/bin/ps -ax"
_osxcommands["top"] = "/usr/bin/top -l1"
_osxcommands["mounts"] = "/sbin/mount"
_osxcommands["nfsstat"] = "/usr/bin/nfsstat"

# The AIX commands derive from Linux.
_aixcommands = _linuxcommands.copy()
_aixcommands["arp"] = "/usr/sbin/arp -a | grep -v bucket:"
_aixcommands["df"] = "/usr/bin/df -k" # no -l option!
del _aixcommands["dmesg"]
_aixcommands["ifconfig"] = "/etc/ifconfig -a"
del _aixcommands["lsmod"]
_aixcommands["mounts"] = "/usr/sbin/mount"
_aixcommands["netstata"] = "/usr/sbin/netstat -an"
_aixcommands["rpcinfo"] = "/usr/bin/rpcinfo -p"
_aixcommands["showmounta"] = "/usr/bin/showmount -a"
_aixcommands["showmounte"] = "/usr/bin/showmount -e"
del _aixcommands["top"]

def do_winlogin(table):
    from sigma.gather import winlogin
    from io import StringIO
    save_stdout = sys.stdout
    io = StringIO()
    try:
        sys.stdout = io
        winlogin.main()
        text = io.getvalue()
        if len(text) > _maxlength:
            text = text[:_maxlength - len (_truncatestring)] \
                   + _truncatestring
        table.add(["winlogin", "winlogin.py", text])
    finally:
        sys.stdout = save_stdout

def _expand_file_name(filename):
    exfilename = filename
    i = filename.find("%")
    if i >= 0:
         j = filename.find("%", i+1)
         if j > i:
             key = filename[i+1: j]
             env = os.getenv(key)
             if env is not None:
                 exfilename = filename[:i] + env + filename[j+1:]
    return exfilename

def do_wintasklist(table):
    from win32com.client import GetObject
    from io import StringIO
    save_stdout = sys.stdout
    io = StringIO()
    try:
        output_cmd = ""
        sys.stdout = io
        exfilename = _expand_file_name("%SystemRoot%\\system32\\qprocess.exe")
        if os.path.exists(exfilename):
            output_cmd = "qprocess"
            print(popen_text("%SystemRoot%\\system32\\qprocess * /system"))
        else:
            output_cmd = "Win32_Process:ProcessId, Caption"
            out_format1 = "PID          Image Name"
            out_format2 = "========================"
            out_format3 = "%5d          %s"  
            print(out_format1)
            print(out_format2)
            try:
                objItems = GetObject("WinMgMts://./root/cimv2").InstancesOf("Win32_Process")
                for item in objItems:
                    astr = out_format3 % (item.Properties_("ProcessId").Value, item.Properties_("Caption").Value)
                    print(astr)
            except:
                logger.exception("Win32_Process")
        text = io.getvalue()
        if len(text) > _maxlength:
            text = text[:_maxlength - len (_truncatestring)] \
                   + _truncatestring
        table.add(["tasklist", output_cmd, text])
    finally:
        sys.stdout = save_stdout

def do_winuser(table):
    exfilename = _expand_file_name("%SystemRoot%\\system32\\qwinsta.exe")
    if os.path.exists(exfilename):
        text = ""
        try:
            text = popen_text("%SystemRoot%\\system32\\qwinsta")
            if len(text) > _maxlength:
                text = text[:_maxlength - len (_truncatestring)] \
                       + _truncatestring
        except:
            logger.exception("qwinsta")
        table.add(["qwinsta", "qwinsta", text])
    else:
        from win32com.client import GetObject
        username = ""
        try:
            objItems = GetObject("WinMgMts://./root/cimv2").InstancesOf("Win32_ComputerSystem")
            for item in objItems:
                username = username + item.Properties_("UserName").Value + "\n"
        except:
            logger.exception("Win32_ComputerSystem")
        table.add(["qwinsta", "Win32_ComputerSystem.UserName", username])

def get_datetime(table):
    import time
    text = time.ctime()
    table.add(["date", "ctime()", text])

# Windows XP commands...
_xpcommands = {
    # Similar to unix commands
    "date": get_datetime,
    "arp": "%SystemRoot%\\system32\\arp -a",
    "netstata": "%SystemRoot%\\system32\\netstat -ano",
    "netstatr": "%SystemRoot%\\system32\\netstat -rn",
    # Windows specific commands
    "ipconfig": "%SystemRoot%\\system32\\ipconfig /all",
    "tasklist": "%SystemRoot%\\system32\\tasklist",
    "qwinsta": do_winuser, #"%SystemRoot%\\system32\\qwinsta",
    "net_share": "%SystemRoot%\\system32\\net share <nul:",
    "net_accounts": "%SystemRoot%\\system32\\net accounts <nul:",
    "net_statistics_server": "%SystemRoot%\\system32\\net statistics server <nul:",
    "net_statistics_workstation": "%SystemRoot%\\system32\\net statistics workstation <nul:",
    #"net_use": "net use",
    #"gpresult": "gpresult",
    "systeminfo": "%SystemRoot%\\system32\\systeminfo",
    "winlogin": do_winlogin,
    }

# Windows 2000 commands...
_w2kcommands = {
    # Similar to unix commands
    "date": get_datetime,
    "arp": "%SystemRoot%\\system32\\arp -a",
    "netstata": "%SystemRoot%\\system32\\netstat -an",
    "netstatr": "%SystemRoot%\\system32\\netstat -rn",
    # Windows specific commands
    "ipconfig": "%SystemRoot%\\system32\\ipconfig /all",
    "tasklist": do_wintasklist, #"%SystemRoot%\\system32\\qprocess * /system",
    "qwinsta": do_winuser, #"%SystemRoot%\\system32\\qwinsta",
    "net_share": "%SystemRoot%\\system32\\net share <nul:",
    "net_accounts": "%SystemRoot%\\system32\\net accounts <nul:",
    "net_statistics_server": "%SystemRoot%\\system32\\net statistics server <nul:",
    "net_statistics_workstation": "%SystemRoot%\\system32\\net statistics workstation <nul:",
    #"net_use": "net use",
    #"gpresult": "gpresult",
    #"systeminfo": "systeminfo",
    "winlogin": do_winlogin,
    }

basicos = platform.uname()[0]
version = platform.uname()[2]
if basicos == "SunOS":
    _commands = _solariscommands
elif basicos == "Linux":
    _commands = _linuxcommands
elif basicos == "HP-UX":
    _commands = _hpuxcommands
elif basicos == "Darwin":
    _commands = _osxcommands
elif basicos == "AIX":
    _commands = _aixcommands
elif basicos in ("Windows", "Microsoft Windows", "Microsoft"):
    if version == '2000':
        _commands = _w2kcommands
    else:
        _commands = _xpcommands
else:
    raise Exception("Can't figure out operating system: %r" % basicos)

def Gather(collector):
    t = collector.table("command", ["shortname", "command", "output"])
    for shortname in _commands.keys():
        command = _commands[shortname]
        if hasattr(command, "__name__"):
            commandname = command.__name__
        else:
            commandname = str(command)
        logger.info("command = %r", commandname)
        if not isinstance(command, str):
            command(t)
            continue
        text = popen_text(command)
        if len(text) > _maxlength:
            text = text[:_maxlength - len (_truncatestring)] \
                   + _truncatestring
        t.add([shortname, command, text])
                    

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

if __name__ == "__main__":
    _test()
