]> git.mxchange.org Git - fba.git/blobdiff - fba/commands.py
projects / fba.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | tree
raw | inline | side by side
Continued:
[fba.git] / fba / commands.py
diff --git a/fba/commands.py b/fba/commands.py
index e21a057644c5d9447d8729f1e2a1dcd404eb3ded..3fb4ea6869ded2a25451493b1172fd5db88889bf 100644 (file)
--- a/fba/commands.py
+++ b/fba/commands.py
@@ -29,7 +29,6 @@ import markdown
 import reqto
 import validators
 
-from fba import csrf
 from fba import database
 from fba import utils
 
@@ -44,6 +43,7 @@ from fba.helpers import processing
 from fba.helpers import software as software_helper
 from fba.helpers import tidyup
 
+from fba.http import csrf
 from fba.http import federation
 from fba.http import network
 
@@ -63,6 +63,7 @@ logger = logging.getLogger(__name__)
 
 def check_instance(args: argparse.Namespace) -> int:
     logger.debug("args.domain='%s' - CALLED!", args.domain)
+
     status = 0
     if not validators.domain(args.domain):
         logger.warning("args.domain='%s' is not valid", args.domain)
@@ -308,10 +309,7 @@ def fetch_blocks(args: argparse.Namespace) -> int:
     for blocker, software, origin, nodeinfo_url in rows:
         logger.debug("blocker='%s',software='%s',origin='%s',nodeinfo_url='%s'", blocker, software, origin, nodeinfo_url)
 
-        if nodeinfo_url is None:
-            logger.debug("blocker='%s',software='%s' has no nodeinfo_url set - SKIPPED!", blocker, software)
-            continue
-        elif not domain_helper.is_wanted(blocker):
+        if not domain_helper.is_wanted(blocker):
             logger.warning("blocker='%s' is not wanted - SKIPPED!", blocker)
             continue
 
@@ -319,42 +317,41 @@ def fetch_blocks(args: argparse.Namespace) -> int:
         instances.set_last_blocked(blocker)
         instances.set_has_obfuscation(blocker, False)
 
-        blocking = list()
-
         # c.s isn't part of oliphant's "hidden" blocklists
-        logger.debug("blocker='%s'", blocker)
-        if blocker != "chaos.social" and not blocklists.is_excluded(blocker):
-            logger.debug("blocker='%s',software='%s'", blocker, software)
+        if blocker == "chaos.social" or blocklists.has(blocker):
+            logger.debug("Skipping blocker='%s', run ./fba.py fetch_cs or fetch_oliphant instead!", blocker)
+            continue
+
+        logger.debug("Invoking federation.fetch_blocks(%s) ...", blocker)
+        blocking = federation.fetch_blocks(blocker)
+
+        logger.info("blocker='%s',software='%s' has %d block entries returned.", blocker, software, len(blocking))
+        if len(blocking) == 0:
+            logger.debug("blocker='%s',software='%s' - fetching blocklist ...", blocker, software)
             if software == "pleroma":
-                logger.info("blocker='%s',software='%s'", blocker, software)
-                blocking = pleroma.fetch_blocks(blocker, nodeinfo_url)
+                blocking = pleroma.fetch_blocks(blocker)
                 logger.debug("blocker='%s' returned %d entries,software='%s'", blocker, len(blocking), software)
             elif software == "mastodon":
-                logger.info("blocker='%s',software='%s'", blocker, software)
-                blocking = mastodon.fetch_blocks(blocker, nodeinfo_url)
+                blocking = mastodon.fetch_blocks(blocker)
                 logger.debug("blocker='%s' returned %d entries,software='%s'", blocker, len(blocking), software)
             elif software == "lemmy":
-                logger.info("blocker='%s',software='%s'", blocker, software)
-                blocking = lemmy.fetch_blocks(blocker, nodeinfo_url)
+                blocking = lemmy.fetch_blocks(blocker)
                 logger.debug("blocker='%s' returned %d entries,software='%s'", blocker, len(blocking), software)
             elif software == "friendica":
