fba/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 sys
  18 import time
  19 import validators
  20
  21 from fba import blacklist
  22 from fba import fba
  23 from fba.helpers import tidyup
  24
  25 def update_reason(reason: str, blocker: str, blocked: str, block_level: str):
  26     # DEBUG: print(f"DEBUG: reason='{reason}',blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
  27     if not isinstance(reason, str) and reason is not None:
  28         raise ValueError(f"Parameter reason[]='{type(reason)}' is not 'str'")
  29     elif not isinstance(blocker, str):
  30         raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not 'str'")
  31     elif blocker == "":
  32         raise ValueError("Parameter 'blocker' is empty")
  33     elif not isinstance(blocked, str):
  34         raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not 'str'")
  35     elif blocked == "":
  36         raise ValueError("Parameter 'blocked' is empty")
  37     elif not isinstance(block_level, str):
  38         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not 'str'")
  39     elif block_level == "":
  40         raise ValueError("Parameter 'block_level' is empty")
  41
  42     # DEBUG: print("DEBUG: Updating block reason:", reason, blocker, blocked, block_level)
  43     try:
  44         fba.cursor.execute(
  45             "UPDATE blocks SET reason = ?, last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? AND (reason IS NULL OR reason IN ('','unknown')) LIMIT 1",
  46             (
  47                 reason,
  48                 time.time(),
  49                 blocker,
  50                 blocked,
  51                 block_level
  52             ),
  53         )
  54
  55         # DEBUG: print(f"DEBUG: fba.cursor.rowcount={fba.cursor.rowcount}")
  56         if fba.cursor.rowcount == 0:
  57             # DEBUG: print(f"DEBUG: Did not update any rows: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}',reason='{reason}' - EXIT!")
  58             return
  59
  60     except BaseException as exception:
  61         print(f"ERROR: failed SQL query: reason='{reason}',blocker='{blocker}',blocked='{blocked}',block_level='{block_level}',exception[{type(exception)}]:'{str(exception)}'")
  62         sys.exit(255)
  63
  64     # DEBUG: print("DEBUG: EXIT!")
  65
  66 def update_last_seen(blocker: str, blocked: str, block_level: str):
  67     # DEBUG: print("DEBUG: Updating last_seen for:", blocker, blocked, block_level)
  68     if not isinstance(blocker, str):
  69         raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not 'str'")
  70     elif blocker == "":
  71         raise ValueError("Parameter 'blocker' is empty")
  72     elif not isinstance(blocked, str):
  73         raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not 'str'")
  74     elif blocked == "":
  75         raise ValueError("Parameter 'blocked' is empty")
  76     elif not isinstance(block_level, str):
  77         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not 'str'")
  78     elif block_level == "":
  79         raise ValueError("Parameter 'block_level' is empty")
  80
  81     try:
  82         fba.cursor.execute(
  83             "UPDATE blocks SET last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
  84             (
  85                 time.time(),
  86                 blocker,
  87                 blocked,
  88                 block_level
  89             )
  90         )
  91
  92         # DEBUG: print(f"DEBUG: fba.cursor.rowcount={fba.cursor.rowcount}")
  93         if fba.cursor.rowcount == 0:
  94             # DEBUG: print(f"DEBUG: Did not update any rows: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' - EXIT!")
  95             return
  96
  97     except BaseException as exception:
  98         print(f"ERROR: failed SQL query: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}',exception[{type(exception)}]:'{str(exception)}'")
  99         sys.exit(255)
 100
 101     # DEBUG: print("DEBUG: EXIT!")
 102
 103 def is_instance_blocked(blocker: str, blocked: str, block_level: str) -> bool:
 104     # DEBUG: print(f"DEBUG: blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
 105     if not isinstance(blocker, str):
 106         raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not of type 'str'")
 107     elif blocker == "":
 108         raise ValueError("Parameter 'blocker' is empty")
 109     elif not isinstance(blocked, str):
 110         raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not of type 'str'")
 111     elif blocked == "":
 112         raise ValueError("Parameter 'blocked' is empty")
 113     elif not isinstance(block_level, str):
 114         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
 115     elif block_level == "":
 116         raise ValueError("Parameter 'block_level' is empty")
 117
 118     fba.cursor.execute(
 119         "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
 120         (
 121             blocker,
 122             blocked,
 123             block_level
 124         ),
 125     )
 126
 127     is_blocked = fba.cursor.fetchone() is not None
 128
 129     # DEBUG: print(f"DEBUG: is_blocked='{is_blocked}' - EXIT!")
 130     return is_blocked
 131
 132 def add_instance(blocker: str, blocked: str, reason: str, block_level: str):
 133     # DEBUG: print("DEBUG: blocker,blocked,reason,block_level:", blocker, blocked, reason, block_level)
 134     if not isinstance(blocker, str):
 135         raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not 'str'")
 136     elif blocker == "":
 137         raise ValueError("Parameter 'blocker' is empty")
 138     elif not validators.domain(blocker.split("/")[0]):
 139         raise ValueError(f"Bad blocker='{blocker}'")
 140     elif not isinstance(blocked, str):
 141         raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not 'str'")
 142     elif blocked == "":
 143         raise ValueError("Parameter 'blocked' is empty")
 144     elif not isinstance(block_level, str):
 145         raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not of type 'str'")
 146     elif block_level == "":
 147         raise ValueError("Parameter 'block_level' is empty")
 148     elif not validators.domain(blocked.split("/")[0]):
 149         raise ValueError(f"Bad blocked='{blocked}'")
 150     elif blacklist.is_blacklisted(blocker):
 151         raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
 152     elif blacklist.is_blacklisted(blocked):
 153         raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
 154
 155     if reason is not None:
 156         # Maybe needs cleaning
 157         reason = tidyup.reason(reason)
 158
 159     print(f"INFO: New block: blocker='{blocker}',blocked='{blocked}', reason='{reason}', block_level='{block_level}'")
 160     try:
 161         fba.cursor.execute(
 162             "INSERT INTO blocks (blocker, blocked, reason, block_level, first_seen, last_seen) VALUES(?, ?, ?, ?, ?, ?)",
 163              (
 164                  blocker,
 165                  blocked,
 166                  reason,
 167                  block_level,
 168                  time.time(),
 169                  time.time()
 170              ),
 171         )
 172     except BaseException as exception:
 173         print(f"ERROR: failed SQL query: blocker='{blocker}',blocked='{blocked}',reason='{reason}',block_level='{block_level}',exception[{type(exception)}]:'{str(exception)}'")
 174         sys.exit(255)
 175
 176     # DEBUG: print("DEBUG: EXIT!")