fetch_instances.py

   1 from reqto import get
   2 from hashlib import sha256
   3 import sqlite3
   4 import sys
   5 import json
   6
   7 with open("config.json") as f:
   8     config = json.loads(f.read())
   9
  10 domain = sys.argv[1]
  11
  12 blacklist = [
  13     "activitypub-troll.cf",
  14     "gab.best",
  15     "4chan.icu",
  16     "social.shrimpcam.pw",
  17     "mastotroll.netz.org"
  18 ]
  19
  20 headers = {
  21     "user-agent": config["useragent"]
  22 }
  23
  24 def get_hash(domain: str) -> str:
  25     return sha256(domain.encode("utf-8")).hexdigest()
  26
  27 def get_peers(domain: str) -> str:
  28     try:
  29         res = get(f"https://{domain}/api/v1/instance/peers", headers=headers, timeout=5)
  30         return res.json()
  31     except:
  32         print("WARNING: Cannot fetch peers:", domain)
  33         return None
  34
  35 peerlist = get_peers(domain)
  36
  37 if (peerlist is None):
  38     print("FATAL: CANNOT FETCH PEERS:", domain)
  39     sys.exit(255)
  40
  41 def get_type(instdomain: str) -> str:
  42     try:
  43         res = get(f"https://{instdomain}/nodeinfo/2.1.json", headers=headers, timeout=5)
  44         if res.status_code == 404:
  45             res = get(f"https://{instdomain}/nodeinfo/2.0", headers=headers, timeout=5)
  46         if res.status_code == 404:
  47             res = get(f"https://{instdomain}/nodeinfo/2.0.json", headers=headers, timeout=5)
  48         if res.ok and "text/html" in res.headers["content-type"]:
  49             res = get(f"https://{instdomain}/nodeinfo/2.1", headers=headers, timeout=5)
  50         if res.ok:
  51             if res.json()["software"]["name"] in ["akkoma", "rebased"]:
  52                 return "pleroma"
  53             elif res.json()["software"]["name"] in ["hometown", "ecko"]:
  54                 return "mastodon"
  55             elif res.json()["software"]["name"] in ["calckey", "groundpolis", "foundkey", "cherrypick"]:
  56                 return "misskey"
  57             else:
  58                 return res.json()["software"]["name"]
  59         elif res.status_code == 404:
  60             res = get(f"https://{instdomain}/api/v1/instance", headers=headers, timeout=5)
  61         if res.ok:
  62             return "mastodon"
  63     except:
  64         return None
  65
  66
  67 conn = sqlite3.connect("blocks.db")
  68 c = conn.cursor()
  69
  70 c.execute(
  71     "SELECT domain FROM instances WHERE 1"
  72 )
  73
  74 for instance in peerlist:
  75     instance = instance.lower()
  76
  77     blacklisted = False
  78     for domain in blacklist:
  79         if domain in instance:
  80             blacklisted = True
  81
  82     if blacklisted:
  83         continue
  84
  85     print(instance)
  86     try:
  87         c.execute(
  88             "SELECT domain FROM instances WHERE domain = ?", (instance,)
  89         )
  90         if c.fetchone() == None:
  91             c.execute(
  92                 "INSERT INTO instances SELECT ?, ?, ?",
  93                 (instance, get_hash(instance), get_type(instance)),
  94             )
  95         conn.commit()
  96     except Exception as e:
  97         print("error:", e, instance)
  98 conn.close()