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