-def update_block_reason(reason: str, blocker: str, blocked: str, block_level: str):
- # DEBUG: print(f"DEBUG: reason='{reason}',blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
- if type(reason) != str and reason != None:
- raise ValueError(f"Parameter reason[]='{type(reason)}' is not 'str'")
- elif type(blocker) != str:
- raise ValueError(f"Parameter blocker[]='{type(blocker)}' is not 'str'")
- elif type(blocked) != str:
- raise ValueError(f"Parameter blocked[]='{type(blocked)}' is not 'str'")
- elif type(block_level) != str:
- raise ValueError(f"Parameter block_level[]='{type(block_level)}' is not 'str'")
-
- # DEBUG: print("DEBUG: Updating block reason:", reason, blocker, blocked, block_level)
- try:
- cursor.execute(
- "UPDATE blocks SET reason = ?, last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? AND reason IN ('','unknown') LIMIT 1",
- (
- reason,
- time.time(),
- blocker,
- blocked,
- block_level
- ),
- )
-
- # DEBUG: print(f"DEBUG: cursor.rowcount={cursor.rowcount}")
- if cursor.rowcount == 0:
- # DEBUG: print(f"DEBUG: Did not update any rows: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}',reason='{reason}' - EXIT!")
- return
-
- except BaseException as e:
- print(f"ERROR: failed SQL query: reason='{reason}',blocker='{blocker}',blocked='{blocked}',block_level='{block_level}',exception[{type(e)}]:'{str(e)}'")
- sys.exit(255)
-
- # DEBUG: print("DEBUG: EXIT!")
-
-def update_last_seen(blocker: str, blocked: str, block_level: str):
- # DEBUG: print("DEBUG: Updating last_seen for:", blocker, blocked, block_level)
- try:
- cursor.execute(
- "UPDATE blocks SET last_seen = ? WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
- (
- time.time(),
- blocker,
- blocked,
- block_level
- )
- )
-
- # DEBUG: print(f"DEBUG: cursor.rowcount={cursor.rowcount}")
- if cursor.rowcount == 0:
- # DEBUG: print(f"DEBUG: Did not update any rows: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}' - EXIT!")
- return
-
- except BaseException as e:
- print(f"ERROR: failed SQL query: blocker='{blocker}',blocked='{blocked}',block_level='{block_level}',exception[{type(e)}]:'{str(e)}'")
- sys.exit(255)
-
- # DEBUG: print("DEBUG: EXIT!")
-
-def is_instance_blocked(blocker: str, blocked: str, block_level: str) -> bool:
- # DEBUG: print(f"DEBUG: blocker={blocker},blocked={blocked},block_level={block_level} - CALLED!")
- if type(blocker) != str:
- raise ValueError(f"Parameter blocker[]={type(blocker)} is not of type 'str'")
- elif blocker == "":
- raise ValueError("Parameter 'blocker' cannot be empty")
- elif type(blocked) != str:
- raise ValueError(f"Parameter blocked[]={type(blocked)} is not of type 'str'")
- elif blocked == "":
- raise ValueError("Parameter 'blocked' cannot be empty")
- elif type(block_level) != str:
- raise ValueError(f"Parameter block_level[]={type(block_level)} is not of type 'str'")
- elif block_level == "":
- raise ValueError("Parameter 'block_level' cannot be empty")
-
- cursor.execute(
- "SELECT * FROM blocks WHERE blocker = ? AND blocked = ? AND block_level = ? LIMIT 1",
- (
- blocker,
- blocked,
- block_level
- ),
- )
-
- is_blocked = cursor.fetchone() != None
-
- # DEBUG: print(f"DEBUG: is_blocked='{is_blocked}' - EXIT!")
- return is_blocked
-
-def block_instance(blocker: str, blocked: str, reason: str, block_level: str):
- # DEBUG: print("DEBUG: blocker,blocked,reason,block_level:", blocker, blocked, reason, block_level)
- if type(blocker) != str:
- raise ValueError(f"Parameter blocker[]={type(blocker)} is not 'str'")
- elif blocker == "":
- raise ValueError(f"Parameter 'blocker' cannot be empty")
- elif not validators.domain(blocker.split("/")[0]):
- raise ValueError(f"Bad blocker='{blocker}'")
- elif type(blocked) != str:
- raise ValueError(f"Parameter blocked[]={type(blocked)} is not 'str'")
- elif blocked == "":
- raise ValueError(f"Parameter 'blocked' cannot be empty")
- elif not validators.domain(blocked.split("/")[0]):
- raise ValueError(f"Bad blocked='{blocked}'")
- elif is_blacklisted(blocker):
- raise Exception(f"blocker='{blocker}' is blacklisted but function invoked")
- elif is_blacklisted(blocked):
- raise Exception(f"blocked='{blocked}' is blacklisted but function invoked")
-
- if reason != None:
- # Maybe needs cleaning
- reason = tidyup_reason(reason)
-
- print(f"INFO: New block: blocker='{blocker}',blocked='{blocked}', reason='{reason}', block_level='{block_level}'")
- try:
- cursor.execute(
- "INSERT INTO blocks (blocker, blocked, reason, block_level, first_seen, last_seen) VALUES(?, ?, ?, ?, ?, ?)",
- (
- blocker,
- blocked,
- reason,
- block_level,
- time.time(),
- time.time()
- ),
- )
- except BaseException as e:
- print(f"ERROR: failed SQL query: blocker='{blocker}',blocked='{blocked}',reason='{reason}',block_level='{block_level}',exception[{type(e)}]:'{str(e)}'")
- sys.exit(255)
-
- # DEBUG: print("DEBUG: EXIT!")
-
-def is_instance_registered(domain: str) -> bool:
- # DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!")
- if type(domain) != str:
- raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
- elif domain == "":
- raise ValueError(f"Parameter 'domain' cannot be empty")
-
- # DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!")
- if not cache.key_exists("is_registered"):
- # DEBUG: print(f"DEBUG: Cache for 'is_registered' not initialized, fetching all rows ...")
- try:
- cursor.execute("SELECT domain FROM instances")
-
- # Check Set all
- cache.set_all("is_registered", cursor.fetchall(), True)
- except BaseException as e:
- print(f"ERROR: failed SQL query: domain='{domain}',exception[{type(e)}]:'{str(e)}'")
- sys.exit(255)
-
- # Is cache found?
- registered = cache.sub_key_exists("is_registered", domain)
-
- # DEBUG: print(f"DEBUG: registered='{registered}' - EXIT!")
- return registered
-
-def add_instance(domain: str, origin: str, originator: str, path: str = None):
- # DEBUG: print(f"DEBUG: domain='{domain}',origin='{origin}',originator='{originator}',path='{path}' - CALLED!")
- if type(domain) != str:
- raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
- elif domain == "":
- raise ValueError(f"Parameter 'domain' cannot be empty")
- elif type(origin) != str and origin != None:
- raise ValueError(f"origin[]={type(origin)} is not 'str'")
- elif type(originator) != str:
- raise ValueError(f"originator[]={type(originator)} is not 'str'")
- elif originator == "":
- raise ValueError(f"originator cannot be empty")
- elif not validators.domain(domain.split("/")[0]):
- raise ValueError(f"Bad domain name='{domain}'")
- elif origin is not None and not validators.domain(origin.split("/")[0]):
- raise ValueError(f"Bad origin name='{origin}'")
- elif is_blacklisted(domain):
- raise Exception(f"domain='{domain}' is blacklisted, but method invoked")
-
- # DEBUG: print("DEBUG: domain,origin,originator,path:", domain, origin, originator, path)
- software = determine_software(domain, path)
- # DEBUG: print("DEBUG: Determined software:", software)
-
- print(f"INFO: Adding instance domain='{domain}' (origin='{origin}',software='{software}')")
- try:
- cursor.execute(
- "INSERT INTO instances (domain, origin, originator, hash, software, first_seen) VALUES (?, ?, ?, ?, ?, ?)",
- (
- domain,
- origin,
- originator,
- get_hash(domain),
- software,
- time.time()
- ),
- )
-
- cache.set_sub_key("is_registered", domain, True)
-
- if instances.has_pending_instance_data(domain):
- # DEBUG: print(f"DEBUG: domain='{domain}' has pending nodeinfo being updated ...")
- instances.set("last_status_code" , domain, None)
- instances.set("last_error_details", domain, None)
- instances.update_instance_data(domain)
- remove_pending_error(domain)
-
- if domain in pending_errors:
- # DEBUG: print("DEBUG: domain has pending error being updated:", domain)
- update_last_error(domain, pending_errors[domain])
- remove_pending_error(domain)
-
- except BaseException as e:
- print(f"ERROR: failed SQL query: domain='{domain}',exception[{type(e)}]:'{str(e)}'")
- sys.exit(255)
- else:
- # DEBUG: print("DEBUG: Updating nodeinfo for domain:", domain)
- update_last_nodeinfo(domain)
-
- # DEBUG: print("DEBUG: EXIT!")
-
-def send_bot_post(instance: str, blocks: dict):
- # DEBUG: print(f"DEBUG: instance={instance},blocks()={len(blocks)} - CALLED!")
- message = instance + " has blocked the following instances:\n\n"
- truncated = False
-
- if len(blocks) > 20:
- truncated = True
- blocks = blocks[0 : 19]
-
- for block in blocks:
- if block["reason"] == None or block["reason"] == '':
- message = message + block["blocked"] + " with unspecified reason\n"
- else:
- if len(block["reason"]) > 420:
- block["reason"] = block["reason"][0:419] + "[…]"
-
- message = message + block["blocked"] + ' for "' + block["reason"].replace("@", "@\u200b") + '"\n'
-
- if truncated:
- message = message + "(the list has been truncated to the first 20 entries)"
-
- botheaders = {**api_headers, **{"Authorization": "Bearer " + config.get("bot_token")}}
-
- req = reqto.post(
- f"{config.get('bot_instance')}/api/v1/statuses",
- data={
- "status" : message,
- "visibility" : config.get('bot_visibility'),
- "content_type": "text/plain"
- },
- headers=botheaders,
- timeout=10
- ).json()
-
- return True
-
-def get_mastodon_blocks(domain: str) -> dict:
- # DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!")
- if type(domain) != str:
- raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
- elif domain == "":
- raise ValueError(f"Parameter 'domain' cannot be empty")
-
- # DEBUG: print("DEBUG: Fetching mastodon blocks from domain:", domain)
- blocks = {
- "Suspended servers": [],
- "Filtered media" : [],
- "Limited servers" : [],
- "Silenced servers" : [],
- }
-
- try:
- doc = bs4.BeautifulSoup(
- get_response(domain, "/about", headers, (config.get("connection_timeout"), config.get("read_timeout"))).text,
- "html.parser",
- )
- except BaseException as e:
- print("ERROR: Cannot fetch from domain:", domain, e)
- update_last_error(domain, e)
- return {}
-
- for header in doc.find_all("h3"):
- header_text = tidyup_domain(header.text)
-
- if header_text in language_mapping:
- # DEBUG: print(f"DEBUG: header_text='{header_text}'")
- header_text = language_mapping[header_text]
-
- if header_text in blocks or header_text.lower() in blocks:
- # replaced find_next_siblings with find_all_next to account for instances that e.g. hide lists in dropdown menu
- for line in header.find_all_next("table")[0].find_all("tr")[1:]:
- blocks[header_text].append(
- {
- "domain": tidyup_domain(line.find("span").text),
- "hash" : tidyup_domain(line.find("span")["title"][9:]),
- "reason": tidyup_domain(line.find_all("td")[1].text),
- }
- )
-
- # DEBUG: print("DEBUG: Returning blocks for domain:", domain)
- return {
- "reject" : blocks["Suspended servers"],
- "media_removal" : blocks["Filtered media"],
- "followers_only": blocks["Limited servers"] + blocks["Silenced servers"],
- }
-
-def get_friendica_blocks(domain: str) -> dict:
- # DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!")
- if type(domain) != str:
- raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
- elif domain == "":
- raise ValueError(f"Parameter 'domain' cannot be empty")
-
- # DEBUG: print("DEBUG: Fetching friendica blocks from domain:", domain)
- blocks = []
-
- try:
- doc = bs4.BeautifulSoup(
- get_response(domain, "/friendica", headers, (config.get("connection_timeout"), config.get("read_timeout"))).text,
- "html.parser",
- )
- except BaseException as e:
- print("WARNING: Failed to fetch /friendica from domain:", domain, e)
- update_last_error(domain, e)
- return {}
-
- blocklist = doc.find(id="about_blocklist")
-
- # Prevents exceptions:
- if blocklist is None:
- # DEBUG: print("DEBUG: Instance has no block list:", domain)
- return {}
-
- for line in blocklist.find("table").find_all("tr")[1:]:
- # DEBUG: print(f"DEBUG: line='{line}'")
- blocks.append({
- "domain": tidyup_domain(line.find_all("td")[0].text),
- "reason": tidyup_domain(line.find_all("td")[1].text)
- })
-
- # DEBUG: print("DEBUG: Returning blocks() for domain:", domain, len(blocks))
- return {
- "reject": blocks
- }
-
-def get_misskey_blocks(domain: str) -> dict:
- # DEBUG: print(f"DEBUG: domain='{domain}' - CALLED!")
- if type(domain) != str:
- raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
- elif domain == "":
- raise ValueError(f"Parameter 'domain' cannot be empty")
-
- # DEBUG: print("DEBUG: Fetching misskey blocks from domain:", domain)
- blocks = {
- "suspended": [],
- "blocked" : []
- }
-
- offset = 0
- step = config.get("misskey_limit")
- while True:
- # iterating through all "suspended" (follow-only in its terminology)
- # instances page-by-page, since that troonware doesn't support
- # sending them all at once
- try:
- # DEBUG: print(f"DEBUG: Fetching offset='{offset}' from '{domain}' ...")
- if offset == 0:
- # DEBUG: print("DEBUG: Sending JSON API request to domain,step,offset:", domain, step, offset)
- fetched = post_json_api(domain, "/api/federation/instances", json.dumps({
- "sort" : "+pubAt",
- "host" : None,
- "suspended": True,
- "limit" : step
- }), {
- "Origin": domain
- })
- else:
- # DEBUG: print("DEBUG: Sending JSON API request to domain,step,offset:", domain, step, offset)
- fetched = post_json_api(domain, "/api/federation/instances", json.dumps({
- "sort" : "+pubAt",
- "host" : None,
- "suspended": True,
- "limit" : step,
- "offset" : offset - 1
- }), {
- "Origin": domain
- })
-
- # DEBUG: print("DEBUG: fetched():", len(fetched))
- if len(fetched) == 0:
- # DEBUG: print("DEBUG: Returned zero bytes, exiting loop:", domain)
- break
- elif len(fetched) != config.get("misskey_limit"):
- # DEBUG: print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config.get('misskey_limit')}'")
- offset = offset + (config.get("misskey_limit") - len(fetched))
- else:
- # DEBUG: print("DEBUG: Raising offset by step:", step)
- offset = offset + step
-
- count = 0
- for instance in fetched:
- # Is it there?
- if instance["isSuspended"] and not has_key(blocks["suspended"], "domain", instance):
- count = count + 1
- blocks["suspended"].append(
- {
- "domain": tidyup_domain(instance["host"]),
- # no reason field, nothing
- "reason": None
- }
- )
-
- # DEBUG: print(f"DEBUG: count={count}")
- if count == 0:
- # DEBUG: print(f"DEBUG: API is no more returning new instances, aborting loop!")
- break
-
- except BaseException as e:
- print("WARNING: Caught error, exiting loop:", domain, e)
- update_last_error(domain, e)
- offset = 0
- break
-
- while True:
- # same shit, different asshole ("blocked" aka full suspend)
- try:
- if offset == 0:
- # DEBUG: print("DEBUG: Sending JSON API request to domain,step,offset:", domain, step, offset)
- fetched = post_json_api(domain,"/api/federation/instances", json.dumps({
- "sort" : "+pubAt",
- "host" : None,
- "blocked": True,
- "limit" : step
- }), {
- "Origin": domain
- })
- else:
- # DEBUG: print("DEBUG: Sending JSON API request to domain,step,offset:", domain, step, offset)
- fetched = post_json_api(domain,"/api/federation/instances", json.dumps({
- "sort" : "+pubAt",
- "host" : None,
- "blocked": True,
- "limit" : step,
- "offset" : offset - 1
- }), {
- "Origin": domain
- })
-
- # DEBUG: print("DEBUG: fetched():", len(fetched))
- if len(fetched) == 0:
- # DEBUG: print("DEBUG: Returned zero bytes, exiting loop:", domain)
- break
- elif len(fetched) != config.get("misskey_limit"):
- # DEBUG: print(f"DEBUG: Fetched '{len(fetched)}' row(s) but expected: '{config.get('misskey_limit')}'")
- offset = offset + (config.get("misskey_limit") - len(fetched))
- else:
- # DEBUG: print("DEBUG: Raising offset by step:", step)
- offset = offset + step
-
- count = 0
- for instance in fetched:
- # Is it there?
- if instance["isBlocked"] and not has_key(blocks["blocked"], "domain", instance):
- count = count + 1
- blocks["blocked"].append({
- "domain": tidyup_domain(instance["host"]),
- "reason": None
- })
-
- # DEBUG: print(f"DEBUG: count={count}")
- if count == 0:
- # DEBUG: print(f"DEBUG: API is no more returning new instances, aborting loop!")
- break
-
- except BaseException as e:
- print("ERROR: Exception during POST:", domain, e)
- update_last_error(domain, e)
- offset = 0
- break
-
- # DEBUG: print(f"DEBUG: Updating last_instance_fetch for domain='{domain}' ...")
- update_last_instance_fetch(domain)
-
- # DEBUG: print("DEBUG: Returning for domain,blocked(),suspended():", domain, len(blocks["blocked"]), len(blocks["suspended"]))
- return {
- "reject" : blocks["blocked"],
- "followers_only": blocks["suspended"]
- }
-