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