fetch_blocks.py

   1 from requests import get
   2 from hashlib import sha256
   3 import sqlite3
   4 from bs4 import BeautifulSoup
   5
   6 headers = {
   7     "user-agent": "fedi-block-api (https://gitlab.com/EnjuAihara/fedi-block-api)"
   8 }
   9
  10
  11 def get_mastodon_blocks(domain: str) -> dict:
  12     blocks = {
  13         "Suspended servers": [],
  14         "Filtered media": [],
  15         "Limited servers": [],
  16         "Silenced servers": [],
  17     }
  18
  19     try:
  20         doc = BeautifulSoup(
  21             get(f"https://{domain}/about/more", headers=headers, timeout=5).text,
  22             "html.parser",
  23         )
  24     except:
  25         return {}
  26
  27     for header in doc.find_all("h3"):
  28         for line in header.find_next_siblings("table")[0].find_all("tr")[1:]:
  29             if header.text in blocks:
  30                 blocks[header.text].append(
  31                     {
  32                         "domain": line.find("span").text,
  33                         "hash": line.find("span")["title"][9:],
  34                         "reason": line.find_all("td")[1].text.strip(),
  35                     }
  36                 )
  37     return {
  38         "reject": blocks["Suspended servers"],
  39         "media_removal": blocks["Filtered media"],
  40         "federated_timeline_removal": blocks["Limited servers"]
  41         + blocks["Silenced servers"],
  42     }
  43
  44
  45 def get_hash(domain: str) -> str:
  46     return sha256(domain.encode("utf-8")).hexdigest()
  47
  48
  49 def get_type(domain: str) -> str:
  50     try:
  51         res = get(f"https://{domain}/nodeinfo/2.1.json", headers=headers, timeout=5)
  52         if res.status_code == 404:
  53             res = get(f"https://{domain}/nodeinfo/2.0.json", headers=headers, timeout=5)
  54         if res.ok and "text/html" in res.headers["content-type"]:
  55             res = get(f"https://{domain}/nodeinfo/2.1", headers=headers, timeout=5)
  56         if res.ok:
  57             return res.json()["software"]["name"]
  58         elif res.status_code == 404:
  59             res = get(f"https://{domain}/api/v1/instance", headers=headers, timeout=5)
  60         if res.ok:
  61             return "mastodon"
  62     except:
  63         return None
  64
  65
  66 conn = sqlite3.connect("blocks.db")
  67 c = conn.cursor()
  68
  69 c.execute(
  70     "select domain, software from instances where software in ('pleroma', 'mastodon')"
  71 )
  72
  73 for blocker, software in c.fetchall():
  74     if software == "pleroma":
  75         print(blocker)
  76         try:
  77             # Blocks
  78             federation = get(
  79                 f"https://{blocker}/nodeinfo/2.1.json", headers=headers, timeout=5
  80             ).json()["metadata"]["federation"]
  81             if "mrf_simple" in federation:
  82                 for block_level, blocks in (
  83                     federation["mrf_simple"]
  84                     | {"quarantined_instances": federation["quarantined_instances"]}
  85                 ).items():
  86                     for blocked in blocks:
  87                         if blocked == "":
  88                             continue
  89                         c.execute(
  90                             "select domain from instances where domain = ?", (blocked,)
  91                         )
  92                         if c.fetchone() == None:
  93                             c.execute(
  94                                 "insert into instances select ?, ?, ?",
  95                                 (blocked, get_hash(blocked), get_type(blocked)),
  96                             )
  97                         c.execute(
  98                             "select * from blocks where blocker = ? and blocked = ? and block_level = ?",
  99                             (blocker, blocked, block_level),
 100                         )
 101                         if c.fetchone() == None:
 102                             c.execute(
 103                                 "insert into blocks select ?, ?, '', ?",
 104                                 (blocker, blocked, block_level),
 105                             )
 106             conn.commit()
 107             # Reasons
 108             if "mrf_simple_info" in federation:
 109                 for block_level, info in (
 110                     federation["mrf_simple_info"]
 111                     | federation["quarantined_instances_info"]
 112                     if "quarantined_instances_info" in federation
 113                     else {}
 114                 ).items():
 115                     for blocked, reason in info.items():
 116                         c.execute(
 117                             "update blocks set reason = ? where blocker = ? and blocked = ? and block_level = ?",
 118                             (reason["reason"], blocker, blocked, block_level),
 119                         )
 120             conn.commit()
 121         except Exception as e:
 122             print("error:", e, blocker)
 123     elif software == "mastodon":
 124         print(blocker)
 125         try:
 126             json = get_mastodon_blocks(blocker)
 127             for block_level, blocks in json.items():
 128                 for instance in blocks:
 129                     blocked, blocked_hash, reason = instance.values()
 130                     if blocked.count("*") <= 1:
 131                         c.execute(
 132                             "select hash from instances where hash = ?", (blocked_hash,)
 133                         )
 134                         if c.fetchone() == None:
 135                             c.execute(
 136                                 "insert into instances select ?, ?, ?",
 137                                 (blocked, get_hash(blocked), get_type(blocked)),
 138                             )
 139                     c.execute(
 140                         "select * from blocks where blocker = ? and blocked = ? and block_level = ?",
 141                         (blocker, blocked, block_level),
 142                     )
 143                     if c.fetchone() == None:
 144                         c.execute(
 145                             "insert into blocks select ?, ?, ?, ?",
 146                             (
 147                                 blocker,
 148                                 blocked if blocked.count("*") <= 1 else blocked_hash,
 149                                 reason,
 150                                 block_level,
 151                             ),
 152                         )
 153             conn.commit()
 154         except Exception as e:
 155             print("error:", e, blocker)
 156 conn.close()