#!/usr/bin/env python3
# Copyright 2026 SigmaSRC, Inc. All Rights Reserved.

"""Scheduler daemon.

This process handles the scheduling of other processes in the agent.

Usage: %prog [-d] [host:port]

Argument:
  host:port    host and port to serve (xmlrpc control)

"""

# Python imports
import os
import sys
import time
import socket
import shutil
import logging
import datetime
import optparse
import threading

# M2Crypto.Rand replaced by os.urandom (os already imported above)

# Enter Python UTF-8 Mode before any I/O so non-ASCII host data / Fuel source is
# handled correctly regardless of host locale (re-execs once if not already on).
# Children spawned by the scheduler inherit PYTHONUTF8 and start in UTF-8 Mode too.
if __name__ == "__main__":
    from sigma.util.utf8 import ensure_utf8_mode
    ensure_utf8_mode()

# Sigma imports
from sigma.util import avalog, config, cert
from sigma.util.errors import usage, error
from sigma.sched import tasks, scheduling, server, process, remote
from sigma.sched.pingpong import PingPongThread
from sigma.runtime import matrix
from sigma.spool import qmgr
from sigma.protocol import rest, xmlproto

# Global objects
options = config.Options()
schedule_file = options.datapath("schedule.xml")
taskdir = options.datapath("task")
logdir = options.agentlogs

logger = logging.getLogger("scheduler")

parser = optparse.OptionParser(usage=__doc__.strip())
parser.add_option("-d", "--daemon", dest="daemon", action="store_true",
                  help="run as daemon (background, closed files, etc.)")
parser.add_option("-D", "--DAEMON", dest="DAEMON", action="store_true",
                  help="run as daemon (background, closed files, etc.) "
                        "without forking")

def main(args=None):
    # Parse options
    opts, args = parser.parse_args(args)

    # Parse host argument or default
    if args:
        if len(args) > 1:
            return usage("Too many arguments")
        host = args[0]
    else:
        host = options.scheduler
    if ":" not in host:
        return usage("Host:port argument malformed")
    host, port = host.split(":")
    port = int(port)

    # Complain if options.central_server isn't set
    if not options.central_server:
        return error("The central-server option isn't set")

    # Refuse to start when the central-server HOST is blank or an install-time
    # placeholder. A silent install with no server (or a hand-edited config) can
    # leave "central-server = :17747" -- non-empty, so the check above passes,
    # but the host is empty and the agent would loop forever trying to register
    # against nothing. Exit 78 (EX_CONFIG) so the service wrapper (winservice.py)
    # and the Linux systemd unit (RestartPreventExitStatus=78) treat it as a
    # permanent config error and stop cleanly instead of restart-looping.
    cs_host = options.central_server.split(":", 1)[0].strip()
    if not cs_host or cs_host.lower() in ("qali01", "mycshost"):
        sys.stderr.write(
            "Sigma agent is not configured: no central server "
            "(central-server = %r). Set a server in conf/config and restart "
            "the agent.\n" % (options.central_server,))
        return 78

    # See if AVBIN exists
    if not os.path.exists(options.agentbin):
        return error("Directory %s doesn't exist; is $AVBIN set right?",
                     options.agentbin)

    # See if another scheduler is already running
    if isrunning(host, port):
        return error("Another scheduler is already running at %s:%s",
                     host, port)

    # Daemonize
    if opts.daemon:
        if daemonize() != 0:
            # Parent
            return 0
    if opts.DAEMON:
        daemonize(fork=False)

    # Initialize log handlers
    avalog.setupLogging()
    if opts.daemon:
        logger.info("Running as daemon process")

    # Create the data directory if necessary
    if not os.path.exists(options.agentdata):
        logger.info("Creating %s directory", options.agentdata)
        try:
            os.makedirs(options.agentdata)
        except os.error as err:
            logger.error("Can't create %s directory: %s",
                         options.agentdata, err)
            return 1

    # On AIX 5.2, apply arcane SSL workaround.  Python can't seem to open
    # the character special files /dev/[u]random on 5.2, so use system()
    # to sneak some randomness out and seed the openssl PRNG with it.
    if sys.platform == 'aix5':
        p = os.popen('uname -r', 'r')
        osrel = p.read()
        p.close()
        logger.info('Seeding PRNG from /dev/urandom')
        ddcmd = 'dd if=/dev/urandom of=%s/randpool bs=128 count=1' % \
                options.agentdata
        p = os.popen(ddcmd, 'r')
        # not interested in dd's chatty output
        p.close()
        # Rand.load_file replaced: Python 3 ssl uses os.urandom internally
        pass

    # Check if an upgrade conversion is unnecessary
    check_for_upgrade()

    # See if there's a pw_check script to run
    if check_poliwall():
        return 1

    # Regenerate poliwall configuration.  This must be completed
    # before we start the poliwall, so can't be run as a scheduler
    # task.
    regen_pwconf()

    # Create the scheduler
    logger.info("Schedule file: %r", schedule_file)
    logger.info("Task directory: %r", taskdir)
    sched = scheduling.TaskScheduler(schedule_file, taskdir)
    if os.path.exists(schedule_file):
        logger.info("Loading existing tasks from %s", schedule_file)
        sched.load(schedule_file)
    schedule_template = os.path.join(options.agentbin, "schedule.xml")
    if os.path.exists(schedule_template):
        logger.info("Loading template tasks from %s", schedule_template)
        sched.load(schedule_template)
    sched.save()

    # Merge in the default schedule tasks

    # Create XML-RPC server; this fails if another scheduler is
    # already running
    address = (host, port)
    try:
        svr = server.SchedulerServer(address, sched, options,
                                     logRequests=False)
    except socket.error as msg:
        logger.error("Can't create XML-RPC server: %s", msg)
        return 1

    try:

        try:
            # Create the poliwall task if needed
            start_poliwall(sched)

            # Create the systray indicator task if needed
            start_indicator(sched)

            # Persist the modified schedule
            sched.save()

            # Start the scheduler (this starts the tasks)
            sched.start()

            # Start the ping-pong thread if needed
            pp = start_ping_pong_thread(sched, svr)

            # Handle XML-RPC requests until exit requested
            logger.info("Serving on %s:%d", host, port)
            svr.serve_until_stopped()

        except KeyboardInterrupt:
            logger.info("Keyboard interrupt; shutting down server")

    finally:

        # Stop server and scheduler
        try:
            conn = getConnection(True)
            conn.sendShutdown()
        except:
            logger.error("Unable to send shutdown message")
            
        sched.shutdown()
        svr.server_close()

    # Exit (successful unless the ping-pong thread's self-thread aborted us)
    return int(pp.error_exit)

