text: import json
  import re
  import time
  from datetime import datetime
  from hashlib import md5
  from logging import Logger
  from os.path import getmtime, getsize
  from tempfile import TemporaryFile
  from threading import Lock
  from typing import List
  
  
  class LogParser:
      """
      Class for parsing information about attacks from log files
      """
  
      def __init__(self, file_log, rules, service_name=None, logger=None):  # type: (str, List[str], str, Logger) -> None
          """
          Initialize the log parser
          :param file_log: path to the file with logs
          :param rules: list of string filters/rules
          :param service_name: optional name of the service. If not specified then found attacks are not assigned to any
          service
          :param logger: optional logger. If specified, LogParsers fires messages into him
          """
          self.file_log = file_log
          self.rules = [rule if type(rule) == Rule else Rule(rule, service_name) for rule in rules]
          self.logger = logger
  
          self._last_file_size = 0
          self._last_file_modification_date = None
  
          self._attack_cache_file = TemporaryFile()  # here will all attacks stay cached
          self._attack_cache_file_lock = Lock()
  
          # if this last bytes are same then we are sure the file was not modified
          self._last_bytes = {'hash': None, 'len': 0}
  
          self.force_rescan()
  
      def parse_attacks(self, max_age=None, skip_scanning=False):  # type: (float, bool) -> dict
          """
          Parses the attacks from log file and returns them
          :param max_age: optional, in seconds. If attack is older as this then it is ignored
          :param skip_scanning: if set to true then the read content is not analyzed for attacks
          :return: dictionary. Key is the IP that attacked and value is list of dictionaries with data about every attack
          """
          if self.logger is not None:
              self.logger.debug('parsing attacks for %s' % self.file_log)
  
          attacks = {}
  
          curr_file_size = getsize(self.file_log)
          curr_file_modification_time = getmtime(self.file_log)
  
          if self._last_file_size == curr_file_size and curr_file_modification_time == self._last_file_modification_date:
              if self.logger is not None:
                  self.logger.debug('nothing changed, nothing new to parse')
              # it seems that the files has not changed so skip analyzing it
              return attacks
  
          continue_in_scanning = True  # when set to True, only new content in file is analyzed
  
          if self._last_file_size > curr_file_size:
              # when the current file is smaller, something has happened to it. We will rescan it to be sure
              if self.logger is not None:
                  self.logger.debug('file went smaller since las scan, rescan it')
              continue_in_scanning = False
              self.force_rescan()
  
          if self._last_file_size == curr_file_size and self._last_file_modification_date != curr_file_modification_time:
              # the file has the same size but was still modified, we better rescan it
              if self.logger is not None:
                  self.logger.debug('file is the same size but it was modified,rescan it')
              continue_in_scanning = False
              self.force_rescan()
  
          with open(self.file_log, 'r') as f:
              if continue_in_scanning and self._last_bytes['hash'] is not None:
                  # check last few bytes if they are same
                  f.seek(self._last_file_size - self._last_bytes['len'] - 5)
                  if md5(f.read(self._last_bytes['len']).encode('utf8')).hexdigest() != self._last_bytes['hash']:
                      # nope, last few bytes differ, something seems really odd about this file. Better rescan it
                      if self.logger is not None:
                          self.logger.debug('last few scanned bytes differ, rescan it')
                      self.force_rescan()
  
              f.seek(self._last_file_size)  # skip all already analyzed content
              new_content = f.read()
  
          # save last ten bytes so we can know if the file is still the same during new analyze
          content_end = new_content[-256:-5]
          self._last_bytes['hash'] = md5(content_end.encode('utf8')).hexdigest()
          self._last_bytes['len'] = len(content_end)
  
          log_lines = new_content.splitlines()
          del new_content, content_end
  
          if not skip_scanning:
              for log_line in log_lines:
                  if self.logger is not None:
                      self.logger.debug('log line "%s"' % log_line.replace('\n', '\\n'))
                  for rule in self.rules:
                      variables = rule.get_variables(log_line)
                      if variables is not None:
                          if max_age is not None and time.time() - max_age > variables['TIMESTAMP']:
                              break
                          attacker_ip = variables['IP']
                          del variables['IP']
                          item = attacks.get(attacker_ip, [])
                          item.append(variables)
                          attacks[attacker_ip] = item
                          break
  
          self._last_file_modification_date = curr_file_modification_time
          self._last_file_size = curr_file_size
  
          self._attack_cache_file_lock.acquire()
          self._attack_cache_file.seek(0)
          attacks.update(json.loads(self._attack_cache_file.read().decode('utf8')))
          self._attack_cache_file.seek(0)
          self._attack_cache_file.truncate(0)
          self._attack_cache_file.write(json.dumps(attacks).encode('utf8'))
          self._attack_cache_file.flush()
          self._attack_cache_file_lock.release()
  
          return attacks
  
      def get_habitual_offenders(self, min_attack_attempts, attack_attempts_time, max_age=None, attacks=None,
                                 first_load=False):
          # type: (int, int, int, dict, bool) -> dict
          """
          Finds IPs that had performed more than allowed number of attacks in specified time range
          :param min_attack_attempts: minimum allowed number of attacks in time range to be included
          :param attack_attempts_time:  the time range in which all of the attacks must have occurred in seconds
          :param max_age: optional, in seconds. If attack is older as this then it is ignored
          :param attacks: optional. If None, then the value of self.parse_attacks(max_age) is used
          :param first_load: If true, then the log file is read only, not scanned for attacks
          :return: dictionary. Key is the IP that attacked more or equal than min_attack_attempts times and
          value is list of dictionaries with data about every attack in specified time range
          """
          attacks = self.parse_attacks(max_age, first_load) if attacks is None else attacks
          habitual_offenders = {}
  
          for ip, attack_list in attacks.items():
              for attack in attack_list:
                  attacks_in_time_range = []
                  for attack2 in attack_list:
                      attack_time_delta = attack2['TIMESTAMP'] - attack['TIMESTAMP']
                      if 0 <= attack_time_delta <= attack_attempts_time:
                          attacks_in_time_range.append(attack2)
                          if len(attacks_in_time_range) > min_attack_attempts:
                              break
                  if len(attacks_in_time_range) >= min_attack_attempts:
                      habitual_offenders[ip] = attack_list
  
          return habitual_offenders
  
      def force_rescan(self):  # type: () -> None
          """
          Resets progress info about log file, forcing program to perform the next scan from the beginning
          :return: None
          """
          self._last_file_size = 0
          self._last_file_modification_date = None
          self._attack_cache_file_lock.acquire()
          self._attack_cache_file.seek(0)
          self._attack_cache_file.truncate(0)
          self._attack_cache_file.write(json.dumps({}).encode('utf8'))
          self._attack_cache_file.flush()
          self._attack_cache_file_lock.release()
  
  
  class Rule:
      """
      Rule or filter that can be tested on a line from config line. If this rule/filter fits, than it can parse
      variables from that line
      """
  
      def __init__(self, filter_string, service_name=None):  # type: (str, str or None) -> None
          """
          Initializes this rule/filter
          :param filter_string: string representation of this rule/filter with all variables stated as %VAR_NAME%
          :param service_name: optional name of the service. If not specified then found attacks are not assigned to any
          service
          """
          self.__service_name = service_name
          self.__rule_variables = re.findall("%.*?%", filter_string)
  
          # Generate regex for rule detection
          self.__rule_regex = filter_string
          for reserved_char in list("\\+*?^$.[]{}()|/"):  # escape reserved regex characters
              self.__rule_regex = self.__rule_regex.replace(reserved_char, '\\' + reserved_char)
          for variable in self.__rule_variables:  # replace all variables with any regex characters
              self.__rule_regex = self.__rule_regex.replace(variable, '(.+?)')
          if self.__rule_regex.endswith('?)'):  # disable lazy search for last variables so they are found whole
              self.__rule_regex = self.__rule_regex[:-2] + ')'
  
          # Remove %'s from variable names
          self.__rule_variables = [var[1:-1] for var in self.__rule_variables]
  
      def test(self, log_line):  # type: (str) -> bool
          """
          Test this Rule against a line from log file if it fits
          :param log_line: line from a log file
          :return: True if it fits, False if this rule cannot be applied to this line
          """
          return True if re.match(self.__rule_regex, log_line) else False
  
      def get_variables(self, log_line):  # type: (str) -> dict or None
          """
          Parses variables from log line that fits this rule
          :param log_line: line from a log file
          :return: None if this rule cannot be applied to this line, otherwise returns a dictionary with parsed variables
          from this line
          """
          data = {}
  
          # Parse all variables from log line
          variable_search = re.match(self.__rule_regex, log_line)
          if not variable_search:  # this rule is not for this line
              return None
          # noinspection PyTypeChecker
          for i, variable in enumerate(self.__rule_variables):
              data[variable] = variable_search.group(i + 1).strip()
  
          # attempt to parse raw IP
          if 'IP' in data:
              ip = re.findall('[0-9]{3}\\.[0-9]{3}\\.[0-9]{3}\\.[0-9]{3}', data['IP'])
              if len(ip) == 0:
                  ip = re.findall('([0-9a-f]{1,4}|::)((:{0,2}[0-9a-f]{0,4}){0,7})', data['IP'])
              if len(ip) == 0:
                  data['IP-RAW'], data['IP'] = data['IP'], 'NO VALID IP FOUND'
  
          if self.__service_name is not None:
              data['SERVICE'] = self.__service_name
  
          date_format = '%Y %b %d %H:%M:%S'
          date_string = None
          if 'D:M' in data and 'D:D' in data and 'TIME' in data:
              date_string = '%s %s %s %s' % (datetime.now().strftime('%Y'), data['D:M'],
                                             data['D:D'], data['TIME'])
          elif 'D:M' in data and 'D:D' in data:
              date_string = '%s %s %s 00:00:00' % (datetime.now().strftime('%Y'), data['D:M'], data['D:D'])
          elif 'TIME' in data:
              # noinspection PyTypeChecker
              date_string = datetime.now().strftime('%Y %b %d') + ' ' + data['TIME']
  
          data['TIMESTAMP'] = time.time() if date_string is None else \
              time.mktime(datetime.strptime(date_string, date_format).timetuple())
          return data
  
  
  # If launched directly, perform a quick proof of work in file debug.log
  if __name__ == '__main__':
      all_rules = ["%D:M% %D:D% %TIME% %IP% attacked on user %USER%"]
      file = 'debug.log'
  
      parser = LogParser(file, all_rules)
      offenders = parser.get_habitual_offenders(3, 100000)
      for off_ip, off_attacks in offenders.items():
          print(off_ip + ':', off_attacks)
  