1 from requests import get
2 from hashlib import sha256
4 from bs4 import BeautifulSoup
7 "user-agent": "fedi-block-api (https://gitlab.com/EnjuAihara/fedi-block-api)"
11 def get_mastodon_blocks(domain: str) -> dict:
13 "Suspended servers": [],
15 "Limited servers": [],
16 "Silenced servers": [],
21 get(f"https://{domain}/about/more", headers=headers, timeout=5).text,
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(
32 "domain": line.find("span").text,
33 "hash": line.find("span")["title"][9:],
34 "reason": line.find_all("td")[1].text.strip(),
38 "reject": blocks["Suspended servers"],
39 "media_removal": blocks["Filtered media"],
40 "federated_timeline_removal": blocks["Limited servers"]
41 + blocks["Silenced servers"],
45 def get_hash(domain: str) -> str:
46 return sha256(domain.encode("utf-8")).hexdigest()
49 def get_type(domain: str) -> str:
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)
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)
66 conn = sqlite3.connect("blocks.db")
70 "select domain, software from instances where software in ('pleroma', 'mastodon')"
73 for blocker, software in c.fetchall():
74 if software == "pleroma":
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"]}
86 for blocked in blocks:
90 "select domain from instances where domain = ?", (blocked,)
92 if c.fetchone() == None:
94 "insert into instances select ?, ?, ?",
95 (blocked, get_hash(blocked), get_type(blocked)),
98 "select * from blocks where blocker = ? and blocked = ? and block_level = ?",
99 (blocker, blocked, block_level),
101 if c.fetchone() == None:
103 "insert into blocks select ?, ?, '', ?",
104 (blocker, blocked, block_level),
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
115 for blocked, reason in info.items():
117 "update blocks set reason = ? where blocker = ? and blocked = ? and block_level = ?",
118 (reason["reason"], blocker, blocked, block_level),
121 except Exception as e:
122 print("error:", e, blocker)
123 elif software == "mastodon":
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:
132 "select hash from instances where hash = ?", (blocked_hash,)
134 if c.fetchone() == None:
136 "insert into instances select ?, ?, ?",
137 (blocked, get_hash(blocked), get_type(blocked)),
140 "select * from blocks where blocker = ? and blocked = ? and block_level = ?",
141 (blocker, blocked, block_level),
143 if c.fetchone() == None:
145 "insert into blocks select ?, ?, ?, ?",
148 blocked if blocked.count("*") <= 1 else blocked_hash,
154 except Exception as e:
155 print("error:", e, blocker)