1 # Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
2 # Copyright (C) 2023 Free Software Foundation
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.
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.
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/>.
20 from fba import blacklist
21 from fba import blocks
22 from fba import config
24 from fba import instances
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'")
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'")
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")
43 federation = fba.get_response(domain, f"{fba.fetch_peers_url}?filter=suspended", fba.api_headers, (config.get("connection_timeout"), config.get("read_timeout"))).json()
45 if (federation == None):
46 print("WARNING: No valid response:", domain);
47 elif "error" in federation:
48 print("WARNING: API returned error:", federation["error"])
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)
58 print("WARNING: blocked is empty:", domain)
60 elif blacklist.is_blacklisted(blocked):
61 print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
63 elif blocked.count("*") > 0:
64 # GTS does not have hashes for obscured domains, so we have to guess it
66 "SELECT domain, origin, nodeinfo_url FROM instances WHERE domain LIKE ? ORDER BY rowid LIMIT 1", [blocked.replace("*", "_")]
68 searchres = fba.cursor.fetchone()
71 print(f"WARNING: Cannot deobsfucate blocked='{blocked}' - SKIPPED!")
74 blocked = searchres[0]
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!")
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!")
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)
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")
98 print(f"DEBUG: Updating block last seen for domain='{domain}',blocked='{blocked}' ...")
99 blocks.update_last_seen(domain, blocked, "reject")
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")
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"]
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)}'")
115 print("DEBUG: EXIT!")