from fba import blacklist
from fba import blocks
from fba import fba
+from fba import instances
def fetch_blocks(domain: str, origin: str, nodeinfo_url: str):
- print(f"DEBUG: domain='{domain}',origin='{origin}',nodeinfo_url='{nodeinfo_url}' - CALLED!")
- if type(domain) != str:
+ # DEBUG: print(f"DEBUG: domain='{domain}',origin='{origin}',nodeinfo_url='{nodeinfo_url}' - CALLED!")
+ if not isinstance(domain, str):
raise ValueError(f"Parameter domain[]={type(domain)} is not 'str'")
elif domain == "":
- raise ValueError(f"Parameter 'domain' is empty")
- elif type(origin) != str and origin != None:
+ raise ValueError("Parameter 'domain' is empty")
+ elif not isinstance(origin, str) and origin is not None:
raise ValueError(f"Parameter origin[]={type(origin)} is not 'str'")
elif origin == "":
- raise ValueError(f"Parameter 'origin' is empty")
- elif type(nodeinfo_url) != str:
+ raise ValueError("Parameter 'origin' is empty")
+ elif not isinstance(nodeinfo_url, str):
raise ValueError(f"Parameter nodeinfo_url[]={type(nodeinfo_url)} is not 'str'")
elif nodeinfo_url == "":
- raise ValueError(f"Parameter 'nodeinfo_url' is empty")
+ raise ValueError("Parameter 'nodeinfo_url' is empty")
try:
# Blocks
blockdict = list()
- json = fba.fetch_nodeinfo(domain, nodeinfo_url)
+ rows = fba.fetch_nodeinfo(domain, nodeinfo_url)
- if json is None:
+ if rows is None:
print("WARNING: Could not fetch nodeinfo from domain:", domain)
return
- elif not "metadata" in json:
- print(f"WARNING: json()={len(json)} does not have key 'metadata', domain='{domain}'")
+ elif not "metadata" in rows:
+ print(f"WARNING: rows()={len(rows)} does not have key 'metadata', domain='{domain}'")
return
- elif not "federation" in json["metadata"]:
- print(f"WARNING: json()={len(json['metadata'])} does not have key 'federation', domain='{domain}'")
+ elif not "federation" in rows["metadata"]:
+ print(f"WARNING: rows()={len(rows['metadata'])} does not have key 'federation', domain='{domain}'")
return
# DEBUG: print("DEBUG: Updating nodeinfo:", domain)
- fba.update_last_nodeinfo(domain)
+ instances.update_last_nodeinfo(domain)
- federation = json["metadata"]["federation"]
+ federation = rows["metadata"]["federation"]
if "enabled" in federation:
# DEBUG: print("DEBUG: Instance has no block list to analyze:", domain)
searchres = fba.cursor.fetchone()
# DEBUG: print("DEBUG: searchres[]:", type(searchres))
- if searchres == None:
+ if searchres is None:
print(f"WARNING: Cannot deobsfucate blocked='{blocked}' - SKIPPED!")
continue
if not validators.domain(blocked):
print(f"WARNING: blocked='{blocked}',software='pleroma' is not a valid domain name - skipped!")
continue
- elif not fba.is_instance_registered(blocked):
+ elif not instances.is_registered(blocked):
# DEBUG: print(f"DEBUG: Domain blocked='{blocked}' wasn't found, adding ..., domain='{domain}',origin='{origin}',nodeinfo_url='{nodeinfo_url}'")
instances.add(blocked, domain, inspect.currentframe().f_code.co_name, nodeinfo_url)
# DEBUG: print(f"DEBUG: Checking {len(info.items())} entries from domain='{domain}',software='pleroma',block_level='{block_level}' ...")
for blocked, reason in info.items():
- # DEBUG: print("DEBUG: BEFORE blocked:", blocked)
+ # DEBUG: print(f"DEBUG: blocked='{blocked}',reason='{reason}' - BEFORE!")
blocked = fba.tidyup_domain(blocked)
- # DEBUG: print("DEBUG: AFTER blocked:", blocked)
+ reason = fba.tidyup_reason(reason) if reason is not None and reason != "" else None
+ # DEBUG: print(f"DEBUG: blocked='{blocked}',reason='{reason}' - AFTER!")
if blocked == "":
print("WARNING: blocked is empty after fba.tidyup_domain():", domain, block_level)
continue
+ elif not validators.domain(blocked):
+ print(f"WARNING: blocked='{blocked}',software='pleroma' is not a valid domain name - skipped!")
+ continue
elif blacklist.is_blacklisted(blocked):
# DEBUG: print(f"DEBUG: blocked='{blocked}' is blacklisted - skipping!")
continue
)
searchres = fba.cursor.fetchone()
- if searchres == None:
+ if searchres is 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='pleroma' is not a valid domain name - skipped!")
- continue
# DEBUG: print("DEBUG: Looking up instance by domain:", blocked)
- if not validators.domain(blocked):
- print(f"WARNING: blocked='{blocked}',software='pleroma' is not a valid domain name - skipped!")
- continue
- elif not fba.is_instance_registered(blocked):
+ if not instances.is_registered(blocked):
# DEBUG: print(f"DEBUG: Domain blocked='{blocked}' wasn't found, adding ..., domain='{domain}',origin='{origin}',nodeinfo_url='{nodeinfo_url}'")
instances.add(blocked, domain, inspect.currentframe().f_code.co_name, nodeinfo_url)
# DEBUG: print("DEBUG: Updating block reason:", domain, blocked, reason["reason"])
blocks.update_reason(reason["reason"], domain, blocked, block_level)
- # DEBUG: print(f"DEBUG: blockdict()={count(blockdict)")
+ # DEBUG: print(f"DEBUG: blockdict()={len(blockdict)}")
for entry in blockdict:
if entry["blocked"] == blocked:
# DEBUG: print("DEBUG: Updating entry reason:", blocked)
entry["reason"] = reason["reason"]
fba.connection.commit()
- except Exception as e:
- print(f"ERROR: domain='{domain}',software='pleroma',exception[{type(e)}]:'{str(e)}'")
+ except BaseException as exception:
+ print(f"ERROR: domain='{domain}',software='pleroma',exception[{type(exception)}]:'{str(exception)}'")
# DEBUG: print("DEBUG: EXIT!")
projects / fba.git / blobdiff