api.py

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