def check_poliwall():
    """Run the pw_check[.bat] script, if it exists.

    If the pw_check script exists, call it, passing it the '-p' option
    if poliwall enforcement is disabled.

    """
    pw_check = "pw_check"
    if os.name == "nt":
        pw_check += ".bat"
    pw_check = os.path.join(options.agentlib, "modules", pw_check)
    if not os.path.exists(pw_check):
        return 0
    args = [pw_check]
    if os.name == "nt":
        args = ["cmd", "/c"] + args
    if not options.poliwall_enforce:
        args.append("-p")
    env = os.environ.copy()
    env["AVBIN"] = options.agentbin
    env["AVLIB"] = options.agentlib
    env["AVDATA"] = options.agentdata
    proc = process.createProcess(args, env=env)
    proc.wait()
    return 0

def check_for_upgrade():
    """Check if an upgrade conversion is necessary."""
    librevsfile = os.path.join(options.agentlib, "revs.txt")
    libversion = read_version_file(librevsfile, "agent")
    if not libversion:
        logger.info("No agent version in %s; skipping upgrade check",
                    librevsfile)
        return
    datarevsfile = options.datapath("revs.txt")
    dataversion = read_version_file(datarevsfile, "agent")
    if not dataversion:
        logger.info("Writing initial data revs file")
        update_version_file(datarevsfile, "agent", libversion)
        return
    if libversion == dataversion:
        logger.info("Version %s unchanged; no upgrade conversion needed",
                    dataversion)
        return
    logger.info("Need upgrade conversion from %s to %s",
                dataversion, libversion)
    if do_upgrade_conversion(dataversion, libversion):
        update_version_file(datarevsfile, "agent", libversion)
    # XXX else, should this be considered a critical error?

def do_upgrade_conversion(dataversion, libversion):
    """Do the upgrade conversion.

    This invokes an external program, conversion.py.

    """
    logger.info("Invoking upgrade conversion script...")
    script = os.path.join(options.agentbin, "conversion.py")
    args = [sys.executable, script, dataversion, libversion]
    logger.debug("args = %s", args)
    proc = process.createProcess(args)
    proc.wait()
    sts = proc.getstatus()
    if sts == 0:
        logger.info("Upgrade conversion script finished.")
        return True
    else:
        logger.error("Upgrade conversion script failed (sts=0x%x)", sts)
        return False

