api.py

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