# Copyright 2026 SigmaSRC, Inc. All Rights Reserved.

"""REST-style protocol.

TO DO:

- access /Fuel/...

MAYBE:

- uniformly return False for errors or raise RuntimeError?  (why bother?)
- when using SSL, validate server cert (this turns out to be a pain)

"""

import ssl
import time
import select
import socket
import urllib.parse
import http.client as httplib
import logging
import threading

from sigma.util import config, cert, xmltime, xmlparse
from sigma.protocol import xmlproto, hostcache

# The exceptions that can be raised by operations in this module
ALLERRORS = (EnvironmentError, RuntimeError,
             socket.error, ssl.SSLError, select.error,
             xmlparse.ParseError, httplib.HTTPException)

logger = logging.getLogger("protocol.rest")

class Connection(object):

    """A connection to the ECS 2.0 server.

    This uses either SSL or plain HTTP.  Most requests can be handled
    by either connection type, with the exception of registration
    requests, which must use SSL because a private key is sent back
    from the server.  A connection may be reused for several
    consecutive requests; i f the server cooperates this will keep the
    TCP/IP socket open, reducing connection roundtrips and (in the SSL
    case) authentication costs.  If the server closes the connection
    after servicing a request, the next request will transparently
    open a new connection.  If the server closes the connection in the
    middle of servicing a request, an exception will be raised.

    Several methods take an argument and return a value representing
    the requested object's last-modified time.  This is a string
    representing the time the object was last modified on the server,
    in an HTTP-specific format.  When this argument is None, the
    object is retrieved unconditionally.  When it is a string, the
    object is only retrieved if it was modified since the indicated
    time; if the object was not modified, None is returned instead of
    the object.  See the descrption of the If-Modified-Since header in
    the HTTP/1.1 spec.

    """

    def __init__(self, host=None, port=None,
                 ssl=True, certstore=None, options=None,
                 useTimeout=True, debugLevel=0):
        """Constructor.

        Arguments:

        host - the host to use
        port - the port to use
        ssl - bool indicating whether SSL should be used
        certstore - optional CertificateStore object
        options - optional Options object
        useTimeout - bool indicating whether I/O should be done in a
                     separate thread to allow timeout detection
                     (see method doIO() below)
        debugLevel - integer debugging level for the connection

        Defaults:

        debugLevel - defaults to 0
        useTimeout - defaults to True
        options - defaults to a new Options object
        certstore - defaults to a new CertificateStore object
        ssl - defaults to True
        host, port - default to options.central_server; 2 is added
                     to the default port when ssl is false.  If
                     options.central_server is not set, the defaults
                     are 'localhost' and ES_PORT (17747).

        In addition, if host is a string of the form 'host:port', that
        port value is used as the default port (and 2 is added when ssl
        is false).

        """
        self.useTimeout = useTimeout
        if options is None:
            options = config.Options()
        self.options = options
        if certstore is None:
            certstore = cert.CertificateStore(ssl=ssl, options=options)
        self.certstore = certstore
        self.ssl = ssl
        defport = config.ES_PORT
        if not host:
            host = options.central_server or "localhost"
        if ":" in host:
            host, _, sport = host.rpartition(":")
            if sport:
                defport = int(sport) # Too bad if this fails
        if not ssl:
            defport += 2
        if port is None:
            port = defport
        host = hostcache.cache_host(host, options)
        if ssl:
            from sigma.protocol.httpslib import HTTPSConnection
            ctx = certstore.getSSLContext()
            self.conn = HTTPSConnection(host, port, ssl_context=ctx)
        else:
            self.conn = httplib.HTTPConnection(host, port)
        self.conn.set_debuglevel(debugLevel)
        self.conn.auto_open = False

    def close(self):
        """Closes the connection.

        Multiple calls to this method are allowed.

        """
        self.conn.close()

    def connect(self):
        """Connect to the server if we haven't already."""
        if self.conn.sock is None:
            self.conn.connect()
        return self.conn

    def loadCertificate(self):
        """Loads the root certificate from the server.

        This retrieves /Certificate from the server and writes it to
        cacert.pem in the agent cert directory.

        """
        cacert, slmtime = self.simpleRequest("/Certificate", None, None)
        if self.certstore.create_cert_dir():
            self.certstore.write_cacert(cacert)

    def getAgentHome(self, slmtime):
        """Returns the parsed agent home page and its last-modified time."""
        if not self.certstore.agentID:
            logger.error("Too early to get agent home")
            return None, None
        url = "/Agent/agent.%d.xml" % self.certstore.agentID
        return self.simpleRequest(url, slmtime, xmlproto.Agent, strict=False)

    def getMatrix(self, full, slmtime, base=None, id=None):
        """Returns the parsed matrix and its last-modified time."""
        if full:
            url = "/Matrix/full.xml"
        else:
            if (base is not None) and (id is not None):
                url = "/Matrix/delta-" + str(base) + "-" + str(id) + ".xml"
            else:
                url = "/Matrix/delta.xml"
        return self.simpleRequest(url, slmtime, xmlproto.Matrix, strict=False)

    def getDirective(self, key, slmtime):
        """Returns the requested directive and its last-modified time.

        The directive is returned as a string in XML format; it is
        not parsed.

        """
        headers = {"X-ECS-upload-agent-id": str(self.certstore.agentID)}
        url = "/Directive/dir.%s.xml" % key
        return self.simpleRequest(url, slmtime, None, strict=True, headers=headers)

    def getConfig(self, slmtime):
        """Returns the contents of /Config/agent.config and its last mtime.

        If the file doesn't exist, return None, None.

        """
        url = "/Config/agent.config"
        try:
            return self.simpleRequest(url, slmtime)
        except RuntimeError:
            return None, None

    def getModule(self, rtversion, module, modversion):
        """Downloads a Fuel module."""
        url = "/Fuel/%s/%s.%s.fuel" % (rtversion,
                                       module.replace(".", "/"), modversion)
        try:
            return self.simpleRequest(url, None)[0]
        except RuntimeError:
            return None

    def simpleRequest(self, url, slmtime, cls=None, __filler=0, strict=True, headers=None):
        """Issues a conditional GET request.

        If cls is None, the retrieved data is returned as a string;
        otherwise, cls is assumed to be a subclass of
        sigma.util.xmlparse.ElementClass and used to parse the
        retrieved data (in particular, the class method
        __parseStream__() is called with the response object as an
        argument).  In this case, the keyword argument strict=False
        may be given to enable non-strict XML parsing.

        If slmtime is not None and represents a valid time equal to or
        later than the requested object's last-modified time, None is
        returned instead of the requested object.

        The return value is always of the form (obj, slmtime) where
        obj is either None or the requested object (parsed if cls is
        not None) and slmtime is the value of the Last-Modified header
        in the response, or None if no such header is found.

        This raises RuntimeError when an unexpected HTTP status code
        is received.  It may raise various other errors (socket errors
        etc.) when the connection is closed by the server in the
        middle of the request or when the server grossly misbehaves.

        """
        return self.doIO(self._simpleRequest, url, slmtime, cls, strict, headers)

    def _simpleRequest(self, url, slmtime, cls, strict, headers):
        """Internal version of simpleRequest()."""
        conn = self.connect()
        if headers is None:
            headers= {}

        if slmtime:
            headers["If-Modified-Since"] = slmtime
        t0 = time.time()
        conn.request("GET", url, None, headers)
        resp = conn.getresponse()
        self.logStatus(url, resp)
        t1 = time.time()
        self.roundtripTime = t1-t0
        self.dateHeader = resp.getheader("Date")
        if resp.status == 200:
            try:
                if cls is None:
                    data = resp.read()
                    if isinstance(data, bytes):
                        data = data.decode("utf-8")
                elif strict:
                    data = cls.__parseStream__(resp)
                else:
                    data = cls.__parseStream__(resp, strict=strict)
                return data, resp.getheader("Last-Modified")
            finally:
                resp.read() # Make sure to read until the end
        if resp.status == 304:
            resp.read()
            return None, resp.getheader("Last-Modified") or slmtime
        resp.read()
        raise RuntimeError("%s: HTTP status %s - %s" %
                           (url, resp.status, resp.reason), resp.status)

    def uploadFile(self, data, upload):
        """Uploads a file to the server.

        The data argument is the contents of the file as a string.

        The upload argument is an xmlproto.Upload instance
        representing the file's metadata; its type, startTime and
        stopTime attributes are used.

        Returns +1 if the upload was successful, 0 if not, -1 if the
        upload was successful but the server closed the connection.

        If the agent isn't registered yet, this always returns False.

        """
        if not self.certstore.agentID:
            logger.error("Too early to upload file")
            return 0
        return self.doIO(self._uploadFile, data, upload)

    def _uploadFile(self, data, upload):
        """Internal version of uploadFile()."""
        conn = self.connect()
        if isinstance(data, bytes):
            is_xml = data.startswith(b"<")
        else:
            is_xml = data.startswith("<")
        if is_xml:
            ctype = "text/xml"
        else:
            ctype = "application/octet-stream"
        headers = {"X-ECS-upload-type": upload.type,
                   "X-ECS-upload-agent-id": str(self.certstore.agentID),
                   "X-ECS-upload-start-time":
                       xmltime.formatdatetime(upload.startTime),
                   "X-ECS-upload-stop-time":
                       xmltime.formatdatetime(upload.stopTime),
                   "Content-Type": ctype}
        url = "/Upload"
        conn.request("POST", url, data, headers)
        resp = conn.getresponse()
        self.logStatus(url, resp)
        ok = int(resp.status in (200, 204))
        if not ok:
            logger.error("Upload status: %s - %s", resp.status, resp.reason)
            if resp.headers:
                logger.info("Headers = %r", resp.headers.items())
        try:
            # Read the body of the response
            resp.read()
        except IOError as err:
            if ok:
                logger.warn("Connection closed on upload (IOError: %s)", err)
                ok = -1
            else:
                logger.error("IOError: %s", err)
        return ok

    def sendShutdown(self):
        """Signal a shutdown to the server.

        This sends a message to the shutdown URL of the server.  It has
        a configurable timeout parameter to prevent the agent from holding up
        shutdown due to a network problem or server being down.
        """
        if not self.certstore.agentID:
            logger.error("Can't send shutdown message until agent registeres")
            return False
        assert self.ssl, "Shutdown must be done over SSL"
        return self.doIO(self._sendShutdown)

    def _sendShutdown(self):
        """Internal version of sendShutdown"""
        conn = self.connect()
        headers = {"X-ECS-upload-agent-id": str(self.certstore.agentID)}
        url = "/Shutdown"
        conn.request("POST", url, "", headers)
        resp = conn.getresponse()
        self.logStatus(url, resp)
        ok = int(resp.status in (200, 204))
        if not ok:
            logger.error("Shutdown status: %s - %s", resp.status, resp.reason)
            if resp.headers:
                logger.info("Headers = %r", resp.headers.items())

        return ok

    def requestCertificate(self):
        """Requests an new certificate for the agents..

        Connects to the server and requests a new certificate for the agent
        from the url /CertificateRequest.  This method can only be used to
        request a new certificate after the agent has registered.  If the
        agent hasn't registered, this will return False.

        """

        if not self.certstore.agentID:
            logger.error("Too early to request a renewal certificate")
            return False
        assert self.ssl, "Registration MUST be done using SSL."
        return self.doIO(self._requestCertificate)

    def _requestCertificate(self):
        """Internal version of requestCertificate."""
        conn = self.connect()
        url = "/CertificateRequest"
        headers = {"X-ECS-upload-agent-id": str(self.certstore.agentID)}
        # all of the necessary data is encoded in the headers 
        conn.request("POST", url, None, headers)
        resp = conn.getresponse()
        self.logStatus(url, resp)
        if (resp.status != 200):
            logger.error("CertificateRequest status: %s - %s",
                         resp.status, resp.reason)
            if resp.headers:
                logger.info("Headers = %r", resp.headers.items())
            return False
        cred = xmlproto.Credentials.__parseStream__(resp)
        if self.certstore.agentID != cred.agentID:
            logger.error("CertificateRequest send back a bad cert")
            return False
        agentCert = cred.cert.text
        agentKey = cred.privKey.text
        self.certstore.write_cert(agentCert)
        self.certstore.write_key(agentKey)
        return True

    def registerAgent(self, enterprise=False):
        """Requests an agent registration from the server.

        The optional Boolean argument indicates whether the agent
        wishes to register as an enterprise agent; it defaults to
        False.

        Returns True if the registration is successful, False
        otherwise.  Upon a successful registration, the agent ID,
        certificate and private key are written to the certificate
        store (the agent cert directory).

        """
        assert self.ssl, "Registration MUST be done using SSL."
        return self.doIO(self._registerAgent, enterprise)

    def _registerAgent(self, enterprise):
        """Internal version of registerAgent()."""
        conn = self.connect()
        url = "/Register"
        if enterprise:
            url += "?enterprise=true"
        conn.request("POST", url) # POST without sending data!
        resp = conn.getresponse()
        self.logStatus(url, resp)
        if resp.status != 200:
            logger.error("Registration status: %s - %s",
                         resp.status, resp.reason)
            if resp.headers:
                logger.info("Headers = %r", resp.headers.items())
            return False
        cred = xmlproto.Credentials.__parseStream__(resp)
        agentID = cred.agentID
        agentCert = cred.cert.text
        agentKey = cred.privKey.text
        if not self.certstore.create_cert_dir():
            return False
        self.certstore.write_id(str(agentID))
        self.certstore.write_cert(agentCert)
        self.certstore.write_key(agentKey)
        return True

    def doIO(self, callable, *args):
        """Executes IO in a separate thread.

        See class IOThread below.

        If the useTimeout argument to the Connection constructor is
        false, don't use the separate thread here; insteadl, just call
        the callable with the args.

        """
        if not self.useTimeout:
            return callable(*args)
        timeout = self.options.timeout_per_msg
        t = IOThread(callable, args)
        try:
            t.start()
        except threading.ThreadError as err:
            logger.error("Can't start IOThread: %s", err)
            raise RuntimeError("Can't start IOThread: %s" % err)
        t.join(timeout)
        if not t.done:
            raise RuntimeError("Request timed out after %s seconds" % timeout)
        if t.exception is not None:
            raise t.exception
        return t.result

    def logStatus(self, url, resp):
        """Log the status.

        We keep 4 separate status files in $AVDATA, based on whether
        there was an error and whether we used SSL or not.

        """
        files = ["last_error.txt",     "last_good.txt",
                 "last_error_ssl.txt", "last_good_ssl.txt"]
        bit0 = bool(200 <= resp.status <= 299)
        bit1 = bool(self.ssl)
        filename = files[bit0 + 2*bit1]
        filename = self.options.datapath(filename)
        try:
            f = open(filename, "w")
            try:
                f.write(time.strftime("%Y-%m-%d %H:%M:%S\n"))
                f.write("%s\n" % url)
                f.write("%s %s\n" % (resp.status, resp.reason))
                for key, value in resp.getheaders():
                    f.write("%s: %s\n" % (key, value))
            finally:
                f.close()
        except IOError as err:
            logger.error("Can't write %s: %s", filename, err)
            return

class IOThread(threading.Thread):

    """Separate thread for doing the actual I/O so we can time out on it.

    See method doIO() above.

    """

    def __init__(self, callable, args):
        threading.Thread.__init__(self, name="IOThread")
        self.daemon = True
        self.callable = callable
        self.args = args
        self.result = None
        self.exception = None
        self.done = False

    def run(self):
        try:
            self.result = self.callable(*self.args)
            self.done = True
        except Exception as err:
            self.exception = err
            self.done = True
            logger.debug("IOThread caught exception", exc_info=True)