def read_version_file(revsfile, key):
    """Helper to read a specific key from a revisions file."""
    if not os.path.exists(revsfile):
        logger.debug("No file %s", revsfile)
        return None
    try:
        f = open(revsfile, "r")
    except IOError as err:
        logger.error("Can't open existing %s: %s", revsfile, err)
        return None
    version = None
    try:
        key += "="
        for line in f:
            if line.startswith(key):
                version = line[len(key):].strip()
                break
    finally:
        f.close()
    return version

def update_version_file(revsfile, key, version):
    """Helper to update a specific key in the revisions file."""
    try:
        f = open(revsfile, "r")
    except IOError as err:
        lines = []
    else:
        try:
            lines = f.readlines()
        finally:
            f.close()
    try:
        f = open(revsfile, "w")
    except IOError as err:
        logger.error("Can't write %s: %s", revsfile, err)
        return
    try:
        key += "="
        updated = 0
        newline = key + version + "\n"
        for line in lines:
            if line.startswith(key):
                line = newline
                updated += 1
            f.write(line)
        if not updated:
            f.write(newline)
    finally:
        f.close()
    logger.info("Wrote %s%s to %s", key, version, revsfile)

def regen_pwconf():
    """Regenerate poliwall configuration."""
    directive = options.datapath("policy", "dir.pw.1.xml")
    if not os.path.exists(directive):
        logger.info("Not generating poliwall configuration.")
    else:
        logger.info("Generating poliwall configuration...")
        engine = os.path.join(options.agentbin, "engine.py")
        args = [sys.executable, engine, "-p", "pw.1"]
        proc = process.createProcess(args)
        proc.wait()
        logger.info("Poliwall configuration generated.")

def isrunning(host, port):
    """Find out whether another scheduler is already running.

    Try to connect to 'our' host and port.

    """
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.connect((host, port))
        s.close()
    except socket.error:
        return False
    else:
        return True

def daemonize(fork=True):
    # Make sure the log directory exists
    if not os.path.isdir(logdir):
        try:
            os.makedirs(logdir)
        except os.error as err:
            if not os.path.isdir(logdir):
                raise
    # Fork a child, or not
    posix = hasattr(os, "fork")
    if posix:
        if fork:
            pid = os.fork()
            if pid != 0:
                return pid # Parent
            if sys.platform == "aix5":
                # Must exec on this platform to avoid problems
                argv = [arg for arg in sys.argv
                            if arg != "-d" and not arg.startswith("--d")]
                argv.insert(0, sys.executable)
                argv.append("--DAEMON")
                try:
                    os.execv(sys.executable, argv)
                except Exception as err:
                    print("Can't exec:", err, file=sys.stderr)
                    os._exit(127)
        devnull = "/dev/null"
    else:
        # Windows has a different name for /dev/null
        devnull = "nul:"
    # Child, or Windows
    os.chdir("/")
    # Redirect input to devnull
    os.close(0)
    sys.stdin = sys.__stdin__ = open(devnull, "r")
    # Redirect output to $AVDATA/log/debug.YYYY-MM-DD.txt
    ts = time.localtime()
    debugout = os.path.join(
        logdir, options.debug_prefix + ".%04d-%02d-%02d.txt" % ts[:3])
    os.close(1)
    fd = os.open(debugout, os.O_WRONLY|os.O_CREAT|os.O_APPEND, 0o600)
    os.dup2(fd, 2)
    sys.stdout = sys.__stdout__ = f = os.fdopen(fd, "a", 1) # Line buffering
    sys.stderr = sys.__stderr__ = f # Use the same file object
    print("-"*70)
    print("%04d-%02d-%02d %02d:%02d:%02d Scheduler starting" % ts[:6])
    # More ancient rituals
    if hasattr(os, "setsid"):
        try:
            os.setsid()
        except os.error as err:
            pass # Seems to occur naturally on AIX
    os.umask(0o22)
    if not posix:
        return 0
    if not fork:
        return 0 # XXX Should do the watchdog here
    # Watchdog code, introducing another level of parent/child
    avalog.setupLogging()
    n = 0
    while True:
        t0 = time.time()
        pid = os.fork()
        if pid == 0:
            return 0 # Child does the work
        # Parent is watchdog
        pid, sts = os.waitpid(pid, 0)
        if sts == 0:
            logger.info("Subprocess exited normally; stopping")
            os._exit(0)
        logger.warn("Subprocess %s", decode_status(sts))
        t1 = time.time()
        if t1 - t0 < 60:
            n += 1
            if n > 10:
                logger.error("Subprocess keeps exiting; giving up")
                os._exit(1)
            dt = 2**n
            logger.warn("Subprocess exited quickly; sleep %d", dt)
            time.sleep(dt)
        else:
            n = 0
    assert 0 # Not reached

