Source code for PyFunceble.cli.file_preloader

The tool to check the availability or syntax of domain, IP or URL.


    ██████╗ ██╗   ██╗███████╗██╗   ██╗███╗   ██╗ ██████╗███████╗██████╗ ██╗     ███████╗
    ██╔══██╗╚██╗ ██╔╝██╔════╝██║   ██║████╗  ██║██╔════╝██╔════╝██╔══██╗██║     ██╔════╝
    ██████╔╝ ╚████╔╝ █████╗  ██║   ██║██╔██╗ ██║██║     █████╗  ██████╔╝██║     █████╗
    ██╔═══╝   ╚██╔╝  ██╔══╝  ██║   ██║██║╚██╗██║██║     ██╔══╝  ██╔══██╗██║     ██╔══╝
    ██║        ██║   ██║     ╚██████╔╝██║ ╚████║╚██████╗███████╗██████╔╝███████╗███████╗
    ╚═╝        ╚═╝   ╚═╝      ╚═════╝ ╚═╝  ╚═══╝ ╚═════╝╚══════╝╚═════╝ ╚══════╝╚══════╝

Provides the interface for the preloading of a given file.

    Nissar Chababy, @funilrys, contactTATAfunilrysTODTODcom

Special thanks:


Project link:

Project documentation:

Project homepage:


    Copyright 2017, 2018, 2019, 2020, 2022, 2023 Nissar Chababy

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.

import copy
import functools
import os
from datetime import datetime, timedelta
from typing import Any, Optional

from domain2idna import domain2idna

import PyFunceble.facility
from PyFunceble.cli.continuous_integration.base import ContinuousIntegrationBase
from PyFunceble.cli.processes.workers.tester import TesterWorker
from PyFunceble.cli.utils.stdout import print_single_line
from PyFunceble.cli.utils.testing import get_subjects_from_line
from PyFunceble.converter.adblock_input_line2subject import AdblockInputLine2Subject
from PyFunceble.converter.cidr2subject import CIDR2Subject
from PyFunceble.converter.input_line2subject import InputLine2Subject
from PyFunceble.converter.rpz_input_line2subject import RPZInputLine2Subject
from PyFunceble.converter.rpz_policy2subject import RPZPolicy2Subject
from PyFunceble.converter.subject2complements import Subject2Complements
from PyFunceble.converter.url2netloc import Url2Netloc
from PyFunceble.converter.wildcard2subject import Wildcard2Subject
from PyFunceble.dataset.autocontinue.base import ContinueDatasetBase
from PyFunceble.dataset.autocontinue.csv import CSVContinueDataset
from PyFunceble.dataset.inactive.base import InactiveDatasetBase
from PyFunceble.helpers.dict import DictHelper
from PyFunceble.helpers.file import FileHelper
from PyFunceble.helpers.hash import HashHelper

