From 213d26eb30a0b472ce4b0cfaf95a2e96e6cbc28e Mon Sep 17 00:00:00 2001
From: =?utf8?q?Roland=20H=C3=A4der?= <roland@mxchange.org>
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