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 router = fastapi.FastAPI(docs_url=fba.config["base_url"] + "/docs", redoc_url=fba.config["base_url"] + "/redoc")
31 templates = Jinja2Templates(directory="templates")
33 @router.get(fba.config["base_url"] + "/api/info.json", response_class=JSONResponse)
35 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)")
36 known, indexed, blocks, errorous = fba.cursor.fetchone()
39 "known_instances" : known,
40 "indexed_instances" : indexed,
41 "blocks_recorded" : blocks,
42 "errorous_instances": errorous,
43 "slogan" : fba.config["slogan"]
46 @router.get(fba.config["base_url"] + "/api/top.json", response_class=JSONResponse)
47 def top(blocked: int = None, blockers: int = None, reference: int = None, software: int = None):
50 raise HTTPException(status_code=400, detail="Too many results")
51 fba.cursor.execute("SELECT blocked, COUNT(blocked) FROM blocks WHERE block_level = 'reject' GROUP BY blocked ORDER BY COUNT(blocked) DESC LIMIT ?", [blocked])
52 elif blockers != None:
54 raise HTTPException(status_code=400, detail="Too many results")
55 fba.cursor.execute("SELECT blocker, COUNT(blocker) FROM blocks WHERE block_level = 'reject' GROUP BY blocker ORDER BY COUNT(blocker) DESC LIMIT ?", [blockers])
56 elif reference != None:
58 raise HTTPException(status_code=400, detail="Too many results")
59 fba.cursor.execute("SELECT origin, COUNT(domain) FROM instances WHERE software IS NOT NULL GROUP BY origin ORDER BY COUNT(domain) DESC LIMIT ?", [reference])
60 elif software != None:
62 raise HTTPException(status_code=400, detail="Too many results")
63 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])
65 raise HTTPException(status_code=400, detail="No filter specified")
67 scores = fba.cursor.fetchall()
71 for domain, highscore in scores:
74 "highscore": highscore
79 @router.get(fba.config["base_url"] + "/api/index.json", response_class=JSONResponse)
80 def blocked(domain: str = None, reason: str = None, reverse: str = None):
81 if domain == None and reason == None and reverse == None:
82 raise HTTPException(status_code=400, detail="No filter specified")
85 reason = re.sub("(%|_)", "", reason)
87 raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
90 wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
91 punycode = domain.encode('idna').decode('utf-8')
92 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",
93 (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
95 fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE blocker = ? ORDER BY first_seen ASC", [reverse])
97 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 + "%"])
99 blocks = fba.cursor.fetchall()
102 for blocker, blocked, block_level, reason, first_seen, last_seen in blocks:
107 "first_seen": first_seen,
108 "last_seen" : last_seen
110 if block_level in result:
111 result[block_level].append(entry)
113 result[block_level] = [entry]
117 @router.get(fba.config["base_url"] + "/api/mutual.json", response_class=JSONResponse)
118 def mutual(domains: list[str] = Query()):
119 """Return 200 if federation is open between the two, 4xx otherwise"""
121 "SELECT block_level FROM blocks " \
122 "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
123 "AND block_level = 'reject' " \
128 "aw": "*." + domains[0],
129 "bw": "*." + domains[1],
132 res = fba.cursor.fetchone()
136 return JSONResponse(status_code=418, content={})
139 return JSONResponse(status_code=200, content={})
141 @router.get(fba.config["base_url"] + "/scoreboard")
142 def index(request: Request, blockers: int = None, blocked: int = None, reference: int = None, software: int = None):
145 if blockers != None and blockers > 0:
146 res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top.json?blockers={blockers}")
147 elif blocked != None and blocked > 0:
148 res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top.json?blocked={blocked}")
149 elif reference != None and reference > 0:
150 res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top.json?reference={reference}")
151 elif software != None and software > 0:
152 res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top.json?software={software}")
154 raise HTTPException(status_code=400, detail="No filter specified")
157 raise HTTPException(status_code=500, detail="Could not determine scores")
159 raise HTTPException(status_code=res.status_code, detail=res.text)
161 return templates.TemplateResponse("scoreboard.html", {
162 "base_url" : fba.config["base_url"],
163 "slogan" : fba.config["slogan"],
166 "blockers" : blockers,
168 "reference" : reference,
169 "software" : software,
170 "scores" : res.json()
173 @router.get(fba.config["base_url"] + "/")
174 def index(request: Request, domain: str = None, reason: str = None, reverse: str = None):
175 if domain == "" or reason == "" or reverse == "":
176 return fastapi.responses.RedirectResponse("/")
181 if domain == None and reason == None and reverse == None:
182 info = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/info.json")
185 raise HTTPException(status_code=info.status_code, detail=info.text)
189 blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/index.json?domain={domain}")
191 blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/index.json?reason={reason}")
192 elif reverse != None:
193 blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/index.json?reverse={reverse}")
197 raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
198 blocks = blocks.json()
199 for block_level in blocks:
200 for block in blocks[block_level]:
201 block["first_seen"] = datetime.utcfromtimestamp(block["first_seen"]).strftime('%Y-%m-%d %H:%M')
202 block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime('%Y-%m-%d %H:%M')
204 return templates.TemplateResponse("index.html", {
213 @router.get(fba.config["base_url"] + "/rss")
214 def rss(request: Request, domain: str = None):
216 wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
217 punycode = domain.encode('idna').decode('utf-8')
218 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",
219 (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
221 fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks ORDER BY first_seen DESC LIMIT 50")
223 blocks = fba.cursor.fetchall()
226 for blocker, blocked, block_level, reason, first_seen, last_seen in blocks:
227 first_seen = utils.format_datetime(datetime.fromtimestamp(first_seen))
228 if reason == None or reason == '':
229 reason = "No reason provided."
231 reason = "Provided reason: '" + reason + "'"
236 "block_level": block_level,
238 "first_seen" : first_seen
241 timestamp = utils.format_datetime(datetime.now())
243 return templates.TemplateResponse("rss.xml", {
245 "timestamp": timestamp,
249 "Content-Type": "routerlication/rss+xml"
252 @router.get(fba.config["base_url"] + "/robots.txt", response_class=PlainTextResponse)
253 def robots(request: Request):
254 return templates.TemplateResponse("robots.txt", {
256 "base_url": fba.config["base_url"]
259 if __name__ == "__main__":
260 uvicorn.run("api:router", host=fba.config["host"], port=fba.config["port"], log_level=fba.config["log_level"])