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     if not isinstance(reason, str) and reason is not None:
  33         raise ValueError(f"Parameter reason[]='{type(reason)}' is not of type 'str'")
  34     elif not isinstance(blocker, str):
  35         raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not of type 'str'")
  36     elif blocker == "":
  37         raise ValueError("Parameter 'blocker' is empty")
  38     elif blocker.lower() != blocker:
  39         raise ValueError(f"Parameter blocker='{blocker}' must be all lower-case")
  40     elif not isinstance(blocked, str):
  41         raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not of type 'str'")
  42     elif blocked == "":
  43         raise ValueError("Parameter 'blocked' is empty")
  44     elif blocked.lower() != blocked:
  45         raise ValueError(f"Parameter blocked='{blocked}' must be all lower-case")
  46     elif not isinstance(block_level, str):
  47         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
  48     elif block_level == "":
  49         raise ValueError("Parameter 'block_level' is empty")
  50     elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
  51         raise ValueError(f"block_level='{block_level}' is not wanted.")
  52
  53     logger.debug("Updating block reason='%s',blocker='%s',blocked='%s',block_level='%s'", reason, blocker, blocked, block_level)
  54     database.cursor.execute(
  55         "UPDATE blocks SET reason = ?, last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? AND (reason IS NULL OR reason = '') LIMIT 1",
  56         [
  57             reason,
  58             time.time(),
  59             blocker,
  60             blocked,
  61             block_level
  62         ])
  63
  64     logger.debug("EXIT!")
  65
  66 def update_last_seen(blocker: str, blocked: str, block_level: str):
  67     logger.debug("blocker='%s',blocked='%s',block_level='%s' - CALLED!", blocker, blocked, block_level)
  68     if not isinstance(blocker, str):
  69         raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not of type 'str'")
  70     elif blocker == "":
  71         raise ValueError("Parameter 'blocker' is empty")
  72     elif blocker.lower() != blocker:
  73         raise ValueError(f"Parameter blocker='{blocker}' must be all lower-case")
  74     elif not isinstance(blocked, str):
  75         raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not of type 'str'")
  76     elif blocked == "":
  77         raise ValueError("Parameter 'blocked' is empty")
  78     elif blocked.lower() != blocked:
  79         raise ValueError(f"Parameter blocked='{blocked}' must be all lower-case")
  80     elif not isinstance(block_level, str):
  81         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
  82     elif block_level == "":
  83         raise ValueError("Parameter 'block_level' is empty")
  84     elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
  85         raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
  86
  87     database.cursor.execute(
  88         "UPDATE blocks SET last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
  89         [
  90             time.time(),
  91             blocker,
  92             blocked,
  93             block_level
  94         ])
  95
  96     logger.debug("EXIT!")
  97
  98 def is_instance_blocked(blocker: str, blocked: str, block_level: str = None) -> bool:
  99     logger.debug("blocker='%s',blocked='%s',block_level='%s' - CALLED!", blocker, blocked, block_level)
 100     if not isinstance(blocker, str):
 101         raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not of type 'str'")
 102     elif blocker == "":
 103         raise ValueError("Parameter 'blocker' is empty")
 104     elif blocker.lower() != blocker:
 105         raise ValueError(f"Parameter blocker='{blocker}' must be all lower-case")
 106     elif not isinstance(blocked, str):
 107         raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not of type 'str'")
 108     elif blocked == "":
 109         raise ValueError("Parameter 'blocked' is empty")
 110     elif blocked.lower() != blocked:
 111         raise ValueError(f"Parameter blocked='{blocked}' must be all lower-case")
 112     elif not isinstance(block_level, str) and block_level is not None:
 113         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
 114     elif block_level == "":
 115         raise ValueError("Parameter 'block_level' is empty")
 116     elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
 117         raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
 118
 119     if block_level is None:
 120         database.cursor.execute(
 121             "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? LIMIT 1",
 122             [
 123                 blocker,
 124                 blocked
 125             ]
 126          )
 127     else:
 128         database.cursor.execute(
 129             "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
 130             [
 131                 blocker,
 132                 blocked,
 133                 block_level
 134             ]
 135         )
 136
 137     is_blocked = database.cursor.fetchone() is not None
 138
 139     logger.debug("is_blocked='%s' - EXIT!", is_blocked)
 140     return is_blocked
 141
 142 def add_instance(blocker: str, blocked: str, reason: str, block_level: str):
 143     logger.debug("blocker='%s',blocked='%s',reason='%s',block_level='%s' - CALLED!", blocker, blocked, reason, block_level)
 144     domain_helper.raise_on(blocker)
 145     domain_helper.raise_on(blocked)
 146
 147     if not isinstance(block_level, str):
 148         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
 149     elif block_level == "":
 150         raise ValueError("Parameter 'block_level' is empty")
 151     elif blacklist.is_blacklisted(blocker):
 152         raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
 153     elif blacklist.is_blacklisted(blocked):
 154         raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
 155     elif block_level in ["accept", "suspend", "silence", "nsfw", "quarantined_instances"]:
 156         raise ValueError(f"blocked='{blocked}' has unwanted block_level='{block_level}'")
 157
 158     if reason is not None:
 159         # Maybe needs cleaning
 160         reason = tidyup.reason(reason)
 161
 162     logger.info("New block: blocker='%s',blocked='%s',reason='%s',block_level='%s'", blocker, blocked, reason, block_level)
 163
 164     database.cursor.execute(
 165         "INSERT INTO blocks (blocker, blocked, reason, block_level, first_seen, last_seen) VALUES (?, ?, ?, ?, ?, ?)",
 166         [
 167              blocker,
 168              blocked,
 169              reason,
 170              block_level,
 171              time.time(),
 172              time.time()
 173         ])
 174
 175     logger.debug("EXIT!")
 176
 177 def valid(value: str, column: str) -> bool:
 178     logger.debug("value='%s' - CALLED!", value)
 179     if not isinstance(value, str):
 180         raise ValueError(f"Parameter value[]='{type(value)}' is not of type 'str'")
 181     elif value == "":
 182         raise ValueError("Parameter 'value' is empty")
 183     elif not isinstance(column, str):
 184         raise ValueError(f"Parameter column[]='{type(column)}' is not of type 'str'")
 185     elif column == "":
 186         raise ValueError("Parameter 'column' is empty")
 187
 188     # Query database
 189     database.cursor.execute(
 190         f"SELECT {column} FROM blocks WHERE {column} = ? LIMIT 1", [value]
 191     )
 192
 193     is_valid = database.cursor.fetchone() is not None
 194
 195     logger.debug("is_valid='%s' - EXIT!", is_valid)
 196     return is_valid
 197
 198 def translate_idnas(rows: list, column: str):
 199     logger.debug("rows[]='%s' - CALLED!", type(rows))
 200     if not isinstance(rows, list):
 201         raise ValueError(f"rows[]='{type(rows)}' is not of type 'list'")
 202     elif len(rows) == 0:
 203         raise ValueError("Parameter 'rows' is an empty list")
 204     elif not isinstance(column, str):
 205         raise ValueError(f"column='{type(column)}' is not of type 'str'")
 206     elif column == "":
 207         raise ValueError("Parameter 'column' is empty")
 208     elif column not in ["blocker", "blocked"]:
 209         raise ValueError(f"column='{column}' is not supported")
 210
 211     logger.info("Checking/converting %d domain names ...", len(rows))
 212     for row in rows:
 213         logger.debug("row[]='%s'", type(row))
 214
 215         translated = row[column].encode("idna").decode("utf-8")
 216         logger.debug("translated='%s',row[%s]='%s'", translated, column, row[column])
 217
 218         if translated != row[column]:
 219             logger.info("Translated row[%s]='%s' to '%s'", column, row[column], translated)
 220             database.cursor.execute(f"UPDATE blocks SET {column} = ? WHERE {column} = ?", [translated, row[column]])
 221
 222             logger.debug("Invoking commit() ...")
 223             database.connection.commit()
 224
 225     logger.debug("EXIT!")
 226
 227 def alias_block_level(block_level: str) -> str:
 228     logger.debug("block_level='%s' - CALLED!", block_level)
 229     if not isinstance(block_level, str):
 230         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
 231     elif block_level == "":
 232         raise ValueError("Parameter 'block_level' is empty")
 233
 234     if block_level == "silence":
 235         logger.debug("Block level 'silence' has been changed to 'silenced'")
 236         block_level = "silenced"
 237     elif block_level == "suspend":
 238         logger.debug("Block level 'suspend' has been changed to 'suspended'")
 239         block_level = "suspended"
 240     elif block_level == "nsfw":
 241         logger.debug("Block level 'nsfw' has been changed to 'media_nsfw'")
 242         block_level = "media_nsfw"
 243     elif block_level == "quarantined_instances":
 244         logger.debug("Block level 'quarantined_instances' has been changed to 'quarantined'")
 245         block_level = "quarantined"
 246
 247     logger.debug("block_level='%s' - EXIT!", block_level)
 248     return block_level