X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=fba%2Fnetworks%2Fpleroma.py;h=5a52f9cd58e823cdb37497d01489a13d8888dad2;hb=bd36d58fadcf377a7982103ca4d7e5c4376ef463;hp=f8fb931f9c8f69c628eac9a661bc1ccc364818e1;hpb=0aafe5700db84daee4c5ff1ea789e4e0e1037610;p=fba.git diff --git a/fba/networks/pleroma.py b/fba/networks/pleroma.py index f8fb931..5a52f9c 100644 --- a/fba/networks/pleroma.py +++ b/fba/networks/pleroma.py @@ -25,9 +25,10 @@ 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) @@ -36,7 +37,17 @@ logger = logging.getLogger(__name__) # Language mapping X -> English language_mapping = { # English -> English - "Reject": "Suspended servers", + "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", + "removal from \"the whole known network\" timeline": "federated_timeline_removal", } def fetch_blocks(domain: str, nodeinfo_url: str) -> list: @@ -44,7 +55,7 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: domain_helper.raise_on(domain) if not isinstance(nodeinfo_url, str): - raise ValueError(f"Parameter nodeinfo_url[]='{type(nodeinfo_url)}' is not 'str'") + raise ValueError(f"Parameter nodeinfo_url[]='{type(nodeinfo_url)}' is not of type 'str'") elif nodeinfo_url == "": raise ValueError("Parameter 'nodeinfo_url' is empty") @@ -52,19 +63,33 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: rows = None try: logger.debug("Fetching nodeinfo: domain='%s',nodeinfo_url='%s'", domain, nodeinfo_url) - rows = federation.fetch_nodeinfo(domain, nodeinfo_url) + rows = nodeinfo.fetch_nodeinfo(domain, nodeinfo_url) + + if "error_message" in rows: + logger.warning("Error message '%s' during fetching nodeinfo for domain='%s',nodeinfo_url='%s'", rows["error_message"], domain, nodeinfo_url) + instances.set_last_error(domain, rows) + + logger.debug("Returning empty list ... - EXIT!") + return list() + elif "exception" in rows: + logger.warning("Exception '%s' during fetching nodeinfo for domain='%s',nodeinfo_url='%s' - EXIT!", type(rows["exception"]), domain, nodeinfo_url) + return list() + elif "json" in rows: + logger.debug("rows[json] found for domain='%s',nodeinfo_url='%s'", domain, nodeinfo_url) + 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"] @@ -92,12 +117,8 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: elif block_level == "accept": logger.debug("domain='%s' skipping block_level='accept'", domain) continue - elif block_level == "suspend": - logger.debug("domain='%s', mapping 'suspend' to 'suspended'", domain) - block_level = "suspended" - elif block_level == "silence": - logger.debug("domain='%s', mapping 'silence' to 'silenced'", domain) - block_level = "silenced" + + 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: @@ -107,17 +128,17 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: logger.debug("blocked='%s' - AFTER!", blocked) if blocked == "": - logger.warning("blocked is empty after tidyup.domain(): domain='%s',block_level='%s'", domain, block_level) + 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): + elif not domain_helper.is_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("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 not domain_helper.is_wanted(blocked): logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked) continue @@ -142,15 +163,15 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: 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): + elif not domain_helper.is_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("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 not domain_helper.is_wanted(blocked): logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked) continue @@ -186,14 +207,10 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: logger.warning("block_level is now empty!") continue elif block_level == "accept": - logger.debug("domain='%s' skipping block_level='accept'", domain) + logger.debug("domain='%s': Skipping block_level='%s' ...", domain, block_level) continue - elif block_level == "suspend": - logger.debug("domain='%s', mapping 'suspend' to 'suspended'", domain) - block_level = "suspended" - elif block_level == "silence": - logger.debug("domain='%s', mapping 'silence' to 'silenced'", domain) - block_level = "silenced" + + 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(): @@ -206,7 +223,7 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: reason = tidyup.reason(reason) elif isinstance(reason, dict) and "reason" in reason: logger.debug("reason[] is a dict") - reason = tidyup.reason(reason["reason"]) + reason = tidyup.reason(reason["reason"]) if isinstance(reason["reason"], str) else None elif reason is not None: raise ValueError(f"Cannot handle reason[]='{type(reason)}'") @@ -215,20 +232,20 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: 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): + elif not domain_helper.is_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("Invoking utils.deobfuscate(%s, %s) ...", blocked, domain) + blocked = utils.deobfuscate(blocked, domain) logger.debug("blocked='%s' - DEOBFUSCATED!", blocked) logger.debug("Checking %d blockdict records ...", len(blockdict)) for block in blockdict: - logger.debug("block[blocked]='%s',blocked='%s'", block['blocked'], blocked) - if block['blocked'] == blocked: - logger.debug("Updating reason='%s' for blocker='%s'", reason, block['blocked']) - block['reason'] = reason + logger.debug("block[blocked]='%s',blocked='%s'", block["blocked"], blocked) + if block["blocked"] == blocked: + logger.debug("Updating reason='%s' for blocker='%s'", reason, block["blocked"]) + block["reason"] = reason elif "quarantined_instances_info" in data and "quarantined_instances" in data["quarantined_instances_info"]: logger.debug("Found 'quarantined_instances_info' in JSON response: domain='%s'", domain) @@ -239,34 +256,34 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: rows = data["quarantined_instances_info"]["quarantined_instances"] for blocked in rows: logger.debug("blocked='%s' - BEFORE!", blocked) + reason = tidyup.reason(rows[blocked]["reason"]) blocked = tidyup.domain(blocked) - logger.debug("blocked='%s' - AFTER!", blocked) + logger.debug("blocked='%s',reason='%s' - AFTER!", blocked, reason) if blocked not in rows or "reason" not in rows[blocked]: logger.warning("Cannot find blocked='%s' in rows()=%d,domain='%s' - BREAK!", blocked, len(rows), domain) break - - reason = rows[blocked]["reason"] - logger.debug("reason='%s'", reason) - - if blocked == "": + elif blocked == "": logger.warning("blocked is empty after tidyup.domain(): domain='%s',block_level='%s'", domain, block_level) continue + elif not domain_helper.is_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("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 not domain_helper.is_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("block[blocked]='%s',blocked='%s'", block['blocked'], blocked) - if block['blocked'] == blocked: - logger.debug("Updating reason='%s' for blocker='%s'", reason, block['blocked']) - block['reason'] = reason + logger.debug("block[blocked]='%s',blocked='%s'", block["blocked"], blocked) + if block["blocked"] == blocked: + logger.debug("Updating reason='%s' for blocker='%s'", reason, block["blocked"]) + block["reason"] = reason else: logger.warning("Cannot find 'mrf_simple_info' or 'quarantined_instances_info' in JSON reply: domain='%s'", domain) @@ -276,34 +293,26 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: logger.debug("blocklist()=%d", len(blocklist)) if len(blocklist) > 0: - logger.info("Checking %d record(s) ...", len(blocklist)) + logger.info("Checking %d different blocklists ...", len(blocklist)) for block_level in blocklist: logger.debug("block_level='%s'", block_level) rows = blocklist[block_level] logger.debug("rows[%s]()=%d'", type(rows), len(rows)) - for record in rows: - logger.debug("record[]='%s'", type(record)) - blocked = tidyup.domain(record["blocked"]) - reason = tidyup.reason(record["reason"]) - logger.debug("blocked='%s',reason='%s' - AFTER!", blocked, reason) + for block in rows: + logger.debug("Invoking utils.deobfuscate(%s, %s) ...", block["blocked"], domain) + block["blocked"] = utils.deobfuscate(block["blocked"], domain) - 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.warning("blocked='%s' is not wanted - SKIPPED!", blocked) + logger.debug("block[blocked]='%s' - DEOBFUSCATED!", block["blocked"]) + if not domain_helper.is_wanted(block["blocked"]): + logger.debug("block[blocked]='%s' is not wanted - SKIPPED!", block["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("Appending blocker='%s',blocked='%s',reason='%s',block_level='%s' ...",domain, blocked, reason, block_level) + 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, - "blocked" : blocked, - "reason" : reason, + "blocked" : block["blocked"], + "reason" : block["reason"], "block_level": block_level, }) @@ -329,7 +338,7 @@ def fetch_blocks_from_about(domain: str) -> dict: (config.get("connection_timeout"), config.get("read_timeout")) ) - logger.debug("response.ok='%s',response.status_code='%d',response.text()=%d", response.ok, response.status_code, len(response.text)) + logger.debug("response.ok='%s',response.status_code=%d,response.text()=%d", response.ok, response.status_code, len(response.text)) if not response.ok or response.text.strip() == "": logger.warning("path='%s' does not exist on domain='%s' - SKIPPED!", path, domain) continue @@ -351,10 +360,13 @@ def fetch_blocks_from_about(domain: str) -> dict: break blocklist = { - "Suspended servers": [], - "Filtered media" : [], - "Limited servers" : [], - "Silenced servers" : [], + "reject" : [], + "filtered_media": [], + "followers_only": [], + "silenced" : [], + "media_nsfw" : [], + "media_removal" : [], + "federated_timeline_removal": [], } logger.debug("doc[]='%s'", type(doc)) @@ -362,32 +374,41 @@ def fetch_blocks_from_about(domain: str) -> dict: logger.warning("Cannot fetch any /about pages for domain='%s' - EXIT!", domain) return list() - for header in doc.find_all("h2"): - header_text = tidyup.reason(header.text) + headers = doc.find_all("h2") - logger.debug("header_text='%s' - BEFORE!", header_text) - if header_text in language_mapping: - logger.debug("header_text='%s' - FOUND!", header_text) - header_text = language_mapping[header_text] + logger.debug("headers[]='%s'", type(headers)) + if headers is None: + logger.warning("Cannot fetch any /about pages for domain='%s' - EXIT!", domain) + return list() + + logger.info("Checking %d headers ...", len(headers)) + for header in headers: + logger.debug("header[%s]='%s'", type(header), header) + block_level = tidyup.reason(header.text).lower() + + logger.debug("block_level='%s' - BEFORE!", block_level) + if block_level in language_mapping: + logger.debug("block_level='%s' - FOUND!", block_level) + block_level = language_mapping[block_level].lower() else: - logger.warning("header_text='%s' not found in language mapping table", header_text) + logger.warning("block_level='%s' not found in language mapping table", block_level) - logger.debug("header_text='%s - AFTER!'", header_text) - if header_text in blocklist or header_text.lower() in blocklist: + logger.debug("block_level='%s - AFTER!'", block_level) + if block_level in blocklist: # replaced find_next_siblings with find_all_next to account for instances that e.g. hide lists in dropdown menu - logger.debug("Found header_text='%s', importing domain blocks ...", header_text) + 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)) - blocklist[header_text].append({ - "blocked": tidyup.domain(line.find_all("td")[0].text), - "reason" : tidyup.reason(line.find_all("td")[1].text), + blocked = tidyup.domain(line.find_all("td")[0].text) + reason = tidyup.reason(line.find_all("td")[1].text) + + logger.debug("Appending block_level='%s',blocked='%s',reason='%s' ...", block_level, blocked, reason) + blocklist[block_level].append({ + "blocked": blocked, + "reason" : reason, }) else: - logger.warning("header_text='%s' not found in blocklist()=%d", header_text, len(blocklist)) + logger.warning("block_level='%s' not found in blocklist()=%d", block_level, len(blocklist)) logger.debug("Returning blocklist for domain='%s' - EXIT!", domain) - return { - "reject" : blocklist["Suspended servers"], - "media_removal" : blocklist["Filtered media"], - "followers_only": blocklist["Limited servers"] + blocklist["Silenced servers"], - } + return blocklist