api.py

   1 import uvicorn
   2 from fastapi import FastAPI, Request, HTTPException, responses
   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
  11 with open("config.json") as f:
  12     config = loads(f.read())
  13     base_url = config["base_url"]
  14     port = config["port"]
  15 app = FastAPI(docs_url=base_url+"/docs", redoc_url=base_url+"/redoc")
  16 templates = Jinja2Templates(directory=".")
  17
  18 def get_hash(domain: str) -> str:
  19     return sha256(domain.encode("utf-8")).hexdigest()
  20
  21 @app.get(base_url+"/info")
  22 def info():
  23     conn = sqlite3.connect("blocks.db")
  24     c = conn.cursor()
  25     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)")
  26     known, indexed, blocks = c.fetchone()
  27     c.close()
  28     return {
  29         "known_instances": known,
  30         "indexed_instances": indexed,
  31         "blocks_recorded": blocks
  32     }
  33
  34 @app.get(base_url+"/top")
  35 def top(blocked: int = None, blockers: int = None):
  36     conn = sqlite3.connect("blocks.db")
  37     c = conn.cursor()
  38     if blocked == None and blockers == None:
  39         raise HTTPException(status_code=400, detail="No filter specified")
  40     elif blocked != None:
  41         if blocked > 500:
  42             raise HTTPException(status_code=400, detail="Too many results")
  43         c.execute("select blocked, count(blocked) from blocks where block_level = 'reject' group by blocked order by count(blocked) desc limit ?", (blocked,))
  44     elif blockers != None:
  45         if blockers > 500:
  46             raise HTTPException(status_code=400, detail="Too many results")
  47         c.execute("select blocker, count(blocker) from blocks where block_level = 'reject' group by blocker order by count(blocker) desc limit ?", (blockers,))
  48     scores = c.fetchall()
  49     c.close()
  50
  51     scoreboard = []
  52     print(scores)
  53     for domain, highscore in scores:
  54         scoreboard.append({"domain": domain, "highscore": highscore})
  55
  56     return scoreboard
  57
  58 @app.get(base_url+"/api")
  59 def blocked(domain: str = None, reason: str = None, reverse: str = None):
  60     if domain == None and reason == None and reverse == None:
  61         raise HTTPException(status_code=400, detail="No filter specified")
  62     if reason != None:
  63         reason = sub("(%|_)", "", reason)
  64         if len(reason) < 3:
  65             raise HTTPException(status_code=400, detail="Keyword is shorter than three characters")
  66     conn = sqlite3.connect("blocks.db")
  67     c = conn.cursor()
  68     if domain != None:
  69         wildchar = "*." + ".".join(domain.split(".")[-domain.count("."):])
  70         punycode = domain.encode('idna').decode('utf-8')
  71         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",
  72                   (domain, "*." + domain, wildchar, get_hash(domain), punycode, "*." + punycode))
  73     elif reverse != None:
  74         c.execute("select blocker, blocked, block_level, reason, first_added, last_seen from blocks where blocker = ? order by first_added asc", (reverse,))
  75     else:
  76         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+"%",))
  77     blocks = c.fetchall()
  78     c.close()
  79
  80     result = {}
  81     for blocker, blocked, block_level, reason, first_added, last_seen in blocks:
  82         entry = {"blocker": blocker, "blocked": blocked, "reason": reason, "first_added": first_added, "last_seen": last_seen}
  83         if block_level in result:
  84             result[block_level].append(entry)
  85         else:
  86             result[block_level] = [entry]
  87
  88     return result
  89
  90 @app.get(base_url+"/scoreboard")
  91 def index(request: Request, blockers: int = None, blocked: int = None):
  92     if blockers == None and blocked == None:
  93         raise HTTPException(status_code=400, detail="No filter specified")
  94     elif blockers != None:
  95         scores = get(f"http://127.0.0.1:{port}{base_url}/top?blockers={blockers}")
  96     elif blocked != None:
  97         scores = get(f"http://127.0.0.1:{port}{base_url}/top?blocked={blocked}")
  98     if scores != None:
  99         if not scores.ok:
 100             raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
 101         scores = scores.json()
 102     return templates.TemplateResponse("index.html", {"request": request, "scoreboard": True, "blockers": blockers, "blocked": blocked, "scores": scores})
 103
 104 @app.get(base_url+"/")
 105 def index(request: Request, domain: str = None, reason: str = None, reverse: str = None):
 106     if domain == "" or reason == "" or reverse == "":
 107         return responses.RedirectResponse("/")
 108     info = None
 109     blocks = None
 110     if domain == None and reason == None and reverse == None:
 111         info = get(f"http://127.0.0.1:{port}{base_url}/info")
 112         if not info.ok:
 113             raise HTTPException(status_code=info.status_code, detail=info.text)
 114         info = info.json()
 115     elif domain != None:
 116         blocks = get(f"http://127.0.0.1:{port}{base_url}/api?domain={domain}")
 117     elif reason != None:
 118         blocks = get(f"http://127.0.0.1:{port}{base_url}/api?reason={reason}")
 119     elif reverse != None:
 120         blocks = get(f"http://127.0.0.1:{port}{base_url}/api?reverse={reverse}")
 121     if blocks != None:
 122         if not blocks.ok:
 123             raise HTTPException(status_code=blocks.status_code, detail=blocks.text)
 124         blocks = blocks.json()
 125         for block_level in blocks:
 126             for block in blocks[block_level]:
 127                 block["first_added"] = datetime.utcfromtimestamp(block["first_added"]).strftime('%Y-%m-%d %H:%M')
 128                 block["last_seen"] = datetime.utcfromtimestamp(block["last_seen"]).strftime('%Y-%m-%d %H:%M')
 129
 130     return templates.TemplateResponse("index.html", {"request": request, "domain": domain, "blocks": blocks, "reason": reason, "reverse": reverse, "info": info})
 131
 132 if __name__ == "__main__":
 133     uvicorn.run("api:app", host="127.0.0.1", port=port, log_level="info")