gotosocial.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 inspect
  18 import validators
  19
  20 from fba import blacklist
  21 from fba import blocks
  22 from fba import config
  23 from fba import fba
  24 from fba import instances
  25
  26 def fetch_blocks(domain: str, origin: str, nodeinfo_url: str):
  27     print(f"DEBUG: domain='{domain}',origin='{origin}',nodeinfo_url='{nodeinfo_url}' - CALLED!")
  28     if type(domain) != str:
  29         raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
  30     elif domain == "":
  31         raise ValueError(f"Parameter 'domain' is empty")
  32     elif type(origin) != str and origin != None:
  33         raise ValueError(f"Parameter origin[]={type(origin)} is not 'str'")
  34     elif origin == "":
  35         raise ValueError(f"Parameter 'origin' is empty")
  36     elif type(nodeinfo_url) != str:
  37         raise ValueError(f"Parameter nodeinfo_url[]={type(nodeinfo_url)} is not 'str'")
  38     elif nodeinfo_url == "":
  39         raise ValueError(f"Parameter 'nodeinfo_url' is empty")
  40
  41     try:
  42         # Blocks
  43         federation = fba.get_response(domain, f"/api/v1/instance/peers?filter=suspended", fba.api_headers, (config.get("connection_timeout"), config.get("read_timeout"))).json()
  44
  45         if (federation == None):
  46             print("WARNING: No valid response:", domain);
  47         elif "error" in federation:
  48             print("WARNING: API returned error:", federation["error"])
  49         else:
  50             print(f"INFO: Checking {len(federation)} entries from domain='{domain}',software='gotosocial' ...")
  51             for peer in federation:
  52                 blocked = peer["domain"].lower()
  53                 print("DEBUG: BEFORE blocked:", blocked)
  54                 blocked = fba.tidyup_domain(blocked)
  55                 print("DEBUG: AFTER blocked:", blocked)
  56
  57                 if blocked == "":
  58                     print("WARNING: blocked is empty:", domain)
  59                     continue
  60                 elif blacklist.is_blacklisted(blocked):
  61                     print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
  62                     continue
  63                 elif blocked.count("*") > 0:
  64                     # GTS does not have hashes for obscured domains, so we have to guess it
  65                     fba.cursor.execute(
  66                         "SELECT domain, origin, nodeinfo_url FROM instances WHERE domain LIKE ? ORDER BY rowid LIMIT 1", [blocked.replace("*", "_")]
  67                     )
  68                     searchres = fba.cursor.fetchone()
  69
  70                     if searchres == None:
  71                         print(f"WARNING: Cannot deobsfucate blocked='{blocked}' - SKIPPED!")
  72                         continue
  73
  74                     blocked = searchres[0]
  75                     origin = searchres[1]
  76                     nodeinfo_url = searchres[2]
  77                 elif not validators.domain(blocked):
  78                     print(f"WARNING: blocked='{blocked}',software='gotosocial' is not a valid domain name - skipped!")
  79                     continue
  80
  81                 print("DEBUG: Looking up instance by domain:", blocked)
  82                 if not validators.domain(blocked):
  83                     print(f"WARNING: blocked='{blocked}',software='gotosocial' is not a valid domain name - skipped!")
  84                     continue
  85                 elif not instances.is_registered(blocked):
  86                     print(f"DEBUG: Domain blocked='{blocked}' wasn't found, adding ..., domain='{domain}',origin='{origin}',nodeinfo_url='{nodeinfo_url}'")
  87                     instances.add(blocked, domain, inspect.currentframe().f_code.co_name, nodeinfo_url)
  88
  89                 if not blocks.is_instance_blocked(domain, blocked, "reject"):
  90                     print(f"DEBUG: domain='{domain}' is blocking '{blocked}' for unknown reason at this point")
  91                     blocks.add_instance(domain, blocked, "unknown", "reject")
  92
  93                     blockdict.append({
  94                         "blocked": blocked,
  95                         "reason" : None
  96                     })
  97                 else:
  98                     print(f"DEBUG: Updating block last seen for domain='{domain}',blocked='{blocked}' ...")
  99                     blocks.update_last_seen(domain, blocked, "reject")
 100
 101                 if "public_comment" in peer:
 102                     print("DEBUG: Updating block reason:", domain, blocked, peer["public_comment"])
 103                     blocks.update_reason(peer["public_comment"], domain, blocked, "reject")
 104
 105                     for entry in blockdict:
 106                         if entry["blocked"] == blocked:
 107                             print(f"DEBUG: Setting block reason for blocked='{blocked}':'{peer['public_comment']}'")
 108                             entry["reason"] = peer["public_comment"]
 109
 110             print("DEBUG: Committing changes ...")
 111             fba.connection.commit()
 112     except Exception as e:
 113         print(f"ERROR: domain='{domain}',software='gotosocial',exception[{type(e)}]:'{str(e)}'")
 114
 115     print("DEBUG: EXIT!")