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['port']}{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
  92     if scores != None:
  93         if not scores.ok:
  94             raise HTTPException(status_code=scores.status_code, detail=scores.text)
  95
  96         scores = scores.json()
  97
  98     return templates.TemplateResponse("index.html", {
  99         "base_url": fba.config["base_url"],
 100         "request": request,
 101         "scoreboard": True,
 102         "blockers": blockers,
 103         "blocked": blocked,
 104         "scores": scores
 105     })
 106
 107 @app.get(fba.config["base_url"] + "/")
 108 def index(request: Request, domain: str = None, reason: str = None, reverse: str = None):
 109     if domain == "" or reason == "" or reverse == "":
 110         return responses.RedirectResponse("/")
 111
 112     info = None
 113     blocks = None
 114
 115     if domain == None and reason == None and reverse == None:
 116         info = get(f"http://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/info")
 117
 118         if not info.ok:
 119             raise HTTPException(status_code=info.status_code, detail=info.text)
 120
 121         info = info.json()
 122     elif domain != None:
 123         blocks = get(f"http://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/api?domain={domain}")
 124     elif reason != None:
 125         blocks = get(f"http://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/api?reason={reason}")
 126     elif reverse != None:
 127         blocks = get(f"http://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/api?reverse={reverse}")
 128
 129     if blocks != None:
 130         if not blocks.ok:
 131             raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
 132         blocks = blocks.json()
 133         for block_level in blocks:
 134             for block in blocks[block_level]:
 135                 block["first_added"] = datetime.utcfromtimestamp(block["first_added"]).strftime('%Y-%m-%d %H:%M')
 136                 block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime('%Y-%m-%d %H:%M')
 137
 138     return templates.TemplateResponse("index.html", {"request": request, "domain": domain, "blocks": blocks, "reason": reason, "reverse": reverse, "info": info})
 139
 140 @app.get(fba.config["base_url"] + "/api/mutual")
 141 def mutual(domains: list[str] = Query()):
 142     """Return 200 if federation is open between the two, 4xx otherwise"""
 143     fba.c.execute(
 144         "SELECT block_level FROM blocks " \
 145         "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
 146         "AND block_level = 'reject' " \
 147         "LIMIT 1",
 148         {
 149             "a": domains[0],
 150             "b": domains[1],
 151             "aw": "*." + domains[0],
 152             "bw": "*." + domains[1],
 153         },
 154     )
 155     res = fba.c.fetchone()
 156
 157     if res is not None:
 158         # Blocks found
 159         return responses.JSONResponse(status_code=418, content={})
 160
 161     # No known blocks
 162     return responses.JSONResponse(status_code=200, content={})
 163
 164 @app.get(fba.config["base_url"] + "/rss")
 165 def rss(request: Request, domain: str = None):
 166     if domain != None:
 167         wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
 168         punycode = domain.encode('idna').decode('utf-8')
 169         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",
 170                   (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
 171     else:
 172         fba.c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM blocks ORDER BY first_added DESC LIMIT 50")
 173
 174     blocks = fba.c.fetchall()
 175
 176     result = []
 177     for blocker, blocked, block_level, reason, first_added, last_seen in blocks:
 178         first_added = utils.format_datetime(datetime.fromtimestamp(first_added))
 179         if reason == None or reason == '':
 180             reason = "No reason provided."
 181         else:
 182             reason = "Provided reason: '" + reason + "'"
 183
 184         result.append({
 185             "blocker": blocker,
 186             "blocked": blocked,
 187             "block_level": block_level,
 188             "reason": reason,
 189             "first_added": first_added
 190         })
 191
 192     timestamp = utils.format_datetime(datetime.now())
 193
 194     return templates.TemplateResponse("rss.xml", {
 195         "request": request,
 196         "timestamp": timestamp,
 197         "domain": domain,
 198         "blocks": result
 199     }, headers={
 200         "Content-Type": "application/rss+xml"
 201     })
 202
 203 if __name__ == "__main__":
 204     uvicorn.run("api:app", host="127.0.0.1", port=fba.config["port"], log_level=fba.config["log_level"])