import uvicorn
-from fastapi import FastAPI, Request, HTTPException, responses
+from fastapi import FastAPI, Request, HTTPException, responses, Query
import sqlite3
from hashlib import sha256
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"]
+ log_level = config["log_level"]
app = FastAPI(docs_url=base_url+"/docs", redoc_url=base_url+"/redoc")
templates = Jinja2Templates(directory=".")
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')), (select count(blocker) from blocks)")
+ 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()
return {
"known_instances": known,
"indexed_instances": indexed,
- "blocks_recorded": blocks,
- "source_code": "https://git.kiwifarms.net/mint/fedi-block-api",
+ "blocks_recorded": blocks
}
+@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:
+ 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,))
+ 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()
+
+ scoreboard = []
+ print(scores)
+ for domain, highscore in scores:
+ scoreboard.append({"domain": domain, "highscore": highscore})
+
+ return scoreboard
+
@app.get(base_url+"/api")
-def blocked(domain: str = None, reason: str = None):
- if domain == None and reason == None:
+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)
+ 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, block_level, reason from blocks where blocked = ? or blocked = ? or blocked = ? or blocked = ? or blocked = ? or blocked = ?",
+ 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))
+ 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,))
else:
- c.execute("select * from blocks where reason like ? and reason != ''", ("%"+reason+"%",))
+ 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()
- conn.close()
+ c.close()
result = {}
- reasons = {}
- if domain != None:
- for domain, block_level, reason in blocks:
- if block_level in result:
- result[block_level].append(domain)
- else:
- result[block_level] = [domain]
-
- if reason != "":
- if block_level in reasons:
- reasons[block_level][domain] = reason
- else:
- reasons[block_level] = {domain: reason}
- return {"blocks": result, "reasons": reasons}
-
- for blocker, blocked, reason, block_level in blocks:
+ 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}
if block_level in result:
- result[block_level].append({"blocker": blocker, "blocked": blocked, "reason": reason})
+ result[block_level].append(entry)
else:
- result[block_level] = [{"blocker": blocker, "blocked": blocked, "reason": reason}]
- return {"blocks": result}
+ result[block_level] = [entry]
+
+ return result
+
+@app.get(base_url+"/scoreboard")
+def index(request: Request, blockers: int = None, blocked: int = None):
+ if blockers == None and blocked == None:
+ 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+"/")
-def index(request: Request, domain: str = None, reason: str = None):
- if domain == "" or reason == "":
+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:
+ if domain == None and reason == None and reverse == None:
info = get(f"http://127.0.0.1:{port}{base_url}/info")
if not info.ok:
raise HTTPException(status_code=info.status_code, detail=info.text)
blocks = get(f"http://127.0.0.1:{port}{base_url}/api?domain={domain}")
elif reason != None:
blocks = get(f"http://127.0.0.1:{port}{base_url}/api?reason={reason}")
+ elif reverse != None:
+ blocks = get(f"http://127.0.0.1:{port}{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()
- return templates.TemplateResponse("index.html", {"request": request, "domain": domain, "blocks": blocks, "reason": reason, "info": info})
+ 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["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})
+
+@app.get(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(
+ "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],
+ "aw": "*." + domains[0],
+ "bw": "*." + domains[1],
+ },
+ )
+ res = c.fetchone()
+ c.close()
+ if res is not None:
+ # Blocks found
+ return responses.JSONResponse(status_code=418, content={})
+ # No known blocks
+ return responses.JSONResponse(status_code=200, content={})
+
+@app.get(base_url+"/rss")
+def rss(request: Request, domain: str = None):
+ 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 DESC LIMIT 50",
+ (domain, "*." + domain, wildchar, get_hash(domain), punycode, "*." + punycode))
+ else:
+ c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM blocks ORDER BY first_added DESC LIMIT 50")
+ blocks = c.fetchall()
+ c.close()
+
+ result = []
+ for blocker, blocked, block_level, reason, first_added, last_seen in blocks:
+ first_added = utils.format_datetime(datetime.fromtimestamp(first_added))
+ 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_added": first_added})
+
+ 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="127.0.0.1", port=port, log_level="info")
+ uvicorn.run("api:app", host="127.0.0.1", port=port, log_level=log_level)
projects / fba.git / blobdiff