-                logger.info("blocker='%s',software='%s'", blocker, software)
                 blocking = friendica.fetch_blocks(blocker)
                 logger.debug("blocker='%s' returned %d entries,software='%s'", blocker, len(blocking), software)
             elif software == "misskey":
-                logger.info("blocker='%s',software='%s'", blocker, software)
                 blocking = misskey.fetch_blocks(blocker)
                 logger.debug("blocker='%s' returned %d entries,software='%s'", blocker, len(blocking), software)
             else:
                 logger.warning("Unknown software: blocker='%s',software='%s'", blocker, software)
 
-            logger.debug("Invoking instances.set_total_blocks(%s, %d) ...", blocker, len(blocking))
-            instances.set_total_blocks(blocker, blocking)
-        else:
-            logger.debug("Skipping blocker='%s', run ./fba.py fetch_cs or fetch_oliphant instead!", blocker)
+        logger.debug("Invoking instances.set_total_blocks(%s, %d) ...", blocker, len(blocking))
+        instances.set_total_blocks(blocker, blocking)
 
-        logger.info("Checking %d entries from blocker='%s',software='%s' ...", len(blocking), blocker, software)
         blockdict = list()
+
+        logger.info("Checking %d entries from blocker='%s',software='%s' ...", len(blocking), blocker, software)
         for block in blocking:
             logger.debug("blocked='%s',block_level='%s',reason='%s'", block["blocked"], block["block_level"], block["reason"])
 
@@ -778,7 +775,7 @@ def fetch_fba_rss(args: argparse.Namespace) -> int:
     response = utils.fetch_url(args.feed, network.web_headers, (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))
-    if response.ok and response.status_code < 300 and len(response.text) > 0:
+    if response.ok and response.status_code == 200 and len(response.text) > 0:
         logger.debug("Parsing RSS feed (%d Bytes) ...", len(response.text))
         rss = atoma.parse_rss_bytes(response.content)
 
@@ -856,7 +853,7 @@ def fetch_fbabot_atom(args: argparse.Namespace) -> int:
     response = utils.fetch_url(feed, network.web_headers, (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))
-    if response.ok and response.status_code < 300 and len(response.text) > 0:
+    if response.ok and response.status_code == 200 and len(response.text) > 0:
         logger.debug("Parsing ATOM feed (%d Bytes) ...", len(response.text))
         atom = atoma.parse_atom_bytes(response.content)
 
@@ -1018,7 +1015,7 @@ def fetch_oliphant(args: argparse.Namespace) -> int:
         response = utils.fetch_url(f"{base_url}/{block['csv_url']}", network.web_headers, (config.get("connection_timeout"), config.get("read_timeout")))
 
         logger.debug("response.ok='%s',response.status_code=%d,response.content()=%d", response.ok, response.status_code, len(response.content))
-        if not response.ok or response.status_code >= 300 or response.content == "":
+        if not response.ok or response.status_code > 200 or response.content == "":
             logger.warning("Could not fetch csv_url='%s' for blocker='%s' - SKIPPED!", block["csv_url"], block["blocker"])
             continue
 
@@ -1108,7 +1105,7 @@ def fetch_oliphant(args: argparse.Namespace) -> int:
                 processing.block(block["blocker"], domain, None, "reject_reports")
 
         logger.debug("block[blocker]='%s'", block["blocker"])
-        if not blocklists.is_excluded(block["blocker"]):
+        if not blocklists.has(block["blocker"]):
             logger.debug("Invoking instances.set_total_blocks(%s, domains()=%d) ...", block["blocker"], len(domains))
             instances.set_total_blocks(block["blocker"], domains)
 
@@ -1146,7 +1143,7 @@ def fetch_txt(args: argparse.Namespace) -> int:
         response = utils.fetch_url(row["url"], network.web_headers, (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))
-        if response.ok and response.status_code < 300 and response.text != "":
+        if response.ok and response.status_code == 200 and response.text != "":
             logger.debug("Returned %d Bytes for processing", len(response.text.strip()))
             domains = response.text.split("\n")
 
@@ -1199,7 +1196,7 @@ def fetch_fedipact(args: argparse.Namespace) -> int:
     )
 
     logger.debug("response.ok='%s',response.status_code=%d,response.text()=%d", response.ok, response.status_code, len(response.text))
