.

. BSA HACKER
Server : Apache
System : Linux nusantara.hosteko.com 4.18.0-553.16.1.lve.el8.x86_64 #1 SMP Tue Aug 13 17:45:03 UTC 2024 x86_64
User : koperas1 ( 1254)
PHP Version : 7.4.33
Disable Function : NONE
Directory : /opt/cloudlinux/venv/lib64/python3.11/site-packages/xray/continuous/
Upload File :
Current File : //opt/cloudlinux/venv/lib64/python3.11/site-packages/xray/continuous/tracing.py
# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

"""
This module contains X Ray continuous tracing implementation
"""
import re
from collections import defaultdict
from typing import List

from xray import gettext as _
from .common import ContinuousCommon
from .mailer import Mailer
from ..apiclient import get_client
from ..console_utils.validations import regex
from ..internal.exceptions import XRayError, XRayAPIError
from ..internal.types import url_split
from ..internal.utils import read_sys_id, prev_date, date_of_timestamp, \
    get_formatted_date, get_html_formatted_links, get_text_formatted_links
from ..manager import initialize_manager


class ContinuousTracing(ContinuousCommon):
    """
    Continuous tracing implementation
    """

    def __init__(self):
        super().__init__()
        system_id = read_sys_id()
        client_object = get_client('auto_tasks')
        self.client = client_object(system_id=system_id)
        self.manager_instance = initialize_manager(system_id)
        self.mail_sender = Mailer()

    def __call__(self):
        self.clear_zombie_auto_tasks()
        self.start_tracing()
        self.generate_mail()

    def current_auto_tasks(self) -> list:
        """
        Get auto tasks list
        """
        try:
            tasks = self.client.get_tasks()
        except XRayAPIError as e:
            self.logger.error('Failed to get auto tasks',
                              extra={'err': str(e)})
            tasks = list()
        return tasks

    def clear_zombie_auto_tasks(self) -> None:
        """
        Stop and complete running auto tasks
        """
        auto_running_tasks = {t.task_id: t.status for t in
                              self.current_auto_tasks() if
                              t.auto_task and t.status != 'completed'
                              and t.user != '*autotracing*'}
        for t_id in auto_running_tasks:
            try:
                if auto_running_tasks[t_id] == 'running':
                    self.manager_instance.stop(t_id)
                self.manager_instance.complete(t_id)
            except XRayError as e:
                self.logger.error('%s', str(e))

    @staticmethod
    def _url(domain: str) -> str:
        """
        Construct URL for auto tracing: {domain}/*
        """
        u = f"{domain.strip('/')}/*"
        if not re.match(regex, u):
            raise ValueError(_('invalid url'), u)
        return u

    def executed(self, domain: str) -> None:
        """
        Increment execution_count of given domain
        """
        self.tracing_conf[domain]['execution_count'] += 1

    def start_tracing(self) -> List[dict]:
        """
        Start auto tasks with status 'running'
        """
        err_text = 'Start auto task failed'
        for continuous_task in self.get_continuous_tasks():
            if continuous_task.status == 'running':
                try:
                    self.manager_instance.start_auto(
                        url=self._url(continuous_task.original_url))
                except ValueError as e:
                    self.logger.error(err_text,
                                      extra={'err': str(e)})
                except XRayError as e:
                    self.logger.error(err_text,
                                      extra={
                                          'err': str(e),
                                          'continuous_task': continuous_task
                                      })
                else:
                    self.executed(continuous_task.domain)
        return self.dump_tracing_configuration()

    def retrieve_mail_addrs(self) -> dict:
        """
        Map e-mail addresses to urls in existing continuous tracing configuration
        """
        url_mail_map = dict()
        for t in self.get_continuous_tasks():
            try:
                url_mail_map[self._url(t.original_url)] = t.email
            except ValueError:
                pass
        return url_mail_map

    def filter_tasks_by_date(self) -> list:
        """
        Select previous date completed tasks
        """

        def check_date(ts: int) -> bool:
            """
            Verify if task timestamp corresponds to previous date
            """
            try:
                return date_of_timestamp(ts) == prev_date()
            except TypeError:
                # in case of starttime == None
                return False

        return [t for t in
                self.current_auto_tasks() if t.status == 'completed' and
                check_date(t.starttime)]

    def generate_mail(self) -> None:
        """
        Generate e-mails with links and send them to recipients
        """
        self.logger.info('Generating e-mails started')
        email_conf = defaultdict(list)
        url_mail_mapping = self.retrieve_mail_addrs()
        # filter tasks by date
        yesterday_tasks = self.filter_tasks_by_date()
        # generate links along with mapping them to e-mail addresses
        for task in yesterday_tasks:
            mail = url_mail_mapping.get(task.url)
            if mail:
                email_conf[mail].append(
                    {url_split(task.url)[0]: task.shared_link})
                self.logger.info('Selected task %s for mailing to %s',
                                 task.task_id, mail)

        # send e-mails
        for k, v in email_conf.items():
            self.mail_sender.send_mail(k, template='report',
                                       date=get_formatted_date(),
                                       links=get_text_formatted_links(v),
                                       html_links=get_html_formatted_links(
                                           v))
        self.logger.info('E-mails sending finished')