From 213d26eb30a0b472ce4b0cfaf95a2e96e6cbc28e Mon Sep 17 00:00:00 2001 From: =?utf8?q?Roland=20H=C3=A4der?= Date: Thu, 30 Nov 2023 04:03:02 +0100 Subject: [PATCH] Continued: - `foo is None or foo == ""` can be simplified to `foo in [None, ""]` - added more blacklist check - improved/added some debug messages --- fba/commands.py | 27 ++++++++++++++------------- fba/helpers/processing.py | 4 +++- fba/http/federation.py | 8 ++++---- fba/http/network.py | 19 ++++++++++++++----- fba/networks/friendica.py | 2 +- fba/networks/lemmy.py | 8 ++++---- fba/networks/mastodon.py | 2 +- fba/networks/misskey.py | 6 +++--- fba/networks/pleroma.py | 8 ++++---- 9 files changed, 48 insertions(+), 36 deletions(-) diff --git a/fba/commands.py b/fba/commands.py index f24a888..88095c3 100644 --- a/fba/commands.py +++ b/fba/commands.py @@ -148,7 +148,7 @@ def fetch_pixelfed_api(args: argparse.Namespace) -> int: if "domain" not in row: logger.warning("row='%s' does not contain element 'domain' - SKIPPED!", row) continue - elif row["domain"] is None or row["domain"] == "": + elif row["domain"] in [None, ""]: logger.debug("row[domain]='%s' is empty - SKIPPED!", row["domain"]) continue @@ -224,7 +224,7 @@ def fetch_bkali(args: argparse.Namespace) -> int: if "domain" not in entry: logger.warning("entry()=%d does not contain 'domain' - SKIPPED!", len(entry)) continue - elif entry["domain"] is None or entry["domain"] == "": + elif entry["domain"] in [None, ""]: logger.debug("entry[domain]='%s' is empty - SKIPPED!", entry["domain"]) continue elif not domain_helper.is_wanted(entry["domain"]): @@ -367,7 +367,7 @@ def fetch_blocks(args: argparse.Namespace) -> int: block["reason"] = tidyup.reason(block["reason"]) if block["reason"] is not None and block["reason"] != "" else None logger.debug("blocked='%s',reason='%s' - AFTER!", block["blocked"], block["reason"]) - if block["blocked"] is None or block["blocked"] == "": + if block["blocked"] in [None, ""]: logger.warning("block[blocked]='%s' is empty, blocker='%s'", block["blocked"], blocker) continue elif block["blocked"].endswith(".onion"): @@ -418,7 +418,7 @@ def fetch_blocks(args: argparse.Namespace) -> int: nodeinfo_url = row["nodeinfo_url"] logger.debug("Looking up instance by domain, blocked='%s'", block["blocked"]) - if block["blocked"] is None or block["blocked"] == "": + if block["blocked"] in [None, ""]: logger.debug("block[blocked]='%s' is empty - SKIPPED!", block["blocked"]) continue @@ -550,7 +550,7 @@ def fetch_observer(args: argparse.Namespace) -> int: domain = tidyup.domain(domain) if domain not in [None, ""] else None logger.debug("domain='%s' - AFTER!", domain) - if domain is None or domain == "": + if domain in [None, ""]: logger.debug("domain='%s' is empty after tidyup.domain() - SKIPPED!", domain) continue @@ -807,7 +807,7 @@ def fetch_fba_rss(args: argparse.Namespace) -> int: domain = tidyup.domain(domain) if domain not in[None, ""] else None logger.debug("domain='%s' - AFTER!", domain) - if domain is None or domain == "": + if domain in [None, ""]: logger.debug("domain='%s' is empty after tidyup.domain() - SKIPPED!", domain) continue @@ -892,7 +892,7 @@ def fetch_fbabot_atom(args: argparse.Namespace) -> int: domain = tidyup.domain(href) if href not in [None, ""] else None logger.debug("domain='%s' - AFTER!", domain) - if domain is None or domain == "": + if domain in [None, ""]: logger.debug("domain='%s' is empty after tidyup.domain() - SKIPPED!", domain) continue @@ -1082,7 +1082,7 @@ def fetch_txt(args: argparse.Namespace) -> int: domain = tidyup.domain(domain) if domain not in[None, ""] else None logger.debug("domain='%s' - AFTER!", domain) - if domain is None or domain == "": + if domain in [None, ""]: logger.debug("domain='%s' is empty after tidyup.domain() - SKIPPED!", domain) continue elif not domain_helper.is_wanted(domain): @@ -1138,7 +1138,7 @@ def fetch_fedipact(args: argparse.Namespace) -> int: domain = tidyup.domain(row.contents[0]) if row.contents[0] not in [None, ""] else None logger.debug("domain='%s' - AFTER!", domain) - if domain is None or domain == "": + if domain in [None, ""]: logger.debug("domain='%s' is empty after tidyup.domain() - SKIPPED!", domain) continue @@ -1284,6 +1284,7 @@ def recheck_obfuscation(args: argparse.Namespace) -> int: logger.debug("blocking()=%d", len(blocking)) if len(blocking) == 0: + logger.debug("Empty blocking list, trying individual fetch_blocks() for row[software]='%s' ...", row["software"]) if row["software"] == "pleroma": logger.debug("domain='%s',software='%s'", row["domain"], row["software"]) blocking = pleroma.fetch_blocks(row["domain"]) @@ -1440,7 +1441,7 @@ def fetch_fedilist(args: argparse.Namespace) -> int: domain = tidyup.domain(row["hostname"]) if row["hostname"] not in [None, ""] else None logger.debug("domain='%s' - AFTER!", domain) - if domain is None or domain == "": + if domain in [None, ""]: logger.debug("domain='%s' is empty after tidyup.domain(): row[hostname]='%s' - SKIPPED!", domain, row["hostname"]) continue @@ -1780,7 +1781,7 @@ def fetch_relays(args: argparse.Namespace) -> int: domain = tidyup.domain(domain) if domain not in[None, ""] else None logger.debug("domain='%s' - AFTER!", domain) - if domain is None or domain == "": + if domain in [None, ""]: logger.debug("domain='%s' is empty after tidyup.domain() from origin='%s' - SKIPPED!", domain, row["domain"]) continue elif domain not in peers: @@ -1822,7 +1823,7 @@ def fetch_relays(args: argparse.Namespace) -> int: domain = tidyup.domain(domain) if domain not in[None, ""] else None logger.debug("domain='%s' - AFTER!", domain) - if domain is None or domain == "": + if domain in [None, ""]: logger.debug("domain='%s' is empty after tidyup.domain() from origin='%s' - SKIPPED!", domain, row["domain"]) continue elif domain not in peers: @@ -1846,7 +1847,7 @@ def fetch_relays(args: argparse.Namespace) -> int: domain = tidyup.domain(domain) if domain not in[None, ""] else None logger.debug("domain='%s' - AFTER!", domain) - if domain is None or domain == "": + if domain in [None, ""]: logger.debug("domain='%s' is empty after tidyup.domain() from origin='%s' - SKIPPED!", domain, row["domain"]) continue elif domain not in peers: diff --git a/fba/helpers/processing.py b/fba/helpers/processing.py index defabdd..d8f3d12 100644 --- a/fba/helpers/processing.py +++ b/fba/helpers/processing.py @@ -54,10 +54,12 @@ def instance(blocked: str, blocker: str, command: str) -> bool: blocked = utils.deobfuscate(blocked, blocker) logger.debug("blocked='%s' - AFTER!", blocked) + logger.debug("Checking if blocker='%s' has pending data ...", blocker) if instances.has_pending(blocker): logger.debug("Flushing updates for blocker='%s' ...", blocker) instances.update(blocker) + logger.debug("Checking blocked='%s' if wanted and recent ...", blocked) if not domain_helper.is_wanted(blocked): logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked) return False @@ -193,7 +195,7 @@ def csv_block(blocker: str, url: str, command: str): cnt = cnt + 1 logger.debug("domain='%s',severity='%s',reject_media='%s',reject_reports='%s'", domain, severity, reject_media, reject_reports) - if domain is None or domain == "": + if domain in [None, ""]: logger.debug("domain='%s' is empty - SKIPPED!", domain) continue elif domain.endswith(".onion"): diff --git a/fba/http/federation.py b/fba/http/federation.py index 136a4ea..f04fd3f 100644 --- a/fba/http/federation.py +++ b/fba/http/federation.py @@ -141,7 +141,7 @@ def fetch_instances(domain: str, origin: str, software: str, command: str, path: logger.info("Checking %d instance(s) from domain='%s',software='%s',depth=%d ...", len(peerlist), domain, software, _DEPTH) for instance in peerlist: logger.debug("instance[%s]='%s'", type(instance), instance) - if instance is None or instance == "": + if instance in [None, ""]: logger.debug("instance[%s]='%s' is either None or empty - SKIPPED!", type(instance), instance) continue @@ -149,7 +149,7 @@ def fetch_instances(domain: str, origin: str, software: str, command: str, path: instance = tidyup.domain(instance) if isinstance(instance, str) and instance != "" else None logger.debug("instance='%s' - AFTER!", instance) - if instance is None or instance == "": + if instance in [None, ""]: logger.warning("instance='%s' is empty after tidyup.domain(), domain='%s'", instance, domain) continue elif ".." in instance: @@ -447,7 +447,7 @@ def determine_software(domain: str, path: str = None) -> str: logger.debug("Generator for domain='%s' is: '%s'", domain, software) logger.debug("software[%s]='%s'", type(software), software) - if software is None or software == "": + if software in [None, ""]: logger.debug("Returning None - EXIT!") return None @@ -537,7 +537,7 @@ def add_peers(rows: dict) -> list: logger.debug("Adding %d peer(s) to peers list ...", len(rows[key])) for peer in rows[key]: logger.debug("peer[%s]='%s' - BEFORE!", type(peer), peer) - if peer is None or peer == "": + if peer in [None, ""]: logger.debug("peer is empty - SKIPPED") continue elif isinstance(peer, dict) and "domain" in peer: diff --git a/fba/http/network.py b/fba/http/network.py index 1e694d0..a5ad140 100644 --- a/fba/http/network.py +++ b/fba/http/network.py @@ -23,6 +23,7 @@ import urllib3 from fba import utils +from fba.helpers import blacklist from fba.helpers import config from fba.helpers import cookies from fba.helpers import domain as domain_helper @@ -61,7 +62,9 @@ def post_json_api(domain: str, path: str, data: str = "", headers: dict = dict() logger.debug("domain='%s',path='%s',data='%s',headers()=%d - CALLED!", domain, path, data, len(headers)) domain_helper.raise_on(domain) - if not isinstance(path, str): + if blacklist.is_blacklisted(domain): + raise ValueError(f"domain='{domain}' is blacklisted but function was invoked") + elif not isinstance(path, str): raise ValueError(f"path[]='{type(path)}' is not of type 'str'") elif path == "": raise ValueError("Parameter 'path' is empty") @@ -158,7 +161,9 @@ def get_json_api(domain: str, path: str, headers: dict, timeout: tuple) -> dict: logger.debug("domain='%s',path='%s',timeout()=%d - CALLED!", domain, path, len(timeout)) domain_helper.raise_on(domain) - if not isinstance(path, str): + if blacklist.is_blacklisted(domain): + raise ValueError(f"domain='{domain}' is blacklisted but function was invoked") + elif not isinstance(path, str): raise ValueError(f"path[]='{type(path)}' is not of type 'str'") elif path == "": raise ValueError("Parameter 'path' is empty") @@ -204,7 +209,9 @@ def send_bot_post(domain: str, blocklist: list): logger.debug("domain='%s',blocklist()=%d - CALLED!", domain, len(blocklist)) domain_helper.raise_on(domain) - if not isinstance(blocklist, list): + if blacklist.is_blacklisted(domain): + raise ValueError(f"domain='{domain}' is blacklisted but function was invoked") + elif not isinstance(blocklist, list): raise ValueError(f"Parameter blocklist[]='{type(blocklist)}' is not of type 'list'") elif len(blocklist) == 0: raise ValueError("Parameter 'blocklist' is empty") @@ -222,7 +229,7 @@ def send_bot_post(domain: str, blocklist: list): logger.debug("blocklist()=%d", len(blocklist)) for block in blocklist: logger.debug("block[%s]='%s'", type(block), block) - if block["reason"] is None or block["reason"] == '': + if block["reason"] in [None, ""]: logger.debug("block[blocked]='%s' is being blocked with no reason specified", block["blocked"]) message = message + block["blocked"] + " with unspecified reason\n" else: @@ -254,7 +261,9 @@ def fetch_response(domain: str, path: str, headers: dict, timeout: tuple, allow_ logger.debug("domain='%s',path='%s',headers()=%d,timeout='%s',allow_redirects='%s' - CALLED!", domain, path, len(headers), timeout, allow_redirects) domain_helper.raise_on(domain) - if not isinstance(path, str): + if blacklist.is_blacklisted(domain): + raise ValueError(f"domain='{domain}' is blacklisted but function was invoked") + elif not isinstance(path, str): raise ValueError(f"Parameter path[]='{type(path)}' is not of type 'str'") elif path == "": raise ValueError("Parameter 'path' is empty") diff --git a/fba/networks/friendica.py b/fba/networks/friendica.py index e48e08c..d370818 100644 --- a/fba/networks/friendica.py +++ b/fba/networks/friendica.py @@ -91,7 +91,7 @@ def fetch_blocks(domain: str) -> list: reason = tidyup.reason(line.find_all("td")[1].text) logger.debug("blocked='%s',reason='%s' - AFTER!", blocked, reason) - if blocked is None or blocked == "": + if blocked in [None, ""]: logger.warning("line[]='%s' returned empty blocked domain - SKIPPED!", type(line)) continue elif not domain_helper.is_wanted(blocked): diff --git a/fba/networks/lemmy.py b/fba/networks/lemmy.py index 52b2195..8c6e336 100644 --- a/fba/networks/lemmy.py +++ b/fba/networks/lemmy.py @@ -165,7 +165,7 @@ def fetch_blocks(domain: str) -> list: content = str(header.contents[0]) logger.debug("content[%s]='%s' - AFTER!", type(content), content) - if content is None or content == "": + if content in [None, ""]: logger.debug("domain='%s' has returned empty header='%s' - SKIPPED!", domain, header) continue elif not isinstance(content, str): @@ -211,7 +211,7 @@ def fetch_blocks(domain: str) -> list: blocked = tidyup.domain(tag.contents[0]) if tag.contents[0] != "" else None logger.debug("blocked='%s'", blocked) - if blocked is None or blocked == "": + if blocked in [None, ""]: logger.warning("blocked='%s' is empty after tidyup.domain() - SKIPPED!", tag.contents[0]) continue elif not domain_helper.is_wanted(blocked): @@ -280,7 +280,7 @@ def fetch_instances(domain: str, origin: str) -> list: peer = tidyup.domain(text) if text != "" else None logger.debug("peer='%s' - AFTER", peer) - if peer is None or peer == "": + if peer in [None, ""]: logger.warning("peer='%s' is empty, text='%s' - SKIPPED!", peer, text) continue elif not domain_helper.is_wanted(peer): @@ -380,7 +380,7 @@ def parse_script(doc: bs4.BeautifulSoup, only: str = None) -> list: peer = tidyup.domain(row["domain"]) logger.debug("peer='%s' - AFTER!", peer) - if peer is None or peer == "": + if peer in [None, ""]: logger.warning("peer='%s' is empty, row[domain]='%s' - SKIPPED!", peer, row["domain"]) continue elif not domain_helper.is_wanted(peer): diff --git a/fba/networks/mastodon.py b/fba/networks/mastodon.py index 6dacdfd..127c33a 100644 --- a/fba/networks/mastodon.py +++ b/fba/networks/mastodon.py @@ -127,7 +127,7 @@ def fetch_blocks_from_about(domain: str) -> dict: reason = tidyup.reason(reason) if reason != "" else None logger.debug("domain='%s',reason='%s' - AFTER!", domain, reason) - if domain is None or domain == "": + if domain in [None, ""]: logger.warning("domain='%s' is empty,line='%s' - SKIPPED!", domain, line) continue diff --git a/fba/networks/misskey.py b/fba/networks/misskey.py index ddbfa46..e544d3a 100644 --- a/fba/networks/misskey.py +++ b/fba/networks/misskey.py @@ -210,7 +210,7 @@ def fetch_blocks(domain: str) -> list: if "host" not in instance: logger.warning("instance(%d)='%s' has no key 'host' - SKIPPED!", len(instance), instance) continue - elif instance["host"] is None or instance["host"] == "": + elif instance["host"] in [None, ""]: logger.debug("instance[host]='%s' is None or empty - SKIPPED!", instance["host"]) continue @@ -218,7 +218,7 @@ def fetch_blocks(domain: str) -> list: blocked = tidyup.domain(instance["host"]) logger.debug("blocked[%s]='%s' - AFTER!", type(blocked), blocked) - if blocked is None or blocked == "": + if blocked in [None, ""]: logger.warning("instance[host]='%s' is None or empty after tidyup.domain() - SKIPPED!", instance["host"]) continue elif not domain_helper.is_wanted(blocked): @@ -298,7 +298,7 @@ def fetch_blocks(domain: str) -> list: blocked = tidyup.domain(instance["host"]) logger.debug("blocked='%s'", blocked) - if blocked is None or blocked == "": + if blocked in [None, ""]: logger.warning("instance[host]='%s' is None or empty after tidyup.domain() - SKIPPED!", instance["host"]) continue elif not domain_helper.is_wanted(blocked): diff --git a/fba/networks/pleroma.py b/fba/networks/pleroma.py index 41d8500..c637f65 100644 --- a/fba/networks/pleroma.py +++ b/fba/networks/pleroma.py @@ -129,7 +129,7 @@ def fetch_blocks(domain: str) -> list: blocked = tidyup.domain(blocked) if blocked != "" else None logger.debug("blocked='%s' - AFTER!", blocked) - if blocked is None or blocked == "": + if blocked in [None, ""]: logger.warning("blocked='%s' is empty after tidyup.domain(): domain='%s',block_level='%s' - SKIPPED!", blocked, domain, block_level) continue @@ -137,7 +137,7 @@ def fetch_blocks(domain: str) -> list: blocked = utils.deobfuscate(blocked, domain) logger.debug("blocked='%s' - DEOBFUSCATED!", blocked) - if blocked is None or blocked == "": + if blocked in [None, ""]: logger.warning("instance[host]='%s' is None or empty after tidyup.domain() - SKIPPED!", instance["host"]) continue elif not domain_helper.is_wanted(blocked): @@ -163,7 +163,7 @@ def fetch_blocks(domain: str) -> list: blocked = tidyup.domain(blocked) if blocked != "" else None logger.debug("blocked='%s' - AFTER!", blocked) - if blocked is None or blocked == "": + if blocked in [None, ""]: logger.warning("blocked is empty after tidyup.domain(): domain='%s',block_level='%s'", domain, block_level) continue elif not domain_helper.is_wanted(blocked): @@ -198,7 +198,7 @@ def fetch_blocks(domain: str) -> list: block_level = tidyup.domain(block_level) if block_level != "" else None logger.debug("block_level='%s' - AFTER!", block_level) - if block_level is None or block_level == "": + if block_level in [None, ""]: logger.warning("block_level='%s' is now empty!", block_level) continue elif block_level == "accept": -- 2.39.5