DEFAULT_CFG_PATH = "/etc/ripe-atlas-monitor/config.cfg"
    DEFAULT_CFG = {
        "var_dir": "",
        "verbosity": 0,
        "logging": {
            "file_path": "",
            "max_bytes": 10000000,
            "backup_cnt": 3,
            "email": {
                "to_addr": "",

                # on parse, if one of the following parameters is null
                # it's set with the same taken from default_smtp
                "smtp_host": str,
                "smtp_port": int,
                "from_addr": str,
                "subject": str,
                "use_ssl": bool,
                "username": str,
                "password": str,
                "timeout": int
            }
        },
        "ip_cache": {
            "dir": "",  # on load, it will be set to var_dir if missing
            "use_ixps_info": True,
            "lifetime": 604800
        },
        "default_smtp": {
            "smtp_host": "",
            "smtp_port": 25,
            "from_addr": "",
            "to_addr": "",
            "subject": "",
            "use_ssl": False,
            "username": "",
            "password": "",
            "timeout": 60
        },
        "default_syslog": {
            "socket": "",
            "host": "",
            "port": 514,
            "file": "",
            "facility": "user",
            "priority": "warning"
        },
        "misc": {
            "email_addr_re": "^[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}$",
            "env_prefix": "RIPE_ATLAS_MONITOR_",
            "msm_results_days_limit": 7,
            "editor": "/usr/bin/vim"
        }
    }
class GlobalConfig(object):
    def __init__(self):
    def _test_dir(dir_path, config_param):
    def get_default_path(self):
    def init_file(self, path):
    def load(self, cfg_path=DEFAULT_CFG_PATH):
    def verify_dirs(self):
    def parse(self, custom):
    def merge(self, orig, new):
    def _get(self, cfg, param):
    def get(self, param): class IPCache(object):
    def __init__(self):
        self.ip_cache = None

    def setup(self, **kwargs):
        logger.debug("Initializing IP cache...")

        if "_dir" in kwargs:
            _dir = kwargs["_dir"]

        if "IP_ADDRESSES_CACHE_FILE" in kwargs:
            IP_ADDRESSES_CACHE_FILE = kwargs["IP_ADDRESSES_CACHE_FILE"]
        else:
            IP_ADDRESSES_CACHE_FILE = "{}/ip_addr.cache".format(_dir)

        if "IP_PREFIXES_CACHE_FILE" in kwargs:
            IP_PREFIXES_CACHE_FILE = kwargs["IP_PREFIXES_CACHE_FILE"]
        else:
            IP_PREFIXES_CACHE_FILE = "{}/ip_pref.cache".format(_dir)

        if "IXP_CACHE_FILE" in kwargs:
            IXP_CACHE_FILE = kwargs["IXP_CACHE_FILE"]
        else:
            IXP_CACHE_FILE = "{}/ixps.cache".format(_dir)

        try:
            self.ip_cache = IPDetailsCache(
                IP_ADDRESSES_CACHE_FILE=IP_ADDRESSES_CACHE_FILE,
                IP_PREFIXES_CACHE_FILE=IP_PREFIXES_CACHE_FILE,
                MAX_CACHE=kwargs["lifetime"],
                dont_save_on_del=True
            )
        except Exception as e:
            raise ProgramError(
                "Error while setting up the IP cache: {}".format(str(e))
            )

        try:
            if kwargs["use_ixps_info"]:
                self.ip_cache.UseIXPs(
                    WhenUse=1,
                    IXP_CACHE_FILE=IXP_CACHE_FILE
                )
        except IPDetailsCacheIXPInformationError as e:
            raise ConfigError(
                "An error occurred while setting up the IP addresses cache: "
                "{} - "
                "IXPs information are not available at the moment; please "
                "consider setting the ip_cache.use_ixps_info to False to "
                "temporary avoid problems.".format(str(e))
            )
        except Exception as e:
            raise ProgramError(
                "Error while setting up the IXPs cache: {}".format(str(e))
            )
        except KeyboardInterrupt:
            raise ConfigError(
                "Aborting IP cache initialization"
            )

        logger.debug("IP cache initialized.")

    def get_ip_info(self, IP):
        return self.ip_cache.GetIPInformation(IP)

    def save(self):
        if self.ip_cache:
            self.ip_cache.SaveCache() class IPCacheManager(BaseManager):
    pass class LockFile(object):

    def __init__(self):
        self.fd = None
        self.path = None

    def acquire(self, path):
        if not path:
            return True

        self.path = path

        self.fd = os.open(self.path, os.O_CREAT)
        try:
            fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            return True
        except (OSError, IOError) as e:
            if e.errno == errno.EAGAIN:
                return False
            else:
                raise

    def release(self):
        if self.fd:
            os.close(self.fd)
            os.remove(self.path) class ProbesFilter(object):
    def __init__(self, probe_ids=None, countries=None):
        self.probe_ids = probe_ids or []
        self.countries = countries or []

    def __contains__(self, probe):
        # probe = Probe object
        if not self.probe_ids and not self.countries:
            return True
        if self.probe_ids and probe.id not in self.probe_ids:
            return False
        if self.countries and probe.country_code not in self.countries:
            return False
        return True LOG_LVL_RESULT = 25
