404Forbiden

✏️ Editing: schedule.py

from anomaly import Anomaly
from datetime import datetime
from threshold import Threshold
import logging
import traceback
import sys
import time

try:
    import json
except ImportError:
    try:
        import simplejson as json
    # it's possible that we may not need json for the action that we're taking.
    # for example, for the rpm post install script, on a python version that
    # doesn't have json, we'll get this far in the code.  but the post
    # install doesn't use json, so we're fine
    except ImportError:
        json = None

# This represents a check that needs to be run, and all the associated
# processes on the check result that need to occur.
class Schedule(object):
    def __init__(self, schedule_data):
        self.log = logging.getLogger(self.__class__.__name__)
        self.update(schedule_data)

self.last_check_value = None
        self.next_check_time = datetime.now()
        self.number_of_checks = 0

self.cached_results = {}

# TODO: These two methods are also in Threshold and Anomaly; we should make
    # these three classes inherit from an Entity class that implements these.
    # The logging library interferes with cPickle, so we must remove the logger
    # instance then reset it when we serialize/unserialize.
    def __getstate__(self):
        state = dict(self.__dict__)
        del state["log"]
        return state

def __setstate__(self, state):
        self.__dict__.update(state)
        self.log = logging.getLogger(self.__class__.__name__)

@classmethod
    def create_test_schedule(cls):
        # none of this stuff really matters for the individual plugin tests
        data = {
            "id": 1,
            "plugin_textkey": "",
            "resource_textkey": "",
            "option": "null",
            "frequency": 60,
            "thresholds": [
                {
                    "id": 1,
                    "delay": 0,
                    "operator": "gt",
                    "value": 0,
                }
            ],
        }
        return cls(data)

def update(self, schedule_data):
        self.id = schedule_data["id"]
        self.plugin_textkey = schedule_data["plugin_textkey"]
        self.resource_textkey = schedule_data["resource_textkey"]
        if type(schedule_data["option"]) == dict:
            self.option = schedule_data["option"]
        else:
            self.option = json.loads(schedule_data["option"] or "null")
        self.frequency = schedule_data["frequency"]
        self.thresholds = []
        self.server_key = schedule_data.get("server_key", None)

def __repr__(self):
        return "<Schedule %d, %s.%s, %d>" % (
            self.id,
            self.plugin_textkey,
            self.resource_textkey,
            self.number_of_checks,
        )

def check(self, plugin_manager, anomalies):
        value = plugin_manager.check(self)
        self.last_check_value = value
        self.number_of_checks += 1
        if value is not None:
            self.log.debug(
                "Schedule %s, check #%d: %d", self.id, self.number_of_checks, value
            )
        else:
            self.log.debug(
                "Schedule %s, check #%d: <No value>", self.id, self.number_of_checks
            )

return value, anomalies

Sid Gifari From Gifari Industries - BD Cyber Security Team