api.py

   1 import uvicorn
   2 from fastapi import FastAPI, Request, HTTPException, responses, Query
   3 import sqlite3
   4 from fastapi.templating import Jinja2Templates
   5 from requests import get
   6 from datetime import datetime
   7 from email import utils
   8 import re
   9 import fba
  10
  11 app = FastAPI(docs_url=fba.config["base_url"] + "/docs", redoc_url=fba.config["base_url"] + "/redoc")
  12 templates = Jinja2Templates(directory=".")
  13
  14 @app.get(fba.config["base_url"] + "/info")
  15 def info():
  16     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)")
  17     known, indexed, blocks = fba.c.fetchone()
  18     fba.c.close()
  19     return {
  20         "known_instances": known,
  21         "indexed_instances": indexed,
  22         "blocks_recorded": blocks,
  23         "slogan": fba.config["slogan"]
  24     }
  25
  26 @app.get(fba.config["base_url"] + "/top")
  27 def top(blocked: int = None, blockers: int = None):
  28     if blocked == None and blockers == None:
  29         raise HTTPException(status_code=400, detail="No filter specified")
  30     elif blocked != None:
  31         if blocked > 500:
  32             raise HTTPException(status_code=400, detail="Too many results")
  33         fba.c.execute("SELECT blocked, count(blocked) FROM blocks WHERE block_level = 'reject' group by blocked ORDER BY count(blocked) DESC LIMIT ?", (blocked,))
  34     elif blockers != None:
  35         if blockers > 500:
  36             raise HTTPException(status_code=400, detail="Too many results")
  37         fba.c.execute("SELECT blocker, count(blocker) FROM blocks WHERE block_level = 'reject' group by blocker ORDER BY count(blocker) DESC LIMIT ?", (blockers,))
  38     scores = fba.c.fetchall()
  39     fba.c.close()
  40
  41     scoreboard = []
  42     print(scores)
  43     for domain, highscore in scores:
  44         scoreboard.append({"domain": domain, "highscore": highscore})
  45
  46     return scoreboard
  47
  48 @app.get(fba.config["base_url"] + "/api")
  49 def blocked(domain: str = None, reason: str = None, reverse: str = None):
  50     if domain == None and reason == None and reverse == None:
  51         raise HTTPException(status_code=400, detail="No filter specified")
  52     if reason != None:
  53         reason = re.sub("(%|_)", "", reason)
  54         if len(reason) < 3:
  55             raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
  56     if domain != None:
  57         wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
  58         punycode = domain.encode('idna').decode('utf-8')
  59         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",
  60                   (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
  61     elif reverse != None:
  62         fba.c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM blocks WHERE blocker = ? ORDER BY first_added asc", (reverse,))
  63     else:
  64         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+"%",))
  65     blocks = fba.c.fetchall()
  66     fba.c.close()
  67
  68     result = {}
  69     for blocker, blocked, block_level, reason, first_added, last_seen in blocks:
  70         entry = {"blocker": blocker, "blocked": blocked, "reason": reason, "first_added": first_added, "last_seen": last_seen}
  71         if block_level in result:
  72             result[block_level].append(entry)
  73         else:
  74             result[block_level] = [entry]
  75
  76     return result
  77
  78 @app.get(fba.config["base_url"] + "/scoreboard")
  79 def index(request: Request, blockers: int = None, blocked: int = None):
  80     if blockers == None and blocked == None:
  81         raise HTTPException(status_code=400, detail="No filter specified")
  82     elif blockers != None:
  83         scores = get(f"http://127.0.0.1:{fba.config['base_url']}{fba.config['base_url']}/top?blockers={blockers}")
  84     elif blocked != None:
  85         scores = get(f"http://127.0.0.1:{fba.config['base_url']}{fba.config['base_url']}/top?blocked={blocked}")
  86     if scores != None:
  87         if not scores.ok:
  88             raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
  89         scores = scores.json()
  90     return templates.TemplateResponse("index.html", {"request": request, "scoreboard": True, "blockers": blockers, "blocked": blocked, "scores": scores})
  91
  92 @app.get(fba.config["base_url"] + "/")
  93 def index(request: Request, domain: str = None, reason: str = None, reverse: str = None):
  94     if domain == "" or reason == "" or reverse == "":
  95         return responses.RedirectResponse("/")
  96     info = None
  97     blocks = None
  98     if domain == None and reason == None and reverse == None:
  99         info = get(f"http://127.0.0.1:{fba.config['base_url']}{fba.config['base_url']}/info")
 100         if not info.ok:
 101             raise HTTPException(status_code=info.status_code, detail=info.text)
 102         info = info.json()
 103     elif domain != None:
 104         blocks = get(f"http://127.0.0.1:{fba.config['base_url']}{fba.config['base_url']}/api?domain={domain}")
 105     elif reason != None:
 106         blocks = get(f"http://127.0.0.1:{fba.config['base_url']}{fba.config['base_url']}/api?reason={reason}")
 107     elif reverse != None:
 108         blocks = get(f"http://127.0.0.1:{fba.config['base_url']}{fba.config['base_url']}/api?reverse={reverse}")
 109     if blocks != None:
 110         if not blocks.ok:
 111             raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
 112         blocks = blocks.json()
 113         for block_level in blocks:
 114             for block in blocks[block_level]:
 115                 block["first_added"] = datetime.utcfromtimestamp(block["first_added"]).strftime('%Y-%m-%d %H:%M')
 116                 block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime('%Y-%m-%d %H:%M')
 117
 118     return templates.TemplateResponse("index.html", {"request": request, "domain": domain, "blocks": blocks, "reason": reason, "reverse": reverse, "info": info})
 119
 120 @app.get(fba.config["base_url"] + "/api/mutual")
 121 def mutual(domains: list[str] = Query()):
 122     """Return 200 if federation is open between the two, 4xx otherwise"""
 123     fba.c.execute(
 124         "SELECT block_level FROM blocks " \
 125         "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
 126         "AND block_level = 'reject' " \
 127         "LIMIT 1",
 128         {
 129             "a": domains[0],
 130             "b": domains[1],
 131             "aw": "*." + domains[0],
 132             "bw": "*." + domains[1],
 133         },
 134     )
 135     res = fba.c.fetchone()
 136     fba.c.close()
 137     if res is not None:
 138         # Blocks found
 139         return responses.JSONResponse(status_code=418, content={})
 140     # No known blocks
 141     return responses.JSONResponse(status_code=200, content={})
 142
 143 @app.get(fba.config["base_url"] + "/rss")
 144 def rss(request: Request, domain: str = None):
 145     if domain != None:
 146         wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
 147         punycode = domain.encode('idna').decode('utf-8')
 148         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",
 149                   (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
 150     else:
 151         fba.c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM blocks ORDER BY first_added DESC LIMIT 50")
 152
 153     blocks = fba.c.fetchall()
 154     fba.c.close()
 155
 156     result = []
 157     for blocker, blocked, block_level, reason, first_added, last_seen in blocks:
 158         first_added = utils.format_datetime(datetime.fromtimestamp(first_added))
 159         if reason == None or reason == '':
 160             reason = "No reason provided."
 161         else:
 162             reason = "Provided reason: '" + reason + "'"
 163
 164         result.append(
 165             {
 166             "blocker": blocker,
 167             "blocked": blocked,
 168             "block_level": block_level,
 169             "reason": reason,
 170             "first_added": first_added
 171             }
 172         )
 173
 174     timestamp = utils.format_datetime(datetime.now())
 175
 176     return templates.TemplateResponse("rss.xml", {"request": request, "timestamp": timestamp, "domain": domain, "blocks": result}, headers={"Content-Type": "application/rss+xml"})
 177
 178 if __name__ == "__main__":
 179     uvicorn.run("api:app", host="127.0.0.1", port=fba.config["port"], log_level=fba.config["log_level"])