X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=api.py;h=30e06bdcdad875e96b92e72cd6c2d5150ab790b2;hb=967ddffa548f610101d9ef74df19f3aecafaed70;hp=43086fde79d45330ee6921a13fb435ed292f61dc;hpb=e4caf63d48037c1d5597155e68d82993c56bb9f1;p=fba.git diff --git a/api.py b/api.py index 43086fd..30e06bd 100644 --- a/api.py +++ b/api.py @@ -1,52 +1,41 @@ import uvicorn 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 +import re +import fba -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") +app = FastAPI(docs_url=fba.config["base_url"] + "/docs", redoc_url=fba.config["base_url"] + "/redoc") templates = Jinja2Templates(directory=".") -def get_hash(domain: str) -> str: - return sha256(domain.encode("utf-8")).hexdigest() - -@app.get(base_url+"/info") +@app.get(fba.config["base_url"] + "/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.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 = fba.c.fetchone() + return { "known_instances": known, "indexed_instances": indexed, - "blocks_recorded": blocks + "blocks_recorded": blocks, + "slogan": fba.config["slogan"] } -@app.get(base_url+"/top") +@app.get(fba.config["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,)) + fba.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() + fba.c.execute("SELECT blocker, count(blocker) FROM blocks WHERE block_level = 'reject' group by blocker ORDER BY count(blocker) DESC LIMIT ?", (blockers,)) + scores = fba.c.fetchall() scoreboard = [] print(scores) @@ -55,27 +44,24 @@ def top(blocked: int = None, blockers: int = None): 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.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, 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.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 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.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 = fba.c.fetchall() result = {} for blocker, blocked, block_level, reason, first_added, last_seen in blocks: @@ -87,37 +73,37 @@ def blocked(domain: str = None, reason: str = None, reverse: str = None): return result -@app.get(base_url+"/scoreboard") +@app.get(fba.config["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}") + scores = get(f"http://127.0.0.1:{fba.config['base_url']}{fba.config['base_url']}/top?blockers={blockers}") elif blocked != None: - scores = get(f"http://127.0.0.1:{port}{base_url}/top?blocked={blocked}") + scores = get(f"http://127.0.0.1:{fba.config['port']}{fba.config['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+"/") +@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 = get(f"http://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/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 = get(f"http://127.0.0.1:{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 = get(f"http://127.0.0.1:{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 = get(f"http://127.0.0.1:{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) @@ -129,29 +115,63 @@ def index(request: Request, domain: str = None, reason: str = None, reverse: str 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.c.execute( "SELECT block_level FROM blocks " \ - "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a)) " \ + "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() + res = fba.c.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.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, fba.get_hash(domain), punycode, "*." + punycode)) + else: + fba.c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM blocks ORDER BY first_added DESC LIMIT 50") + blocks = fba.c.fetchall() + + 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=fba.config["port"], log_level=fba.config["log_level"])