fba/models/blocks.py

   1 # Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
   2 # Copyright (C) 2023 Free Software Foundation
   3 #
   4 # This program is free software: you can redistribute it and/or modify
   5 # it under the terms of the GNU Affero General Public License as published
   6 # by the Free Software Foundation, either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU Affero General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Affero General Public License
  15 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
  16
  17 import logging
  18
  19 import time
  20
  21 from fba import database
  22
  23 from fba.helpers import blacklist
  24 from fba.helpers import domain as domain_helper
  25 from fba.helpers import tidyup
  26
  27 logging.basicConfig(level=logging.INFO)
  28 logger = logging.getLogger(__name__)
  29
  30 def update_reason(reason: str, blocker: str, blocked: str, block_level: str):
  31     logger.debug("reason='%s',blocker='%s',blocked='%s',block_level='%s' - CALLED!", reason, blocker, blocked, block_level)
  32     domain_helper.raise_on(blocker)
  33     domain_helper.raise_on(blocked)
  34
  35     if not isinstance(reason, str) and reason is not None:
  36         raise ValueError(f"Parameter reason[]='{type(reason)}' is not of type 'str'")
  37     elif not isinstance(block_level, str):
  38         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
  39     elif block_level == "":
  40         raise ValueError("Parameter 'block_level' is empty")
  41     elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
  42         raise ValueError(f"block_level='{block_level}' is not wanted.")
  43     elif blacklist.is_blacklisted(blocker):
  44         raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
  45     elif blacklist.is_blacklisted(blocked):
  46         raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
  47     elif not is_instance_blocked(blocker, blocked, block_level):
  48         raise Exception(f"blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' is not registered but function is invoked")
  49
  50     logger.debug("Updating block reason='%s',blocker='%s',blocked='%s',block_level='%s'", reason, blocker, blocked, block_level)
  51     database.cursor.execute(
  52         "UPDATE blocks SET reason = ?, last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? AND (reason IS NULL OR reason = '') LIMIT 1",
  53         [
  54             reason,
  55             time.time(),
  56             blocker,
  57             blocked,
  58             block_level
  59         ])
  60
  61     logger.debug("EXIT!")
  62
  63 def update_last_seen(blocker: str, blocked: str, block_level: str):
  64     logger.debug("blocker='%s',blocked='%s',block_level='%s' - CALLED!", blocker, blocked, block_level)
  65     domain_helper.raise_on(blocker)
  66     domain_helper.raise_on(blocked)
  67
  68     if not isinstance(block_level, str):
  69         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
  70     elif block_level == "":
  71         raise ValueError("Parameter 'block_level' is empty")
  72     elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
  73         raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
  74     elif blacklist.is_blacklisted(blocker):
  75         raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
  76     elif blacklist.is_blacklisted(blocked):
  77         raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
  78     elif not is_instance_blocked(blocker, blocked, block_level):
  79         raise Exception(f"blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' is not registered but function is invoked")
  80
  81     database.cursor.execute(
  82         "UPDATE blocks SET last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
  83         [
  84             time.time(),
  85             blocker,
  86             blocked,
  87             block_level
  88         ])
  89
  90     logger.debug("EXIT!")
  91
  92 def is_instance_blocked(blocker: str, blocked: str, block_level: str = None) -> bool:
  93     logger.debug("blocker='%s',blocked='%s',block_level='%s' - CALLED!", blocker, blocked, block_level)
  94     domain_helper.raise_on(blocker)
  95     domain_helper.raise_on(blocked)
  96
  97     if not isinstance(block_level, str) and block_level is not None:
  98         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
  99     elif block_level == "":
 100         raise ValueError("Parameter 'block_level' is empty")
 101     elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
 102         raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
 103     elif blacklist.is_blacklisted(blocker):
 104         raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
 105     elif blacklist.is_blacklisted(blocked):
 106         raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
 107
 108     if block_level is None:
 109         database.cursor.execute(
 110             "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? LIMIT 1",
 111             [
 112                 blocker,
 113                 blocked
 114             ]
 115          )
 116     else:
 117         database.cursor.execute(
 118             "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
 119             [
 120                 blocker,
 121                 blocked,
 122                 block_level
 123             ]
 124         )
 125
 126     is_blocked = database.cursor.fetchone() is not None
 127
 128     logger.debug("is_blocked='%s' - EXIT!", is_blocked)
 129     return is_blocked
 130
 131 def add(blocker: str, blocked: str, reason: str, block_level: str):
 132     logger.debug("blocker='%s',blocked='%s',reason='%s',block_level='%s' - CALLED!", blocker, blocked, reason, block_level)
 133     domain_helper.raise_on(blocker)
 134     domain_helper.raise_on(blocked)
 135
 136     if not isinstance(block_level, str):
 137         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
 138     elif block_level == "":
 139         raise ValueError("Parameter 'block_level' is empty")
 140     elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
 141         raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
 142     elif blacklist.is_blacklisted(blocker):
 143         raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
 144     elif blacklist.is_blacklisted(blocked):
 145         raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
 146     elif reason is not None and reason == "":
 147         raise Exception(f"blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' has empty (not 'None') block reason set")
 148
 149     if reason is not None:
 150         # Maybe needs cleaning
 151         logger.debug("reason='%s' - BEFORE!")
 152         reason = tidyup.reason(reason)
 153         logger.debug("reason='%s' - AFTER!")
 154
 155     logger.info("New block: blocker='%s',blocked='%s',reason='%s',block_level='%s'", blocker, blocked, reason, block_level)
 156
 157     database.cursor.execute(
 158         "INSERT INTO blocks (blocker, blocked, reason, block_level, first_seen, last_seen) VALUES (?, ?, ?, ?, ?, ?)",
 159         [
 160              blocker,
 161              blocked,
 162              reason,
 163              block_level,
 164              time.time(),
 165              time.time()
 166         ])
 167
 168     logger.debug("EXIT!")
 169
 170 def valid(value: str, column: str) -> bool:
 171     logger.debug("value='%s' - CALLED!", value)
 172
 173     if not isinstance(value, str):
 174         raise ValueError(f"Parameter value[]='{type(value)}' is not of type 'str'")
 175     elif value == "":
 176         raise ValueError("Parameter 'value' is empty")
 177     elif not isinstance(column, str):
 178         raise ValueError(f"Parameter column[]='{type(column)}' is not of type 'str'")
 179     elif column == "":
 180         raise ValueError("Parameter 'column' is empty")
 181
 182     # Query database
 183     database.cursor.execute(
 184         f"SELECT {column} FROM blocks WHERE {column} = ? LIMIT 1", [value]
 185     )
 186
 187     is_valid = database.cursor.fetchone() is not None
 188
 189     logger.debug("is_valid='%s' - EXIT!", is_valid)
 190     return is_valid
 191
 192 def translate_idnas(rows: list, column: str):
 193     logger.debug("rows[]='%s' - CALLED!", type(rows))
 194
 195     if not isinstance(rows, list):
 196         raise ValueError(f"rows[]='{type(rows)}' is not of type 'list'")
 197     elif len(rows) == 0:
 198         raise ValueError("Parameter 'rows' is an empty list")
 199     elif not isinstance(column, str):
 200         raise ValueError(f"column='{type(column)}' is not of type 'str'")
 201     elif column not in ["blocker", "blocked"]:
 202         raise ValueError(f"column='{column}' is not supported")
 203
 204     logger.info("Checking/converting %d domain names ...", len(rows))
 205     for row in rows:
 206         logger.debug("row[]='%s'", type(row))
 207
 208         translated = row[column].encode("idna").decode("utf-8")
 209         logger.debug("translated='%s',row[%s]='%s'", translated, column, row[column])
 210
 211         if translated != row[column]:
 212             logger.info("Translated row[%s]='%s' to '%s'", column, row[column], translated)
 213             database.cursor.execute(f"UPDATE blocks SET {column} = ? WHERE {column} = ?", [translated, row[column]])
 214
 215             logger.debug("Invoking commit() ...")
 216             database.connection.commit()
 217
 218     logger.debug("EXIT!")
 219
 220 def alias_block_level(block_level: str) -> str:
 221     logger.debug("block_level='%s' - CALLED!", block_level)
 222
 223     if not isinstance(block_level, str):
 224         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
 225     elif block_level in ["accept", "accepted"]:
 226         raise ValueError(f"Parameter block_level='{block_level}' is not accepted but function was invoked")
 227     elif block_level == "silence":
 228         logger.debug("Block level 'silence' has been changed to 'silenced'")
 229         block_level = "silenced"
 230     elif block_level == "suspend":
 231         logger.debug("Block level 'suspend' has been changed to 'suspended'")
 232         block_level = "suspended"
 233     elif block_level == "nsfw":
 234         logger.debug("Block level 'nsfw' has been changed to 'media_nsfw'")
 235         block_level = "media_nsfw"
 236     elif block_level == "quarantined_instances":
 237         logger.debug("Block level 'quarantined_instances' has been changed to 'quarantined'")
 238         block_level = "quarantined"
 239
 240     logger.debug("block_level='%s' - EXIT!", block_level)
 241     return block_level