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
30 from fba import config
33 router = fastapi.FastAPI(docs_url=config.get("base_url") + "/docs", redoc_url=config.get("base_url") + "/redoc")
34 templates = Jinja2Templates(directory="templates")
36 @router.get(config.get("base_url") + "/api/info.json", response_class=JSONResponse)
38 fba.cursor.execute("SELECT (SELECT COUNT(domain) FROM instances), (SELECT COUNT(domain) FROM instances WHERE software IN ('pleroma', 'mastodon', 'misskey', 'friendica', 'bookwyrm', 'takahe', 'peertube')), (SELECT COUNT(blocker) FROM blocks), (SELECT COUNT(domain) FROM instances WHERE last_status_code IS NOT NULL)")
39 known, indexed, blocklist, errorous = fba.cursor.fetchone()
42 "known_instances" : known,
43 "indexed_instances" : indexed,
44 "blocks_recorded" : blocklist,
45 "errorous_instances": errorous,
46 "slogan" : config.get("slogan")
49 @router.get(config.get("base_url") + "/api/top.json", response_class=JSONResponse)
50 def top(blocked: int = None, blockers: int = None, reference: int = None, software: int = None, command: int = None, error_code: int = None):
51 if blocked is not None:
53 raise HTTPException(status_code=400, detail="Too many results")
54 fba.cursor.execute("SELECT blocked, COUNT(blocked) FROM blocks WHERE block_level = 'reject' GROUP BY blocked ORDER BY COUNT(blocked) DESC LIMIT ?", [blocked])
55 elif blockers is not None:
57 raise HTTPException(status_code=400, detail="Too many results")
58 fba.cursor.execute("SELECT blocker, COUNT(blocker) FROM blocks WHERE block_level = 'reject' GROUP BY blocker ORDER BY COUNT(blocker) DESC LIMIT ?", [blockers])
59 elif reference is not None:
61 raise HTTPException(status_code=400, detail="Too many results")
62 fba.cursor.execute("SELECT origin, COUNT(domain) FROM instances WHERE software IS NOT NULL GROUP BY origin ORDER BY COUNT(domain) DESC LIMIT ?", [reference])
63 elif software is not None:
65 raise HTTPException(status_code=400, detail="Too many results")
66 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])
67 elif command is not None:
69 raise HTTPException(status_code=400, detail="Too many results")
70 fba.cursor.execute("SELECT command, COUNT(domain) FROM instances WHERE command IS NOT NULL GROUP BY command ORDER BY COUNT(domain) DESC, command ASC LIMIT ?", [command])
71 elif error_code is not None:
73 raise HTTPException(status_code=400, detail="Too many results")
74 fba.cursor.execute("SELECT last_status_code, COUNT(domain) AS cnt FROM instances WHERE last_status_code IS NOT NULL AND last_status_code != '200' GROUP BY last_status_code ORDER BY cnt DESC LIMIT ?", [error_code])
76 raise HTTPException(status_code=400, detail="No filter specified")
78 scores = fba.cursor.fetchall()
82 for domain, highscore in scores:
85 "highscore": highscore
90 @router.get(config.get("base_url") + "/api/index.json", response_class=JSONResponse)
91 def blocked(domain: str = None, reason: str = None, reverse: str = None):
92 if domain is None and reason is None and reverse is None:
93 raise HTTPException(status_code=400, detail="No filter specified")
95 if reason is not None:
96 reason = re.sub("(%|_)", "", reason)
98 raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
100 if domain is not None:
101 wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
102 punycode = domain.encode('idna').decode('utf-8')
103 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",
104 (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
105 elif reverse is not None:
106 fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE blocker = ? ORDER BY first_seen ASC", [reverse])
108 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 + "%"])
110 blocklist = fba.cursor.fetchall()
113 for blocker, blocked, block_level, reason, first_seen, last_seen in blocklist:
114 if reason is not None and reason != "":
115 reason = reason.replace(",", " ").replace(" ", " ")
121 "first_seen": first_seen,
122 "last_seen" : last_seen
125 if block_level in result:
126 result[block_level].append(entry)
128 result[block_level] = [entry]
132 @router.get(config.get("base_url") + "/api/mutual.json", response_class=JSONResponse)
133 def mutual(domains: list[str] = Query()):
134 """Return 200 if federation is open between the two, 4xx otherwise"""
136 "SELECT block_level FROM blocks " \
137 "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
138 "AND block_level = 'reject' " \
143 "aw": "*." + domains[0],
144 "bw": "*." + domains[1],
147 response = fba.cursor.fetchone()
149 if response is not None:
151 return JSONResponse(status_code=418, content={})
154 return JSONResponse(status_code=200, content={})
156 @router.get(config.get("base_url") + "/scoreboard")
157 def scoreboard(request: Request, blockers: int = None, blocked: int = None, reference: int = None, software: int = None, command: int = None, error_code: int = None):
158 if blockers is not None and blockers > 0:
159 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?blockers={blockers}")
160 elif blocked is not None and blocked > 0:
161 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?blocked={blocked}")
162 elif reference is not None and reference > 0:
163 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?reference={reference}")
164 elif software is not None and software > 0:
165 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?software={software}")
166 elif command is not None and command > 0:
167 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?command={command}")
168 elif error_code is not None and error_code > 0:
169 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/top.json?error_code={error_code}")
171 raise HTTPException(status_code=400, detail="No filter specified")
174 raise HTTPException(status_code=500, detail="Could not determine scores")
175 elif not response.ok:
176 raise HTTPException(status_code=response.status_code, detail=response.text)
178 return templates.TemplateResponse("views/scoreboard.html", {
179 "base_url" : config.get("base_url"),
180 "slogan" : config.get("slogan"),
183 "blockers" : blockers,
185 "reference" : reference,
186 "software" : software,
188 "error_code": error_code,
189 "scores" : fba.json_from_response(response)
192 @router.get(config.get("base_url") + "/")
193 def index(request: Request):
195 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/info.json")
198 raise HTTPException(status_code=response.status_code, detail=response.text)
200 return templates.TemplateResponse("views/index.html", {
202 "info" : response.json()
205 @router.get(config.get("base_url") + "/top")
206 def top(request: Request, domain: str = None, reason: str = None, reverse: str = None):
207 if domain == "" or reason == "" or reverse == "":
208 raise HTTPException(status_code=500, detail="Insufficient parameter provided")
210 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/info.json")
213 raise HTTPException(status_code=response.status_code, detail=response.text)
215 info = response.json()
218 if domain is not None:
219 if not validators.domain(domain):
220 raise HTTPException(status_code=500, detail="Invalid domain")
222 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/index.json?domain={domain}")
223 elif reason is not None:
224 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/index.json?reason={reason}")
225 elif reverse is not None:
226 if not validators.domain(reverse):
227 raise HTTPException(status_code=500, detail="Invalid domain")
229 response = requests.get(f"http://{config.get('host')}:{config.get('port')}{config.get('base_url')}/api/index.json?reverse={reverse}")
231 if response is not None:
233 raise HTTPException(status_code=response.status_code, detail=response.text)
234 blocklist = response.json()
235 for block_level in blocklist:
236 for block in blocklist[block_level]:
237 block["first_seen"] = datetime.utcfromtimestamp(block["first_seen"]).strftime(config.get("timestamp_format"))
238 block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime(config.get("timestamp_format"))
240 return templates.TemplateResponse("views/top.html", {
243 "blocks" : blocklist,
249 @router.get(config.get("base_url") + "/rss")
250 def rss(request: Request, domain: str = None):
251 if domain is not None:
252 wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
253 punycode = domain.encode('idna').decode('utf-8')
254 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",
255 (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
257 fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks ORDER BY first_seen DESC LIMIT 50")
259 result = fba.cursor.fetchall()
262 for blocker, blocked, block_level, reason, first_seen, last_seen in result:
263 first_seen = utils.format_datetime(datetime.fromtimestamp(first_seen))
264 if reason is None or reason == '':
265 reason = "No reason provided."
267 reason = "Provided reason: '" + reason + "'"
272 "block_level": block_level,
274 "first_seen" : first_seen,
275 "last_seen" : last_seen,
278 return templates.TemplateResponse("rss.xml", {
280 "timestamp": utils.format_datetime(datetime.now()),
282 "hostname" : config.get("hostname"),
285 "Content-Type": "routerlication/rss+xml"
288 @router.get(config.get("base_url") + "/robots.txt", response_class=PlainTextResponse)
289 def robots(request: Request):
290 return templates.TemplateResponse("robots.txt", {
292 "base_url": config.get("base_url")
295 if __name__ == "__main__":
296 uvicorn.run("api:router", host=config.get("host"), port=config.get("port"), log_level=config.get("log_level"))