api.py

   1 from fastapi import FastAPI, Request, HTTPException, responses, Query
   2 from fastapi.templating import Jinja2Templates
   3 from datetime import datetime
   4 from email import utils
   5
   6 import uvicorn
   7 import requests
   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, reference: int = None):
  29     if blocked == None and blockers == None and reference == 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     elif reference != None:
  40         if reference > 500:
  41             raise HTTPException(status_code=400, detail="Too many results")
  42         fba.c.execute("SELECT origin, COUNT(domain) FROM instances ORDER BY COUNT(domain) DESC LIMIT ?", [reference])
  43
  44     scores = fba.c.fetchall()
  45
  46     scoreboard = []
  47
  48     for domain, highscore in scores:
  49         scoreboard.append({"domain": domain, "highscore": highscore})
  50
  51     return scoreboard
  52
  53 @app.get(fba.config["base_url"] + "/api")
  54 def blocked(domain: str = None, reason: str = None, reverse: str = None):
  55     if domain == None and reason == None and reverse == None:
  56         raise HTTPException(status_code=400, detail="No filter specified")
  57     if reason != None:
  58         reason = re.sub("(%|_)", "", reason)
  59         if len(reason) < 3:
  60             raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
  61     if domain != None:
  62         wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
  63         punycode = domain.encode('idna').decode('utf-8')
  64         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",
  65                   (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
  66     elif reverse != None:
  67         fba.c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM blocks WHERE blocker = ? ORDER BY first_added ASC", [reverse])
  68     else:
  69         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 + "%"])
  70     blocks = fba.c.fetchall()
  71
  72     result = {}
  73     for blocker, blocked, block_level, reason, first_added, last_seen in blocks:
  74         entry = {
  75             "blocker": blocker,
  76             "blocked": blocked,
  77             "reason": reason,
  78             "first_added": first_added,
  79             "last_seen": last_seen
  80         }
  81         if block_level in result:
  82             result[block_level].append(entry)
  83         else:
  84             result[block_level] = [entry]
  85
  86     return result
  87
  88 @app.get(fba.config["base_url"] + "/scoreboard")
  89 def index(request: Request, blockers: int = None, blocked: int = None, reference: int = None, ):
  90     scores = None
  91
  92     if blockers == None and blocked == None and reference == None:
  93         raise HTTPException(status_code=400, detail="No filter specified")
  94     elif blockers != None:
  95         scores = requests.get(f"http://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/top?blockers={blockers}")
  96     elif blocked != None:
  97         scores = requests.get(f"http://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/top?blocked={blocked}")
  98     elif reference != None:
  99         scores = requests.get(f"http://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/top?reference={reference}")
 100
 101     if scores = None:
 102         raise HTTPException(status_code=500, detail="Could not determine scores")
 103     elif not scores.ok:
 104         raise HTTPException(status_code=scores.status_code, detail=scores.text)
 105
 106     return templates.TemplateResponse("index.html", {
 107         "base_url"  : fba.config["base_url"],
 108         "request"   : request,
 109         "scoreboard": True,
 110         "blockers"  : blockers,
 111         "blocked"   : blocked,
 112         "reference" : reference,
 113         "scores"    : scores.json()
 114     })
 115
 116 @app.get(fba.config["base_url"] + "/")
 117 def index(request: Request, domain: str = None, reason: str = None, reverse: str = None):
 118     if domain == "" or reason == "" or reverse == "":
 119         return responses.RedirectResponse("/")
 120
 121     info = None
 122     blocks = None
 123
 124     if domain == None and reason == None and reverse == None:
 125         info = requests.get(f"http://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/info")
 126
 127         if not info.ok:
 128             raise HTTPException(status_code=info.status_code, detail=info.text)
 129
 130         info = info.json()
 131     elif domain != None:
 132         blocks = requests.get(f"http://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/api?domain={domain}")
 133     elif reason != None:
 134         blocks = requests.get(f"http://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/api?reason={reason}")
 135     elif reverse != None:
 136         blocks = requests.get(f"http://127.0.0.1:{fba.config['port']}{fba.config['base_url']}/api?reverse={reverse}")
 137
 138     if blocks != None:
 139         if not blocks.ok:
 140             raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
 141         blocks = blocks.json()
 142         for block_level in blocks:
 143             for block in blocks[block_level]:
 144                 block["first_added"] = datetime.utcfromtimestamp(block["first_added"]).strftime('%Y-%m-%d %H:%M')
 145                 block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime('%Y-%m-%d %H:%M')
 146
 147     return templates.TemplateResponse("index.html", {"request": request, "domain": domain, "blocks": blocks, "reason": reason, "reverse": reverse, "info": info})
 148
 149 @app.get(fba.config["base_url"] + "/api/mutual")
 150 def mutual(domains: list[str] = Query()):
 151     """Return 200 if federation is open between the two, 4xx otherwise"""
 152     fba.c.execute(
 153         "SELECT block_level FROM blocks " \
 154         "WHERE ((blocker = :a OR blocker = :b) AND (blocked = :b OR blocked = :a OR blocked = :aw OR blocked = :bw)) " \
 155         "AND block_level = 'reject' " \
 156         "LIMIT 1",
 157         {
 158             "a": domains[0],
 159             "b": domains[1],
 160             "aw": "*." + domains[0],
 161             "bw": "*." + domains[1],
 162         },
 163     )
 164     res = fba.c.fetchone()
 165
 166     if res is not None:
 167         # Blocks found
 168         return responses.JSONResponse(status_code=418, content={})
 169
 170     # No known blocks
 171     return responses.JSONResponse(status_code=200, content={})
 172
 173 @app.get(fba.config["base_url"] + "/rss")
 174 def rss(request: Request, domain: str = None):
 175     if domain != None:
 176         wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
 177         punycode = domain.encode('idna').decode('utf-8')
 178         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",
 179                   (domain, "*." + domain, wildchar, fba.get_hash(domain), punycode, "*." + punycode))
 180     else:
 181         fba.c.execute("SELECT blocker, blocked, block_level, reason, first_added, last_seen FROM blocks ORDER BY first_added DESC LIMIT 50")
 182
 183     blocks = fba.c.fetchall()
 184
 185     result = []
 186     for blocker, blocked, block_level, reason, first_added, last_seen in blocks:
 187         first_added = utils.format_datetime(datetime.fromtimestamp(first_added))
 188         if reason == None or reason == '':
 189             reason = "No reason provided."
 190         else:
 191             reason = "Provided reason: '" + reason + "'"
 192
 193         result.append({
 194             "blocker": blocker,
 195             "blocked": blocked,
 196             "block_level": block_level,
 197             "reason": reason,
 198             "first_added": first_added
 199         })
 200
 201     timestamp = utils.format_datetime(datetime.now())
 202
 203     return templates.TemplateResponse("rss.xml", {
 204         "request": request,
 205         "timestamp": timestamp,
 206         "domain": domain,
 207         "blocks": result
 208     }, headers={
 209         "Content-Type": "application/rss+xml"
 210     })
 211
 212 if __name__ == "__main__":
 213     uvicorn.run("api:app", host="127.0.0.1", port=fba.config["port"], log_level=fba.config["log_level"])