2 from fastapi import FastAPI, Request, HTTPException, responses, Query
4 from hashlib import sha256
5 from fastapi.templating import Jinja2Templates
6 from requests import get
9 from datetime import datetime
10 from email import utils
12 with open("config.json") as f:
13 config = loads(f.read())
14 base_url = config["base_url"]
16 log_level = config["log_level"]
17 app = FastAPI(docs_url=base_url+"/docs", redoc_url=base_url+"/redoc")
18 templates = Jinja2Templates(directory=".")
20 def get_hash(domain: str) -> str:
21 return sha256(domain.encode("utf-8")).hexdigest()
23 @app.get(base_url+"/info")
25 conn = sqlite3.connect("blocks.db")
27 c.execute("SELECT (SELECT count(domain) FROM instances), (SELECT count(domain) FROM instances WHERE software in ('pleroma', 'mastodon', 'misskey', 'gotosocial', 'friendica')), (SELECT count(blocker) FROM blocks)")
28 known, indexed, blocks = c.fetchone()
31 "known_instances": known,
32 "indexed_instances": indexed,
33 "blocks_recorded": blocks
36 @app.get(base_url+"/top")
37 def top(blocked: int = None, blockers: int = None):
38 conn = sqlite3.connect("blocks.db")
40 if blocked == None and blockers == None:
41 raise HTTPException(status_code=400, detail="No filter specified")
44 raise HTTPException(status_code=400, detail="Too many results")
45 c.execute("SELECT blocked, count(blocked) FROM blocks WHERE block_level = 'reject' group by blocked order by count(blocked) desc limit ?", (blocked,))
46 elif blockers != None:
48 raise HTTPException(status_code=400, detail="Too many results")
49 c.execute("SELECT blocker, count(blocker) FROM blocks WHERE block_level = 'reject' group by blocker order by count(blocker) desc limit ?", (blockers,))
55 for domain, highscore in scores:
56 scoreboard.append({"domain": domain, "highscore": highscore})
60 @app.get(base_url+"/api")
61 def blocked(domain: str = None, reason: str = None, reverse: str = None):
62 if domain == None and reason == None and reverse == None:
63 raise HTTPException(status_code=400, detail="No filter specified")
65 reason = sub("(%|_)", "", reason)
67 raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
68 conn = sqlite3.connect("blocks.db")
71 wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
72 punycode = domain.encode('idna').decode('utf-8')
73 c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM blocks WHERE blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? order by first_added asc",
74 (domain, "*." + domain, wildchar, get_hash(domain), punycode, "*." + punycode))
76 c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM blocks WHERE blocker = ? order by first_added asc", (reverse,))
78 c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM blocks WHERE reason like ? AND reason != '' order by first_added asc", ("%"+reason+"%",))
83 for blocker, blocked, block_level, reason, first_added, last_seen in blocks:
84 entry = {"blocker": blocker, "blocked": blocked, "reason": reason, "first_added": first_added, "last_seen": last_seen}
85 if block_level in result:
86 result[block_level].append(entry)
88 result[block_level] = [entry]
92 @app.get(base_url+"/scoreboard")
93 def index(request: Request, blockers: int = None, blocked: int = None):
94 if blockers == None and blocked == None:
95 raise HTTPException(status_code=400, detail="No filter specified")
96 elif blockers != None:
97 scores = get(f"http://127.0.0.1:{port}{base_url}/top?blockers={blockers}")
99 scores = get(f"http://127.0.0.1:{port}{base_url}/top?blocked={blocked}")
102 raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
103 scores = scores.json()
104 return templates.TemplateResponse("index.html", {"request": request, "scoreboard": True, "blockers": blockers, "blocked": blocked, "scores": scores})
106 @app.get(base_url+"/")
107 def index(request: Request, domain: str = None, reason: str = None, reverse: str = None):
108 if domain == "" or reason == "" or reverse == "":
109 return responses.RedirectResponse("/")
112 if domain == None and reason == None and reverse == None:
113 info = get(f"http://127.0.0.1:{port}{base_url}/info")
115 raise HTTPException(status_code=info.status_code, detail=info.text)
118 blocks = get(f"http://127.0.0.1:{port}{base_url}/api?domain={domain}")
120 blocks = get(f"http://127.0.0.1:{port}{base_url}/api?reason={reason}")
121 elif reverse != None:
122 blocks = get(f"http://127.0.0.1:{port}{base_url}/api?reverse={reverse}")
125 raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
126 blocks = blocks.json()
127 for block_level in blocks:
128 for block in blocks[block_level]:
129 block["first_added"] = datetime.utcfromtimestamp(block["first_added"]).strftime('%Y-%m-%d %H:%M')
130 block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime('%Y-%m-%d %H:%M')
132 return templates.TemplateResponse("index.html", {"request": request, "domain": domain, "blocks": blocks, "reason": reason, "reverse": reverse, "info": info})
134 @app.get(base_url+"/api/mutual")
135 def mutual(domains: list[str] = Query()):
136 """Return 200 if federation is open between the two, 4xx otherwise"""
137 conn = sqlite3.connect('blocks.db')
140 "SELECT block_level FROM blocks " \
141 "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
142 "AND block_level = 'reject' " \
147 "aw": "*." + domains[0],
148 "bw": "*." + domains[1],
155 return responses.JSONResponse(status_code=418, content={})
157 return responses.JSONResponse(status_code=200, content={})
159 @app.get(base_url+"/rss")
160 def rss(request: Request, domain: str = None):
161 conn = sqlite3.connect("blocks.db")
164 wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
165 punycode = domain.encode('idna').decode('utf-8')
166 c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM blocks WHERE blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? order by first_added desc limit 50",
167 (domain, "*." + domain, wildchar, get_hash(domain), punycode, "*." + punycode))
169 c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM blocks order by first_added desc limit 50")
170 blocks = c.fetchall()
174 for blocker, blocked, block_level, reason, first_added, last_seen in blocks:
175 first_added = utils.format_datetime(datetime.fromtimestamp(first_added))
176 if reason == None or reason == '':
177 reason = "No reason provided."
179 reason = "Provided reason: '" + reason + "'"
180 result.append({"blocker": blocker, "blocked": blocked, "block_level": block_level, "reason": reason, "first_added": first_added})
182 timestamp = utils.format_datetime(datetime.now())
184 return templates.TemplateResponse("rss.xml", {"request": request, "timestamp": timestamp, "domain": domain, "blocks": result}, headers={"Content-Type": "application/rss+xml"})
186 if __name__ == "__main__":
187 uvicorn.run("api:app", host="127.0.0.1", port=port, log_level=log_level)