X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=fba%2Fnetworks%2Fpleroma.py;h=435d7ce2d3099ea38f5e4ca2cc51f93474365e6b;hb=2d6883fad6ba05ee90224274ec57979f47a568d9;hp=8bc28831d7df3c8ee3f32a36f2253b4f8cc4e631;hpb=27ce6a581c57c698b0e9424fe05ea03ec9ec022c;p=fba.git diff --git a/fba/networks/pleroma.py b/fba/networks/pleroma.py index 8bc2883..435d7ce 100644 --- a/fba/networks/pleroma.py +++ b/fba/networks/pleroma.py @@ -14,7 +14,6 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -import inspect import logging import bs4 @@ -22,7 +21,6 @@ import bs4 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 @@ -30,7 +28,6 @@ from fba.helpers import tidyup from fba.http import federation from fba.http import network -from fba.models import blocks from fba.models import instances logging.basicConfig(level=logging.INFO) @@ -39,7 +36,10 @@ logger = logging.getLogger(__name__) # Language mapping X -> English language_mapping = { # English -> English - "Reject": "Suspended servers", + "limited servers" : "followers_only", + "suspended servers": "reject", + "silenced servers" : "silenced", + "filtered media" : "filtered_media", } def fetch_blocks(domain: str, nodeinfo_url: str) -> list: @@ -54,7 +54,7 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: blockdict = list() rows = None try: - logger.debug(f"Fetching nodeinfo: domain='{domain}',nodeinfo_url='{nodeinfo_url}'") + logger.debug("Fetching nodeinfo: domain='%s',nodeinfo_url='%s'", domain, nodeinfo_url) rows = federation.fetch_nodeinfo(domain, nodeinfo_url) except network.exceptions as exception: logger.warning("Exception '%s' during fetching nodeinfo from domain='%s'", type(exception), domain) @@ -67,7 +67,7 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: logger.warning("rows()=%d does not have key 'metadata', domain='%s'", 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'", len(rows["metadata"]), domain) return list() data = rows["metadata"]["federation"] @@ -96,6 +96,8 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: logger.debug("domain='%s' skipping block_level='accept'", domain) continue + block_level = utils.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: @@ -104,44 +106,16 @@ 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) - continue - elif blocked.endswith(".arpa"): - logger.debug("blocked='%s' is a reverse IP address - SKIPPED!", blocked) + logger.warning("blocked is empty after tidyup.domain(): domain='%s',block_level='%s' - SKIPPED!", domain, block_level) continue - elif blocked.endswith(".tld"): - logger.debug("blocked='%s' is a fake domain - SKIPPED!", blocked) + elif not utils.is_domain_wanted(blocked): + logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked) continue - elif blocked.count("*") > 0: - logger.debug("domain='%s' uses obfucated domains, marking ...", domain) - instances.set_has_obfucation(domain, True) - - # Obscured domain name with no hash - row = instances.deobfucate("*", blocked) - - logger.debug("row[]='%s'", type(row)) - if row is None: - logger.warning("Cannot deobfucate blocked='%s',domain='%s' - SKIPPED!", blocked, domain) - continue - - logger.debug("blocked='%s' de-obscured to '%s'", blocked, row[0]) - blocked = row[0] - elif blocked.count("?") > 0: - logger.debug("domain='%s' uses obfucated domains, marking ...", domain) - instances.set_has_obfucation(domain, True) - - # Obscured domain name with no hash - row = instances.deobfucate("?", blocked) - - logger.debug("row[]='%s'", type(row)) - if row is None: - logger.warning("Cannot deobfucate blocked='%s',domain='%s' - SKIPPED!", blocked, domain) - continue - logger.debug("blocked='%s' de-obscured to '%s'", blocked, row[0]) - blocked = row[0] + logger.debug("Invoking utils.deobfuscate_domain(%s, %s) ...", blocked, domain) + blocked = utils.deobfuscate_domain(blocked, domain) - logger.debug("blocked='%s'", blocked) + logger.debug("blocked='%s' - DEOBFUSCATED!", blocked) if not utils.is_domain_wanted(blocked): logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked) continue @@ -167,42 +141,14 @@ 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 blocked.endswith(".arpa"): - logger.debug("blocked='%s' is a reverse IP address - SKIPPED!", blocked) - continue - elif blocked.endswith(".tld"): - logger.debug("blocked='%s' is a fake domain - SKIPPED!", blocked) + elif not utils.is_domain_wanted(blocked): + logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked) continue - elif blocked.count("*") > 0: - logger.debug("domain='%s' uses obfucated domains, marking ...", domain) - instances.set_has_obfucation(domain, True) - - # Obscured domain name with no hash - row = instances.deobfucate("*", blocked) - - logger.debug("row[]='%s'", type(row)) - if row is None: - logger.warning("Cannot deobfucate blocked='%s',domain='%s' - SKIPPED!", blocked, domain) - continue - - logger.debug("blocked='%s' de-obscured to '%s'", blocked, row[0]) - blocked = row[0] - elif blocked.count("?") > 0: - logger.debug("domain='%s' uses obfucated domains, marking ...", domain) - instances.set_has_obfucation(domain, True) - # Obscured domain name with no hash - row = instances.deobfucate("?", blocked) - - logger.debug("row[]='%s'", type(row)) - if row is None: - logger.warning("Cannot deobfucate blocked='%s',domain='%s' - SKIPPED!", blocked, domain) - continue + logger.debug("Invoking utils.deobfuscate_domain(%s, %s) ...", blocked, domain) + blocked = utils.deobfuscate_domain(blocked, domain) - logger.debug("blocked='%s' de-obscured to '%s'", blocked, row[0]) - blocked = row[0] - - logger.debug("blocked='%s' - DEOBFUCATED!", blocked) + logger.debug("blocked='%s' - DEOBFUSCATED!", blocked) if not utils.is_domain_wanted(blocked): logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked) continue @@ -239,9 +185,11 @@ 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 + block_level = utils.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) @@ -253,7 +201,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)}'") @@ -262,46 +210,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 blocked.count("*") > 0: - logger.debug("domain='%s' uses obfucated domains, marking ...", domain) - instances.set_has_obfucation(domain, True) - - # Obscured domain name with no hash - row = instances.deobfucate("*", blocked) - - logger.debug("row[]='%s'", type(row)) - if row is None: - logger.warning("Cannot deobfucate blocked='%s',domain='%s' - SKIPPED!", blocked, domain) - continue - - logger.debug("blocked='%s' de-obscured to '%s'", blocked, row[0]) - blocked = row[0] - elif blocked.count("?") > 0: - logger.debug("domain='%s' uses obfucated domains, marking ...", domain) - instances.set_has_obfucation(domain, True) - - # Obscured domain name with no hash - row = instances.deobfucate("?", blocked) - - logger.debug("row[]='%s'", type(row)) - if row is None: - logger.warning("Cannot deobfucate blocked='%s',domain='%s' - SKIPPED!", blocked, domain) - continue - - logger.debug("blocked='%s' de-obscured to '%s'", blocked, row[0]) - blocked = row[0] - - logger.debug("blocked='%s' - DEOBFUCATED!", blocked) - if not utils.is_domain_wanted(blocked): + 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: - logger.debug("block[blocker]='%s',blocker='%s'", block['blocker'], blocker) - 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) @@ -313,58 +235,33 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: for blocked in rows: logger.debug("blocked='%s' - BEFORE!", blocked) blocked = tidyup.domain(blocked) - logger.debug("blocked='%s' - AFTER!", blocked) + reason = tidyup.reason(rows[blocked]["reason"]) + 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 blocked.count("*") > 0: - logger.debug("domain='%s' uses obfucated domains, marking ...", domain) - instances.set_has_obfucation(domain, True) - - # Obscured domain name with no hash - row = instances.deobfucate("*", blocked) - - logger.debug("row[]='%s'", type(row)) - if row is None: - logger.warning("Cannot deobfucate blocked='%s',domain='%s' - SKIPPED!", blocked, domain) - continue - - logger.debug("blocked='%s' de-obscured to '%s'", blocked, row[0]) - blocked = row[0] - elif blocked.count("?") > 0: - logger.debug("domain='%s' uses obfucated domains, marking ...", domain) - instances.set_has_obfucation(domain, True) - - # Obscured domain name with no hash - row = instances.deobfucate("?", blocked) - - logger.debug("row[]='%s'", type(row)) - if row is None: - logger.warning("Cannot deobfucate blocked='%s',domain='%s' - SKIPPED!", blocked, domain) - continue + elif not utils.is_domain_wanted(blocked): + logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked) + continue - logger.debug("blocked='%s' de-obscured to '%s'", blocked, row[0]) - blocked = row[0] + logger.debug("Invoking utils.deobfuscate_domain(%s, %s) ...", blocked, domain) + blocked = utils.deobfuscate_domain(blocked, domain) - logger.debug("blocked='%s'", blocked) + 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("block[blocker]='%s',blocker='%s'", block['blocker'], blocker) - 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) @@ -374,60 +271,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) - - if blocked == "": - logger.warning("blocked is empty after tidyup.domain(): domain='%s',block_level='%s'", domain, block_level) - continue - elif blocked.count("*") > 0: - logger.debug("domain='%s' uses obfucated domains, marking ...", domain) - instances.set_has_obfucation(domain, True) - - # Obscured domain name with no hash - row = instances.deobfucate("*", blocked) - - logger.debug("row[]='%s'", type(row)) - if row is None: - logger.warning("Cannot deobfucate blocked='%s',domain='%s' - SKIPPED!", blocked, domain) - continue - - logger.debug("blocked='%s' de-obscured to '%s'", blocked, row[0]) - blocked = row[0] - elif blocked.count("?") > 0: - logger.debug("domain='%s' uses obfucated domains, marking ...", domain) - instances.set_has_obfucation(domain, True) - # Obscured domain name with no hash - row = instances.deobfucate("?", blocked) - - logger.debug("row[]='%s'", type(row)) - if row is None: - logger.warning("Cannot deobfucate blocked='%s',domain='%s' - SKIPPED!", blocked, domain) - continue - - logger.debug("blocked='%s' de-obscured to '%s'", blocked, row[0]) - blocked = row[0] + 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("blocked='%s' - DEOBFUCATED!", blocked) - if 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 utils.is_domain_wanted(block["blocked"]): + logger.debug("block[blocked]='%s' is not wanted - SKIPPED!", block["blocked"]) continue - 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, }) @@ -435,7 +298,7 @@ def fetch_blocks(domain: str, nodeinfo_url: str) -> list: return blockdict def fetch_blocks_from_about(domain: str) -> dict: - logger.debug("domain(%d)='%s' - CALLED!", len(domain), domain) + logger.debug("domain='%s' - CALLED!", domain) domain_helper.raise_on(domain) logger.debug("Fetching mastodon blocks from domain='%s'", domain) @@ -453,7 +316,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 @@ -475,10 +338,10 @@ def fetch_blocks_from_about(domain: str) -> dict: break blocklist = { - "Suspended servers": [], - "Filtered media" : [], - "Limited servers" : [], - "Silenced servers" : [], + "reject" : [], + "filtered_media": [], + "followers_only": [], + "silenced" : [], } logger.debug("doc[]='%s'", type(doc)) @@ -486,32 +349,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("headers[]='%s'", type(headers)) + if headers is None: + logger.warning("Cannot fetch any /about pages for domain='%s' - EXIT!", domain) + return list() - 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.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