-    if response.ok and response.status_code < 300 and response.text != "":
+    if response.ok and response.status_code == 200 and response.text != "":
         logger.debug("Parsing %d Bytes ...", len(response.text))
 
         doc = bs4.BeautifulSoup(response.text, "html.parser")
@@ -1532,28 +1529,32 @@ def recheck_obfuscation(args: argparse.Namespace) -> int:
             logger.debug("row[domain]='%s' has been recently checked, args.force[]='%s' - SKIPPED!", row["domain"], type(args.force))
             continue
 
-        blocking = list()
-        if row["software"] == "pleroma":
-            logger.debug("domain='%s',software='%s'", row["domain"], row["software"])
-            blocking = pleroma.fetch_blocks(row["domain"], row["nodeinfo_url"])
-        elif row["software"] == "mastodon":
-            logger.debug("domain='%s',software='%s'", row["domain"], row["software"])
-            blocking = mastodon.fetch_blocks(row["domain"], row["nodeinfo_url"])
-        elif row["software"] == "lemmy":
-            logger.debug("domain='%s',software='%s'", row["domain"], row["software"])
-            blocking = lemmy.fetch_blocks(row["domain"], row["nodeinfo_url"])
-        elif row["software"] == "friendica":
-            logger.debug("domain='%s',software='%s'", row["domain"], row["software"])
-            blocking = friendica.fetch_blocks(row["domain"])
-        elif row["software"] == "misskey":
-            logger.debug("domain='%s',software='%s'", row["domain"], row["software"])
-            blocking = misskey.fetch_blocks(row["domain"])
-        else:
-            logger.warning("Unknown software: domain='%s',software='%s'", row["domain"], row["software"])
+        logger.debug("Invoking federation.fetch_blocks(%s) ...", row["domain"])
+        blocking = federation.fetch_blocks(row["domain"])
+
+        logger.debug("blocking()=%d", len(blocking))
+        if len(blocking) == 0:
+            if row["software"] == "pleroma":
+                logger.debug("domain='%s',software='%s'", row["domain"], row["software"])
+                blocking = pleroma.fetch_blocks(row["domain"])
+            elif row["software"] == "mastodon":
+                logger.debug("domain='%s',software='%s'", row["domain"], row["software"])
+                blocking = mastodon.fetch_blocks(row["domain"])
+            elif row["software"] == "lemmy":
+                logger.debug("domain='%s',software='%s'", row["domain"], row["software"])
+                blocking = lemmy.fetch_blocks(row["domain"])
+            elif row["software"] == "friendica":
+                logger.debug("domain='%s',software='%s'", row["domain"], row["software"])
+                blocking = friendica.fetch_blocks(row["domain"])
+            elif row["software"] == "misskey":
+                logger.debug("domain='%s',software='%s'", row["domain"], row["software"])
+                blocking = misskey.fetch_blocks(row["domain"])
+            else:
+                logger.warning("Unknown software: domain='%s',software='%s'", row["domain"], row["software"])
 
         # c.s isn't part of oliphant's "hidden" blocklists
         logger.debug("row[domain]='%s'", row["domain"])
-        if row["domain"] != "chaos.social" and not blocklists.is_excluded(row["domain"]):
+        if row["domain"] != "chaos.social" and not blocklists.has(row["domain"]):
             logger.debug("Invoking instances.set_total_blocks(%s, %d) ...", row["domain"], len(blocking))
             instances.set_last_blocked(row["domain"])
             instances.set_total_blocks(row["domain"], blocking)
@@ -1662,7 +1663,7 @@ def fetch_fedilist(args: argparse.Namespace) -> int:
     )
 
     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.status_code >= 300 or len(response.content) == 0:
+    if not response.ok or response.status_code > 200 or len(response.content) == 0:
         logger.warning("Failed fetching url='%s': response.ok='%s',response.status_code=%d,response.content()=%d - EXIT!", url, response.ok, response.status_code, len(response.text))
         return 1
 
Personal clone of "fedi-block-api", if you like to use the command fetch_instances_social you need to get an API key from https://instances.social .