WhakerPy 2.4

https://github.com/brigitte-bigi/WhakerPy/

Module whakerpy.httpd

Class BanList

Description

Manager of a temporary list of client addresses to be rejected.

An address is recorded when a request met a forbidden URL path, and it is rejected during the duration given by the configuration.

The list is stored in a file, so that the processes of a server share it and it survives a reload. A line of the file is an address and the time it was recorded:

192.0.2.4 1761220000.0

An address matches when it is written in the file and its line is not older than the duration. The comparison is exact: an address is never a part of another one.

Constructor

__init__

Create a ban list with no file and no duration.

View Source
def __init__(self):
    """Create a ban list with no file and no duration.

    """
    self.__filename = ''
    self.__duration = 0

Public functions

configure

Configure the ban list.

The section of the given key is a dict with two required values: "file" is the path of the file where the addresses are stored, and "duration" is the number of seconds an address is rejected.

If the key is missing, the ban list is disabled.

Parameters
  • config: (dict) Configuration dictionary.
  • json_key: (str) Key of the ban list dict in dict.
Raises
  • TypeError: the section is not a dict.
  • ValueError: a required value is missing or invalid.
Returns
  • (dict) Loaded settings: {"file": str, "duration": int}.
View Source
def configure(self, config: dict, json_key: str='banlist') -> dict:
    """Configure the ban list.

        The section of the given key is a dict with two required values:
        "file" is the path of the file where the addresses are stored, and
        "duration" is the number of seconds an address is rejected.

        If the key is missing, the ban list is disabled.

        :param config: (dict) Configuration dictionary.
        :param json_key: (str) Key of the ban list dict in dict.
        :raises: TypeError: the section is not a dict.
        :raises: ValueError: a required value is missing or invalid.
        :return: (dict) Loaded settings: {"file": str, "duration": int}.

        """
    _cfg = config.get(json_key, None)
    if _cfg is None:
        self.__filename = ''
        self.__duration = 0
        return {'file': '', 'duration': 0}
    if isinstance(_cfg, dict) is False:
        raise TypeError(f"JSON key '{json_key}' must be a dict.")
    _filename = _cfg.get('file', '')
    if type(_filename) is not str or len(_filename) == 0:
        raise ValueError(f'{json_key}.file must be a non-empty string.')
    _duration = _cfg.get('duration', 0)
    if isinstance(_duration, int) is False or _duration <= 0:
        raise ValueError(f'{json_key}.duration must be a number of seconds.')
    self.__filename = _filename
    self.__duration = _duration
    return {'file': _filename, 'duration': _duration}

match

Return True if the given address is banned.

Parameters
  • address: (str) The address of a client.
Raises
  • TypeError: address is not a string.
Returns
  • (bool) True if the address was recorded and did not expire.
View Source
def match(self, address: str) -> bool:
    """Return True if the given address is banned.

        :param address: (str) The address of a client.
        :raises: TypeError: address is not a string.
        :return: (bool) True if the address was recorded and did not expire.

        """
    if type(address) is not str:
        raise TypeError('Address must be a string.')
    if len(self.__filename) == 0 or len(address) == 0:
        return False
    return address in self.__read()

add

Record the given address for the configured duration.

The addresses that expired are dropped when the file is written.

Parameters
  • address: (str) The address of a client.
Raises
  • TypeError: address is not a string.
View Source
def add(self, address: str) -> None:
    """Record the given address for the configured duration.

        The addresses that expired are dropped when the file is written.

        :param address: (str) The address of a client.
        :raises: TypeError: address is not a string.

        """
    if type(address) is not str:
        raise TypeError('Address must be a string.')
    if len(self.__filename) == 0 or len(address) == 0:
        return
    entries = self.__read()
    entries[address] = time.time()
    self.__write(entries)
    logging.info("Banned address '" + address + "' for " + str(self.__duration) + ' seconds')

Protected functions

__read

Return the addresses of the file that did not expire.

Returns
  • (dict) key=address, value=time it was recorded.
View Source
def __read(self) -> dict:
    """Return the addresses of the file that did not expire.

        :return: (dict) key=address, value=time it was recorded.

        """
    if os.path.exists(self.__filename) is False:
        return dict()
    oldest = time.time() - self.__duration
    entries = dict()
    try:
        with open(self.__filename, 'r', encoding='utf-8') as fp:
            for line in fp:
                items = line.split()
                if len(items) != 2:
                    continue
                try:
                    recorded = float(items[1])
                except ValueError:
                    continue
                if recorded > oldest:
                    entries[items[0]] = recorded
    except OSError as e:
        logging.error("Ban list file can't be read: " + str(e))
    return entries

__write

Write the given addresses in the file.

The file is written aside and moved into place, so that a reader never gets an incomplete content.

Parameters
  • entries: (dict) key=address, value=time it was recorded.
View Source
def __write(self, entries: dict) -> None:
    """Write the given addresses in the file.

        The file is written aside and moved into place, so that a reader
        never gets an incomplete content.

        :param entries: (dict) key=address, value=time it was recorded.

        """
    folder = os.path.dirname(self.__filename)
    if len(folder) == 0:
        folder = '.'
    temporary = ''
    try:
        fd, temporary = tempfile.mkstemp(dir=folder)
        with os.fdopen(fd, 'w', encoding='utf-8') as fp:
            for address in entries:
                fp.write(address + ' ' + str(entries[address]) + '\n')
        os.replace(temporary, self.__filename)
    except OSError as e:
        logging.error("Ban list file can't be written: " + str(e))
        if len(temporary) > 0 and os.path.exists(temporary) is True:
            os.remove(temporary)