LOG_LVL_LOG_ACTION = 27
class CustomLogger(object):
class CustomSysLogLogger(object):
    def __init__(self):
    def setup(self, verbosity_lvl, stdout=True):
    def log(self, lvl, msg, exc_info=False):
    def error(self, msg, exc_info=False):
    def warning(self, msg):
    def info(self, msg):
    def debug(self, msg):
    def result(self, msg):
    def action_log(self, msg):
    def __init__(self, name):
    def setup(self, socket, address, facility):
    def log(self, lvl, msg, *args, **kwargs): class Monitor(BasicConfigElement, MsmProcessingUnit):
    """Monitor

    A monitor allows to process results from a measurement.

    `descr` (optional): monitor's brief description.

    `measurement-id` (optional): measurement ID used to gather results. It can
    be given (and/or overwritten) via command line argument `--measurement-id`.

    `matching_rules`: list of rules to match probes against. When a probe
    matches one of these rules, its expected results are processed and its
    actions are performed.

    `expected_results` (optional): list of expected results. Probe's expected
    results contain references to this list.

    `actions` (optional): list of actions to be executed on the basis of
    probe's expected results.

    `stream` (optional): boolean indicating if results streaming must be used.
    It can be given (and/or overwritten) via command line argument `--stream`.

    `stream_timeout` (optional): how long to wait (in seconds) before stopping
    a streaming monitor if no results are received on the stream.

    `key` (optional): RIPE Atlas key to access the measurement. It can be
    given (and/or overwritten) via command line argument `--key`.

    `key_file` (optional): a file containing the RIPE Atlas key to access the
    measurement. The file must contain only the RIPE Atlas key, in plain text.
    If `key` is given, this field is ignored.
    """

    MANDATORY_CFG_FIELDS = ["matching_rules"]

    OPTIONAL_CFG_FIELDS = ["measurement-id", "stream", "stream_timeout",
                           "actions", "key", "key_file", "descr",
                           "expected_results"]

    def _get_statusfile_path(self):
        if self.monitor_name:
            return "{}/status/{}.{}.json".format(
                Config.get("var_dir"),
                self.monitor_name,
                self.msm_id
            )
        return None

    def _get_lockfile_path(self):
        if self.monitor_name:
            return "{}/locks/{}.{}.lock".format(
                Config.get("var_dir"),
                self.monitor_name,
                self.msm_id
            )
        return None

    def _load_from_file(self, monitor_name):
        file_path = "{}/monitors/{}.yaml".format(
            Config.get("var_dir"), monitor_name
        )

        if not os.path.isfile(file_path):
            raise MissingFileError(path=file_path)

        try:
            with open(file_path, "r") as monitor_file:
                try:
                    return yaml.load(monitor_file.read())
                except yaml.parser.ParserError as e:
                    raise ConfigError(
                        "Error in YAML syntax: {}".format(str(e))
                    )
        except Exception as e:
            raise ConfigError(
                "Can't read from the monitor configuration file {}: {}".format(
                    file_path, str(e)
                )
            )

    def __init__(self, cfg_or_name, ip_cache=None, msm_id=None, key=None):

        if isinstance(cfg_or_name, str):
            self.monitor_name = cfg_or_name
            cfg = self._load_from_file(self.monitor_name)

        elif isinstance(cfg_or_name, dict):
            self.monitor_name = None
            cfg = cfg_or_name

        else:
            raise ConfigError(
                "Invalid monitor name or configuration type: {}".format(
                    type(cfg_or_name)
                )
            )

        BasicConfigElement.__init__(self, cfg)

        self.normalize_fields()

        self.descr = self._enforce_param("descr", str)

        self._enforce_param("matching_rules", list)

        self._enforce_param("expected_results", dict) or {}

        self._enforce_param("actions", dict)

        MsmProcessingUnit.__init__(
            self,
            ip_cache=ip_cache,
            msm_id=msm_id or self._enforce_param("measurement-id", int),
            key=key or self._enforce_param("key", str),
            key_file=self._enforce_param("key_file", str)
        )

        self.stream = self._enforce_param("stream", bool) or False

        self.stream_timeout = self._enforce_param("stream_timeout", int)

        if self.stream:
            self.ensure_streaming_enabled(ConfigError)

        # Expected results normalization

        self.expected_results = {}

        if self.cfg["expected_results"]:
            for expres_name in self.cfg["expected_results"]:
                expres_cfg = self.cfg["expected_results"][expres_name]
                try:
                    expres = ExpectedResult(self, expres_name, expres_cfg)
                    self.expected_results[expres_name] = expres
                except ConfigError as e:
                    raise ConfigError(
                        "Syntax error for expected result {} - {}".format(
                            expres_name, str(e)
                        )
                    )

        # Actions normalization

        self.actions = {}

        if self.cfg["actions"]:
            for action_name in self.cfg["actions"]:
                action_cfg = self.cfg["actions"][action_name]
                try:
                    if "kind" in action_cfg:
                        action_kind = action_cfg["kind"]
                        action = None

                        for action_class in ACTION_CLASSES:
                            if action_class.CFG_ACTION_KIND == action_kind:
                                action = action_class(self, action_name,
                                                      action_cfg)
                                break

                        if action is None:
                            raise ConfigError(
                                "Unknown action kind: {}".format(
                                    action_kind
                                )
                            )

                        self.actions[action_name] = action
                    else:
                        raise ConfigError("Missing action kind")
                except ConfigError as e:
                    raise ConfigError(
                        "Syntax error for action {} - {}".format(
                            action_name, str(e)
                        )
                    )

        # Rules normalization

        self.rules = []

        rule_n = 0
        for rule_cfg in self.cfg["matching_rules"]:
            rule_n += 1

            try:
                rule = Rule(self, rule_cfg)
                self.rules.append(rule)

                if rule.expected_results and rule.expected_results != []:
                    for expres_name in rule.expected_results:
                        if expres_name not in self.expected_results:
                            raise ConfigError(
                                "Expected result not found: "
                                "{}".format(expres_name)
                            )

                if rule.actions and rule.actions != []:
                    for action_name in rule.actions:
                        if action_name not in self.actions:
                            raise ConfigError(
                                "Action not found: "
                                "{}".format(action_name)
                            )

            except ConfigError as e:
                raise ConfigError(
                    "Syntax error for rule n. {} - {}".format(
                        rule_n, str(e)
                    )
                )

        self.internal_labels = {
            "probes": {},
            "results": {}
        }

        self.exit_thread = False
        self.results_queue = None

        self.lock_fd = None
        self.lock_file = LockFile()

        self.status = {}
        self.load_status()
        self._epoch = datetime.datetime.fromtimestamp(0, tz=pytz.UTC)

    def load_status(self):
        status_filepath = self._get_statusfile_path()
        if status_filepath:
            if os.path.isfile(status_filepath):
                try:
                    with open(status_filepath, "r") as f:
                        self.status = json.loads(f.read())
                except:
                    raise ProgramError(
                        "Can't read status from {}".format(status_filepath)
                    )

    def write_status(self):
        status_filepath = self._get_statusfile_path()
        if status_filepath:
            try:
                with open("{}.tmp".format(status_filepath), "w") as f:
                    f.write(json.dumps(self.status))
                os.rename("{}.tmp".format(status_filepath), status_filepath)
            except:
                raise ProgramError(
                    "Can't write status to {}".format(status_filepath)
                )

    def __str__(self):
        if self.descr:
            return "monitor '{}'".format(self.descr)
        else:
            if self.monitor_name:
                tpl = ("monitor '{name}' "
                       "(measurement ID {msm_id}, {msm_type}, IPv{af})")
            else:
                tpl = ("monitor for measurement ID {msm_id} "
                       "({msm_type}, IPv{af})")

            return tpl.format(
                name=self.monitor_name,
                msm_id=self.msm_id,
                msm_type=self.msm_type,
                af=self.msm_af
            )

    def display(self):
        if self.msm_id:
            print("Measurement ID: {}".format(self.msm_id))
        else:
            print(
                "Measurement ID not specified, it must be given using the "
                "command line argument --measurement-id."
            )
        print("")

        if self.stream:
            if self.stream_timeout:
                print("Streaming of results enabled - "
                      "{} seconds timeout.".format(self.stream_timeout))
            else:
                print("Streaming of results enabled - no timeout.")
            print("")

        rule_n = 0
        for rule in self.rules:
            rule_n += 1

            print("Matching rule n. {}".format(rule_n))
            print("")

            rule.display()

            if len(rule.expected_results) > 0:
                print("  Expected results:")
                print("")
                for expres_name in rule.expected_results:
                    self.expected_results[expres_name].display()
            else:
                print("  No expected results for this rule.")
                print("")

            if len(rule.actions) > 0:
                print("  Actions:")
                print("")
                for action_name in rule.actions:
                    if self.actions[action_name].when == "on_match":
                        tpl = "    Action fired on match: {}"
                    elif self.actions[action_name].when == "on_mismatch":
                        tpl = "    Action fired on mismatch: {}"
                    else:
                        tpl = "    Action always fired: {}"
                    print(tpl.format(self.actions[action_name]))
                    print("")

    def process_matching_probe(self, probe, rule_n, rule, result):
        logger.info("  probe ID {} matches".format(probe.id))

        if len(rule.expected_results) == 0:
            logger.result(
                "Rule n. {} ({}), {}, "
                "no expected result.".format(
                    rule_n, str(rule), probe
                )
            )
            rule.perform_actions(result=result)
            return

        for expres_name in rule.expected_results:
            expres = self.expected_results[expres_name]

            logger.info(
                "Verifying expected result {}: {}...".format(
                    expres_name, str(expres)
                )
            )

            result_matches = expres.result_matches(result)

            logger.result(
                "Rule n. {} ({}), {}, "
                "expected result {}: {}.".format(
                    rule_n, str(rule), probe, expres_name,
                    "OK" if result_matches else "MISMATCH"
                )
            )

            rule.perform_actions(result=result, expres=expres,
                                 result_matches=result_matches)

    def update_latest_result_ts(self, result):
        ts = (result.created - self._epoch).total_seconds()

        if "latest_result_ts" in self.status:
            if self.status["latest_result_ts"] > ts:
                return
        self.status["latest_result_ts"] = ts
        self.write_status()

    def process_results(self, results, probes_filter):
        logger.info("Processing results...")

        # Be sure to have info for every probe in the resultset
        self.update_probes(results)

        if len(results) == 0:
            logger.info("No results found.")
            return

        # Processing results...

        for json_result in results:
            result = Result.get(json_result, on_error=Result.ACTION_IGNORE,
                                on_malformation=Result.ACTION_IGNORE)

            probe = self.get_probe(result)
            if probe not in probes_filter:
                logger.debug(
                    "  skipping {} because of probes filter".format(probe)
                )
                continue

            self.process_result(result)
            self.update_latest_result_ts(result)
            self.internal_labels["results"] = {}

        self.ip_cache.save()

    def process_result(self, result):
        self.internal_labels["results"] = {}

        is_success = not result.is_error and not result.is_malformed

        if not is_success:
            return

        probe = self.get_probe(result)

        logger.info(
            "Processing result for {} at {}...".format(
                str(probe), result.created
            )
        )

        rule_n = 0
        for rule in self.rules:
            rule_n += 1

            logger.info(
                "Testing rule n. {}: {}...".format(rule_n, str(rule))
            )

            if rule.probe_matches(probe):
                self.process_matching_probe(probe, rule_n, rule, result)

                if rule.process_next is True:
                    logger.info("  next rule processing forcedly enabled")
                else:
                    break
            else:
                logger.info(
                    "  {} does not match".format(probe)
                )

                if rule.process_next is False:
                    logger.info("  next rule processing forcedly inhibited")
                    break

    def on_result_response(self, result):
        self.results_queue.put(result)

    def acquire_lock(self):
        if not self.monitor_name:
            return

        if not self.lock_file.acquire(self._get_lockfile_path()):
            raise LockError(
                "Another instance of this program is already "
                "executing this monitor."
            )

    def release_lock(self):
        self.lock_file.release()

    def ensure_streaming_enabled(self, exception_class):
        err = "Can't use results streaming for this measurement: {}"

        if self.msm_is_oneoff:
            raise exception_class(
                err.format("it's a one-off measurement.")
            )

        if not self.msm_is_running:
            raise exception_class(
                err.format("it is not running anymore.")
            )

        if not self.msm_is_public:
            raise exception_class(
                err.format("it is not public.")
            )

    def run_stream(self, probes_filter):
        logger.info(" - using real-time results streaming")

        self.ensure_streaming_enabled(MeasurementProcessingError)

        try:
            atlas_stream = AtlasStream()
            atlas_stream.connect()
            atlas_stream.bind_channel("result", self.on_result_response)
            stream_params = {"msm": self.msm_id}
        except Exception as e:
            raise MeasurementProcessingError(
                "Error while creating the stream: {}".format(str(e))
            )

        self.results_queue = Queue()
        thread = MonitorResultsThread(self, probes_filter)

        try:
            thread.start()

            atlas_stream.start_stream(stream_type="result",
                                      **stream_params)
            atlas_stream.timeout(seconds=self.stream_timeout)
            atlas_stream.disconnect()
        except:
            try:
                atlas_stream.disconnect()
            except:
                pass
        finally:
            try:
                atlas_stream.disconnect()
            except:
                pass
            self.exit_thread = True
            thread.join(timeout=10)

    def get_latest_result_ts(self):
        if "latest_result_ts" in self.status:
            return self.status["latest_result_ts"]
        return None

    def run_once(self, probes_filter, start=None, stop=None,
                 latest_results=None):
        fetch_only_probe_ids = probes_filter.probe_ids

        results = self.download(start=start, stop=stop,
                                latest_results=latest_results,
                                probe_ids=fetch_only_probe_ids)

        self.process_results(results, probes_filter)

    def run_continously(self, start, probes_filter):
        try:
            while True:
                self.run_once(probes_filter, start=start)

                logger.info(
                    "Waiting {} seconds (measurement's interval) before "
                    "downloading new results...".format(self.msm_interval)
                )

                time.sleep(self.msm_interval)
        except KeyboardInterrupt:
            pass

    def run(self, start=None, stop=None, latest_results=None, dont_wait=False,
            probes_filter=None):
        self.acquire_lock()

        if not probes_filter:
            probes_filter = ProbesFilter()

        logger.info("Starting {}".format(str(self)))

        try:
            if self.stream:
                self.run_stream(probes_filter)

            elif not self.msm_is_oneoff and self.msm_is_running and \
                    not latest_results and not stop and not dont_wait:

                self.run_continously(start, probes_filter)

            else:
                self.run_once(probes_filter, start=start, stop=stop,
                              latest_results=latest_results)
        finally:
            self.release_lock() class RunMonitorProcess(Process):
    def __init__(self, monitor_name, ip_cache):
        Process.__init__(self, name=monitor_name)
        self.monitor_name = monitor_name
        self.ip_cache = ip_cache
        self.name = self.monitor_name

        self.monitor = None

    def run(self):
        logger.info(
            "Starting process {} (PID {})".format(
                self.name, self.pid
            )
        )
        try:
            self.monitor = Monitor(self.monitor_name, self.ip_cache)
            self.monitor.run()
        except KeyboardInterrupt:
            pass
        except RIPEAtlasMonitorError as e:
            logger.error(e)
        except Exception as e:
            logger.error(e, exc_info=True)
        logger.info("Process {} completed.".format(str(self.monitor_name))) class ArgumentError(RIPEAtlasMonitorError):
    pass class LockError(RIPEAtlasMonitorError):
    pass
from ..Config import Config
from ..Helpers import IPCache, IPCacheManager, LockFile, ProbesFilter
from ..Logging import logger
from ..Monitor import Monitor
from ..Processes import RunMonitorProcess
from ..Errors import ArgumentError, LockError
# Copyright (C) 2016 Pier Carlo Chiodi
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.



def run_one(args):
    ip_cache = IPCache()
    ip_cache.setup(
        _dir=Config.get("ip_cache.dir"),
        lifetime=Config.get("ip_cache.lifetime"),
        use_ixps_info=Config.get("ip_cache.use_ixps_info")
    )


    monitor = Monitor(