[docs]class FilePreloader: """ Provides an interface for the file preloading. The main idea behind this interface is to read all lines of the given file parse each lines into something our autocontinue dataset understand. Once everything preloaded in the autocontinue dataset, one can use the :meth:`PyFunceble.dataset.autocontinue.base.ContinueDataset.get_to_test` to get the next subject to test. By doing this, we don't have to re-read a file completely once we are sure that the hash of the file didn't changed. :param authorized: The authorization to launch. If :py:class:`None` is given, we will try to guess the best value. :param protocol: The protocol describing the file to test. """ STD_AUTHORIZED: bool = False _authorized: Optional[bool] = False _protocol: Optional[dict] = None __description: list = [] __description_file: Optional[str] = None __matching_index: int = 0 continuous_integration: Optional[ContinuousIntegrationBase] = None continue_dataset: Optional[ContinueDatasetBase] = None checker_type: Optional[str] = None subject2complements: Optional[Subject2Complements] = None inputline2subject: Optional[InputLine2Subject] = None adblock_inputline2subject: Optional[AdblockInputLine2Subject] = None wildcard2subject: Optional[Wildcard2Subject] = None rpz_policy2subject: Optional[RPZPolicy2Subject] = None rpz_inputline2subject: Optional[RPZInputLine2Subject] = None url2netloc: Optional[Url2Netloc] = None cidr2subject: Optional[CIDR2Subject] = None def __init__( self, *, authorized: Optional[bool] = None, protocol: Optional[dict] = None, continuous_integration: Optional[ContinuousIntegrationBase] = None, checker_type: Optional[str] = None, adblock_inputline2subject: Optional[AdblockInputLine2Subject] = None, wildcard2subject: Optional[Wildcard2Subject] = None, rpz_policy2subject: Optional[RPZPolicy2Subject] = None, rpz_inputline2subject: Optional[RPZInputLine2Subject] = None, inputline2subject: Optional[InputLine2Subject] = None, subject2complements: Optional[Subject2Complements] = None, url2netloc: Optional[Url2Netloc] = None, continue_dataset: Optional[ContinueDatasetBase] = None, inactive_dataset: Optional[InactiveDatasetBase] = None, cidr2subject: Optional[CIDR2Subject] = None, ) -> None: if authorized is not None: self.authorized = authorized else: self.guess_and_set_authorized() if protocol is not None: self.file_path = protocol self.continuous_integration = continuous_integration self.checker_type = checker_type self.adblock_inputline2subject = adblock_inputline2subject self.wildcard2subject = wildcard2subject self.rpz_policy2subject = rpz_policy2subject self.rpz_inputline2subject = rpz_inputline2subject self.inputline2subject = inputline2subject self.subject2complements = subject2complements self.url2netloc = url2netloc self.continue_dataset = continue_dataset self.inactive_dataset = inactive_dataset self.cidr2subject = cidr2subject
[docs] def execute_if_authorized(default: Any = None): # pylint: disable=no-self-argument """ Executes the decorated method only if we are authorized to process. Otherwise, apply the given :code:`default`. .. warning:: If :py:class:`None` is given as default value, this method will return the :code:`self` object. """ def inner_method(func): @functools.wraps(func) def wrapper(self, *args, **kwargs): if self.authorized: return func(self, *args, **kwargs) # pylint: disable=not-callable return self if default is None else default return wrapper return inner_method
[docs] def ensure_protocol_is_given(func): # pylint: disable=no-self-argument """ Ensures that the protocol is given before launching the decorated method. :raise RuntimeError: When the protocol is not declared yet. """ @functools.wraps(func) def wrapper(self, *args, **kwargs): if self.protocol is None: raise RuntimeError("<self.protocol> is not given.") return func(self, *args, **kwargs) # pylint: disable=not-callable return wrapper
@property def authorized(self) -> Optional[bool]: """ Provides the current state of the :code:`_authorized` attribute. """ return self._authorized @authorized.setter def authorized(self, value: bool) -> None: """ Sets the value of the :code:`_authorized` attribute. :param value: The value to set. :raise TypeError: When the given :code:`value` is not a :py:class:`bool`. """ if not isinstance(value, bool): raise TypeError(f"<value> should be {bool}, {type(value)} given.") self._authorized = value
[docs] def set_authorized(self, value: bool) -> "FilePreloader": """ Sets the value of the :code:`_authorized` attribute. :param value: The value to set. """ self.authorized = value return self
@property def protocol(self) -> Optional[dict]: """ Provides the current state of the :code:`_file_path` attribute. """ return self._protocol @protocol.setter def protocol(self, value: dict) -> None: """ Sets the value of the :code:`_protocol` attribute. :param value: The value to set. :raise TypeError: When the given :code:`value` is not a :py:class:`dict`. :raise ValueError: When the given :code:`value` is empty. """ if not isinstance(value, dict): raise TypeError(f"<value> should be {str}, {type(value)} given.") self._protocol = value self.__description_file = os.path.join( self._protocol["output_dir"], )
[docs] def set_protocol(self, value: dict) -> "FilePreloader": """ Sets the value of the :code:`_protocol` attribute. :param value: The value to set. """ self.protocol = value return self
[docs] def guess_and_set_authorized(self) -> "FilePreloader": """ Try to guess and set the value of the :code:`_authorized` attribute. """ if PyFunceble.facility.ConfigLoader.is_already_loaded(): if bool( and bool( ): self.authorized = True else: self.authorized = self.STD_AUTHORIZED else: self.authorized = self.STD_AUTHORIZED return self
[docs] @execute_if_authorized(False) @ensure_protocol_is_given def does_preloader_description_file_exists(self) -> bool: """ Checks if our preloader file exists. """ return FileHelper(self.__description_file).exists()
def __load_description(self) -> "FilePreloader": """ Loads the descriptoin into the interface. """ def set_new_dataset(*, append: bool = False, new_index: int = 0) -> None: """ Sets the default dataset into the given index. :param new_index: The index to write into. :param append: Append instead. """ new_dataset = copy.deepcopy(self.protocol) new_dataset["previous_hash"] = None new_dataset["hash"] = None new_dataset["line_number"] = 1 if append: self.__description.append(new_dataset) elif self.__description: self.__description[new_index] = new_dataset else: self.__description = [new_dataset] if self.does_preloader_description_file_exists(): dataset = DictHelper().from_json_file(self.__description_file) if not isinstance(dataset, list): dataset = [dataset] found = False for index, descr in enumerate(dataset): if all(x in descr and descr[x] == y for x, y in self.protocol.items()): self.__matching_index = index found = True break self.__description = dataset if not found: set_new_dataset(append=True) self.__matching_index = len(self.__description) - 1 else: set_new_dataset() return self def __save_description(self) -> "FilePreloader": """ Saves the description at its destination. """ self.__description[self.__matching_index].update(self.protocol) DictHelper(self.__description).to_json_file(self.__description_file)
[docs] @execute_if_authorized(None) @ensure_protocol_is_given def start(self, print_dots: bool = False) -> "FilePreloader": """ Starts the pre-loading of the currently set file path. """ self.__load_description() broken = False file_helper = FileHelper(self.protocol["subject"]) self.__description[self.__matching_index]["hash"] = HashHelper().hash_file( file_helper.path ) if isinstance(self.continue_dataset, CSVContinueDataset): self.continue_dataset.set_base_directory(self.protocol["output_dir"]) if ( self.__description[self.__matching_index]["checker_type"] != self.protocol["checker_type"] or self.__description[self.__matching_index]["subject_type"] != self.protocol["subject_type"] ): try: self.continue_dataset.cleanup() except TypeError: self.continue_dataset.cleanup(session_id=self.protocol["session_id"]) if ( self.__description[self.__matching_index]["previous_hash"] and self.__description[self.__matching_index]["hash"] != self.__description[self.__matching_index]["previous_hash"] ): # Forces the reading of each lines because there is literally no # way to know where something has been changed. self.__description[self.__matching_index]["line_number"] = 1 if ( self.__description[self.__matching_index]["checker_type"] != self.protocol["checker_type"] or self.__description[self.__matching_index]["subject_type"] != self.protocol["subject_type"] or self.__description[self.__matching_index]["hash"] != self.__description[self.__matching_index]["previous_hash"] ): try: with"r", encoding="utf-8") as file_stream: line_num = 1 for line in file_stream: if ( line_num < self.__description[self.__matching_index]["line_number"] ): line_num += 1 continue if ( self.continuous_integration and self.continuous_integration.is_time_exceeded() ): broken = True break line = line.strip() if self.rpz_policy2subject and "SOA" in line: self.rpz_policy2subject.set_soa(line.split()[0]) for subject in get_subjects_from_line( line, self.checker_type, subject_type=self.protocol["subject_type"], adblock_inputline2subject=self.adblock_inputline2subject, wildcard2subject=self.wildcard2subject, rpz_policy2subject=self.rpz_policy2subject, rpz_inputline2subject=self.rpz_inputline2subject, inputline2subject=self.inputline2subject, subject2complements=self.subject2complements, url2netloc=self.url2netloc, cidr2subject=self.cidr2subject, ): to_send = copy.deepcopy(self.protocol) to_send["subject"] = subject to_send["idna_subject"] = domain2idna(subject) to_send["tested_at"] = datetime.utcnow() - timedelta( days=365.25 * 20 ) if self.inactive_dataset.exists(to_send): print_single_line("I") continue if TesterWorker.should_be_ignored( subject=to_send["idna_subject"] ): print_single_line("X") continue self.continue_dataset.update(to_send, ignore_if_exist=True) if print_dots: print_single_line() self.__description[self.__matching_index]["line_number"] += 1 line_num += 1 except KeyboardInterrupt as exception: self.__save_description() raise exception if not broken: self.__description[self.__matching_index][ "previous_hash" ] = self.__description[self.__matching_index]["hash"] self.__save_description() return self