- # DEBUG: print("DEBUG: Looking up instance by domain:", blocked)
- if not validators.domain(blocked):
- print(f"WARNING: blocked='{blocked}',software='{software}' is not a valid domain name - skipped!")
- continue
- elif not fba.is_instance_registered(blocked):
- # DEBUG: print("DEBUG: Hash wasn't found, adding:", blocked, blocker)
- fba.add_instance(blocked, blocker, inspect.currentframe().f_code.co_name, nodeinfo_url)
-
- if not blocks.is_instance_blocked(blocker, blocked, block_level):
- blocks.add_instance(blocker, blocked, reason, block_level)
-
- if block_level == "reject":
- blockdict.append({
- "blocked": blocked,
- "reason" : reason
- })
- else:
- # DEBUG: print(f"DEBUG: Updating block last seen and reason for blocker='{blocker}',blocked='{blocked}' ...")
- fba.update_last_seen(blocker, blocked, block_level)
- blocks.update_reason(reason, blocker, blocked, block_level)
-
- # DEBUG: print("DEBUG: Committing changes ...")
- fba.connection.commit()
- except Exception as e:
- print(f"ERROR: blocker='{blocker}',software='{software}',exception[{type(e)}]:'{str(e)}'")
- elif software == "gotosocial":
- print(f"INFO: blocker='{blocker}',software='{software}'")
- try:
- # Blocks
- federation = fba.get_response(blocker, f"{fba.get_peers_url}?filter=suspended", fba.api_headers, (config.get("connection_timeout"), config.get("read_timeout"))).json()
-
- if (federation == None):
- print("WARNING: No valid response:", blocker);
- elif "error" in federation:
- print("WARNING: API returned error:", federation["error"])
- else:
- print(f"INFO: Checking {len(federation)} entries from blocker='{blocker}',software='{software}' ...")
- for peer in federation:
- blocked = peer["domain"].lower()
- # DEBUG: print("DEBUG: BEFORE blocked:", blocked)
- blocked = fba.tidyup_domain(blocked)
- # DEBUG: print("DEBUG: AFTER blocked:", blocked)
-
- if blocked == "":
- print("WARNING: blocked is empty:", blocker)
- continue
- elif blacklist.is_blacklisted(blocked):
- # DEBUG: print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
- continue
- elif blocked.count("*") > 0:
- # GTS does not have hashes for obscured domains, so we have to guess it
- fba.cursor.execute(
- "SELECT domain, origin, nodeinfo_url FROM instances WHERE domain LIKE ? ORDER BY rowid LIMIT 1", [blocked.replace("*", "_")]
- )
- searchres = fba.cursor.fetchone()
-
- if searchres == None:
- print(f"WARNING: Cannot deobsfucate blocked='{blocked}' - SKIPPED!")
- continue
-
- blocked = searchres[0]
- origin = searchres[1]
- nodeinfo_url = searchres[2]
- elif not validators.domain(blocked):
- print(f"WARNING: blocked='{blocked}',software='{software}' is not a valid domain name - skipped!")
- continue