fba/helpers/processing.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
  18 from fba import utils
  19
  20 from fba.helpers import blacklist
  21 from fba.helpers import domain as domain_helper
  22
  23 from fba.http import federation
  24 from fba.http import network
  25
  26 from fba.models import blocks
  27 from fba.models import instances
  28
  29 logging.basicConfig(level=logging.INFO)
  30 logger = logging.getLogger(__name__)
  31
  32 def instance(blocked: str, blocker: str, command: str) -> bool:
  33     logger.debug("blocked='%s',blocker='%s',command='%s' - CALLED!", blocked, blocker, command)
  34     domain_helper.raise_on(blocked)
  35     domain_helper.raise_on(blocker)
  36
  37     if not isinstance(command, str):
  38         raise ValueError(f"Parameter command[]='{type(command)}' is not of type 'str'")
  39     elif command == "":
  40         raise ValueError("Parameter 'command' is empty")
  41
  42     logger.debug("blocked='%s' - BEFORE!", blocked)
  43     blocked = utils.deobfuscate(blocked, blocker)
  44
  45     logger.debug("blocked='%s' - DEOBFUSCATED!", blocked)
  46     if instances.has_pending(blocker):
  47         logger.debug("Flushing updates for blocker='%s' ...", blocker)
  48         instances.update(blocker)
  49
  50     if not domain_helper.is_wanted(blocked):
  51         logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked)
  52         return False
  53     elif instances.is_recent(blocked):
  54         logger.debug("blocked='%s' has been recently checked - SKIPPED!", blocked)
  55         return False
  56
  57     processed = False
  58     try:
  59         logger.info("Fetching instances for blocked='%s',blocker='%s',command='%s' ...", blocked, blocker, command)
  60         federation.fetch_instances(blocked, blocker, None, command)
  61         processed = True
  62     except network.exceptions as exception:
  63         logger.warning("Exception '%s' during fetching instances (%s) from blocked='%s'", type(exception), command, blocked)
  64         instances.set_last_error(blocked, exception)
  65
  66     logger.debug("Checking if blocked='%s' has pending updates ...", blocked)
  67     if instances.has_pending(blocked):
  68         logger.debug("Flushing updates for blocked='%s' ...", blocked)
  69         instances.update(blocked)
  70
  71     logger.debug("processed='%s' - EXIT!", processed)
  72     return processed
  73
  74 def block(blocker: str, blocked: str, reason: str, block_level: str) -> bool:
  75     logger.debug("blocker='%s',blocked='%s',reason='%s',block_level='%s' - CALLED!", blocker, blocked, reason, block_level)
  76     domain_helper.raise_on(blocker)
  77     domain_helper.raise_on(blocked)
  78
  79     if not isinstance(reason, str) and reason is not None:
  80         raise ValueError(f"Parameter reason[]='{type(reason)}' is not of type 'str'")
  81     elif not isinstance(block_level, str):
  82         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
  83     elif block_level == "":
  84         raise ValueError("Parameter block_level is empty")
  85     elif blacklist.is_blacklisted(blocked):
  86         raise ValueError(f"blocked='{blocked}' is blacklisted but function was invoked")
  87
  88     added = False
  89     if not blocks.is_instance_blocked(blocker, blocked, block_level):
  90         logger.debug("Invoking blocks.add(%s, %s, %s, %s) ...", blocker, blocked, reason, block_level)
  91         blocks.add(blocker, blocked, reason, block_level)
  92         added = True
  93     else:
  94         logger.debug("Updating last_seen for blocker='%s',blocked='%s',block_level='%s' ...", blocker, blocked, block_level)
  95         blocks.update_last_seen(blocker, blocked, block_level)
  96
  97     logger.debug("added='%s' - EXIT!", added)
  98     return added