-import uvicorn
-from fastapi import FastAPI, Request, HTTPException, responses, Query
-import sqlite3
-from hashlib import sha256
+# Fedi API Block - An aggregator for fetching blocking data from fediverse nodes
+# Copyright (C) 2023 Free Software Foundation
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+from fastapi import Request, HTTPException, responses, Query
from fastapi.templating import Jinja2Templates
-from requests import get
-from json import loads
-from re import sub
from datetime import datetime
+from email import utils
-with open("config.json") as f:
- config = loads(f.read())
- base_url = config["base_url"]
- port = config["port"]
-app = FastAPI(docs_url=base_url+"/docs", redoc_url=base_url+"/redoc")
-templates = Jinja2Templates(directory=".")
+import fastapi
+import uvicorn
+import requests
+import re
+import fba
-def get_hash(domain: str) -> str:
- return sha256(domain.encode("utf-8")).hexdigest()
+app = fastapi.FastAPI(docs_url=fba.config["base_url"] + "/docs", redoc_url=fba.config["base_url"] + "/redoc")
+templates = Jinja2Templates(directory=".")
-@app.get(base_url+"/info")
+@app.get(fba.config["base_url"] + "/api/info")
def info():
- conn = sqlite3.connect("blocks.db")
- c = conn.cursor()
- 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)")
- known, indexed, blocks = c.fetchone()
- c.close()
+ 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)")
+ known, indexed, blocks, errorous = fba.cursor.fetchone()
+
return {
- "known_instances": known,
- "indexed_instances": indexed,
- "blocks_recorded": blocks
+ "known_instances" : known,
+ "indexed_instances" : indexed,
+ "blocks_recorded" : blocks,
+ "errorous_instances": errorous,
+ "slogan" : fba.config["slogan"]
}
-@app.get(base_url+"/top")
-def top(blocked: int = None, blockers: int = None):
- conn = sqlite3.connect("blocks.db")
- c = conn.cursor()
- if blocked == None and blockers == None:
- raise HTTPException(status_code=400, detail="No filter specified")
- elif blocked != None:
+@app.get(fba.config["base_url"] + "/api/top")
+def top(blocked: int = None, blockers: int = None, reference: int = None, software: int = None):
+ if blocked != None:
if blocked > 500:
raise HTTPException(status_code=400, detail="Too many results")
- c.execute("select blocked, count(blocked) from blocks where block_level = 'reject' group by blocked order by count(blocked) desc limit ?", (blocked,))
+ fba.cursor.execute("SELECT blocked, COUNT(blocked) FROM blocks WHERE block_level = 'reject' GROUP BY blocked ORDER BY COUNT(blocked) DESC LIMIT ?", [blocked])
elif blockers != None:
if blockers > 500:
raise HTTPException(status_code=400, detail="Too many results")
- c.execute("select blocker, count(blocker) from blocks where block_level = 'reject' group by blocker order by count(blocker) desc limit ?", (blockers,))
- scores = c.fetchall()
- c.close()
+ fba.cursor.execute("SELECT blocker, COUNT(blocker) FROM blocks WHERE block_level = 'reject' GROUP BY blocker ORDER BY COUNT(blocker) DESC LIMIT ?", [blockers])
+ elif reference != None:
+ if reference > 500:
+ raise HTTPException(status_code=400, detail="Too many results")
+ fba.cursor.execute("SELECT origin, COUNT(domain) FROM instances GROUP BY origin ORDER BY COUNT(domain) DESC LIMIT ?", [reference])
+ elif software != None:
+ if software > 500:
+ raise HTTPException(status_code=400, detail="Too many results")
+ 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])
+ else:
+ raise HTTPException(status_code=400, detail="No filter specified")
+
+ scores = fba.cursor.fetchall()
scoreboard = []
- print(scores)
+
for domain, highscore in scores:
- scoreboard.append({"domain": domain, "highscore": highscore})
+ scoreboard.append({
+ "domain" : domain,
+ "highscore": highscore
+ })
return scoreboard
-@app.get(base_url+"/api")
+@app.get(fba.config["base_url"] + "/api")
def blocked(domain: str = None, reason: str = None, reverse: str = None):
if domain == None and reason == None and reverse == None:
raise HTTPException(status_code=400, detail="No filter specified")
+
if reason != None:
- reason = sub("(%|_)", "", reason)
+ reason = re.sub("(%|_)", "", reason)
if len(reason) < 3:
raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
- conn = sqlite3.connect("blocks.db")
- c = conn.cursor()
+
if domain != None:
wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
punycode = domain.encode('idna').decode('utf-8')
- 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",
- (domain, "*." + domain, wildchar, get_hash(domain), punycode, "*." + punycode))
+ 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",
+ (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
elif reverse != None:
- c.execute("select blocker, blocked, block_level, reason, first_added, last_seen from blocks where blocker = ? order by first_added asc", (reverse,))
+ fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks WHERE blocker = ? ORDER BY first_seen ASC", [reverse])
else:
- 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+"%",))
- blocks = c.fetchall()
- c.close()
+ 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 + "%"])
+
+ blocks = fba.cursor.fetchall()
result = {}
- for blocker, blocked, block_level, reason, first_added, last_seen in blocks:
- entry = {"blocker": blocker, "blocked": blocked, "reason": reason, "first_added": first_added, "last_seen": last_seen}
+ for blocker, blocked, block_level, reason, first_seen, last_seen in blocks:
+ entry = {
+ "blocker" : blocker,
+ "blocked" : blocked,
+ "reason" : reason,
+ "first_seen": first_seen,
+ "last_seen" : last_seen
+ }
if block_level in result:
result[block_level].append(entry)
else:
return result
-@app.get(base_url+"/scoreboard")
-def index(request: Request, blockers: int = None, blocked: int = None):
- if blockers == None and blocked == None:
+@app.get(fba.config["base_url"] + "/scoreboard")
+def index(request: Request, blockers: int = None, blocked: int = None, reference: int = None, software: int = None):
+ scores = None
+
+ if blockers != None and blockers > 0:
+ res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top?blockers={blockers}")
+ elif blocked != None and blocked > 0:
+ res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top?blocked={blocked}")
+ elif reference != None and reference > 0:
+ res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top?reference={reference}")
+ elif software != None and software > 0:
+ res = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/top?software={software}")
+ else:
raise HTTPException(status_code=400, detail="No filter specified")
- elif blockers != None:
- scores = get(f"http://127.0.0.1:{port}{base_url}/top?blockers={blockers}")
- elif blocked != None:
- scores = get(f"http://127.0.0.1:{port}{base_url}/top?blocked={blocked}")
- if scores != None:
- if not scores.ok:
- raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
- scores = scores.json()
- return templates.TemplateResponse("index.html", {"request": request, "scoreboard": True, "blockers": blockers, "blocked": blocked, "scores": scores})
-@app.get(base_url+"/")
+ if res == None:
+ raise HTTPException(status_code=500, detail="Could not determine scores")
+ elif not res.ok:
+ raise HTTPException(status_code=res.status_code, detail=res.text)
+
+ return templates.TemplateResponse("index.html", {
+ "base_url" : fba.config["base_url"],
+ "request" : request,
+ "scoreboard": True,
+ "blockers" : blockers,
+ "blocked" : blocked,
+ "reference" : reference,
+ "software" : software,
+ "scores" : res.json()
+ })
+
+@app.get(fba.config["base_url"] + "/")
def index(request: Request, domain: str = None, reason: str = None, reverse: str = None):
if domain == "" or reason == "" or reverse == "":
return responses.RedirectResponse("/")
+
info = None
blocks = None
+
if domain == None and reason == None and reverse == None:
- info = get(f"http://127.0.0.1:{port}{base_url}/info")
+ info = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api/info")
+
if not info.ok:
raise HTTPException(status_code=info.status_code, detail=info.text)
+
info = info.json()
elif domain != None:
- blocks = get(f"http://127.0.0.1:{port}{base_url}/api?domain={domain}")
+ blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api?domain={domain}")
elif reason != None:
- blocks = get(f"http://127.0.0.1:{port}{base_url}/api?reason={reason}")
+ blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api?reason={reason}")
elif reverse != None:
- blocks = get(f"http://127.0.0.1:{port}{base_url}/api?reverse={reverse}")
+ blocks = requests.get(f"http://{fba.config['host']}:{fba.config['port']}{fba.config['base_url']}/api?reverse={reverse}")
+
if blocks != None:
if not blocks.ok:
raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
blocks = blocks.json()
for block_level in blocks:
for block in blocks[block_level]:
- block["first_added"] = datetime.utcfromtimestamp(block["first_added"]).strftime('%Y-%m-%d %H:%M')
+ block["first_seen"] = datetime.utcfromtimestamp(block["first_seen"]).strftime('%Y-%m-%d %H:%M')
block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime('%Y-%m-%d %H:%M')
- return templates.TemplateResponse("index.html", {"request": request, "domain": domain, "blocks": blocks, "reason": reason, "reverse": reverse, "info": info})
+ return templates.TemplateResponse("index.html", {
+ "request": request,
+ "domain" : domain,
+ "blocks" : blocks,
+ "reason" : reason,
+ "reverse": reverse,
+ "info" : info
+ })
-@app.get(base_url+"/api/mutual")
+@app.get(fba.config["base_url"] + "/api/mutual")
def mutual(domains: list[str] = Query()):
"""Return 200 if federation is open between the two, 4xx otherwise"""
- conn = sqlite3.connect('blocks.db')
- c = conn.cursor()
- c.execute(
+ fba.cursor.execute(
"SELECT block_level FROM blocks " \
"WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
"AND block_level = 'reject' " \
"LIMIT 1",
{
- "a": domains[0],
- "b": domains[1],
+ "a" : domains[0],
+ "b" : domains[1],
"aw": "*." + domains[0],
"bw": "*." + domains[1],
},
)
- res = c.fetchone()
- c.close()
+ res = fba.cursor.fetchone()
+
if res is not None:
# Blocks found
return responses.JSONResponse(status_code=418, content={})
+
# No known blocks
return responses.JSONResponse(status_code=200, content={})
-if __name__ == "__main__":
- uvicorn.run("api:app", host="127.0.0.1", port=port, log_level="info")
+@app.get(fba.config["base_url"] + "/rss")
+def rss(request: Request, domain: str = None):
+ if domain != None:
+ wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
+ punycode = domain.encode('idna').decode('utf-8')
+ 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",
+ (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
+ else:
+ fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_seen, last_seen FROM blocks ORDER BY first_seen DESC LIMIT 50")
+
+ blocks = fba.cursor.fetchall()
+
+ result = []
+ for blocker, blocked, block_level, reason, first_seen, last_seen in blocks:
+ first_seen = utils.format_datetime(datetime.fromtimestamp(first_seen))
+ if reason == None or reason == '':
+ reason = "No reason provided."
+ else:
+ reason = "Provided reason: '" + reason + "'"
+
+ result.append({
+ "blocker" : blocker,
+ "blocked" : blocked,
+ "block_level": block_level,
+ "reason" : reason,
+ "first_seen" : first_seen
+ })
+
+ timestamp = utils.format_datetime(datetime.now())
+ return templates.TemplateResponse("rss.xml", {
+ "request" : request,
+ "timestamp": timestamp,
+ "domain" : domain,
+ "blocks" : result
+ }, headers={
+ "Content-Type": "application/rss+xml"
+ })
+
+if __name__ == "__main__":
+ uvicorn.run("api:app", host=fba.config["host"], port=fba.config["port"], log_level=fba.config["log_level"])