X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=api.py;h=7b18ead8947b0e798ae1c31a4d77672c44cdfa1f;hb=ab9778ff76580cb17aa79d3e654466a9b1e06728;hp=37b990d8851a4eeca124f2396cf95cf153d8514a;hpb=5a8dd9c68ef847b5652c206f445785b5a6f9c5ed;p=fba.git diff --git a/api.py b/api.py index 37b990d..7b18ead 100644 --- a/api.py +++ b/api.py @@ -1,30 +1,48 @@ -from fastapi import FastAPI, Request, HTTPException, responses, Query +# 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 . + +from fastapi import Request, HTTPException, responses, Query +from fastapi.responses import PlainTextResponse from fastapi.templating import Jinja2Templates from datetime import datetime from email import utils +import fastapi import uvicorn import requests -import sqlite3 import re import fba -app = FastAPI(docs_url=fba.config["base_url"] + "/docs", redoc_url=fba.config["base_url"] + "/redoc") -templates = Jinja2Templates(directory=".") +router = fastapi.FastAPI(docs_url=fba.config["base_url"] + "/docs", redoc_url=fba.config["base_url"] + "/redoc") +templates = Jinja2Templates(directory="templates") -@app.get(fba.config["base_url"] + "/info") +@router.get(fba.config["base_url"] + "/api/info") def info(): - fba.cursor.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.cursor.fetchone() + 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, - "slogan" : fba.config["slogan"] + "known_instances" : known, + "indexed_instances" : indexed, + "blocks_recorded" : blocks, + "errorous_instances": errorous, + "slogan" : fba.config["slogan"] } -@app.get(fba.config["base_url"] + "/top") +@router.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: @@ -41,7 +59,7 @@ def top(blocked: int = None, blockers: int = None, reference: int = None, softwa elif software != None: if software > 500: raise HTTPException(status_code=400, detail="Too many results") - fba.cursor.execute("SELECT software, COUNT(domain) FROM instances GROUP BY software ORDER BY COUNT(domain) DESC LIMIT ?", [software]) + 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") @@ -57,33 +75,36 @@ def top(blocked: int = None, blockers: int = None, reference: int = None, softwa return scoreboard -@app.get(fba.config["base_url"] + "/api") +@router.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 = re.sub("(%|_)", "", reason) if len(reason) < 3: raise HTTPException(status_code=400, detail="Keyword is shorter than three characters") + 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_added, last_seen FROM blocks WHERE blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? ORDER BY first_added ASC", + 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: - fba.cursor.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: - fba.cursor.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM blocks WHERE reason like ? AND reason != '' ORDER BY first_added ASC", ["%" + reason + "%"]) + 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: + for blocker, blocked, block_level, reason, first_seen, last_seen in blocks: entry = { - "blocker" : blocker, - "blocked" : blocked, - "reason" : reason, - "first_added": first_added, - "last_seen" : last_seen + "blocker" : blocker, + "blocked" : blocked, + "reason" : reason, + "first_seen": first_seen, + "last_seen" : last_seen } if block_level in result: result[block_level].append(entry) @@ -92,18 +113,18 @@ def blocked(domain: str = None, reason: str = None, reverse: str = None): return result -@app.get(fba.config["base_url"] + "/scoreboard") +@router.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://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/top?blockers={blockers}") + 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://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/top?blocked={blocked}") + 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://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/top?reference={reference}") + 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://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/top?software={software}") + 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") @@ -112,8 +133,9 @@ def index(request: Request, blockers: int = None, blocked: int = None, reference elif not res.ok: raise HTTPException(status_code=res.status_code, detail=res.text) - return templates.TemplateResponse("index.html", { + return templates.TemplateResponse("scoreboard.html", { "base_url" : fba.config["base_url"], + "slogan" : fba.config["slogan"], "request" : request, "scoreboard": True, "blockers" : blockers, @@ -123,7 +145,7 @@ def index(request: Request, blockers: int = None, blocked: int = None, reference "scores" : res.json() }) -@app.get(fba.config["base_url"] + "/") +@router.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("/") @@ -132,18 +154,18 @@ def index(request: Request, domain: str = None, reason: str = None, reverse: str blocks = None if domain == None and reason == None and reverse == None: - info = requests.get(f"http://127.0.0.1:{fba.config['port']}{fba.config['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 = requests.get(f"http://127.0.0.1:{fba.config['port']}{fba.config['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 = requests.get(f"http://127.0.0.1:{fba.config['port']}{fba.config['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 = requests.get(f"http://127.0.0.1:{fba.config['port']}{fba.config['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: @@ -151,7 +173,7 @@ def index(request: Request, domain: str = None, reason: str = None, reverse: str 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", { @@ -163,7 +185,7 @@ def index(request: Request, domain: str = None, reason: str = None, reverse: str "info" : info }) -@app.get(fba.config["base_url"] + "/api/mutual") +@router.get(fba.config["base_url"] + "/api/mutual") def mutual(domains: list[str] = Query()): """Return 200 if federation is open between the two, 4xx otherwise""" fba.cursor.execute( @@ -172,8 +194,8 @@ def mutual(domains: list[str] = Query()): "AND block_level = 'reject' " \ "LIMIT 1", { - "a": domains[0], - "b": domains[1], + "a" : domains[0], + "b" : domains[1], "aw": "*." + domains[0], "bw": "*." + domains[1], }, @@ -187,44 +209,53 @@ def mutual(domains: list[str] = Query()): # No known blocks return responses.JSONResponse(status_code=200, content={}) -@app.get(fba.config["base_url"] + "/rss") +@router.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_added, last_seen FROM blocks WHERE blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? OR blocked = ? ORDER BY first_added DESC LIMIT 50", + 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_added, last_seen FROM blocks ORDER BY first_added DESC LIMIT 50") + 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_added, last_seen in blocks: - first_added = utils.format_datetime(datetime.fromtimestamp(first_added)) + 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, + "blocker" : blocker, + "blocked" : blocked, "block_level": block_level, - "reason": reason, - "first_added": first_added + "reason" : reason, + "first_seen" : first_seen }) timestamp = utils.format_datetime(datetime.now()) return templates.TemplateResponse("rss.xml", { - "request": request, + "request" : request, "timestamp": timestamp, - "domain": domain, - "blocks": result + "domain" : domain, + "blocks" : result + }, headers={ + "Content-Type": "routerlication/rss+xml" + }) + +@router.get(fba.config["base_url"] + "/robots.txt", response_class=PlainTextResponse) +def robots(request: Request): + return templates.TemplateResponse("robots.txt", { + "request" : request, + "base_url": fba.config["base_url"] }, headers={ - "Content-Type": "application/rss+xml" + "Content-Type": "text/plain" }) if __name__ == "__main__": - uvicorn.run("api:app", host="127.0.0.1", port=fba.config["port"], log_level=fba.config["log_level"]) + uvicorn.run("api:router", host=fba.config["host"], port=fba.config["port"], log_level=fba.config["log_level"])