from fba import database
from fba import utils
+from fba.helpers import blacklist
from fba.helpers import config
from fba.helpers import domain as domain_helper
from fba.helpers import tidyup
-from fba.http import federation
from fba.http import network
+from fba.http import nodeinfo
+from fba.models import blocks
from fba.models import instances
logging.basicConfig(level=logging.INFO)
# Language mapping X -> English
language_mapping = {
# English -> English
+ "filtered media" : "filtered_media",
"limited servers" : "followers_only",
+ "followers-only" : "followers_only",
+ "media removal" : "media_removal",
+ "media_removal" : "media_removal",
+ "media force-set as sensitive": "media_nsfw",
+ "nsfw" : "media_nsfw",
+ "reject" : "reject",
"suspended servers": "reject",
"silenced servers" : "silenced",
- "filtered media" : "filtered_media",
+ "removal from \"the whole known network\" timeline": "federated_timeline_removal",
}
-def fetch_blocks(domain: str, nodeinfo_url: str) -> list:
- logger.debug("domain='%s',nodeinfo_url='%s' - CALLED!", domain, nodeinfo_url)
+def fetch_blocks(domain: str) -> list:
+ logger.debug("domain='%s' - CALLED!", domain)
domain_helper.raise_on(domain)
- if not isinstance(nodeinfo_url, str):
- raise ValueError(f"Parameter nodeinfo_url[]='{type(nodeinfo_url)}' is not 'str'")
- elif nodeinfo_url == "":
- raise ValueError("Parameter 'nodeinfo_url' is empty")
+ if blacklist.is_blacklisted(domain):
+ raise Exception(f"domain='{domain}' is blacklisted but function is invoked.")
+ elif not instances.is_registered(domain):
+ raise Exception(f"domain='{domain}' is not registered but function is invoked.")
blockdict = list()
rows = None
+
try:
- logger.debug("Fetching nodeinfo: domain='%s',nodeinfo_url='%s'", domain, nodeinfo_url)
- rows = federation.fetch_nodeinfo(domain, nodeinfo_url)
+ logger.debug("Fetching nodeinfo: domain='%s'", domain)
+ rows = nodeinfo.fetch(domain, update_mode=False)
+
+ if "error_message" in rows:
+ logger.warning("Error message '%s' during fetching nodeinfo for domain='%s'", rows["error_message"], domain)
+ instances.set_last_error(domain, rows)
+ instances.update(domain)
+
+ logger.debug("Returning empty list ... - EXIT!")
+ return list()
+ elif "exception" in rows:
+ logger.warning("Exception '%s' during fetching nodeinfo for domain='%s' - EXIT!", type(rows["exception"]), domain)
+ return list()
+ elif "json" in rows:
+ logger.debug("rows[json] found for domain='%s'", domain)
+ rows = rows["json"]
+
except network.exceptions as exception:
logger.warning("Exception '%s' during fetching nodeinfo from domain='%s'", type(exception), domain)
instances.set_last_error(domain, exception)
if rows is None:
- logger.warning("Could not fetch nodeinfo from domain='%s'", domain)
+ logger.warning("Could not fetch nodeinfo from domain='%s' - EXIT!", domain)
return list()
elif "metadata" not in rows:
- logger.warning("rows()=%d does not have key 'metadata', domain='%s'", len(rows), domain)
+ logger.warning("rows()=%d does not have key 'metadata', domain='%s' - EXIT!", len(rows), domain)
return list()
elif "federation" not in rows["metadata"]:
- logger.warning("rows()=%d does not have key 'federation', domain='%s'", len(rows["metadata"]), domain)
+ logger.warning("rows()=%d does not have key 'federation', domain='%s' - EXIT!", len(rows["metadata"]), domain)
return list()
data = rows["metadata"]["federation"]
}
).items():
logger.debug("block_level='%s', blocklist()=%d", block_level, len(blocklist))
- block_level = tidyup.domain(block_level)
+ block_level = tidyup.domain(block_level) if block_level != "" else None
logger.debug("block_level='%s' - AFTER!", block_level)
if block_level == "":
logger.debug("domain='%s' skipping block_level='accept'", domain)
continue
- block_level = utils.alias_block_level(block_level)
+ block_level = blocks.alias_block_level(block_level)
logger.debug("Checking %d entries from domain='%s',block_level='%s' ...", len(blocklist), domain, block_level)
if len(blocklist) > 0:
for blocked in blocklist:
logger.debug("blocked='%s' - BEFORE!", blocked)
- blocked = tidyup.domain(blocked)
+ blocked = tidyup.domain(blocked) if blocked != "" else None
logger.debug("blocked='%s' - AFTER!", blocked)
- if blocked == "":
- logger.warning("blocked is empty after tidyup.domain(): domain='%s',block_level='%s' - SKIPPED!", domain, block_level)
- continue
- elif not utils.is_domain_wanted(blocked):
- logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked)
+ if blocked is None or blocked == "":
+ logger.warning("blocked='%s' is empty after tidyup.domain(): domain='%s',block_level='%s' - SKIPPED!", blocked, domain, block_level)
continue
- logger.debug("Invoking utils.deobfuscate_domain(%s, %s) ...", blocked, domain)
- blocked = utils.deobfuscate_domain(blocked, domain)
+ logger.debug("Invoking utils.deobfuscate(%s, %s) ...", blocked, domain)
+ blocked = utils.deobfuscate(blocked, domain)
logger.debug("blocked='%s' - DEOBFUSCATED!", blocked)
- if not utils.is_domain_wanted(blocked):
+ if blocked is None or blocked == "":
+ logger.warning("instance[host]='%s' is None or empty after tidyup.domain() - SKIPPED!", instance["host"])
+ continue
+ elif not domain_helper.is_wanted(blocked):
logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked)
continue
for blocked in data["quarantined_instances"]:
logger.debug("blocked='%s' - BEFORE!", blocked)
- blocked = tidyup.domain(blocked)
- logger.debug("blocked='%s' - AFTER!", blocked)
+ blocked = tidyup.domain(blocked) if blocked != "" else None
- if blocked == "":
+ logger.debug("blocked='%s' - AFTER!", blocked)
+ if blocked is None or blocked == "":
logger.warning("blocked is empty after tidyup.domain(): domain='%s',block_level='%s'", domain, block_level)
continue
- elif not utils.is_domain_wanted(blocked):
- logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked)
- continue
-
- logger.debug("Invoking utils.deobfuscate_domain(%s, %s) ...", blocked, domain)
- blocked = utils.deobfuscate_domain(blocked, domain)
-
- logger.debug("blocked='%s' - DEOBFUSCATED!", blocked)
- if not utils.is_domain_wanted(blocked):
+ elif not domain_helper.is_wanted(blocked):
logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked)
continue
}
).items():
logger.debug("block_level='%s', info.items()=%d", block_level, len(info.items()))
- block_level = tidyup.domain(block_level)
+ block_level = tidyup.domain(block_level) if block_level != "" else None
logger.debug("block_level='%s' - AFTER!", block_level)
- if block_level == "":
- logger.warning("block_level is now empty!")
+ if block_level is None or block_level == "":
+ logger.warning("block_level='%s' is now empty!", block_level)
continue
elif block_level == "accept":
logger.debug("domain='%s': Skipping block_level='%s' ...", domain, block_level)
continue
- block_level = utils.alias_block_level(block_level)
+ block_level = blocks.alias_block_level(block_level)
logger.debug("Checking %d entries from domain='%s',block_level='%s' ...", len(info.items()), domain, block_level)
for blocked, reason in info.items():
logger.debug("blocked='%s',reason[%s]='%s' - BEFORE!", blocked, type(reason), reason)
- blocked = tidyup.domain(blocked)
+ blocked = tidyup.domain(blocked) if blocked != "" else None
logger.debug("blocked='%s' - AFTER!", blocked)
if isinstance(reason, str):
if blocked == "":
logger.warning("blocked is empty after tidyup.domain(): domain='%s',block_level='%s'", domain, block_level)
continue
- elif not utils.is_domain_wanted(blocked):
- logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked)
- continue
-
- logger.debug("Invoking utils.deobfuscate_domain(%s, %s) ...", blocked, domain)
- blocked = utils.deobfuscate_domain(blocked, domain)
- logger.debug("blocked='%s' - DEOBFUSCATED!", blocked)
logger.debug("Checking %d blockdict records ...", len(blockdict))
for block in blockdict:
rows = data["quarantined_instances_info"]["quarantined_instances"]
for blocked in rows:
logger.debug("blocked='%s' - BEFORE!", blocked)
- blocked = tidyup.domain(blocked)
reason = tidyup.reason(rows[blocked]["reason"])
+ blocked = tidyup.domain(blocked) if blocked != "" else None
logger.debug("blocked='%s',reason='%s' - AFTER!", blocked, reason)
if blocked not in rows or "reason" not in rows[blocked]:
elif blocked == "":
logger.warning("blocked is empty after tidyup.domain(): domain='%s',block_level='%s'", domain, block_level)
continue
- elif not utils.is_domain_wanted(blocked):
- logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked)
- continue
-
- logger.debug("Invoking utils.deobfuscate_domain(%s, %s) ...", blocked, domain)
- blocked = utils.deobfuscate_domain(blocked, domain)
-
- logger.debug("blocked='%s' - DEOBFUSCATED!", blocked)
- if not utils.is_domain_wanted(blocked):
- logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked)
- continue
logger.debug("Checking %d blockdict records ...", len(blockdict))
for block in blockdict:
logger.debug("rows[%s]()=%d'", type(rows), len(rows))
for block in rows:
- logger.debug("Invoking utils.deobfuscate_domain(%s, %s) ...", block["blocked"], domain)
- block["blocked"] = utils.deobfuscate_domain(block["blocked"], domain)
-
- logger.debug("block[blocked]='%s' - DEOBFUSCATED!", block["blocked"])
- if not utils.is_domain_wanted(block["blocked"]):
- logger.debug("block[blocked]='%s' is not wanted - SKIPPED!", block["blocked"])
- continue
-
logger.debug("Appending blocker='%s',block[blocked]='%s',block[reason]='%s',block_level='%s' ...",domain, block["blocked"], block["reason"], block_level)
blockdict.append({
"blocker" : domain,
logger.debug("domain='%s' - CALLED!", domain)
domain_helper.raise_on(domain)
+ if blacklist.is_blacklisted(domain):
+ raise Exception(f"domain='{domain}' is blacklisted but function is invoked.")
+ elif not instances.is_registered(domain):
+ raise Exception(f"domain='{domain}' is not registered but function is invoked.")
+
logger.debug("Fetching mastodon blocks from domain='%s'", domain)
doc = None
for path in ["/instance/about/index.html"]:
"filtered_media": [],
"followers_only": [],
"silenced" : [],
+ "media_nsfw" : [],
+ "media_removal" : [],
+ "federated_timeline_removal": [],
}
logger.debug("doc[]='%s'", type(doc))
logger.debug("Found block_level='%s', importing domain blocks ...", block_level)
for line in header.find_next("table").find_all("tr")[1:]:
logger.debug("line[]='%s'", type(line))
- blocked = tidyup.domain(line.find_all("td")[0].text)
+ blocked = line.find_all("td")[0].text
+ logger.debug("blocked='%s'", blocked)
+
+ blocked = tidyup.domain(blocked) if blocked != "" else None
reason = tidyup.reason(line.find_all("td")[1].text)
+ logger.debug("blocked='%s',reason='%s' - AFTER!", blocked, reason)
+
+ if blocked is None or blocked == "":
+ logger.debug("domain='%s',block_level='%s': blocked is empty - SKIPPED!", domain, block_level)
+ continue
+ elif not domain_helper.is_wanted(blocked):
+ logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked)
+ continue
logger.debug("Appending block_level='%s',blocked='%s',reason='%s' ...", block_level, blocked, reason)
blocklist[block_level].append({