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