# Copyright 2026 SigmaSRC, Inc. All Rights Reserved.

"""REST-style Ping-Pong thread."""

import os
import sys
import time
import datetime
import random
import http.client as httplib
import logging
import calendar
import threading

from sigma.util import config, cert, ostypes, xmltime, timeskew
from sigma.util import dance
from sigma.util.enterprise import readEnterpriseFlag, writeEnterpriseFlag
from sigma.protocol import rest, xmlproto
from sigma.spool import qmgr
from sigma.runtime import matrix
from sigma.sched import tasks, remote, process, scheduling

ALLERRORS = rest.ALLERRORS

class PingPongThread(threading.Thread):

    """A Thread subclass managing the server connections.

    All server connections are made here.  Most of the action starts
    in mainLoop(), which runs until told to stop.

    The scheduler should create an instance and start() it.

    """

    # Error exit status for the scheduler.  We set this instance
    # variable to True when we exit with an error condition.
    error_exit = False

    def __init__(self, sched, svr, options=None):
        """Constructor."""
        threading.Thread.__init__(self, name=self.__class__.__name__)
        self.sched = sched
        self.svr = svr
        if options is None:
            options = config.Options()
        self.options = options
        self.logger = logging.getLogger("sched.pingpong")
        self.certstore = cert.CertificateStore(options=self.options)
        self.customize() # Hook for synthetic agent
        self.directives = self.readCurrentDirectives()
        self.enterprise = readEnterpriseFlag()
        self.checkInterfaces() # Initialize pwi_mtime and pwi_filename
        self.skewer = timeskew.TimeSkew(options)
        self.restart = False

    def customize(self):
        """This is a hook for the synthetic agent."""

    def readCurrentDirectives(self):
        """Returns the set of currently active directives.

        This uses both the directory listing of the policy directory
        and the task list, and returns the intersection of valid
        directive IDs found.  The 'pw.1' special directive is excluded.

        """
        taskids = set([tid[7:]
                       for tid in self.sched.list_tasks()
                       if tid.startswith("engine.")])
        dirids = set([xmlproto.get_policy_key(name, self.options)
                      for name in xmlproto.list_policydir(self.options)])
        # Subtract "pw.1" since it's not a real directive.  Subtract
        # None since get_policy_key() returns None for unknown files.
        result = dirids & taskids - set(["pw.1", None])
        self.logger.info("Active directives: %s", " ".join(result))
        return result

    def run(self):
        """Runs the main loop code.

        If it fails, log an error with a traceback.  If it exits
        without an exception, log an info message.

        This is a standard Thread API.

        """
        try:
            self.mainLoop()
        except:
            self.logger.exception("Ping-pong thread died of exception")
            self.shutDownServer(True)
        else:
            self.logger.info("Ping-pong thread exited")

    def mainLoop(self):
        """The main loop.

        This alternates bursts of activity and sleeps.

        """
        DELAY = max(self.options.rfa_interval, 1)
        FACTOR = max(self.options.backoff_factor, 1)
        LIMIT = max(self.options.backoff_limit, DELAY)
        delay = DELAY
        while True:
            if not self.pingScheduler():
                self.shutDownServer(True)
                break
            if self.repeatedTasks():
                if delay != DELAY:
                    delay = DELAY
                    self.logger.info("Reset delay to %.4g", delay)
            else:
                delay = min(delay * FACTOR, LIMIT)
                self.logger.info("Set delay to %.4g", delay)
            if self.restart:
                break
            sleeptime = self.randomizeDelay(DELAY, delay, LIMIT)
            if not self.sleep(sleeptime):
                self.shutDownServer(False)
                break

    def randomizeDelay(self, DELAY, delay, LIMIT):
        """Randomize the sleep time when we're in exponential back-off mode.

        This is a separate method for the benefit of the synthetic agent.

        """
        if delay > DELAY:
            # XXX Should really have a statistician look at this
            dt = delay - DELAY
            delay = random.uniform(DELAY, min(delay+dt, LIMIT))
            self.logger.debug("randomized sleep time %.1f", delay)
        return delay

    def pingScheduler(self):
        """Pings the local scheduler's remote API.

        This is a check for certain catastrophic errors that cause the
        agent to be uninstalled without killing the scheduler.  If we
        detect something like that, we should exit ungraciously.

        Returns True unless an error occurred.

        """
        sched = remote.Scheduler()
        try:
            sched.ping()
        except Exception as err:
            # Don't log a critical error if we're exiting already
            if not self.checkServerExit():
                self.logError("self-ping failed", err, logging.CRITICAL)
            return False
        else:
            return True

    def sleep(self, delay):
        """Sleeps for a given time, in 1-second increments.

        While sleeping, checks for server exit requests and for
        changes in the set of active interfaces.

        The sleep time is rounded up to whole seconds.

        Returns True unless we should drop out of the main loop.

        """
        while True:
            if self.checkServerExit():
                return False
            if self.checkInterfaces():
                self.recomputePoliwallConfiguration()
                break
            time.sleep(1)
            delay -= 1
            if delay <= 0:
                break
        return True

    def checkServerExit(self):
        """Checks whether the server has requested us to exit.

        Returns True if we should exit now, False otherwise.

        """
        return self.svr.handler.exit

    pwi_filename = None
    pwi_mtime = None

    def checkInterfaces(self):
        """Check whether the set of active interfaces has changed.

        Returns True if the set has changed since the last call, False
        otherwise.

        """
        # XXX There's a kind of race condition here.  If we ignore the
        # initial mtime, we'll miss the (rare) case where the set of
        # interfaces changes after the poliwall has started but before
        # the PingPongThread instance is created.  But if we honor the
        # initial change, we'll always recompute pwconf.txt twice at
        # startup.  For now, we're ignoring the initial mtime change,
        # by virtue of calling this method in __init__() and ignoring
        # its return value there.  (We may still recompute it twice if
        # the poliwall is slow starting up.)
        if self.pwi_filename is None:
            self.pwi_filename = self.options.datapath("interfaces.txt")
        try:
            mtime = os.path.getmtime(self.pwi_filename)
        except os.error:
            mtime = None
        if mtime == self.pwi_mtime:
            return False
        self.pwi_mtime = mtime
        self.logger.info("Interface up/down event, rechecking")
        return True

    def repeatedTasks(self):
        """Generates a burst of energy.

        Returns True unless an error occurred.

        """
        if not self.certstore.check_registration():
            if not self.tryToRegister():
                return False
        if not self.probeVarious():
            return False
        qm = qmgr.QueueManager(options=self.options)
        if qm.is_empty():
            return True
        return self.uploadFiles(qm)

    def tryToRegister(self):
        """Tries to register the agent.

        Returns True unless an error occurred.

        """
        self.logger.info("AGENT REGISTERING WITH %s...",
                         self.options.central_server)
        conn = self.getConnection(False)
        try:
            try:
                conn.loadCertificate()
            except ALLERRORS as err:
                self.logError("AGENT REGISTRATION FAILED LOADING CERT", err)
                return False
        finally:
            conn.close()
        conn = self.getConnection(True)
        try:
            try:
                ok = conn.registerAgent(self.enterprise)
            except ALLERRORS as err:
                self.logError("AGENT REGISTRATION FAILED", err)
                return False
        finally:
            conn.close()
        if not ok:
            self.logger.error("AGENT REGISTRATION FAILED")
            return False
        self.logger.info("AGENT REGISTERED: %s", self.certstore.agentID)
        return True

    def probeVarious(self):
        """Probes the server for changes to various files.

        This probes the matrix, the per-agent home page, and the
        global agent configuration file, and takes appropriate action
        if any one is changed.

        Returns True unless an error occurred.

        """
        conn = self.getConnection(False)
        try:
            return (self.probeMatrix(conn) and
                    self.probeIndex(conn) and
                    self.probeConfig(conn))
        finally:
            conn.close()

    matrix_id = None
    matrix_slmtime = None

    def probeMatrix(self, conn):
        """Probes /Matrix/{full,delta}.xml for changes, and acts on them.

        Returns True unless an error occurred.

        """

        if self.matrix_id is None:
            self.matrix_id = matrix.getmatrixid(self.options)

        try:
            # get latest delta
            update, matrix_slmtime = conn.getMatrix(False, self.matrix_slmtime)
            if (update is None) or (update.id == self.matrix_id):
                if matrix_slmtime:
                    self.matrix_slmtime = matrix_slmtime
                self.logger.debug("Matrix unchanged")
                return True
            m = matrix.loadmatrix(self.options)
            # download previous deltas or full matrix
            initial_id = m.id
            while self.matrix_id < update.base:
                try:
                    u, ms = conn.getMatrix(False, None, self.matrix_id,
                                           self.matrix_id + 1)
                except RuntimeError as err: 
                    if err.args[1] == 404:
                        # previous delta does not exist, so download full matrix
                        update, matrix_slmtime = conn.getMatrix(True, None)
                        break
                    else: 
                        if initial_id != m.id:
                            # we got some deltas, save them
                            m.store()
                            self.recomputePoliwallConfiguration()
                            self.logger.info("Got few matrix deltas from id %r to %r", initial_id, m.id)
                            return True
                        else:
                            # we did not get any deltas
                            self.logError("matrix", err)
                            return False
                m.update(u)
                self.matrix_id = m.id
                self.matrix_slmtime = ms
            # update matrix
            m.update(update)
            self.matrix_id = m.id
            self.matrix_slmtime = matrix_slmtime
            m.store()
            self.recomputePoliwallConfiguration()
            return True
        except ALLERRORS as err:
            self.logError("matrix", err)
            return False

    ahome = None
    ahome_slmtime = None

    def probeIndex(self, conn):
        """Probes /Agent/agent.<agentid>.xml for changes and takes actions.

        Returns True unless an error occurred.

        """
        try:
            t0 = time.time()
            # download the home page unconditionally if agent does not 
            # have any directives (5666)
            if len(self.directives) == 0:
                ahome, ahome_slmtime = conn.getAgentHome(None)
            else:
                ahome, ahome_slmtime = conn.getAgentHome(self.ahome_slmtime)
            t1 = time.time()
        except ALLERRORS as err:
            self.logError("agent home", err)
            return False
        roundtrip = int(conn.roundtripTime*1000)
        self.logger.info("Roundtrip time %d msec", roundtrip)
        ok = True
        if roundtrip > self.options.slow_rfa_cutoff:
            self.logger.warn("Network is too slow")
            # Don't bail out right away -- if there's a change to the
            # home page, process it normally.  However we'll return
            # False at the end to cause backing off the polling
            # frequency and to skip uploads.
            ok = False
        else:
            self.setTimeSkew(t0, conn, t1)
        if ahome_slmtime:
            self.ahome_slmtime = ahome_slmtime
        if ahome is None:
            self.logger.debug("Agent home unchanged")
            return ok
        if ahome == self.ahome:
            self.logger.info("Agent home unchanged")
            # We must still process upgrades,
            # unless the network is too slow
            # XXX Why not in the "ahome is None" branch above?
            if ok:
                self.processUpgrades()
            return ok
        self.logger.info("Agent home changed")
        self.logger.debug("New agent home:\n%s", ahome)
        self.ahome = ahome
        if (ahome.enterprise is not None and
            ahome.enterprise != self.enterprise):
            self.logger.info("Changing enterprise to %s", ahome.enterprise)
            self.enterprise = ahome.enterprise
            writeEnterpriseFlag(self.enterprise)
        self.processUpgrades()
        self.processExemptions()
        if not self.processDirectives(conn):
            ok = False
        if (ahome.poliwall_enforce is not None and
            ahome.poliwall_enforce != self.options.poliwall_enforce):
            self.logger.info("Changing poliwall-enforce to %s; restarting",
                             ahome.poliwall_enforce)
            self.options.poliwall_enforce = ahome.poliwall_enforce
            self.options.write()
            # We must restart the scheduler now, because it needs to
            # re-run the pwcheck script with the changed option, and
            # then restart the poliwall binary picked by that script.
            # XXX We could avoid the restart like this:
            # 1. stop the poliwall task
            # 2. run the pwcheck script
            # 3. restart the poliwall task
            # But that requires refactoring the scheduler startup magic.
            self.restartServer()
            ok = False
        return ok

    def setTimeSkew(self, t0, conn, t1):
        """Computes the clock offset between server and agent."""
        if not conn.dateHeader:
            self.logger.warn("Server sent no Date; not setting time skew")
            return
        try:
            servertime = time.strptime(conn.dateHeader,
                                       "%a, %d %b %Y %H:%M:%S GMT")
        except ValueError as err:
            self.logger.warn("Can't parse Date header (%s): %s",
                             conn.dateHeader, err)
            return
        ts = calendar.timegm(servertime)
        t0 = int(t0)
        t1 = int(t1)
        if t0 != t1:
            self.logger.debug("not changing skew (request crossed second)")
            return
        olds = self.skewer.get_skew()
        news = ts - t0
        self.logger.debug("skew %s -> %s", olds, news)
        if abs(olds - news) <= 1 and news != 0:
            self.logger.debug("not changing skew (within 1 sec of old skew)")
        elif news != olds:
            self.logger.info("Setting time skew to %d", news)
            self.skewer.set_skew(news)

    def processDirectives(self, conn):
        """Makes sure the set of active directives matches self.ahome.

        Returns True unless an error occurred.

        """
        olddirectives = self.directives
        newdirectives = set()
        for d in self.ahome.directives:
            key = "%s.%s" % (d.id, d.version)
            newdirectives.add(key)
        if newdirectives == olddirectives:
            self.logger.debug("No changes in directives")
            return True
        for key in newdirectives:
            if key not in olddirectives:
                if not self.copyDirective(key, conn):
                    return False
        self.directives = newdirectives.copy()
        try:
            names = xmlproto.list_policydir(self.options)
        except os.error as err:
            self.logError("list policydir", err)
        else:
            newdirectives.add("pw.1")
            pdir = xmlproto.get_policydir(self.options)
            for name in names:
                key = xmlproto.get_policy_key(name, self.options)
                if not key or key not in newdirectives:
                    fullname = os.path.join(pdir, name)
                    try:
                        os.remove(fullname)
                    except os.error as err:
                        self.logError("remove", err)
                    else:
                        self.logger.info("Deleted %s", fullname)
        return self.scheduleDirectives(conn)

    def scheduleDirectives(self, conn):
        """Makes sure the schedule matches the set of loaded directives.

        This also computes the pw.1 directive and schedules it.

        """
        tasklist = [self.getPoliwallTask()]
        pwdirective = xmlproto.Directive(id=None, version=1,
                                         name="[Internal Poliwall]")
        pwmodekey = None
        for key in self.directives:
            taskid = "engine." + key
            command = "engine.py " + key
            try:
                dir = xmlproto.load_directive(key, self.options)
            except ALLERRORS as err:
                self.logError("load directive %s" % key, err)
                return False
            task = dir.task or tasks.newtask()
            task.id = taskid
            task.command = command
            tasklist.append(task)
            pwmode = None
            for rule in dir.rules:
                if rule.category == "packet filter":
                    pwdirective.rules.append(rule)
                    pwmode = dir.poliwall_mode
            if pwmode:
                if not pwdirective.poliwall_mode:
                    pwdirective.poliwall_mode = pwmode
                    pwmodekey = key
                elif pwmode != pwdirective.poliwall_mode:
                    self.logger.error("Inconsistent pwmode: %s->%s, %s->%s",
                                      pwmodekey, pwdirective.poliwall_mode,
                                      key, pwmode)
        xmlproto.create_policydir(self.options)
        f = open(xmlproto.dir_filename("pw.1", self.options), "w")
        try:
            pwdirective.__render__(f)
        finally:
            f.close()
        for taskid in self.sched.list_tasks():
            if taskid.startswith("engine."):
                key = taskid[7:]
                if key != "pw.1" and key not in self.directives:
                    tasklist.append(tasks.newtask(taskid))
        tasknames = [(t.command and "%s+" or "%s-") % t.id for t in tasklist]
        tasknames.sort()
        self.logger.info("Tasks: %s", " ".join(tasknames))
        self.sched.add_list(tasklist)
        self.sched.save()
        return True

    def copyDirective(self, key, conn):
        """Copies a directive from the server to the policy directory.

        Returns True unless an error occurred.

        """
        try:
            data, slmtime = conn.getDirective(key, None)
        except ALLERRORS as err:
            self.logError("copy directive", err)
            return False
        xmlproto.create_policydir(self.options)
        fn = xmlproto.dir_filename(key, self.options)
        f = open(fn, "w")
        try:
            f.write(data)
        finally:
            f.close()
        self.logger.info("Wrote %s", fn)
        return True

    def processUpgrades(self):
        """Processes any <upgrade> elements present in the agent home page."""
        if not self.ahome.upgrades:
            return # Nothing to do
        cmd = xmlproto.Command(upgradeList=self.ahome.upgrades)
        fn = self.options.datapath("upgrades.xml")
        try:
            f = open(fn, "w")
        except IOError as err:
            self.logError("write upgrades", err)
            return # Not server-related
        try:
            cmd.__render__(f)
        finally:
            f.close()
        self.logger.info("Wrote upgrades to %s", fn)
        self.logger.info("Starting upgrade.py as independent subprocess")
        args = [sys.executable, os.path.join(self.options.agentbin,
                                             "upgrade.py"),
                "xml", fn]
        self.logger.debug("Starting process %s", args)
        try:
            proc = process.createProcess(args)
        except process.Error as err:
            self.logger.error("Can't start process %s: %s", args, err)
        else:
            self.logger.info("Upgrade subprocess spawned")

    def processExemptions(self):
        """Writes the rule exemptions from the home page to a file.

        We do this even if there are no exemptions -- there might have
        been exemptions before that are now lifted.

        We don't bother checking if there were any changes -- I don't
        see anything we'd gain by doing that.

        We don't return a status -- this shouldn't fail.

        """
        if not self.ahome.exempt:
            ruleids = []
        else:
            ruleids = self.ahome.exempt.get_rule_ids()
        fn = self.options.datapath("exempt.txt")
        fnn = fn + ".new"
        f = open(fnn, "w")
        try:
            for ruleid in ruleids:
                f.write("%s\n" % ruleid)
        finally:
            f.close()
        err = dance.rename(fnn, fn)
        if err:
            self.error("Can't move %s to %s: %s", fnn, fn, err)

    def uploadFiles(self, qm, filter=None):
        """Uploads files to the server.

        Returns True unless an error occurred.

        """
        qm.consolidate_duplicates()
        conn = self.getConnection(True)
        try:
            return self._uploadFiles(qm, conn, filter)
        finally:
            conn.close()

    def _uploadFiles(self, qm, conn, filter):
        """Internal version of uploadFiles()."""
        uploads = []
        for qitem in qm.list_items():
            type = qitem.metadata.get("type")
            if filter:
                # Only upload files of the given type
                if type not in filter:
                    continue
            startTime = xmltime.parse(qitem.metadata.get("starttime"))
            stopTime = xmltime.parse(qitem.metadata.get("endtime"))
            upload = xmlproto.Upload(type=type,
                                     startTime=startTime,
                                     stopTime=stopTime)
            upload.qitem = qitem
            uploads.append(upload)
        if not uploads:
            if filter:
                self.logger.info("No uploads found matching %s", filter)
            else:
                self.logger.info("No uploads found")
            return True
        uploads.sort(key=lambda u: u.startTime)
        totalkb = 0
        totalcount = 0
        t0 = time.time()
        ok = 1
        for upload in uploads:
            if (ok < 0 or
                totalkb >= self.options.upload_max_kb or
                totalcount >= self.options.upload_max_files):
                dt = time.time() - t0
                self.logger.info(
                    "Stop uploading after %s kb, %s files, %.3f secs",
                    totalkb, totalcount, dt)
                return True
            qitem = upload.qitem
            md = qitem.metadata
            num_upload_attempts = int(md.get("num-upload-attempts", 0))
            if num_upload_attempts >= self.options.max_upload_attempts:
                # config file changed
                self.logger.debug(md["qfile"] + ": num_upload_attempts="
                                  + str(num_upload_attempts))
                self.logger.warn("Discarding queue item")
                qitem.discard()
                continue
            if "prev-upload-attempt" in qitem.metadata:
                prev_upload_attempt = qitem.metadata["prev-upload-attempt"]
                prev_upload_attempt_dt = xmltime.parse(prev_upload_attempt)
                now_dt = datetime.datetime.fromtimestamp(time.time(), datetime.timezone.utc).replace(tzinfo=None)
                td = now_dt - prev_upload_attempt_dt
                if (scheduling.timedelta_to_seconds(td)
                    < self.options.upload_attempt_period):
                    continue
            totalcount += 1
            type = upload.type
            if type == "compliance":
                type += ":" + qitem.metadata.get("deployment_id", "")
            try:
                kb = (os.path.getsize(qitem.filename) + 999) // 1000
                totalkb += kb
                self.logger.info("Uploading %s: %d kb (%s)",
                                 type, kb, upload.startTime)
                f = open(qitem.filename, "rb")
                try:
                    data = f.read()
                finally:
                    f.close()
                ok = conn.uploadFile(data, upload)
            except ALLERRORS as err:
                self.logError("upload file", err)
                break
            if ok:
                qitem.discard()
            else:
                num_upload_attempts += 1
                self.logger.debug(md["qfile"] + ": num_upload_attempts="
                                  + str(num_upload_attempts))
                if (num_upload_attempts >= self.options.max_upload_attempts):
                    self.logger.warn("Upload error; discarding queue item")
                    qitem.discard()
                else:
                    self.logger.warn("Upload error; keeping queue item")
                    md["num-upload-attempts"] = num_upload_attempts
                    prev_upload_attempt = xmltime.formattimestamp(time.time())
                    md["prev-upload-attempt"] = prev_upload_attempt
                    qitem.rewrite_metadata()
                return False
        else:
            dt = time.time() - t0
            if filter:
                self.logger.info(
                    "All queue items matching %s uploaded: "
                    "%s kb, %s files, %.3f secs",
                    filter, totalkb, totalcount, dt)
            else:
                self.logger.info(
                    "All queue items uploaded: %s kb, %s files, %.3f secs",
                    totalkb, totalcount, dt)
            return True

    def recomputePoliwallConfiguration(self):
        """Schedules an event that recomputes the poliwall configuration."""
        fn = xmlproto.dir_filename("pw.1")
        if not os.path.exists(fn):
            self.logger.info("pw.1 directive doesn't exist yet")
            return
        self.logger.info("Scheduling engine.pw.1 task to recompute pwconf.txt")
        self.sched.add_list([self.getPoliwallTask()])

    aconfig = None
    aconfig_slmtime = None

    def probeConfig(self, conn):
        """Probes /Config/agent.config and take actions."""
        try:
            aconfig, slmtime = conn.getConfig(self.aconfig_slmtime)
        except ALLERRORS as err:
            self.logError("get configuration", err)
            return False
        self.aconfig_slmtime = slmtime
        if aconfig is None:
            self.logger.debug("Configuration unchanged")
            return True
        if aconfig == self.aconfig:
            self.logger.info("Configuration unchanged")
            return True
        self.aconfig = aconfig
        configtext = aconfig.strip()
        try:
            changed = self.options.updatefromtext(configtext)
        except Exception as err:
            self.logger.exception("Failed to update configuration")
            self.logger.info("Offending configuration text:")
            for line in configtext.splitlines():
                self.logger.info("    %s", line)
        else:
            if not changed:
                self.logger.info("Configuration is up to date")
            else:
                self.logger.info("Configuration updated; restarting")
                self.logger.info("Added configuration text:")
                for line in configtext.splitlines():
                    self.logger.info("    %s", line)
                self.restartServer()
        return True

    def getPoliwallTask(self):
        """Returns a new Task instance that runs the pw.1 directive."""
        return tasks.newtask("engine.pw.1", "engine.py -p pw.1")

    def shutDownServer(self, error_exit):
        """Shuts down the server and scheduler."""
        self.logger.info("Shutting down from inside PingPong thread")
        self.error_exit = error_exit
        self.svr.handler.exit = True
        self.svr.force_exit()
        self.sched.shutdown()
        self.svr.server_close()

    def restartServer(self):
        """Restarts the server.  Hackish."""
        self.logger.info("Restarting from inside PingPong thread")
        if os.name == "nt":
            # Restarting the service on Windows is a pain;
            # instead, we exit and the watchdog will restart us.
            self.shutDownServer(True)
        else:
            # This looks silly: we're using an RPC call to shut down
            # this very process.  But all attempts to invoke the same
            # functionality directly on self.sched and self.svr have
            # remained fruitless; the RPC approach is solid.
            sched = remote.Scheduler()
            sched.restart()
        self.restart = True

    def getConnection(self, 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.

        """
        return rest.Connection(self.options.central_server,
                               ssl=ssl, certstore=self.certstore)

    def logError(self, msg, err, level=None):
        """Logs an error.

        The first argument is a short string indicating the action
        that failed.  The second argument is an Exception instance.
        The third is an optional logging level, which defaults to
        ERROR.

        Special case: if the error is BadStatusLine with an empty
        line, log a less severe-looking warning.

        """
        if isinstance(err, httplib.BadStatusLine) and err.line == "":
            if level is None:
                level = logging.WARNING
            self.logger.log(level, "%s: connection closed early", msg)
            return
        if level is None:
            level = logging.ERROR
        name = err.__class__.__name__
        mod = err.__class__.__module__
        if mod != "exceptions":
            name = "%s.%s" % (mod, name)
        self.logger.log(level, "%s: %s: %s", msg, name, err)
