fba/models/error_log.py

   1 # Copyright (C) 2023 Free Software Foundation
   2 #
   3 # This program is free software: you can redistribute it and/or modify
   4 # it under the terms of the GNU Affero General Public License as published
   5 # by the Free Software Foundation, either version 3 of the License, or
   6 # (at your option) any later version.
   7 #
   8 # This program is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 # GNU Affero General Public License for more details.
  12 #
  13 # You should have received a copy of the GNU Affero General Public License
  14 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
  15
  16 import logging
  17 import json
  18 import time
  19
  20 import validators
  21
  22 from fba import fba
  23
  24 from fba.helpers import config
  25
  26 logging.basicConfig(level=logging.INFO)
  27 logger = logging.getLogger(__name__)
  28
  29 def add(domain: str, error: dict):
  30     logger.debug("domain,error[]:", domain, type(error))
  31     if not isinstance(domain, str):
  32         raise ValueError(f"Parameter domain[]='{type(domain)}' is not 'str'")
  33     elif domain == "":
  34         raise ValueError("Parameter 'domain' is empty")
  35     elif domain.lower() != domain:
  36         raise ValueError(f"Parameter domain='{domain}' must be all lower-case")
  37     elif not validators.domain(domain.split("/")[0]):
  38         raise ValueError(f"domain='{domain}' is not a valid domain")
  39     elif domain.endswith(".arpa"):
  40         raise ValueError(f"domain='{domain}' is a domain for reversed IP addresses, please don't crawl them!")
  41     elif domain.endswith(".tld"):
  42         raise ValueError(f"domain='{domain}' is a fake domain, please don't crawl them!")
  43     elif config.get("write_error_log").lower() != "true":
  44         logger.debug(f"Writing to error_log is disabled in configuruation file - EXIT!")
  45         return
  46
  47     logger.debug("BEFORE error[]:", type(error))
  48     if isinstance(error, BaseException, error, json.decoder.JSONDecodeError):
  49         error = f"error[{type(error)}]='{str(error)}'"
  50
  51     logger.debug("AFTER error[]:", type(error))
  52     if isinstance(error, str):
  53         fba.cursor.execute("INSERT INTO error_log (domain, error_code, error_message, created) VALUES (?, 999, ?, ?)",[
  54             domain,
  55             error,
  56             time.time()
  57         ])
  58     else:
  59         fba.cursor.execute("INSERT INTO error_log (domain, error_code, error_message, created) VALUES (?, ?, ?, ?)",[
  60             domain,
  61             error["status_code"],
  62             error["error_message"],
  63             time.time()
  64         ])
  65
  66     # Cleanup old entries
  67     logger.debug(f"Purging old records (distance: {config.get('error_log_cleanup')})")
  68     fba.cursor.execute("DELETE FROM error_log WHERE created < ?", [time.time() - config.get("error_log_cleanup")])
  69
  70     logger.debug("EXIT!")