fetch_blocks.py

   1 from requests import get
   2 from requests import post
   3 from hashlib import sha256
   4 import sqlite3
   5 from bs4 import BeautifulSoup
   6
   7 headers = {
   8     "user-agent": "Mozilla/5.0 (Windows NT 10.0; rv:102.0) Gecko/20100101 Firefox/102.0"
   9 }
  10
  11
  12 def get_mastodon_blocks(domain: str) -> dict:
  13     blocks = {
  14         "Suspended servers": [],
  15         "Filtered media": [],
  16         "Limited servers": [],
  17         "Silenced servers": [],
  18     }
  19
  20     translations = {
  21         "Silenced instances": "Silenced servers",
  22         "Suspended instances": "Suspended servers",
  23         "Gesperrte Server": "Suspended servers",
  24         "Gefilterte Medien": "Filtered media",
  25         "Stummgeschaltete Server": "Silenced servers",
  26         "停止済みのサーバー": "Suspended servers",
  27         "メディアを拒否しているサーバー": "Filtered media",
  28         "サイレンス済みのサーバー": "Silenced servers",
  29         "Serveurs suspendus": "Suspended servers",
  30         "Médias filtrés": "Filtered media",
  31         "Serveurs limités": "Silenced servers",
  32     }
  33
  34     try:
  35         doc = BeautifulSoup(
  36             get(f"https://{domain}/about/more", headers=headers, timeout=5).text,
  37             "html.parser",
  38         )
  39     except:
  40         return {}
  41
  42     for header in doc.find_all("h3"):
  43         header_text = header.text
  44         if header_text in translations:
  45             header_text = translations[header_text]
  46         if header_text in blocks:
  47             for line in header.find_next_siblings("table")[0].find_all("tr")[1:]:
  48                 blocks[header_text].append(
  49                     {
  50                         "domain": line.find("span").text,
  51                         "hash": line.find("span")["title"][9:],
  52                         "reason": line.find_all("td")[1].text.strip(),
  53                     }
  54                 )
  55     return {
  56         "reject": blocks["Suspended servers"],
  57         "media_removal": blocks["Filtered media"],
  58         "federated_timeline_removal": blocks["Limited servers"]
  59         + blocks["Silenced servers"],
  60     }
  61
  62 def get_friendica_blocks(domain: str) -> dict:
  63     blocks = []
  64
  65     try:
  66         doc = BeautifulSoup(
  67             get(f"https://{domain}/friendica", headers=headers, timeout=5).text,
  68             "html.parser",
  69         )
  70     except:
  71         return {}
  72
  73     blocklist = doc.find(id="about_blocklist")
  74     for line in blocklist.find("table").find_all("tr")[1:]:
  75             blocks.append(
  76                 {
  77                     "domain": line.find_all("td")[0].text.strip(),
  78                     "reason": line.find_all("td")[1].text.strip()
  79                 }
  80             )
  81
  82     return {
  83         "reject": blocks
  84     }
  85
  86 def get_pisskey_blocks(domain: str) -> dict:
  87     blocks = {
  88         "suspended": [],
  89         "blocked": []
  90     }
  91
  92     try:
  93         counter = 0
  94         step = 99
  95         while True:
  96             # iterating through all "suspended" (follow-only in its terminology) instances page-by-page, since that troonware doesn't support sending them all at once
  97             try:
  98                 if counter == 0:
  99                     doc = post(f"https://{domain}/api/federation/instances", data=dumps({"sort":"+caughtAt","host":None,"suspended":True,"limit":step}), headers=headers, timeout=5).json()
 100                     if doc == []: raise
 101                 else:
 102                     doc = post(f"https://{domain}/api/federation/instances", data=dumps({"sort":"+caughtAt","host":None,"suspended":True,"limit":step,"offset":counter-1}), headers=headers, timeout=5).json()
 103                     if doc == []: raise
 104                 for instance in doc:
 105                     # just in case
 106                     if instance["isSuspended"]:
 107                         blocks["suspended"].append(
 108                             {
 109                                 "domain": instance["host"],
 110                                 # no reason field, nothing
 111                                 "reason": ""
 112                             }
 113                         )
 114                 counter = counter + step
 115             except:
 116                 counter = 0
 117                 break
 118
 119         while True:
 120             # same shit, different asshole ("blocked" aka full suspend)
 121             try:
 122                 if counter == 0:
 123                     doc = post(f"https://{domain}/api/federation/instances", data=dumps({"sort":"+caughtAt","host":None,"blocked":True,"limit":step}), headers=headers, timeout=5).json()
 124                     if doc == []: raise
 125                 else:
 126                     doc = post(f"https://{domain}/api/federation/instances", data=dumps({"sort":"+caughtAt","host":None,"blocked":True,"limit":step,"offset":counter-1}), headers=headers, timeout=5).json()
 127                     if doc == []: raise
 128                 for instance in doc:
 129                     if instance["isBlocked"]:
 130                         blocks["blocked"].append(
 131                             {
 132                                 "domain": instance["host"],
 133                                 "reason": ""
 134                             }
 135                         )
 136                 counter = counter + step
 137             except:
 138                 counter = 0
 139                 break
 140
 141         return {
 142             "reject": blocks["blocked"],
 143             "followers_only": blocks["suspended"]
 144         }
 145
 146     except:
 147         return {}
 148
 149 def get_hash(domain: str) -> str:
 150     return sha256(domain.encode("utf-8")).hexdigest()
 151
 152
 153 def get_type(domain: str) -> str:
 154     try:
 155         res = get(f"https://{domain}/nodeinfo/2.1.json", headers=headers, timeout=5)
 156         if res.status_code == 404:
 157             res = get(f"https://{domain}/nodeinfo/2.0", headers=headers, timeout=5)
 158         if res.status_code == 404:
 159             res = get(f"https://{domain}/nodeinfo/2.0.json", headers=headers, timeout=5)
 160         if res.ok and "text/html" in res.headers["content-type"]:
 161             res = get(f"https://{domain}/nodeinfo/2.1", headers=headers, timeout=5)
 162         if res.ok:
 163             if res.json()["software"]["name"] == "akkoma":
 164                 return "pleroma"
 165             elif res.json()["software"]["name"] == "rebased":
 166                 return "pleroma"
 167             elif res.json()["software"]["name"] == "hometown":
 168                 return "mastodon"
 169             elif res.json()["software"]["name"] == "ecko":
 170                 return "mastodon"
 171             elif res.json()["software"]["name"] == "calckey":
 172                 return "misskey"
 173             else:
 174                 return res.json()["software"]["name"]
 175         elif res.status_code == 404:
 176             res = get(f"https://{domain}/api/v1/instance", headers=headers, timeout=5)
 177         if res.ok:
 178             return "mastodon"
 179     except:
 180         return None
 181
 182
 183 conn = sqlite3.connect("blocks.db")
 184 c = conn.cursor()
 185
 186 c.execute(
 187     "select domain, software from instances where software in ('pleroma', 'mastodon', 'friendica', 'misskey')"
 188 )
 189
 190 for blocker, software in c.fetchall():
 191     if software == "pleroma":
 192         print(blocker)
 193         try:
 194             # Blocks
 195             federation = get(
 196                 f"https://{blocker}/nodeinfo/2.1.json", headers=headers, timeout=5
 197             ).json()["metadata"]["federation"]
 198             if "mrf_simple" in federation:
 199                 for block_level, blocks in (
 200                     {**federation["mrf_simple"],
 201                     **{"quarantined_instances": federation["quarantined_instances"]}}
 202                 ).items():
 203                     for blocked in blocks:
 204                         if blocked == "":
 205                             continue
 206                         blocked == blocked.lower()
 207                         blocker == blocker.lower()
 208                         c.execute(
 209                             "select domain from instances where domain = ?", (blocked,)
 210                         )
 211                         if c.fetchone() == None:
 212                             c.execute(
 213                                 "insert into instances select ?, ?, ?",
 214                                 (blocked, get_hash(blocked), get_type(blocked)),
 215                             )
 216                         c.execute(
 217                             "select * from blocks where blocker = ? and blocked = ? and block_level = ?",
 218                             (blocker, blocked, block_level),
 219                         )
 220                         if c.fetchone() == None:
 221                             c.execute(
 222                                 "insert into blocks select ?, ?, '', ?",
 223                                 (blocker, blocked, block_level),
 224                             )
 225             conn.commit()
 226             # Reasons
 227             if "mrf_simple_info" in federation:
 228                 for block_level, info in (
 229                     {**federation["mrf_simple_info"],
 230                     **(federation["quarantined_instances_info"]
 231                     if "quarantined_instances_info" in federation
 232                     else {})}
 233                 ).items():
 234                     for blocked, reason in info.items():
 235                         blocker == blocker.lower()
 236                         blocked == blocked.lower()
 237                         c.execute(
 238                             "update blocks set reason = ? where blocker = ? and blocked = ? and block_level = ?",
 239                             (reason["reason"], blocker, blocked, block_level),
 240                         )
 241             conn.commit()
 242         except Exception as e:
 243             print("error:", e, blocker)
 244     elif software == "mastodon":
 245         print(blocker)
 246         try:
 247             json = get_mastodon_blocks(blocker)
 248             for block_level, blocks in json.items():
 249                 for instance in blocks:
 250                     blocked, blocked_hash, reason = instance.values()
 251                     blocked == blocked.lower()
 252                     blocker == blocker.lower()
 253                     if blocked.count("*") <= 1:
 254                         c.execute(
 255                             "select hash from instances where hash = ?", (blocked_hash,)
 256                         )
 257                         if c.fetchone() == None:
 258                             c.execute(
 259                                 "insert into instances select ?, ?, ?",
 260                                 (blocked, get_hash(blocked), get_type(blocked)),
 261                             )
 262                     c.execute(
 263                         "select * from blocks where blocker = ? and blocked = ? and block_level = ?",
 264                         (blocker, blocked if blocked.count("*") <= 1 else blocked_hash, block_level),
 265                     )
 266                     if c.fetchone() == None:
 267                         c.execute(
 268                             "insert into blocks select ?, ?, ?, ?",
 269                             (
 270                                 blocker,
 271                                 blocked if blocked.count("*") <= 1 else blocked_hash,
 272                                 reason,
 273                                 block_level,
 274                             ),
 275                         )
 276             conn.commit()
 277         except Exception as e:
 278             print("error:", e, blocker)
 279     elif software == "friendica" or software == "misskey":
 280         print(blocker)
 281         try:
 282             if software == "friendica":
 283                 json = get_friendica_blocks(blocker)
 284             elif software == "misskey":
 285                 json = get_pisskey_blocks(blocker)
 286             for block_level, blocks in json.items():
 287                 for instance in blocks:
 288                     blocked, reason = instance.values()
 289                     blocked == blocked.lower()
 290                     blocker == blocker.lower()
 291                     c.execute(
 292                         "select domain from instances where domain = ?", (blocked,)
 293                     )
 294                     if c.fetchone() == None:
 295                         c.execute(
 296                             "insert into instances select ?, ?, ?",
 297                             (blocked, get_hash(blocked), get_type(blocked)),
 298                         )
 299                     c.execute(
 300                         "select * from blocks where blocker = ? and blocked = ?",
 301                         (blocker, blocked),
 302                     )
 303                     if c.fetchone() == None:
 304                         c.execute(
 305                             "insert into blocks select ?, ?, ?, ?",
 306                             (
 307                                 blocker,
 308                                 blocked,
 309                                 reason,
 310                                 block_level,
 311                             ),
 312                         )
 313             conn.commit()
 314         except Exception as e:
 315             print("error:", e, blocker)
 316 conn.close()