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