1 # Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
2 # Copyright (C) 2023 Free Software Foundation
4 # This program is free software: you can redistribute it and/or modify
5 # it under the terms of the GNU Affero General Public License as published
6 # by the Free Software Foundation, either version 3 of the License, or
7 # (at your option) any later version.
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 # GNU Affero General Public License for more details.
14 # You should have received a copy of the GNU Affero General Public License
15 # along with this program. If not, see <https://www.gnu.org/licenses/>.
17 from fastapi import Request, HTTPException, Query
18 from fastapi.responses import JSONResponse
19 from fastapi.responses import PlainTextResponse
20 from fastapi.templating import Jinja2Templates
21 from datetime import datetime
22 from email import utils
32 router = fastapi.FastAPI(docs_url=config.get("base_url") + "/docs", redoc_url=config.get("base_url") + "/redoc")
33 templates = Jinja2Templates(directory="templates")
35 @router.get(config.get("base_url") + "/api/info.json", response_class=JSONResponse)
37 fba.cursor.execute("SELECT (SELECT COUNT(domain) FROM instances), (SELECT COUNT(domain) FROM instances WHERE software IN ('pleroma', 'mastodon', 'misskey', 'gotosocial', 'friendica', 'bookwyrm', 'takahe', 'peertube')), (SELECT COUNT(blocker) FROM blocks), (SELECT COUNT(domain) FROM instances WHERE last_status_code IS NOT NULL)")
38 known, indexed, blocklist, errorous = fba.cursor.fetchone()
41 "known_instances" : known,
42 "indexed_instances" : indexed,
43 "blocks_recorded" : blocklist,
44 "errorous_instances": errorous,
45 "slogan" : config.get("slogan")
48 @router.get(config.get("base_url") + "/api/top.json", response_class=JSONResponse)
49 def top(blocked: int = None, blockers: int = None, reference: int = None, software: int = None, originator: int = None):
52 raise HTTPException(status_code=400, detail="Too many results")
53 fba.cursor.execute("SELECT blocked, COUNT(blocked) FROM blocks WHERE block_level = 'reject' GROUP BY blocked ORDER BY COUNT(blocked) DESC LIMIT ?", [blocked])
54 elif blockers != None:
56 raise HTTPException(status_code=400, detail="Too many results")
57 fba.cursor.execute("SELECT blocker, COUNT(blocker) FROM blocks WHERE block_level = 'reject' GROUP BY blocker ORDER BY COUNT(blocker) DESC LIMIT ?", [blockers])
58 elif reference != None:
60 raise HTTPException(status_code=400, detail="Too many results")
61 fba.cursor.execute("SELECT origin, COUNT(domain) FROM instances WHERE software IS NOT NULL GROUP BY origin ORDER BY COUNT(domain) DESC LIMIT ?", [reference])
62 elif software != None:
64 raise HTTPException(status_code=400, detail="Too many results")
65 fba.cursor.execute("SELECT software, COUNT(domain) FROM instances WHERE software IS NOT NULL GROUP BY software ORDER BY COUNT(domain) DESC, software ASC LIMIT ?", [software])
66 elif originator != None:
68 raise HTTPException(status_code=400, detail="Too many results")
69 fba.cursor.execute("SELECT originator, COUNT(domain) FROM instances WHERE originator IS NOT NULL GROUP BY originator ORDER BY COUNT(domain) DESC, originator ASC LIMIT ?", [originator])
71 raise HTTPException(status_code=400, detail="No filter specified")
73 scores = fba.cursor.fetchall()
77 for domain, highscore in scores:
80 "highscore": highscore
85 @router.get(config.get("base_url") + "/api/index.json", response_class=JSONResponse)
86 def blocked(domain: str = None, reason: str = None, reverse: str = None):
87 if domain == None and reason == None and reverse == None:
88 raise HTTPException(status_code=400, detail="No filter specified")
91 reason = re.sub("(%|_)", "", reason)
93 raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
96 wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
97 punycode = domain.encode('idna').decode('utf-8')
98 fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? ORDER BY first_seen ASC",
99 (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
100 elif reverse != None:
101 fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE blocker = ? ORDER BY first_seen ASC", [reverse])
103 fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE reason like ? AND reason != '' ORDER BY first_seen ASC", ["%" + reason + "%"])
105 blocklist = fba.cursor.fetchall()
108 for blocker, blocked, block_level, reason, first_seen, last_seen in blocklist:
113 "first_seen": first_seen,
114 "last_seen" : last_seen
116 if block_level in result:
117 result[block_level].append(entry)
119 result[block_level] = [entry]
123 @router.get(config.get("base_url") + "/api/mutual.json", response_class=JSONResponse)
124 def mutual(domains: list[str] = Query()):
125 """Return 200 if federation is open between the two, 4xx otherwise"""
127 "SELECT block_level FROM blocks " \
128 "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
129 "AND block_level = 'reject' " \
134 "aw": "*." + domains[0],
135 "bw": "*." + domains[1],
138 response = fba.cursor.fetchone()
140 if response is not None:
142 return JSONResponse(status_code=418, content={})
145 return JSONResponse(status_code=200, content={})
147 @router.get(config.get("base_url") + "/scoreboard")
148 def index(request: Request, blockers: int = None, blocked: int = None, reference: int = None, software: int = None, originator: int = None):
151 if blockers != None and blockers > 0:
152 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?blockers={blockers}")
153 elif blocked != None and blocked > 0:
154 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?blocked={blocked}")
155 elif reference != None and reference > 0:
156 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?reference={reference}")
157 elif software != None and software > 0:
158 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?software={software}")
159 elif originator != None and originator > 0:
160 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?originator={originator}")
162 raise HTTPException(status_code=400, detail="No filter specified")
165 raise HTTPException(status_code=500, detail="Could not determine scores")
166 elif not response.ok:
167 raise HTTPException(status_code=response.status_code, detail=response.text)
169 return templates.TemplateResponse("scoreboard.html", {
170 "base_url" : config.get("base_url"),
171 "slogan" : config.get("slogan"),
174 "blockers" : blockers,
176 "reference" : reference,
177 "software" : software,
178 "originator": originator,
179 "scores" : fba.json_from_response(response)
182 @router.get(config.get("base_url") + "/")
183 def index(request: Request, domain: str = None, reason: str = None, reverse: str = None):
184 if domain == "" or reason == "" or reverse == "":
185 return fastapi.responses.RedirectResponse("/")
187 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/info.json")
190 raise HTTPException(status_code=response.status_code, detail=response.text)
192 info = response.json()
196 if not validators.domain(domain):
197 raise HTTPException(status_code=500, detail="Invalid domain")
199 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/index.json?domain={domain}")
201 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/index.json?reason={reason}")
202 elif reverse != None:
203 if not validators.domain(reverse):
204 raise HTTPException(status_code=500, detail="Invalid domain")
206 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/index.json?reverse={reverse}")
210 raise HTTPException(status_code=response.status_code, detail=response.text)
211 blocklist = response.json()
212 for block_level in blocklist:
213 for block in blocklist[block_level]:
214 block["first_seen"] = datetime.utcfromtimestamp(block["first_seen"]).strftime(config.get("timestamp_format"))
215 block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime(config.get("timestamp_format"))
217 return templates.TemplateResponse("index.html", {
220 "blocks" : blocklist,
226 @router.get(config.get("base_url") + "/rss")
227 def rss(request: Request, domain: str = None):
229 wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
230 punycode = domain.encode('idna').decode('utf-8')
231 fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? ORDER BY first_seen DESC LIMIT 50",
232 (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
234 fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks ORDER BY first_seen DESC LIMIT 50")
236 result = fba.cursor.fetchall()
239 for blocker, blocked, block_level, reason, first_seen, last_seen in result:
240 first_seen = utils.format_datetime(datetime.fromtimestamp(first_seen))
241 if reason == None or reason == '':
242 reason = "No reason provided."
244 reason = "Provided reason: '" + reason + "'"
249 "block_level": block_level,
251 "first_seen" : first_seen
254 return templates.TemplateResponse("rss.xml", {
256 "timestamp": utils.format_datetime(datetime.now()),
258 "hostname" : config.get("hostname"),
261 "Content-Type": "routerlication/rss+xml"
264 @router.get(config.get("base_url") + "/robots.txt", response_class=PlainTextResponse)
265 def robots(request: Request):
266 return templates.TemplateResponse("robots.txt", {
268 "base_url": config.get("base_url")
271 if __name__ == "__main__":
272 uvicorn.run("api:router", host=config.get("host"), port=config.get("port"), log_level=config.get("log_level"))