]> git.mxchange.org Git - fba.git/blobdiff - fba/networks/pleroma.py
Continued:
[fba.git] / fba / networks / pleroma.py
index 3b580560cba02ddf154c3b5dae1ad95f30d4fc67..bdccdc2997414657599f216267393a9882aeaa54 100644 (file)
@@ -17,6 +17,7 @@
 import logging
 
 import bs4
+import validators
 
 from fba import database
 from fba import utils
@@ -60,6 +61,7 @@ def fetch_blocks(domain: str) -> list:
     elif not instances.is_registered(domain):
         raise Exception(f"domain='{domain}' is not registered but function is invoked.")
 
+    # Init variables
     blockdict = list()
     rows = None
 
@@ -85,6 +87,7 @@ def fetch_blocks(domain: str) -> list:
         logger.warning("Exception '%s' during fetching nodeinfo from domain='%s'", type(exception), domain)
         instances.set_last_error(domain, exception)
 
+    logger.debug("rows[]='%s'", type(rows))
     if rows is None:
         logger.warning("Could not fetch nodeinfo from domain='%s' - EXIT!", domain)
         return list()
@@ -95,21 +98,14 @@ def fetch_blocks(domain: str) -> list:
         logger.warning("rows()=%d does not have key 'federation', domain='%s' - EXIT!", len(rows["metadata"]), domain)
         return list()
 
-    data = rows["metadata"]["federation"]
     found = False
-
+    data = rows["metadata"]["federation"]
     logger.debug("data[]='%s'", type(data))
+
     if "mrf_simple" in data:
         logger.debug("Found mrf_simple in API response from domain='%s'", domain)
         found = True
-        for block_level, blocklist in (
-            {
-                **data["mrf_simple"],
-                **{
-                    "quarantined_instances": data["quarantined_instances"]
-                }
-            }
-        ).items():
+        for block_level, blocklist in data["mrf_simple"].items():
             logger.debug("block_level='%s', blocklist()=%d", block_level, len(blocklist))
             block_level = tidyup.domain(block_level) if block_level != "" else None
             logger.debug("block_level='%s' - AFTER!", block_level)
@@ -124,46 +120,49 @@ def fetch_blocks(domain: str) -> list:
             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) if blocked != "" else None
-                    logger.debug("blocked='%s' - AFTER!", 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(%s, %s) ...", blocked, domain)
-                    blocked = utils.deobfuscate(blocked, domain)
-
-                    logger.debug("blocked='%s' - DEOBFUSCATED!", 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
-
-                    logger.debug("Appending blocker='%s',blocked='%s',block_level='%s' ...", domain, blocked, block_level)
-                    blockdict.append({
-                        "blocker"    : domain,
-                        "blocked"    : blocked,
-                        "reason"     : None,
-                        "block_level": block_level,
-                    })
+            for blocked in blocklist:
+                logger.debug("blocked='%s' - BEFORE!", blocked)
+                blocked = tidyup.domain(blocked) if blocked != "" else None
+                logger.debug("blocked='%s' - AFTER!", 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
+                elif validators.domain(blocked) and blacklist.is_blacklisted(blocked):
+                    logger.debug("blocked='%s' is blacklisted - SKIPPED!")
+                    continue
+
+                logger.debug("Invoking utils.deobfuscate(%s, %s) ...", blocked, domain)
+                blocked = utils.deobfuscate(blocked, domain)
+                logger.debug("blocked[%s]='%s' - DEOBFUSCATED!", type(blocked), 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):
+                    logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked)
+                    continue
+
+                logger.debug("Appending blocker='%s',blocked='%s',block_level='%s' ...", domain, blocked, block_level)
+                blockdict.append({
+                    "blocker"    : domain,
+                    "blocked"    : blocked,
+                    "reason"     : None,
+                    "block_level": block_level,
+                })
 
     elif "quarantined_instances" in data:
         logger.debug("Found 'quarantined_instances' in JSON response: domain='%s'", domain)
         found = True
         block_level = "quarantined"
 
+        logger.debug("Checking %d quarantined instance(s) ...", len(data["quarantined_instances"]))
         for blocked in data["quarantined_instances"]:
             logger.debug("blocked='%s' - BEFORE!", blocked)
             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,11 +197,8 @@ 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:
-                logger.warning("block_level is now None!")
-                continue
-            elif block_level == "":
-                logger.warning("block_level is now empty!")
+            if block_level in [None, ""]:
+                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)
@@ -247,7 +243,7 @@ def fetch_blocks(domain: 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"])
+            reason  = tidyup.reason(rows[blocked]["reason"]) if rows[blocked]["reason"] != "" else None
             blocked = tidyup.domain(blocked) if blocked != "" else None
             logger.debug("blocked='%s',reason='%s' - AFTER!", blocked, reason)
 
@@ -258,7 +254,7 @@ def fetch_blocks(domain: str) -> list:
                 logger.warning("blocked is empty after tidyup.domain(): domain='%s',block_level='%s'", domain, block_level)
                 continue
 
-            logger.debug("Checking %d blockdict records ...", len(blockdict))
+            logger.debug("Checking %d blockdict record(s) ...", len(blockdict))
             for block in blockdict:
                 logger.debug("block[blocked]='%s',blocked='%s'", block["blocked"], blocked)
                 if block["blocked"] == blocked:
@@ -267,19 +263,17 @@ def fetch_blocks(domain: str) -> list:
     else:
         logger.warning("Cannot find 'mrf_simple_info' or 'quarantined_instances_info' in JSON reply: domain='%s'", domain)
 
+    logger.debug("found='%s'", found)
     if not found:
         logger.debug("Did not find any useable JSON elements, domain='%s', continuing with /about page ...", domain)
         blocklist = fetch_blocks_from_about(domain)
 
         logger.debug("blocklist()=%d", len(blocklist))
         if len(blocklist) > 0:
-            logger.info("Checking %d different blocklists ...", len(blocklist))
+            logger.info("Checking %d different blocklist(s) ...", 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 block in rows:
+                logger.debug("Checking blocklist[%s]()=%d entries ...", block_level, blocklist[block_level])
+                for block in blocklist[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,
@@ -300,8 +294,10 @@ def fetch_blocks_from_about(domain: str) -> dict:
     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)
+    # Init variables
     doc = None
+
+    logger.debug("Fetching mastodon blocks from domain='%s'", domain)
     for path in ["/instance/about/index.html"]:
         try:
             # Resetting doc type
@@ -377,14 +373,15 @@ def fetch_blocks_from_about(domain: str) -> dict:
             for line in header.find_next("table").find_all("tr")[1:]:
                 logger.debug("line[]='%s'", type(line))
                 blocked = line.find_all("td")[0].text
-                logger.debug("blocked='%s'", blocked)
+                reason  = line.find_all("td")[1].text
 
+                logger.debug("blocked='%s',reason='%s' - BEFORE!", blocked, reason)
                 blocked = tidyup.domain(blocked) if blocked != "" else None
-                reason = tidyup.reason(line.find_all("td")[1].text)
+                reason  = tidyup.reason(reason)  if reason  != "" else None
                 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)
+                if blocked in [None, ""]:
+                    logger.debug("domain='%s',block_level='%s': blocked='%s' is empty - SKIPPED!", domain, block_level, blocked)
                     continue
                 elif not domain_helper.is_wanted(blocked):
                     logger.debug("blocked='%s' is not wanted - SKIPPED!", blocked)