def decode_status(sts):
    if sts == 0:
        return "exited normally"
    if hasattr(os, "WIFSIGNALED") and os.WIFSIGNALED(sts):
        return "killed by signal %d" % os.WTERMSIG(sts)
    if hasattr(os, "WIFEXITED") and os.WIFEXITED(sts):
        return "exited with status %d" % os.WEXITSTATUS(sts)
    return "raw exit status %d" % sts

def start_poliwall(sched):
    """Maybe start the poliwall.

    This needs to be too flexible to fit in the initial schedule.

    """
    # If it's unix and we're not root, delete instead of add the task
    command = None
    # The poliwall policy directive (dir.pw.1.xml) is what regen_pwconf()
    # compiles into the daemon's rule file; if the server has not assigned any
    # poliwall policy to this host it is absent (regen_pwconf logs "Not
    # generating poliwall configuration").
    pw_directive = options.datapath("policy", "dir.pw.1.xml")
    if not options.poliwall_enforce:
       # Enforcement is disabled.  The packet-filter daemon installs traffic
       # redirect rules (e.g. NFQUEUE iptables rules) whenever it runs, so
       # "monitor but don't enforce" would still divert -- and potentially
       # black-hole -- the host's traffic.  Treat enforce=false as "poliwall
       # off" and don't start it at all (matches gather getpwmode(), which
       # reports "disabled" in this case).  command stays None, so add_task()
       # removes any existing poliwall task.
       logger.info("poliwall-enforce is false; not starting poliwall")
    elif not os.path.exists(pw_directive):
       # No poliwall policy has been assigned to this host, so there is no rule
       # file to enforce.  Starting the daemon anyway opens an all-traffic DROP
       # interception (WinDivert filter="true"+FLAG_DROP, or NFQUEUE) with
       # nothing to allow, which black-holes the host.  No policy => nothing to
       # enforce => don't start it (command stays None, removing any existing
       # poliwall task).
       logger.info("No poliwall policy (%s absent); not starting poliwall",
                   pw_directive)
    elif not hasattr(os, "getuid") or os.getuid() == 0:
       # 'pwfilter' is either a duplicate or a link to the backend chosen by
       # pw_check: 'pwnfqueue' (Linux userspace, primary), 'pwnwext' (macOS
       # system extension), 'pwpcap'/'pwlinux' (legacy), or 'pwwinnt' (Windows).
       command = "pwfilter"
    args = [command] + (options.poliwall_options or "").split()
    add_task(sched, "poliwall", args)

def start_indicator(sched):
    """Maybe start the systray indicator.

    This needs to be too flexible to fit in the initial schedule.

    Actually, we don't do this any more.  The code here now exists to
    expunge the indicator task from the schedule might it exist
    previously.

    """
    task = tasks.newtask("indicator")
    args = [None]
    add_task(sched, "indicator", args)

def add_task(sched, id, args):
    task = tasks.newtask(id)
    command = args[0]
    if command:
        fullpath = os.path.join(options.agentbin, command)
        if os.name == "nt" and not command.endswith(".py"):
            fullpath += ".exe"
        if not os.path.exists(fullpath):
            logger.warn("Can't find %s", fullpath)
        else:
            task.command = " ".join(args)
            task.freq = 10
    sched.add_list([task]) # This deletes it if command not set

def start_ping_pong_thread(sched, svr):
    """Start the ping-pong thread and return the thread object.

    This creates a new thread that will ping the server every N
    seconds.  N is configurable (rfa-interval) and defaults to 60.
    Based upon the server response, further downloads and/or uploads
    are then carried out.

    The PingPong thread also periodocially carries out a self-check,
    and aborts the scheduler process if this fails.

    """
    pp = PingPongThread(sched, svr)
    pp.start()
    return pp

def getConnection(ssl):
        """Returns a new connection object.

        The argument is True if an SSL connection is requested, False
        otherwise.

        Don't bother catching socket or I/O errors here; the
        rest.Connection() class constructor doesn't do any I/O.

        """
        store = cert.CertificateStore(options=options)
        return rest.Connection(options.central_server,
                               ssl=ssl, certstore=store)

if __name__ == "__main__":
    sys.exit(main())
